diff --git a/docs/docs/configuration/cameras.md b/docs/docs/configuration/cameras.md index 640b78841f..72305d27ad 100644 --- a/docs/docs/configuration/cameras.md +++ b/docs/docs/configuration/cameras.md @@ -83,11 +83,12 @@ A camera is enabled by default but can be disabled by using `enabled: False`. Ca Each role can only be assigned to one input per camera. The options for roles are as follows: -| Role | Description | -| -------- | ----------------------------------------------------------------------------------- | -| `detect` | Main feed for object detection. [docs](object_detectors.md) | -| `record` | Saves segments of the video feed based on configuration settings. [docs](record.md) | -| `audio` | Feed for audio based detection. [docs](audio_detectors.md) | +| Role | Description | +| ------------ | ------------------------------------------------------------------------------------------------------------ | +| `detect` | Main feed for object detection. [docs](object_detectors.md) | +| `record` | Saves segments of the video feed based on configuration settings. [docs](record.md) | +| `record_sub` | Saves segments of a second, lower quality stream with its own retention. [docs](record.md#sub-stream-recording) | +| `audio` | Feed for audio based detection. [docs](audio_detectors.md) | diff --git a/docs/docs/configuration/record.md b/docs/docs/configuration/record.md index c7a61b2c2c..522db9935c 100644 --- a/docs/docs/configuration/record.md +++ b/docs/docs/configuration/record.md @@ -280,7 +280,7 @@ This configuration will retain recording segments that overlap with alerts and d In addition to the main recording stream, Frigate can record a second, lower quality stream for each camera. This serves two purposes: - **Quality selection during playback**: A quality selector (`Auto`, `Original`, or `Low`) appears in History view for cameras with sub stream recording enabled. `Original` and `Low` play only that stream's recordings. Time ranges where the selected stream has no footage are skipped during playback, and the selector notes when the selected stream has no recordings at all in the viewed time range. With `Auto` (the default), playback prefers the original quality and automatically falls back to the low quality stream when the connection cannot keep up, or for time ranges where the original recordings have expired. The selector shows each stream's video codec and audio details beneath the options; footage recorded by older Frigate versions shows no details. -- **Extended retention**: Sub stream recordings have their own retention settings, fully independent of the main recordings. By giving the low quality recordings a longer retention period, you can keep weeks or months of low quality history using a fraction of the storage, and that history remains playable after the main recordings expire. Playback falls back to the low quality recordings automatically, and the timeline shows a muted treatment for time ranges where only low quality footage remains. +- **Extended retention**: Sub stream recordings have their own retention settings, fully independent of the main recordings. By giving the low quality recordings a longer retention period, you can keep weeks or months of low quality history using a fraction of the storage, and that history remains playable after the main recordings expire. Playback falls back to the low quality recordings automatically, and the timeline shows a muted treatment for time ranges where only low quality footage remains. Timeline previews are kept for as long as either stream still has recordings, so scrubbing works across the whole retained history. ### Configuring sub stream recording @@ -427,7 +427,7 @@ This table covers only features that read recordings from disk. Tracked object s ### Trade-offs -- Recording a second stream increases overall storage use. The increase is typically small relative to the main recordings, since the low quality stream is much smaller. +- Recording a second stream increases overall storage use. The increase is typically small relative to the main recordings, since the low quality stream is much smaller. Both streams are cached before being written to disk, so cache use goes up as well. See [the `/tmp/cache` area is separate](#the-tmpcache-area-is-separate) if you start seeing `No space left on device` errors after enabling it. - The go2rtc transcode approach continuously encodes the low quality stream, which uses CPU or GPU resources. This cost only applies to the transcode path; recording the camera's native sub stream does not re-encode. See the [go2rtc hardware acceleration documentation](https://github.com/AlexxIT/go2rtc?tab=readme-ov-file#source-ffmpeg) for accelerating the transcode. - Many camera sub streams do not include audio. If the source stream has no audio, the low quality recordings will not have audio. - **Matching video codecs and audio settings between the two streams gives the smoothest playback.** When playback combines both qualities on one timeline (the default `Auto` behavior: for example original quality during events with low quality in between, or low quality history after the original recordings expire) and the streams use different video codecs or audio settings, for example H.265 on the main stream and H.264 on the sub stream, or 16 kHz audio on one and 8 kHz on the other, playback still works: Frigate inserts a decoder reset at each quality transition, which can cause a barely-perceptible pause there. Configuring both streams in the camera's firmware to use the same video codec, audio codec, and sample rate makes transitions fully seamless, and a mismatched audio sample rate can also be corrected with [sub stream output args](#sub-stream-output-args). If one stream has audio and the other does not, combined time ranges play **without audio**; selecting a single quality with the playback selector always keeps that stream's audio. diff --git a/docs/docs/integrations/mqtt.md b/docs/docs/integrations/mqtt.md index 002feafe3d..6363021188 100644 --- a/docs/docs/integrations/mqtt.md +++ b/docs/docs/integrations/mqtt.md @@ -304,7 +304,7 @@ Topic with current state of notifications. Published values are `ON` and `OFF`. ### `frigate//status/` -Publishes the current health status of each role that is enabled (`audio`, `detect`, `record`). Possible values are: +Publishes the current health status of each role that is enabled (`audio`, `detect`, `record`, `record_sub`). `record_sub` is only published for cameras with [sub stream recording](/configuration/record#sub-stream-recording) enabled, and is tracked separately from `record` so a healthy main stream can't hide a stalled sub stream. Possible values are: - `online`: Stream is running and being processed - `offline`: Stream is offline and is being restarted diff --git a/frigate/comms/dispatcher.py b/frigate/comms/dispatcher.py index 1ff79fcd41..ef260aa821 100644 --- a/frigate/comms/dispatcher.py +++ b/frigate/comms/dispatcher.py @@ -6,6 +6,8 @@ import logging from collections.abc import Callable, Iterable from typing import Any, cast +from peewee import IntegrityError + from frigate.camera import PTZMetrics from frigate.camera.activity_manager import AudioActivityManager, CameraActivityManager from frigate.comms.base_communicator import Communicator @@ -254,7 +256,21 @@ class Dispatcher: restart_frigate() def handle_insert_many_recordings() -> None: - Recordings.insert_many(payload).execute() + try: + Recordings.insert_many(payload).execute() + except IntegrityError: + logger.warning( + "Batch recording insert failed, inserting rows individually" + ) + + for recording in payload: + try: + Recordings.insert(recording).execute() + except IntegrityError: + logger.warning( + "Skipping recording that is already stored: %s", + recording.get(Recordings.path.name), + ) def handle_request_region_grid() -> Any: camera = payload diff --git a/frigate/comms/recordings_updater.py b/frigate/comms/recordings_updater.py index 249c2f6076..88645dd648 100644 --- a/frigate/comms/recordings_updater.py +++ b/frigate/comms/recordings_updater.py @@ -18,7 +18,10 @@ class RecordingsDataTypeEnum(str, Enum): class RecordingsDataPublisher(Publisher[Any]): - """Publishes latest recording data.""" + """Publishes latest recording data. + + Payloads are (camera, stream_type, timestamp, cache_path) on every topic. + """ topic_base = "recordings/" diff --git a/frigate/config/camera/camera.py b/frigate/config/camera/camera.py index a278c5af8d..0ef6f3664a 100644 --- a/frigate/config/camera/camera.py +++ b/frigate/config/camera/camera.py @@ -220,16 +220,21 @@ class CameraConfig(FrigateBaseModel): # add roles to the input if there is only one if len(config["ffmpeg"]["inputs"]) == 1: - has_audio = "audio" in config["ffmpeg"]["inputs"][0].get("roles", []) + existing_roles = config["ffmpeg"]["inputs"][0].get("roles", []) config["ffmpeg"]["inputs"][0]["roles"] = [ "record", "detect", ] - if has_audio: + if "audio" in existing_roles: config["ffmpeg"]["inputs"][0]["roles"].append("audio") + # kept so role validation can report the real problem rather than + # claiming the role was never assigned + if "record_sub" in existing_roles: + config["ffmpeg"]["inputs"][0]["roles"].append("record_sub") + super().__init__(**config) @property diff --git a/frigate/config/camera/record.py b/frigate/config/camera/record.py index 341299113d..d9453257d2 100644 --- a/frigate/config/camera/record.py +++ b/frigate/config/camera/record.py @@ -2,7 +2,7 @@ from enum import Enum from pydantic import Field -from frigate.const import MAX_PRE_CAPTURE +from frigate.const import MAX_PRE_CAPTURE, STREAM_TYPE_SUB from frigate.review.types import SeverityEnum from ..base import FrigateBaseModel @@ -191,6 +191,13 @@ class RecordConfig(FrigateBaseModel): description="Indicates whether recording was enabled in the original static configuration.", ) + def stream_enabled(self, stream_type: str) -> bool: + """Whether the given record stream type should currently be recording.""" + if stream_type == STREAM_TYPE_SUB: + return self.enabled and self.sub.enabled + + return self.enabled + @property def effective_alert_days(self) -> float: """Alert retention extended to the sub stream window when sub is enabled. diff --git a/frigate/config/config.py b/frigate/config/config.py index e0c3af9993..1998364754 100644 --- a/frigate/config/config.py +++ b/frigate/config/config.py @@ -269,6 +269,12 @@ def verify_config_roles(camera_config: CameraConfig) -> None: f"Camera {camera_config.name} has sub stream recording enabled, but record_sub is not assigned to an input." ) + for ffmpeg_input in camera_config.ffmpeg.inputs: + if "record" in ffmpeg_input.roles and "record_sub" in ffmpeg_input.roles: + raise ValueError( + f"Camera {camera_config.name} has record and record_sub assigned to the same input, which would record the same stream twice." + ) + if camera_config.audio.enabled and "audio" not in assigned_roles: raise ValueError( f"Camera {camera_config.name} has audio events enabled, but audio is not assigned to an input." diff --git a/frigate/const.py b/frigate/const.py index 909e21c4a8..d313c64dd4 100644 --- a/frigate/const.py +++ b/frigate/const.py @@ -28,6 +28,9 @@ REDACTED_CREDENTIAL_SENTINEL = "__FRIGATE_SAVED_CREDENTIAL__" STREAM_TYPE_MAIN = "main" STREAM_TYPE_SUB = "sub" SUB_CACHE_TAG = "@sub" +RECORD_STREAM_TYPES = (STREAM_TYPE_MAIN, STREAM_TYPE_SUB) +ROLE_TO_STREAM_TYPE = {"record": STREAM_TYPE_MAIN, "record_sub": STREAM_TYPE_SUB} +STREAM_TYPE_TO_ROLE = {v: k for k, v in ROLE_TO_STREAM_TYPE.items()} # Attribute & Object constants diff --git a/frigate/data_processing/post/review_descriptions.py b/frigate/data_processing/post/review_descriptions.py index 544e823e4a..cbd4e88cc2 100644 --- a/frigate/data_processing/post/review_descriptions.py +++ b/frigate/data_processing/post/review_descriptions.py @@ -23,6 +23,7 @@ from frigate.const import ( ATTRIBUTE_LABEL_DISPLAY_MAP, CACHE_DIR, CLIPS_DIR, + STREAM_TYPE_MAIN, UPDATE_REVIEW_DESCRIPTION, ) from frigate.data_processing.types import PostProcessDataEnum @@ -441,6 +442,7 @@ class ReviewDescriptionProcessor(PostProcessorApi): ) .where((ts >= Recordings.start_time) & (ts <= Recordings.end_time)) .where(Recordings.camera == camera) + .where(Recordings.stream_type == STREAM_TYPE_MAIN) .order_by(Recordings.start_time.desc()) .limit(1) .get() diff --git a/frigate/embeddings/maintainer.py b/frigate/embeddings/maintainer.py index 03309a7ad4..17b09383d9 100644 --- a/frigate/embeddings/maintainer.py +++ b/frigate/embeddings/maintainer.py @@ -714,7 +714,9 @@ class EmbeddingMaintainer(threading.Thread): topic = str(raw_topic) if topic.endswith(RecordingsDataTypeEnum.saved.value): - camera, recordings_available_through_timestamp, _ = payload + camera, _stream_type, recordings_available_through_timestamp, _ = ( + payload + ) self.recordings_available_through[camera] = ( recordings_available_through_timestamp diff --git a/frigate/record/cleanup.py b/frigate/record/cleanup.py index 03e0ceb2f7..c71b366d36 100644 --- a/frigate/record/cleanup.py +++ b/frigate/record/cleanup.py @@ -149,8 +149,12 @@ class RecordingCleanup(threading.Thread): detections_retain_mode: RetainModeEnum, config: CameraConfig, reviews: list[Any], - ) -> set[Path]: - """Delete recordings for existing camera based on retention config.""" + ) -> tuple[set[Path], list[tuple[float, float]]]: + """Delete recordings for one stream of an existing camera based on retention config. + + Returns the directories to check for emptiness and the segments that + were kept, which the caller feeds to expire_camera_previews. + """ # Get the timestamp for cutoff of retained days # Get recordings to check for expiration @@ -257,9 +261,23 @@ class RecordingCleanup(threading.Thread): Recordings.id << deleted_recordings_list[i : i + max_deletes] ).execute() - # previews follow main retention, so only the main pass expires them - if stream_type != STREAM_TYPE_MAIN: - return maybe_empty_dirs + return maybe_empty_dirs, kept_recordings + + def expire_camera_previews( + self, + config: CameraConfig, + continuous_expire_date: float, + motion_expire_date: float, + kept_recordings: list[tuple[float, float]], + ) -> set[Path]: + """Delete previews that no longer have recordings on any stream. + + Previews aren't recorded per stream, so the cutoffs must be the oldest + of the per stream values and kept_recordings must cover every stream, + sorted by start time. Otherwise a short main retention expires previews + the sub recordings still need. + """ + maybe_empty_dirs: set[Path] = set() previews = ( Previews.select( @@ -438,7 +456,7 @@ class RecordingCleanup(threading.Thread): .namedtuples() ) - maybe_empty_dirs |= self.expire_existing_camera_recordings( + main_dirs, main_kept = self.expire_existing_camera_recordings( STREAM_TYPE_MAIN, continuous_expire_date, motion_expire_date, @@ -452,10 +470,11 @@ class RecordingCleanup(threading.Thread): config.record.detections.retain.days, ), ) + maybe_empty_dirs |= main_dirs # runs even when sub recording is disabled so old rows still # expire - maybe_empty_dirs |= self.expire_existing_camera_recordings( + sub_dirs, sub_kept = self.expire_existing_camera_recordings( STREAM_TYPE_SUB, sub_continuous_expire_date, sub_motion_expire_date, @@ -469,6 +488,14 @@ class RecordingCleanup(threading.Thread): config.record.sub.detections.days, ), ) + maybe_empty_dirs |= sub_dirs + + maybe_empty_dirs |= self.expire_camera_previews( + config, + min(continuous_expire_date, sub_continuous_expire_date), + min(motion_expire_date, sub_motion_expire_date), + sorted(main_kept + sub_kept), + ) logger.debug(f"End camera: {camera}.") logger.debug("End all cameras.") diff --git a/frigate/record/maintainer.py b/frigate/record/maintainer.py index c347566b07..6da72bdb0d 100644 --- a/frigate/record/maintainer.py +++ b/frigate/record/maintainer.py @@ -79,6 +79,54 @@ def parse_cache_segment_name(basename: str) -> tuple[str, str, str] | None: return (prefix, STREAM_TYPE_MAIN, date) +def format_segment_details(cache_path: str, segment_info: dict[str, Any]) -> str: + """Comma separated facts about a segment, for discard warnings.""" + details: list[str] = [] + + duration = segment_info.get("duration", -1) + + if duration != -1: + details.append(f"duration: {duration:.2f}s") + + try: + details.append(f"size: {os.path.getsize(cache_path) / 1024:.1f} KB") + except OSError: + pass + + details.append(f"video: {segment_info.get('video_codec') or 'none'}") + + if segment_info.get("has_audio"): + audio = segment_info.get("audio_codec") or "unknown" + rate = segment_info.get("audio_rate") + details.append(f"audio: {audio} {rate}Hz" if rate else f"audio: {audio}") + else: + details.append("audio: none") + + return ", ".join(details) + + +def segment_path_time(cache_path: str) -> datetime.datetime | None: + """Timestamp a segment's recording path is built from, or None if unparsable. + + Recording paths carry one second of resolution, and so does ffmpeg's cache + segment template, which makes a cache file name unique per camera stream + and second. Resolved start times are not: a stream cutting segments faster + than once a second resolves consecutive segments into the same second, and + building the path from those collides on the unique path index. + """ + parsed = parse_cache_segment_name(Path(cache_path).stem) + + if parsed is None: + return None + + try: + return datetime.datetime.strptime(parsed[2], CACHE_SEGMENT_FORMAT).astimezone( + datetime.UTC + ) + except ValueError: + return None + + class SegmentInfo: def __init__( self, @@ -241,51 +289,55 @@ class RecordingMaintainer(threading.Thread): and not d.startswith("preview_") ] - # publish newest cached segment per camera (including in use files) - newest_cache_segments: dict[str, dict[str, Any]] = {} + # publish newest cached segment per camera stream (including in use files) + newest_cache_segments: dict[tuple[str, str], dict[str, Any]] = {} for cache in cache_files: cache_path = os.path.join(CACHE_DIR, cache) basename = os.path.splitext(cache)[0] parsed = parse_cache_segment_name(basename) if parsed is None: if not self.unexpected_cache_files_logged: - logger.warning("Skipping unexpected files in cache") + logger.warning(f"Skipping unexpected files in cache, e.g. {cache}") self.unexpected_cache_files_logged = True continue camera, stream_type, date = parsed - # this topic feeds main-stream health/sync consumers only - if stream_type == STREAM_TYPE_SUB: - continue - start_time = datetime.datetime.strptime( date, CACHE_SEGMENT_FORMAT ).astimezone(datetime.UTC) + key = (camera, stream_type) if ( - camera not in newest_cache_segments - or start_time > newest_cache_segments[camera]["start_time"] + key not in newest_cache_segments + or start_time > newest_cache_segments[key]["start_time"] ): - newest_cache_segments[camera] = { + newest_cache_segments[key] = { "start_time": start_time, "cache_path": cache_path, } - for camera, newest in newest_cache_segments.items(): + for (camera, stream_type), newest in newest_cache_segments.items(): self.recordings_publisher.publish( ( camera, + stream_type, newest["start_time"].timestamp(), newest["cache_path"], ), RecordingsDataTypeEnum.latest.value, ) - # publish None for cameras with no cache files (but only if we know the camera exists) - for camera_name in self.config.cameras: - if camera_name not in newest_cache_segments: - self.recordings_publisher.publish( - (camera_name, None, None), - RecordingsDataTypeEnum.latest.value, - ) + # publish None for streams with no cache files (but only if we know the camera exists) + for camera_name, camera_config in self.config.cameras.items(): + stream_types = [STREAM_TYPE_MAIN] + + if camera_config.record.sub.enabled: + stream_types.append(STREAM_TYPE_SUB) + + for stream_type in stream_types: + if (camera_name, stream_type) not in newest_cache_segments: + self.recordings_publisher.publish( + (camera_name, stream_type, None, None), + RecordingsDataTypeEnum.latest.value, + ) files_in_use = [] for process in psutil.process_iter(): @@ -314,7 +366,7 @@ class RecordingMaintainer(threading.Thread): parsed = parse_cache_segment_name(basename) if parsed is None: if not self.unexpected_cache_files_logged: - logger.warning("Skipping unexpected files in cache") + logger.warning(f"Skipping unexpected files in cache, e.g. {cache}") self.unexpected_cache_files_logged = True continue camera, stream_type, date = parsed @@ -447,6 +499,7 @@ class RecordingMaintainer(threading.Thread): self.recordings_publisher.publish( ( camera, + stream_type, recordings[0]["start_time"].timestamp() if camera_cfg and camera_cfg.record.enabled else None, @@ -540,13 +593,13 @@ class RecordingMaintainer(threading.Thread): if not segment_info.get("has_valid_video", False): logger.warning( - f"Invalid or missing video stream in segment {cache_path}. Discarding." + f"Invalid or missing video stream in segment {cache_path} " + f"({format_segment_details(cache_path, segment_info)}). Discarding." + ) + self.recordings_publisher.publish( + (camera, stream_type, start_time.timestamp(), cache_path), + RecordingsDataTypeEnum.invalid.value, ) - if stream_type == STREAM_TYPE_MAIN: - self.recordings_publisher.publish( - (camera, start_time.timestamp(), cache_path), - RecordingsDataTypeEnum.invalid.value, - ) self.drop_segment(cache_path) return None @@ -583,21 +636,22 @@ class RecordingMaintainer(threading.Thread): if duration == -1: logger.warning(f"Failed to probe corrupt segment {cache_path}") - logger.warning(f"Discarding a corrupt recording segment: {cache_path}") - if stream_type == STREAM_TYPE_MAIN: - self.recordings_publisher.publish( - (camera, start_time.timestamp(), cache_path), - RecordingsDataTypeEnum.invalid.value, - ) + logger.warning( + f"Discarding a corrupt recording segment: {cache_path} " + f"({format_segment_details(cache_path, segment_info)})" + ) + self.recordings_publisher.publish( + (camera, stream_type, start_time.timestamp(), cache_path), + RecordingsDataTypeEnum.invalid.value, + ) self.drop_segment(cache_path) return None # this segment has a valid duration and has video data, so publish an update - if stream_type == STREAM_TYPE_MAIN: - self.recordings_publisher.publish( - (camera, start_time.timestamp(), cache_path), - RecordingsDataTypeEnum.valid.value, - ) + self.recordings_publisher.publish( + (camera, stream_type, start_time.timestamp(), cache_path), + RecordingsDataTypeEnum.valid.value, + ) record_config = self.config.cameras[camera].record @@ -863,18 +917,20 @@ class RecordingMaintainer(threading.Thread): video_codec: str | None = None, keyframes: list[int] | None = None, ) -> dict[str, Any] | None: - # directory will be in utc due to start_time being in utc + path_time = segment_path_time(cache_path) or start_time + + # directory will be in utc due to path_time being in utc # sub segments get a tagged directory to avoid filename collisions directory = os.path.join( RECORD_DIR, - start_time.strftime("%Y-%m-%d/%H"), + path_time.strftime("%Y-%m-%d/%H"), camera if stream_type == STREAM_TYPE_MAIN else f"{camera}{SUB_CACHE_TAG}", ) os.makedirs(directory, exist_ok=True) - # file will be in utc due to start_time being in utc - file_name = f"{start_time.strftime('%M.%S.mp4')}" + # file will be in utc due to path_time being in utc + file_name = f"{path_time.strftime('%M.%S.mp4')}" file_path = os.path.join(directory, file_name) try: @@ -946,10 +1002,9 @@ class RecordingMaintainer(threading.Thread): Recordings.video_codec.name: video_codec, Recordings.keyframes.name: keyframes, } - except Exception as e: - logger.error(f"Unable to store recording segment {cache_path}") + except Exception: + logger.exception(f"Unable to store recording segment {cache_path}") Path(cache_path).unlink(missing_ok=True) - logger.error(e) # clear end_time cache self.end_time_cache.pop(cache_path, None) diff --git a/frigate/test/test_camera_watchdog.py b/frigate/test/test_camera_watchdog.py new file mode 100644 index 0000000000..0167c1aa45 --- /dev/null +++ b/frigate/test/test_camera_watchdog.py @@ -0,0 +1,220 @@ +"""Tests for per stream recording health tracking in the camera watchdog.""" + +import unittest +from datetime import UTC, datetime, timedelta +from unittest.mock import MagicMock, patch + +from frigate.config import FrigateConfig +from frigate.const import STREAM_TYPE_MAIN, STREAM_TYPE_SUB +from frigate.video.ffmpeg import CameraWatchdog + + +class TestCameraWatchdogStreamHealth(unittest.TestCase): + def _build_watchdog( + self, sub_enabled: bool = True, output_args: dict | None = None + ) -> CameraWatchdog: + config = FrigateConfig( + **{ + "mqtt": {"host": "mqtt"}, + "cameras": { + "front_door": { + "ffmpeg": { + "output_args": output_args or {}, + "inputs": [ + { + "path": "rtsp://10.0.0.1:554/video", + "roles": ["record"], + }, + { + "path": "rtsp://10.0.0.1:554/video2", + "roles": ["detect", "record_sub"], + }, + ], + }, + "record": { + "enabled": True, + "sub": {"enabled": sub_enabled}, + }, + } + }, + } + ) + camera_config = config.cameras["front_door"] + + with ( + patch("frigate.video.ffmpeg.LogPipe"), + patch("frigate.video.ffmpeg.InterProcessRequestor"), + patch("frigate.video.ffmpeg.RecordingsDataSubscriber"), + patch("frigate.video.ffmpeg.CameraConfigUpdateSubscriber"), + ): + watchdog = CameraWatchdog( + camera_config, + 1, + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + MagicMock(), + ) + + watchdog.requestor = MagicMock() + return watchdog + + def test_stale_sub_does_not_mark_main_stale(self): + watchdog = self._build_watchdog() + now = datetime.now().astimezone(UTC) + stale = (now - timedelta(hours=1)).timestamp() + + watchdog.latest_cache_segment_time[STREAM_TYPE_MAIN] = now.timestamp() + watchdog.latest_valid_segment_time[STREAM_TYPE_MAIN] = now.timestamp() + watchdog.latest_cache_segment_time[STREAM_TYPE_SUB] = stale + watchdog.latest_valid_segment_time[STREAM_TYPE_SUB] = stale + + assert watchdog._stream_staleness(STREAM_TYPE_MAIN, now) is None + assert watchdog._stream_staleness(STREAM_TYPE_SUB, now) is not None + + def test_stale_main_does_not_mark_sub_stale(self): + watchdog = self._build_watchdog() + now = datetime.now().astimezone(UTC) + stale = (now - timedelta(hours=1)).timestamp() + + watchdog.latest_cache_segment_time[STREAM_TYPE_MAIN] = stale + watchdog.latest_valid_segment_time[STREAM_TYPE_MAIN] = stale + watchdog.latest_cache_segment_time[STREAM_TYPE_SUB] = now.timestamp() + watchdog.latest_valid_segment_time[STREAM_TYPE_SUB] = now.timestamp() + + assert watchdog._stream_staleness(STREAM_TYPE_MAIN, now) is not None + assert watchdog._stream_staleness(STREAM_TYPE_SUB, now) is None + + def test_grace_period_suppresses_staleness(self): + watchdog = self._build_watchdog() + now = datetime.now().astimezone(UTC) + watchdog.record_enable_time = now - timedelta(seconds=10) + watchdog.latest_cache_segment_time[STREAM_TYPE_SUB] = ( + now - timedelta(hours=1) + ).timestamp() + + assert watchdog._stream_staleness(STREAM_TYPE_SUB, now) is None + + def test_status_goes_to_the_matching_role_topic(self): + watchdog = self._build_watchdog() + + watchdog._send_record_status(STREAM_TYPE_MAIN, "online", 100.0) + watchdog._send_record_status(STREAM_TYPE_SUB, "offline", 100.0) + + watchdog.requestor.send_data.assert_any_call( + "front_door/status/record", "online" + ) + watchdog.requestor.send_data.assert_any_call( + "front_door/status/record_sub", "offline" + ) + + def test_status_is_cached_per_stream(self): + watchdog = self._build_watchdog() + + watchdog._send_record_status(STREAM_TYPE_MAIN, "online", 100.0) + watchdog._send_record_status(STREAM_TYPE_SUB, "online", 100.0) + watchdog._send_record_status(STREAM_TYPE_MAIN, "online", 100.0) + + assert watchdog.requestor.send_data.call_count == 2 + + def test_recorded_streams_follows_config(self): + watchdog = self._build_watchdog() + assert watchdog._recorded_streams(["record"]) == [STREAM_TYPE_MAIN] + assert watchdog._recorded_streams(["detect", "record_sub"]) == [STREAM_TYPE_SUB] + assert watchdog._recorded_streams(["detect"]) == [] + + disabled = self._build_watchdog(sub_enabled=False) + assert disabled._recorded_streams(["detect", "record_sub"]) == [] + + def test_restart_grace_suppresses_repeat_staleness(self): + watchdog = self._build_watchdog() + now = datetime.now().astimezone(UTC) + stale = (now - timedelta(hours=1)).timestamp() + + watchdog.latest_cache_segment_time[STREAM_TYPE_MAIN] = stale + watchdog.latest_valid_segment_time[STREAM_TYPE_MAIN] = stale + assert watchdog._stream_staleness(STREAM_TYPE_MAIN, now) is not None + + watchdog._grant_restart_grace([STREAM_TYPE_MAIN], now) + + assert watchdog._stream_staleness(STREAM_TYPE_MAIN, now) is None + assert ( + watchdog._stream_staleness(STREAM_TYPE_MAIN, now + timedelta(seconds=89)) + is None + ) + assert ( + watchdog._stream_staleness(STREAM_TYPE_MAIN, now + timedelta(seconds=91)) + is not None + ) + + def test_restart_grace_is_per_stream(self): + watchdog = self._build_watchdog() + now = datetime.now().astimezone(UTC) + stale = (now - timedelta(hours=1)).timestamp() + + for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB): + watchdog.latest_cache_segment_time[stream_type] = stale + watchdog.latest_valid_segment_time[stream_type] = stale + + watchdog._grant_restart_grace([STREAM_TYPE_MAIN], now) + + assert watchdog._stream_staleness(STREAM_TYPE_MAIN, now) is None + assert watchdog._stream_staleness(STREAM_TYPE_SUB, now) is not None + + def test_detect_reset_grants_the_shared_sub_stream_grace(self): + watchdog = self._build_watchdog() + watchdog.detect_process_records_sub = True + watchdog.ffmpeg_detect_process = MagicMock() + watchdog.capture_thread = MagicMock() + watchdog.capture_thread.is_alive.return_value = False + watchdog.start_ffmpeg_detect = MagicMock() + + now = datetime.now().astimezone(UTC) + stale = (now - timedelta(hours=1)).timestamp() + watchdog.latest_cache_segment_time[STREAM_TYPE_SUB] = stale + watchdog.latest_valid_segment_time[STREAM_TYPE_SUB] = stale + assert watchdog._stream_staleness(STREAM_TYPE_SUB, now) is not None + + watchdog.reset_capture_thread(terminate=False) + + # the sub check runs later in the same tick against a stale can_restart, + # so without this grace it would kill the just-restarted process again + assert ( + watchdog._stream_staleness(STREAM_TYPE_SUB, datetime.now().astimezone(UTC)) + is None + ) + + def test_detect_reset_leaves_sub_alone_when_not_shared(self): + watchdog = self._build_watchdog() + watchdog.detect_process_records_sub = False + watchdog.ffmpeg_detect_process = MagicMock() + watchdog.capture_thread = MagicMock() + watchdog.capture_thread.is_alive.return_value = False + watchdog.start_ffmpeg_detect = MagicMock() + + now = datetime.now().astimezone(UTC) + stale = (now - timedelta(hours=1)).timestamp() + watchdog.latest_cache_segment_time[STREAM_TYPE_SUB] = stale + watchdog.latest_valid_segment_time[STREAM_TYPE_SUB] = stale + + watchdog.reset_capture_thread(terminate=False) + + assert ( + watchdog._stream_staleness(STREAM_TYPE_SUB, datetime.now().astimezone(UTC)) + is not None + ) + + def test_stale_threshold_follows_each_stream_segment_time(self): + watchdog = self._build_watchdog( + output_args={ + "record": "-f segment -segment_time 10 -c copy", + "record_sub": "-f segment -segment_time 60 -c copy", + } + ) + + assert watchdog.record_stale_threshold[STREAM_TYPE_MAIN] == 120 + assert watchdog.record_stale_threshold[STREAM_TYPE_SUB] == 150 diff --git a/frigate/test/test_config.py b/frigate/test/test_config.py index 0eb571fcc3..c1b9531a31 100644 --- a/frigate/test/test_config.py +++ b/frigate/test/test_config.py @@ -1229,6 +1229,36 @@ class TestConfig(unittest.TestCase): lambda: FrigateConfig(**config).cameras, ) + def test_fails_on_record_and_record_sub_on_same_input(self): + config = self._sub_record_config() + config["cameras"]["back"]["ffmpeg"]["inputs"] = [ + { + "path": "rtsp://10.0.0.1:554/video", + "roles": ["detect", "record", "record_sub"], + }, + {"path": "rtsp://10.0.0.1:554/video2", "roles": ["audio"]}, + ] + + self.assertRaisesRegex( + ValueError, + "record and record_sub assigned to the same input", + lambda: FrigateConfig(**config).cameras, + ) + + def test_fails_on_record_sub_with_a_single_input(self): + # the single input case has record forced onto it, so record_sub can + # only ever duplicate that same stream + config = self._sub_record_config() + config["cameras"]["back"]["ffmpeg"]["inputs"] = [ + {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect", "record_sub"]}, + ] + + self.assertRaisesRegex( + ValueError, + "record and record_sub assigned to the same input", + lambda: FrigateConfig(**config).cameras, + ) + def test_record_sub_segment_time_not_checked_when_disabled(self): config = self._sub_record_config( { diff --git a/frigate/test/test_dispatcher_recordings_insert.py b/frigate/test/test_dispatcher_recordings_insert.py new file mode 100644 index 0000000000..c7ffcaa242 --- /dev/null +++ b/frigate/test/test_dispatcher_recordings_insert.py @@ -0,0 +1,67 @@ +"""Tests for the recordings batch insert handler.""" + +import unittest +from unittest.mock import MagicMock, patch + +from playhouse.sqlite_ext import SqliteExtDatabase + +from frigate.comms.dispatcher import Dispatcher +from frigate.const import INSERT_MANY_RECORDINGS +from frigate.models import Recordings + + +def _recording(id: str, path: str) -> dict: + return { + Recordings.id.name: id, + Recordings.camera.name: "front_door", + Recordings.stream_type.name: "main", + Recordings.path.name: path, + Recordings.start_time.name: 1000.0, + Recordings.end_time.name: 1010.0, + Recordings.duration.name: 10.0, + Recordings.motion.name: 0, + Recordings.objects.name: 0, + Recordings.dBFS.name: 0, + Recordings.segment_size.name: 1.0, + } + + +class TestInsertManyRecordings(unittest.TestCase): + """A duplicate path must not cost the rest of the batch.""" + + def setUp(self): + self.db = SqliteExtDatabase(":memory:") + self.db.bind([Recordings]) + self.db.create_tables([Recordings]) + + with ( + patch("frigate.comms.dispatcher.CameraActivityManager"), + patch("frigate.comms.dispatcher.AudioActivityManager"), + ): + self.dispatcher = Dispatcher(MagicMock(), MagicMock(), MagicMock(), {}, []) + + def tearDown(self): + self.db.close() + + def test_batch_with_duplicate_keeps_the_other_rows(self): + Recordings.insert(_recording("existing", "/rec/00.10.mp4")).execute() + + self.dispatcher._receive( + INSERT_MANY_RECORDINGS, + [ + _recording("a", "/rec/00.20.mp4"), + _recording("b", "/rec/00.10.mp4"), + _recording("c", "/rec/00.30.mp4"), + ], + ) + + paths = {r.path for r in Recordings.select()} + self.assertEqual(paths, {"/rec/00.10.mp4", "/rec/00.20.mp4", "/rec/00.30.mp4"}) + + def test_clean_batch_inserts_every_row(self): + self.dispatcher._receive( + INSERT_MANY_RECORDINGS, + [_recording("a", "/rec/00.20.mp4"), _recording("b", "/rec/00.30.mp4")], + ) + + self.assertEqual(Recordings.select().count(), 2) diff --git a/frigate/test/test_record_cleanup_sub.py b/frigate/test/test_record_cleanup_sub.py index edbae0f8e3..b5425e8cc2 100644 --- a/frigate/test/test_record_cleanup_sub.py +++ b/frigate/test/test_record_cleanup_sub.py @@ -73,6 +73,21 @@ class TestRecordingCleanupSubRetention(unittest.TestCase): stream_type=stream_type, ) + def _insert_preview( + self, id: str, age_days: float, camera: str = "front_door" + ) -> None: + end_time = ( + datetime.datetime.now() - datetime.timedelta(days=age_days) + ).timestamp() + Previews.create( + id=id, + camera=camera, + path=f"/media/frigate/previews/{id}.mp4", + start_time=end_time - 10, + end_time=end_time, + duration=10, + ) + def test_sub_recordings_expire_independently(self): # main retention 7 days, sub retention 30 days; rows 10 days old # -> main row deleted, sub row kept @@ -91,6 +106,60 @@ class TestRecordingCleanupSubRetention(unittest.TestCase): assert Recordings.get_or_none(Recordings.id == "m1") is None assert Recordings.get_or_none(Recordings.id == "s1") is not None + def test_previews_survive_while_sub_recordings_remain(self): + # main retention 7 days, sub retention 30 days; only the sub row + # survives at 10 days, and the preview covering it must survive too + cleanup = self._build_cleanup( + { + "enabled": True, + "continuous": {"days": 7}, + "sub": {"enabled": True, "continuous": {"days": 30}}, + } + ) + self._insert_recording("m1", "main", 10) + self._insert_recording("s1", "sub", 10) + self._insert_preview("p1", 10) + + cleanup.expire_recordings() + + assert Recordings.get_or_none(Recordings.id == "m1") is None + assert Previews.get_or_none(Previews.id == "p1") is not None + + def test_previews_expire_once_every_stream_has(self): + # both streams expired at 40 days -> the preview goes with them + cleanup = self._build_cleanup( + { + "enabled": True, + "continuous": {"days": 7}, + "sub": {"enabled": True, "continuous": {"days": 30}}, + } + ) + self._insert_recording("m1", "main", 40) + self._insert_recording("s1", "sub", 40) + self._insert_preview("p1", 40) + + cleanup.expire_recordings() + + assert Recordings.get_or_none(Recordings.id == "s1") is None + assert Previews.get_or_none(Previews.id == "p1") is None + + def test_preview_retention_unchanged_when_sub_disabled(self): + cleanup = self._build_cleanup( + { + "enabled": True, + "continuous": {"days": 7}, + "sub": {"enabled": False}, + } + ) + self._insert_recording("m1", "main", 10) + self._insert_preview("p_old", 10) + self._insert_preview("p_new", 1) + + cleanup.expire_recordings() + + assert Previews.get_or_none(Previews.id == "p_old") is None + assert Previews.get_or_none(Previews.id == "p_new") is not None + def test_sub_recordings_expire_after_sub_retention(self): # sub retention 30 days; sub row 40 days old -> deleted cleanup = self._build_cleanup( diff --git a/frigate/test/test_record_sub_maintainer.py b/frigate/test/test_record_sub_maintainer.py index 070e2f4680..b60a69a63b 100644 --- a/frigate/test/test_record_sub_maintainer.py +++ b/frigate/test/test_record_sub_maintainer.py @@ -15,6 +15,7 @@ from frigate.record.maintainer import ( RecordingMaintainer, SegmentInfo, parse_cache_segment_name, + segment_path_time, ) @@ -349,6 +350,98 @@ class TestSegmentAudioPresence(unittest.IsolatedAsyncioTestCase): self.assertEqual(result[Recordings.video_codec.name], video_codec) +class TestSegmentPathTime(unittest.IsolatedAsyncioTestCase): + """The recording path must stay unique when segments are shorter than a second.""" + + def _build_maintainer(self) -> RecordingMaintainer: + camera_config = MagicMock() + camera_config.record.enabled = True + camera_config.record.continuous.days = 1 + camera_config.record.motion.days = 0 + + config = MagicMock() + config.cameras = {"test_cam": camera_config} + + maintainer = RecordingMaintainer.__new__(RecordingMaintainer) + maintainer.config = config + maintainer.end_time_cache = {} + maintainer.object_recordings_info = defaultdict(list) + maintainer.audio_recordings_info = defaultdict(list) + maintainer.recordings_publisher = MagicMock() + maintainer.last_segment_end = {("test_cam", "main"): 0.0} + return maintainer + + def test_parses_main_and_sub_names(self): + expected = datetime.datetime(2026, 6, 10, 14, 30, 22, tzinfo=datetime.UTC) + self.assertEqual( + segment_path_time("/tmp/cache/test_cam@20260610143022+0000.mp4"), expected + ) + self.assertEqual( + segment_path_time("/tmp/cache/test_cam@sub@20260610143022+0000.mp4"), + expected, + ) + + def test_returns_none_for_unparsable_names(self): + self.assertIsNone(segment_path_time("/tmp/cache/garbage.mp4")) + self.assertIsNone(segment_path_time("/tmp/cache/test_cam@notadate.mp4")) + + async def test_sub_second_segments_get_distinct_paths(self): + # two cache files a second apart whose resolved starts both land in + # second 22; deriving the path from the resolved start collides + segments = [ + ("test_cam@20260610143022+0000.mp4", 100_000), + ("test_cam@20260610143023+0000.mp4", 980_000), + ] + paths = [] + + with tempfile.TemporaryDirectory() as tmpdir: + for name, microsecond in segments: + maintainer = self._build_maintainer() + maintainer.config.ffmpeg.ffmpeg_path = "ffmpeg" + + start_time = datetime.datetime( + 2026, 6, 10, 14, 30, 22, microsecond, tzinfo=datetime.UTC + ) + cache_path = os.path.join(tmpdir, name) + with open(cache_path, "wb") as f: + f.write(b"\x00" * 16) + + proc = MagicMock() + proc.returncode = 0 + proc.wait = AsyncMock(return_value=0) + + with ( + patch( + "frigate.record.maintainer.RECORD_DIR", + os.path.join(tmpdir, "recordings"), + ), + patch( + "frigate.record.maintainer.asyncio.create_subprocess_exec", + AsyncMock(return_value=proc), + ), + ): + result = await maintainer.move_segment( + "test_cam", + "main", + start_time, + start_time + datetime.timedelta(seconds=0.96), + 0.96, + cache_path, + SegmentInfo(0, 0, 0, 0), + ) + + self.assertIsNotNone(result) + paths.append(result[Recordings.path.name]) + # the row keeps the resolved start even though the path doesn't + self.assertEqual( + result[Recordings.start_time.name], start_time.timestamp() + ) + + self.assertEqual(len(set(paths)), 2, paths) + self.assertTrue(paths[0].endswith("30.22.mp4"), paths[0]) + self.assertTrue(paths[1].endswith("30.23.mp4"), paths[1]) + + class TestSegmentStartChaining(unittest.IsolatedAsyncioTestCase): """Contiguous segments must chain start times across filename truncation. diff --git a/frigate/util/builtin.py b/frigate/util/builtin.py index 6bf5c6f380..fbc53e2cd9 100644 --- a/frigate/util/builtin.py +++ b/frigate/util/builtin.py @@ -20,7 +20,12 @@ from typing import TYPE_CHECKING, Any import numpy as np from ruamel.yaml import YAML -from frigate.const import REGEX_HTTP_CAMERA_USER_PASS, REGEX_RTSP_CAMERA_USER_PASS +from frigate.const import ( + REGEX_HTTP_CAMERA_USER_PASS, + REGEX_RTSP_CAMERA_USER_PASS, + STREAM_TYPE_MAIN, + STREAM_TYPE_SUB, +) if TYPE_CHECKING: from frigate.config import CameraConfig @@ -137,9 +142,16 @@ def get_ffmpeg_arg_list(arg: Any) -> list: DEFAULT_RECORD_SEGMENT_TIME = 10 -def get_record_segment_time(config: "CameraConfig") -> int: - """Extract -segment_time from the camera's record output args.""" - record_args = get_ffmpeg_arg_list(config.ffmpeg.output_args.record) +def get_record_segment_time( + config: "CameraConfig", stream_type: str = STREAM_TYPE_MAIN +) -> int: + """Extract -segment_time from the camera's record output args for a stream.""" + output_args = ( + config.ffmpeg.output_args.effective_record_sub + if stream_type == STREAM_TYPE_SUB + else config.ffmpeg.output_args.record + ) + record_args = get_ffmpeg_arg_list(output_args) if record_args and record_args[0].startswith("preset"): return DEFAULT_RECORD_SEGMENT_TIME diff --git a/frigate/video/ffmpeg.py b/frigate/video/ffmpeg.py index d4645efc78..c55c127498 100644 --- a/frigate/video/ffmpeg.py +++ b/frigate/video/ffmpeg.py @@ -5,7 +5,7 @@ import queue import subprocess as sp import threading import time -from collections import deque +from collections import defaultdict, deque from datetime import UTC, datetime, timedelta from multiprocessing import Queue, Value from multiprocessing.synchronize import Event as MpEvent @@ -22,7 +22,14 @@ from frigate.config.camera.updater import ( CameraConfigUpdateEnum, CameraConfigUpdateSubscriber, ) -from frigate.const import PROCESS_PRIORITY_HIGH +from frigate.const import ( + PROCESS_PRIORITY_HIGH, + RECORD_STREAM_TYPES, + ROLE_TO_STREAM_TYPE, + STREAM_TYPE_MAIN, + STREAM_TYPE_SUB, + STREAM_TYPE_TO_ROLE, +) from frigate.log import LogPipe from frigate.util.builtin import EventsPerSecond, get_record_segment_time from frigate.util.ffmpeg import start_or_restart_ffmpeg, stop_ffmpeg @@ -34,6 +41,8 @@ from frigate.util.process import FrigateProcess logger = logging.getLogger(__name__) +RECORD_GRACE_SECONDS = 90 + def capture_frames( ffmpeg_process: sp.Popen[Any], @@ -150,16 +159,26 @@ class CameraWatchdog(threading.Thread): self.was_record_sub_enabled = self.config.record.sub.enabled self.segment_subscriber = RecordingsDataSubscriber(RecordingsDataTypeEnum.all) - self.latest_valid_segment_time: float = 0 - self.latest_invalid_segment_time: float = 0 - self.latest_cache_segment_time: float = 0 + self.latest_valid_segment_time: dict[str, float] = defaultdict(float) + self.latest_invalid_segment_time: dict[str, float] = defaultdict(float) + self.latest_cache_segment_time: dict[str, float] = defaultdict(float) self.record_enable_time: datetime | None = None + self.stream_grace_until: dict[str, datetime] = {} # `valid` segments are published with the segment's start time, so the # gap between consecutive publishes can reach 2 * segment_time. Pad the # staleness threshold so it's never tighter than that worst case. - segment_time = get_record_segment_time(self.config) - self.record_stale_threshold = max(120, 2 * segment_time + 30) + self.record_stale_threshold: dict[str, int] = { + stream_type: max( + 120, 2 * get_record_segment_time(self.config, stream_type) + 30 + ) + for stream_type in RECORD_STREAM_TYPES + } + + # the sub stream usually shares its input, and therefore its ffmpeg + # process, with detect, so it isn't in ffmpeg_other_processes and needs + # its own staleness check + self.detect_process_records_sub = False # Stall tracking (based on last processed frame) self._stall_timestamps: deque[float] = deque() @@ -167,7 +186,7 @@ class CameraWatchdog(threading.Thread): # Status caching to reduce message volume self._last_detect_status: str | None = None - self._last_record_status: str | None = None + self._last_record_status: dict[str, str] = {} self._last_status_update_time: float = 0.0 def _send_detect_status(self, status: str, now: float) -> None: @@ -180,16 +199,78 @@ class CameraWatchdog(threading.Thread): self._last_detect_status = status self._last_status_update_time = now - def _send_record_status(self, status: str, now: float) -> None: - """Send record status only if changed or retry_interval has elapsed.""" + def _send_record_status(self, stream_type: str, status: str, now: float) -> None: + """Send a record stream's status only if changed or retry_interval has elapsed.""" if ( - status != self._last_record_status + status != self._last_record_status.get(stream_type) or (now - self._last_status_update_time) >= self.sleeptime ): - self.requestor.send_data(f"{self.config.name}/status/record", status) - self._last_record_status = status + self.requestor.send_data( + f"{self.config.name}/status/{STREAM_TYPE_TO_ROLE[stream_type]}", status + ) + self._last_record_status[stream_type] = status self._last_status_update_time = now + def _reset_segment_times(self) -> None: + self.latest_valid_segment_time.clear() + self.latest_invalid_segment_time.clear() + self.latest_cache_segment_time.clear() + self.stream_grace_until.clear() + + def _grant_restart_grace(self, stream_types: list[str], now_utc: datetime) -> None: + for stream_type in stream_types: + self.stream_grace_until[stream_type] = now_utc + timedelta( + seconds=RECORD_GRACE_SECONDS + ) + + def _stream_staleness(self, stream_type: str, now_utc: datetime) -> str | None: + """Return why the stream's segments are stale, or None if they're healthy.""" + # ffmpeg needs time to create a first segment after recording is + # enabled and after a restart, per stream + in_grace_period = ( + self.record_enable_time is not None + and (now_utc - self.record_enable_time) + < timedelta(seconds=RECORD_GRACE_SECONDS) + ) or now_utc < self.stream_grace_until.get(stream_type, now_utc) + + if in_grace_period: + return None + + latest_cache = self.latest_cache_segment_time[stream_type] + latest_valid = self.latest_valid_segment_time[stream_type] + latest_invalid = self.latest_invalid_segment_time[stream_type] + + def as_dt(timestamp: float) -> datetime: + if timestamp > 0: + return datetime.fromtimestamp(timestamp, tz=UTC) + + return now_utc - timedelta(seconds=1) + + stale_window = timedelta(seconds=self.record_stale_threshold[stream_type]) + + if now_utc > (as_dt(latest_cache) + stale_window): + return "No new recording segments were created" + + if now_utc > (as_dt(latest_valid) + stale_window): + return "No new valid recording segments were created" + + if ( + latest_invalid > 0 + and now_utc > (as_dt(latest_invalid) + stale_window) + and latest_valid <= latest_invalid + ): + return "No valid segments created since last invalid segment" + + return None + + def _recorded_streams(self, roles: list[Any]) -> list[str]: + """Record stream types the given roles cover that are currently recording.""" + return [ + stream_type + for role, stream_type in ROLE_TO_STREAM_TYPE.items() + if role in roles and self.config.record.stream_enabled(stream_type) + ] + def _check_config_updates(self) -> dict[str, list[str]]: """Check for config updates and return the update dict.""" return self.config_subscriber.check_for_updates() @@ -245,6 +326,11 @@ class CameraWatchdog(threading.Thread): self.logger.info("Restarting ffmpeg...") self.start_ffmpeg_detect() + # this process produces the sub stream's segments too, so it gets the + # same startup grace however the reset was triggered + if self.detect_process_records_sub: + self._grant_restart_grace([STREAM_TYPE_SUB], datetime.now().astimezone(UTC)) + def run(self) -> None: if self._update_enabled_state(): self.start_all_ffmpeg() @@ -267,9 +353,7 @@ class CameraWatchdog(threading.Thread): ) self.stop_all_ffmpeg() self.start_all_ffmpeg() - self.latest_valid_segment_time = 0 - self.latest_invalid_segment_time = 0 - self.latest_cache_segment_time = 0 + self._reset_segment_times() self.record_enable_time = datetime.now().astimezone(UTC) last_restart_time = datetime.now().timestamp() continue @@ -281,9 +365,7 @@ class CameraWatchdog(threading.Thread): self.start_all_ffmpeg() # reset all timestamps and record the enable time for grace period - self.latest_valid_segment_time = 0 - self.latest_invalid_segment_time = 0 - self.latest_cache_segment_time = 0 + self._reset_segment_times() self.record_enable_time = datetime.now().astimezone(UTC) else: self.logger.debug(f"Disabling camera {self.config.name}") @@ -293,7 +375,10 @@ class CameraWatchdog(threading.Thread): # update camera status now = datetime.now().timestamp() self._send_detect_status("disabled", now) - self._send_record_status("disabled", now) + self._send_record_status(STREAM_TYPE_MAIN, "disabled", now) + # cameras without a sub stream never get a record_sub topic + if self.config.record.sub.enabled: + self._send_record_status(STREAM_TYPE_SUB, "disabled", now) self.was_enabled = enabled continue @@ -305,9 +390,7 @@ class CameraWatchdog(threading.Thread): ) self.stop_all_ffmpeg() self.start_all_ffmpeg() - self.latest_valid_segment_time = 0 - self.latest_invalid_segment_time = 0 - self.latest_cache_segment_time = 0 + self._reset_segment_times() self.record_enable_time = datetime.now().astimezone(UTC) last_restart_time = datetime.now().timestamp() self.was_record_enabled_in_config = record_enabled_in_config @@ -323,9 +406,7 @@ class CameraWatchdog(threading.Thread): ) self.stop_all_ffmpeg() self.start_all_ffmpeg() - self.latest_valid_segment_time = 0 - self.latest_invalid_segment_time = 0 - self.latest_cache_segment_time = 0 + self._reset_segment_times() self.record_enable_time = datetime.now().astimezone(UTC) last_restart_time = datetime.now().timestamp() self.was_record_sub_enabled = record_sub_enabled @@ -343,26 +424,25 @@ class CameraWatchdog(threading.Thread): raw_topic, payload = update if raw_topic and payload: topic = str(raw_topic) - camera, segment_time, _ = payload + camera, stream_type, segment_time, _ = payload if camera != self.config.name: continue if topic.endswith(RecordingsDataTypeEnum.invalid.value): self.logger.warning( - f"Invalid recording segment detected for {camera} at {segment_time}" + f"Invalid recording segment detected for {camera} ({stream_type}) at {segment_time}" ) - self.latest_invalid_segment_time = segment_time + self.latest_invalid_segment_time[stream_type] = segment_time elif topic.endswith(RecordingsDataTypeEnum.valid.value): self.logger.debug( - f"Latest valid recording segment time on {camera}: {segment_time}" + f"Latest valid recording segment time on {camera} ({stream_type}): {segment_time}" ) - self.latest_valid_segment_time = segment_time + self.latest_valid_segment_time[stream_type] = segment_time elif topic.endswith(RecordingsDataTypeEnum.latest.value): - if segment_time is not None: - self.latest_cache_segment_time = segment_time - else: - self.latest_cache_segment_time = 0 + self.latest_cache_segment_time[stream_type] = ( + segment_time if segment_time is not None else 0 + ) now = datetime.now().timestamp() @@ -409,63 +489,26 @@ class CameraWatchdog(threading.Thread): for p in self.ffmpeg_other_processes: poll = p["process"].poll() - if self.config.record.enabled and "record" in p["roles"]: + recorded_streams = self._recorded_streams(p["roles"]) + + if recorded_streams: now_utc = datetime.now().astimezone(UTC) - # Check if we're within the grace period after enabling recording - # Grace period: 90 seconds allows time for ffmpeg to start and create first segment - in_grace_period = self.record_enable_time is not None and ( - now_utc - self.record_enable_time - ) < timedelta(seconds=90) + # ensure segments are still being created and that they have + # valid video data. each stream is tracked separately so a + # healthy one can't mask a stalled one. + stale_stream = None + stale_reason = None + for stream_type in recorded_streams: + stale_reason = self._stream_staleness(stream_type, now_utc) - latest_cache_dt = ( - datetime.fromtimestamp(self.latest_cache_segment_time, tz=UTC) - if self.latest_cache_segment_time > 0 - else now_utc - timedelta(seconds=1) - ) - - latest_valid_dt = ( - datetime.fromtimestamp(self.latest_valid_segment_time, tz=UTC) - if self.latest_valid_segment_time > 0 - else now_utc - timedelta(seconds=1) - ) - - latest_invalid_dt = ( - datetime.fromtimestamp(self.latest_invalid_segment_time, tz=UTC) - if self.latest_invalid_segment_time > 0 - else now_utc - timedelta(seconds=1) - ) - - # ensure segments are still being created and that they have valid video data - # Skip checks during grace period to allow segments to start being created - stale_window = timedelta(seconds=self.record_stale_threshold) - cache_stale = not in_grace_period and now_utc > ( - latest_cache_dt + stale_window - ) - valid_stale = not in_grace_period and now_utc > ( - latest_valid_dt + stale_window - ) - invalid_stale_condition = ( - self.latest_invalid_segment_time > 0 - and not in_grace_period - and now_utc > (latest_invalid_dt + stale_window) - and self.latest_valid_segment_time - <= self.latest_invalid_segment_time - ) - invalid_stale = invalid_stale_condition - - if cache_stale or valid_stale or invalid_stale: - if cache_stale: - reason = "No new recording segments were created" - elif valid_stale: - reason = "No new valid recording segments were created" - else: # invalid_stale - reason = ( - "No valid segments created since last invalid segment" - ) + if stale_reason is not None: + stale_stream = stream_type + break + if stale_stream is not None and can_restart: self.logger.error( - f"{reason} for {self.config.name} in the last {self.record_stale_threshold}s. Restarting the ffmpeg record process..." + f"{stale_reason} for {self.config.name} ({stale_stream}) in the last {self.record_stale_threshold[stale_stream]}s. Restarting the ffmpeg record process..." ) p["process"] = start_or_restart_ffmpeg( p["cmd"], @@ -479,10 +522,18 @@ class CameraWatchdog(threading.Thread): f"{self.config.name}/status/{role.value}", "offline" ) + self._grant_restart_grace(recorded_streams, now_utc) + last_restart_time = now + continue - else: - self._send_record_status("online", now) - p["latest_segment_time"] = self.latest_cache_segment_time + elif stale_stream is None: + for stream_type in recorded_streams: + self._send_record_status(stream_type, "online", now) + + p["latest_segment_time"] = max( + self.latest_cache_segment_time[stream_type] + for stream_type in recorded_streams + ) if poll is None: continue @@ -497,6 +548,25 @@ class CameraWatchdog(threading.Thread): p["cmd"], self.logger, p["logpipe"], ffmpeg_process=p["process"] ) + if ( + self.detect_process_records_sub + and self.config.record.stream_enabled(STREAM_TYPE_SUB) + and self.capture_thread is not None + and self.capture_thread.is_alive() + ): + now_utc = datetime.now().astimezone(UTC) + stale_reason = self._stream_staleness(STREAM_TYPE_SUB, now_utc) + + if stale_reason is None: + self._send_record_status(STREAM_TYPE_SUB, "online", now) + elif can_restart: + self.logger.error( + f"{stale_reason} for {self.config.name} (sub, shared with detect) in the last {self.record_stale_threshold[STREAM_TYPE_SUB]}s. Restarting ffmpeg..." + ) + self._send_record_status(STREAM_TYPE_SUB, "offline", now) + self.reset_capture_thread() + last_restart_time = now + # Prune expired reconnect timestamps now = datetime.now().timestamp() while ( @@ -539,9 +609,9 @@ class CameraWatchdog(threading.Thread): self.segment_subscriber.stop() def start_ffmpeg_detect(self): - ffmpeg_cmd = [ - c["cmd"] for c in self.config.ffmpeg_cmds if "detect" in c["roles"] - ][0] + detect_cmd = [c for c in self.config.ffmpeg_cmds if "detect" in c["roles"]][0] + ffmpeg_cmd = detect_cmd["cmd"] + self.detect_process_records_sub = "record_sub" in detect_cmd["roles"] self.ffmpeg_detect_process = start_or_restart_ffmpeg( ffmpeg_cmd, self.logger, self.logpipe, self.frame_size )