Implemented terminal command processing, data post processing, data filters, and refresh rate tracking

.gitignore

@@ -50,4 +50,8 @@ __pycache__/
 firmware/panorama/.cache/
 
 # Runtime directory
-rundir/
+rundir/
+
+
+# Firmware related:
+firmware/panorama/.cache/

client/CMakeLists.txt

@@ -8,6 +8,12 @@ file(GLOB CLIENT_HEADERS CONFIGURE_DEPENDS include/client/*.hpp include/client/*
 set(WXMATHPLOT_BUILD_EXAMPLES OFF CACHE BOOL "" FORCE)
 add_subdirectory(external/wxMathPlot_0.2.0/mathplot)
 
+option(BENCHMARK "ENABLE" OFF)
+
+if (BENCHMARK)
+    add_compile_definitions(BENCHMARK)
+endif()
+
 # Executable settings
 if(APPLE)
     add_executable(panorama-client MACOSX_BUNDLE ${CLIENT_SOURCES} ${COMMON_HEADERS} ${CLIENT_HEADERS})
@@ -79,7 +85,7 @@ if(APPLE)
 
     message(STATUS "Using wxWidgets prebuilt from ${WXWIDGETS_PREBUILT_PATH}")
 else() # linux
-	
+
 
 	find_program(wxWidgets_CONFIG_EXECUTABLE wx-config)
     #set(wxWidgets_CONFIG_EXECUTABLE "${WXWIDGETS_PREBUILT_PATH}/bin/wx-config")

client/include/client/DataBuffer.hpp

@@ -63,6 +63,8 @@ class DataBuffer : public BufferBase<buffer_data_t> {
 
     std::string toStringAll();
 
+    void printAll();
+
     void exportBuffer(std::string exportPath);
 
 private:

client/include/client/buffer_base.hpp

@@ -99,5 +99,5 @@ class BufferBase {
     std::list<T> buffer_;
     mutable std::mutex mutex_;
     int MAX_BUFFER_SIZE = 500;
-    int FLUSH_THRESHOLD = 50; //percentage
+    int FLUSH_THRESHOLD = 100; //percentage
 };

client/include/client/command_processor.hpp

@@ -0,0 +1,25 @@
+#pragma once
+
+#include "client/DataBuffer.hpp"
+#include "client/post_processing.hpp"
+
+#include <string>
+#include <iostream>
+#include <atomic>
+#include <memory>
+
+class CommandProcessor {
+public:
+    CommandProcessor(std::shared_ptr<DataBuffer> dataBuffer, std::shared_ptr<PostProcessing> postProcessor);
+    void start(); //run command loop
+    void stop();
+
+private:
+    std::shared_ptr<DataBuffer> dataBuffer_;
+    std::shared_ptr<PostProcessing> postProcessor_;
+    std::string command;
+    std::atomic<bool> running_{true};
+
+    void processCommand(const std::string& command);
+};
+

client/include/client/data_filters.hpp

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <iostream>
+#include <vector>
+#include <deque>
+
+class DataFilters {
+public:
+    DataFilters();
+
+    int KalmanFilter(double input);
+    double MovingAverageFilter(double input);
+
+private:
+    size_t MAX_KALMAN_SIZE = 10;    
+    std::vector<double> kalmanList; 
+
+    size_t MAX_MOVINGAVERAGE = 10;
+    std::deque<double> movingAverageList;
+
+};

client/include/client/graph_panel.hpp

@@ -5,10 +5,12 @@
 #include <vector> 
 #include <set>
 #include <string>
+#include <memory>
+#include "client/post_processing.hpp"
 
 class GraphPanel : public wxPanel {
 public:
-    GraphPanel(wxWindow* parent);
+    GraphPanel(wxWindow* parent, std::shared_ptr<PostProcessing> postProcessor);
 
     void AddDataPoint(const std::string& sensorName, double value, double timestamp);
     void SetVisibleSensors(const std::set<std::string>& visisble);
@@ -30,4 +32,6 @@ class GraphPanel : public wxPanel {
     void UpdateGraph();
 
     wxDECLARE_EVENT_TABLE();
+
+    std::shared_ptr<PostProcessing> postProcessor_;
 };

client/include/client/mainframe.hpp

@@ -10,12 +10,14 @@
 #include <wx/socket.h>
 #include <vector>
 #include <memory>
+#include <atomic>
 #include <map>
 #include <atomic>
 #include "client/sensor_data_panel.h"
 #include "client/sensor_manager.hpp"
 #include "client/sensor.hpp"
 #include "client/graph_panel.hpp"
+#include "client/post_processing.hpp"
 #include <set>
 
 class MessageModel;
@@ -45,7 +47,7 @@ class MainFrame : public wxFrame {
     };
 
     MainFrame(const wxString& title, std::shared_ptr<MessageModel> model,
-        std::shared_ptr<DataBuffer> dataBuffer,
+        std::shared_ptr<DataBuffer> dataBuffer, std::shared_ptr<PostProcessing> postProcessor,
         TcpClient* tcpClient,
         const wxPoint& pos = wxDefaultPosition,
         const wxSize& size = wxSize(1200, 800));
@@ -91,12 +93,16 @@ class MainFrame : public wxFrame {
 
     void OnUpdateTimer(wxTimerEvent& event);
 
+    std::shared_ptr<PostProcessing> postProcessor_;
+
     // Auto detecting ESP32s
     std::unique_ptr<Esp32Scanner> esp32Scanner_;
     wxPanel* esp32Banner_ = nullptr;
     wxBoxSizer* mainSizer_ = nullptr;
     std::atomic<bool> esp32BannerPending_{false};
-    bool esp32BannerVisible_ = false;
+    bool esp32BannerVisible_ = false;    
+
+
 };
 
 #endif // __MAINFRAME__

client/include/client/post_processing.hpp

@@ -0,0 +1,21 @@
+#pragma once
+
+#include <mutex>
+#include <iostream>
+#include <list>
+
+class PostProcessing {
+public:
+    PostProcessing();
+    float processData(float data);
+    void reset();
+    void addOffset(float offset);
+    void addScaling(float scaleFactor);
+    void updateDataBase();
+
+private:
+    float currentOffset = 0.0;
+    float currentScaleFactor = 1.0;
+    std::mutex mutex_;
+
+};

client/include/client/tcp_client.hpp

@@ -33,11 +33,21 @@ class TcpClient {
     std::atomic<bool> running_;
     std::thread clientThread_;
 
+
+
 #ifdef _WIN32
     unsigned long long socket_; // SOCKET type on Windows
 #else
     int socket_;
 #endif
+
+#ifdef BENCHMARK
+    // track bitrate
+    std::time_t prevRefresh_ = std::time(nullptr);
+    double dataPerSecond_ = 0.0;
+    int dataCount_ = 0;
+
+#endif
 };
 
 #endif

client/src/DataBuffer.cpp

@@ -12,29 +12,18 @@ DataBuffer::~DataBuffer() {
 }
 
 void DataBuffer::writeData(buffer_data_t jsonChunk) {
-    // Append the new chunk of raw JSON data to the buffer. This functions only job is to store raw
-    // inbound data in a way that doesn't lose anything later the parser will call extractNextJson()
-    // parseNextJson()
-   // std::cout << "[DataBuffer] Writing data to buffer:" << std::endl;
-   // std::cout << "[DataBuffer]   a: '" << jsonChunk.a << "' a_data: " << jsonChunk.a_data << std::endl;
-   // std::cout << "[DataBuffer]   b: '" << jsonChunk.b << "' b_data: " << jsonChunk.b_data << std::endl;
-   // std::cout << "[DataBuffer]   Buffer size before write: " << size() << std::endl;
-
-   //get the runtime directory path from DataLogger to export buffer if it exceeds threshold
-
-   //DataLogger logger;
-   //std::string exportPath = logger.getLogFilePath();
 
+    //before writing, do necessary post processing on jsonChunk
+    //jsonChunk.data = PostProcessing.processData(jsonChunk.data);
+
     write(jsonChunk);
 
     if ((int)size() > FLUSH_THRESHOLD * MAX_BUFFER_SIZE / 100) {
         popFront();
         exportBuffer(logFilePath_);
     }
 
-    // std::cout << "[DataBuffer]   Buffer size after write: " << size() << std::endl;
-    // std::cout << buffer_.size();
-    // std::cout << "[DataBuffer] : " << toStringAll() << std::endl;
+
 }
 
 void DataBuffer::setData(buffer_data_t jsonData) {
@@ -153,6 +142,11 @@ std::string DataBuffer::toStringAll() {
     return res;
 }
 
+void DataBuffer::printAll() {
+    //print all buffer_ as string
+    std::cout << toStringAll() << std::endl;
+}
+
 void DataBuffer::exportBuffer(std::string exportPath) {
     //Export the entire buffer (make a local JSON file under client/src/)
 
@@ -170,4 +164,32 @@ void DataBuffer::exportBuffer(std::string exportPath) {
     fputs(json_content.c_str(), fp);
     fclose(fp);
     return;
-}
+}
+
+/*
+void DataBuffer::replaceData(std::string dataType, float targetValue) {
+    //replace all data of type dataType in buffer_ with targetValue
+    std::lock_guard<std::mutex> lock(mutex_);
+
+    std::cout << "Replacing data of type " << dataType << " with value " << targetValue << std::endl;
+    std::cout << "Buffer before replaceData: " << toStringAll() << std::endl;
+
+    for (auto& item : readAll()) {
+        if (item.datatype == dataType) {
+            item.data = targetValue;
+            std::cout << "Replaced data of type " << dataType << " with value " << targetValue << std::endl;
+        }
+    }
+}
+
+void DataBuffer::addOffset(std::string dataType, float offsetValue) {
+    //add offsetValue to all data of type dataType in buffer_
+    std::lock_guard<std::mutex> lock(mutex_);
+
+    for (auto& item : readAll()) {
+        if (item.datatype == dataType) {
+            item.data += offsetValue;
+        }
+    }
+}
+*/

client/src/command_processor.cpp

@@ -0,0 +1,72 @@
+#include "client/command_processor.hpp"
+#include "client/DataBuffer.hpp"
+
+/*
+TO DO: add stop mechanism for command processor thread, currently it runs indefinitely and can only be stopped by exiting the program
+TO DO: add command_processor in onExit() in main.cpp
+TO DO: handle race conditions between command processor and tcp client both accessing data buffer using mutex
+*/
+
+CommandProcessor::CommandProcessor(std::shared_ptr<DataBuffer> dataBuffer, std::shared_ptr<PostProcessing> postProcessor)
+    : dataBuffer_(dataBuffer), postProcessor_(postProcessor) {
+
+}
+
+void CommandProcessor::start() {
+    while(running_) {
+            std::cout << "Enter command (type 'exit' to quit): ";
+            std::getline(std::cin, command);
+
+            if (command == "exit") {
+                break;
+            }
+
+
+            processCommand(command);
+        }
+}
+
+void CommandProcessor::stop() {
+    running_ = false;
+}
+
+void CommandProcessor::processCommand(const std::string& command) {
+    /*
+    command:
+        <function> <parameter1> <parameter2>
+    */
+    std::string cmdType = command.substr(0, command.find(' '));
+
+    //get the parameters from command
+    size_t firstSpace = command.find(' ');
+    size_t secondSpace = command.find(' ', firstSpace + 1);
+    std::string firstParameter = command.substr(firstSpace + 1, secondSpace - firstSpace - 1);
+    std::string secondParameter = command.substr(secondSpace + 1);
+
+    if(cmdType == "printAll") {
+        dataBuffer_->printAll();
+
+    } else if (cmdType == "clear") {
+        dataBuffer_->clear();
+
+    } else if (cmdType == "size"){
+        std::cout << "Buffer size: " << dataBuffer_->size() << std::endl;
+
+    } else if (cmdType == "reset"){
+        //reset all post processing parameters to default values
+        postProcessor_->reset();
+
+    } else if (cmdType == "setOffset"){
+        float offset = std::stof(firstParameter);
+        postProcessor_->addOffset(offset);
+
+    } else if (cmdType == "setScale"){
+        float scaleFactor = std::stof(firstParameter);
+        postProcessor_->addScaling(scaleFactor);
+
+    } else {
+        std::cout << "Unknown command: " << command << std::endl;
+    }
+
+    //add more commands as needed
+}

client/src/data_filters.cpp

@@ -0,0 +1,28 @@
+#include "client/data_filters.hpp";
+
+
+DataFilters::DataFilters() {
+}
+
+
+int DataFilters::KalmanFilter(double input) {
+
+    // QUEUE DATA TYPE, FIRST IN FIRST OUT
+    return 0.0;
+}
+
+double DataFilters::MovingAverageFilter(double input) {
+    movingAverageList.push_front(input);
+    if (MAX_MOVINGAVERAGE <= movingAverageList.size()) {
+        movingAverageList.pop_back();
+    }
+
+    double sum = 0;
+    int i = 0;
+    while (i < movingAverageList.size()) {
+        sum += movingAverageList[i];
+        i++;
+    }
+
+    return sum / i;
+}