Implement brightspot detection

modules/detect_target/detect_target_brightspot.py

@@ -0,0 +1,153 @@
+"""
+Detects bright spots in images.
+"""
+
+import time
+
+import cv2
+import numpy as np
+
+from . import base_detect_target
+from .. import detections_and_time
+from .. import image_and_time
+from ..common.modules.logger import logger
+
+
+BRIGHTSPOT_PERCENTILE = 99.9
+
+# Label for brightspots; is 1 since 0 is used for blue landing pads
+DETECTION_LABEL = 1
+# SimpleBlobDetector is a binary detector, so a detection has confidence 1.0 by default
+CONFIDENCE = 1.0
+
+
+class DetectTargetBrightspot(base_detect_target.BaseDetectTarget):
+    """
+    Detects bright spots in images.
+    """
+
+    def __init__(
+        self,
+        local_logger: logger.Logger,
+        show_annotations: bool = False,
+        save_name: str = "",
+    ) -> None:
+        """
+        Initializes the bright spot detector.
+
+        show_annotations: Display annotated images.
+        save_name: Filename prefix for logging detections and annotated images.
+        """
+        self.__counter = 0
+        self.__local_logger = local_logger
+        self.__show_annotations = show_annotations
+        self.__filename_prefix = ""
+        if save_name != "":
+            self.__filename_prefix = f"{save_name}_{int(time.time())}_"
+
+    def run(
+        self, data: image_and_time.ImageAndTime
+    ) -> tuple[True, detections_and_time.DetectionsAndTime] | tuple[False, None]:
+        """
+        Runs brightspot detection on the provided image and returns the detections.
+
+        data: Image with a timestamp.
+
+        Return: Success, detections.
+        """
+        start_time = time.time()
+
+        image = data.image
+        try:
+            grey_image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        # Catching all exceptions for library call
+        # pylint: disable-next=broad-exception-caught
+        except Exception as exception:
+            self.__local_logger.error(
+                f"{time.time()}: Failed to convert to greyscale, exception: {exception}"
+            )
+            return False, None
+
+        brightspot_threshold = np.percentile(grey_image, BRIGHTSPOT_PERCENTILE)
+
+        # Apply thresholding to isolate bright spots
+        threshold_used, bw_image = cv2.threshold(
+            grey_image, brightspot_threshold, 255, cv2.THRESH_BINARY
+        )
+        if threshold_used == 0:
+            self.__local_logger.error(f"{time.time()}: Failed to threshold image.")
+            return False, None
+
+        # Set up SimpleBlobDetector
+        params = cv2.SimpleBlobDetector_Params()
+        params.filterByColor = True
+        params.blobColor = 255
+        params.filterByCircularity = False
+        params.filterByInertia = True
+        params.minInertiaRatio = 0.2
+        params.filterByConvexity = False
+        params.filterByArea = True
+        params.minArea = 50  # pixels
+
+        detector = cv2.SimpleBlobDetector_create(params)
+        keypoints = detector.detect(bw_image)
+
+        # A lack of detections is not an error, but should still not be forwarded
+        if len(keypoints) == 0:
+            self.__local_logger.info(f"{time.time()}: No brightspots detected.")
+            return False, None
+
+        # Annotate the image (green circle) with detected keypoints
+        image_annotated = cv2.drawKeypoints(
+            image, keypoints, None, (0, 255, 0), cv2.DRAW_MATCHES_FLAGS_DRAW_RICH_KEYPOINTS
+        )
+
+        # Process bright spot detection
+        result, detections = detections_and_time.DetectionsAndTime.create(data.timestamp)
+        if not result:
+            self.__local_logger.error(f"{time.time()}: Failed to create detections for image.")
+            return False, None
+
+        # Get Pylance to stop complaining
+        assert detections is not None
+
+        # Draw bounding boxes around detected keypoints
+        for keypoint in keypoints:
+            x, y = keypoint.pt
+            size = keypoint.size
+            bounds = np.array([x - size / 2, y - size / 2, x + size / 2, y + size / 2])
+            result, detection = detections_and_time.Detection.create(
+                bounds, DETECTION_LABEL, CONFIDENCE
+            )
+            if not result:
+                self.__local_logger.error(f"{time.time()}: Failed to create bounding boxes.")
+                return False, None
+
+            # Get Pylance to stop complaining
+            assert detections is not None
+
+            detections.append(detection)
+
+        # Logging is identical to detect_target_ultralytics.py
+        # pylint: disable=duplicate-code
+        end_time = time.time()
+
+        # Logging
+        self.__local_logger.info(
+            f"{time.time()}: Count: {self.__counter}. Target detection took {end_time - start_time} seconds. Objects detected: {detections}."
+        )
+
+        if self.__filename_prefix != "":
+            filename = self.__filename_prefix + str(self.__counter)
+
+            # Annotated image
+            cv2.imwrite(filename + ".png", image_annotated)  # type: ignore
+
+            self.__counter += 1
+
+        if self.__show_annotations:
+            cv2.imshow("Annotated", image_annotated)  # type: ignore
+
+        # pylint: enable=duplicate-code
+
+        return True, detections

modules/detect_target/detect_target_factory.py

@@ -5,6 +5,7 @@
 import enum
 
 from . import base_detect_target
+from . import detect_target_brightspot
 from . import detect_target_ultralytics
 from ..common.modules.logger import logger
 
