Adding apparatus to send HSV filter from the onboard drone component.

cmusatyalab · Mar 7, 2024 · b91724e · b91724e
1 parent 6cbd9aa
commit b91724e
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 0 deletions.
diff --git a/cnc/protocol/cnc.proto b/cnc/protocol/cnc.proto
@@ -43,6 +43,12 @@ message DroneStatus {
   int64 mag = 5;
 }
 
+message HSV {
+  int64 H = 1;
+  int64 S = 2;
+  int64 V = 3;
+}
+
 message Extras {
     bool registering = 1;
     string drone_id = 2;
@@ -51,4 +57,8 @@ message Extras {
     Location location = 5;
     DroneStatus status = 6;
     string detection_model = 7;
+
+    //for HSV filtering of object detection results
+    optional HSV lower_bound = 8;
+    optional HSV upper_bound = 9;
 }
diff --git a/hermes/python/interfaces/CloudletItf.py b/hermes/python/interfaces/CloudletItf.py
@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: GPL-2.0-only
 
 from abc import ABC, abstractmethod
+from typing import Tuple
 
 class CloudletItf(ABC):
 
@@ -22,6 +23,10 @@ def stopStreaming(self):
     def switchModel(self, model):
         pass
 
+    @abstractmethod
+    def setHSVFilter(self, lower_bound: Tuple[int, int, int], upper_bound: Tuple[int, int, int]):
+        pass
+
     @abstractmethod
     def sendFrame(self, frame):
         pass

diff --git a/onboard/python/implementation/cloudlets/PureOffloadCloudlet.py b/onboard/python/implementation/cloudlets/PureOffloadCloudlet.py
@@ -11,6 +11,7 @@
 import asyncio
 from syncer import sync
 import cv2
+from typing import Tuple
 
 from cnc_protocol import cnc_pb2
 from gabriel_protocol import gabriel_pb2
@@ -28,6 +29,8 @@ def __init__(self):
         self.drone = None
         self.sample_rate = 1
         self.stop = True
+        self.hsv_upper = [50,255,255]
+        self.hsv_lower = [30,100,100]
 
     def processResults(self, result_wrapper):
         if len(result_wrapper.results) != 1:
@@ -61,12 +64,22 @@ def stopStreaming(self):
     def switchModel(self, model):
         self.model = model
 
+    def setHSVFilter(self, lower_bound: Tuple[int, int, int], upper_bound: Tuple[int, int, int]):
+        self.hsv_lower = lower_bound
+        self.hsv_upper = upper_bound
+
     def produce_extras(self):
         extras = cnc_pb2.Extras()
         extras.drone_id = sync(self.drone.getName())
         extras.location.latitude = sync(self.drone.getLat())
         extras.location.longitude = sync(self.drone.getLng())
         extras.detection_model = self.model
+        extras.lower_bound.H = self.hsv_lower[0]
+        extras.lower_bound.S = self.hsv_lower[1]
+        extras.lower_bound.V = self.hsv_lower[2]
+        extras.upper_bound.H = self.hsv_upper[0]
+        extras.upper_bound.S = self.hsv_upper[1]
+        extras.upper_bound.V = self.hsv_upper[2]
         return extras
 
     def sendFrame(self):