Increase ruff coverage (#23644)

* Pin ruff

* Add python upgrade fixes

This enables python upgrade checks in ruff to look for deprecated types and patterns. This namely fixes:
- usage of deprecated `Typing` which is now built in
- some specific exceptions which are caught and have new aliases

Some specific UP checks were also ignored as they are stylistic / unimportant and likely to cause bugs

* Remove async blocking calls

Use asyncio.to_thread on two remaining blocking calls to fix hanging event thread loop. Enable this specific rule to block it in the future.

* Use proper logging mechanism

* Correctly format logs

* Raise with context

When raising an exception include the from context to improve debugging

* Cleanup

frigate/util/audio.py

@@ -3,7 +3,6 @@
 import logging
 import os
 import subprocess as sp
-from typing import Optional
 
 from pathvalidate import sanitize_filename
 
@@ -19,7 +18,7 @@ def get_audio_from_recording(
     start_ts: float,
     end_ts: float,
     sample_rate: int = 16000,
-) -> Optional[bytes]:
+) -> bytes | None:
     """Extract audio from recording files between start_ts and end_ts in WAV format suitable for sherpa-onnx.
 
     Args:

frigate/util/builtin.py

@@ -15,7 +15,7 @@ from collections import deque
 from collections.abc import Mapping
 from multiprocessing.managers import ValueProxy
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any
 
 import numpy as np
 from ruamel.yaml import YAML
@@ -152,7 +152,7 @@ def get_record_segment_time(config: "CameraConfig") -> int:
 
 
 def load_labels(
-    path: Optional[str], encoding="utf-8", prefill=91, indexed: bool | None = None
+    path: str | None, encoding="utf-8", prefill=91, indexed: bool | None = None
 ):
     """Loads labels from file (with or without index numbers).
     Args:
@@ -164,7 +164,7 @@ def load_labels(
     if path is None:
         return {}
 
-    with open(path, "r", encoding=encoding) as f:
+    with open(path, encoding=encoding) as f:
         labels = {index: "unknown" for index in range(prefill)}
         lines = f.readlines()
         if not lines:
@@ -180,8 +180,8 @@ def load_labels(
 
 
 def to_relative_box(
-    width: int, height: int, box: Tuple[int, int, int, int]
-) -> Tuple[int | float, int | float, int | float, int | float]:
+    width: int, height: int, box: tuple[int, int, int, int]
+) -> tuple[int | float, int | float, int | float, int | float]:
     return (
         box[0] / width,  # x
         box[1] / height,  # y
@@ -195,7 +195,7 @@ def create_mask(frame_shape, mask):
     mask_img[:] = 255
 
 
-def process_config_query_string(query_string: Dict[str, list]) -> Dict[str, Any]:
+def process_config_query_string(query_string: dict[str, list]) -> dict[str, Any]:
     updates = {}
     for key_path_str, new_value_list in query_string.items():
         # use the string key as-is for updates dictionary
@@ -213,8 +213,8 @@ def process_config_query_string(query_string: Dict[str, list]) -> Dict[str, Any]
 
 
 def flatten_config_data(
-    config_data: Dict[str, Any], parent_key: str = ""
-) -> Dict[str, Any]:
+    config_data: dict[str, Any], parent_key: str = ""
+) -> dict[str, Any]:
     items = []
     for key, value in config_data.items():
         escaped_key = escape_config_key_segment(str(key))
@@ -261,12 +261,12 @@ def split_config_key_path(key_path_str: str) -> list[str]:
     return parts
 
 
-def update_yaml_file_bulk(file_path: str, updates: Dict[str, Any]):
+def update_yaml_file_bulk(file_path: str, updates: dict[str, Any]):
     yaml = YAML()
     yaml.indent(mapping=2, sequence=4, offset=2)
 
     try:
-        with open(file_path, "r") as f:
+        with open(file_path) as f:
             data = yaml.load(f)
     except FileNotFoundError:
         logger.error(
@@ -438,9 +438,7 @@ def generate_color_palette(n):
     return colors
 
 
-def serialize(
-    vector: Union[list[float], np.ndarray, float], pack: bool = True
-) -> bytes:
+def serialize(vector: list[float] | np.ndarray | float, pack: bool = True) -> bytes:
     """Serializes a list of floats, numpy array, or single float into a compact "raw bytes" format"""
     if isinstance(vector, np.ndarray):
         # Convert numpy array to list of floats
@@ -459,7 +457,7 @@ def serialize(
         else:
             return vector
     except struct.error as e:
-        raise ValueError(f"Failed to pack vector: {e}. Vector: {vector}")
+        raise ValueError(f"Failed to pack vector: {e}. Vector: {vector}") from e
 
 
 def deserialize(bytes_data: bytes) -> list[float]:

frigate/util/classification.py

@@ -84,7 +84,7 @@ def read_training_metadata(model_name: str) -> dict[str, any] | None:
         return None
 
     try:
-        with open(metadata_path, "r") as f:
+        with open(metadata_path) as f:
             metadata = json.load(f)
         return metadata
     except Exception as e:
@@ -294,7 +294,7 @@ class ClassificationTrainingProcess(FrigateProcess):
             return True
 
         except Exception as e:
-            logger.error(f"Training failed for {self.model_name}: {e}", exc_info=True)
+            logger.exception(f"Training failed for {self.model_name}: {e}")
             return False
 
 
@@ -732,7 +732,7 @@ def collect_object_classification_examples(
 
     # Step 1: Query events for the specified label and cameras
     events = list(
-        Event.select().where((Event.label == label)).order_by(Event.start_time.asc())
+        Event.select().where(Event.label == label).order_by(Event.start_time.asc())
     )
 
     if not events:

frigate/util/config.py

@@ -4,7 +4,7 @@ import asyncio
 import logging
 import os
 import shutil
-from typing import Any, Optional, Union
+from typing import Any
 
 from ruamel.yaml import YAML
 
@@ -78,7 +78,7 @@ def migrate_frigate_config(config_file: str):
 
     yaml = YAML()
     yaml.indent(mapping=2, sequence=4, offset=2)
-    with open(config_file, "r") as f:
+    with open(config_file) as f:
         config: dict[str, dict[str, Any]] = yaml.load(f)
 
     if config is None:
@@ -477,7 +477,7 @@ def migrate_017_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]
 
 
 def _convert_legacy_mask_to_dict(
-    mask: Optional[Union[str, list]], mask_type: str = "motion_mask", label: str = ""
+    mask: str | list | None, mask_type: str = "motion_mask", label: str = ""
 ) -> dict[str, dict[str, Any]]:
     """Convert legacy mask format (str or list[str]) to new dict format.
 
@@ -659,10 +659,10 @@ def migrate_018_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]
 
 
 def get_relative_coordinates(
-    mask: Optional[Union[str, list]],
+    mask: str | list | None,
     frame_shape: tuple[int, int],
     camera_name: str = "",
-) -> Union[str, list]:
+) -> str | list:
     # masks and zones are saved as relative coordinates
     # we know if any points are > 1 then it is using the
     # old native resolution coordinates
@@ -720,7 +720,7 @@ def get_relative_coordinates(
 
 
 def convert_area_to_pixels(
-    area_value: Union[int, float], frame_shape: tuple[int, int]
+    area_value: int | float, frame_shape: tuple[int, int]
 ) -> int:
     """
     Convert area specification to pixels.
@@ -762,7 +762,7 @@ class StreamInfoRetriever:
         return info
 
 
-def apply_section_update(camera_config, section: str, update: dict) -> Optional[str]:
+def apply_section_update(camera_config, section: str, update: dict) -> str | None:
     """Merge an update dict into a camera config section and rebuild runtime variants.
 
     For motion and object filter sections, the plain Pydantic models are rebuilt

frigate/util/downloader.py

@@ -1,8 +1,8 @@
 import logging
 import os
 import threading
+from collections.abc import Callable
 from pathlib import Path
-from typing import Callable, List
 
 import requests
 
@@ -19,7 +19,7 @@ class ModelDownloader:
         self,
         model_name: str,
         download_path: str,
-        file_names: List[str],
+        file_names: list[str],
         download_func: Callable[[str], None],
         complete_func: Callable[[], None] | None = None,
         silent: bool = False,

frigate/util/ffmpeg.py

@@ -2,7 +2,8 @@
 
 import logging
 import subprocess as sp
-from typing import Any, Callable, Optional
+from collections.abc import Callable
+from typing import Any
 
 from frigate.const import PROCESS_PRIORITY_LOW
 from frigate.log import LogPipe
@@ -68,9 +69,9 @@ def run_ffmpeg_with_progress(
     cmd: list[str],
     *,
     expected_duration_seconds: float,
-    on_progress: Optional[Callable[[float], None]] = None,
-    stdin_payload: Optional[str] = None,
-    process_started: Optional[Callable[[sp.Popen], None]] = None,
+    on_progress: Callable[[float], None] | None = None,
+    stdin_payload: str | None = None,
+    process_started: Callable[[sp.Popen], None] | None = None,
     use_low_priority: bool = True,
 ) -> tuple[int, str]:
     """Run an ffmpeg command, streaming progress via `-progress pipe:2`.

frigate/util/file.py

@@ -7,7 +7,7 @@ import os
 import time
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any
 
 import cv2
 from numpy import ndarray
@@ -323,7 +323,7 @@ class FileLock:
         self.timeout = timeout
         self.poll_interval = poll_interval
         self.stale_timeout = stale_timeout
-        self._fd: Optional[int] = None
+        self._fd: int | None = None
         self._acquired = False
 
         if cleanup_stale_on_init:
@@ -367,7 +367,7 @@ class FileLock:
 
         return False
 
-    def acquire(self, timeout: Optional[int] = None) -> bool:
+    def acquire(self, timeout: int | None = None) -> bool:
         """
         Acquire the file lock using fcntl.flock().
 
@@ -400,7 +400,7 @@ class FileLock:
                     self._acquired = True
                     logger.debug(f"Acquired lock: {self.lock_path}")
                     return True
-                except (OSError, IOError):
+                except OSError:
                     # Lock is held by another process
                     if time.time() - start_time >= timeout:
                         logger.warning(f"Timeout waiting for lock: {self.lock_path}")

frigate/util/image.py

@@ -8,7 +8,7 @@ from abc import ABC, abstractmethod
 from multiprocessing import resource_tracker as _mprt
 from multiprocessing import shared_memory as _mpshm
 from string import printable
-from typing import Any, AnyStr, Optional
+from typing import Any, AnyStr
 
 import cv2
 import numpy as np
@@ -270,7 +270,7 @@ def draw_box_with_label(
     )
 
 
-def get_image_quality_params(ext: str, quality: Optional[int]) -> list[int]:
+def get_image_quality_params(ext: str, quality: int | None) -> list[int]:
     if ext in ("jpg", "jpeg"):
         return [int(cv2.IMWRITE_JPEG_QUALITY), quality if quality is not None else 70]
 
@@ -921,7 +921,7 @@ def yuv_region_2_bgr(frame, region):
         raise
 
 
-def intersection(box_a, box_b) -> Optional[list[int]]:
+def intersection(box_a, box_b) -> list[int] | None:
     """Return intersection box or None if boxes do not intersect."""
     if (
         box_a[2] < box_b[0]
@@ -994,7 +994,7 @@ class FrameManager(ABC):
         pass
 
     @abstractmethod
-    def write(self, name: str) -> Optional[memoryview]:
+    def write(self, name: str) -> memoryview | None:
         pass
 
     @abstractmethod
@@ -1021,7 +1021,7 @@ class UntrackedSharedMemory(_mpshm.SharedMemory):
 
     def __init__(
         self,
-        name: Optional[str] = None,
+        name: str | None = None,
         create: bool = False,
         size: int = 0,
         *,
@@ -1075,7 +1075,7 @@ class SharedMemoryFrameManager(FrameManager):
         self.shm_store[name] = shm
         return shm.buf
 
-    def write(self, name: str) -> Optional[memoryview]:
+    def write(self, name: str) -> memoryview | None:
         try:
             if name in self.shm_store:
                 shm = self.shm_store[name]
@@ -1087,7 +1087,7 @@ class SharedMemoryFrameManager(FrameManager):
             logger.info(f"the file {name} not found")
             return None
 
-    def get(self, name: str, shape) -> Optional[np.ndarray]:
+    def get(self, name: str, shape) -> np.ndarray | None:
         try:
             required = int(np.prod(shape))
             shm = self.shm_store.get(name)
@@ -1185,10 +1185,10 @@ def run_ffmpeg_snapshot(
     ffmpeg,
     input_path: str,
     codec: str,
-    seek_time: Optional[float] = None,
-    height: Optional[int] = None,
-    timeout: Optional[int] = None,
-) -> tuple[Optional[bytes], str]:
+    seek_time: float | None = None,
+    height: int | None = None,
+    timeout: int | None = None,
+) -> tuple[bytes | None, str]:
     """Run ffmpeg to extract a snapshot/image from a video source."""
     ffmpeg_cmd = [
         ffmpeg.ffmpeg_path,
@@ -1238,8 +1238,8 @@ def get_image_from_recording(
     file_path: str,
     relative_frame_time: float,
     codec: str,
-    height: Optional[int] = None,
-) -> Optional[Any]:
+    height: int | None = None,
+) -> Any | None:
     """retrieve a frame from given time in recording file."""
 
     image_data, _ = run_ffmpeg_snapshot(
@@ -1261,7 +1261,7 @@ def get_histogram(image, x_min, y_min, x_max, y_max):
 
 def create_thumbnail(
     yuv_frame: np.ndarray, box: tuple[int, int, int, int], height=500
-) -> Optional[bytes]:
+) -> bytes | None:
     """Return jpg thumbnail of a region of the frame."""
     frame = cv2.cvtColor(yuv_frame, cv2.COLOR_YUV2BGR_I420)
     region = calculate_region(

frigate/util/media.py

@@ -5,9 +5,9 @@ import errno
 import logging
 import os
 import subprocess as sp
+from collections.abc import Iterable
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Iterable
 
 from peewee import DatabaseError, chunked
 
@@ -110,7 +110,7 @@ def sync_recordings(
         # start checking on the hour 36 hours ago
         check_point = datetime.datetime.now().replace(
             minute=0, second=0, microsecond=0
-        ).astimezone(datetime.timezone.utc) - datetime.timedelta(hours=36)
+        ).astimezone(datetime.UTC) - datetime.timedelta(hours=36)
 
         # Gather DB recordings to inspect
         if limited:
@@ -793,7 +793,7 @@ def write_orphan_report(
             f.write("# Media Sync Orphan Report\n")
             f.write(f"# Job: {job_id}\n")
             f.write(
-                f"# Date: {datetime.datetime.now().astimezone(datetime.timezone.utc).isoformat()}\n"
+                f"# Date: {datetime.datetime.now().astimezone(datetime.UTC).isoformat()}\n"
             )
             f.write(f"# Mode: dry_run={dry_run}\n\n")
 

frigate/util/process.py

@@ -6,9 +6,9 @@ import os
 import pathlib
 import subprocess
 import threading
+from collections.abc import Callable
 from logging.handlers import QueueHandler
 from multiprocessing.synchronize import Event as MpEvent
-from typing import Callable, Optional
 
 from setproctitle import setproctitle
 
@@ -23,11 +23,11 @@ class BaseProcess(mp.Process):
         stop_event: MpEvent,
         priority: int,
         *,
-        name: Optional[str] = None,
-        target: Optional[Callable] = None,
+        name: str | None = None,
+        target: Callable | None = None,
         args: tuple = (),
         kwargs: dict = {},
-        daemon: Optional[bool] = None,
+        daemon: bool | None = None,
     ):
         self.priority = priority
         self.stop_event = stop_event
@@ -121,7 +121,7 @@ class FrigateProcess(BaseProcess):
                 f"If process crashes, manually generate with: memray flamegraph {binary_file}"
             )
         except Exception as e:
-            self.logger.error(f"Failed to setup memray profiling: {e}", exc_info=True)
+            self.logger.exception(f"Failed to setup memray profiling: {e}")
 
     def _cleanup_memray(self, safe_name: str, binary_file: pathlib.Path) -> None:
         """Stop memray tracking and generate HTML report."""
@@ -156,4 +156,4 @@ class FrigateProcess(BaseProcess):
         except subprocess.TimeoutExpired:
             self.logger.error("Memray report generation timed out")
         except Exception as e:
-            self.logger.error(f"Failed to cleanup memray profiling: {e}", exc_info=True)
+            self.logger.exception(f"Failed to cleanup memray profiling: {e}")

frigate/util/rknn_converter.py

@@ -6,7 +6,6 @@ import subprocess
 import sys
 import time
 from pathlib import Path
-from typing import Optional
 
 from frigate.const import SUPPORTED_RK_SOCS
 from frigate.util.file import FileLock
@@ -139,7 +138,7 @@ def ensure_rknn_toolkit() -> bool:
         return False
 
 
-def get_soc_type() -> Optional[str]:
+def get_soc_type() -> str | None:
     """Get the SoC type from device tree."""
     try:
         with open("/proc/device-tree/compatible") as file:
@@ -160,7 +159,7 @@ def convert_onnx_to_rknn(
     output_path: str,
     model_type: str,
     quantization: bool = False,
-    soc: Optional[str] = None,
+    soc: str | None = None,
 ) -> bool:
     """
     Convert ONNX model to RKNN format.
@@ -345,7 +344,7 @@ def wait_for_conversion_completion(
 
 def auto_convert_model(
     model_path: str, model_type: str | None = None, quantization: bool = False
-) -> Optional[str]:
+) -> str | None:
     """
     Automatically convert a model to RKNN format if needed.
 

frigate/util/schema.py

@@ -1,11 +1,11 @@
 """JSON schema utilities for Frigate."""
 
-from typing import Any, Dict, Type
+from typing import Any
 
 from pydantic import BaseModel, TypeAdapter
 
 
-def get_config_schema(config_class: Type[BaseModel]) -> Dict[str, Any]:
+def get_config_schema(config_class: type[BaseModel]) -> dict[str, Any]:
     """
     Returns the JSON schema for FrigateConfig with polymorphic detectors.
 

frigate/util/services.py

@@ -12,7 +12,7 @@ import subprocess as sp
 import time
 import traceback
 from datetime import datetime
-from typing import Any, List, Optional, Tuple
+from typing import Any
 
 import cv2
 import psutil
@@ -59,7 +59,7 @@ def get_cgroups_version() -> str:
         return "unknown"
 
     try:
-        with open("/proc/mounts", "r") as f:
+        with open("/proc/mounts") as f:
             mounts = f.readlines()
 
         for mount in mounts:
@@ -89,7 +89,7 @@ def get_docker_memlimit_bytes() -> int:
         memlimit_path = "/sys/fs/cgroup/memory.max"
 
         try:
-            with open(memlimit_path, "r") as f:
+            with open(memlimit_path) as f:
                 value = f.read().strip()
 
             if value.isnumeric():
@@ -127,7 +127,7 @@ def get_cpu_stats() -> dict[str, dict]:
             if not any(keyword in cmdline for keyword in keywords):
                 continue
 
-            with open(f"/proc/{pid}/stat", "r") as f:
+            with open(f"/proc/{pid}/stat") as f:
                 stats = f.readline().split()
             utime = int(stats[13])
             stime = int(stats[14])
@@ -146,7 +146,7 @@ def get_cpu_stats() -> dict[str, dict]:
             process_usage_sec = process_utime_sec + process_stime_sec
             cpu_average_usage = process_usage_sec * 100 // process_elapsed_sec
 
-            with open(f"/proc/{pid}/statm", "r") as f:
+            with open(f"/proc/{pid}/statm") as f:
                 mem_stats = f.readline().split()
             mem_res = int(mem_stats[1]) * os.sysconf("SC_PAGE_SIZE") / 1024
 
@@ -171,7 +171,7 @@ def get_physical_interfaces(interfaces) -> list:
     if not interfaces:
         return []
 
-    with open("/proc/net/dev", "r") as file:
+    with open("/proc/net/dev") as file:
         lines = file.readlines()
 
     physical_interfaces = []
@@ -238,7 +238,7 @@ def is_vaapi_amd_driver() -> bool:
         return any("AMD Radeon Graphics" in line for line in output)
 
 
-def get_amd_gpu_stats() -> Optional[dict[str, str]]:
+def get_amd_gpu_stats() -> dict[str, str] | None:
     """Get stats using radeontop."""
     radeontop_command = ["radeontop", "-d", "-", "-l", "1"]
 
@@ -287,7 +287,7 @@ _XE_ENGINE_KEYS = {
 }
 
 
-def _resolve_intel_gpu_pdev(device: Optional[str]) -> Optional[str]:
+def _resolve_intel_gpu_pdev(device: str | None) -> str | None:
     """Map a configured GPU hint (/dev/dri/card1, renderD128, or a PCI bus
     address) to its drm-pdev string so we can filter fdinfo entries to that
     device. Returns None when no hint is supplied or it cannot be resolved."""
@@ -304,7 +304,7 @@ def _resolve_intel_gpu_pdev(device: Optional[str]) -> Optional[str]:
         return None
 
 
-def _read_intel_drm_fdinfo(target_pdev: Optional[str]) -> dict:
+def _read_intel_drm_fdinfo(target_pdev: str | None) -> dict:
     """Snapshot DRM fdinfo for every Intel client visible in /proc.
 
     Returns a dict keyed by (pdev, drm-client-id, pid) so the same context
@@ -394,8 +394,8 @@ def _read_intel_drm_fdinfo(target_pdev: Optional[str]) -> dict:
 
 
 def get_intel_gpu_stats(
-    intel_gpu_device: Optional[str],
-) -> Optional[dict[str, dict[str, Any]]]:
+    intel_gpu_device: str | None,
+) -> dict[str, dict[str, Any]] | None:
     """Get stats by reading DRM fdinfo files, bucketed per-pdev.
 
     Each DRM client FD exposes monotonic per-engine busy counters via
@@ -509,12 +509,12 @@ def get_intel_gpu_stats(
     return results
 
 
-def get_openvino_npu_stats() -> Optional[dict[str, str]]:
+def get_openvino_npu_stats() -> dict[str, str] | None:
     """Get NPU stats using openvino."""
     NPU_RUNTIME_PATH = "/sys/devices/pci0000:00/0000:00:0b.0/power/runtime_active_time"
 
     try:
-        with open(NPU_RUNTIME_PATH, "r") as f:
+        with open(NPU_RUNTIME_PATH) as f:
             initial_runtime = float(f.read().strip())
 
         initial_time = time.time()
@@ -523,7 +523,7 @@ def get_openvino_npu_stats() -> Optional[dict[str, str]]:
         time.sleep(1.0)
 
         # Read runtime value again
-        with open(NPU_RUNTIME_PATH, "r") as f:
+        with open(NPU_RUNTIME_PATH) as f:
             current_runtime = float(f.read().strip())
 
         current_time = time.time()
@@ -542,10 +542,10 @@ def get_openvino_npu_stats() -> Optional[dict[str, str]]:
         return None
 
 
-def get_rockchip_gpu_stats() -> Optional[dict[str, str | float]]:
+def get_rockchip_gpu_stats() -> dict[str, str | float] | None:
     """Get GPU stats using rk."""
     try:
-        with open("/sys/kernel/debug/rkrga/load", "r") as f:
+        with open("/sys/kernel/debug/rkrga/load") as f:
             content = f.read()
     except FileNotFoundError:
         return None
@@ -563,7 +563,7 @@ def get_rockchip_gpu_stats() -> Optional[dict[str, str | float]]:
     stats: dict[str, str | float] = {"gpu": average_load, "mem": "-%"}
 
     try:
-        with open("/sys/class/thermal/thermal_zone5/temp", "r") as f:
+        with open("/sys/class/thermal/thermal_zone5/temp") as f:
             line = f.readline().strip()
             stats["temp"] = round(int(line) / 1000, 1)
     except (FileNotFoundError, OSError, ValueError):
@@ -572,10 +572,10 @@ def get_rockchip_gpu_stats() -> Optional[dict[str, str | float]]:
     return stats
 
 
-def get_rockchip_npu_stats() -> Optional[dict[str, float | str]]:
+def get_rockchip_npu_stats() -> dict[str, float | str] | None:
     """Get NPU stats using rk."""
     try:
-        with open("/sys/kernel/debug/rknpu/load", "r") as f:
+        with open("/sys/kernel/debug/rknpu/load") as f:
             npu_output = f.read()
 
             if "Core0:" in npu_output:
@@ -595,7 +595,7 @@ def get_rockchip_npu_stats() -> Optional[dict[str, float | str]]:
     stats: dict[str, float | str] = {"npu": mean, "mem": "-%"}
 
     try:
-        with open("/sys/class/thermal/thermal_zone6/temp", "r") as f:
+        with open("/sys/class/thermal/thermal_zone6/temp") as f:
             line = f.readline().strip()
             stats["temp"] = round(int(line) / 1000, 1)
     except (FileNotFoundError, OSError, ValueError):
@@ -604,7 +604,7 @@ def get_rockchip_npu_stats() -> Optional[dict[str, float | str]]:
     return stats
 
 
-def get_axcl_npu_stats() -> Optional[dict[str, str | float]]:
+def get_axcl_npu_stats() -> dict[str, str | float] | None:
     """Get NPU stats using axcl."""
     # Check if axcl-smi exists
     axcl_smi_path = "/usr/bin/axcl/axcl-smi"
@@ -721,18 +721,18 @@ def get_nvidia_gpu_stats() -> dict[int, dict]:
         return results
 
 
-def get_jetson_stats() -> Optional[dict[int, dict]]:
+def get_jetson_stats() -> dict[int, dict] | None:
     results = {}
 
     try:
         results["mem"] = "-"  # no discrete gpu memory
 
         if os.path.exists("/sys/devices/gpu.0/load"):
-            with open("/sys/devices/gpu.0/load", "r") as f:
+            with open("/sys/devices/gpu.0/load") as f:
                 gpuload = float(f.readline()) / 10
                 results["gpu"] = f"{gpuload}%"
         elif os.path.exists("/sys/devices/platform/gpu.0/load"):
-            with open("/sys/devices/platform/gpu.0/load", "r") as f:
+            with open("/sys/devices/platform/gpu.0/load") as f:
                 gpuload = float(f.readline()) / 10
                 results["gpu"] = f"{gpuload}%"
         else:
@@ -793,7 +793,7 @@ def get_hailo_temps() -> dict[str, float]:
 def is_go2rtc_arbitrary_exec_allowed() -> bool:
     """Read the GO2RTC_ALLOW_ARBITRARY_EXEC override from env, docker
     secrets, or the Home Assistant add-on options file."""
-    raw: Optional[str] = None
+    raw: str | None = None
     if "GO2RTC_ALLOW_ARBITRARY_EXEC" in os.environ:
         raw = os.environ.get("GO2RTC_ALLOW_ARBITRARY_EXEC")
     elif (
@@ -839,7 +839,7 @@ def ffprobe_stream(ffmpeg, path: str, detailed: bool = False) -> sp.CompletedPro
     else:
         format_entries = None
 
-    def run(rtsp_transport: Optional[str] = None) -> sp.CompletedProcess:
+    def run(rtsp_transport: str | None = None) -> sp.CompletedProcess:
         cmd = [ffmpeg.ffprobe_path]
         if rtsp_transport:
             cmd += ["-rtsp_transport", rtsp_transport]
@@ -883,14 +883,14 @@ KEYFRAME_PROBE_WINDOW_SECONDS = 20
 KEYFRAME_GAP_WARNING_SECONDS = 4.0
 
 
-def parse_keyframe_packets(output: str) -> Tuple[List[float], Optional[float]]:
+def parse_keyframe_packets(output: str) -> tuple[list[float], float | None]:
     """Parse ffprobe CSV `pts_time,flags` output.
 
     Returns the presentation timestamps of keyframes (flags containing "K")
     and the maximum timestamp observed across all packets.
     """
-    keyframe_pts: List[float] = []
-    max_pts: Optional[float] = None
+    keyframe_pts: list[float] = []
+    max_pts: float | None = None
 
     for line in output.splitlines():
         parts = line.split(",")
@@ -909,7 +909,7 @@ def parse_keyframe_packets(output: str) -> Tuple[List[float], Optional[float]]:
 
 
 def classify_keyframe_gaps(
-    keyframe_pts: List[float], segment_time: int
+    keyframe_pts: list[float], segment_time: int
 ) -> dict[str, Any]:
     """Classify keyframe spacing for recording suitability.
 
@@ -990,7 +990,7 @@ async def analyze_record_keyframes(
             stderr=asyncio.subprocess.PIPE,
         )
         stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=window + 15)
-    except asyncio.TimeoutError:
+    except TimeoutError:
         logger.warning("Keyframe probe timed out for record stream")
         proc.kill()
         return classify_keyframe_gaps([], segment_time)
@@ -1004,7 +1004,7 @@ async def analyze_record_keyframes(
     return result
 
 
-def vainfo_hwaccel(device_name: Optional[str] = None) -> sp.CompletedProcess:
+def vainfo_hwaccel(device_name: str | None = None) -> sp.CompletedProcess:
     """Run vainfo."""
     if not device_name:
         cmd = ["vainfo"]
@@ -1081,8 +1081,8 @@ async def get_video_properties(
 ) -> dict[str, Any]:
     async def probe_with_ffprobe(
         url: str,
-        rtsp_transport: Optional[str] = None,
-    ) -> tuple[bool, int, int, Optional[str], float]:
+        rtsp_transport: str | None = None,
+    ) -> tuple[bool, int, int, str | None, float]:
         """Fallback using ffprobe: returns (valid, width, height, codec, duration)."""
         cmd = [ffmpeg.ffprobe_path]
         if rtsp_transport:
@@ -1105,7 +1105,7 @@ async def get_video_properties(
             )
             try:
                 stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=6)
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 logger.info(
                     "ffprobe timed out while probing %s (transport=%s)",
                     clean_camera_user_pass(url),
@@ -1137,7 +1137,7 @@ async def get_video_properties(
         except (json.JSONDecodeError, ValueError, KeyError, sp.SubprocessError):
             return False, 0, 0, None, -1
 
-    def probe_with_cv2(url: str) -> tuple[bool, int, int, Optional[str], float]:
+    def probe_with_cv2(url: str) -> tuple[bool, int, int, str | None, float]:
         """Primary attempt using cv2: returns (valid, width, height, fourcc, duration)."""
         cap = cv2.VideoCapture(url)
         if not cap.isOpened():
@@ -1197,10 +1197,10 @@ async def get_video_properties(
 
 def process_logs(
     contents: str,
-    service: Optional[str] = None,
-    start: Optional[int] = None,
-    end: Optional[int] = None,
-) -> Tuple[int, List[str]]:
+    service: str | None = None,
+    start: int | None = None,
+    end: int | None = None,
+) -> tuple[int, list[str]]:
     log_lines = []
     last_message = None
     last_timestamp = None

frigate/util/time.py

@@ -2,7 +2,6 @@
 
 import datetime
 import logging
-from typing import Tuple
 from zoneinfo import ZoneInfoNotFoundError
 
 import pytz
@@ -11,7 +10,7 @@ from tzlocal import get_localzone
 logger = logging.getLogger(__name__)
 
 
-def get_tz_modifiers(tz_name: str) -> Tuple[str, str, float]:
+def get_tz_modifiers(tz_name: str) -> tuple[str, str, float]:
     seconds_offset = (
         datetime.datetime.now(pytz.timezone(tz_name)).utcoffset().total_seconds()
     )
@@ -27,24 +26,18 @@ def get_tomorrow_at_time(hour: int) -> datetime.datetime:
     try:
         tomorrow = datetime.datetime.now(get_localzone()) + datetime.timedelta(days=1)
     except ZoneInfoNotFoundError:
-        tomorrow = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
-            days=1
-        )
+        tomorrow = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1)
         logger.warning(
             "Using utc for maintenance due to missing or incorrect timezone set"
         )
 
-    return tomorrow.replace(hour=hour, minute=0, second=0).astimezone(
-        datetime.timezone.utc
-    )
+    return tomorrow.replace(hour=hour, minute=0, second=0).astimezone(datetime.UTC)
 
 
 def is_current_hour(timestamp: int) -> bool:
     """Returns if timestamp is in the current UTC hour."""
     start_of_next_hour = (
-        datetime.datetime.now(datetime.timezone.utc).replace(
-            minute=0, second=0, microsecond=0
-        )
+        datetime.datetime.now(datetime.UTC).replace(minute=0, second=0, microsecond=0)
         + datetime.timedelta(hours=1)
     ).timestamp()
     return timestamp < start_of_next_hour