Mirrors/blakeblackshear.frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2024-11-21 19:07:46 +01:00
Code Issues Projects Releases Wiki Activity

Modernizing Typing

Browse Source 
All Dict, List were converted to dict, list, see: https://mypy.readthedocs.io/en/stable/builtin_types.html#generic-types
This commit is contained in:
Sebastian Englbrecht 2022-04-16 17:38:07 +02:00 committed by Blake Blackshear
parent a1afade9ba
commit ebf4e43ced
4 changed files with 9 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
frigate/edgetpu.py
View File

@ -6,7 +6,6 @@ import queue
import signal import signal
import threading import threading
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import Dict
import numpy as np import numpy as np
import tflite_runtime.interpreter as tflite import tflite_runtime.interpreter as tflite
@ -110,7 +109,7 @@ class LocalObjectDetector(ObjectDetector):
def run_detector( def run_detector(
name: str, name: str,
detection_queue: mp.Queue, detection_queue: mp.Queue,
out_events: Dict[str, mp.Event], out_events: dict[str, mp.Event],
avg_speed, avg_speed,
start, start,
model_path, model_path,

8
frigate/object_processing.py
View File

@ -11,7 +11,7 @@ import threading
import time import time
from collections import Counter, defaultdict from collections import Counter, defaultdict
from statistics import mean, median from statistics import mean, median
from typing import Callable, Dict from typing import Callable
import cv2 import cv2
import numpy as np import numpy as np
@ -362,9 +362,9 @@ class CameraState:
self.config = config self.config = config
self.camera_config = config.cameras[name] self.camera_config = config.cameras[name]
self.frame_manager = frame_manager self.frame_manager = frame_manager
self.best_objects: Dict[str, TrackedObject] = {} self.best_objects: dict[str, TrackedObject] = {}
self.object_counts = defaultdict(int) self.object_counts = defaultdict(int)
self.tracked_objects: Dict[str, TrackedObject] = {} self.tracked_objects: dict[str, TrackedObject] = {}
self.frame_cache = {} self.frame_cache = {}
self.zone_objects = defaultdict(list) self.zone_objects = defaultdict(list)
self._current_frame = np.zeros(self.camera_config.frame_shape_yuv, np.uint8) self._current_frame = np.zeros(self.camera_config.frame_shape_yuv, np.uint8)
@ -650,7 +650,7 @@ class TrackedObjectProcessor(threading.Thread):
self.video_output_queue = video_output_queue self.video_output_queue = video_output_queue
self.recordings_info_queue = recordings_info_queue self.recordings_info_queue = recordings_info_queue
self.stop_event = stop_event self.stop_event = stop_event
self.camera_states: Dict[str, CameraState] = {} self.camera_states: dict[str, CameraState] = {}
self.frame_manager = SharedMemoryFrameManager() self.frame_manager = SharedMemoryFrameManager()
def start(camera, obj: TrackedObject, current_frame_time): def start(camera, obj: TrackedObject, current_frame_time):

5
frigate/video.py
View File

@ -9,7 +9,6 @@ import subprocess as sp
import threading import threading
import time import time
from collections import defaultdict from collections import defaultdict
from typing import Dict, List
import numpy as np import numpy as np
from cv2 import cv2, reduce from cv2 import cv2, reduce
@ -476,8 +475,8 @@ def process_frames(
object_detector: RemoteObjectDetector, object_detector: RemoteObjectDetector,
object_tracker: ObjectTracker, object_tracker: ObjectTracker,
detected_objects_queue: mp.Queue, detected_objects_queue: mp.Queue,
process_info: Dict, process_info: dict,
objects_to_track: List[str], objects_to_track: list[str],
object_filters, object_filters,
detection_enabled: mp.Value, detection_enabled: mp.Value,
stop_event, stop_event,

4
frigate/watchdog.py
View File

@ -8,13 +8,13 @@ import signal
from frigate.edgetpu import EdgeTPUProcess from frigate.edgetpu import EdgeTPUProcess
from frigate.util import restart_frigate from frigate.util import restart_frigate
from multiprocessing.synchronize import Event from multiprocessing.synchronize import Event
from typing import Dict from typing import dict
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class FrigateWatchdog(threading.Thread): class FrigateWatchdog(threading.Thread):
def __init__(self, detectors: Dict[str, EdgeTPUProcess], stop_event: Event): def __init__(self, detectors: dict[str, EdgeTPUProcess], stop_event: Event):
threading.Thread.__init__(self) threading.Thread.__init__(self)
self.name = "frigate_watchdog" self.name = "frigate_watchdog"
self.detectors = detectors self.detectors = detectors