ensure lpr events are always detections, typing fixes

frigate/camera/state.py

@@ -410,7 +410,11 @@ class CameraState:
             self.previous_frame_id = frame_name
 
     def save_manual_event_image(
-        self, frame: np.ndarray, event_id: str, label: str, draw: dict[str, list[dict]]
+        self,
+        frame: np.ndarray | None,
+        event_id: str,
+        label: str,
+        draw: dict[str, list[dict]],
     ) -> None:
         img_frame = frame if frame is not None else self.get_current_frame()
 

frigate/comms/detections_updater.py

@@ -11,6 +11,7 @@ class DetectionTypeEnum(str, Enum):
     api = "api"
     video = "video"
     audio = "audio"
+    lpr = "lpr"
 
 
 class DetectionPublisher(Publisher):

frigate/config/camera/camera.py

@@ -108,7 +108,7 @@ class CameraConfig(FrigateBaseModel):
     onvif: OnvifConfig = Field(
         default_factory=OnvifConfig, title="Camera Onvif Configuration."
     )
-    type: str = Field(default=CameraTypeEnum.generic, title="Camera Type")
+    type: CameraTypeEnum = Field(default=CameraTypeEnum.generic, title="Camera Type")
     ui: CameraUiConfig = Field(
         default_factory=CameraUiConfig, title="Camera UI Modifications."
     )

frigate/data_processing/common/license_plate/mixin.py

@@ -897,7 +897,7 @@ class LicensePlateProcessingMixin:
         return event_id
 
     def lpr_process(
-        self, obj_data: dict[str, any], frame: np.ndarray, dedicated_lpr: bool = None
+        self, obj_data: dict[str, any], frame: np.ndarray, dedicated_lpr: bool = False
     ):
         """Look for license plates in image."""
         camera = obj_data if dedicated_lpr else obj_data["camera"]

frigate/data_processing/real_time/license_plate.py

@@ -36,7 +36,10 @@ class LicensePlateRealTimeProcessor(LicensePlateProcessingMixin, RealTimeProcess
         super().__init__(config, metrics)
 
     def process_frame(
-        self, obj_data: dict[str, any], frame: np.ndarray, dedicated_lpr: bool = None
+        self,
+        obj_data: dict[str, any],
+        frame: np.ndarray,
+        dedicated_lpr: bool | None = False,
     ):
         """Look for license plates in image."""
         self.lpr_process(obj_data, frame, dedicated_lpr)

frigate/record/maintainer.py

@@ -577,7 +577,7 @@ class RecordingMaintainer(threading.Thread):
                                 audio_detections,
                             )
                         )
-                elif topic == DetectionTypeEnum.api:
+                elif topic == DetectionTypeEnum.api or DetectionTypeEnum.lpr:
                     continue
 
                 if frame_time < run_start - stale_frame_count_threshold:

frigate/review/maintainer.py

@@ -506,7 +506,7 @@ class ReviewSegmentMaintainer(threading.Thread):
                     _,
                     audio_detections,
                 ) = data
-            elif topic == DetectionTypeEnum.api:
+            elif topic == DetectionTypeEnum.api or DetectionTypeEnum.lpr:
                 (
                     camera,
                     frame_time,
@@ -565,13 +565,21 @@ class ReviewSegmentMaintainer(threading.Thread):
                             or audio in camera_config.review.detections.labels
                         ) and camera_config.review.detections.enabled:
                             current_segment.audio.add(audio)
-                elif topic == DetectionTypeEnum.api:
+                elif topic == DetectionTypeEnum.api or topic == DetectionTypeEnum.lpr:
                     if manual_info["state"] == ManualEventState.complete:
                         current_segment.detections[manual_info["event_id"]] = (
                             manual_info["label"]
                         )
-                        if self.config.cameras[camera].review.alerts.enabled:
+                        if (
+                            topic == DetectionTypeEnum.api
+                            and self.config.cameras[camera].review.alerts.enabled
+                        ):
                             current_segment.severity = SeverityEnum.alert
+                        elif (
+                            topic == DetectionTypeEnum.lpr
+                            and self.config.cameras[camera].review.detections.enabled
+                        ):
+                            current_segment.severity = SeverityEnum.detection
                         current_segment.last_update = manual_info["end_time"]
                     elif manual_info["state"] == ManualEventState.start:
                         self.indefinite_events[camera][manual_info["event_id"]] = (
@@ -580,8 +588,16 @@ class ReviewSegmentMaintainer(threading.Thread):
                         current_segment.detections[manual_info["event_id"]] = (
                             manual_info["label"]
                         )
-                        if self.config.cameras[camera].review.alerts.enabled:
+                        if (
+                            topic == DetectionTypeEnum.api
+                            and self.config.cameras[camera].review.alerts.enabled
+                        ):
                             current_segment.severity = SeverityEnum.alert
+                        elif (
+                            topic == DetectionTypeEnum.lpr
+                            and self.config.cameras[camera].review.detections.enabled
+                        ):
+                            current_segment.severity = SeverityEnum.detection
 
                         # temporarily make it so this event can not end
                         current_segment.last_update = sys.maxsize
@@ -669,6 +685,34 @@ class ReviewSegmentMaintainer(threading.Thread):
                         logger.warning(
                             f"Manual event API has been called for {camera}, but alerts are disabled. This manual event will not appear as an alert."
                         )
+                elif topic == DetectionTypeEnum.lpr:
+                    if self.config.cameras[camera].review.detections.enabled:
+                        self.active_review_segments[camera] = PendingReviewSegment(
+                            camera,
+                            frame_time,
+                            SeverityEnum.detection,
+                            {manual_info["event_id"]: manual_info["label"]},
+                            {},
+                            [],
+                            set(),
+                        )
+
+                        if manual_info["state"] == ManualEventState.start:
+                            self.indefinite_events[camera][manual_info["event_id"]] = (
+                                manual_info["label"]
+                            )
+                            # temporarily make it so this event can not end
+                            self.active_review_segments[
+                                camera
+                            ].last_update = sys.maxsize
+                        elif manual_info["state"] == ManualEventState.complete:
+                            self.active_review_segments[
+                                camera
+                            ].last_update = manual_info["end_time"]
+                    else:
+                        logger.warning(
+                            f"Dedicated LPR camera API has been called for {camera}, but detections are disabled. LPR events will not appear as a detection."
+                        )
 
         self.record_config_subscriber.stop()
         self.review_config_subscriber.stop()

frigate/track/object_processing.py

@@ -501,6 +501,21 @@ class TrackedObjectProcessor(threading.Thread):
             )
         )
 
+        self.ongoing_manual_events[event_id] = camera_name
+        self.detection_publisher.publish(
+            (
+                camera_name,
+                frame_time,
+                {
+                    "state": ManualEventState.start,
+                    "label": f"{label}: {sub_label}" if sub_label else label,
+                    "event_id": event_id,
+                    "end_time": None,
+                },
+            ),
+            DetectionTypeEnum.lpr.value,
+        )
+
     def end_manual_event(self, payload: tuple) -> None:
         (event_id, end_time) = payload