administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-01 20:36:51 +03:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Add sub stream recording with adaptive quality playback (#24009)

Browse Source 
* add sub stream recording with adaptive quality playback

Optionally record a second, lower bitrate stream alongside the main
recording stream via a `record_sub` input role and `record.sub` config block, with its own retention windows.
Recordings rows now carry the stream type plus the media details needed to serve both streams from one manifest: video codec, audio presence, audio codec and rate, and a record-time keyframe index.

Playback resolves coverage across both streams and merges them into a single VOD sequence, falling back to a discontinuity manifest with per-clip init segments when the media signatures differ. The player exposes a quality selector, and an auto governor picks the stream from stall time, bandwidth, codec support, and the save-data hint.

* fix tests and i18n
This commit is contained in:
Josh Hawkins
2026-09-12 07:30:04 -06:00
committed by Nicolas Mowen
parent f7c5500ea8
commit 1498231eb9
68 changed files with 6794 additions and 602 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/api/media.py
+243 -125
View File
@@ -8,6 +8,7 @@ import os
import subprocess as sp
import time
from datetime import UTC, datetime, timedelta
from enum import Enum
from pathlib import Path as FilePath
from typing import Any
from urllib.parse import unquote
@@ -39,8 +40,9 @@ from frigate.config.camera.snapshots import SnapshotsConfig
from frigate.const import (
CACHE_DIR,
INSTALL_DIR,
MAX_SEGMENT_DURATION,
PREVIEW_FRAME_TYPE,
STREAM_TYPE_MAIN,
STREAM_TYPE_SUB,
)
from frigate.models import Event, Previews, Recordings, Regions, ReviewSegment
from frigate.output.preview import get_most_recent_preview_frame
@@ -52,12 +54,34 @@ from frigate.util.file import (
load_event_snapshot_image,
)
from frigate.util.image import get_image_from_recording, get_image_quality_params
from frigate.util.media import get_keyframe_before
from frigate.util.object import create_empty_regions_grid
from frigate.util.recording_coverage import (
build_spans,
null_audio_glitches,
plan_clip,
resolve_coverage,
stream_has_audio,
)
logger = logging.getLogger(__name__)
# must match the patched MAX_CLIPS in docker/main/build_nginx.sh; a
# normal hour needs ~360, one clip per recording file
NGINX_VOD_MAX_CLIPS = 1080
class VodStreamPreference(str, Enum):
"""Stream pin for the path-segment VOD route.
nginx-vod derives its mapping fetch URI from the playlist URL path
(query params are dropped), so the preference must be a path segment.
"""
main = STREAM_TYPE_MAIN
sub = STREAM_TYPE_SUB
router = APIRouter(tags=[Tags.media])
@@ -319,7 +343,7 @@ async def get_snapshot_from_recording(
& (frame_time <= Recordings.end_time)
)
.where(Recordings.camera == camera_name)
.order_by(Recordings.start_time.desc())
.order_by(Recordings.stream_type.asc(), Recordings.start_time.desc())
.limit(1)
.get()
)
@@ -338,7 +362,7 @@ async def get_snapshot_from_recording(
& (frame_time <= Recordings.end_time)
)
.where(Recordings.camera == camera_name)
.order_by(Recordings.start_time.desc())
.order_by(Recordings.stream_type.asc(), Recordings.start_time.desc())
.limit(1)
.get()
)
@@ -398,7 +422,7 @@ async def submit_recording_snapshot_to_plus(
(frame_time >= Recordings.start_time) & (frame_time <= Recordings.end_time)
)
.where(Recordings.camera == camera_name)
.order_by(Recordings.start_time.desc())
.order_by(Recordings.stream_type.asc(), Recordings.start_time.desc())
.limit(1)
)
@@ -472,20 +496,29 @@ async def recording_clip(
FilePath(file_path).unlink(missing_ok=True)
break
recordings = (
Recordings.select(
Recordings.path,
Recordings.start_time,
Recordings.end_time,
def get_clip_query(stream_type: str):
return (
Recordings.select(
Recordings.path,
Recordings.start_time,
Recordings.end_time,
)
.where(
(Recordings.start_time.between(start_ts, end_ts))
| (Recordings.end_time.between(start_ts, end_ts))
| ((start_ts > Recordings.start_time) & (end_ts < Recordings.end_time))
)
.where(Recordings.camera == camera_name)
.where(Recordings.stream_type == stream_type)
.order_by(Recordings.start_time.asc())
)
.where(
(Recordings.start_time.between(start_ts, end_ts))
| (Recordings.end_time.between(start_ts, end_ts))
| ((start_ts > Recordings.start_time) & (end_ts < Recordings.end_time))
)
.where(Recordings.camera == camera_name)
.order_by(Recordings.start_time.asc())
)
# never mix streams in one concat; use main when available and
# fall back to sub for expired-main history
recordings = get_clip_query(STREAM_TYPE_MAIN)
if recordings.count() == 0:
recordings = get_clip_query(STREAM_TYPE_SUB)
if recordings.count() == 0:
return JSONResponse(
@@ -549,17 +582,52 @@ async def recording_clip(
)
@router.get(
"/vod/{camera_name}/start/{start_ts}/end/{end_ts}",
dependencies=[Depends(require_camera_access)],
description="Returns an HLS playlist for the specified timestamp-range on the specified camera. Append /master.m3u8 or /index.m3u8 for HLS playback.",
)
async def vod_ts(
def _build_vod_clip(
row: Any, start: float, end: float
) -> tuple[dict[str, Any], int] | None:
"""Build one nginx-vod clip dict + duration (ms) for a recording row trimmed to [start, end).
Realization comes entirely from the shared plan_clip, so the coverage
endpoint's realized timelines match this manifest by construction.
"""
plan = plan_clip(row, start, end)
if plan.skipped:
return None
clip: dict[str, Any] = {"type": "source", "path": row.path}
if plan.clip_from_ms is not None:
clip["clipFrom"] = plan.clip_from_ms
clip["keyFrameDurations"] = [plan.duration_ms]
logger.debug(
"VOD: added clip %s duration_ms=%s clipFrom=%s",
row.path,
plan.duration_ms,
clip.get("clipFrom"),
)
return clip, plan.duration_ms
async def _vod_response(
camera_name: str,
start_ts: float,
end_ts: float,
force_discontinuity: bool = False,
):
stream_preference: str | None = None,
) -> JSONResponse:
"""Build an nginx-vod mapping JSON for a camera over a timestamp range.
Always a single-sequence mapping; quality selection happens in the
frontend by choosing between this route and the stream-pinned routes.
Args:
camera_name: The camera to build the mapping for
start_ts: Range start as a unix timestamp
end_ts: Range end as a unix timestamp
force_discontinuity: Emit HLS discontinuity markers between clips
stream_preference: Pin the manifest to one stream type ("main" or
"sub"), serving only that stream's recordings
"""
logger.debug(
"VOD: Generating VOD for %s from %s to %s with force_discontinuity=%s",
camera_name,
@@ -567,104 +635,85 @@ async def vod_ts(
end_ts,
force_discontinuity,
)
recordings = (
Recordings.select(
Recordings.path,
Recordings.duration,
Recordings.end_time,
Recordings.start_time,
)
.where(
Recordings.camera == camera_name,
Recordings.start_time >= start_ts - MAX_SEGMENT_DURATION,
Recordings.start_time <= end_ts,
Recordings.end_time >= start_ts,
)
.order_by(Recordings.start_time.asc())
.iterator()
intervals = resolve_coverage(camera_name, start_ts, end_ts)
# rows contradicting their stream's audio composition are
# truncated-shutdown glitches
main_audio = stream_has_audio(intervals, main=True)
sub_audio = stream_has_audio(intervals, main=False)
spans = build_spans(
null_audio_glitches(intervals, main_audio, sub_audio),
stream_preference,
)
clips = []
durations = []
min_duration_ms = 100 # Minimum 100ms to ensure at least one video frame
max_duration_ms = MAX_SEGMENT_DURATION * 1000
recording: Recordings
for recording in recordings:
durations: list[int] = []
clips: list[dict[str, Any]] = []
# gathered after glitch-nulling and span building, so the policy
# decisions below reflect the manifest's real contents
video_codecs: set[str] = set()
audio_presence: set[bool] = set()
audio_params: set[tuple[str | None, int | None]] = set()
span_streams: set[bool] = set()
for row, span_start, span_end, span_is_main in spans:
logger.debug(
"VOD: processing recording: %s start=%s end=%s duration=%s",
recording.path,
recording.start_time,
recording.end_time,
recording.duration,
row.path,
row.start_time,
row.end_time,
row.duration,
)
built = _build_vod_clip(row, span_start, span_end)
clip = {"type": "source", "path": recording.path}
duration = int(recording.duration * 1000)
# adjust start offset if start_ts is after recording.start_time
if start_ts > recording.start_time:
inpoint = int((start_ts - recording.start_time) * 1000)
clip["clipFrom"] = inpoint
duration -= inpoint
logger.debug(
"VOD: applied clipFrom %sms to %s",
inpoint,
recording.path,
)
# adjust end if recording.end_time is after end_ts
if recording.end_time > end_ts:
duration -= int((recording.end_time - end_ts) * 1000)
# nginx-vod-module pushes clipFrom forward to the next keyframe,
# which can leave too few frames and produce an empty/unplayable
# segment. Snap clipFrom back to the preceding keyframe so the
# segment always starts with a decodable frame.
if "clipFrom" in clip:
keyframe_ms = get_keyframe_before(recording.path, clip["clipFrom"])
if keyframe_ms is not None:
gained = clip["clipFrom"] - keyframe_ms
clip["clipFrom"] = keyframe_ms
duration += gained
logger.debug(
"VOD: snapped clipFrom to keyframe at %sms for %s, duration now %sms",
keyframe_ms,
recording.path,
duration,
)
else:
# could not read keyframes, remove clipFrom to use full recording
logger.debug(
"VOD: no keyframe info for %s, removing clipFrom to use full recording",
recording.path,
)
del clip["clipFrom"]
duration = int(recording.duration * 1000)
if recording.end_time > end_ts:
duration -= int((recording.end_time - end_ts) * 1000)
if duration < min_duration_ms:
# skip if the clip has no valid duration (too short to contain frames)
logger.debug(
"VOD: skipping recording %s - resulting duration %sms too short",
recording.path,
duration,
)
if built is None:
continue
if min_duration_ms <= duration < max_duration_ms:
clip["keyFrameDurations"] = [duration]
clips.append(clip)
durations.append(duration)
logger.debug(
"VOD: added clip %s duration_ms=%s clipFrom=%s",
recording.path,
duration,
clip.get("clipFrom"),
)
else:
logger.warning(f"Recording clip is missing or empty: {recording.path}")
clips.append(built[0])
durations.append(built[1])
span_streams.add(span_is_main)
if row.video_codec is not None:
video_codecs.add(row.video_codec)
audio_presence.add(row.has_audio is not False)
# legacy rows contribute no signature, so uniformly-unknown
# history keeps the legacy shape
if row.has_audio is not False and (
row.audio_codec is not None or row.audio_rate is not None
):
audio_params.add((row.audio_codec, row.audio_rate))
# nginx-vod requires a uniform track count per sequence, and adding or
# removing an audio track across an MSE discontinuity is unproven
if len(audio_presence) > 1:
logger.debug(
"VOD: %s mixes audio-bearing and audio-less recordings between "
"%s and %s; serving the range without audio",
camera_name,
start_ts,
end_ts,
)
for clip in clips:
clip["tracks"] = "v"
# discontinuity mode emits per-clip init segments, letting the decoder
# reconfigure at each boundary. Stream type counts as a signature of
# its own: the two encoders differ in SPS/PPS even when codec name and
# audio params match, and a single-init manifest then decode-fails on
# players that only configure from the init segment (iOS)
use_discontinuity = (
len(video_codecs) > 1 or len(audio_params) > 1 or len(span_streams) > 1
)
if use_discontinuity:
logger.debug(
"VOD: %s mixes media signatures between %s and %s (video codecs "
"%s, audio params %s, streams %s); serving a discontinuity "
"manifest with per-clip init segments",
camera_name,
start_ts,
end_ts,
sorted(video_codecs),
sorted(audio_params, key=str),
sorted(span_streams),
)
if not clips:
logger.error(
@@ -678,16 +727,49 @@ async def vod_ts(
status_code=404,
)
if len(clips) > NGINX_VOD_MAX_CLIPS:
logger.warning(
"VOD: %s needs %d clips between %s and %s, exceeding nginx's "
"limit of %d; playback of this range will fail. This usually "
"means the camera produced abnormally short recording segments "
"(check the stream's timestamps)",
camera_name,
len(clips),
start_ts,
end_ts,
NGINX_VOD_MAX_CLIPS,
)
hour_ago = datetime.now() - timedelta(hours=1)
return JSONResponse(
content={
"cache": hour_ago.timestamp() > start_ts,
"discontinuity": force_discontinuity,
"consistentSequenceMediaInfo": True,
"durations": durations,
"segment_duration": max(durations),
"sequences": [{"clips": clips}],
}
content = {
"cache": hour_ago.timestamp() > start_ts,
"discontinuity": force_discontinuity or use_discontinuity,
"consistentSequenceMediaInfo": True,
"durations": durations,
# aligns segments to recording file boundaries
"segment_duration": max(durations),
"sequences": [{"clips": clips}],
}
if use_discontinuity:
# clip-indexed naming is what makes nginx-vod emit per-clip
# EXT-X-MAP outside of its live mode
content["initialClipIndex"] = 1
return JSONResponse(content=content)
@router.get(
"/vod/{camera_name}/start/{start_ts}/end/{end_ts}",
dependencies=[Depends(require_camera_access)],
description="Returns an HLS playlist for the specified timestamp-range on the specified camera. Append /master.m3u8 or /index.m3u8 for HLS playback.",
)
async def vod_ts(
camera_name: str,
start_ts: float,
end_ts: float,
force_discontinuity: bool = False,
):
return await _vod_response(
camera_name, start_ts, end_ts, force_discontinuity=force_discontinuity
)
@@ -776,7 +858,43 @@ async def vod_clip(
start_ts: float,
end_ts: float,
):
return await vod_ts(camera_name, start_ts, end_ts, force_discontinuity=True)
# the tracking-details player corrects its timeline from
# sequences[0].clips[0].clipFrom
return await _vod_response(
camera_name,
start_ts,
end_ts,
force_discontinuity=True,
)
# registered after /vod/clip/... on purpose: both routes are six path
# segments, Starlette matches structurally in registration order, and the
# enum validation on {stream} would otherwise 422 every /vod/clip request
@router.get(
"/vod/{camera_name}/{stream}/start/{start_ts}/end/{end_ts}",
dependencies=[Depends(require_camera_access)],
description="Returns an HLS playlist pinned to one stream type (main or sub) for the specified timestamp-range on the specified camera. Append /master.m3u8 or /index.m3u8 for HLS playback.",
)
async def vod_ts_stream(
camera_name: str,
stream: VodStreamPreference,
start_ts: float,
end_ts: float,
force_discontinuity: bool = False,
):
"""VOD for a timestamp range pinned to one stream type.
How the frontend selects quality, now that mappings are always
single-sequence.
"""
return await _vod_response(
camera_name,
start_ts,
end_ts,
force_discontinuity=force_discontinuity,
stream_preference=stream.value,
)
@router.get(
frigate/api/record.py
+106 -16
View File
@@ -25,8 +25,20 @@ from frigate.api.defs.query.recordings_query_parameters import (
)
from frigate.api.defs.response.generic_response import GenericResponse
from frigate.api.defs.tags import Tags
from frigate.const import MAX_SEGMENT_DURATION, RECORD_DIR
from frigate.const import (
MAX_SEGMENT_DURATION,
RECORD_DIR,
STREAM_TYPE_MAIN,
STREAM_TYPE_SUB,
)
from frigate.models import Event, Recordings
from frigate.util.recording_coverage import (
coverage_spans,
known_video_codecs,
realized_timelines,
resolve_coverage,
stream_media_summary,
)
from frigate.util.time import get_dst_transitions
logger = logging.getLogger(__name__)
@@ -149,23 +161,28 @@ async def recordings_summary(camera_name: str, timezone: str = "utc"):
period_hour_modifier = f"{hours_offset} hour"
period_minute_modifier = f"{minutes_offset} minute"
hour_expression = fn.strftime(
"%Y-%m-%d %H",
fn.datetime(
Recordings.start_time,
"unixepoch",
period_hour_modifier,
period_minute_modifier,
),
)
# sub rows duplicate the camera's motion/object stats, so
# aggregating them too would double-count
recording_groups = (
Recordings.select(
fn.strftime(
"%Y-%m-%d %H",
fn.datetime(
Recordings.start_time,
"unixepoch",
period_hour_modifier,
period_minute_modifier,
),
).alias("hour"),
hour_expression.alias("hour"),
fn.SUM(Recordings.duration).alias("duration"),
fn.SUM(Recordings.motion).alias("motion"),
fn.SUM(Recordings.objects).alias("objects"),
)
.where(
(Recordings.camera == camera_name)
& (Recordings.stream_type == STREAM_TYPE_MAIN)
& (Recordings.end_time >= period_start)
& (Recordings.start_time <= period_end)
)
@@ -174,6 +191,23 @@ async def recordings_summary(camera_name: str, timezone: str = "utc"):
.namedtuples()
)
# sub recordings can outlive main, so hours covered only by sub
# rows are reported too, flagged as sub_only
sub_groups = (
Recordings.select(
hour_expression.alias("hour"),
fn.SUM(Recordings.duration).alias("duration"),
)
.where(
(Recordings.camera == camera_name)
& (Recordings.stream_type == STREAM_TYPE_SUB)
& (Recordings.end_time >= period_start)
& (Recordings.start_time <= period_end)
)
.group_by((Recordings.start_time + period_offset).cast("int") / 3600)
.namedtuples()
)
event_groups = (
Event.select(
fn.strftime(
@@ -197,17 +231,43 @@ async def recordings_summary(camera_name: str, timezone: str = "utc"):
event_map = {g.hour: g.count for g in event_groups}
for recording_group in recording_groups:
parts = recording_group.hour.split()
hour_stats = [
(
g.hour,
{
"motion": g.motion,
"objects": g.objects,
"duration": round(g.duration),
},
)
for g in recording_groups
]
main_hours = {group_hour for group_hour, _ in hour_stats}
hour_stats.extend(
(
g.hour,
{
"motion": 0,
"objects": 0,
"duration": round(g.duration),
"sub_only": True,
},
)
for g in sub_groups
if g.hour not in main_hours
)
# restore the most-recent-first ordering after merging in sub hours
hour_stats.sort(key=lambda entry: entry[0], reverse=True)
for group_hour, stats in hour_stats:
parts = group_hour.split()
hour = parts[1]
day = parts[0]
events_count = event_map.get(recording_group.hour, 0)
events_count = event_map.get(group_hour, 0)
hour_data = {
"hour": hour,
"events": events_count,
"motion": recording_group.motion,
"objects": recording_group.objects,
"duration": round(recording_group.duration),
**stats,
}
if day in days:
# merge counts if already present (edge-case at DST boundary)
@@ -223,6 +283,35 @@ async def recordings_summary(camera_name: str, timezone: str = "utc"):
return JSONResponse(content=list(days.values()))
@router.get(
"/{camera_name}/recordings/coverage",
dependencies=[Depends(require_camera_access)],
)
async def recordings_coverage(
camera_name: str, after: float, before: float, timelines: bool = False
):
"""Returns merged recording coverage spans plus codec compatibility.
codecs_compatible is false only when more than one known video codec
appears across the range's rows, the case where the merged vod route
degrades to a single-stream manifest.
"""
intervals = resolve_coverage(camera_name, after, before)
content = {
"spans": coverage_spans(intervals),
"codecs_compatible": len(known_video_codecs(intervals)) <= 1,
"streams": stream_media_summary(intervals),
}
# pure computation (shared plan_clip, record-time keyframe index), but
# opt-in for payload hygiene: day-level requests need only the spans
if timelines:
content["timelines"] = realized_timelines(intervals)
return JSONResponse(content=content)
@router.get("/{camera_name}/recordings", dependencies=[Depends(require_camera_access)])
async def recordings(
camera_name: str,
@@ -243,6 +332,7 @@ async def recordings(
)
.where(
Recordings.camera == camera_name,
Recordings.stream_type == STREAM_TYPE_MAIN,
Recordings.start_time >= after - MAX_SEGMENT_DURATION,
Recordings.end_time >= after,
Recordings.start_time <= before,
frigate/api/review.py
+3
View File
@@ -33,6 +33,7 @@ from frigate.api.defs.response.review_response import (
ReviewSummaryResponse,
)
from frigate.api.defs.tags import Tags
from frigate.const import STREAM_TYPE_MAIN
from frigate.embeddings import EmbeddingsContext
from frigate.models import Recordings, ReviewSegment, UserReviewStatus
from frigate.review.types import SeverityEnum
@@ -598,6 +599,8 @@ def motion_activity(
clauses = [(Recordings.start_time > after) & (Recordings.end_time < before)]
clauses.append(Recordings.motion > 0)
# sub rows duplicate the camera's motion stats, so only count main rows
clauses.append(Recordings.stream_type == STREAM_TYPE_MAIN)
if cameras != "all":
requested = set(cameras.split(","))
frigate/config/camera/camera.py
+28 -1
View File
@@ -3,7 +3,12 @@ from enum import Enum
from pydantic import Field, PrivateAttr, model_validator
from frigate.const import CACHE_DIR, CACHE_SEGMENT_FORMAT, REGEX_CAMERA_NAME
from frigate.const import (
CACHE_DIR,
CACHE_SEGMENT_FORMAT,
REGEX_CAMERA_NAME,
SUB_CACHE_TAG,
)
from frigate.ffmpeg_presets import (
parse_preset_hardware_acceleration_decode,
parse_preset_hardware_acceleration_scale,
@@ -294,6 +299,28 @@ class CameraConfig(FrigateBaseModel):
+ ffmpeg_output_args
)
if (
"record_sub" in ffmpeg_input.roles
and self.record.enabled
and self.record.sub.enabled
):
sub_output_args = self.ffmpeg.output_args.effective_record_sub
record_args = get_ffmpeg_arg_list(
parse_preset_output_record(
sub_output_args,
self.ffmpeg.apple_compatibility,
)
or sub_output_args
)
ffmpeg_output_args = (
record_args
+ [
f"{os.path.join(CACHE_DIR, self.name)}{SUB_CACHE_TAG}@{CACHE_SEGMENT_FORMAT}.mp4"
]
+ ffmpeg_output_args
)
# if there aren't any outputs enabled for this input
if len(ffmpeg_output_args) == 0:
return None
frigate/config/camera/ffmpeg.py
+15
View File
@@ -42,6 +42,20 @@ class FfmpegOutputArgsConfig(FrigateBaseModel):
title="Record output arguments",
description="Default output arguments for record role streams.",
)
record_sub: str | list[str] = Field(
default_factory=list,
title="Sub stream record output arguments",
description="Output arguments for record_sub role streams. The record output arguments are used when this is not set.",
)
@property
def effective_record_sub(self) -> str | list[str]:
"""Output arguments used for the record_sub role.
Falls back to the record arguments rather than to the stock preset so
that a customized record value keeps applying to both recorded streams.
"""
return self.record_sub or self.record
class FfmpegConfig(FrigateBaseModel):
@@ -99,6 +113,7 @@ class FfmpegConfig(FrigateBaseModel):
class CameraRoleEnum(str, Enum):
audio = "audio"
record = "record"
record_sub = "record_sub"
detect = "detect"
frigate/config/camera/record.py
+52
View File
@@ -13,6 +13,7 @@ __all__ = [
"RecordExportConfig",
"RecordPreviewConfig",
"RecordQualityEnum",
"RecordSubConfig",
"EventsConfig",
"ReviewRetainConfig",
"RecordRetainConfig",
@@ -110,6 +111,34 @@ class RecordExportConfig(FrigateBaseModel):
)
class RecordSubConfig(FrigateBaseModel):
enabled: bool = Field(
default=False,
title="Enable sub stream recording",
description="Enable recording of a second, lower quality stream for adaptive quality playback and extended retention.",
)
continuous: RecordRetainConfig = Field(
default_factory=RecordRetainConfig,
title="Sub stream continuous retention",
description="Number of days to retain sub stream recordings regardless of tracked objects or motion.",
)
motion: RecordRetainConfig = Field(
default_factory=RecordRetainConfig,
title="Sub stream motion retention",
description="Number of days to retain sub stream recordings triggered by motion.",
)
alerts: ReviewRetainConfig = Field(
default_factory=ReviewRetainConfig,
title="Sub stream alert retention",
description="Retention settings for sub stream recordings of alerts.",
)
detections: ReviewRetainConfig = Field(
default_factory=ReviewRetainConfig,
title="Sub stream detection retention",
description="Retention settings for sub stream recordings of detections.",
)
class RecordConfig(FrigateBaseModel):
enabled: bool = Field(
default=False,
@@ -151,12 +180,35 @@ class RecordConfig(FrigateBaseModel):
title="Preview config",
description="Settings controlling the quality of recording previews shown in the UI.",
)
sub: RecordSubConfig = Field(
default_factory=RecordSubConfig,
title="Sub stream recording",
description="Settings for recording a second, lower quality stream.",
)
enabled_in_config: bool | None = Field(
default=None,
title="Original recording state",
description="Indicates whether recording was enabled in the original static configuration.",
)
@property
def effective_alert_days(self) -> float:
"""Alert retention extended to the sub stream window when sub is enabled.
Review items and tracked objects must stay visible for as long as
either stream still has recordings.
"""
if self.sub.enabled:
return max(self.alerts.retain.days, self.sub.alerts.days)
return self.alerts.retain.days
@property
def effective_detection_days(self) -> float:
"""Detection retention extended to the sub window when sub is enabled."""
if self.sub.enabled:
return max(self.detections.retain.days, self.sub.detections.days)
return self.detections.retain.days
@property
def event_pre_capture(self) -> int:
return max(
frigate/config/camera/updater.py
+6 -1
View File
@@ -129,8 +129,13 @@ class CameraConfigUpdateSubscriber:
config.objects = updated_config
elif update_type == CameraConfigUpdateEnum.record:
old_enabled_in_config = config.record.enabled_in_config
old_sub_enabled = config.record.sub.enabled
config.record = updated_config
if old_enabled_in_config != updated_config.enabled_in_config:
# the record and record_sub ffmpeg outputs are gated on these
if (
old_enabled_in_config != updated_config.enabled_in_config
or old_sub_enabled != updated_config.sub.enabled
):
config.recreate_ffmpeg_cmds()
elif update_type == CameraConfigUpdateEnum.review:
config.review = updated_config
frigate/config/config.py
+31 -8
View File
@@ -255,6 +255,15 @@ def verify_config_roles(camera_config: CameraConfig) -> None:
f"Camera {camera_config.name} has record enabled, but record is not assigned to an input."
)
if (
camera_config.record.enabled
and camera_config.record.sub.enabled
and "record_sub" not in assigned_roles
):
raise ValueError(
f"Camera {camera_config.name} has sub stream recording enabled, but record_sub is not assigned to an input."
)
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."
@@ -275,13 +284,11 @@ def verify_valid_live_stream_names(
)
def verify_recording_segments_setup_with_reasonable_time(
camera_config: CameraConfig,
def verify_record_output_args_segment_time(
camera_config: CameraConfig, output_args: str | list[str], role: str
) -> None:
"""Verify that recording segments are setup and segment time is not greater than 60."""
record_args: list[str] = get_ffmpeg_arg_list(
camera_config.ffmpeg.output_args.record
)
"""Verify that a recording role's output args segment at a reasonable time."""
record_args: list[str] = get_ffmpeg_arg_list(output_args)
if record_args[0].startswith("preset"):
return
@@ -291,16 +298,32 @@ def verify_recording_segments_setup_with_reasonable_time(
except ValueError:
raise ValueError(
f"Camera {camera_config.name} has no segment_time in \
recording output args, segment args are required for record."
{role} output args, segment args are required for record."
) from None
if int(record_args[seg_arg_index + 1]) > 60:
raise ValueError(
f"Camera {camera_config.name} has invalid segment_time output arg, \
f"Camera {camera_config.name} has invalid segment_time in {role} output args, \
segment_time must be 60 or less."
)
def verify_recording_segments_setup_with_reasonable_time(
camera_config: CameraConfig,
) -> None:
"""Verify that recording segments are setup and segment time is not greater than 60."""
verify_record_output_args_segment_time(
camera_config, camera_config.ffmpeg.output_args.record, "recording"
)
if camera_config.record.sub.enabled:
verify_record_output_args_segment_time(
camera_config,
camera_config.ffmpeg.output_args.effective_record_sub,
"sub stream recording",
)
def verify_zone_objects_are_tracked(camera_config: CameraConfig) -> None:
"""Verify that user has not entered zone objects that are not in the tracking config."""
for zone_name, zone in camera_config.zones.items():
frigate/const.py
+6
View File
@@ -23,6 +23,12 @@ SHM_FRAMES_VAR = "SHM_MAX_FRAMES"
REDACTED_CREDENTIAL_SENTINEL = "__FRIGATE_SAVED_CREDENTIAL__"
# Stream type constants
STREAM_TYPE_MAIN = "main"
STREAM_TYPE_SUB = "sub"
SUB_CACHE_TAG = "@sub"
# Attribute & Object constants
DEFAULT_ATTRIBUTE_LABEL_MAP = {
frigate/events/cleanup.py
+8 -8
View File
@@ -197,9 +197,11 @@ class EventCleanup(threading.Thread):
def expire_clips(self) -> list[str]:
## Expire events from unlisted cameras based on the global config
# effective days cover the sub window, keeping tracked objects in
# Explore while sub recordings and review items still exist
expire_days = max(
self.config.record.alerts.retain.days,
self.config.record.detections.retain.days,
self.config.record.effective_alert_days,
self.config.record.effective_detection_days,
)
file_extension = None # mp4 clips are no longer stored in /clips
update_params = {"has_clip": False}
@@ -278,15 +280,13 @@ class EventCleanup(threading.Thread):
## Expire events from cameras based on the camera config
for name, camera in self.config.cameras.items():
expire_days = max(
camera.record.alerts.retain.days,
camera.record.detections.retain.days,
)
# effective days cover the sub window, keeping tracked objects
# in Explore while sub recordings and review items still exist
alert_expire_date = (
now - datetime.timedelta(days=camera.record.alerts.retain.days)
now - datetime.timedelta(days=camera.record.effective_alert_days)
).timestamp()
detection_expire_date = (
now - datetime.timedelta(days=camera.record.detections.retain.days)
now - datetime.timedelta(days=camera.record.effective_detection_days)
).timestamp()
# grab all events after specific time
expired_events = (
frigate/jobs/motion_search.py
+2 -1
View File
@@ -15,7 +15,7 @@ import numpy as np
from frigate.comms.inter_process import InterProcessRequestor
from frigate.config import FrigateConfig
from frigate.const import UPDATE_JOB_STATE
from frigate.const import STREAM_TYPE_MAIN, UPDATE_JOB_STATE
from frigate.jobs.job import Job
from frigate.jobs.manager import (
get_job_by_id,
@@ -485,6 +485,7 @@ class MotionSearchRunner(threading.Thread):
)
)
.where(Recordings.camera == camera_name)
.where(Recordings.stream_type == STREAM_TYPE_MAIN)
.order_by(Recordings.start_time.asc())
)
frigate/models.py
+6
View File
@@ -79,6 +79,12 @@ class Recordings(Model):
segment_size = FloatField(default=0) # this should be stored as MB
regions = IntegerField(null=True)
motion_heatmap = JSONField(null=True) # 16x16 grid, 256 values (0-255)
keyframes = JSONField(null=True) # ms offsets; NULL = unprobed (legacy rows)
stream_type = CharField(default="main", max_length=8)
has_audio = BooleanField(null=True) # NULL = unknown (legacy rows)
audio_rate = IntegerField(null=True) # Hz; NULL = unknown (legacy rows)
audio_codec = CharField(null=True, max_length=20) # NULL = unknown (legacy rows)
video_codec = CharField(null=True, max_length=20) # NULL = unknown (legacy rows)
class ExportCase(Model):
frigate/record/cleanup.py
+96 -13
View File
@@ -12,7 +12,14 @@ from typing import Any
from playhouse.sqlite_ext import SqliteExtDatabase
from frigate.config import CameraConfig, FrigateConfig, RetainModeEnum
from frigate.const import CACHE_DIR, CLIPS_DIR, MAX_WAL_SIZE, RECORD_DIR
from frigate.const import (
CACHE_DIR,
CLIPS_DIR,
MAX_WAL_SIZE,
RECORD_DIR,
STREAM_TYPE_MAIN,
STREAM_TYPE_SUB,
)
from frigate.models import Previews, Recordings, ReviewSegment, UserReviewStatus
from frigate.util.builtin import clear_and_unlink
from frigate.util.media import remove_empty_directories
@@ -20,6 +27,29 @@ from frigate.util.media import remove_empty_directories
logger = logging.getLogger(__name__)
def _filter_reviews_for_pass(
reviews: list[Any],
now: datetime.datetime,
alerts_days: float,
detections_days: float,
) -> list[Any]:
"""Limit reviews to those still within this pass's per-severity retention window.
Review rows survive to the longer of the main and sub retention windows,
so a pass that honored all of them would let extended sub retention keep
main recordings alive too. Filtering preserves sort order for the overlap
loop in expire_existing_camera_recordings.
"""
alert_cutoff = (now - datetime.timedelta(days=alerts_days)).timestamp()
detection_cutoff = (now - datetime.timedelta(days=detections_days)).timestamp()
return [
r
for r in reviews
if r.end_time is None
or (r.end_time >= (alert_cutoff if r.severity == "alert" else detection_cutoff))
]
class RecordingCleanup(threading.Thread):
"""Cleanup existing recordings based on retention config."""
@@ -65,11 +95,14 @@ class RecordingCleanup(threading.Thread):
self, config: CameraConfig, now: datetime.datetime
) -> set[Path]:
"""Delete review segments that are expired"""
alert_expire_date = (
now - datetime.timedelta(days=config.record.alerts.retain.days)
).timestamp()
# review rows survive to the longer of the main and sub windows so
# they stay visible while either stream still has recordings
alert_days = config.record.effective_alert_days
detection_days = config.record.effective_detection_days
alert_expire_date = (now - datetime.timedelta(days=alert_days)).timestamp()
detection_expire_date = (
now - datetime.timedelta(days=config.record.detections.retain.days)
now - datetime.timedelta(days=detection_days)
).timestamp()
expired_reviews = (
ReviewSegment.select(ReviewSegment.id, ReviewSegment.thumb_path)
@@ -109,8 +142,11 @@ class RecordingCleanup(threading.Thread):
def expire_existing_camera_recordings(
self,
stream_type: str,
continuous_expire_date: float,
motion_expire_date: float,
alerts_retain_mode: RetainModeEnum,
detections_retain_mode: RetainModeEnum,
config: CameraConfig,
reviews: list[Any],
) -> set[Path]:
@@ -130,6 +166,7 @@ class RecordingCleanup(threading.Thread):
)
.where(
(Recordings.camera == config.name)
& (Recordings.stream_type == stream_type)
& (
(
(Recordings.end_time < continuous_expire_date)
@@ -175,9 +212,9 @@ class RecordingCleanup(threading.Thread):
):
keep = True
mode = (
config.record.alerts.retain.mode
alerts_retain_mode
if review.severity == "alert"
else config.record.detections.retain.mode
else detections_retain_mode
)
break
@@ -216,6 +253,10 @@ 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
previews = (
Previews.select(
Previews.id,
@@ -342,6 +383,20 @@ class RecordingCleanup(threading.Thread):
)
).timestamp()
# computed here so the reviews window below covers both passes
sub_continuous_expire_date = (
now - datetime.timedelta(days=config.record.sub.continuous.days)
).timestamp()
sub_motion_expire_date = (
now
- datetime.timedelta(
days=max(
config.record.sub.motion.days,
config.record.sub.continuous.days,
) # can't keep motion for less than continuous
)
).timestamp()
# Get all the reviews to check against
reviews = (
ReviewSegment.select(
@@ -351,18 +406,46 @@ class RecordingCleanup(threading.Thread):
)
.where(
ReviewSegment.camera == camera,
# candidate recordings can extend up to continuous_expire_date
# (the no-motion no-audio branch of the recordings query),
# so reviews must cover that full range to avoid deleting
# segments that overlap recent alerts/detections.
ReviewSegment.start_time < continuous_expire_date,
# candidate recordings reach the later of the two passes'
# continuous cutoffs, so reviews must cover that whole
# range or segments overlapping recent alerts get deleted
ReviewSegment.start_time
< max(continuous_expire_date, sub_continuous_expire_date),
)
.order_by(ReviewSegment.start_time)
.namedtuples()
)
maybe_empty_dirs |= self.expire_existing_camera_recordings(
continuous_expire_date, motion_expire_date, config, reviews
STREAM_TYPE_MAIN,
continuous_expire_date,
motion_expire_date,
config.record.alerts.retain.mode,
config.record.detections.retain.mode,
config,
_filter_reviews_for_pass(
reviews,
now,
config.record.alerts.retain.days,
config.record.detections.retain.days,
),
)
# runs even when sub recording is disabled so old rows still
# expire
maybe_empty_dirs |= self.expire_existing_camera_recordings(
STREAM_TYPE_SUB,
sub_continuous_expire_date,
sub_motion_expire_date,
config.record.sub.alerts.mode,
config.record.sub.detections.mode,
config,
_filter_reviews_for_pass(
reviews,
now,
config.record.sub.alerts.days,
config.record.sub.detections.days,
),
)
logger.debug(f"End camera: {camera}.")
frigate/record/export.py
+38 -30
View File
@@ -12,6 +12,7 @@ import threading
from collections.abc import Callable
from enum import Enum
from pathlib import Path
from typing import Any
import pytz # type: ignore[import-untyped]
from peewee import DoesNotExist
@@ -24,6 +25,8 @@ from frigate.const import (
EXPORT_DIR,
MAX_PLAYLIST_SECONDS,
PREVIEW_FRAME_TYPE,
STREAM_TYPE_MAIN,
STREAM_TYPE_SUB,
)
from frigate.ffmpeg_presets import (
EncodeTypeEnum,
@@ -284,6 +287,29 @@ class RecordingExporter(threading.Thread):
return input_duration * factor
def _get_recordings_for_range(self, stream_type: str) -> list[Any]:
"""Fetch one stream type's recording rows overlapping the export range."""
return list(
Recordings.select(
Recordings.start_time,
Recordings.end_time,
)
.where(
Recordings.start_time.between(self.start_time, self.end_time)
| Recordings.end_time.between(self.start_time, self.end_time)
| (
(self.start_time > Recordings.start_time)
& (self.end_time < Recordings.end_time)
)
)
.where(
(Recordings.camera == self.camera)
& (Recordings.stream_type == stream_type)
)
.order_by(Recordings.start_time.asc())
.iterator()
)
def _sum_source_duration_seconds(self) -> float | None:
"""Sum saved-video seconds inside [start_time, end_time].
@@ -294,19 +320,12 @@ class RecordingExporter(threading.Thread):
"""
try:
if self.playback_source == PlaybackSourceEnum.recordings:
rows = (
Recordings.select(Recordings.start_time, Recordings.end_time)
.where(
Recordings.start_time.between(self.start_time, self.end_time)
| Recordings.end_time.between(self.start_time, self.end_time)
| (
(self.start_time > Recordings.start_time)
& (self.end_time < Recordings.end_time)
)
)
.where(Recordings.camera == self.camera)
.iterator()
)
# never mix streams in one estimate; use main when available
# and fall back to sub for expired-main history
rows = self._get_recordings_for_range(STREAM_TYPE_MAIN)
if not rows:
rows = self._get_recordings_for_range(STREAM_TYPE_SUB)
else:
rows = (
Previews.select(Previews.start_time, Previews.end_time)
@@ -692,23 +711,12 @@ class RecordingExporter(threading.Thread):
if type(internal_port) is str:
internal_port = int(internal_port.split(":")[-1])
recordings = list(
Recordings.select(
Recordings.start_time,
Recordings.end_time,
)
.where(
Recordings.start_time.between(self.start_time, self.end_time)
| Recordings.end_time.between(self.start_time, self.end_time)
| (
(self.start_time > Recordings.start_time)
& (self.end_time < Recordings.end_time)
)
)
.where(Recordings.camera == self.camera)
.order_by(Recordings.start_time.asc())
.iterator()
)
# never mix streams in one playlist; use main when available and
# fall back to sub for expired-main history
recordings = self._get_recordings_for_range(STREAM_TYPE_MAIN)
if not recordings:
recordings = self._get_recordings_for_range(STREAM_TYPE_SUB)
playlist_lines: list[str] = []
if (self.end_time - self.start_time) <= MAX_PLAYLIST_SECONDS:
frigate/record/maintainer.py
+336 -87
View File
@@ -15,6 +15,7 @@ from typing import Any
import numpy as np
import psutil
from peewee import fn
from frigate.comms.detections_updater import DetectionSubscriber, DetectionTypeEnum
from frigate.comms.inter_process import InterProcessRequestor
@@ -35,15 +36,48 @@ from frigate.const import (
MAX_SEGMENT_DURATION,
MAX_SEGMENTS_IN_CACHE,
RECORD_DIR,
STREAM_TYPE_MAIN,
STREAM_TYPE_SUB,
SUB_CACHE_TAG,
)
from frigate.models import Recordings, ReviewSegment
from frigate.review.types import SeverityEnum
from frigate.util.media import get_keyframe_offsets
from frigate.util.services import get_video_properties
logger = logging.getLogger(__name__)
STALE_RECORDINGS_INFO_TTL = MAX_SEGMENTS_IN_CACHE * MAX_SEGMENT_DURATION * 2
# cache filenames have whole-second resolution, so a contiguous segment's
# parsed start lands up to 1s before the previous segment's true end
SEGMENT_CHAIN_TOLERANCE_S = 1.0
# against an mtime-measured start, disagreement beyond this means
# accumulated probe-duration error and the chain re-anchors on the mtime
SEGMENT_CHAIN_DRIFT_LIMIT_S = 0.5
# probing every cached segment at once starves the camera and detection
# processes, and the probes then blow their own timeouts together, so
# segments get discarded as corrupt and the record watchdog restarts ffmpeg
MAX_CONCURRENT_SEGMENT_PROBES = 4
def parse_cache_segment_name(basename: str) -> tuple[str, str, str] | None:
"""Parse a cache segment basename into (camera, stream_type, date).
Main segments are named {camera}@{date}; sub segments {camera}@sub@{date}.
"""
try:
prefix, date = basename.rsplit("@", maxsplit=1)
except ValueError:
return None
if prefix.endswith(SUB_CACHE_TAG):
return (prefix[: -len(SUB_CACHE_TAG)], STREAM_TYPE_SUB, date)
return (prefix, STREAM_TYPE_MAIN, date)
class SegmentInfo:
def __init__(
@@ -83,6 +117,10 @@ class SegmentInfo:
class RecordingMaintainer(threading.Thread):
# move_files replaces this per cycle: an asyncio primitive binds to the
# first event loop that contends it, and every cycle runs in a new loop
probe_semaphore = asyncio.Semaphore(MAX_CONCURRENT_SEGMENT_PROBES)
def __init__(self, config: FrigateConfig, stop_event: MpEvent):
super().__init__(name="recording_maintainer")
self.config = config
@@ -100,10 +138,101 @@ class RecordingMaintainer(threading.Thread):
self.stop_event = stop_event
self.object_recordings_info: dict[str, list] = defaultdict(list)
self.audio_recordings_info: dict[str, list] = defaultdict(list)
self.end_time_cache: dict[str, tuple[datetime.datetime, float]] = {}
# cache_path -> (end_time, duration, has_audio, audio_rate,
# audio_codec, video_codec, keyframes)
self.end_time_cache: dict[
str,
tuple[
datetime.datetime,
float,
bool | None,
int | None,
str | None,
str | None,
list[int] | None,
],
] = {}
# last known capture end per (camera, stream_type); 0.0 marks a key
# whose DB seed found no rows
self.last_segment_end: dict[tuple[str, str], float] = {}
self.unexpected_cache_files_logged: bool = False
def _get_last_segment_end(self, camera: str, stream_type: str) -> float | None:
"""Return the last known capture end time for a camera stream.
Lazily seeds from the most recent stored recording so start-time
chains survive restarts.
"""
key = (camera, stream_type)
if key not in self.last_segment_end:
last_db_end = (
Recordings.select(fn.MAX(Recordings.end_time))
.where(
Recordings.camera == camera,
Recordings.stream_type == stream_type,
)
.scalar()
)
# the 0.0 sentinel keeps the seed query from repeating
self.last_segment_end[key] = last_db_end if last_db_end is not None else 0.0
return self.last_segment_end[key] or None
def _resolve_segment_start(
self,
camera: str,
stream_type: str,
filename_start: datetime.datetime,
duration: float,
cache_path: str,
) -> datetime.datetime:
"""Resolve a segment's true start time from its cache file.
Cache filenames carry whole-second resolution, so the parsed start
sits up to 1s early. The cache file's mtime is the wall clock when
ffmpeg rolled the segment, so mtime minus the probed duration
restores the fractional start. Contiguous segments still chain to
the previous segment's end so rows stay exactly adjacent.
"""
filename_ts = filename_start.timestamp()
measured: float | None = None
try:
mtime = os.path.getmtime(cache_path)
except OSError:
mtime = None
if mtime is not None:
candidate = mtime - duration
# media shorter than its wall span (a stalled stream, an early
# close) derives a start past the truncation window, where the
# floored filename start is safer
if 0 <= candidate - filename_ts < SEGMENT_CHAIN_TOLERANCE_S:
measured = candidate
last_end = self._get_last_segment_end(camera, stream_type)
if measured is not None:
if (
last_end is not None
and abs(last_end - measured) < SEGMENT_CHAIN_DRIFT_LIMIT_S
):
return datetime.datetime.fromtimestamp(last_end, tz=datetime.UTC)
return datetime.datetime.fromtimestamp(measured, tz=datetime.UTC)
# no usable mtime: capture is continuous within a run, so a
# filename start just before the previous end chains to that end
if (
last_end is not None
and 0 <= last_end - filename_ts < SEGMENT_CHAIN_TOLERANCE_S
):
return datetime.datetime.fromtimestamp(last_end, tz=datetime.UTC)
return filename_start
async def move_files(self) -> None:
self.probe_semaphore = asyncio.Semaphore(MAX_CONCURRENT_SEGMENT_PROBES)
cache_files = [
d
for d in os.listdir(CACHE_DIR)
@@ -117,13 +246,17 @@ class RecordingMaintainer(threading.Thread):
for cache in cache_files:
cache_path = os.path.join(CACHE_DIR, cache)
basename = os.path.splitext(cache)[0]
try:
camera, date = basename.rsplit("@", maxsplit=1)
except ValueError:
parsed = parse_cache_segment_name(basename)
if parsed is None:
if not self.unexpected_cache_files_logged:
logger.warning("Skipping unexpected files in 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
@@ -167,8 +300,10 @@ class RecordingMaintainer(threading.Thread):
except psutil.Error:
continue
# group recordings by camera (skip in-use for validation/moving)
grouped_recordings: defaultdict[str, list[dict[str, Any]]] = defaultdict(list)
# group recordings by camera and stream type (skip in-use for validation/moving)
grouped_recordings: defaultdict[tuple[str, str], list[dict[str, Any]]] = (
defaultdict(list)
)
for cache in cache_files:
# Skip files currently in use
if cache in files_in_use:
@@ -176,32 +311,35 @@ class RecordingMaintainer(threading.Thread):
cache_path = os.path.join(CACHE_DIR, cache)
basename = os.path.splitext(cache)[0]
try:
camera, date = basename.rsplit("@", maxsplit=1)
except ValueError:
parsed = parse_cache_segment_name(basename)
if parsed is None:
if not self.unexpected_cache_files_logged:
logger.warning("Skipping unexpected files in cache")
self.unexpected_cache_files_logged = True
continue
camera, stream_type, date = parsed
# important that start_time is utc because recordings are stored and compared in utc
start_time = datetime.datetime.strptime(
date, CACHE_SEGMENT_FORMAT
).astimezone(datetime.UTC)
grouped_recordings[camera].append(
grouped_recordings[(camera, stream_type)].append(
{
"cache_path": cache_path,
"start_time": start_time,
"stream_type": stream_type,
}
)
# delete all cached files past the most recent MAX_SEGMENTS_IN_CACHE
keep_count = MAX_SEGMENTS_IN_CACHE
for camera in grouped_recordings.keys():
for key in grouped_recordings.keys():
camera, stream_type = key
# sort based on start time
grouped_recordings[camera] = sorted(
grouped_recordings[camera], key=lambda s: s["start_time"]
grouped_recordings[key] = sorted(
grouped_recordings[key], key=lambda s: s["start_time"]
)
camera_info = self.object_recordings_info[camera]
@@ -216,7 +354,7 @@ class RecordingMaintainer(threading.Thread):
r["start_time"].timestamp()
< most_recently_processed_frame_time
),
grouped_recordings[camera],
grouped_recordings[key],
)
)
)
@@ -226,103 +364,133 @@ class RecordingMaintainer(threading.Thread):
logger.warning(
f"Unable to keep up with recording segments in cache for {camera}. Keeping the {keep_count} most recent segments out of {processed_segment_count} and discarding the rest..."
)
to_remove = grouped_recordings[camera][:-keep_count]
to_remove = grouped_recordings[key][:-keep_count]
for rec in to_remove:
cache_path = rec["cache_path"]
Path(cache_path).unlink(missing_ok=True)
self.end_time_cache.pop(cache_path, None)
grouped_recordings[camera] = grouped_recordings[camera][-keep_count:]
grouped_recordings[key] = grouped_recordings[key][-keep_count:]
# see if detection has failed and unprocessed segments need to be deleted
unprocessed_segment_count = (
len(grouped_recordings[camera]) - processed_segment_count
len(grouped_recordings[key]) - processed_segment_count
)
if unprocessed_segment_count > keep_count:
logger.warning(
f"Too many unprocessed recording segments in cache for {camera}. This likely indicates an issue with the detect stream, keeping the {keep_count} most recent segments out of {unprocessed_segment_count} and discarding the rest..."
)
to_remove = grouped_recordings[camera][:-keep_count]
to_remove = grouped_recordings[key][:-keep_count]
for rec in to_remove:
cache_path = rec["cache_path"]
Path(cache_path).unlink(missing_ok=True)
self.end_time_cache.pop(cache_path, None)
grouped_recordings[camera] = grouped_recordings[camera][-keep_count:]
grouped_recordings[key] = grouped_recordings[key][-keep_count:]
tasks = []
for camera, recordings in grouped_recordings.items():
# frame stats are shared per camera across stream types, so trimming
# to one stream's oldest cache would pop frames the other still needs
min_start_per_camera: dict[str, float] = {}
for key, recordings in grouped_recordings.items():
camera, _ = key
oldest_start = recordings[0]["start_time"].timestamp()
if (
camera not in min_start_per_camera
or oldest_start < min_start_per_camera[camera]
):
min_start_per_camera[camera] = oldest_start
for camera, min_start in min_start_per_camera.items():
# clear out all the object recording info for old frames
while (
len(self.object_recordings_info[camera]) > 0
and self.object_recordings_info[camera][0][0]
< recordings[0]["start_time"].timestamp()
and self.object_recordings_info[camera][0][0] < min_start
):
self.object_recordings_info[camera].pop(0)
# clear out all the audio recording info for old frames
while (
len(self.audio_recordings_info[camera]) > 0
and self.audio_recordings_info[camera][0][0]
< recordings[0]["start_time"].timestamp()
and self.audio_recordings_info[camera][0][0] < min_start
):
self.audio_recordings_info[camera].pop(0)
# get all reviews with the end time after the start of the oldest cache file
# or with end_time None
reviews = (
ReviewSegment.select(
ReviewSegment.start_time,
ReviewSegment.end_time,
ReviewSegment.severity,
ReviewSegment.data,
tasks = []
reviews_by_camera: dict[str, Any] = {}
for key, recordings in grouped_recordings.items():
camera, stream_type = key
# get all reviews with the end time after the start of the oldest
# cache file or with end_time None; shared across stream types
if camera not in reviews_by_camera:
reviews_by_camera[camera] = (
ReviewSegment.select(
ReviewSegment.start_time,
ReviewSegment.end_time,
ReviewSegment.severity,
ReviewSegment.data,
)
.where(
ReviewSegment.camera == camera,
(ReviewSegment.end_time == None)
| (ReviewSegment.end_time >= min_start_per_camera[camera]),
)
.order_by(ReviewSegment.start_time)
)
.where(
ReviewSegment.camera == camera,
(ReviewSegment.end_time == None)
| (
ReviewSegment.end_time
>= recordings[0]["start_time"].timestamp()
),
)
.order_by(ReviewSegment.start_time)
)
reviews = reviews_by_camera[camera]
tasks.extend(
[self.validate_and_move_segment(camera, reviews, r) for r in recordings]
)
# publish most recently available recording time and None if disabled
camera_cfg = self.config.cameras.get(camera)
self.recordings_publisher.publish(
(
camera,
recordings[0]["start_time"].timestamp()
if camera_cfg and camera_cfg.record.enabled
else None,
None,
),
RecordingsDataTypeEnum.saved.value,
)
if stream_type == STREAM_TYPE_MAIN:
camera_cfg = self.config.cameras.get(camera)
self.recordings_publisher.publish(
(
camera,
recordings[0]["start_time"].timestamp()
if camera_cfg and camera_cfg.record.enabled
else None,
None,
),
RecordingsDataTypeEnum.saved.value,
)
self._expire_stale_recordings_info(grouped_recordings)
recordings_to_insert: list[dict[str, Any] | None] = await asyncio.gather(*tasks)
# fire and forget recordings entries
self.requestor.send_data(
INSERT_MANY_RECORDINGS,
[r for r in recordings_to_insert if r is not None],
# one segment must not abort the cycle: an exception propagating out
# of gather would abandon the other segments' in-flight probes
results: list[dict[str, Any] | None | BaseException] = await asyncio.gather(
*tasks, return_exceptions=True
)
recordings_to_insert: list[dict[str, Any]] = []
for result in results:
if isinstance(result, BaseException):
logger.error(
"Failed to validate and move a recording segment", exc_info=result
)
continue
if result is not None:
recordings_to_insert.append(result)
# fire and forget recordings entries
self.requestor.send_data(INSERT_MANY_RECORDINGS, recordings_to_insert)
def _expire_stale_recordings_info(
self, grouped_recordings: defaultdict[str, list[dict[str, Any]]]
self, grouped_recordings: defaultdict[tuple[str, str], list[dict[str, Any]]]
) -> None:
expire_before = datetime.datetime.now().timestamp() - STALE_RECORDINGS_INFO_TTL
# a camera is still active when any of its streams cached segments
cameras_with_cache = {camera for camera, _ in grouped_recordings}
for recordings_info in (
self.object_recordings_info,
self.audio_recordings_info,
):
for camera in list(recordings_info.keys()):
if camera in grouped_recordings:
if camera in cameras_with_cache:
continue
info = recordings_info[camera]
while info and info[0][0] < expire_before:
@@ -337,64 +505,121 @@ class RecordingMaintainer(threading.Thread):
) -> dict[str, Any] | None:
cache_path: str = recording["cache_path"]
start_time: datetime.datetime = recording["start_time"]
stream_type: str = recording["stream_type"]
# Just delete files if camera removed or recordings are turned off
if (
camera not in self.config.cameras
or not self.config.cameras[camera].record.enabled
or (
stream_type == STREAM_TYPE_SUB
and not self.config.cameras[camera].record.sub.enabled
)
):
self.drop_segment(cache_path)
return None
if cache_path in self.end_time_cache:
end_time, duration = self.end_time_cache[cache_path]
(
end_time,
duration,
has_audio,
audio_rate,
audio_codec,
video_codec,
keyframes,
) = self.end_time_cache[cache_path]
# recover the resolved start rather than reusing the truncated
# filename timestamp
start_time = end_time - datetime.timedelta(seconds=duration)
else:
segment_info = await get_video_properties(
self.config.ffmpeg, cache_path, get_duration=True
)
async with self.probe_semaphore:
segment_info = await get_video_properties(
self.config.ffmpeg, cache_path, get_duration=True
)
if not segment_info.get("has_valid_video", False):
logger.warning(
f"Invalid or missing video stream in segment {cache_path}. Discarding."
)
self.recordings_publisher.publish(
(camera, 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
duration = float(segment_info.get("duration", -1))
has_audio = segment_info.get("has_audio")
audio_rate = segment_info.get("audio_rate")
audio_codec = segment_info.get("audio_codec")
video_codec = segment_info.get("video_codec")
# ensure duration is within expected length
if 0 < duration < MAX_SEGMENT_DURATION:
# playback snaps mid-file entry points against these offsets
# instead of probing files on demand
async with self.probe_semaphore:
keyframes = await get_keyframe_offsets(cache_path)
start_time = self._resolve_segment_start(
camera, stream_type, start_time, duration, cache_path
)
end_time = start_time + datetime.timedelta(seconds=duration)
self.end_time_cache[cache_path] = (end_time, duration)
self.end_time_cache[cache_path] = (
end_time,
duration,
has_audio,
audio_rate,
audio_codec,
video_codec,
keyframes,
)
# segments later discarded by retention still advance the
# chain for the next kept segment
self.last_segment_end[(camera, stream_type)] = end_time.timestamp()
else:
if duration == -1:
logger.warning(f"Failed to probe corrupt segment {cache_path}")
logger.warning(f"Discarding a corrupt recording segment: {cache_path}")
self.recordings_publisher.publish(
(camera, 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
# this segment has a valid duration and has video data, so publish an update
self.recordings_publisher.publish(
(camera, start_time.timestamp(), cache_path),
RecordingsDataTypeEnum.valid.value,
)
if stream_type == STREAM_TYPE_MAIN:
self.recordings_publisher.publish(
(camera, start_time.timestamp(), cache_path),
RecordingsDataTypeEnum.valid.value,
)
record_config = self.config.cameras[camera].record
# sub's alerts/detections carry the retain mode directly, unlike
# main's nested retain config
if stream_type == STREAM_TYPE_SUB:
continuous_days = record_config.sub.continuous.days
motion_days = record_config.sub.motion.days
alerts_retain_mode = record_config.sub.alerts.mode
detections_retain_mode = record_config.sub.detections.mode
else:
continuous_days = record_config.continuous.days
motion_days = record_config.motion.days
alerts_retain_mode = record_config.alerts.retain.mode
detections_retain_mode = record_config.detections.retain.mode
segment_stats: SegmentInfo | None = None
highest = None
if record_config.continuous.days > 0:
if continuous_days > 0:
highest = "continuous"
elif record_config.motion.days > 0:
elif motion_days > 0:
highest = "motion"
# if we have continuous or motion recording enabled
@@ -426,11 +651,17 @@ class RecordingMaintainer(threading.Thread):
if not segment_stats.should_discard_segment(record_mode):
return await self.move_segment(
camera,
stream_type,
start_time,
end_time,
duration,
cache_path,
segment_stats,
has_audio,
audio_rate,
audio_codec,
video_codec,
keyframes,
)
# we fell through the continuous / motion check, so we need to check the review items
@@ -459,9 +690,9 @@ class RecordingMaintainer(threading.Thread):
if overlaps:
record_mode = (
record_config.alerts.retain.mode
alerts_retain_mode
if review.severity == "alert"
else record_config.detections.retain.mode
else detections_retain_mode
)
if segment_stats is None:
@@ -471,11 +702,17 @@ class RecordingMaintainer(threading.Thread):
# move from cache to recordings immediately
return await self.move_segment(
camera,
stream_type,
start_time,
end_time,
duration,
cache_path,
segment_stats,
has_audio,
audio_rate,
audio_codec,
video_codec,
keyframes,
)
else:
self.drop_segment(cache_path)
@@ -614,17 +851,24 @@ class RecordingMaintainer(threading.Thread):
async def move_segment(
self,
camera: str,
stream_type: str,
start_time: datetime.datetime,
end_time: datetime.datetime,
duration: float,
cache_path: str,
segment_info: SegmentInfo,
has_audio: bool | None = None,
audio_rate: int | None = None,
audio_codec: str | None = None,
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
# sub segments get a tagged directory to avoid filename collisions
directory = os.path.join(
RECORD_DIR,
start_time.strftime("%Y-%m-%d/%H"),
camera,
camera if stream_type == STREAM_TYPE_MAIN else f"{camera}{SUB_CACHE_TAG}",
)
os.makedirs(directory, exist_ok=True)
@@ -684,6 +928,7 @@ class RecordingMaintainer(threading.Thread):
return {
Recordings.id.name: f"{start_time.timestamp()}-{rand_id}",
Recordings.camera.name: camera,
Recordings.stream_type.name: stream_type,
Recordings.path.name: file_path,
Recordings.start_time.name: start_time.timestamp(),
Recordings.end_time.name: end_time.timestamp(),
@@ -695,6 +940,11 @@ class RecordingMaintainer(threading.Thread):
Recordings.dBFS.name: segment_info.average_dBFS,
Recordings.segment_size.name: segment_size,
Recordings.motion_heatmap.name: segment_info.motion_heatmap,
Recordings.has_audio.name: has_audio,
Recordings.audio_rate.name: audio_rate,
Recordings.audio_codec.name: audio_codec,
Recordings.video_codec.name: video_codec,
Recordings.keyframes.name: keyframes,
}
except Exception as e:
logger.error(f"Unable to store recording segment {cache_path}")
@@ -788,11 +1038,10 @@ class RecordingMaintainer(threading.Thread):
try:
asyncio.run(self.move_files())
except Exception as e:
logger.error(
except Exception:
logger.exception(
"Error occurred when attempting to maintain recording cache"
)
logger.error(e)
duration = datetime.datetime.now().timestamp() - run_start
wait_time = max(0, 5 - duration)
frigate/storage.py
+42 -24
View File
@@ -9,7 +9,12 @@ from pathlib import Path
from peewee import SQL, fn
from frigate.config import FrigateConfig
from frigate.const import RECORD_DIR, REPLAY_CAMERA_PREFIX
from frigate.const import (
RECORD_DIR,
REPLAY_CAMERA_PREFIX,
STREAM_TYPE_MAIN,
STREAM_TYPE_SUB,
)
from frigate.models import Event, Recordings
from frigate.util.builtin import clear_and_unlink
@@ -49,30 +54,43 @@ class StorageMaintainer(threading.Thread):
)
}
# calculate MB/hr from last 100 segments
try:
# Subquery to get last 100 segments, then average their bandwidth
last_100 = (
Recordings.select(bandwidth_equation.alias("bw"))
.where(Recordings.camera == camera, Recordings.segment_size > 0)
.order_by(Recordings.start_time.desc())
.limit(100)
.alias("recent")
)
bandwidth = round(
Recordings.select(fn.AVG(SQL("bw"))).from_(last_100).scalar()
* 3600,
2,
)
if bandwidth > MAX_CALCULATED_BANDWIDTH:
logger.warning(
f"{camera} has a bandwidth of {bandwidth} MB/hr which exceeds the expected maximum. This typically indicates an issue with the cameras recordings."
# calculate MB/hr from the last 100 segments of each stream
# type and sum the rates; mixing streams would average small
# sub segments against large main segments and underestimate
# the true write rate
bandwidth = 0
for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB):
try:
# Subquery to get last 100 segments, then average their bandwidth
last_100 = (
Recordings.select(bandwidth_equation.alias("bw"))
.where(
Recordings.camera == camera,
Recordings.segment_size > 0,
Recordings.stream_type == stream_type,
)
.order_by(Recordings.start_time.desc())
.limit(100)
.alias("recent")
)
bandwidth = MAX_CALCULATED_BANDWIDTH
except TypeError:
bandwidth = 0
bandwidth += round(
Recordings.select(fn.AVG(SQL("bw")))
.from_(last_100)
.scalar()
* 3600,
2,
)
except TypeError:
pass
bandwidth = round(bandwidth, 2)
if bandwidth > MAX_CALCULATED_BANDWIDTH:
logger.warning(
f"{camera} has a bandwidth of {bandwidth} MB/hr which exceeds the expected maximum. This typically indicates an issue with the cameras recordings."
)
bandwidth = MAX_CALCULATED_BANDWIDTH
self.camera_storage_stats[camera]["bandwidth"] = bandwidth
logger.debug(f"{camera} has a bandwidth of {bandwidth} MiB/hr.")
frigate/test/http_api/test_http_media.py
+1227 -49
View File
File diff suppressed because it is too large Load Diff
frigate/test/test_camera_config_updater.py
+102
View File
@@ -0,0 +1,102 @@
"""Tests for dynamic camera config updates recreating ffmpeg commands."""
import unittest
from unittest.mock import patch
from frigate.config import CameraConfig, FrigateConfig
from frigate.config.camera.updater import (
CameraConfigUpdateEnum,
CameraConfigUpdateSubscriber,
)
from frigate.const import SUB_CACHE_TAG
def _build_camera_config(sub_enabled: bool) -> CameraConfig:
config = FrigateConfig(
**{
"mqtt": {"host": "mqtt"},
"cameras": {
"front_door": {
"ffmpeg": {
"inputs": [
{
"path": "rtsp://10.0.0.1:554/video",
"roles": ["detect", "record"],
},
{
"path": "rtsp://10.0.0.1:554/video2",
"roles": ["record_sub"],
},
]
},
"record": {"enabled": True, "sub": {"enabled": sub_enabled}},
}
},
}
)
return config.cameras["front_door"]
def _has_sub_output(camera_config: CameraConfig) -> bool:
return any(
SUB_CACHE_TAG in part for c in camera_config.ffmpeg_cmds for part in c["cmd"]
)
class TestRecordUpdateRecreatesFfmpegCmds(unittest.TestCase):
def setUp(self):
# avoid binding a real ZMQ socket; updates are fed directly through
# the mocked subscriber below
patcher = patch("frigate.config.camera.updater.ConfigSubscriber")
patcher.start()
self.addCleanup(patcher.stop)
def _push_record_update(
self, subscriber: CameraConfigUpdateSubscriber, record_config
) -> None:
subscriber.subscriber.check_for_update.side_effect = [
("config/cameras/front_door/record", record_config),
(None, None),
]
subscriber.check_for_updates()
def test_enabling_sub_recreates_ffmpeg_cmds(self):
camera_config = _build_camera_config(sub_enabled=False)
subscriber = CameraConfigUpdateSubscriber(
None, {"front_door": camera_config}, [CameraConfigUpdateEnum.record]
)
assert not _has_sub_output(camera_config)
self._push_record_update(
subscriber, _build_camera_config(sub_enabled=True).record
)
assert _has_sub_output(camera_config)
def test_disabling_sub_recreates_ffmpeg_cmds(self):
camera_config = _build_camera_config(sub_enabled=True)
subscriber = CameraConfigUpdateSubscriber(
None, {"front_door": camera_config}, [CameraConfigUpdateEnum.record]
)
assert _has_sub_output(camera_config)
self._push_record_update(
subscriber, _build_camera_config(sub_enabled=False).record
)
assert not _has_sub_output(camera_config)
def test_unchanged_record_update_keeps_existing_cmds(self):
camera_config = _build_camera_config(sub_enabled=False)
subscriber = CameraConfigUpdateSubscriber(
None, {"front_door": camera_config}, [CameraConfigUpdateEnum.record]
)
cmds_before = camera_config.ffmpeg_cmds
# neither enabled_in_config nor sub.enabled changed, so the
# commands should not be rebuilt
self._push_record_update(
subscriber, _build_camera_config(sub_enabled=False).record
)
assert camera_config.ffmpeg_cmds is cmds_before
frigate/test/test_config.py
+196 -1
View File
@@ -1,13 +1,14 @@
import json
import os
import unittest
from copy import deepcopy
from unittest.mock import patch
import numpy as np
from pydantic import ValidationError
from ruamel.yaml.constructor import DuplicateKeyError
from frigate.config import FrigateConfig
from frigate.config import FrigateConfig, RetainModeEnum
from frigate.const import MODEL_CACHE_DIR
from frigate.detectors import DetectorTypeEnum
from frigate.util.builtin import deep_merge
@@ -968,6 +969,162 @@ class TestConfig(unittest.TestCase):
assert len(ffmpeg_cmds) == 1
assert "clips" not in ffmpeg_cmds[0]["roles"]
def test_record_sub_cmd_writes_sub_cache_path(self):
config = {
"mqtt": {"host": "mqtt"},
"cameras": {
"back": {
"ffmpeg": {
"inputs": [
{
"path": "rtsp://10.0.0.1:554/video",
"roles": ["detect", "record"],
},
{
"path": "rtsp://10.0.0.1:554/video2",
"roles": ["record_sub"],
},
]
},
"record": {"enabled": True, "sub": {"enabled": True}},
}
},
}
frigate_config = FrigateConfig(**config)
cmds = frigate_config.cameras["back"].ffmpeg_cmds
sub_cmds = [c for c in cmds if "record_sub" in c["roles"]]
assert len(sub_cmds) == 1
joined = " ".join(sub_cmds[0]["cmd"])
assert "back@sub@" in joined
def test_record_sub_disabled_no_sub_cache_path(self):
config = {
"mqtt": {"host": "mqtt"},
"cameras": {
"back": {
"ffmpeg": {
"inputs": [
{
"path": "rtsp://10.0.0.1:554/video",
"roles": ["detect", "record"],
},
]
},
"record": {"enabled": True, "sub": {"enabled": False}},
}
},
}
frigate_config = FrigateConfig(**config)
cmds = frigate_config.cameras["back"].ffmpeg_cmds
assert all("@sub@" not in " ".join(c["cmd"]) for c in cmds)
def _sub_record_config(self, ffmpeg_extra: dict | None = None) -> dict:
return {
"mqtt": {"host": "mqtt"},
"cameras": {
"back": {
"ffmpeg": {
"inputs": [
{
"path": "rtsp://10.0.0.1:554/video",
"roles": ["detect", "record"],
},
{
"path": "rtsp://10.0.0.1:554/video2",
"roles": ["record_sub"],
},
],
**(ffmpeg_extra or {}),
},
"record": {"enabled": True, "sub": {"enabled": True}},
}
},
}
def _sub_record_cmd(self, config: dict) -> str:
cmds = FrigateConfig(**config).cameras["back"].ffmpeg_cmds
sub_cmds = [c for c in cmds if "record_sub" in c["roles"]]
assert len(sub_cmds) == 1
return " ".join(sub_cmds[0]["cmd"])
def test_record_sub_output_args_inherit_record(self):
config = self._sub_record_config(
{"output_args": {"record": "preset-record-generic-audio-copy"}}
)
cmd = self._sub_record_cmd(config)
# the customized record args, not the stock aac default
assert "-c copy" in cmd
assert "-c:a aac" not in cmd
def test_record_sub_output_args_override_record(self):
config = self._sub_record_config(
{
"output_args": {
"record": "preset-record-generic-audio-aac",
"record_sub": "preset-record-generic",
}
}
)
cmd = self._sub_record_cmd(config)
assert "-c copy -an" in cmd
assert "-c:a aac" not in cmd
def test_record_output_args_unaffected_by_record_sub(self):
config = self._sub_record_config(
{
"output_args": {
"record": "preset-record-generic-audio-aac",
"record_sub": "preset-record-generic",
}
}
)
cmds = FrigateConfig(**config).cameras["back"].ffmpeg_cmds
record_cmd = " ".join(next(c for c in cmds if "record" in c["roles"])["cmd"])
assert "-c:a aac" in record_cmd
def test_record_sub_manual_output_args(self):
config = self._sub_record_config(
{
"output_args": {
"record_sub": "-f segment -segment_time 10 -segment_format mp4 -reset_timestamps 1 -strftime 1 -c:v copy -c:a aac -ar 16000"
}
}
)
assert "-ar 16000" in self._sub_record_cmd(config)
def test_fails_on_bad_record_sub_segment_time(self):
config = self._sub_record_config(
{
"output_args": {
"record_sub": "-f segment -segment_time 70 -segment_format mp4 -reset_timestamps 1 -strftime 1 -c copy -an"
}
}
)
self.assertRaisesRegex(
ValueError,
"segment_time",
lambda: FrigateConfig(**config).cameras,
)
def test_record_sub_segment_time_not_checked_when_disabled(self):
config = self._sub_record_config(
{
"output_args": {
"record_sub": "-f segment -segment_time 70 -segment_format mp4 -reset_timestamps 1 -strftime 1 -c copy -an"
}
}
)
config["cameras"]["back"]["record"]["sub"]["enabled"] = False
FrigateConfig(**config).cameras
def test_max_disappeared_default(self):
config = {
"mqtt": {"host": "mqtt"},
@@ -1210,6 +1367,44 @@ class TestConfig(unittest.TestCase):
self.assertRaises(ValueError, lambda: FrigateConfig(**config))
def test_record_sub_config_defaults(self):
config = FrigateConfig(**self.minimal)
record = config.cameras["back"].record
assert record.sub.enabled is False
assert record.sub.continuous.days == 0
assert record.sub.alerts.mode == RetainModeEnum.motion
def test_record_sub_enabled_requires_role(self):
config = deepcopy(self.minimal)
config["cameras"]["back"]["ffmpeg"]["inputs"] = [
{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect", "record"]},
]
config["cameras"]["back"]["record"] = {
"enabled": True,
"sub": {"enabled": True},
}
# no record_sub role assigned -> must raise
self.assertRaisesRegex(
ValueError,
"record_sub is not assigned",
lambda: FrigateConfig(**config),
)
def test_record_sub_role_accepted(self):
config = deepcopy(self.minimal)
config["cameras"]["back"]["ffmpeg"]["inputs"] = [
{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect", "record"]},
{"path": "rtsp://10.0.0.1:554/video2", "roles": ["record_sub"]},
]
config["cameras"]["back"]["record"] = {
"enabled": True,
"sub": {"enabled": True, "continuous": {"days": 30}},
}
parsed = FrigateConfig(**config)
assert parsed.cameras["back"].record.sub.continuous.days == 30
def test_works_on_missing_role_multiple_cams(self):
config = {
"mqtt": {"host": "mqtt"},
frigate/test/test_event_cleanup_sub.py
+127
View File
@@ -0,0 +1,127 @@
"""Tests for sub stream retention extending event clip lifetimes."""
import datetime
import unittest
from unittest.mock import MagicMock
from playhouse.sqlite_ext import SqliteExtDatabase
from frigate.config import FrigateConfig
from frigate.events.cleanup import EventCleanup
from frigate.models import Event, Timeline
class TestEventCleanupSubRetention(unittest.TestCase):
def setUp(self):
# in-memory database keeps these tests isolated from the shared
# on-disk test.db used by the http api tests
self.db = SqliteExtDatabase(":memory:")
models = [Event, Timeline]
self.db.bind(models)
self.db.create_tables(models)
def tearDown(self):
self.db.close()
def _build_cleanup(self, record_config: dict) -> EventCleanup:
config = FrigateConfig(
**{
"mqtt": {"host": "mqtt"},
"cameras": {
"front_door": {
"ffmpeg": {
"inputs": [
{
"path": "rtsp://10.0.0.1:554/video",
"roles": ["detect", "record"],
},
{
"path": "rtsp://10.0.0.1:554/video2",
"roles": ["record_sub"],
},
]
},
"record": record_config,
}
},
}
)
return EventCleanup(config, MagicMock(), MagicMock())
def _insert_event(self, id: str, age_days: float, severity: str = "alert") -> None:
end_time = (
datetime.datetime.now() - datetime.timedelta(days=age_days)
).timestamp()
Event.create(
id=id,
label="person",
camera="front_door",
start_time=end_time - 10,
end_time=end_time,
top_score=0.9,
score=0.9,
false_positive=False,
zones=[],
thumbnail="",
has_clip=True,
has_snapshot=False,
region=[],
box=[],
area=0,
retain_indefinitely=False,
plus_id="",
model_hash="",
detector_type="cpu",
model_type="ssd",
data={"max_severity": severity},
)
def test_sub_alerts_days_extends_event_clip_retention(self):
# a 20-day-old alert event keeps its clip for the 60 day sub window
# so Explore stays coherent with the surviving sub recordings
cleanup = self._build_cleanup(
{
"enabled": True,
"alerts": {"retain": {"days": 10}},
"sub": {"enabled": True, "alerts": {"days": 60}},
}
)
self._insert_event("e1", 20)
expired = cleanup.expire_clips()
assert "e1" not in expired
assert Event.get(Event.id == "e1").has_clip is True
def test_event_clip_expires_when_sub_disabled(self):
# with sub recording disabled, the 20-day-old alert event expires
# under the 10 day main alerts retention exactly as before
cleanup = self._build_cleanup(
{
"enabled": True,
"alerts": {"retain": {"days": 10}},
"sub": {"enabled": False, "alerts": {"days": 60}},
}
)
self._insert_event("e1", 20)
expired = cleanup.expire_clips()
assert "e1" in expired
assert Event.get(Event.id == "e1").has_clip is False
def test_sub_detections_days_extends_event_clip_retention(self):
# detection severity uses the sub detections window
cleanup = self._build_cleanup(
{
"enabled": True,
"detections": {"retain": {"days": 10}},
"sub": {"enabled": True, "detections": {"days": 60}},
}
)
self._insert_event("e1", 20, severity="detection")
expired = cleanup.expire_clips()
assert "e1" not in expired
assert Event.get(Event.id == "e1").has_clip is True
frigate/test/test_maintainer.py
+10 -3
View File
@@ -98,7 +98,9 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
end_time = now - datetime.timedelta(seconds=10)
cache_path = "/tmp/cache/test_cam@20260417150000+0000.mp4"
maintainer.end_time_cache = {cache_path: (end_time, 10.0)}
maintainer.end_time_cache = {
cache_path: (end_time, 10.0, None, None, None, None, None)
}
# Single processed frame well past end_time with no motion/objects.
maintainer.object_recordings_info["test_cam"] = [(now.timestamp(), [], [], [])]
maintainer.audio_recordings_info["test_cam"] = []
@@ -109,7 +111,11 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
result = await maintainer.validate_and_move_segment(
"test_cam",
reviews=[],
recording={"start_time": start_time, "cache_path": cache_path},
recording={
"start_time": start_time,
"cache_path": cache_path,
"stream_type": "main",
},
)
self.assertIsNone(result)
@@ -137,7 +143,8 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
(recent, 0, []),
]
grouped_recordings = {"present_cam": [{"start_time": ancient}]}
# keyed by (camera, stream_type), matching what move_files passes
grouped_recordings = {("present_cam", "main"): [{"start_time": ancient}]}
maintainer._expire_stale_recordings_info(grouped_recordings)
frigate/test/test_record_cleanup_sub.py
+210
View File
@@ -0,0 +1,210 @@
"""Tests for independent sub stream retention in recording cleanup."""
import datetime
import unittest
from unittest.mock import MagicMock
from playhouse.sqlite_ext import SqliteExtDatabase
from frigate.config import FrigateConfig
from frigate.models import Previews, Recordings, ReviewSegment, UserReviewStatus
from frigate.record.cleanup import RecordingCleanup
class TestRecordingCleanupSubRetention(unittest.TestCase):
def setUp(self):
# in-memory database keeps these tests isolated from the shared
# on-disk test.db used by the http api tests
self.db = SqliteExtDatabase(":memory:")
models = [Previews, Recordings, ReviewSegment, UserReviewStatus]
self.db.bind(models)
self.db.create_tables(models)
def tearDown(self):
self.db.close()
def _build_cleanup(self, record_config: dict) -> RecordingCleanup:
config = FrigateConfig(
**{
"mqtt": {"host": "mqtt"},
"cameras": {
"front_door": {
"ffmpeg": {
"inputs": [
{
"path": "rtsp://10.0.0.1:554/video",
"roles": ["detect", "record"],
},
{
"path": "rtsp://10.0.0.1:554/video2",
"roles": ["record_sub"],
},
]
},
"record": record_config,
}
},
}
)
return RecordingCleanup(config, MagicMock())
def _insert_recording(
self, id: str, stream_type: str, age_days: float, motion: int = 0
) -> None:
end_time = (
datetime.datetime.now() - datetime.timedelta(days=age_days)
).timestamp()
Recordings.create(
id=id,
camera="front_door",
path=f"/media/frigate/recordings/{id}.mp4",
start_time=end_time - 10,
end_time=end_time,
duration=10,
motion=motion,
objects=0,
dBFS=0,
segment_size=0,
stream_type=stream_type,
)
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
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)
cleanup.expire_recordings()
assert Recordings.get_or_none(Recordings.id == "m1") is None
assert Recordings.get_or_none(Recordings.id == "s1") 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(
{
"enabled": True,
"continuous": {"days": 7},
"sub": {"enabled": True, "continuous": {"days": 30}},
}
)
self._insert_recording("s_old", "sub", 40)
self._insert_recording("s_new", "sub", 10)
cleanup.expire_recordings()
assert Recordings.get_or_none(Recordings.id == "s_old") is None
assert Recordings.get_or_none(Recordings.id == "s_new") is not None
def test_sub_recordings_overlapping_recent_reviews_survive(self):
# sub retention shorter than main: the reviews window must reach the
# sub pass cutoff or sub segments overlapping recent alerts are
# deleted on the first cleanup after being stored
cleanup = self._build_cleanup(
{
"enabled": True,
"continuous": {"days": 7},
"sub": {"enabled": True, "alerts": {"mode": "motion"}},
}
)
self._insert_recording("m1", "main", 1, motion=10)
self._insert_recording("s1", "sub", 1, motion=10)
# alert review covering the same time window as the recordings
recording = Recordings.get(Recordings.id == "s1")
ReviewSegment.create(
id="r1",
camera="front_door",
start_time=recording.start_time,
end_time=recording.end_time,
severity="alert",
thumb_path="/media/frigate/clips/review/thumb-r1.webp",
data={},
)
cleanup.expire_recordings()
assert Recordings.get_or_none(Recordings.id == "m1") is not None
assert Recordings.get_or_none(Recordings.id == "s1") is not None
def test_sub_alerts_days_extends_review_and_sub_retention(self):
# a 20-day-old alert review survives for the 60 day sub window and
# keeps the sub row alive, but must not hold the main row past the
# 10 day main alerts window
cleanup = self._build_cleanup(
{
"enabled": True,
"continuous": {"days": 7},
"sub": {"enabled": True, "alerts": {"days": 60}},
}
)
self._insert_recording("m1", "main", 20, motion=10)
self._insert_recording("s1", "sub", 20, motion=10)
# alert review covering the same time window as the recordings
recording = Recordings.get(Recordings.id == "s1")
ReviewSegment.create(
id="r1",
camera="front_door",
start_time=recording.start_time,
end_time=recording.end_time,
severity="alert",
thumb_path="/media/frigate/clips/review/thumb-r1.webp",
data={},
)
cleanup.expire_recordings()
assert ReviewSegment.get_or_none(ReviewSegment.id == "r1") is not None
assert Recordings.get_or_none(Recordings.id == "m1") is None
assert Recordings.get_or_none(Recordings.id == "s1") is not None
def test_main_review_lifetime_unchanged_when_sub_disabled(self):
# with sub recording disabled, the 20-day-old alert review expires
# under the 10 day main alerts retention exactly as before
cleanup = self._build_cleanup(
{
"enabled": True,
"continuous": {"days": 7},
"sub": {"enabled": False, "alerts": {"days": 60}},
}
)
end_time = (datetime.datetime.now() - datetime.timedelta(days=20)).timestamp()
ReviewSegment.create(
id="r1",
camera="front_door",
start_time=end_time - 10,
end_time=end_time,
severity="alert",
thumb_path="/media/frigate/clips/review/thumb-r1.webp",
data={},
)
cleanup.expire_recordings()
assert ReviewSegment.get_or_none(ReviewSegment.id == "r1") is None
def test_disabled_sub_still_expires_old_sub_rows(self):
# sub disabled but old sub rows remain: the sub pass still runs and
# expires them by the sub config dates
cleanup = self._build_cleanup(
{
"enabled": True,
"continuous": {"days": 7},
"sub": {"enabled": False, "continuous": {"days": 30}},
}
)
self._insert_recording("s_old", "sub", 40)
self._insert_recording("s_new", "sub", 10)
cleanup.expire_recordings()
assert Recordings.get_or_none(Recordings.id == "s_old") is None
assert Recordings.get_or_none(Recordings.id == "s_new") is not None
frigate/test/test_record_sub_maintainer.py
+544
View File
@@ -0,0 +1,544 @@
"""Tests for sub stream cache segment handling in the recording maintainer."""
import datetime
import os
import tempfile
import unittest
from collections import defaultdict
from unittest.mock import AsyncMock, MagicMock, patch
from playhouse.sqlite_ext import SqliteExtDatabase
from frigate.config import FrigateConfig
from frigate.models import Recordings
from frigate.record.maintainer import (
RecordingMaintainer,
SegmentInfo,
parse_cache_segment_name,
)
def _build_chaining_maintainer(
t0: float, retention: str = "continuous"
) -> RecordingMaintainer:
"""Build a maintainer for start-time chaining tests.
Built without __init__ to avoid the IPC scaffolding; move_segment and
drop_segment are mocked so the times passed downstream are observable.
"""
camera_config = MagicMock()
camera_config.record.enabled = True
camera_config.record.event_pre_capture = 0
camera_config.detect.width = 1920
camera_config.detect.height = 1080
camera_config.record.continuous.days = 1 if retention == "continuous" else 0
camera_config.record.motion.days = 0 if retention == "continuous" else 1
# no spec: pydantic fields like config.ffmpeg are not visible to
# spec'd mocks, and the probe path needs them
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.drop_segment = MagicMock()
maintainer.move_segment = AsyncMock(return_value=None)
# pre-seed the chain state so the lazy DB seed query is not attempted
maintainer.last_segment_end = {("test_cam", "main"): 0.0}
# a processed frame far past any segment end marks segments ready
maintainer.object_recordings_info["test_cam"] = [(t0 + 1000, [], [], [])]
return maintainer
async def _validate_segment(
maintainer: RecordingMaintainer,
t0: float,
offset: float,
duration: float,
mtime: float | None = None,
):
"""Run validate_and_move_segment for a segment named t0+offset.
mtime injects the cache file's close time; None simulates a missing file.
"""
start = datetime.datetime.fromtimestamp(t0 + offset, tz=datetime.UTC)
probe = AsyncMock(return_value={"has_valid_video": True, "duration": duration})
getmtime = (
MagicMock(side_effect=OSError("missing"))
if mtime is None
else MagicMock(return_value=mtime)
)
with (
patch("frigate.record.maintainer.get_video_properties", probe),
patch(
"frigate.record.maintainer.get_keyframe_offsets",
AsyncMock(return_value=[0]),
),
patch("frigate.record.maintainer.os.path.getmtime", getmtime),
):
return await maintainer.validate_and_move_segment(
"test_cam",
reviews=[],
recording={
"start_time": start,
"cache_path": f"/tmp/cache/test_cam@chain{offset}.mp4",
"stream_type": "main",
},
)
class TestParseCacheSegmentName(unittest.TestCase):
def test_parses_main_segment(self):
camera, stream_type, date = parse_cache_segment_name(
"front_door@20260610143022+0000"
)
assert camera == "front_door"
assert stream_type == "main"
assert date == "20260610143022+0000"
def test_parses_sub_segment(self):
camera, stream_type, date = parse_cache_segment_name(
"front_door@sub@20260610143022+0000"
)
assert camera == "front_door"
assert stream_type == "sub"
assert date == "20260610143022+0000"
def test_returns_none_for_unexpected_name(self):
assert parse_cache_segment_name("no_at_sign_here") is None
class TestValidateAndMoveSubSegment(unittest.IsolatedAsyncioTestCase):
"""Behavioral tests for sub stream segments in validate_and_move_segment.
The maintainer is built without __init__ to avoid the ZMQ scaffolding
the real constructor sets up.
"""
def _build_maintainer(self, camera_config: MagicMock) -> RecordingMaintainer:
config = MagicMock(spec=FrigateConfig)
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.drop_segment = MagicMock()
maintainer.recordings_publisher = MagicMock()
maintainer.move_segment = AsyncMock(return_value=None)
return maintainer
async def test_drops_sub_segment_when_sub_recording_disabled(self):
camera_config = MagicMock()
camera_config.record.enabled = True
camera_config.record.sub.enabled = False
maintainer = self._build_maintainer(camera_config)
cache_path = "/tmp/cache/test_cam@sub@20260610143022+0000.mp4"
result = await maintainer.validate_and_move_segment(
"test_cam",
reviews=[],
recording={
"start_time": datetime.datetime.now(datetime.UTC),
"cache_path": cache_path,
"stream_type": "sub",
},
)
self.assertIsNone(result)
maintainer.drop_segment.assert_called_once_with(cache_path)
maintainer.move_segment.assert_not_awaited()
async def test_sub_segment_uses_sub_continuous_retention(self):
# main continuous/motion are disabled, but sub continuous is enabled;
# the sub segment must be kept based on the sub retention config
camera_config = MagicMock()
camera_config.record.enabled = True
camera_config.record.sub.enabled = True
camera_config.record.continuous.days = 0
camera_config.record.motion.days = 0
camera_config.record.sub.continuous.days = 1
camera_config.record.sub.motion.days = 0
maintainer = self._build_maintainer(camera_config)
now = datetime.datetime.now(datetime.UTC)
start_time = now - datetime.timedelta(seconds=20)
end_time = now - datetime.timedelta(seconds=10)
cache_path = "/tmp/cache/test_cam@sub@20260610143022+0000.mp4"
# pre-fill the end time cache so no ffprobe is attempted; the
# audio and codec fields stay unknown without one
maintainer.end_time_cache = {
cache_path: (end_time, 10.0, None, None, None, None, None)
}
# a processed frame past end_time so the segment is considered ready
maintainer.object_recordings_info["test_cam"] = [(now.timestamp(), [], [], [])]
result = await maintainer.validate_and_move_segment(
"test_cam",
reviews=[],
recording={
"start_time": start_time,
"cache_path": cache_path,
"stream_type": "sub",
},
)
self.assertIsNone(result)
maintainer.move_segment.assert_awaited_once()
call_args = maintainer.move_segment.await_args.args
self.assertEqual(call_args[0], "test_cam")
self.assertEqual(call_args[1], "sub")
maintainer.drop_segment.assert_not_called()
class TestSegmentAudioPresence(unittest.IsolatedAsyncioTestCase):
"""Audio presence must flow from the segment probe into the DB insert."""
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
# no spec: pydantic fields like config.ffmpeg are not visible to
# spec'd mocks, and the probe/move paths need them
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()
# pre-seed the chain state so the lazy DB seed query is not attempted
maintainer.last_segment_end = {("test_cam", "main"): 0.0}
return maintainer
async def test_probe_audio_presence_reaches_move_segment(self):
for has_audio, audio_rate, audio_codec, video_codec in (
(True, 16000, "aac", "hevc"),
(True, 8000, "pcm_alaw", "h264"),
(False, None, None, "h264"),
(False, None, None, None),
):
with self.subTest(
has_audio=has_audio,
audio_rate=audio_rate,
audio_codec=audio_codec,
video_codec=video_codec,
):
maintainer = self._build_maintainer()
maintainer.move_segment = AsyncMock(return_value=None)
now = datetime.datetime.now(datetime.UTC)
start_time = now - datetime.timedelta(seconds=20)
cache_path = "/tmp/cache/test_cam@20260610143022+0000.mp4"
# a processed frame past the segment end marks it ready
maintainer.object_recordings_info["test_cam"] = [
(now.timestamp(), [], [], [])
]
probe = AsyncMock(
return_value={
"has_valid_video": True,
"width": 1920,
"height": 1080,
"duration": 10.0,
"has_audio": has_audio,
"audio_rate": audio_rate,
"audio_codec": audio_codec,
"video_codec": video_codec,
}
)
with (
patch("frigate.record.maintainer.get_video_properties", probe),
patch(
"frigate.record.maintainer.get_keyframe_offsets",
AsyncMock(return_value=[0, 2000]),
),
):
await maintainer.validate_and_move_segment(
"test_cam",
reviews=[],
recording={
"start_time": start_time,
"cache_path": cache_path,
"stream_type": "main",
},
)
maintainer.move_segment.assert_awaited_once()
call_args = maintainer.move_segment.await_args.args
self.assertEqual(call_args[7], has_audio)
self.assertEqual(call_args[8], audio_rate)
self.assertEqual(call_args[9], audio_codec)
self.assertEqual(call_args[11], [0, 2000])
self.assertEqual(call_args[10], video_codec)
# the probe result is cached alongside the end time so the
# cached path stays as informed as the probed path
self.assertEqual(maintainer.end_time_cache[cache_path][2], has_audio)
self.assertEqual(maintainer.end_time_cache[cache_path][3], audio_rate)
self.assertEqual(maintainer.end_time_cache[cache_path][4], audio_codec)
self.assertEqual(maintainer.end_time_cache[cache_path][5], video_codec)
async def test_move_segment_insert_includes_has_audio(self):
for has_audio, audio_rate, audio_codec, video_codec in (
(True, 16000, "aac", "hevc"),
(False, None, None, None),
):
with self.subTest(
has_audio=has_audio,
audio_rate=audio_rate,
audio_codec=audio_codec,
video_codec=video_codec,
):
maintainer = self._build_maintainer()
maintainer.config.ffmpeg.ffmpeg_path = "ffmpeg"
start_time = datetime.datetime.now(datetime.UTC)
end_time = start_time + datetime.timedelta(seconds=10)
proc = MagicMock()
proc.returncode = 0
proc.wait = AsyncMock(return_value=0)
with tempfile.TemporaryDirectory() as tmpdir:
cache_path = os.path.join(
tmpdir, "test_cam@20260610143022+0000.mp4"
)
with open(cache_path, "wb") as f:
f.write(b"\x00" * 16)
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,
end_time,
10.0,
cache_path,
SegmentInfo(0, 0, 0, 0),
has_audio,
audio_rate,
audio_codec,
video_codec,
)
self.assertIsNotNone(result)
self.assertEqual(result[Recordings.has_audio.name], has_audio)
self.assertEqual(result[Recordings.audio_rate.name], audio_rate)
self.assertEqual(result[Recordings.audio_codec.name], audio_codec)
self.assertEqual(result[Recordings.video_codec.name], video_codec)
class TestSegmentStartChaining(unittest.IsolatedAsyncioTestCase):
"""Contiguous segments must chain start times across filename truncation.
A contiguous segment's parsed start lands up to 1s before the previous
segment's fractional end, and must snap to it without ever snapping
across a genuine gap.
"""
T0 = datetime.datetime(2026, 6, 10, 14, 30, 22, tzinfo=datetime.UTC).timestamp()
async def test_contiguous_segment_snaps_to_previous_end(self):
maintainer = _build_chaining_maintainer(self.T0)
# segment A named at a whole second with a fractional duration
await _validate_segment(maintainer, self.T0, 0, 10.4)
# segment B's filename truncates its true start (T0 + 10.4) to T0 + 10
await _validate_segment(maintainer, self.T0, 10, 10.4)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 2)
self.assertEqual(calls[0].args[2].timestamp(), self.T0)
self.assertAlmostEqual(calls[0].args[3].timestamp(), self.T0 + 10.4, places=3)
self.assertAlmostEqual(calls[1].args[2].timestamp(), self.T0 + 10.4, places=3)
self.assertAlmostEqual(calls[1].args[3].timestamp(), self.T0 + 20.8, places=3)
async def test_genuine_gap_is_not_snapped(self):
maintainer = _build_chaining_maintainer(self.T0)
await _validate_segment(maintainer, self.T0, 0, 10.4)
# a segment starting well after the previous end is a genuine gap
await _validate_segment(maintainer, self.T0, 30, 10.4)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 2)
self.assertEqual(calls[1].args[2].timestamp(), self.T0 + 30)
self.assertAlmostEqual(calls[1].args[3].timestamp(), self.T0 + 40.4, places=3)
async def test_difference_over_tolerance_is_not_snapped(self):
# a last end far ahead of the parsed start (dual recorders or
# cache-pressure deletions) must never pull the start forward
maintainer = _build_chaining_maintainer(self.T0)
maintainer.last_segment_end = {("test_cam", "main"): self.T0 + 5.0}
await _validate_segment(maintainer, self.T0, 0, 10.4)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 1)
self.assertEqual(calls[0].args[2].timestamp(), self.T0)
async def test_start_after_last_end_within_tolerance_is_not_snapped(self):
maintainer = _build_chaining_maintainer(self.T0)
await _validate_segment(maintainer, self.T0, 0, 10.4)
# parsed start T0 + 11 is after the last end (T0 + 10.4): a real
# sub-second gap, never snapped backwards
await _validate_segment(maintainer, self.T0, 11, 10.4)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 2)
self.assertEqual(calls[1].args[2].timestamp(), self.T0 + 11)
async def test_chain_survives_retention_discarded_segment(self):
# motion retention: A and C overlap motion frames, B has none and is
# discarded, but capture continued through B so C chains to B's end
maintainer = _build_chaining_maintainer(self.T0, retention="motion")
maintainer.object_recordings_info["test_cam"] = [
(self.T0 + 5, [], [(0, 0, 10, 10)], []),
(self.T0 + 25, [], [(0, 0, 10, 10)], []),
(self.T0 + 1000, [], [], []),
]
await _validate_segment(maintainer, self.T0, 0, 10.4)
await _validate_segment(maintainer, self.T0, 10, 10.4)
await _validate_segment(maintainer, self.T0, 20, 10.4)
calls = maintainer.move_segment.await_args_list
# B was discarded by motion retention, only A and C moved
self.assertEqual(len(calls), 2)
maintainer.drop_segment.assert_called_once_with(
"/tmp/cache/test_cam@chain10.mp4"
)
# C snaps to B's end (T0 + 20.8), not A's end (T0 + 10.4)
self.assertAlmostEqual(calls[1].args[2].timestamp(), self.T0 + 20.8, places=3)
self.assertAlmostEqual(calls[1].args[3].timestamp(), self.T0 + 31.2, places=3)
class TestSegmentStartMtimeAnchoring(unittest.IsolatedAsyncioTestCase):
"""Segment starts must anchor to the cache file's close time.
mtime is the wall clock when ffmpeg rolled the segment, so mtime minus
the probed duration restores the fractional start the filename floored
away. Contiguous segments still chain within probe jitter.
"""
T0 = datetime.datetime(2026, 8, 7, 9, 15, 42, tzinfo=datetime.UTC).timestamp()
async def test_mtime_restores_fractional_start(self):
maintainer = _build_chaining_maintainer(self.T0)
# true start T0 + 0.437; the filename floored it to T0
await _validate_segment(
maintainer, self.T0, 0, 10.0, mtime=self.T0 + 0.437 + 10.0
)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 1)
self.assertAlmostEqual(calls[0].args[2].timestamp(), self.T0 + 0.437, places=3)
self.assertAlmostEqual(calls[0].args[3].timestamp(), self.T0 + 10.437, places=3)
async def test_chain_wins_within_probe_jitter(self):
maintainer = _build_chaining_maintainer(self.T0)
await _validate_segment(
maintainer, self.T0, 0, 10.0, mtime=self.T0 + 0.437 + 10.0
)
# B measures T0 + 10.420, a 17ms disagreement with A's chained end
# (T0 + 10.437): the chain wins so the rows stay exactly adjacent
await _validate_segment(
maintainer, self.T0, 10, 10.0, mtime=self.T0 + 10.420 + 10.0
)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 2)
self.assertAlmostEqual(calls[1].args[2].timestamp(), self.T0 + 10.437, places=3)
self.assertAlmostEqual(calls[1].args[3].timestamp(), self.T0 + 20.437, places=3)
async def test_chain_reanchors_when_drifted(self):
maintainer = _build_chaining_maintainer(self.T0)
# accumulated probe error left the chain 0.7s past the measured start
maintainer.last_segment_end = {("test_cam", "main"): self.T0 + 0.9}
await _validate_segment(
maintainer, self.T0, 0, 10.0, mtime=self.T0 + 0.2 + 10.0
)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 1)
self.assertAlmostEqual(calls[0].args[2].timestamp(), self.T0 + 0.2, places=3)
async def test_implausible_mtime_falls_back_to_filename_chain(self):
maintainer = _build_chaining_maintainer(self.T0)
maintainer.last_segment_end = {("test_cam", "main"): self.T0 + 0.4}
# a stalled stream: media (8s) is shorter than the wall span to the
# close time (9.5s), so mtime - duration lands outside the window
await _validate_segment(maintainer, self.T0, 0, 8.0, mtime=self.T0 + 9.5)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 1)
self.assertAlmostEqual(calls[0].args[2].timestamp(), self.T0 + 0.4, places=3)
class TestSegmentChainSeeding(unittest.IsolatedAsyncioTestCase):
"""Chain state must lazily seed from the DB so chains survive restarts."""
T0 = datetime.datetime(2026, 6, 10, 14, 30, 22, tzinfo=datetime.UTC).timestamp()
def setUp(self):
self.db = SqliteExtDatabase(":memory:")
self.db.bind([Recordings])
self.db.create_tables([Recordings])
def tearDown(self):
self.db.close()
async def test_seeds_chain_from_db_once(self):
maintainer = _build_chaining_maintainer(self.T0)
# empty chain state forces the lazy DB seed on first encounter
maintainer.last_segment_end = {}
Recordings.create(
id="seed-row",
camera="test_cam",
path="/recordings/seed.mp4",
start_time=self.T0 - 10,
end_time=self.T0 + 0.35,
duration=10.35,
stream_type="main",
)
with patch.object(Recordings, "select", wraps=Recordings.select) as select_spy:
await _validate_segment(maintainer, self.T0, 0, 10.4)
await _validate_segment(maintainer, self.T0, 10, 10.4)
# the seed query runs once; the second segment uses in-memory state
self.assertEqual(select_spy.call_count, 1)
calls = maintainer.move_segment.await_args_list
self.assertEqual(len(calls), 2)
# first segment snaps to the stored fractional end_time
self.assertAlmostEqual(calls[0].args[2].timestamp(), self.T0 + 0.35, places=3)
self.assertAlmostEqual(calls[0].args[3].timestamp(), self.T0 + 10.75, places=3)
# second segment chains off the first's in-memory end
self.assertAlmostEqual(calls[1].args[2].timestamp(), self.T0 + 10.75, places=3)
frigate/test/test_recording_coverage.py
+305
View File
@@ -0,0 +1,305 @@
"""Tests for the unified recording coverage resolver."""
import asyncio
import json
import unittest
from types import SimpleNamespace
from playhouse.sqlite_ext import SqliteExtDatabase
from frigate.api.media import _vod_response
from frigate.const import MAX_SEGMENT_DURATION
from frigate.models import Recordings
from frigate.util.recording_coverage import (
_rows_query,
coverage_spans,
plan_clip,
realized_timeline,
resolve_coverage,
stream_media_summary,
)
class CoverageDbTestCase(unittest.TestCase):
def setUp(self):
# in-memory database keeps these tests isolated from the shared
# on-disk test.db used by the http api tests
self.db = SqliteExtDatabase(":memory:")
models = [Recordings]
self.db.bind(models)
self.db.create_tables(models)
def tearDown(self):
self.db.close()
def _insert(
self,
id,
start,
end,
stream_type,
camera="front_door",
video_codec=None,
audio_rate=None,
audio_codec=None,
has_audio=None,
segment_size=0,
keyframes=None,
):
Recordings.create(
id=id,
camera=camera,
path=f"/tmp/{id}.mp4",
start_time=start,
end_time=end,
duration=end - start,
stream_type=stream_type,
video_codec=video_codec,
audio_rate=audio_rate,
audio_codec=audio_codec,
has_audio=has_audio,
segment_size=segment_size,
keyframes=keyframes,
)
class TestRecordingCoverage(CoverageDbTestCase):
def test_both_streams_offset_boundaries(self):
# main: [1000,1010) [1010,1020) sub: [1003,1013) [1013,1023)
self._insert("m1", 1000.0, 1010.0, "main")
self._insert("m2", 1010.0, 1020.0, "main")
self._insert("s1", 1003.0, 1013.0, "sub")
self._insert("s2", 1013.0, 1023.0, "sub")
intervals = resolve_coverage("front_door", 1000.0, 1020.0)
# boundaries: 1000,1003,1010,1013,1020
assert [round(i.start_time) for i in intervals] == [1000, 1003, 1010, 1013]
assert intervals[0].main is not None and intervals[0].sub is None
assert intervals[1].main is not None and intervals[1].sub is not None
def test_sub_only_when_main_expired(self):
self._insert("s1", 1000.0, 1010.0, "sub")
intervals = resolve_coverage("front_door", 1000.0, 1010.0)
assert len(intervals) == 1
assert intervals[0].main is None and intervals[0].sub is not None
def test_gap_when_neither_exists(self):
self._insert("m1", 1000.0, 1010.0, "main")
self._insert("m2", 1030.0, 1040.0, "main")
intervals = resolve_coverage("front_door", 1000.0, 1040.0)
assert len(intervals) == 2 # the [1010,1030) gap produces no interval
def test_spans_merge_same_availability(self):
self._insert("m1", 1000.0, 1010.0, "main")
self._insert("m2", 1010.0, 1020.0, "main")
spans = coverage_spans(resolve_coverage("front_door", 1000.0, 1020.0))
assert spans == [
{"start_time": 1000.0, "end_time": 1020.0, "streams": ["main"]}
]
def test_requested_range_clamps_intervals(self):
# query range partially overlaps the rows; the first and last
# intervals must be clamped to the requested [after, before]
self._insert("m1", 1000.0, 1010.0, "main")
self._insert("m2", 1010.0, 1020.0, "main")
intervals = resolve_coverage("front_door", 1005.0, 1015.0)
assert intervals[0].start_time == 1005.0
assert intervals[-1].end_time == 1015.0
spans = coverage_spans(intervals)
assert spans == [
{"start_time": 1005.0, "end_time": 1015.0, "streams": ["main"]}
]
def test_stream_media_summary_newest_known_wins(self):
# the newer main row's NULL codecs must not mask the older row's
# known codecs, while its known audio fields take precedence
self._insert(
"m1",
1000.0,
1010.0,
"main",
video_codec="hevc",
audio_rate=16000,
audio_codec="aac",
)
self._insert("m2", 1010.0, 1020.0, "main", audio_rate=8000, has_audio=True)
self._insert("s1", 1000.0, 1010.0, "sub", has_audio=False)
summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1020.0))
assert summary == {
"main": {
"video_codec": "hevc",
"audio_rate": 8000,
"audio_codec": "aac",
"has_audio": True,
"bitrate": None,
},
"sub": {
"video_codec": None,
"audio_rate": None,
"audio_codec": None,
"has_audio": False,
"bitrate": None,
},
}
def test_stream_media_summary_omits_absent_stream(self):
self._insert("m1", 1000.0, 1010.0, "main", video_codec="h264")
summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1010.0))
assert list(summary.keys()) == ["main"]
assert stream_media_summary([]) == {}
def test_stream_media_summary_bitrate_weighted_by_duration(self):
# main: 10 MiB over 10s + 30 MiB over 20s -> 40 MiB / 30s
self._insert("m1", 1000.0, 1010.0, "main", segment_size=10)
self._insert("m2", 1010.0, 1030.0, "main", segment_size=30)
# sub rows without a stored size (legacy default 0) report None
self._insert("s1", 1000.0, 1030.0, "sub")
summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1030.0))
expected = int(40 * 1024 * 1024 * 8 / 30)
assert summary["main"]["bitrate"] == expected
assert summary["sub"]["bitrate"] is None
def test_stream_media_summary_bitrate_skips_zero_size_rows(self):
# the zero-size glitch row contributes neither bytes nor seconds
self._insert("m1", 1000.0, 1010.0, "main", segment_size=5)
self._insert("m2", 1010.0, 1020.0, "main", segment_size=0)
summary = stream_media_summary(resolve_coverage("front_door", 1000.0, 1020.0))
expected = int(5 * 1024 * 1024 * 8 / 10)
assert summary["main"]["bitrate"] == expected
def test_other_camera_rows_excluded(self):
self._insert("m1", 1000.0, 1010.0, "main")
self._insert("o1", 1000.0, 1010.0, "main", camera="back_yard")
self._insert("o2", 1000.0, 1010.0, "sub", camera="back_yard")
intervals = resolve_coverage("front_door", 1000.0, 1010.0)
assert len(intervals) == 1
assert intervals[0].main is not None and intervals[0].sub is None
assert resolve_coverage("side_gate", 1000.0, 1010.0) == []
def test_realized_timeline_snap_lead_in(self):
"""A mid-file resume's back-snap lead-in appears in the realized duration.
Sub resumes at 1010 with a 7000ms inpoint; the stored keyframe
index snaps back to 6000ms, so the clip serves 1000ms of pre-span
content on top of the 3000ms span.
"""
self._insert("m1", 1000.0, 1010.0, "main")
self._insert("s1", 1003.0, 1013.0, "sub", keyframes=[0, 2000, 4000, 6000, 8000])
timeline = realized_timeline(
resolve_coverage("front_door", 1000.0, 1013.0), None
)
assert timeline == [
{"start_time": 1000.0, "end_time": 1010.0, "duration": 10000},
{"start_time": 1010.0, "end_time": 1013.0, "duration": 4000},
]
def test_realized_timeline_no_keyframe_index_serves_whole_file(self):
"""A row without a stored keyframe index serves the whole file.
NULL keyframes mean the record-time probe failed, so a mid-file
entry cannot be snapped to a decodable frame.
"""
self._insert("m1", 1000.0, 1010.0, "main")
self._insert("s1", 1003.0, 1013.0, "sub")
timeline = realized_timeline(
resolve_coverage("front_door", 1000.0, 1013.0), None
)
# the resume clip serves all 10000ms of the sub file for a 3000ms span
assert timeline[1] == {
"start_time": 1010.0,
"end_time": 1013.0,
"duration": 10000,
}
def test_realized_timeline_pinned_stream_no_mid_range_lead_in(self):
"""A pinned stream's spans start at its own file starts: no lead-in."""
self._insert("s1", 1000.0, 1010.0, "sub", keyframes=[0, 5000])
self._insert("s2", 1010.0, 1020.0, "sub", keyframes=[0, 5000])
timeline = realized_timeline(
resolve_coverage("front_door", 1000.0, 1020.0), "sub"
)
assert [t["duration"] for t in timeline] == [10000, 10000]
def test_rows_query_scan_is_bounded_below(self):
"""The row query must give SQLite a lower bound on start_time.
Without one the best index range is (camera=? AND start_time<?),
which walks the camera's entire history on every coverage call:
~20s on a cold page cache, blocking the API event loop.
"""
# the index the planner picks in production (migration 020)
self.db.execute_sql(
"CREATE INDEX recordings_api_recordings_summary ON recordings "
'("camera", "start_time" DESC, "duration", "motion", "objects")'
)
sql, params = _rows_query("front_door", 2000.0, 3000.0, "main").sql()
plan = " ".join(
row[-1] for row in self.db.execute_sql("EXPLAIN QUERY PLAN " + sql, params)
)
assert "start_time>?" in plan and "start_time<?" in plan, plan
def test_longest_segment_spanning_window_start_included(self):
# a maximum-length segment overlapping `after` sits right at the
# start_time lower bound the query is allowed to apply
self._insert("m1", 1000.0, 1000.0 + MAX_SEGMENT_DURATION - 0.5, "main")
intervals = resolve_coverage("front_door", 1599.0, 1700.0)
assert len(intervals) == 1
assert intervals[0].main is not None
def test_plan_clip_skips_too_short(self):
"""A clip left with under 100ms of media is skipped (duration 0 in timelines)."""
row = SimpleNamespace(
path="/tmp/x.mp4",
start_time=1000.0,
end_time=1000.15,
duration=0.15,
keyframes=[0, 100],
)
plan = plan_clip(row, 1000.1, 1000.15)
assert plan.skipped
assert plan.duration_ms == 0
class TestVodManifestPolicy(CoverageDbTestCase):
"""Discontinuity policy of the vod mapping builder."""
def _mapping(self, start, end, stream=None):
response = asyncio.run(
_vod_response("front_door", start, end, stream_preference=stream)
)
return json.loads(response.body)
def _insert_stream(self, prefix, stream_type, spans):
for idx, (start, end) in enumerate(spans):
self._insert(
f"{prefix}{idx}",
start,
end,
stream_type,
video_codec="h264",
audio_codec="aac",
audio_rate=16000,
has_audio=True,
)
def test_cross_stream_handoff_forces_discontinuity(self):
# codec and audio params match on both streams, so only the stream
# mix can force per-clip init segments here
self._insert_stream("m", "main", [(1000.0, 1010.0)])
self._insert_stream("s", "sub", [(1000.0, 1010.0), (1010.0, 1020.0)])
mapping = self._mapping(1000.0, 1020.0)
assert mapping["discontinuity"] is True
assert mapping["initialClipIndex"] == 1
assert len(mapping["sequences"][0]["clips"]) == 2
def test_single_stream_range_stays_consistent(self):
self._insert_stream("s", "sub", [(1000.0, 1010.0), (1010.0, 1020.0)])
mapping = self._mapping(1000.0, 1020.0)
assert mapping["discontinuity"] is False
assert "initialClipIndex" not in mapping
def test_pinned_stream_over_mixed_coverage_stays_consistent(self):
self._insert_stream("m", "main", [(1000.0, 1010.0), (1010.0, 1020.0)])
self._insert_stream("s", "sub", [(1000.0, 1010.0), (1010.0, 1020.0)])
mapping = self._mapping(1000.0, 1020.0, stream="sub")
assert mapping["discontinuity"] is False
assert "initialClipIndex" not in mapping
frigate/util/audio.py
+31 -15
View File
@@ -6,12 +6,33 @@ import subprocess as sp
from pathvalidate import sanitize_filename
from frigate.const import CACHE_DIR
from frigate.const import CACHE_DIR, STREAM_TYPE_MAIN, STREAM_TYPE_SUB
from frigate.models import Recordings
logger = logging.getLogger(__name__)
def _get_recordings_for_range(
camera_name: str, start_ts: float, end_ts: float, stream_type: str
) -> list[Recordings]:
"""Fetch one stream type's recording rows overlapping the requested range."""
return list(
Recordings.select(
Recordings.path,
Recordings.start_time,
Recordings.end_time,
)
.where(
(Recordings.start_time.between(start_ts, end_ts))
| (Recordings.end_time.between(start_ts, end_ts))
| ((start_ts > Recordings.start_time) & (end_ts < Recordings.end_time))
)
.where(Recordings.camera == camera_name)
.where(Recordings.stream_type == stream_type)
.order_by(Recordings.start_time.asc())
)
def get_audio_from_recording(
ffmpeg,
camera_name: str,
@@ -31,22 +52,17 @@ def get_audio_from_recording(
Returns:
Bytes of WAV audio data or None if extraction failed
"""
# Fetch all relevant recording segments
recordings = (
Recordings.select(
Recordings.path,
Recordings.start_time,
Recordings.end_time,
)
.where(
(Recordings.start_time.between(start_ts, end_ts))
| (Recordings.end_time.between(start_ts, end_ts))
| ((start_ts > Recordings.start_time) & (end_ts < Recordings.end_time))
)
.where(Recordings.camera == camera_name)
.order_by(Recordings.start_time.asc())
# Fetch all relevant recording segments; never mix streams in one
# concat, so prefer main and fall back to sub for expired-main history
recordings = _get_recordings_for_range(
camera_name, start_ts, end_ts, STREAM_TYPE_MAIN
)
if not recordings:
recordings = _get_recordings_for_range(
camera_name, start_ts, end_ts, STREAM_TYPE_SUB
)
if not recordings:
logger.debug(
f"No recordings found for {camera_name} between {start_ts} and {end_ts}"
frigate/util/classification.py
+2
View File
@@ -18,6 +18,7 @@ from frigate.const import (
CLIPS_DIR,
MODEL_CACHE_DIR,
PROCESS_PRIORITY_LOW,
STREAM_TYPE_MAIN,
UPDATE_MODEL_STATE,
)
from frigate.log import redirect_output_to_logger, suppress_stderr_during
@@ -555,6 +556,7 @@ def _extract_keyframes(
(timestamp >= Recordings.start_time)
& (timestamp <= Recordings.end_time)
& (Recordings.camera == camera)
& (Recordings.stream_type == STREAM_TYPE_MAIN)
)
.order_by(Recordings.start_time.desc())
.limit(1)
frigate/util/media.py
+33 -31
View File
@@ -1,10 +1,11 @@
"""Recordings Utilities."""
import asyncio
import contextlib
import datetime
import errno
import logging
import os
import subprocess as sp
from collections.abc import Iterable
from dataclasses import dataclass, field
from pathlib import Path
@@ -879,38 +880,43 @@ def sync_all_media(
return results
def get_keyframe_before(path: str, offset_ms: int) -> int | None:
"""Get the timestamp (ms) of the last keyframe at or before offset_ms.
async def get_keyframe_offsets(path: str) -> list[int] | None:
"""Get every video keyframe offset (ms from segment start) in an mp4.
Uses ffprobe packet index to read keyframe positions from the mp4 file.
Returns None if ffprobe fails or no keyframe is found before the offset.
Runs at record time so playback never needs to probe. Returns None if
ffprobe fails, so the caller stores NULL and playback falls back to
serving whole files.
"""
proc = None
try:
result = sp.run(
[
FFPROBE_PATH,
"-select_streams",
"v:0",
"-show_entries",
"packet=pts_time,flags",
"-of",
"csv=p=0",
"-loglevel",
"error",
path,
],
capture_output=True,
timeout=5,
proc = await asyncio.create_subprocess_exec(
FFPROBE_PATH,
"-select_streams",
"v:0",
"-show_entries",
"packet=pts_time,flags",
"-of",
"csv=p=0",
"-loglevel",
"error",
path,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
)
except (sp.TimeoutExpired, FileNotFoundError):
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=5)
except (TimeoutError, FileNotFoundError):
if proc is not None and proc.returncode is None:
with contextlib.suppress(ProcessLookupError):
proc.kill()
with contextlib.suppress(TimeoutError):
await asyncio.wait_for(proc.communicate(), timeout=2)
return None
if result.returncode != 0:
if proc.returncode != 0:
return None
offset_s = offset_ms / 1000.0
best_ms = None
for line in result.stdout.decode().strip().splitlines():
offsets: list[int] = []
for line in stdout.decode().strip().splitlines():
parts = line.strip().split(",")
if len(parts) != 2:
continue
@@ -918,12 +924,8 @@ def get_keyframe_before(path: str, offset_ms: int) -> int | None:
if "K" not in flags:
continue
try:
ts = float(ts_str)
offsets.append(int(float(ts_str) * 1000))
except ValueError:
continue
if ts <= offset_s:
best_ms = int(ts * 1000)
else:
break
return best_ms
return offsets
frigate/util/recording_coverage.py
+422
View File
@@ -0,0 +1,422 @@
"""Merge main and sub stream recording rows into a unified coverage timeline."""
import logging
from dataclasses import dataclass
from typing import Any
from frigate.const import MAX_SEGMENT_DURATION, STREAM_TYPE_MAIN, STREAM_TYPE_SUB
from frigate.models import Recordings
logger = logging.getLogger(__name__)
# intervals shorter than this are boundary artifacts, not playable content
MIN_INTERVAL_S = 0.1
@dataclass
class CoverageInterval:
"""A time span annotated with the recording row covering it per stream."""
start_time: float
end_time: float
main: Any | None
sub: Any | None
def _rows_query(camera: str, after: float, before: float, stream_type: str) -> Any:
return (
Recordings.select(
Recordings.path,
Recordings.start_time,
Recordings.end_time,
Recordings.duration,
Recordings.has_audio,
Recordings.audio_rate,
Recordings.audio_codec,
Recordings.video_codec,
Recordings.segment_size,
Recordings.keyframes,
)
.where(
(Recordings.camera == camera)
& (Recordings.stream_type == stream_type)
& (Recordings.end_time > after)
& (Recordings.start_time < before)
& (Recordings.start_time > after - MAX_SEGMENT_DURATION)
)
.order_by(Recordings.start_time.asc())
.namedtuples()
)
def _get_rows(camera: str, after: float, before: float, stream_type: str) -> list[Any]:
return list(_rows_query(camera, after, before, stream_type))
def _covering(
rows: list[Any], idx: int, start: float, end: float
) -> tuple[Any | None, int]:
"""Find the row covering [start, end), advancing idx (rows are sorted, non-overlapping)."""
while idx < len(rows) and rows[idx].end_time <= start:
idx += 1
if idx < len(rows) and rows[idx].start_time <= start and rows[idx].end_time >= end:
return rows[idx], idx
return None, idx
def resolve_coverage(
camera: str, after: float, before: float
) -> list[CoverageInterval]:
"""Resolve the unified coverage timeline for a camera and time range.
Returns intervals bounded by the union of both streams' segment edges,
clamped to [after, before]. Ranges covered by neither stream produce no
interval (a gap).
"""
main_rows = _get_rows(camera, after, before, STREAM_TYPE_MAIN)
sub_rows = _get_rows(camera, after, before, STREAM_TYPE_SUB)
boundaries: set[float] = set()
for row in main_rows + sub_rows:
boundaries.add(max(row.start_time, after))
boundaries.add(min(row.end_time, before))
ordered = sorted(boundaries)
intervals: list[CoverageInterval] = []
main_idx = 0
sub_idx = 0
for i in range(len(ordered) - 1):
start, end = ordered[i], ordered[i + 1]
if end - start < MIN_INTERVAL_S:
continue
main_row, main_idx = _covering(main_rows, main_idx, start, end)
sub_row, sub_idx = _covering(sub_rows, sub_idx, start, end)
if main_row is None and sub_row is None:
continue
intervals.append(CoverageInterval(start, end, main_row, sub_row))
return intervals
def known_video_codecs(intervals: list[CoverageInterval]) -> set[str]:
"""Collect the known video codecs across all rows in a coverage window.
NULL codecs (legacy rows probed before the column existed) are
excluded, so uniformly-unknown data reports an empty set.
"""
return {
row.video_codec
for interval in intervals
for row in (interval.main, interval.sub)
if row is not None and row.video_codec is not None
}
def stream_media_summary(
intervals: list[CoverageInterval],
) -> dict[str, dict[str, Any]]:
"""Summarize the most recently known media details per stream.
Every field reports the newest non-NULL value across that stream's
rows, so an older row can still supply a value a legacy newer row
lacks. A stream with no rows is omitted entirely.
"""
fields = ("video_codec", "audio_rate", "audio_codec", "has_audio")
summary: dict[str, dict[str, Any]] = {}
for stream_type in (STREAM_TYPE_MAIN, STREAM_TYPE_SUB):
# the same row can back many intervals, so dedupe by path
rows = {
row.path: row
for interval in intervals
if (row := getattr(interval, stream_type)) is not None
}
if not rows:
continue
newest_first = sorted(rows.values(), key=lambda r: r.start_time, reverse=True)
stream_summary: dict[str, Any] = {field: None for field in fields}
for field in fields:
for row in newest_first:
value = getattr(row, field)
if value is not None:
stream_summary[field] = value
break
# segment_size is stored in MiB; totalling bytes and seconds
# weights by duration, unlike averaging per-row ratios
total_bytes = 0.0
total_seconds = 0.0
for row in rows.values():
size = row.segment_size
if size is None or size <= 0 or row.duration is None or row.duration <= 0:
continue
total_bytes += size * 1024 * 1024
total_seconds += row.duration
stream_summary["bitrate"] = (
int(total_bytes * 8 / total_seconds) if total_seconds > 0 else None
)
summary[stream_type] = stream_summary
return summary
def coverage_spans(intervals: list[CoverageInterval]) -> list[dict[str, Any]]:
"""Collapse intervals into contiguous spans of identical stream availability."""
spans: list[dict[str, Any]] = []
for interval in intervals:
streams = [
t
for t, row in (
(STREAM_TYPE_MAIN, interval.main),
(STREAM_TYPE_SUB, interval.sub),
)
if row is not None
]
if (
spans
and spans[-1]["end_time"] == interval.start_time
and spans[-1]["streams"] == streams
):
spans[-1]["end_time"] = interval.end_time
else:
spans.append(
{
"start_time": interval.start_time,
"end_time": interval.end_time,
"streams": streams,
}
)
return spans
def stream_has_audio(intervals: list[CoverageInterval], main: bool) -> bool:
"""Whether a stream is audio-bearing over a coverage window.
A stream counts as audio-bearing unless EVERY one of its rows reports
has_audio False; NULL (legacy or undetermined) counts as audio.
"""
return any(
row is not None and row.has_audio is not False
for row in ((interval.main if main else interval.sub) for interval in intervals)
)
def null_audio_glitches(
intervals: list[CoverageInterval], main_audio: bool, sub_audio: bool
) -> list[CoverageInterval]:
"""Treat video-only glitch rows on audio-bearing streams as no recording.
nginx-vod requires every clip in a sequence to carry the same track
count, so a truncated video-only segment (a backend restart can flush
a sub-second file before any audio packet landed) poisons every
manifest that includes it. Nulling the row turns the glitch into a
hole the span builder skips like any recording gap.
"""
result: list[CoverageInterval] = []
for interval in intervals:
main = interval.main
sub = interval.sub
if main is not None and main_audio and main.has_audio is False:
main = None
if sub is not None and sub_audio and sub.has_audio is False:
sub = None
if main is None and sub is None:
continue
if main is interval.main and sub is interval.sub:
result.append(interval)
else:
result.append(
CoverageInterval(interval.start_time, interval.end_time, main, sub)
)
return result
def build_spans(
intervals: list[CoverageInterval], stream: str | None
) -> list[list[Any]]:
"""Merge coverage intervals into single-sequence spans of one row each.
Each span is [row, start, end, is_main]; is_main lets the manifest
builder detect cross-stream hand-offs. A pinned stream serves only
its own rows; otherwise main is preferred and sub fills the gaps.
Intervals served by the same row merge on row identity alone, since
splitting a row mid-file would re-snap to a keyframe and repeat
content.
"""
spans: list[list[Any]] = []
last_is_main: bool | None = None
for interval in intervals:
if stream == STREAM_TYPE_MAIN:
row, is_main = interval.main, True
elif stream == STREAM_TYPE_SUB:
row, is_main = interval.sub, False
elif interval.main is not None:
row, is_main = interval.main, True
else:
row, is_main = interval.sub, False
if row is None:
continue
if spans and spans[-1][0] == row:
spans[-1][2] = interval.end_time
else:
start = interval.start_time
# adjacent same-stream rows routinely overlap; trimming the
# previous span's end is free, where starting this span
# mid-file would cost a clipFrom keyframe snap. Inclusive on
# the span end, since sub-MIN_INTERVAL_S overlaps are dropped
# by the resolver and leave the previous span ending a hair
# past this row's start
if (
spans
and is_main == last_is_main
and spans[-1][1] < row.start_time <= spans[-1][2]
):
spans[-1][2] = row.start_time
start = row.start_time
spans.append([row, start, interval.end_time, is_main])
last_is_main = is_main
return spans
def _keyframe_before(keyframes: Any, offset_ms: int) -> int | None:
"""Last stored keyframe offset at or before offset_ms.
keyframes is the row's record-time keyframe index (ms from segment
start). Returns None when the row has no usable index, which callers
treat as "serve the whole file".
"""
if not keyframes:
return None
candidates = [k for k in keyframes if k <= offset_ms]
return max(candidates) if candidates else None
@dataclass
class ClipPlan:
"""The exact playlist realization of one span.
clip_from_ms is the keyframe-snapped clipFrom, or None when the whole
file is served. skipped means the clip is omitted from the manifest.
"""
clip_from_ms: int | None
duration_ms: int
skipped: bool
def plan_clip(row: Any, start: float, end: float) -> ClipPlan:
"""Plan one nginx-vod clip for a recording row trimmed to [start, end).
The single source of truth for clip realization: the vod mapping
builder emits exactly this plan and the coverage timelines report it
to the frontend, so the playhead model matches the playlist by
construction rather than accumulating drift at each hand-off.
"""
min_duration_ms = 100 # Minimum 100ms to ensure at least one video frame
max_duration_ms = MAX_SEGMENT_DURATION * 1000
clip_from: int | None = None
duration = int(row.duration * 1000)
# adjust start offset if start is after the recording start
inpoint = int((start - row.start_time) * 1000) if start > row.start_time else 0
if inpoint > 0:
clip_from = inpoint
duration -= inpoint
# adjust end if the recording ends after the requested end
if row.end_time > end:
duration -= int((row.end_time - end) * 1000)
# nginx-vod-module pushes clipFrom forward to the next keyframe,
# which can leave too few frames for a playable segment; snapping
# back to the preceding keyframe always starts on a decodable frame
if clip_from is not None:
keyframe_ms = _keyframe_before(row.keyframes, clip_from)
if keyframe_ms is not None:
gained = clip_from - keyframe_ms
clip_from = keyframe_ms
duration += gained
logger.debug(
"VOD: snapped clipFrom to keyframe at %sms for %s, duration now %sms",
keyframe_ms,
row.path,
duration,
)
else:
logger.debug(
"VOD: no keyframe index for %s, removing clipFrom to use full recording",
row.path,
)
clip_from = None
duration = int(row.duration * 1000)
if row.end_time > end:
duration -= int((row.end_time - end) * 1000)
if duration < min_duration_ms:
logger.debug(
"VOD: skipping recording %s - resulting duration %sms too short",
row.path,
duration,
)
return ClipPlan(None, 0, True)
if duration >= max_duration_ms:
logger.warning(f"Recording clip is missing or empty: {row.path}")
return ClipPlan(None, 0, True)
return ClipPlan(clip_from, duration, False)
def realized_timeline(
intervals: list[CoverageInterval], stream: str | None
) -> list[dict[str, Any]]:
"""The exact playlist timeline a vod route will realize for a range.
Each item pairs a span's wall-clock bounds with the duration (ms) of
the manifest clip serving it, including keyframe back-snap lead-in,
so summing durations reproduces playlist time exactly. A span whose
clip is skipped reports duration 0.
"""
return [
{
"start_time": span_start,
"end_time": span_end,
"duration": plan.duration_ms,
}
for row, span_start, span_end, _ in build_spans(intervals, stream)
for plan in (plan_clip(row, span_start, span_end),)
]
def realized_timelines(
intervals: list[CoverageInterval],
) -> dict[str, list[dict[str, Any]]]:
"""All three variant timelines for a coverage window.
Applies the same glitch-nulling as the manifest builder, then
assembles each variant's realized spans. Keyframe snapping reads the
per-row index stored at record time, so no file is touched.
"""
main_audio = stream_has_audio(intervals, main=True)
sub_audio = stream_has_audio(intervals, main=False)
nulled = null_audio_glitches(intervals, main_audio, sub_audio)
return {
"auto": realized_timeline(nulled, None),
"main": realized_timeline(nulled, STREAM_TYPE_MAIN),
"sub": realized_timeline(nulled, STREAM_TYPE_SUB),
}
frigate/util/services.py
+164 -22
View File
@@ -1,6 +1,7 @@
"""Utilities for services."""
import asyncio
import contextlib
import glob
import json
import logging
@@ -1256,8 +1257,22 @@ async def get_video_properties(
async def probe_with_ffprobe(
url: str,
rtsp_transport: str | None = None,
) -> tuple[bool, int, int, str | None, float]:
"""Fallback using ffprobe: returns (valid, width, height, codec, duration)."""
) -> tuple[
bool,
int,
int,
str | None,
str | None,
float,
bool | None,
int | None,
str | None,
]:
"""Probe using ffprobe: returns (valid, width, height, fourcc, video_codec, duration, has_audio, audio_rate, audio_codec).
ffprobe reports the codec name directly, so fourcc and
video_codec are the same value on this path.
"""
cmd = [ffmpeg.ffprobe_path]
if rtsp_transport:
cmd += ["-rtsp_transport", rtsp_transport]
@@ -1285,19 +1300,17 @@ async def get_video_properties(
clean_camera_user_pass(url),
rtsp_transport or "default",
)
proc.kill()
await proc.wait()
return False, 0, 0, None, -1
return False, 0, 0, None, None, -1, None, None, None
if proc.returncode != 0:
return False, 0, 0, None, -1
return False, 0, 0, None, None, -1, None, None, None
data = json.loads(stdout.decode())
video_streams = [
s for s in data.get("streams", []) if s.get("codec_type") == "video"
]
if not video_streams:
return False, 0, 0, None, -1
return False, 0, 0, None, None, -1, None, None, None
v = video_streams[0]
width = int(v.get("width", 0))
@@ -1307,16 +1320,70 @@ async def get_video_properties(
duration_str = data.get("format", {}).get("duration")
duration = float(duration_str) if duration_str else -1.0
return True, width, height, codec, duration
except (json.JSONDecodeError, ValueError, KeyError, sp.SubprocessError):
return False, 0, 0, None, -1
audio_streams = [
s for s in data.get("streams", []) if s.get("codec_type") == "audio"
]
has_audio = bool(audio_streams)
def probe_with_cv2(url: str) -> tuple[bool, int, int, str | None, float]:
"""Primary attempt using cv2: returns (valid, width, height, fourcc, duration)."""
# codec and sample rate distinguish audio tracks whose decoder
# configs cannot share an HLS sequence
audio_rate: int | None = None
audio_codec: str | None = None
if audio_streams:
try:
audio_rate = int(audio_streams[0]["sample_rate"])
except (KeyError, TypeError, ValueError):
audio_rate = None
audio_codec = audio_streams[0].get("codec_name")
return (
True,
width,
height,
codec,
codec,
duration,
has_audio,
audio_rate,
audio_codec,
)
except (json.JSONDecodeError, ValueError, KeyError, sp.SubprocessError):
return False, 0, 0, None, None, -1, None, None, None
finally:
# callers run in a per-cycle event loop, and an ffprobe still
# running when that loop closes is finalized against a dead loop
# ("Event loop is closed"). Draining after the kill is what
# closes the pipes and their transports
if proc is not None and proc.returncode is None:
with contextlib.suppress(ProcessLookupError):
proc.kill()
with contextlib.suppress(TimeoutError):
await asyncio.wait_for(proc.communicate(), timeout=2)
def probe_with_cv2(
url: str,
) -> tuple[
bool,
int,
int,
str | None,
str | None,
float,
bool | None,
int | None,
str | None,
]:
"""Probe using cv2: returns (valid, width, height, fourcc, video_codec, duration, has_audio, audio_rate, audio_codec).
cv2 cannot report audio streams or a normalized codec name, so
has_audio, audio_rate, audio_codec, and video_codec are always
None (unknown) on this path.
"""
cap = cv2.VideoCapture(url)
if not cap.isOpened():
cap.release()
return False, 0, 0, None, -1
return False, 0, 0, None, None, -1, None, None, None
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
@@ -1335,7 +1402,7 @@ async def get_video_properties(
duration = total_frames / fps
cap.release()
return valid, width, height, fourcc, duration
return valid, width, height, fourcc, None, duration, None, None, None
is_rtsp = url.startswith("rtsp://")
@@ -1343,24 +1410,99 @@ async def get_video_properties(
# skip cv2 for RTSP: its FFmpeg backend has a hardcoded ~30s internal
# timeout that cannot be shortened per-call, and ffprobe bounded by
# -rw_timeout handles RTSP probing reliably
has_video, width, height, fourcc, duration = await probe_with_ffprobe(url)
(
has_video,
width,
height,
fourcc,
video_codec,
duration,
has_audio,
audio_rate,
audio_codec,
) = await probe_with_ffprobe(url)
elif get_duration:
# ffprobe first: segment validation also needs audio presence,
# which cv2 cannot report
(
has_video,
width,
height,
fourcc,
video_codec,
duration,
has_audio,
audio_rate,
audio_codec,
) = await probe_with_ffprobe(url)
# fallback to cv2 if needed; audio stays unknown there
if not has_video or duration < 0:
(
has_video,
width,
height,
fourcc,
video_codec,
duration,
has_audio,
audio_rate,
audio_codec,
) = probe_with_cv2(url)
else:
# try cv2 first for local files, HTTP, RTMP
has_video, width, height, fourcc, duration = probe_with_cv2(url)
(
has_video,
width,
height,
fourcc,
video_codec,
duration,
has_audio,
audio_rate,
audio_codec,
) = probe_with_cv2(url)
# fallback to ffprobe if needed
if not has_video or (get_duration and duration < 0):
has_video, width, height, fourcc, duration = await probe_with_ffprobe(url)
if not has_video:
(
has_video,
width,
height,
fourcc,
video_codec,
duration,
has_audio,
audio_rate,
audio_codec,
) = await probe_with_ffprobe(url)
# last resort for RTSP: try TCP transport, since default UDP may be blocked
if (not has_video or (get_duration and duration < 0)) and is_rtsp:
has_video, width, height, fourcc, duration = await probe_with_ffprobe(
url, rtsp_transport="tcp"
)
(
has_video,
width,
height,
fourcc,
video_codec,
duration,
has_audio,
audio_rate,
audio_codec,
) = await probe_with_ffprobe(url, rtsp_transport="tcp")
result: dict[str, Any] = {"has_valid_video": has_video}
if has_video:
result.update({"width": width, "height": height})
result.update(
{
"width": width,
"height": height,
"has_audio": has_audio,
"audio_rate": audio_rate,
"audio_codec": audio_codec,
"video_codec": video_codec,
}
)
if fourcc:
result["fourcc"] = fourcc
if get_duration:
frigate/video/ffmpeg.py
+19
View File
@@ -147,6 +147,7 @@ class CameraWatchdog(threading.Thread):
self.requestor = InterProcessRequestor()
self.was_enabled = self.config.enabled
self.was_record_enabled_in_config = self.config.record.enabled_in_config
self.was_record_sub_enabled = self.config.record.sub.enabled
self.segment_subscriber = RecordingsDataSubscriber(RecordingsDataTypeEnum.all)
self.latest_valid_segment_time: float = 0
@@ -312,6 +313,24 @@ class CameraWatchdog(threading.Thread):
self.was_record_enabled_in_config = record_enabled_in_config
continue
record_sub_enabled = self.config.record.sub.enabled
if record_sub_enabled != self.was_record_sub_enabled:
# adding and removing the record_sub output both require a
# restart, unlike the main record toggle
if record_enabled_in_config and enabled:
self.logger.debug(
f"Sub stream recording toggled in config for {self.config.name}, restarting ffmpeg"
)
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.record_enable_time = datetime.now().astimezone(UTC)
last_restart_time = datetime.now().timestamp()
self.was_record_sub_enabled = record_sub_enabled
continue
if not enabled:
continue