diff --git a/docker/main/requirements-wheels.txt b/docker/main/requirements-wheels.txt index 7bd098454d..dc67e082b5 100644 --- a/docker/main/requirements-wheels.txt +++ b/docker/main/requirements-wheels.txt @@ -1,4 +1,4 @@ -aiofiles == 24.1.* +anyio == 4.14.* click == 8.1.* # FastAPI aiohttp == 3.12.* diff --git a/frigate/api/app.py b/frigate/api/app.py index 7f78f4b56c..113cf7e87e 100644 --- a/frigate/api/app.py +++ b/frigate/api/app.py @@ -14,8 +14,8 @@ from io import StringIO from pathlib import Path as FilePath from typing import Any -import aiofiles import ruamel.yaml +from anyio import open_file as aopen from fastapi import APIRouter, Body, Path, Request, Response from fastapi.encoders import jsonable_encoder from fastapi.params import Depends @@ -1045,7 +1045,7 @@ async def logs( """Asynchronously stream log lines.""" buffer = "" try: - async with aiofiles.open(file_path) as file: + async with await aopen(file_path) as file: await file.seek(0, 2) while True: line = await file.readline() @@ -1083,7 +1083,7 @@ async def logs( # For full logs initially try: - async with aiofiles.open(service_location) as file: + async with await aopen(service_location) as file: contents = await file.read() total_lines, log_lines = process_logs(contents, service, start, end) diff --git a/frigate/api/camera.py b/frigate/api/camera.py index 29e861b3f5..c2b28f4783 100644 --- a/frigate/api/camera.py +++ b/frigate/api/camera.py @@ -10,6 +10,7 @@ from urllib.parse import quote_plus import httpx import requests +from anyio import open_file as aopen from fastapi import APIRouter, Depends, Query, Request, Response from fastapi.responses import JSONResponse from filelock import FileLock, Timeout @@ -1188,15 +1189,17 @@ async def delete_camera( try: with lock: - with open(config_file) as f: - old_raw_config = f.read() + async with await aopen(config_file) as f: + old_raw_config = await f.read() try: yaml = YAML() yaml.indent(mapping=2, sequence=4, offset=2) - with open(config_file) as f: - data = yaml.load(f) + async with await aopen(config_file) as f: + text = await f.read() + + data = yaml.load(text) # Remove camera from config if "cameras" in data and camera_name in data["cameras"]: @@ -1221,17 +1224,17 @@ async def delete_camera( for role_name in empty_roles: del auth["roles"][role_name] - with open(config_file, "w") as f: + async with await aopen(config_file, "w") as f: yaml.dump(data, f) - with open(config_file) as f: - new_raw_config = f.read() + async with await aopen(config_file) as f: + new_raw_config = await f.read() try: config = FrigateConfig.parse(new_raw_config) except Exception: - with open(config_file, "w") as f: - f.write(old_raw_config) + async with await aopen(config_file, "w") as f: + await f.write(old_raw_config) logger.exception( "Config error after removing camera %s", camera_name, diff --git a/frigate/api/event.py b/frigate/api/event.py index d37a547ba4..7aa62d3cf5 100644 --- a/frigate/api/event.py +++ b/frigate/api/event.py @@ -13,6 +13,7 @@ from pathlib import Path from urllib.parse import unquote import numpy as np +from anyio import Path as AsyncPath from fastapi import APIRouter, Request from fastapi.params import Depends from fastapi.responses import JSONResponse @@ -1455,10 +1456,10 @@ async def set_attributes( dataset_dir = os.path.join(CLIPS_DIR, sanitize_filename(model_key), "dataset") available_labels = set() - if os.path.exists(dataset_dir): + if await AsyncPath(dataset_dir).exists(): for category_name in os.listdir(dataset_dir): category_dir = os.path.join(dataset_dir, category_name) - if os.path.isdir(category_dir): + if await AsyncPath(category_dir).is_dir(): available_labels.add(category_name) if not available_labels: diff --git a/frigate/api/media.py b/frigate/api/media.py index 8d56475c59..b7f57471f1 100644 --- a/frigate/api/media.py +++ b/frigate/api/media.py @@ -15,6 +15,8 @@ from urllib.parse import unquote import cv2 import numpy as np import pytz +from anyio import Path as AsyncPath +from anyio import open_file as aopen from fastapi import APIRouter, Depends, Path, Query, Request, Response from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from pathvalidate import sanitize_filename @@ -497,18 +499,18 @@ async def recording_clip( file_name = sanitize_filename(f"playlist_{camera_name}_{start_ts}-{end_ts}.txt") file_path = os.path.join(CACHE_DIR, file_name) - with open(file_path, "w") as file: + async with await aopen(file_path, "w") as file: clip: Recordings for clip in recordings: - file.write(f"file '{clip.path}'\n") + await file.write(f"file '{clip.path}'\n") # if this is the starting clip, add an inpoint if clip.start_time < start_ts: - file.write(f"inpoint {int(start_ts - clip.start_time)}\n") + await file.write(f"inpoint {int(start_ts - clip.start_time)}\n") # if this is the ending clip, add an outpoint if clip.end_time > end_ts: - file.write(f"outpoint {int(end_ts - clip.start_time)}\n") + await file.write(f"outpoint {int(end_ts - clip.start_time)}\n") if len(file_name) > 1000: return JSONResponse( @@ -1149,8 +1151,8 @@ async def event_snapshot_clean(request: Request, event_id: str, download: bool = ) if image_path.endswith(".webp"): - with open(image_path, "rb") as image_file: - webp_bytes = image_file.read() + async with await aopen(image_path, "rb") as image_file: + webp_bytes = await image_file.read() else: image = load_event_snapshot_image(event, clean_only=True)[0] if image is None: @@ -1366,7 +1368,7 @@ async def preview_gif( # need to generate from existing images preview_dir = os.path.join(CACHE_DIR, "preview_frames") - if not os.path.isdir(preview_dir): + if not await AsyncPath(preview_dir).is_dir(): return JSONResponse( content={"success": False, "message": "Preview not found"}, status_code=404, @@ -1555,7 +1557,7 @@ async def preview_mp4( # need to generate from existing images preview_dir = os.path.join(CACHE_DIR, "preview_frames") - if not os.path.isdir(preview_dir): + if not await AsyncPath(preview_dir).is_dir(): return JSONResponse( content={"success": False, "message": "Preview not found"}, status_code=404, @@ -1633,7 +1635,7 @@ async def preview_mp4( "Content-Description": "File Transfer", "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}", "Content-Type": "video/mp4", - "Content-Length": str(os.path.getsize(path)), + "Content-Length": str((await AsyncPath(path).stat()).st_size), # nginx: https://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_ignore_headers "X-Accel-Redirect": f"/cache/{file_name}", } @@ -1707,10 +1709,10 @@ async def preview_thumbnail(request: Request, file_name: str): preview_dir = os.path.join(CACHE_DIR, "preview_frames") try: - with open( + async with await aopen( os.path.join(preview_dir, safe_file_name_current), "rb" ) as image_file: - jpg_bytes = image_file.read() + jpg_bytes = await image_file.read() except FileNotFoundError: return JSONResponse( content=({"success": False, "message": "Image file not found"}), diff --git a/frigate/api/record.py b/frigate/api/record.py index 5db257f482..76da5b21da 100644 --- a/frigate/api/record.py +++ b/frigate/api/record.py @@ -4,9 +4,9 @@ import datetime as dt import logging from datetime import datetime, timedelta from functools import reduce -from pathlib import Path from urllib.parse import unquote +from anyio import Path as AsyncPath from fastapi import APIRouter, Depends, Request from fastapi import Path as PathParam from fastapi.responses import JSONResponse @@ -443,7 +443,7 @@ async def delete_recordings( recording_ids.append(recording["id"]) try: - Path(recording["path"]).unlink(missing_ok=True) + await AsyncPath(recording["path"]).unlink(missing_ok=True) deleted_count += 1 except Exception as e: logger.error(f"Failed to delete recording file {recording['path']}: {e}") diff --git a/frigate/record/maintainer.py b/frigate/record/maintainer.py index 7f9dbc19da..f175065880 100644 --- a/frigate/record/maintainer.py +++ b/frigate/record/maintainer.py @@ -15,6 +15,7 @@ from typing import Any import numpy as np import psutil +from anyio import Path as AsyncPath from frigate.comms.detections_updater import DetectionSubscriber, DetectionTypeEnum from frigate.comms.inter_process import InterProcessRequestor @@ -105,11 +106,11 @@ class RecordingMaintainer(threading.Thread): async def move_files(self) -> None: cache_files = [ - d - for d in os.listdir(CACHE_DIR) - if os.path.isfile(os.path.join(CACHE_DIR, d)) - and d.endswith(".mp4") - and not d.startswith("preview_") + path.name + async for path in AsyncPath(CACHE_DIR).iterdir() + if await path.is_file() + and path.suffix == ".mp4" + and not path.name.startswith("preview_") ] # publish newest cached segment per camera (including in use files) @@ -229,7 +230,7 @@ class RecordingMaintainer(threading.Thread): to_remove = grouped_recordings[camera][:-keep_count] for rec in to_remove: cache_path = rec["cache_path"] - Path(cache_path).unlink(missing_ok=True) + await AsyncPath(cache_path).unlink(missing_ok=True) self.end_time_cache.pop(cache_path, None) grouped_recordings[camera] = grouped_recordings[camera][-keep_count:] @@ -244,7 +245,7 @@ class RecordingMaintainer(threading.Thread): to_remove = grouped_recordings[camera][:-keep_count] for rec in to_remove: cache_path = rec["cache_path"] - Path(cache_path).unlink(missing_ok=True) + await AsyncPath(cache_path).unlink(missing_ok=True) self.end_time_cache.pop(cache_path, None) grouped_recordings[camera] = grouped_recordings[camera][-keep_count:] @@ -634,7 +635,7 @@ class RecordingMaintainer(threading.Thread): file_path = os.path.join(directory, file_name) try: - if not os.path.exists(file_path): + if not await AsyncPath(file_path).exists(): start_frame = datetime.datetime.now().timestamp() # add faststart to kept segments to improve metadata reading @@ -670,7 +671,9 @@ class RecordingMaintainer(threading.Thread): # get the segment size of the cache file # file without faststart is same size segment_size = round( - float(os.path.getsize(cache_path)) / pow(2, 20), 2 + float((await AsyncPath(cache_path).stat()).st_size) + / pow(2, 20), + 2, ) except OSError: segment_size = 0 @@ -698,7 +701,7 @@ class RecordingMaintainer(threading.Thread): } except Exception as e: logger.error(f"Unable to store recording segment {cache_path}") - Path(cache_path).unlink(missing_ok=True) + await AsyncPath(cache_path).unlink(missing_ok=True) logger.error(e) # clear end_time cache diff --git a/frigate/test/test_maintainer.py b/frigate/test/test_maintainer.py index 715cd5a1a1..8d840c0e4f 100644 --- a/frigate/test/test_maintainer.py +++ b/frigate/test/test_maintainer.py @@ -1,7 +1,7 @@ import datetime import sys import unittest -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch # Mock complex imports before importing maintainer, saving originals so we can # restore them after import and avoid polluting sys.modules for other tests. @@ -42,38 +42,51 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase): # One bad file, one good file files = ["bad_filename.mp4", "camera@20210101000000+0000.mp4"] - with patch("os.listdir", return_value=files): - with patch("os.path.isfile", return_value=True): - with patch( - "frigate.record.maintainer.psutil.process_iter", return_value=[] - ): - with patch("frigate.record.maintainer.logger.warning") as warn: - # Mock validate_and_move_segment to avoid further logic - maintainer.validate_and_move_segment = MagicMock() + mock_paths = [] + for filename in files: + path = MagicMock() + path.name = filename + path.suffix = ".mp4" + path.is_file = AsyncMock(return_value=True) + mock_paths.append(path) - try: - await maintainer.move_files() - except ValueError as e: - if "not enough values to unpack" in str(e): - self.fail("move_files() crashed on bad filename!") - raise e - except Exception: - # Ignore other errors (like DB connection) as we only care about the unpack crash - pass + async def mock_iterdir(): + for path in mock_paths: + yield path - # The bad filename is encountered in multiple loops, but should only warn once. - matching = [ - c - for c in warn.call_args_list - if c.args - and isinstance(c.args[0], str) - and "Skipping unexpected files in cache" in c.args[0] - ] - self.assertEqual( - 1, - len(matching), - f"Expected a single warning for unexpected files, got {len(matching)}", - ) + with patch("frigate.record.maintainer.AsyncPath") as mock_async_path: + mock_async_path.return_value.iterdir = mock_iterdir + + with patch( + "frigate.record.maintainer.psutil.process_iter", return_value=[] + ): + with patch("frigate.record.maintainer.logger.warning") as warn: + # Mock validate_and_move_segment to avoid further logic + maintainer.validate_and_move_segment = MagicMock() + + try: + await maintainer.move_files() + except ValueError as e: + if "not enough values to unpack" in str(e): + self.fail("move_files() crashed on bad filename!") + raise e + except Exception: + # Ignore other errors (like DB connection) as we only care about the unpack crash + pass + + # The bad filename is encountered in multiple loops, but should only warn once. + matching = [ + c + for c in warn.call_args_list + if c.args + and isinstance(c.args[0], str) + and "Skipping unexpected files in cache" in c.args[0] + ] + self.assertEqual( + 1, + len(matching), + f"Expected a single warning for unexpected files, got {len(matching)}", + ) async def test_drops_quiet_segment_when_only_motion_retention(self): # Regression: when motion retention is enabled but a segment has no diff --git a/pyproject.toml b/pyproject.toml index 775db6026c..964a9141c2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,5 +2,20 @@ target-version = "py311" [tool.ruff.lint] -ignore = ["E501","E711","E712","UP031","UP032","UP042","G004"] -extend-select = ["I", "UP", "G", "ASYNC210", "B904"] +ignore = [ + "ASYNC109", # Async function definition with a timeout parameter + "E501", # line-too-long + "E711", # none-comparison + "E712", # true-false-comparison + "UP031", # printf-string-formatting + "UP032", # f-string + "UP042", # replace-str-enum + "G004", # logging-f-string +] +extend-select = [ + "ASYNC", # https://docs.astral.sh/ruff/rules/#flake8-async-async + "B904", # https://docs.astral.sh/ruff/rules/raise-without-from-inside-except/ + "G", # https://docs.astral.sh/ruff/rules/#flake8-logging-format-g + "I", # https://docs.astral.sh/ruff/rules/#isort-i + "UP", # https://docs.astral.sh/ruff/rules/#pyupgrade-up +]