Detections UI (NickM-27#32)

* Cleanups and removing redundant code

* Get detection events from API

* Add detections route

* Fix parsing

* Add cards to detections

* UI tweaks

* Save and get snapshot with event id

* Formatting

* Set correct snapshot decoding

* Add network image

* Add network image

* Ui styling

* Add ability to delete detections

swatch/app.py

@@ -69,6 +69,7 @@ def __init_processing__(self) -> None:
             if config.auto_detect > 0 and config.snapshot_config.url:
                 self.camera_processes[name] = AutoDetector(
                     self.image_processor,
+                    self.snapshot_processor,
                     config,
                     self.stop_event,
                 )

swatch/detection.py

@@ -10,6 +10,7 @@
 from swatch.config import CameraConfig
 from swatch.image import ImageProcessor
 from swatch.models import Detection
+from swatch.snapshot import SnapshotProcessor
 
 
 class AutoDetector(threading.Thread):
@@ -18,11 +19,13 @@ class AutoDetector(threading.Thread):
     def __init__(
         self,
         image_processor: ImageProcessor,
+        snap_processor: SnapshotProcessor,
         camera_config: CameraConfig,
         stop_event: multiprocessing.Event,
     ) -> None:
         threading.Thread.__init__(self)
         self.image_processor = image_processor
+        self.snap_processor = snap_processor
         self.config = camera_config
         self.stop_event = stop_event
         self.obj_data: Dict[str, Any] = {}
@@ -69,6 +72,12 @@ def __handle_detections__(self, detection_result: Dict[str, Any]) -> None:
                 if not self.obj_data.get(non_unique_id):
                     self.obj_data[non_unique_id] = {}
 
+                unique_id = (
+                    f"{non_unique_id}.{''.join(random.choices(string.ascii_lowercase + string.digits, k=6))}"
+                    if not self.obj_data[non_unique_id].get("id")
+                    else self.obj_data[non_unique_id]["id"]
+                )
+
                 self.obj_data[non_unique_id]["object_name"] = object_name
                 self.obj_data[non_unique_id]["zone_name"] = zone_name
                 self.obj_data[non_unique_id]["variant"] = object_result["variant"]
@@ -78,8 +87,15 @@ def __handle_detections__(self, detection_result: Dict[str, Any]) -> None:
                 ):
                     self.obj_data[non_unique_id]["top_area"] = object_result["area"]
 
+                    # save snapshot with best area
+                    self.snap_processor.save_snapshot(
+                        cam_name,
+                        zone_name,
+                        f"{unique_id}.jpg",
+                        None,
+                    )
+
                 if not self.obj_data[non_unique_id].get("id"):
-                    unique_id = f"{non_unique_id}.{''.join(random.choices(string.ascii_lowercase + string.digits, k=6))}"
                     self.obj_data[non_unique_id]["id"] = unique_id
                     self.__handle_db__("new", non_unique_id)
                 else:
@@ -100,7 +116,7 @@ def run(self) -> None:
 
         # ensure db doesn't contain bad data after shutdown
         Detection.update(end_time=datetime.datetime.now().timestamp()).where(
-            Detection.end_time == None
+            Detection.end_time is None
         ).execute()
         print(f"Stopping Auto Detection for {self.config.name}")
 

swatch/http.py

@@ -154,6 +154,7 @@ def get_detections() -> Any:
         Detection.top_area,
         Detection.color_variant,
         Detection.start_time,
+        Detection.end_time,
     ]
 
     if camera != "all":
@@ -185,13 +186,39 @@ def get_detections() -> Any:
 
 
 @bp.route("/detections/<detection_id>", methods=["GET"])
-def get_detection(id: str):
+def get_detection(detection_id: str):
     """Get specific detection."""
     try:
-        return model_to_dict(Detection.get(Detection.id == id))
+        return model_to_dict(Detection.get(Detection.id == detection_id))
     except DoesNotExist:
         return jsonify(
-            {"success": False, "message": f"Detection with id {id} not found."}, 404
+            {
+                "success": False,
+                "message": f"Detection with id {detection_id} not found.",
+            },
+            404,
+        )
+
+
+@bp.route("/detections/<detection_id>", methods=["DELETE"])
+def delete_detection(detection_id: str):
+    """Get specific detection."""
+    try:
+        Detection.delete().where(Detection.id == detection_id).execute()
+        return jsonify(
+            {
+                "success": True,
+                "message": "Deleted successfully.",
+            },
+            200,
+        )
+    except DoesNotExist:
+        return jsonify(
+            {
+                "success": False,
+                "message": f"Detection with id {detection_id} not found.",
+            },
+            404,
         )
 
 
