Fix camera access and improve typing

frigate/api/event.py

@@ -35,8 +35,9 @@
     CLIPS_DIR,
 )
 from frigate.embeddings import EmbeddingsContext
+from frigate.events.external import ExternalEventProcessor
 from frigate.models import Event, ReviewSegment, Timeline
-from frigate.object_processing import TrackedObject
+from frigate.object_processing import TrackedObject, TrackedObjectProcessor
 from frigate.util.builtin import get_tz_modifiers
 
 logger = logging.getLogger(__name__)
@@ -1087,9 +1088,11 @@ def create_event(
         )
 
     try:
-        frame = request.app.detected_frames_processor.get_current_frame(camera_name)
+        frame_processor: TrackedObjectProcessor = request.app.detected_frames_processor
+        external_processor: ExternalEventProcessor = request.app.external_processor
 
-        event_id = request.app.external_processor.create_manual_event(
+        frame = frame_processor.get_current_frame(camera_name)
+        event_id = external_processor.create_manual_event(
             camera_name,
             label,
             body.source_type,

frigate/api/media.py

@@ -36,6 +36,7 @@
     RECORD_DIR,
 )
 from frigate.models import Event, Previews, Recordings, Regions, ReviewSegment
+from frigate.object_processing import TrackedObjectProcessor
 from frigate.util.builtin import get_tz_modifiers
 from frigate.util.image import get_image_from_recording
 
@@ -79,7 +80,11 @@ def mjpeg_feed(
 
 
 def imagestream(
-    detected_frames_processor, camera_name: str, fps: int, height: int, draw_options
+    detected_frames_processor: TrackedObjectProcessor,
+    camera_name: str,
+    fps: int,
+    height: int,
+    draw_options: dict[str, any],
 ):
     while True:
         # max out at specified FPS
@@ -118,6 +123,7 @@ def latest_frame(
     extension: Extension,
     params: MediaLatestFrameQueryParams = Depends(),
 ):
+    frame_processor: TrackedObjectProcessor = request.app.detected_frames_processor
     draw_options = {
         "bounding_boxes": params.bbox,
         "timestamp": params.timestamp,
@@ -129,17 +135,14 @@ def latest_frame(
     quality = params.quality
 
     if camera_name in request.app.frigate_config.cameras:
-        frame = request.app.detected_frames_processor.get_current_frame(
-            camera_name, draw_options
-        )
+        frame = frame_processor.get_current_frame(camera_name, draw_options)
         retry_interval = float(
             request.app.frigate_config.cameras.get(camera_name).ffmpeg.retry_interval
             or 10
         )
 
         if frame is None or datetime.now().timestamp() > (
-            request.app.detected_frames_processor.get_current_frame_time(camera_name)
-            + retry_interval
+            frame_processor.get_current_frame_time(camera_name) + retry_interval
         ):
             if request.app.camera_error_image is None:
                 error_image = glob.glob("/opt/frigate/frigate/images/camera-error.jpg")
@@ -180,7 +183,7 @@ def latest_frame(
         )
     elif camera_name == "birdseye" and request.app.frigate_config.birdseye.restream:
         frame = cv2.cvtColor(
-            request.app.detected_frames_processor.get_current_frame(camera_name),
+            frame_processor.get_current_frame(camera_name),
             cv2.COLOR_YUV2BGR_I420,
         )
 
@@ -813,15 +816,15 @@ def grid_snapshot(
 ):
     if camera_name in request.app.frigate_config.cameras:
         detect = request.app.frigate_config.cameras[camera_name].detect
-        frame = request.app.detected_frames_processor.get_current_frame(camera_name, {})
+        frame_processor: TrackedObjectProcessor = request.app.detected_frames_processor
+        frame = frame_processor.get_current_frame(camera_name, {})
         retry_interval = float(
             request.app.frigate_config.cameras.get(camera_name).ffmpeg.retry_interval
             or 10
         )
 
         if frame is None or datetime.now().timestamp() > (
-            request.app.detected_frames_processor.get_current_frame_time(camera_name)
-            + retry_interval
+            frame_processor.get_current_frame_time(camera_name) + retry_interval
         ):
             return JSONResponse(
                 content={"success": False, "message": "Unable to get valid frame"},

frigate/config/config.py

@@ -230,12 +230,16 @@ def verify_recording_segments_setup_with_reasonable_time(
     try:
         seg_arg_index = record_args.index("-segment_time")
     except ValueError:
-        raise ValueError(f"Camera {camera_config.name} has no segment_time in \
-                         recording output args, segment args are required for record.")
+        raise ValueError(
+            f"Camera {camera_config.name} has no segment_time in \
+                         recording output args, segment args are required for record."
+        )
 
     if int(record_args[seg_arg_index + 1]) > 60:
-        raise ValueError(f"Camera {camera_config.name} has invalid segment_time output arg, \
-                         segment_time must be 60 or less.")
+        raise ValueError(
+            f"Camera {camera_config.name} has invalid segment_time output arg, \
+                         segment_time must be 60 or less."
+        )
 
 
 def verify_zone_objects_are_tracked(camera_config: CameraConfig) -> None:

frigate/events/external.py

@@ -10,6 +10,7 @@
 from typing import Optional
 
 import cv2
+from numpy import ndarray
 
 from frigate.comms.detections_updater import DetectionPublisher, DetectionTypeEnum
 from frigate.comms.events_updater import EventUpdatePublisher
@@ -45,7 +46,7 @@ def create_manual_event(
         duration: Optional[int],
         include_recording: bool,
         draw: dict[str, any],
-        snapshot_frame: any,
+        snapshot_frame: Optional[ndarray],
     ) -> str:
         now = datetime.datetime.now().timestamp()
         camera_config = self.config.cameras.get(camera)
@@ -131,8 +132,11 @@ def _write_images(
         label: str,
         event_id: str,
         draw: dict[str, any],
-        img_frame: any,
-    ) -> str:
+        img_frame: Optional[ndarray],
+    ) -> Optional[str]:
+        if not img_frame:
+            return None
+
         # write clean snapshot if enabled
         if camera_config.snapshots.clean_copy:
             ret, png = cv2.imencode(".png", img_frame)

frigate/object_processing.py

@@ -6,7 +6,7 @@
 import threading
 from collections import Counter, defaultdict
 from multiprocessing.synchronize import Event as MpEvent
-from typing import Callable
+from typing import Callable, Optional
 
 import cv2
 import numpy as np
@@ -784,13 +784,18 @@ def get_best(self, camera, label):
         else:
             return {}
 
-    def get_current_frame(self, camera, draw_options={}):
+    def get_current_frame(
+        self, camera: str, draw_options: dict[str, any] = {}
+    ) -> Optional[np.ndarray]:
         if camera == "birdseye":
             return self.frame_manager.get(
                 "birdseye",
                 (self.config.birdseye.height * 3 // 2, self.config.birdseye.width),
             )
 
+        if camera not in self.camera_states:
+            return None
+
         return self.camera_states[camera].get_current_frame(draw_options)
 
     def get_current_frame_time(self, camera) -> int: