diff --git a/frigate/api/motion_search.py b/frigate/api/motion_search.py index 313f303990..15f692047e 100644 --- a/frigate/api/motion_search.py +++ b/frigate/api/motion_search.py @@ -23,9 +23,6 @@ logger = logging.getLogger(__name__) router = APIRouter(tags=[Tags.motion_search]) -# Covers the UI's day windows (25 h on a DST fall-back day) with room to spare. -MAX_SEARCH_RANGE_SECONDS = 48 * 3600 - class MotionSearchRequest(BaseModel): """Request body for motion search.""" @@ -158,15 +155,6 @@ async def start_motion_search( status_code=400, ) - if body.end_time - body.start_time > MAX_SEARCH_RANGE_SECONDS: - return JSONResponse( - content={ - "success": False, - "message": f"Time range must be {MAX_SEARCH_RANGE_SECONDS // 3600} hours or less", - }, - status_code=400, - ) - try: job_id = start_motion_search_job( config=config, diff --git a/frigate/jobs/motion_search.py b/frigate/jobs/motion_search.py index 75fc86aff5..2d8617972b 100644 --- a/frigate/jobs/motion_search.py +++ b/frigate/jobs/motion_search.py @@ -20,7 +20,6 @@ from frigate.jobs.job import Job from frigate.jobs.motion_search_batch import ( build_segment_time_map, coalesce_runs, - keyframes_cover_stream, stream_time_to_absolute, ) from frigate.jobs.motion_search_decode import ( @@ -54,9 +53,6 @@ FALLBACK_SAMPLE_FPS = 2.0 MAX_ACTIVE_JOBS = 3 # Finished jobs stay pollable for this long before eviction (seconds). FINISHED_JOB_TTL = 600.0 -# Missing stream time beyond this marks a run as truncated, e.g. a segment the -# HLS demuxer skipped after a read timeout. -MAX_STREAM_HOLE_SECONDS = 5.0 # A cancelled job stops holding a slot after this long, even if its thread is wedged. CANCEL_GRACE_SECONDS = 60.0 @@ -654,44 +650,37 @@ class MotionSearchRunner(threading.Thread): kf_pts = probe_vod_keyframe_pts( self.ffprobe_path, vod_url, self._should_stop ) - # Probe and decode fetch the stream separately, so equal counts only - # pair frames with timestamps when the probe saw the whole stream. - if keyframes_cover_stream( - kf_pts, _stream_duration(time_map), MAX_STREAM_HOLE_SECONDS - ): - frames = list( - iter_vod_frames( - self.ffmpeg_path, - vod_url, - self.scaled[0], - self.scaled[1], - self.channels, - self.decode_args, - self.crop, - self.scaled, - True, - self._should_stop, - skip_nonkey=True, - fps_rate=None, - ) + frames = list( + iter_vod_frames( + self.ffmpeg_path, + vod_url, + self.scaled[0], + self.scaled[1], + self.channels, + self.decode_args, + self.crop, + self.scaled, + True, + self._should_stop, + skip_nonkey=True, + fps_rate=None, ) - if len(frames) == len(kf_pts): - abs_times = [stream_time_to_absolute(time_map, p) for p in kf_pts] - indexed = list(enumerate(frames)) + ) + if kf_pts and len(frames) == len(kf_pts): + abs_times = [stream_time_to_absolute(time_map, p) for p in kf_pts] + indexed = list(enumerate(frames)) - def _ts_kf(i: int) -> float: - return abs_times[i] + def _ts_kf(i: int) -> float: + return abs_times[i] - results = self._detect_with_progress(indexed, _ts_kf) - return results, len(frames) + results = self._detect_with_progress(indexed, _ts_kf) + return results, len(frames) - logger.debug( - "Keyframe count mismatch (%d decoded vs %d probed), using cadence", - len(frames), - len(kf_pts), - ) - else: - logger.debug("Keyframe probe has a gap, using cadence") + logger.debug( + "Keyframe count mismatch (%d decoded vs %d probed), using cadence", + len(frames), + len(kf_pts), + ) return self._process_run_cadence(vod_url, time_map) @@ -715,15 +704,6 @@ class MotionSearchRunner(threading.Thread): fps_rate=self.fps_rate, ) ) - expected = _stream_duration(time_map) * self.fps_rate - if ( - not self._should_stop() - and expected - len(frames) > MAX_STREAM_HOLE_SECONDS * self.fps_rate - ): - raise RuntimeError( - f"VOD decode returned {len(frames)} of ~{expected:.0f} frames" - ) - indexed = list(enumerate(frames)) def _ts_fps(i: int) -> float: @@ -886,11 +866,6 @@ def _evict_finished_jobs(now: float) -> None: del _motion_search_jobs[job_id] -def _stream_duration(time_map: list[tuple[float, float, float]]) -> float: - offset, _, duration = time_map[-1] - return offset + duration - - def _is_active(job: MotionSearchJob, now: float) -> bool: """True while the job's thread may still be running and holding a slot.""" if job.end_time is not None: @@ -943,10 +918,9 @@ def start_motion_search_job( now = datetime.now().timestamp() _evict_finished_jobs(now) - # A cancelled job keeps its thread until the runner sets end_time, so - # count threads rather than statuses. + # A cancelled job keeps its thread until the runner sets end_time, so it + # still counts toward the global cap. Its user may start a new search. active = [j for j, _ in _motion_search_jobs.values() if _is_active(j, now)] - # The UI forgets a cancelled job at once, so it only counts globally. if username is not None and any( j.username == username and j.status != JobStatusTypesEnum.cancelled for j in active diff --git a/frigate/jobs/motion_search_batch.py b/frigate/jobs/motion_search_batch.py index 66799b3d93..f916da5e0a 100644 --- a/frigate/jobs/motion_search_batch.py +++ b/frigate/jobs/motion_search_batch.py @@ -73,14 +73,3 @@ def stream_time_to_absolute( idx = 0 stream_offset, abs_start, _duration = time_map[idx] return abs_start + (stream_time - stream_offset) - - -def keyframes_cover_stream(pts: list[float], duration: float, max_hole: float) -> bool: - """True if sorted keyframe ``pts`` span 0..``duration`` with no gap over ``max_hole``. - - A segment skipped or cut short by a read timeout leaves such a gap. - """ - if not pts: - return False - edges = [0.0, *pts, duration] - return max(b - a for a, b in zip(edges, edges[1:])) <= max_hole diff --git a/frigate/jobs/motion_search_decode.py b/frigate/jobs/motion_search_decode.py index ed7d1fd445..ba56ae2d44 100644 --- a/frigate/jobs/motion_search_decode.py +++ b/frigate/jobs/motion_search_decode.py @@ -107,16 +107,7 @@ def keyframe_sampling_eligible( return max(gaps) <= max_gap -# rw_timeout (microseconds) ends a stalled VOD read instead of blocking forever. -# The HLS demuxer logs this at warning level when a segment read fails or times out. -HLS_SEGMENT_SKIPPED = b"failed too many times, skipping" - -VOD_PROTOCOL_ARGS = [ - "-protocol_whitelist", - "pipe,file,http,tcp", - "-rw_timeout", - "30000000", -] +VOD_PROTOCOL_ARGS = ["-protocol_whitelist", "pipe,file,http,tcp"] def build_vod_decode_command( @@ -153,7 +144,7 @@ def build_vod_decode_command( filters.append(f"scale={sw}:{sh}") pix_fmt = "gray" if gray else "bgr24" - cmd = [ffmpeg_path, "-hide_banner", "-loglevel", "warning"] + cmd = [ffmpeg_path, "-hide_banner", "-loglevel", "error"] if skip_nonkey: cmd += ["-skip_frame", "nokey"] cmd += [*decode_args, *VOD_PROTOCOL_ARGS, "-i", vod_url, "-an"] @@ -215,13 +206,8 @@ def _run_vod_decode( yield frame finally: terminate_ffmpeg_stream(proc) - stderr_file.seek(0) - segment_skipped = HLS_SEGMENT_SKIPPED in stderr_file.read() stderr_file.close() - if segment_skipped and not should_stop(): - raise RuntimeError("VOD decode skipped a segment that failed to load") - if count == 0 and software_retry and not should_stop(): logger.warning("Hardware VOD decode produced no frames, retrying in software") yield from _run_vod_decode( @@ -355,7 +341,6 @@ def probe_video_dimensions( ffprobe_path, "-v", "error", - *VOD_PROTOCOL_ARGS, "-select_streams", "v:0", "-show_entries", diff --git a/frigate/test/http_api/test_http_motion_search.py b/frigate/test/http_api/test_http_motion_search.py index 003a2d5801..35a0211679 100644 --- a/frigate/test/http_api/test_http_motion_search.py +++ b/frigate/test/http_api/test_http_motion_search.py @@ -97,14 +97,6 @@ class TestHttpMotionSearch(BaseTestHttp): self.assertEqual(self.start(client, f"user{i}").status_code, 200) self.assertEqual(self.start(client, "late").status_code, 429) - def test_over_long_range_rejected(self): - body = {**BODY, "end_time": BODY["start_time"] + 48 * 3600 + 1} - with AuthTestClient(self.app) as client: - resp = self.start(client, "alice", body) - - self.assertEqual(resp.status_code, 400) - self.assertEqual(motion_search._motion_search_jobs, {}) - def test_finished_jobs_evicted_after_ttl(self): with AuthTestClient(self.app) as client: old_id = self.start(client, "alice").json()["job_id"] diff --git a/frigate/test/test_motion_search_batch.py b/frigate/test/test_motion_search_batch.py index 9c619034d4..0e7fcdfd8a 100644 --- a/frigate/test/test_motion_search_batch.py +++ b/frigate/test/test_motion_search_batch.py @@ -6,7 +6,6 @@ from dataclasses import dataclass from frigate.jobs.motion_search_batch import ( build_segment_time_map, coalesce_runs, - keyframes_cover_stream, stream_time_to_absolute, ) @@ -57,17 +56,3 @@ class TestTimestampMapping(unittest.TestCase): run = [_Seg("a", 1000.0, 1010.0)] time_map = build_segment_time_map(run) self.assertAlmostEqual(stream_time_to_absolute(time_map, 9.9), 1009.9) - - -class TestKeyframesCoverStream(unittest.TestCase): - def test_dense_keyframes_cover(self): - self.assertTrue(keyframes_cover_stream([0.5, 2.5, 4.5, 6.5, 8.5], 10.0, 5.0)) - - def test_gap_from_skipped_segment(self): - self.assertFalse(keyframes_cover_stream([0.5, 2.5, 12.5, 14.5], 20.0, 5.0)) - - def test_truncated_tail(self): - self.assertFalse(keyframes_cover_stream([0.5, 2.5, 4.5], 20.0, 5.0)) - - def test_empty(self): - self.assertFalse(keyframes_cover_stream([], 10.0, 5.0)) diff --git a/frigate/test/test_motion_search_decode.py b/frigate/test/test_motion_search_decode.py index 113bbd18a2..75e29e4351 100644 --- a/frigate/test/test_motion_search_decode.py +++ b/frigate/test/test_motion_search_decode.py @@ -1,6 +1,5 @@ """Tests for the motion search hardware-accelerated decode helpers.""" -import io import subprocess as sp import unittest from types import SimpleNamespace @@ -9,7 +8,6 @@ from unittest import mock from frigate.jobs.motion_search_decode import ( KEYFRAME_MAX_GAP_SECONDS, build_vod_decode_command, - iter_vod_frames, keyframe_sampling_eligible, probe_video_dimensions, probe_vod_keyframe_pts, @@ -104,7 +102,6 @@ class TestVodDecodeCommand(unittest.TestCase): joined = " ".join(cmd) self.assertIn("-skip_frame nokey", joined) self.assertIn("-protocol_whitelist pipe,file,http,tcp", joined) - self.assertLess(cmd.index("-rw_timeout"), cmd.index("-i")) self.assertIn(f"-i {self.URL}", joined) self.assertIn("crop=100:80:10:20", joined) self.assertIn("scale=50:40", joined) @@ -143,44 +140,6 @@ class TestVodDecodeCommand(unittest.TestCase): self.assertIn("format=nv12", joined) -class TestVodDecodeSkippedSegment(unittest.TestCase): - def _decode(self, stderr_output: bytes) -> list: - def popen(cmd, stdout, stderr): - stderr.write(stderr_output) - return mock.Mock(stdout=io.BytesIO(b"\0" * 8)) - - with ( - mock.patch("frigate.jobs.motion_search_decode.sp.Popen", side_effect=popen), - mock.patch("frigate.jobs.motion_search_decode.terminate_ffmpeg_stream"), - ): - return list( - iter_vod_frames( - "ffmpeg", - "http://x/index.m3u8", - 2, - 2, - 1, - [], - None, - None, - True, - lambda: False, - skip_nonkey=False, - fps_rate=2.0, - ) - ) - - def test_clean_decode_returns_frames(self): - self.assertEqual(len(self._decode(b"")), 2) - - def test_skipped_segment_raises(self): - stderr = ( - b"[hls @ 0x1] Segment 1 of playlist 0 failed too many times, skipping\n" - ) - with self.assertRaises(RuntimeError): - self._decode(stderr) - - class TestProbeVodKeyframePts(unittest.TestCase): def _proc(self, stdout="", returncode=0): proc = mock.Mock(returncode=returncode) @@ -197,13 +156,11 @@ class TestProbeVodKeyframePts(unittest.TestCase): with mock.patch( "frigate.jobs.motion_search_decode.sp.Popen", return_value=self._proc(sample), - ) as popen: + ): pts = probe_vod_keyframe_pts( "ffprobe", "http://x/index.m3u8", lambda: False ) self.assertEqual(pts, [0.0, 2.0]) - cmd = popen.call_args.args[0] - self.assertLess(cmd.index("-rw_timeout"), cmd.index("-i")) def test_returns_empty_on_failure(self): with mock.patch( @@ -237,10 +194,9 @@ class TestProbeVideoDimensions(unittest.TestCase): completed = mock.Mock(stdout=sample, returncode=0) with mock.patch( "frigate.jobs.motion_search_decode.sp.run", return_value=completed - ) as run: + ): dims = probe_video_dimensions("ffprobe", "/tmp/a.mp4") assert dims is not None - self.assertIn("-rw_timeout", run.call_args.args[0]) width, height, fps = dims self.assertEqual((width, height), (1920, 1080)) self.assertAlmostEqual(fps, 29.97, places=2) diff --git a/frigate/test/test_motion_search_runner.py b/frigate/test/test_motion_search_runner.py deleted file mode 100644 index 08bc0d578f..0000000000 --- a/frigate/test/test_motion_search_runner.py +++ /dev/null @@ -1,85 +0,0 @@ -"""Tests for motion search run decoding and truncation checks.""" - -import threading -import unittest -from types import SimpleNamespace -from unittest import mock - -import numpy as np - -from frigate.jobs.motion_search import MotionSearchJob, MotionSearchRunner - -# Two contiguous 10 s segments, so the run's stream is 20 s long. -RUN = [ - SimpleNamespace(start_time=1000.0, end_time=1010.0), - SimpleNamespace(start_time=1010.0, end_time=1020.0), -] -FRAME = np.zeros((2, 2), dtype=np.uint8) - - -class TestProcessRun(unittest.TestCase): - def setUp(self): - requestor = mock.patch("frigate.jobs.motion_search.InterProcessRequestor") - requestor.start() - self.addCleanup(requestor.stop) - - self.runner = MotionSearchRunner( - MotionSearchJob(camera="cam"), mock.Mock(), threading.Event() - ) - self.runner.fps_rate = 2.0 - self.runner._detect_with_progress = mock.Mock(return_value=[]) - - def _process(self, kf_pts, keyframe_frames, cadence_frames): - def frames(*args, skip_nonkey, fps_rate): - count = keyframe_frames if skip_nonkey else cadence_frames - return iter([FRAME] * count) - - with ( - mock.patch( - "frigate.jobs.motion_search.probe_vod_keyframe_pts", - return_value=kf_pts, - ), - mock.patch( - "frigate.jobs.motion_search.iter_vod_frames", side_effect=frames - ) as decode, - ): - result = self.runner._process_run(RUN) - return result, [c.kwargs["skip_nonkey"] for c in decode.call_args_list] - - def test_complete_keyframe_probe_pairs_timestamps(self): - pts = [0.5, 2.5, 4.5, 6.5, 8.5, 10.5, 12.5, 14.5, 16.5, 18.5] - (_, frames), modes = self._process(pts, len(pts), 0) - - self.assertEqual(frames, len(pts)) - self.assertEqual(modes, [True]) - ts_fn = self.runner._detect_with_progress.call_args.args[1] - self.assertAlmostEqual(ts_fn(5), 1010.5) - - def test_keyframe_probe_gap_falls_back_to_cadence(self): - # Probe lost the second segment; an equal decode count must not be trusted. - pts = [0.5, 2.5, 4.5, 6.5, 8.5] - (_, frames), modes = self._process(pts, len(pts), 40) - - self.assertEqual(modes, [False]) - self.assertEqual(frames, 40) - - def test_cadence_shortfall_raises(self): - self.runner.use_keyframe = False - with self.assertRaises(RuntimeError): - self._process([], 0, 20) - - def test_zero_frames_raises(self): - self.runner.use_keyframe = False - with self.assertRaises(RuntimeError): - self._process([], 0, 0) - - def test_small_cadence_drift_is_tolerated(self): - self.runner.use_keyframe = False - (_, frames), _ = self._process([], 0, 38) - self.assertEqual(frames, 38) - - def test_shortfall_after_stop_does_not_raise(self): - self.runner.use_keyframe = False - self.runner.cancel_event.set() - (_, frames), _ = self._process([], 0, 3) - self.assertEqual(frames, 3)