@@ -346,6 +373,36 @@ def get_latest_zone_snapshot(camera_name: str, zone_name: str) -> Any:
     return response
 
 
+@bp.route("/detections/<detection_id>/snapshot.jpg", methods=["GET"])
+def get_detection_snapshot(detection_id: str):
+    """Get specific detection snapshot."""
+    try:
+        detection = Detection.get(Detection.id == detection_id)
+        jpg_bytes = current_app.snapshot_processor.get_detection_snapshot(detection)
+
+        if jpg_bytes:
+            response = make_response(jpg_bytes)
+            response.headers["Content-Type"] = "image/jpg"
+            return response
+
+        return jsonify(
+            {
+                "success": False,
+                "message": f"Error loading snapshot for {detection_id}.",
+            },
+            404,
+        )
+
+    except DoesNotExist:
+        return jsonify(
+            {
+                "success": False,
+                "message": f"Detection with id {detection_id} not found.",
+            },
+            404,
+        )
+
+
 @bp.route("/<camera_name>/detection.jpg", methods=["GET"])
 def get_latest_detection(camera_name: str) -> Any:
     """Get the latest detection for <camera_name>."""
@@ -361,7 +418,9 @@ def get_latest_detection(camera_name: str) -> Any:
             {"success": False, "message": f"{camera_name} is not a valid camera."}, 404
         )
 
-    jpg_bytes = current_app.snapshot_processor.get_latest_detection(camera_name)
+    jpg_bytes = current_app.snapshot_processor.get_latest_detection_snapshot(
+        camera_name
+    )
 
     response = make_response(jpg_bytes)
     response.headers["Content-Type"] = "image/jpg"

swatch/image.py

