Moving detection metrics into an Observer class.

1 files changed, 50 insertions, 42 deletions

diff --git a/src/argaze/ArUcoMarkers/ArUcoDetector.py b/src/argaze/ArUcoMarkers/ArUcoDetector.py
index a04bc8f..9f51490 100644
--- a/src/argaze/ArUcoMarkers/ArUcoDetector.py
+++ b/src/argaze/ArUcoMarkers/ArUcoDetector.py
@@ -133,7 +133,7 @@ class DetectorParameters():
 class ArUcoDetector(DataFeatures.PipelineStepObject):
     """OpenCV ArUco library wrapper."""
 
-    def __init__(self, dictionary: ArUcoMarkersDictionary.ArUcoMarkersDictionary = None, optic_parameters: ArUcoOpticCalibrator.OpticParameters = None, parameters: DetectorParameters = None):
+    def __init__(self, dictionary: ArUcoMarkersDictionary.ArUcoMarkersDictionary = None, optic_parameters: ArUcoOpticCalibrator.OpticParameters = None, parameters: DetectorParameters = None, **kwargs):
         """Initialize ArUcoDetector.
 
         Parameters:
@@ -143,7 +143,7 @@ class ArUcoDetector(DataFeatures.PipelineStepObject):
         """
 
         # Init parent class
-        super().__init__()
+        super().__init__(**kwargs)
 
         # Init private attributes
         self.__dictionary = dictionary
@@ -159,10 +159,6 @@ class ArUcoDetector(DataFeatures.PipelineStepObject):
         self.__board_corners = []
         self.__board_corners_ids = []
 
-        # Init detect metrics data
-        self.__detection_count = 0
-        self.__detected_ids = []
-
     @property
     def dictionary(self) -> ArUcoMarkersDictionary.ArUcoMarkersDictionary:
         """Get aruco detector's dictionary object."""
@@ -250,11 +246,19 @@ class ArUcoDetector(DataFeatures.PipelineStepObject):
 
             new_parameters = DetectorParameters()
 
+        # Load temporary pipeline step object from aruco_detector_data then export it as dict
+        temp_pipeline_step_object_data = DataFeatures.PipelineStepObject.from_dict(aruco_detector_data, working_directory).as_dict()
+
         # Create aruco detector
-        return ArUcoDetector(new_dictionary, new_optic_parameters, new_parameters)
+        return ArUcoDetector( \
+            new_dictionary, \
+            new_optic_parameters, \
+            new_parameters, \
+            **temp_pipeline_step_object_data \
+            )
 
     @DataFeatures.PipelineStepMethod
-    def detect_markers(self, timestamp: int|float, image: numpy.array) -> float:
+    def detect_markers(self, timestamp: int|float, image: numpy.array):
         """Detect all ArUco markers into an image.
 
         !!! danger "DON'T MIRROR IMAGE"
@@ -262,32 +266,20 @@ class ArUcoDetector(DataFeatures.PipelineStepObject):
 
         !!! danger "DON'T UNDISTORED IMAGE"
             Camera intrisic parameters and distorsion coefficients are used later during pose estimation.
-
-        Returns:
-            detection time: marker detection time in ms.
         """
 
         # Reset detected markers data
         self.__detected_markers, detected_markers_corners, detected_markers_ids = {}, [], []
 
-        # Store marker detection start date
-        detection_start = time.perf_counter()
-
         # Detect markers into gray picture
         detected_markers_corners, detected_markers_ids, _ = aruco.detectMarkers(cv.cvtColor(image, cv.COLOR_BGR2GRAY), self.__dictionary.markers, parameters = self.__parameters.internal)
         
-        # Assess marker detection time in ms
-        detection_time = (time.perf_counter() - detection_start) * 1e3
-
         # Is there detected markers ?
         if len(detected_markers_corners) > 0:
 
             # Transform markers ids array into list
             detected_markers_ids = detected_markers_ids.T[0]
 
-            # Gather detected markers data and update metrics
-            self.__detection_count += 1
-
             for i, marker_id in enumerate(detected_markers_ids):
 
                 marker = ArUcoMarker.ArUcoMarker(self.__dictionary, marker_id)
@@ -300,11 +292,6 @@ class ArUcoDetector(DataFeatures.PipelineStepObject):
 
                 self.__detected_markers[marker_id] = marker
 
-                # Update metrics
-                self.__detected_ids.append(marker_id)
-
-        return detection_time
-
     def estimate_markers_pose(self, markers_ids: list = []):
         """Estimate pose of current detected markers or of given markers id list."""
 
@@ -398,23 +385,6 @@ class ArUcoDetector(DataFeatures.PipelineStepObject):
 
             cv.drawChessboardCorners(image, ((self.__board.size[0] - 1 ), (self.__board.size[1] - 1)), self.__board_corners, True)
 
-    def reset_detection_metrics(self):
-        """Enable marker detection metrics."""
-
-        self.__detection_count = 0
-        self.__detected_ids = []
-
-    @property
-    def detection_metrics(self) -> Tuple[int, dict]:
-        """Get marker detection metrics.
-
-        Returns:
-            number of detect function call
-            dict with number of detection for each marker identifier
-        """
-
-        return self.__detection_count, Counter(self.__detected_ids)
-
     @property
     def board_corners_number(self) -> int:
         """Get detected board corners number."""
@@ -433,3 +403,41 @@ class ArUcoDetector(DataFeatures.PipelineStepObject):
 
         return self.__board_corners
 
+class Observer(DataFeatures.PipelineStepObserver):
+    """Define ArUcoDetector observer to count how many times detection succeeded and how many times markers are detected."""
+
+    def __init__(self):
+        """Initialize marker detection metrics."""
+
+        self.__try_count = 0
+        self.__success_count = 0
+        self.__detected_ids = []
+
+    @property
+    def metrics(self) -> Tuple[int, dict]:
+        """Get marker detection metrics.
+
+        Returns:
+            number of detect function call
+            dict with number of detection for each marker identifier
+        """
+
+        return self.__try_count, self.__success_count, Counter(self.__detected_ids)
+
+    def reset(self):
+        """Reset marker detection metrics."""
+
+        self.__try_count = 0
+        self.__success_count = 0
+        self.__detected_ids = []
+
+    def on_detect_markers(self, timestamp, aruco_detector, exception):
+        """Update ArUco markers detection metrics."""
+
+        self.__try_count += 1
+        detected_markers_list = list(aruco_detector.detected_markers.keys())
+
+        if len(detected_markers_list):
+
+            self.__success_count += 1
+            self.__detected_ids.extend(detected_markers_list)