Support main+sub stream exports (#24193)

* Support multi resolution exports

* Fix decoder text

* Add dropdown and ability to select export stream selection

* Fix for review comments

* Fix mypy

* Cleanup wording
This commit is contained in:
Nicolas Mowen
2026-09-12 07:30:04 -06:00
parent 70ce193e09
commit 82be9fff5e
15 changed files with 1164 additions and 49 deletions
@@ -1,5 +1,7 @@
from pydantic import BaseModel, Field, model_validator
from frigate.record.export import ExportStreamEnum
MAX_BATCH_EXPORT_ITEMS = 50
@@ -53,6 +55,16 @@ class BatchExportBody(BaseModel):
title="New case description",
description="Optional description for a newly created export case",
)
stream: ExportStreamEnum = Field(
default=ExportStreamEnum.auto,
title="Recorded stream to export",
description=(
"Which recorded stream every item in the batch is exported "
"from. 'auto' uses the merged timeline, preferring the main "
"stream and falling back to the sub stream where main has "
"aged out. 'main' or 'sub' pins the exports to that stream."
),
)
@model_validator(mode="after")
def validate_case_target(self) -> "BatchExportBody":
@@ -3,6 +3,7 @@ from pydantic.json_schema import SkipJsonSchema
from frigate.record.export import (
ChaptersEnum,
ExportStreamEnum,
PlaybackSourceEnum,
)
@@ -27,6 +28,16 @@ class ExportRecordingsBody(BaseModel):
"the camera's configured export chapter mode is used."
),
)
stream: ExportStreamEnum = Field(
default=ExportStreamEnum.auto,
title="Recorded stream to export",
description=(
"Which recorded stream to export. 'auto' uses the merged "
"timeline, preferring the main stream and falling back to the "
"sub stream where main has aged out. 'main' or 'sub' pins the "
"export to that stream alone."
),
)
class ExportRecordingsCustomBody(BaseModel):
+39 -21
View File
@@ -73,6 +73,7 @@ from frigate.record.export import (
DEFAULT_TIME_LAPSE_FFMPEG_ARGS,
DEFAULT_TIME_LAPSE_FFMPEG_INPUT_ARGS,
ChaptersEnum,
ExportStreamEnum,
PlaybackSourceEnum,
export_video_path,
validate_ffmpeg_args,
@@ -149,16 +150,23 @@ def _sanitize_existing_image(
return existing_image, None
def _no_recordings_message(stream: ExportStreamEnum) -> str:
if stream == ExportStreamEnum.auto:
return "No recordings found for time range"
return f"No {stream.value} stream recordings found for time range"
def _validate_export_source(
camera_name: str,
start_time: float,
end_time: float,
playback_source: PlaybackSourceEnum,
stream: ExportStreamEnum = ExportStreamEnum.auto,
) -> str | None:
if playback_source == PlaybackSourceEnum.recordings:
recordings_count = (
Recordings.select()
.where(
query = Recordings.select().where(
(
Recordings.start_time.between(start_time, end_time)
| Recordings.end_time.between(start_time, end_time)
| (
@@ -166,12 +174,16 @@ def _validate_export_source(
& (end_time < Recordings.end_time)
)
)
.where(Recordings.camera == camera_name)
.count()
& (Recordings.camera == camera_name)
)
if recordings_count <= 0:
return "No recordings found for time range"
# a pinned export reads only that stream, so the other stream's
# coverage must not make the range look exportable
if stream != ExportStreamEnum.auto:
query = query.where(Recordings.stream_type == stream.value)
if query.count() <= 0:
return _no_recordings_message(stream)
return None
@@ -195,6 +207,7 @@ def _validate_export_source(
def _get_item_recording_export_errors(
request: Request,
items: list[BatchExportItem],
stream: ExportStreamEnum = ExportStreamEnum.auto,
) -> dict[int, str]:
"""Return {item_index: error message} for items with invalid state.
@@ -224,20 +237,20 @@ def _get_item_recording_export_errors(
min_start = min(r[1] for r in indexed_ranges)
max_end = max(r[2] for r in indexed_ranges)
recording_ranges = list(
Recordings.select(Recordings.start_time, Recordings.end_time)
.where(
Recordings.camera == camera_name,
Recordings.start_time.between(min_start, max_end)
| Recordings.end_time.between(min_start, max_end)
| (
(min_start > Recordings.start_time)
& (max_end < Recordings.end_time)
),
)
.iterator()
query = Recordings.select(Recordings.start_time, Recordings.end_time).where(
Recordings.camera == camera_name,
Recordings.start_time.between(min_start, max_end)
| Recordings.end_time.between(min_start, max_end)
| ((min_start > Recordings.start_time) & (max_end < Recordings.end_time)),
)
# a pinned batch reads only that stream, so the other stream's
# coverage must not make an item look exportable
if stream != ExportStreamEnum.auto:
query = query.where(Recordings.stream_type == stream.value)
recording_ranges = list(query.iterator())
for index, start_time, end_time in indexed_ranges:
has_recording = any(
(
@@ -248,7 +261,7 @@ def _get_item_recording_export_errors(
for rec in recording_ranges
)
if not has_recording:
errors[index] = "No recordings found for time range"
errors[index] = _no_recordings_message(stream)
return errors
@@ -265,6 +278,7 @@ def _build_export_job(
ffmpeg_output_args: str | None = None,
cpu_fallback: bool = False,
chapters: ChaptersEnum | None = None,
stream: ExportStreamEnum = ExportStreamEnum.auto,
) -> ExportJob:
return ExportJob(
id=_generate_export_id(camera_name),
@@ -279,6 +293,7 @@ def _build_export_job(
ffmpeg_output_args=ffmpeg_output_args,
cpu_fallback=cpu_fallback,
chapters=chapters,
stream=stream,
)
@@ -692,7 +707,7 @@ def export_recordings_batch(
return image_validation_error
sanitized_images.append(existing_image)
item_errors = _get_item_recording_export_errors(request, body.items)
item_errors = _get_item_recording_export_errors(request, body.items, body.stream)
queueable_indexes = [
index for index in range(len(body.items)) if index not in item_errors
@@ -761,6 +776,7 @@ def export_recordings_batch(
chapters=request.app.frigate_config.cameras[
item.camera
].record.export.chapters,
stream=body.stream,
)
try:
start_export_job(request.app.frigate_config, export_job)
@@ -868,6 +884,7 @@ def export_recording(
start_time,
end_time,
playback_source,
body.stream,
)
if source_error is not None:
return JSONResponse(
@@ -884,6 +901,7 @@ def export_recording(
playback_source,
export_case_id,
chapters=chapters,
stream=body.stream,
)
try:
start_export_job(request.app.frigate_config, export_job)
+7 -1
View File
@@ -18,7 +18,11 @@ from frigate.config.camera.record import ChaptersEnum
from frigate.const import UPDATE_JOB_STATE
from frigate.jobs.job import Job
from frigate.models import Export
from frigate.record.export import PlaybackSourceEnum, RecordingExporter
from frigate.record.export import (
ExportStreamEnum,
PlaybackSourceEnum,
RecordingExporter,
)
from frigate.types import JobStatusTypesEnum
logger = logging.getLogger(__name__)
@@ -58,6 +62,7 @@ class ExportJob(Job):
ffmpeg_output_args: str | None = None
cpu_fallback: bool = False
chapters: ChaptersEnum | None = None
stream: ExportStreamEnum = ExportStreamEnum.auto
current_step: str = "queued"
progress_percent: float = 0.0
@@ -347,6 +352,7 @@ class ExportJobManager:
job.ffmpeg_output_args,
job.cpu_fallback,
job.chapters,
job.stream,
on_progress=self._make_progress_callback(job),
)
+481 -22
View File
@@ -1,5 +1,6 @@
"""Export recordings to storage."""
import asyncio
import datetime
import logging
import os
@@ -10,6 +11,7 @@ import string
import subprocess as sp
import threading
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Any
@@ -36,6 +38,13 @@ from frigate.ffmpeg_presets import (
from frigate.models import Export, Previews, Recordings, ReviewSegment
from frigate.util.ffmpeg import run_ffmpeg_with_progress
from frigate.util.ownership import chown_to_runtime
from frigate.util.recording_coverage import (
build_spans,
known_video_codecs,
resolve_coverage,
stream_media_summary,
)
from frigate.util.services import get_video_properties
from frigate.util.time import is_current_hour
logger = logging.getLogger(__name__)
@@ -45,6 +54,41 @@ DEFAULT_TIME_LAPSE_FFMPEG_INPUT_ARGS = "-an"
DEFAULT_TIME_LAPSE_FFMPEG_ARGS = "-vf setpts=0.04*PTS -r 30"
TIMELAPSE_DATA_INPUT_ARGS = "-skip_frame nokey"
# nginx-vod repackages each stream into fMP4 with a timescale derived from
# that stream's frame rate, and the concat demuxer rescales every input to
# whatever timebase the first one happens to use. Staging each run with one
# explicit timescale is what keeps a 5fps sub run from being replayed at the
# main stream's rate. 90000 is the RTSP clock rate and divides evenly by
# every common camera frame rate.
EXPORT_TRACK_TIMESCALE = 90000
@dataclass
class StreamRun:
"""A contiguous slice of an export served by a single stream type.
sample_path is one recording from the run, used to probe the stream's
resolution when the runs have to be scaled to a common size.
"""
stream_type: str
start_time: float
end_time: float
sample_path: str
@property
def duration(self) -> float:
return max(0.0, self.end_time - self.start_time)
@dataclass
class _ChapterWindow:
"""A merged-timeline slice, shaped like the recording rows chapters read."""
start_time: float
end_time: float
# Matches the setpts factor used in timelapse exports (e.g. setpts=0.04*PTS).
# Captures the floating-point factor so we can scale expected duration.
SETPTS_FACTOR_RE = re.compile(r"setpts=([0-9]*\.?[0-9]+)\*PTS")
@@ -207,6 +251,20 @@ class PlaybackSourceEnum(str, Enum):
preview = "preview"
class ExportStreamEnum(str, Enum):
"""Which recorded stream an export should be built from.
``auto`` keeps the merged timeline: main where it exists, sub filling
the gaps main has already aged out of. Pinning to one stream trades
that coverage for a uniform source, which is always a plain stream
copy since nothing hands off mid-export.
"""
auto = "auto"
main = STREAM_TYPE_MAIN
sub = STREAM_TYPE_SUB
EXPORT_FILE_NAME_MAX_BYTES = 255
@@ -241,6 +299,7 @@ class RecordingExporter(threading.Thread):
ffmpeg_output_args: str | None = None,
cpu_fallback: bool = False,
chapters: ChaptersEnum | None = None,
stream: ExportStreamEnum = ExportStreamEnum.auto,
on_progress: Callable[[str, float], None] | None = None,
) -> None:
super().__init__()
@@ -257,7 +316,10 @@ class RecordingExporter(threading.Thread):
self.ffmpeg_output_args = ffmpeg_output_args
self.cpu_fallback = cpu_fallback
self.chapters = chapters
self.stream = stream
self.on_progress = on_progress
self.staged_runs: list[str] = []
self._coverage: tuple[list[list[Any]], set[str], bool] | None = None
# ensure export thumb dir
Path(os.path.join(CLIPS_DIR, "export")).mkdir(exist_ok=True)
@@ -305,6 +367,337 @@ class RecordingExporter(threading.Thread):
return input_duration * factor
@property
def pinned_stream(self) -> str | None:
"""The stream type this export is pinned to, or None for auto."""
return None if self.stream == ExportStreamEnum.auto else self.stream.value
def _resolve_coverage(self) -> tuple[list[list[Any]], set[str], bool]:
"""Resolve the export range into the spans the VOD manifest will serve.
Delegates to the same coverage resolution the manifest builder
uses, so what we plan around and what nginx-vod emits agree by
construction. Returns the spans (each [row, start, end, is_main]),
the known video codecs, and whether audio survives the range.
Memoized: several stages of the export ask the same question, and
the recordings backing a finished range do not change under us.
"""
if self._coverage is None:
intervals = resolve_coverage(self.camera, self.start_time, self.end_time)
self._coverage = (
build_spans(intervals, self.pinned_stream),
known_video_codecs(intervals),
self._audio_is_uniform(stream_media_summary(intervals)),
)
return self._coverage
def _audio_is_uniform(self, summary: dict[str, dict[str, Any]]) -> bool:
"""Whether every stream in range carries audio with the same signature.
Stream-copying audio across a hand-off only works when both
streams agree, the same rule the merged manifest applies when it
decides to serve a mixed range without audio.
"""
# legacy rows report None rather than False, and an unknown
# signature is not one we can promise lines up
if not summary or any(
stream["has_audio"] is not True for stream in summary.values()
):
return False
return (
len(
{
(stream["audio_codec"], stream["audio_rate"])
for stream in summary.values()
}
)
== 1
)
def _merged_spans(self) -> list[list[Any]]:
return self._resolve_coverage()[0]
def _prepare_stream_runs(self) -> bool:
"""Stage each stream run to a temp file when the range spans more than one.
Leaves ``staged_runs`` empty for a single-stream range, which is
the overwhelmingly common case and keeps the existing
single-playlist path byte for byte what it was. Returns False only
when staging was needed and failed.
"""
if self.pinned_stream is not None:
# a pinned range is uniform by construction, so there is no
# hand-off to stage around
return True
spans, codecs, keep_audio = self._resolve_coverage()
runs = self._stream_runs(spans)
# a range one stream covers end to end has nothing to hand off,
# so it stays on the existing path however long it is
if len(runs) < 2:
return True
runs = [piece for run in runs for piece in self._split_long_run(run)]
return self._stage_stream_runs(runs, codecs, keep_audio)
def _stream_runs(self, spans: list[list[Any]]) -> list[StreamRun]:
"""Collapse the merged spans into contiguous runs of one stream type.
A run is the largest slice of the export that a single stream
covers end to end, and is therefore the largest chunk we can hand
to ffmpeg without the parameter sets changing underneath it.
"""
runs: list[StreamRun] = []
for row, span_start, span_end, is_main in spans:
stream_type = STREAM_TYPE_MAIN if is_main else STREAM_TYPE_SUB
if runs and runs[-1].stream_type == stream_type:
runs[-1].end_time = span_end
else:
runs.append(StreamRun(stream_type, span_start, span_end, row.path))
return runs
def _split_long_run(self, run: StreamRun) -> list[StreamRun]:
"""Break a run into playlist-sized pieces.
Each run is fetched as one pinned VOD playlist, and nginx-vod caps
how many clips a single mapping may hold. This is the same bound
the unstaged path respects by paging its playlist lines. Splitting
is free here: both halves are the same stream, so they share the
parameter sets and the timebase.
"""
if run.duration <= MAX_PLAYLIST_SECONDS:
return [run]
pieces: list[StreamRun] = []
start = run.start_time
while start < run.end_time:
end = min(start + MAX_PLAYLIST_SECONDS, run.end_time)
pieces.append(StreamRun(run.stream_type, start, end, run.sample_path))
start = end
return pieces
def _internal_port(self) -> int:
"""The API port to fetch VOD playlists from."""
internal_port = self.config.networking.listen.internal
# handle case where internal port is a string with ip:port
if isinstance(internal_port, str):
return int(internal_port.split(":")[-1])
return internal_port
def _vod_url(self, stream_type: str | None, start: float, end: float) -> str:
"""A VOD playlist URL, pinned to one stream type when given."""
pin = f"/{stream_type}" if stream_type else ""
return (
f"http://127.0.0.1:{self._internal_port()}/vod/{self.camera}{pin}"
f"/start/{start}/end/{end}/index.m3u8"
)
def _staged_run_path(self, index: int) -> str:
return os.path.join(CACHE_DIR, f"export_stage_{self.export_id}_{index}.mp4")
def _probe_stream_resolution(self, run: StreamRun) -> tuple[int, int] | None:
"""Probe one recording from a run for its resolution.
Only the scaling path needs this, and one segment per stream is
enough: a stream's resolution is fixed for as long as the camera
keeps its configuration.
"""
try:
properties = asyncio.run(
get_video_properties(self.config.ffmpeg, run.sample_path)
)
except OSError:
logger.exception("Failed to probe %s for export sizing", run.sample_path)
return None
width = properties.get("width")
height = properties.get("height")
if not width or not height:
return None
return int(width), int(height)
def _staged_progress(
self, step: str, base: float, weight: float
) -> Callable[[float], None]:
"""Map one run's 0-100 progress onto its slice of the whole pass."""
def report(percent: float) -> None:
self._emit_progress(step, base + (percent / 100.0) * weight)
return report
def _stage_run_command(
self,
run: StreamRun,
dest: str,
target: tuple[int, int] | None,
keep_audio: bool,
) -> list[str]:
"""Build the ffmpeg command that renders one run to a temp file.
Without a target the run is stream-copied, which is all a mixed
*resolution* export needs. A target is only set when the runs also
disagree on codec, where one mp4 track genuinely cannot hold both
and every run has to be re-encoded to match.
"""
ffmpeg_input = (
"-y -protocol_whitelist pipe,file,http,tcp "
f"-i {self._vod_url(run.stream_type, run.start_time, run.end_time)}"
)
# audio only survives when both streams agree on it, otherwise the
# copied track breaks at the same hand-off the video used to. These
# are output options: "-c:a copy" ahead of -i selects a *decoder*
# named copy, which does not exist
audio_args = "-c:a copy" if keep_audio else "-an"
if target is None:
return (
f"{self.config.ffmpeg.ffmpeg_path} -hide_banner {ffmpeg_input} "
f"{audio_args} -c:v copy "
f"-video_track_timescale {EXPORT_TRACK_TIMESCALE} {dest}"
).split(" ")
width, height = target
# pad rather than stretch: the sub stream is often a different
# aspect ratio than the main one, and letterboxing it is honest
# where distorting the footage is not
scale = (
f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2,setsar=1"
)
# deliberately software-encoded. The vaapi and nvidia encode presets
# set -hwaccel_output_format, which leaves decoded frames in GPU
# memory where scale/pad cannot reach them; making this pass use the
# GPU means per-preset hwdownload/hwupload chains (see the birdseye
# vaapi preset). Codec-mixed ranges are rare enough to eat the CPU
# cost rather than ship an untested filter graph
return parse_preset_hardware_acceleration_encode(
self.config.ffmpeg.ffmpeg_path,
None,
ffmpeg_input,
f"{audio_args} -vf {scale} "
f"-video_track_timescale {EXPORT_TRACK_TIMESCALE} {dest}",
EncodeTypeEnum.timelapse,
).split(" ")
def _stage_stream_runs(
self,
runs: list[StreamRun],
codecs: set[str],
keep_audio: bool,
) -> bool:
"""Render each run to its own temp file, one stream type at a time.
This is what makes a mixed-resolution export work. Handing the
merged playlist straight to ffmpeg looks like it should work,
since nginx-vod marks the stream change with a discontinuity and a
fresh EXT-X-MAP, but ffmpeg's HLS demuxer binds the track's
parameter sets from the *first* init segment only. Every sample
after the hand-off is then decoded against the wrong SPS, which is
what freezes the sub-resolution stretches of the output.
Demuxing each run on its own gives each one its correct parameter
sets, and the mp4 muxer writes them in-band at the hand-off, so
the concatenated result decodes cleanly without re-encoding.
Returns False when staging failed. The caller must abort rather
than fall back to the merged playlist, which is the very thing
that produces the broken file.
"""
target: tuple[int, int] | None = None
if len(codecs) > 1:
# one mp4 track carries one codec, so a range that mixes them
# has to be re-encoded to a common one. Scale up to the
# largest stream so the main footage keeps its detail
sizes = [
size
for size in (self._probe_stream_resolution(run) for run in runs)
if size is not None
]
if not sizes:
logger.error(
"Export %s spans video codecs %s but no run could be probed "
"for its resolution",
self.export_id,
sorted(codecs),
)
return False
target = (max(s[0] for s in sizes), max(s[1] for s in sizes))
logger.debug(
"Export %s spans video codecs %s; re-encoding every run to %dx%d",
self.export_id,
sorted(codecs),
target[0],
target[1],
)
else:
logger.debug(
"Export %s spans %d stream runs; staging each one separately",
self.export_id,
len(runs),
)
total_duration = sum(run.duration for run in runs) or 1.0
step = "encoding" if target is not None else "copying"
completed = 0.0
for index, run in enumerate(runs):
dest = self._staged_run_path(index)
weight = 100.0 * run.duration / total_duration
base = 100.0 * completed / total_duration
# claim the path before ffmpeg can write to it. ffmpeg removes
# its own output on a clean error exit, but a killed one (OOM,
# container stop) leaves whatever it had already muxed, and a
# path that was never recorded is a partial file nothing
# deletes
self.staged_runs.append(dest)
returncode, stderr = run_ffmpeg_with_progress(
self._stage_run_command(run, dest, target, keep_audio),
expected_duration_seconds=run.duration,
on_progress=self._staged_progress(step, base, weight),
use_low_priority=True,
)
if returncode != 0:
logger.error(
"Failed to stage %s stream for export %s between %s and %s",
run.stream_type,
self.export_id,
run.start_time,
run.end_time,
)
logger.error(stderr)
self._cleanup_staged_runs()
return False
completed += run.duration
return True
def _cleanup_staged_runs(self) -> None:
"""Remove any temp files left behind by staging."""
for path in self.staged_runs:
Path(path).unlink(missing_ok=True)
self.staged_runs = []
def _get_recordings_for_range(self, stream_type: str) -> list[Any]:
"""Fetch one stream type's recording rows overlapping the export range."""
return list(
@@ -338,12 +731,16 @@ class RecordingExporter(threading.Thread):
"""
try:
if self.playback_source == PlaybackSourceEnum.recordings:
# 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)
# the merged timeline is what actually gets exported: main
# where it exists, sub filling the gaps it leaves behind.
# Summing one stream alone under-reports a mixed range and
# pins progress at 100% for the rest of the export
return float(
sum(
max(0.0, span_end - span_start)
for _row, span_start, span_end, _is_main in self._merged_spans()
)
)
else:
rows = (
Previews.select(Previews.start_time, Previews.end_time)
@@ -724,21 +1121,40 @@ class RecordingExporter(threading.Thread):
def get_record_export_command(
self, video_path: str, use_hwaccel: bool = True
) -> tuple[list[str], str | list[str]]:
# handle case where internal port is a string with ip:port
internal_port = self.config.networking.listen.internal
if type(internal_port) is str:
internal_port = int(internal_port.split(":")[-1])
if self.staged_runs:
# each run was already rendered to a temp file with a common
# track timescale, so the concat demuxer has nothing left to
# reconcile and every chapter offset lines up with the merged
# timeline the staged files reproduce
recordings = [
_ChapterWindow(span_start, span_end)
for _row, span_start, span_end, _is_main in self._merged_spans()
]
playlist_lines: list[str] = [f"file '{path}'" for path in self.staged_runs]
ffmpeg_input = (
"-y -protocol_whitelist pipe,file -f concat -safe 0 -i /dev/stdin"
)
return self._finish_record_export_command(
video_path, ffmpeg_input, playlist_lines, recordings, use_hwaccel
)
# 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)
pin = self.pinned_stream
if not recordings:
recordings = self._get_recordings_for_range(STREAM_TYPE_SUB)
if pin is not None:
# a pinned export reads that stream and only that stream, so
# its own rows are the ones the chapters describe
recordings = self._get_recordings_for_range(pin)
else:
# 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)
playlist_lines: list[str] = []
if not recordings:
recordings = self._get_recordings_for_range(STREAM_TYPE_SUB)
playlist_lines = []
if (self.end_time - self.start_time) <= MAX_PLAYLIST_SECONDS:
playlist_url = f"http://127.0.0.1:{internal_port}/vod/{self.camera}/start/{self.start_time}/end/{self.end_time}/index.m3u8"
playlist_url = self._vod_url(pin, self.start_time, self.end_time)
ffmpeg_input = (
f"-y -protocol_whitelist pipe,file,http,tcp -i {playlist_url}"
)
@@ -748,12 +1164,26 @@ class RecordingExporter(threading.Thread):
page_size = 1000
for i in range(0, len(recordings), page_size):
chunk = recordings[i : i + page_size]
playlist_lines.append(
f"file 'http://127.0.0.1:{internal_port}/vod/{self.camera}/start/{float(chunk[0].start_time)}/end/{float(chunk[-1].end_time)}/index.m3u8'"
chunk_url = self._vod_url(
pin, float(chunk[0].start_time), float(chunk[-1].end_time)
)
playlist_lines.append(f"file '{chunk_url}'")
ffmpeg_input = "-y -protocol_whitelist pipe,file,http,tcp -f concat -safe 0 -i /dev/stdin"
return self._finish_record_export_command(
video_path, ffmpeg_input, playlist_lines, recordings, use_hwaccel
)
def _finish_record_export_command(
self,
video_path: str,
ffmpeg_input: str,
playlist_lines: list[str],
recordings: list[Any],
use_hwaccel: bool,
) -> tuple[list[str], str | list[str]]:
"""Apply encoding, chapters, and metadata to a prepared input."""
if self.ffmpeg_input_args is not None and self.ffmpeg_output_args is not None:
hwaccel_args = (
self.config.cameras[self.camera].record.export.hwaccel_args
@@ -960,8 +1390,32 @@ class RecordingExporter(threading.Thread):
Export.insert(export_values).execute()
try:
self._run_export(video_path, thumb_path)
finally:
# staged runs hold a full copy of the export, so they must not
# survive a failure the way the small chapter file could
self._cleanup_staged_runs()
def _discard_failed_export(self, video_path: str, thumb_path: str) -> None:
"""Drop the partial output and the row that promised it."""
Path(video_path).unlink(missing_ok=True)
Export.delete().where(Export.id == self.export_id).execute()
Path(thumb_path).unlink(missing_ok=True)
def _run_export(self, video_path: str, thumb_path: str) -> None:
try:
if self.playback_source == PlaybackSourceEnum.recordings:
if not self._prepare_stream_runs():
# the merged playlist is what staging exists to avoid;
# falling back to it would hand the user a file whose
# sub-resolution stretches are frozen
logger.error(
"Failed to stage stream runs for export %s", self.export_id
)
self._discard_failed_export(video_path, thumb_path)
return
ffmpeg_cmd, playlist_lines = self.get_record_export_command(video_path)
else:
ffmpeg_cmd, playlist_lines = self.get_preview_export_command(video_path)
@@ -978,6 +1432,10 @@ class RecordingExporter(threading.Thread):
)
initial_step = "copying" if is_stream_copy else "encoding"
if self.staged_runs:
# staging already reported a full pass under its own step
initial_step = "merging"
returncode, stderr = self._run_ffmpeg_with_progress(
ffmpeg_cmd, playlist_lines, step=initial_step
)
@@ -994,6 +1452,9 @@ class RecordingExporter(threading.Thread):
)
if self.playback_source == PlaybackSourceEnum.recordings:
# staged runs are always software-encoded, so there is no
# hwaccel in them to fall back from; only the merge pass
# is rebuilt here
ffmpeg_cmd, playlist_lines = self.get_record_export_command(
video_path, use_hwaccel=False
)
@@ -1013,9 +1474,7 @@ class RecordingExporter(threading.Thread):
f"Failed to export {self.playback_source.value} for command {' '.join(ffmpeg_cmd)}"
)
logger.error(stderr)
Path(video_path).unlink(missing_ok=True)
Export.delete().where(Export.id == self.export_id).execute()
Path(thumb_path).unlink(missing_ok=True)
self._discard_failed_export(video_path, thumb_path)
return
else:
chown_to_runtime(video_path)
+464 -1
View File
@@ -1,9 +1,21 @@
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from frigate.api.export import _unique_archive_name
from frigate.const import MAX_PLAYLIST_SECONDS
from frigate.models import Export
from frigate.record.export import export_video_path, validate_ffmpeg_args
from frigate.record.export import (
EXPORT_TRACK_TIMESCALE,
ExportStreamEnum,
PlaybackSourceEnum,
RecordingExporter,
StreamRun,
export_video_path,
validate_ffmpeg_args,
)
class TestValidateFfmpegArgs(unittest.TestCase):
@@ -210,3 +222,454 @@ class TestUniqueArchiveName(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class _FakeRow:
def __init__(self, path: str) -> None:
self.path = path
def _span(path: str, start: float, end: float, is_main: bool) -> list:
return [_FakeRow(path), start, end, is_main]
def _make_exporter(spans: list, codecs: set) -> RecordingExporter:
"""Build an exporter with coverage resolution stubbed out.
Bypasses __init__ so no directories are created and no FrigateConfig
is required, then pre-seeds the memoized coverage the same shape
_resolve_coverage would produce.
"""
exporter = RecordingExporter.__new__(RecordingExporter)
exporter.config = MagicMock()
exporter.config.ffmpeg.ffmpeg_path = "ffmpeg"
exporter.config.networking.listen.internal = 5000
exporter.config.cameras = {"front": MagicMock()}
exporter.export_id = "front_abc123"
exporter.camera = "front"
exporter.start_time = 1_000
exporter.end_time = 2_000
exporter.playback_source = PlaybackSourceEnum.recordings
exporter.ffmpeg_input_args = None
exporter.ffmpeg_output_args = None
exporter.chapters = None
exporter.stream = ExportStreamEnum.auto
exporter.staged_runs = []
exporter.staged_transcode = False
exporter._coverage = (spans, codecs, False)
return exporter
class TestStreamRuns(unittest.TestCase):
"""Runs are the largest chunk of an export whose parameter sets hold still."""
def test_consecutive_spans_of_one_stream_collapse(self) -> None:
exporter = _make_exporter([], {"h264"})
runs = exporter._stream_runs(
[
_span("/m1.mp4", 1_000, 1_010, True),
_span("/m2.mp4", 1_010, 1_020, True),
_span("/s1.mp4", 1_020, 1_030, False),
_span("/s2.mp4", 1_030, 1_040, False),
_span("/m3.mp4", 1_040, 1_050, True),
]
)
self.assertEqual(
[(r.stream_type, r.start_time, r.end_time) for r in runs],
[("main", 1_000, 1_020), ("sub", 1_020, 1_040), ("main", 1_040, 1_050)],
)
def test_run_keeps_a_sample_path_to_probe(self) -> None:
exporter = _make_exporter([], {"h264"})
runs = exporter._stream_runs(
[
_span("/m1.mp4", 1_000, 1_010, True),
_span("/m2.mp4", 1_010, 1_020, True),
]
)
self.assertEqual(len(runs), 1)
self.assertEqual(runs[0].sample_path, "/m1.mp4")
def test_long_runs_are_split_to_playlist_size(self) -> None:
"""One pinned playlist per run still has to fit nginx-vod's clip cap."""
exporter = _make_exporter([], {"h264"})
runs = exporter._split_long_run(
StreamRun("main", 0, MAX_PLAYLIST_SECONDS * 2.5, "/m1.mp4")
)
self.assertEqual(len(runs), 3)
self.assertEqual(runs[0].start_time, 0)
self.assertEqual(runs[-1].end_time, MAX_PLAYLIST_SECONDS * 2.5)
# contiguous, no gaps or overlap between the pieces
for earlier, later in zip(runs, runs[1:]):
self.assertEqual(earlier.end_time, later.start_time)
self.assertTrue(all(r.stream_type == "main" for r in runs))
def test_a_long_single_stream_range_is_not_staged(self) -> None:
"""Length alone is not a hand-off; only a stream change is."""
exporter = _make_exporter(
[_span("/m1.mp4", 0, MAX_PLAYLIST_SECONDS * 3, True)], {"h264"}
)
with patch.object(RecordingExporter, "_stage_stream_runs") as stage:
exporter._prepare_stream_runs()
stage.assert_not_called()
def test_mixed_range_is_staged(self) -> None:
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_020, True),
_span("/s1.mp4", 1_020, 1_040, False),
],
{"h264"},
)
with patch.object(RecordingExporter, "_stage_stream_runs") as stage:
exporter._prepare_stream_runs()
stage.assert_called_once()
staged_runs = stage.call_args.args[0]
self.assertEqual([r.stream_type for r in staged_runs], ["main", "sub"])
def test_single_stream_range_is_not_staged(self) -> None:
"""The common case must stay on the untouched single-playlist path."""
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_010, True),
_span("/m2.mp4", 1_010, 1_020, True),
],
{"h264"},
)
with patch.object(RecordingExporter, "_stage_stream_runs") as stage:
exporter._prepare_stream_runs()
stage.assert_not_called()
self.assertEqual(exporter.staged_runs, [])
class TestStageRunCommand(unittest.TestCase):
"""Each run is rendered on its own so its parameter sets travel with it."""
def _run(self, stream_type: str = "sub") -> StreamRun:
return StreamRun(stream_type, 1_020.0, 1_040.0, "/media/s1.mp4")
def test_copy_pins_the_playlist_to_one_stream(self) -> None:
# the merged /vod route is exactly what breaks: ffmpeg binds the
# track's parameter sets from the first init segment only
exporter = _make_exporter([], {"h264"})
cmd = exporter._stage_run_command(self._run(), "/tmp/s.mp4", None, False)
self.assertIn(
"http://127.0.0.1:5000/vod/front/sub/start/1020.0/end/1040.0/index.m3u8",
cmd,
)
def test_copy_forces_a_common_track_timescale(self) -> None:
# without this a 5fps sub run is replayed at the main stream's rate
exporter = _make_exporter([], {"h264"})
cmd = exporter._stage_run_command(self._run(), "/tmp/s.mp4", None, False)
self.assertEqual(
cmd[cmd.index("-video_track_timescale") + 1], str(EXPORT_TRACK_TIMESCALE)
)
self.assertIn("copy", cmd)
self.assertEqual(cmd[-1], "/tmp/s.mp4")
def test_audio_is_dropped_unless_both_streams_agree(self) -> None:
exporter = _make_exporter([], {"h264"})
dropped = exporter._stage_run_command(self._run(), "/tmp/s.mp4", None, False)
kept = exporter._stage_run_command(self._run(), "/tmp/s.mp4", None, True)
self.assertIn("-an", dropped)
self.assertNotIn("-an", kept)
self.assertIn("-c:a", kept)
def test_audio_codec_is_an_output_option(self) -> None:
""" "-c:a copy" ahead of -i selects a decoder named copy, which errors."""
exporter = _make_exporter([], {"h264", "h265"})
for target in (None, (1920, 1080)):
cmd = exporter._stage_run_command(self._run(), "/tmp/s.mp4", target, True)
self.assertGreater(
cmd.index("-c:a"),
cmd.index("-i"),
f"audio codec placed ahead of -i for target={target}",
)
def test_scaling_pass_does_not_use_hwaccel(self) -> None:
# the vaapi/nvidia presets keep frames in GPU memory, out of reach
# of the software scale/pad filters this pass relies on
exporter = _make_exporter([], {"h264", "h265"})
exporter.config.cameras["front"].record.export.hwaccel_args = "preset-vaapi"
cmd = exporter._stage_run_command(
self._run(), "/tmp/s.mp4", (1920, 1080), False
)
self.assertNotIn("-hwaccel", cmd)
self.assertIn("libx264", cmd)
def test_target_scales_and_pads_rather_than_stretching(self) -> None:
exporter = _make_exporter([], {"h264", "h265"})
cmd = exporter._stage_run_command(
self._run(), "/tmp/s.mp4", (1920, 1080), False
)
filtergraph = cmd[cmd.index("-vf") + 1]
self.assertIn(
"scale=1920:1080:force_original_aspect_ratio=decrease", filtergraph
)
self.assertIn("pad=1920:1080", filtergraph)
self.assertIn("setsar=1", filtergraph)
self.assertNotIn("copy", cmd)
class TestAudioUniformity(unittest.TestCase):
"""Audio only survives a hand-off when both streams agree on it."""
def _check(self, summary: dict) -> bool:
return _make_exporter([], {"h264"})._audio_is_uniform(summary)
def test_matching_audio_is_kept(self) -> None:
stream = {"has_audio": True, "audio_codec": "aac", "audio_rate": 48000}
self.assertTrue(self._check({"main": stream, "sub": dict(stream)}))
def test_differing_rate_is_dropped(self) -> None:
self.assertFalse(
self._check(
{
"main": {
"has_audio": True,
"audio_codec": "aac",
"audio_rate": 48000,
},
"sub": {
"has_audio": True,
"audio_codec": "aac",
"audio_rate": 16000,
},
}
)
)
def test_audio_on_only_one_stream_is_dropped(self) -> None:
self.assertFalse(
self._check(
{
"main": {
"has_audio": True,
"audio_codec": "aac",
"audio_rate": 48000,
},
"sub": {
"has_audio": False,
"audio_codec": None,
"audio_rate": None,
},
}
)
)
def test_unknown_legacy_audio_is_dropped(self) -> None:
# NULL means unprobed, not "the same as the other stream"
self.assertFalse(
self._check(
{
"main": {
"has_audio": None,
"audio_codec": None,
"audio_rate": None,
},
"sub": {"has_audio": None, "audio_codec": None, "audio_rate": None},
}
)
)
class TestPinnedStream(unittest.TestCase):
"""Pinning trades merged coverage for a uniform, copy-only source."""
def _pinned(self, stream: ExportStreamEnum) -> RecordingExporter:
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_020, True),
_span("/s1.mp4", 1_020, 1_040, False),
],
{"h264"},
)
exporter.stream = stream
return exporter
def test_auto_reports_no_pin(self) -> None:
self.assertIsNone(self._pinned(ExportStreamEnum.auto).pinned_stream)
def test_pinned_reports_its_stream(self) -> None:
self.assertEqual(self._pinned(ExportStreamEnum.sub).pinned_stream, "sub")
self.assertEqual(self._pinned(ExportStreamEnum.main).pinned_stream, "main")
def test_pinned_range_is_never_staged(self) -> None:
# nothing hands off inside one stream, so there is nothing to stage
exporter = self._pinned(ExportStreamEnum.sub)
with patch.object(RecordingExporter, "_stage_stream_runs") as stage:
self.assertTrue(exporter._prepare_stream_runs())
stage.assert_not_called()
self.assertEqual(exporter.staged_runs, [])
def test_pinned_playlist_url_carries_the_pin(self) -> None:
exporter = self._pinned(ExportStreamEnum.sub)
exporter._get_recordings_for_range = lambda _stream: [] # type: ignore[method-assign]
cmd, _lines = exporter.get_record_export_command("/exports/out.mp4")
self.assertTrue(
any("/vod/front/sub/start/" in token for token in cmd),
f"expected a sub-pinned playlist url in {cmd}",
)
def test_auto_playlist_url_stays_merged(self) -> None:
exporter = self._pinned(ExportStreamEnum.auto)
exporter._get_recordings_for_range = lambda _stream: [] # type: ignore[method-assign]
cmd, _lines = exporter.get_record_export_command("/exports/out.mp4")
self.assertTrue(any("/vod/front/start/" in token for token in cmd))
self.assertFalse(any("/vod/front/main/" in token for token in cmd))
class TestStagedFileCleanup(unittest.TestCase):
"""A staged path must be tracked before ffmpeg can write to it."""
def _exporter(self, tmpdir: str) -> RecordingExporter:
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_020, True),
_span("/s1.mp4", 1_020, 1_040, False),
],
{"h264"},
)
exporter._staged_run_path = lambda index: os.path.join( # type: ignore[method-assign]
tmpdir, f"export_stage_{index}.mp4"
)
return exporter
def test_partial_file_from_a_failed_run_is_removed(self) -> None:
# a killed ffmpeg (OOM, container stop) leaves whatever it muxed
with tempfile.TemporaryDirectory() as tmpdir:
exporter = self._exporter(tmpdir)
runs = [
StreamRun("main", 1_000, 1_020, "/m1.mp4"),
StreamRun("sub", 1_020, 1_040, "/s1.mp4"),
]
def fake_run(cmd, **kwargs):
# ffmpeg opens its output before it fails
Path(cmd[-1]).write_bytes(b"partial")
return (-9, "Killed")
with patch(
"frigate.record.export.run_ffmpeg_with_progress", side_effect=fake_run
):
self.assertFalse(exporter._stage_stream_runs(runs, {"h264"}, False))
self.assertEqual(os.listdir(tmpdir), [])
self.assertEqual(exporter.staged_runs, [])
def test_partial_file_from_a_later_run_is_removed(self) -> None:
"""The failing run must not orphan the runs that already succeeded."""
with tempfile.TemporaryDirectory() as tmpdir:
exporter = self._exporter(tmpdir)
runs = [
StreamRun("main", 1_000, 1_020, "/m1.mp4"),
StreamRun("sub", 1_020, 1_040, "/s1.mp4"),
]
calls = {"n": 0}
def fake_run(cmd, **kwargs):
Path(cmd[-1]).write_bytes(b"data")
calls["n"] += 1
return (0, "") if calls["n"] == 1 else (-9, "Killed")
with patch(
"frigate.record.export.run_ffmpeg_with_progress", side_effect=fake_run
):
self.assertFalse(exporter._stage_stream_runs(runs, {"h264"}, False))
self.assertEqual(os.listdir(tmpdir), [])
class TestStagingFailure(unittest.TestCase):
def test_failed_staging_aborts_rather_than_falling_back(self) -> None:
"""The merged playlist is the thing staging exists to avoid."""
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_020, True),
_span("/s1.mp4", 1_020, 1_040, False),
],
{"h264"},
)
with patch.object(RecordingExporter, "_stage_stream_runs", return_value=False):
self.assertFalse(exporter._prepare_stream_runs())
def test_successful_staging_reports_true(self) -> None:
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_020, True),
_span("/s1.mp4", 1_020, 1_040, False),
],
{"h264"},
)
with patch.object(RecordingExporter, "_stage_stream_runs", return_value=True):
self.assertTrue(exporter._prepare_stream_runs())
def test_single_stream_range_reports_true_without_staging(self) -> None:
exporter = _make_exporter([_span("/m1.mp4", 1_000, 1_020, True)], {"h264"})
with patch.object(RecordingExporter, "_stage_stream_runs") as stage:
self.assertTrue(exporter._prepare_stream_runs())
stage.assert_not_called()
class TestStagedExportCommand(unittest.TestCase):
def test_staged_runs_are_concatenated_with_stream_copy(self) -> None:
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_020, True),
_span("/s1.mp4", 1_020, 1_040, False),
],
{"h264"},
)
exporter.staged_runs = ["/cache/stage_0.mp4", "/cache/stage_1.mp4"]
cmd, playlist_lines = exporter.get_record_export_command("/exports/out.mp4")
self.assertEqual(
playlist_lines,
["file '/cache/stage_0.mp4'", "file '/cache/stage_1.mp4'"],
)
self.assertIn("concat", cmd)
self.assertIn("copy", cmd)
# nothing is left pointing at the merged vod route
self.assertFalse(any("/vod/front/start/" in token for token in cmd))
self.assertEqual(cmd[-1], "/exports/out.mp4")
def test_expected_duration_sums_the_merged_timeline(self) -> None:
"""A mixed range must not be measured by one stream alone."""
exporter = _make_exporter(
[
_span("/m1.mp4", 1_000, 1_020, True),
_span("/s1.mp4", 1_020, 1_050, False),
],
{"h264"},
)
self.assertEqual(exporter._expected_output_duration_seconds(), 50.0)
+9 -1
View File
@@ -12,7 +12,11 @@ from frigate.jobs.export import (
ExportJob,
ExportJobManager,
)
from frigate.record.export import PlaybackSourceEnum, RecordingExporter
from frigate.record.export import (
ExportStreamEnum,
PlaybackSourceEnum,
RecordingExporter,
)
from frigate.types import JobStatusTypesEnum
from frigate.util.ffmpeg import inject_progress_flags
@@ -38,7 +42,11 @@ def _make_exporter(
exporter.ffmpeg_input_args = ffmpeg_input_args
exporter.ffmpeg_output_args = ffmpeg_output_args
exporter.cpu_fallback = False
exporter.stream = ExportStreamEnum.auto
exporter.on_progress = on_progress
exporter.staged_runs = []
exporter.staged_transcode = False
exporter._coverage = ([], set(), False)
return exporter