Improve typing

frigate/object_detection/base.py

@@ -1,11 +1,12 @@
 import datetime
 import logging
-import multiprocessing as mp
 import os
 import queue
 import signal
 import threading
 from abc import ABC, abstractmethod
+from multiprocessing import Queue, Value
+from multiprocessing.synchronize import Event as MpEvent
 
 import numpy as np
 from setproctitle import setproctitle
@@ -15,6 +16,7 @@ from frigate.detectors import create_detector
 from frigate.detectors.detector_config import (
     BaseDetectorConfig,
     InputDTypeEnum,
+    ModelConfig,
 )
 from frigate.util.builtin import EventsPerSecond, load_labels
 from frigate.util.image import SharedMemoryFrameManager, UntrackedSharedMemory
@@ -85,11 +87,11 @@ class LocalObjectDetector(ObjectDetector):
 
 def run_detector(
     name: str,
-    detection_queue: mp.Queue,
-    out_events: dict[str, mp.Event],
-    avg_speed,
-    start,
-    detector_config,
+    detection_queue: Queue,
+    out_events: dict[str, MpEvent],
+    avg_speed: Value,
+    start: Value,
+    detector_config: BaseDetectorConfig,
 ):
     threading.current_thread().name = f"detector:{name}"
     logger = logging.getLogger(f"detector.{name}")
@@ -97,7 +99,7 @@ def run_detector(
     setproctitle(f"frigate.detector.{name}")
     listen()
 
-    stop_event = mp.Event()
+    stop_event = MpEvent()
 
     def receiveSignal(signalNumber, frame):
         stop_event.set()
@@ -145,16 +147,16 @@ def run_detector(
 class ObjectDetectProcess:
     def __init__(
         self,
-        name,
-        detection_queue,
-        out_events,
-        detector_config,
+        name: str,
+        detection_queue: Queue,
+        out_events: dict[str, MpEvent],
+        detector_config: BaseDetectorConfig,
     ):
         self.name = name
         self.out_events = out_events
         self.detection_queue = detection_queue
-        self.avg_inference_speed = mp.Value("d", 0.01)
-        self.detection_start = mp.Value("d", 0.0)
+        self.avg_inference_speed = Value("d", 0.01)
+        self.detection_start = Value("d", 0.0)
         self.detect_process = None
         self.detector_config = detector_config
         self.start_or_restart()
@@ -193,7 +195,15 @@ class ObjectDetectProcess:
 
 
 class RemoteObjectDetector:
-    def __init__(self, name, labels, detection_queue, event, model_config, stop_event):
+    def __init__(
+        self,
+        name: str,
+        labels: dict[int, str],
+        detection_queue: Queue,
+        event: MpEvent,
+        model_config: ModelConfig,
+        stop_event: MpEvent,
+    ):
         self.labels = labels
         self.name = name
         self.fps = EventsPerSecond()

frigate/video.py

@@ -1,12 +1,14 @@
 import datetime
 import logging
-import multiprocessing as mp
 import os
 import queue
 import signal
 import subprocess as sp
 import threading
 import time
+from multiprocessing import Queue, Value
+from multiprocessing.synchronize import Event as MpEvent
+from typing import Any
 
 import cv2
 from setproctitle import setproctitle
@@ -99,10 +101,10 @@ def capture_frames(
     frame_shape: tuple[int, int],
     frame_manager: FrameManager,
     frame_queue,
-    fps: mp.Value,
-    skipped_fps: mp.Value,
-    current_frame: mp.Value,
-    stop_event: mp.Event,
+    fps: Value,
+    skipped_fps: Value,
+    current_frame: Value,
+    stop_event: MpEvent,
 ):
     frame_size = frame_shape[0] * frame_shape[1]
     frame_rate = EventsPerSecond()
@@ -167,7 +169,7 @@ class CameraWatchdog(threading.Thread):
         camera_name,
         config: CameraConfig,
         shm_frame_count: int,
-        frame_queue: mp.Queue,
+        frame_queue: Queue,
         camera_fps,
         skipped_fps,
         ffmpeg_pid,
@@ -402,10 +404,10 @@ class CameraCapture(threading.Thread):
         frame_index: int,
         ffmpeg_process,
         frame_shape: tuple[int, int],
-        frame_queue: mp.Queue,
-        fps,
-        skipped_fps,
-        stop_event,
+        frame_queue: Queue,
+        fps: Value,
+        skipped_fps: Value,
+        stop_event: MpEvent,
     ):
         threading.Thread.__init__(self)
         self.name = f"capture:{config.name}"
@@ -419,7 +421,7 @@ class CameraCapture(threading.Thread):
         self.skipped_fps = skipped_fps
         self.frame_manager = SharedMemoryFrameManager()
         self.ffmpeg_process = ffmpeg_process
-        self.current_frame = mp.Value("d", 0.0)
+        self.current_frame = Value("d", 0.0)
         self.last_frame = 0
 
     def run(self):
@@ -441,7 +443,7 @@ class CameraCapture(threading.Thread):
 def capture_camera(
     name, config: CameraConfig, shm_frame_count: int, camera_metrics: CameraMetrics
 ):
-    stop_event = mp.Event()
+    stop_event = MpEvent()
 
     def receiveSignal(signalNumber, frame):
         stop_event.set()
@@ -469,16 +471,16 @@ def capture_camera(
 def track_camera(
     name,
     config: CameraConfig,
-    model_config,
-    labelmap,
-    detection_queue,
-    result_connection,
+    model_config: ModelConfig,
+    labelmap: dict[int, str],
+    detection_queue: Queue,
+    result_connection: MpEvent,
     detected_objects_queue,
     camera_metrics: CameraMetrics,
     ptz_metrics: PTZMetrics,
-    region_grid,
+    region_grid: list[list[dict[str, Any]]],
 ):
-    stop_event = mp.Event()
+    stop_event = MpEvent()
 
     def receiveSignal(signalNumber, frame):
         stop_event.set()
@@ -584,8 +586,8 @@ def detect(
 def process_frames(
     camera_name: str,
     requestor: InterProcessRequestor,
-    frame_queue: mp.Queue,
-    frame_shape,
+    frame_queue: Queue,
+    frame_shape: tuple[int, int],
     model_config: ModelConfig,
     camera_config: CameraConfig,
     detect_config: DetectConfig,
@@ -593,13 +595,13 @@ def process_frames(
     motion_detector: MotionDetector,
     object_detector: RemoteObjectDetector,
     object_tracker: ObjectTracker,
-    detected_objects_queue: mp.Queue,
+    detected_objects_queue: Queue,
     camera_metrics: CameraMetrics,
     objects_to_track: list[str],
     object_filters,
-    stop_event,
+    stop_event: MpEvent,
     ptz_metrics: PTZMetrics,
-    region_grid,
+    region_grid: list[list[dict[str, Any]]],
     exit_on_empty: bool = False,
 ):
     next_region_update = get_tomorrow_at_time(2)