From 4c72a210a9670da5261a74a1720618bfce11ed26 Mon Sep 17 00:00:00 2001 From: Nicolas Mowen Date: Thu, 26 Mar 2026 10:58:24 -0600 Subject: [PATCH] Cleanup bird mypy --- .../post/audio_transcription.py | 17 ++++++------ frigate/data_processing/real_time/bird.py | 26 ++++++++++++------- 2 files changed, 25 insertions(+), 18 deletions(-) diff --git a/frigate/data_processing/post/audio_transcription.py b/frigate/data_processing/post/audio_transcription.py index 558ab433e..f761ee7c6 100644 --- a/frigate/data_processing/post/audio_transcription.py +++ b/frigate/data_processing/post/audio_transcription.py @@ -4,8 +4,9 @@ import logging import os import threading import time -from typing import Optional +from typing import Any, Optional +from embeddings.embeddings import Embeddings from peewee import DoesNotExist from frigate.comms.inter_process import InterProcessRequestor @@ -31,7 +32,7 @@ class AudioTranscriptionPostProcessor(PostProcessorApi): self, config: FrigateConfig, requestor: InterProcessRequestor, - embeddings, + embeddings: Embeddings, metrics: DataProcessorMetrics, ): super().__init__(config, metrics, None) @@ -40,7 +41,7 @@ class AudioTranscriptionPostProcessor(PostProcessorApi): self.embeddings = embeddings self.recognizer = None self.transcription_lock = threading.Lock() - self.transcription_thread = None + self.transcription_thread: threading.Thread | None = None self.transcription_running = False # faster-whisper handles model downloading automatically @@ -69,7 +70,7 @@ class AudioTranscriptionPostProcessor(PostProcessorApi): self.recognizer = None def process_data( - self, data: dict[str, any], data_type: PostProcessDataEnum + self, data: dict[str, Any], data_type: PostProcessDataEnum ) -> None: """Transcribe audio from a recording. @@ -141,13 +142,13 @@ class AudioTranscriptionPostProcessor(PostProcessorApi): except Exception as e: logger.error(f"Error in audio transcription post-processing: {e}") - def __transcribe_audio(self, audio_data: bytes) -> Optional[tuple[str, float]]: + def __transcribe_audio(self, audio_data: bytes) -> Optional[str]: """Transcribe WAV audio data using faster-whisper.""" if not self.recognizer: logger.debug("Recognizer not initialized") return None - try: + try: # type: ignore[unreachable] # Save audio data to a temporary wav (faster-whisper expects a file) temp_wav = os.path.join(CACHE_DIR, f"temp_audio_{int(time.time())}.wav") with open(temp_wav, "wb") as f: @@ -176,7 +177,7 @@ class AudioTranscriptionPostProcessor(PostProcessorApi): logger.error(f"Error transcribing audio: {e}") return None - def _transcription_wrapper(self, event: dict[str, any]) -> None: + def _transcription_wrapper(self, event: dict[str, Any]) -> None: """Wrapper to run transcription and reset running flag when done.""" try: self.process_data( @@ -194,7 +195,7 @@ class AudioTranscriptionPostProcessor(PostProcessorApi): self.requestor.send_data(UPDATE_AUDIO_TRANSCRIPTION_STATE, "idle") - def handle_request(self, topic: str, request_data: dict[str, any]) -> str | None: + def handle_request(self, topic: str, request_data: dict[str, Any]) -> str | None: if topic == "transcribe_audio": event = request_data["event"] diff --git a/frigate/data_processing/real_time/bird.py b/frigate/data_processing/real_time/bird.py index 38ff1a950..48663f971 100644 --- a/frigate/data_processing/real_time/bird.py +++ b/frigate/data_processing/real_time/bird.py @@ -14,7 +14,7 @@ from frigate.comms.event_metadata_updater import ( from frigate.config import FrigateConfig from frigate.const import MODEL_CACHE_DIR from frigate.log import suppress_stderr_during -from frigate.util.object import calculate_region +from frigate.util.image import calculate_region from ..types import DataProcessorMetrics from .api import RealTimeProcessorApi @@ -35,10 +35,10 @@ class BirdRealTimeProcessor(RealTimeProcessorApi): metrics: DataProcessorMetrics, ): super().__init__(config, metrics) - self.interpreter: Interpreter = None + self.interpreter: Interpreter | None = None self.sub_label_publisher = sub_label_publisher - self.tensor_input_details: dict[str, Any] = None - self.tensor_output_details: dict[str, Any] = None + self.tensor_input_details: list[dict[str, Any]] | None = None + self.tensor_output_details: list[dict[str, Any]] | None = None self.detected_birds: dict[str, float] = {} self.labelmap: dict[int, str] = {} @@ -61,7 +61,7 @@ class BirdRealTimeProcessor(RealTimeProcessorApi): self.downloader = ModelDownloader( model_name="bird", download_path=download_path, - file_names=self.model_files.keys(), + file_names=list(self.model_files.keys()), download_func=self.__download_models, complete_func=self.__build_detector, ) @@ -102,8 +102,12 @@ class BirdRealTimeProcessor(RealTimeProcessorApi): i += 1 line = f.readline() - def process_frame(self, obj_data, frame): - if not self.interpreter: + def process_frame(self, obj_data: dict[str, Any], frame: np.ndarray) -> None: + if ( + not self.interpreter + or not self.tensor_input_details + or not self.tensor_output_details + ): return if obj_data["label"] != "bird": @@ -145,7 +149,7 @@ class BirdRealTimeProcessor(RealTimeProcessorApi): self.tensor_output_details[0]["index"] )[0] probs = res / res.sum(axis=0) - best_id = np.argmax(probs) + best_id = int(np.argmax(probs)) if best_id == 964: logger.debug("No bird classification was detected.") @@ -179,9 +183,11 @@ class BirdRealTimeProcessor(RealTimeProcessorApi): self.config.classification = payload logger.debug("Bird classification config updated dynamically") - def handle_request(self, topic, request_data): + def handle_request( + self, topic: str, request_data: dict[str, Any] + ) -> dict[str, Any] | None: return None - def expire_object(self, object_id, camera): + def expire_object(self, object_id: str, camera: str) -> None: if object_id in self.detected_birds: self.detected_birds.pop(object_id)