Merge pull request #304 from tritonuas/feat/data-aggregation

Add data aggregation structure to retain data on the obc and overwrite it with new data from the gcs

Author: WilyHyperion

include/cv/aggregator.hpp

@@ -11,6 +11,7 @@
 #include <thread>
 #include <unordered_map>
 #include <vector>
+#include <map>
 
 #include "cv/pipeline.hpp"
 #include "cv/utilities.hpp"
@@ -50,7 +51,16 @@ class CVAggregator {
     // For the endpoint to reset the current list of structs
     std::vector<AggregatedRun> popAllRuns();
 
+    // gets the record of all cv results
+    LockPtr<std::map<int, IdentifiedTarget>> getCVRecord();
+
+    void updateRecords(std::vector<IdentifiedTarget>& new_values);
+
  private:
+    std::mutex cv_record_mut;
+
+    std::shared_ptr<std::map<int, IdentifiedTarget>> cv_record;
+
     void worker(ImageData image, int thread_num);
 
     Pipeline pipeline;

include/cv/pipeline.hpp

@@ -3,6 +3,7 @@
 
 #include <cmath>
 #include <memory>
+#include <optional>
 #include <string>
 #include <utility>
 #include <vector>
@@ -29,14 +30,21 @@ struct PipelineResults {
 };
 
 struct PipelineParams {
-    // Added outputPath parameter with default empty string
-    explicit PipelineParams(std::string yoloModelPath, std::string outputPath = "",
+    // yoloModelPath is optional; when absent, no CV models will be loaded.
+    explicit PipelineParams(std::optional<std::string> yoloModelPath = std::nullopt,
+                            std::string outputPath = "",
                             bool do_preprocess = true)
         : yoloModelPath{std::move(yoloModelPath)},
           outputPath(std::move(outputPath)),
           do_preprocess(do_preprocess) {}
 
-    std::string yoloModelPath;
+    explicit PipelineParams(const std::string& yoloModelPath,
+                            std::string outputPath = "",
+                            bool do_preprocess = true)
+        : PipelineParams(std::optional<std::string>(yoloModelPath), std::move(outputPath),
+                         do_preprocess) {}
+
+    std::optional<std::string> yoloModelPath;
     bool do_preprocess;
     std::string outputPath;
 };

src/core/mission_state.cpp

@@ -128,3 +128,5 @@ void MissionState::setCamera(std::shared_ptr<CameraInterface> camera) { this->ca
 bool MissionState::getMappingIsDone() { return this->mappingIsDone; }
 
 void MissionState::setMappingIsDone(bool isDone) { this->mappingIsDone = isDone; }
+
+

src/cv/aggregator.cpp

@@ -9,7 +9,7 @@ CVAggregator::CVAggregator(Pipeline&& p) : pipeline(std::move(p)) {
     this->num_worker_threads = 0;
     this->results = std::make_shared<CVResults>();
     this->matched_results = std::make_shared<MatchedResults>();
-
+    this->cv_record = std::make_shared<std::map<int, IdentifiedTarget>>();
     AirdropTarget dummy;  // Create one dummy template
 
     // Configure the coordinate part of the dummy
@@ -36,6 +36,19 @@ LockPtr<MatchedResults> CVAggregator::getMatchedResults() {
     return LockPtr<MatchedResults>(this->matched_results, &this->mut);
 }
 
+LockPtr<std::map<int, IdentifiedTarget>> CVAggregator::getCVRecord() {
+    return LockPtr<std::map<int, IdentifiedTarget>>(this->cv_record, &this->cv_record_mut);
+}
+void CVAggregator::updateRecords(std::vector<IdentifiedTarget>& new_values) {
+    LockPtr<std::map<int, IdentifiedTarget>> records = this->getCVRecord();
+    for (IdentifiedTarget id : new_values) {
+        if (records.data->contains(id.run_id())) {
+            records.data->at(id.run_id()).CopyFrom(id);
+        } else {
+            LOG_F(WARNING, "Tried to modify with an ID not inside CVRecord. ID: (%d)", id.run_id());
+        }
+    }
+}
 void CVAggregator::runPipeline(const ImageData& image) {
     Lock lock(this->mut);
 

src/cv/pipeline.cpp

@@ -7,9 +7,18 @@
 
 // Pipeline constructor: initialize YOLO detector and the preprocess flag.
 Pipeline::Pipeline(const PipelineParams& p)
-    : yoloDetector(std::make_unique<YOLO>(p.yoloModelPath, 0.05f, 640, 640)),
-      outputPath(p.outputPath),
-      do_preprocess(p.do_preprocess) {}
+    : outputPath(p.outputPath),
+      do_preprocess(p.do_preprocess) {
+    if (p.yoloModelPath.has_value() && !p.yoloModelPath->empty()) {
+        yoloDetector = std::make_unique<YOLO>(*p.yoloModelPath, 0.05f, 640, 640);
+        LOG_F(INFO, "YOLO model loaded from path: %s", p.yoloModelPath->c_str());
+    } else {
+        yoloDetector.reset();
+        LOG_F(WARNING, "No CV models are loaded (no YOLO model path provided).");
+        LOG_F(WARNING, "CVAGGREGATOR WILL NOT BE WORKING AS INTENDED. USE AT YOUR OWN RISK.");
+        LOG_F(WARNING, "Provide a YOLO model path to enable detections.");
+    }
+}
 
 PipelineResults Pipeline::run(const ImageData& imageData) {
     LOG_F(INFO, "Running pipeline on an image");
@@ -88,7 +97,9 @@ PipelineResults Pipeline::run(const ImageData& imageData) {
 
     // 3) DRAW DETECTIONS ON THE IMAGE
     //    (this modifies processedImage in-place)
-    this->yoloDetector->drawAndPrintDetections(processedImage, yoloResults);
+    if (this->yoloDetector) {
+        this->yoloDetector->drawAndPrintDetections(processedImage, yoloResults);
+    }
 
     // Save the annotated image if an output path is specified
     if (!outputPath.empty()) {

src/network/gcs_routes.cpp

@@ -351,7 +351,6 @@ DEF_GCS_HANDLE(Post, takeoff, autonomous) {
 
 DEF_GCS_HANDLE(Get, targets, all) {
     LOG_REQUEST("GET", "/targets/all");
-
     auto aggregator = state->getCV();
     if (!aggregator) {
         LOG_RESPONSE(ERROR, "CV not connected yet", BAD_REQUEST);
@@ -370,13 +369,15 @@ DEF_GCS_HANDLE(Get, targets, all) {
     std::vector<IdentifiedTarget> out_data;
     out_data.reserve(new_runs.size());  // Reserve space for efficiency
 
+    // get aggregate to store a record of these results
+    LockPtr<std::map<int, IdentifiedTarget>> records = aggregator->getCVRecord();
+    std::shared_ptr<std::map<int, IdentifiedTarget>> records_ptr = records.data;
+
     for (const auto& run : new_runs) {
         // Create ONE IdentifiedTarget message per AggregatedRun
         IdentifiedTarget target;
-
         // Set the run ID
         target.set_run_id(run.run_id);
-
         // START COMPRESSION
 
         // Compress the annotated image before converting to base64
@@ -409,10 +410,8 @@ DEF_GCS_HANDLE(Get, targets, all) {
 
         // Convert the compressed image to base64 and set it (once per run)
         std::string b64 = cvMatToBase64(compressed_image);
-
-        // END COMPRESSION
-
         target.set_picture(b64);
+        // END COMPRESSION
 
         // Ensure coords and bboxes vectors are the same size (should be guaranteed by Aggregator
         // logic)
@@ -440,10 +439,15 @@ DEF_GCS_HANDLE(Get, targets, all) {
             proto_bbox->set_y2(run.bboxes[i].y2);
         }
 
+        // copy the target to a record object to store
+        IdentifiedTarget record;
+        record.CopyFrom(target);
+        // remove image
+        record.set_picture("");
+        records_ptr->insert_or_assign(run.run_id, target);
         // Add the completed IdentifiedTarget (representing the whole run) to the output list
         out_data.push_back(std::move(target));
     }  // End loop over AggregatedRuns
-
     // 3) Serialize the vector of IdentifiedTarget messages to JSON
     // Ensure messagesToJson can handle a vector or use iterators correctly
     std::string out_data_json = messagesToJson(out_data.begin(), out_data.end());

src/utilities/datatypes.cpp

@@ -111,3 +111,4 @@ std::string AirdropTypeObjectsToString(const AirdropType& color) {
         default: return "IDFK";
     }
 }
+

tests/unit/modify_runs.cpp

@@ -0,0 +1,78 @@
+#include <gtest/gtest.h>
+#include "cv/aggregator.hpp"
+TEST(ModifyRuns, OverWriteData)
+{
+    PipelineParams params(std::nullopt,
+                          "",
+                          false);
+    Pipeline pipeline(params);
+    CVAggregator aggregator(std::move(pipeline));
+
+    std::vector<IdentifiedTarget> updated;
+    {
+        auto lock = aggregator.getCVRecord();
+        auto ptr = lock.data;
+        // Add some mock runs
+        for (int i = 1; i < 11; i++)
+        {
+            IdentifiedTarget target;
+            target.set_run_id(i);
+            // set mock bounding box
+            GPSCoord *proto_coord = target.add_coordinates();
+            proto_coord->set_latitude(0);
+            proto_coord->set_longitude(0);
+            proto_coord->set_altitude(0);
+
+            BboxProto *proto_bbox = target.add_bboxes();
+            proto_bbox->set_x1(0);
+            proto_bbox->set_y1(0);
+            proto_bbox->set_x2(0);
+            proto_bbox->set_y2(0);
+
+            ptr->insert_or_assign(i, target);
+        }
+
+        for (int i = 1; i < 11; i++)
+        {
+            IdentifiedTarget target;
+            target.set_run_id(i);
+
+            GPSCoord *proto_coord = target.add_coordinates(); // Use the plural field name
+            proto_coord->set_latitude(i);
+            proto_coord->set_longitude(i);
+            proto_coord->set_altitude(i);
+
+            // Add bounding box
+            BboxProto *proto_bbox = target.add_bboxes(); // Use the plural field name
+            proto_bbox->set_x1(i);
+            proto_bbox->set_y1(i);
+            proto_bbox->set_x2(i);
+            proto_bbox->set_y2(i);
+            ptr->insert_or_assign(i, target);
+
+            updated.push_back(target);
+        }
+        // deallocate to unlock the mutex
+    }
+    aggregator.updateRecords(updated);
+
+    for (auto value : updated)
+    {
+        auto lock = aggregator.getCVRecord();
+        auto ptr = lock.data;
+        ASSERT_TRUE(ptr->contains(value.run_id()));
+        for (int i = 0; i < value.bboxes_size(); i++)
+        {
+            ASSERT_EQ(value.bboxes().at(i).x1(), ptr->at(value.run_id()).bboxes().at(i).x1());
+            ASSERT_EQ(value.bboxes().at(i).y1(), ptr->at(value.run_id()).bboxes().at(i).y1());
+            ASSERT_EQ(value.bboxes().at(i).x2(), ptr->at(value.run_id()).bboxes().at(i).x2());
+            ASSERT_EQ(value.bboxes().at(i).y2(), ptr->at(value.run_id()).bboxes().at(i).y2());
+        }
+        for (int i = 0; i < value.coordinates_size(); i++)
+        {
+            ASSERT_EQ(value.coordinates().at(i).altitude(), ptr->at(value.run_id()).coordinates().at(i).altitude());
+            ASSERT_EQ(value.coordinates().at(i).longitude(), ptr->at(value.run_id()).coordinates().at(i).longitude());
+            ASSERT_EQ(value.coordinates().at(i).latitude(), ptr->at(value.run_id()).coordinates().at(i).latitude());
+        }
+    }
+}