@@ -77,13 +77,16 @@ def __check_image__(
 
             output, matches = __mask_image__(crop, color_variant)
 
-            if matches > detectable.min_area and matches < detectable.max_area:
+            if detectable.min_area < matches < detectable.max_area:
                 if snapshot.save_detections and snapshot.mode in [
                     SnapshotModeEnum.ALL,
                     SnapshotModeEnum.MASK,
                 ]:
                     self.snapshot_processor.save_snapshot(
-                        camera_name, f"detected_{variant_name}_{file_name}", output
+                        camera_name,
+                        "",
+                        f"detected_{variant_name}_{file_name}_{datetime.datetime.now().strftime('%f')}.jpg",
+                        output,
                     )
 
                 return {
@@ -92,22 +95,25 @@ def __check_image__(
                     "variant": variant_name,
                     "camera_name": camera_name,
                 }
-            else:
-                if matches > best_fail.get("area", 0):
-                    best_fail = {
-                        "result": False,
-                        "area": matches,
-                        "variant": variant_name,
-                        "camera_name": camera_name,
-                    }
-
-                if snapshot.save_misses and snapshot.mode in [
-                    SnapshotModeEnum.ALL,
-                    SnapshotModeEnum.MASK,
-                ]:
-                    self.snapshot_processor.save_snapshot(
-                        camera_name, f"missed_{variant_name}_{file_name}", output
-                    )
+
+            if matches > best_fail.get("area", 0):
+                best_fail = {
+                    "result": False,
+                    "area": matches,
+                    "variant": variant_name,
+                    "camera_name": camera_name,
+                }
+
+            if snapshot.save_misses and snapshot.mode in [
+                SnapshotModeEnum.ALL,
+                SnapshotModeEnum.MASK,
+            ]:
+                self.snapshot_processor.save_snapshot(
+                    camera_name,
+                    "",
+                    f"missed_{variant_name}_{file_name}_{datetime.datetime.now().strftime('%f')}.jpg",
+                    output,
+                )
 
         return best_fail
 
@@ -121,7 +127,7 @@ def detect(self, camera_name: str, image_url: str) -> Dict[str, Any]:
             response[zone_name] = {}
             imgBytes = requests.get(image_url).content
 
-            if not imgBytes:
+            if imgBytes is None:
                 continue
 
             img = cv2.imdecode(np.asarray(bytearray(imgBytes), dtype=np.uint8), -1)
@@ -151,7 +157,10 @@ def detect(self, camera_name: str, image_url: str) -> Dict[str, Any]:
                     SnapshotModeEnum.CROP,
                 ]:
                     self.snapshot_processor.save_snapshot(
-                        camera_name, f"{zone_name}", crop
+                        camera_name,
+                        zone_name,
+                        f"{zone_name}_{datetime.datetime.now().strftime('%f')}.jpg",
+                        crop,
                     )
 
                 self.latest_results[object_name] = result

swatch/snapshot.py

@@ -14,6 +14,7 @@
 
 from swatch.config import SwatchConfig, CameraConfig
 from swatch.const import CONST_MEDIA_DIR
+from swatch.models import Detection
 
 
 def delete_dir(date_dir: str, camera_name: str):
@@ -39,10 +40,11 @@ def __init__(self, config: SwatchConfig) -> None:
     def save_snapshot(
         self,
         camera_name: str,
+        zone_name: str,
         file_name: str,
         image: ndarray,
     ) -> bool:
-        """Saves the snapshot to the correct snapshot dir."""
+        """Saves the file snapshot to the correct snapshot dir."""
         time = datetime.datetime.now()
 
         file_dir = f"{CONST_MEDIA_DIR}/snapshots/{time.strftime('%m-%d')}/{camera_name}"
@@ -53,20 +55,63 @@ def save_snapshot(
             print(f"after creating {os.listdir('/media/')}")
             return False
 
-        file = f"{file_dir}/{file_name}_{time.strftime('%f')}.jpg"
-        cv2.imwrite(file, image)
+        file = f"{file_dir}/{file_name}"
+
+        if image is not None:
+            cv2.imwrite(file, image)
+        else:
+            imgBytes = requests.get(
+                self.config.cameras[camera_name].snapshot_config.url
+            ).content
+
+            if imgBytes is None:
+                return False
+
+            img = cv2.imdecode(np.asarray(bytearray(imgBytes), dtype=np.uint8), -1)
+
+            coordinates = (
+                self.config.cameras[camera_name]
+                .zones[zone_name]
+                .coordinates.split(", ")
+            )
+
+            if img.size > 0:
+                crop = img[
+                    int(coordinates[1]) : int(coordinates[3]),
+                    int(coordinates[0]) : int(coordinates[2]),
+                ]
+
+            cv2.imwrite(file, crop)
+
         return True
 
+    def get_detection_snapshot(self, detection: Detection) -> Any:
+        """Get file snapshot for a specific detection."""
+        file_dir = f"{CONST_MEDIA_DIR}/snapshots/{datetime.datetime.fromtimestamp(detection.start_time).strftime('%m-%d')}/{detection.camera}"
+
+        if not os.path.exists(file_dir):
+            file_dir = f"{CONST_MEDIA_DIR}/snapshots/{datetime.datetime.fromtimestamp(detection.end_time).strftime('%m-%d')}/{detection.camera}"
+
+        if not os.path.exists(file_dir):
+            return None
+
+        file = f"{file_dir}/{detection.id}.jpg"
+
+        with open(file, "rb") as image_file:
+            jpg_bytes = image_file.read()
+
+        return jpg_bytes
+
     def get_latest_camera_snapshot(
         self,
         camera_name: str,
     ) -> Any:
-        """Get the latest snapshot for <camera_name> and <zone_name>."""
+        """Get the latest web snapshot for <camera_name> and <zone_name>."""
         imgBytes = requests.get(
             self.config.cameras[camera_name].snapshot_config.url
         ).content
 
-        if not imgBytes:
+        if imgBytes is None:
             return None
 
         img = cv2.imdecode(np.asarray(bytearray(imgBytes), dtype=np.uint8), -1)
@@ -78,7 +123,7 @@ def get_latest_zone_snapshot(
         camera_name: str,
         zone_name: str,
     ) -> Any:
-        """Get the latest snapshot for <camera_name>."""
+        """Get the latest web snapshot for <camera_name>."""
         camera_config: CameraConfig = self.config.cameras[camera_name]
         imgBytes = requests.get(camera_config.snapshot_config.url).content
 
@@ -95,14 +140,12 @@ def get_latest_zone_snapshot(
         ret, jpg = cv2.imencode(".jpg", crop, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
         return jpg.tobytes()
 
-    def get_latest_detection(self, camera_name: str) -> Any:
-        """Get the latest detection for a <camera_name>."""
+    def get_latest_detection_snapshot(self, camera_name: str) -> Any:
+        """Get the latest file snapshot for a <camera_name> detection."""
         snaps_dir = f"{CONST_MEDIA_DIR}/snapshots"
-        print(f"snaps dir is {snaps_dir}")
         recent_folder = max(
             [os.path.join(snaps_dir, basename) for basename in os.listdir(snaps_dir)]
         )
-        print(f"")
 
         cam_snaps_dir = f"{recent_folder}/{camera_name}"
         recent_snap = max(