@@ -15,6 +16,7 @@ class DetectTargetOption(enum.Enum):
     """
 
     ML_ULTRALYTICS = 0
+    CV_BRIGHTSPOT = 1
 
 
 def create_detect_target(
@@ -39,5 +41,11 @@ def create_detect_target(
                 show_annotations,
                 save_name,
             )
+        case DetectTargetOption.CV_BRIGHTSPOT:
+            return True, detect_target_brightspot.DetectTargetBrightspot(
+                local_logger,
+                show_annotations,
+                save_name,
+            )
 
     return False, None

tests/brightspot_example/bounding_box_ir_detections_0.txt

@@ -0,0 +1 @@
+1.000000 1.000000 545.888705 202.055392 555.831266 211.997953

tests/brightspot_example/bounding_box_ir_detections_1.txt

@@ -0,0 +1 @@
+1.000000 1.000000 443.409194 379.341292 453.529099 389.461198

tests/brightspot_example/bounding_box_ir_detections_2.txt

@@ -0,0 +1 @@
+1.000000 1.000000 270.872054 249.021590 288.934281 267.083817

tests/brightspot_example/bounding_box_ir_detections_3.txt

@@ -0,0 +1 @@
+1.000000 1.000000 630.931005 406.213048 640.793971 416.076014

tests/brightspot_example/bounding_box_ir_detections_4.txt

@@ -0,0 +1 @@
+1.000000 1.000000 407.681973 162.778408 426.180088 181.276524

tests/brightspot_example/generate_expected.py

@@ -0,0 +1,111 @@
+"""
+Generates expected output for the brightspot detector.
+"""
+
+import pathlib
+
+import cv2
+import numpy as np
+
+from modules import image_and_time
+from modules.common.modules.logger import logger
+from modules.detect_target import detect_target_brightspot
+
+
+TEST_PATH = pathlib.Path("tests", "brightspot_example")
+
+NUMBER_OF_IMAGES_DETECTIONS = 5
+IMAGE_DETECTIONS_FILES = [
+    pathlib.Path(f"ir_detections_{i}.png") for i in range(0, NUMBER_OF_IMAGES_DETECTIONS)
+]
+ANNOTATED_IMAGE_PATHS = [
+    pathlib.Path(TEST_PATH, f"ir_detections_{i}_annotated.png")
+    for i in range(0, NUMBER_OF_IMAGES_DETECTIONS)
+]
+EXPECTED_DETECTIONS_PATHS = [
+    pathlib.Path(TEST_PATH, f"bounding_box_ir_detections_{i}.txt")
+    for i in range(0, NUMBER_OF_IMAGES_DETECTIONS)
+]
+
+NUMBER_OF_IMAGES_NO_DETECTIONS = 2
+IMAGE_NO_DETECTIONS_FILES = [
+    pathlib.Path(f"ir_no_detections_{i}.png") for i in range(0, NUMBER_OF_IMAGES_NO_DETECTIONS)
+]
+
+
+def main() -> int:
+    """
+    Main function.
+    """
+    result, temp_logger = logger.Logger.create("test_logger", False)
+    if not temp_logger:
+        print("ERROR: Failed to create logger.")
+        return 1
+
+    detector = detect_target_brightspot.DetectTargetBrightspot(
+        local_logger=temp_logger, show_annotations=False, save_name=""
+    )
+
+    for image_file, annotated_image_path, expected_detections_path in zip(
+        IMAGE_DETECTIONS_FILES, ANNOTATED_IMAGE_PATHS, EXPECTED_DETECTIONS_PATHS
+    ):
+        image_path = pathlib.Path(TEST_PATH, image_file)
+        image = cv2.imread(str(image_path))  # type: ignore
+        result, image_data = image_and_time.ImageAndTime.create(image)
+        if not result:
+            temp_logger.error(f"Failed to load image {image_path}.")
+            continue
+
+        # Get Pylance to stop complaining
+        assert image_data is not None
+
+        result, detections = detector.run(image_data)
+        if not result:
+            temp_logger.error(f"Unable to get detections for {image_path}.")
+            continue
+
+        # Get Pylance to stop complaining
+        assert detections is not None
+
+        detections_list = []
+        image_annotated = image.copy()
+        for detection in detections.detections:
+            confidence = detection.confidence
+            label = detection.label
+            x_1 = detection.x_1
+            y_1 = detection.y_1
+            x_2 = detection.x_2
+            y_2 = detection.y_2
+            detections_list.append([confidence, label, x_1, y_1, x_2, y_2])
+
+            cv2.rectangle(image_annotated, (int(x_1), int(y_1)), (int(x_2), int(y_2)), (0, 255, 0), 1)  # type: ignore
+
+        detections_array = np.array(detections_list)
+
+        np.savetxt(expected_detections_path, detections_array, fmt="%.6f")
+        temp_logger.info(f"Expected detections saved to {expected_detections_path}.")
+
+        result = cv2.imwrite(str(annotated_image_path), image_annotated)  # type: ignore
+        if not result:
+            temp_logger.error(f"Failed to write image to {annotated_image_path}.")
+            continue
+
+        temp_logger.info(f"Annotated image saved to {annotated_image_path}.")
+
+    for image_file in IMAGE_NO_DETECTIONS_FILES:
+        result, detections = detector.run(image_data)
+        if result:
+            temp_logger.error(f"False positive detections in {image_path}.")
+            continue
+
+        assert detections is None
+
+    return 0
+
+
+if __name__ == "__main__":
+    result_main = main()
+    if result_main < 0:
+        print(f"ERROR: Status code: {result_main}")
+    else:
+        print("Done!")