From abaf91272dbd73468465b0817ac2af30304f6ac8 Mon Sep 17 00:00:00 2001 From: Joe Petrucelli Date: Tue, 27 Oct 2026 12:54:10 -0400 Subject: [PATCH 1/2] fix: size ffmpeg timeouts to the file; input-seek re-encode (0.1.2) A 110 GB day master on a 9p mount took longer than the flat 600 s keyframe probe, so keyframes came back unknown and every range fell back to a re-encode that decoded from the start of the file under a flat 3600 s timeout. Probe, copy and concat timeouts now scale with the bytes read at a 20 MB/s floor; the re-encode seeks on the input side with an output-side -ss 0 so copied audio carries no pre-cut packets, and its timeout scales with the range length. Releases 0.1.2. --- CHANGELOG.md | 16 ++++++- README.md | 11 +++-- pyproject.toml | 2 +- src/keycut/__init__.py | 8 +++- src/keycut/extract.py | 58 ++++++++++++++++++----- src/keycut/keyframes.py | 31 ++++++++++-- tests/conftest.py | 2 + tests/test_extract.py | 81 +++++++++++++++++++++++++++++++- tests/test_integration_ffmpeg.py | 11 +++-- tests/test_keyframes.py | 25 ++++++++++ 10 files changed, 219 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b2db7a..50c4cb1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,10 +4,24 @@ All notable changes to this project are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## Unreleased +## 0.1.2 - 2026-10-27 ### Fixed +- Every ffmpeg and ffprobe step now sizes its timeout to the file it reads, + at a 20 MB/s floor (`MIN_IO_BYTES_PER_SEC`, `io_timeout`), instead of a + flat window. The keyframe probe (600 s floor), a stream copy (1800 s floor) + and the concat (1800 s floor, twice the bytes with `faststart`) all read + the whole file. A 110 GB master on a slow mount took longer than 600 s to + probe, so keyframes came back unknown and every range fell back to a + re-encode. +- The re-encode fallback seeks on the input side (`-ss S -i master -ss 0 + -t D`) instead of the output side, so it decodes at most one GOP before + the cut rather than everything from the start of the file. It is still + frame-exact. The output-side `-ss 0` drops the stream-copied audio's + pre-cut packets, which the concat demuxer would otherwise play. Its + timeout is at least 20 s per second of footage + (`REENCODE_SEC_PER_SEC`), with a 3600 s floor. - `probe_keyframes` no longer caches a failed probe forever: `None` is now remembered for `FAILED_PROBE_RETRY_SEC` (30 s), so a transient ffprobe failure on unchanged media recovers, and repeated failures are re-probed at diff --git a/README.md b/README.md index d560382..9048fa4 100644 --- a/README.md +++ b/README.md @@ -161,8 +161,12 @@ extract_and_concat_ranges( Two things change. The merge refuses to close a gap containing an excluded second, however narrow — merging would re-admit exactly what you banned. And for any range where a copy would still snap back across an excluded second, -that one range is re-encoded with a frame-accurate seek (`-ss` after `-i`), -while every other range is still copied losslessly. +that one range is re-encoded with a frame-accurate seek, while every other +range is still copied losslessly. The re-encode seeks on the input side +(`-ss S -i master -ss 0 -t D`): the demuxer jumps to the keyframe before the +cut, the decoder drops frames up to the exact start, and the output-side +`-ss 0` drops the copied audio's pre-cut packets. That decodes at most one GOP +before the cut, not the whole file before it. The re-encode arguments are derived from the master's own codec, pixel format and colour metadata, because the concat demuxer joins streams without @@ -342,7 +346,8 @@ The pieces, usable on their own: | `snap_start_to_keyframe(start, end_limit, keyframes)` | Move a cut start forward onto a keyframe. | | `keyframe_at_or_before(t, keyframes)` | Where `-ss t -c copy` would actually land. | | `merge_ranges(ranges, max_gap_sec, is_excluded)` | Collapse overlapping, touching and sub-GOP-separated ranges. | -| `stream_copy_range` / `reencode_range` / `concat_demuxer` | The three ffmpeg invocations. | +| `stream_copy_range` / `reencode_range` / `concat_demuxer` | The three ffmpeg invocations. Each timeout is a floor, raised to fit the file's size or the range's length. | +| `io_timeout(floor_sec, nbytes)` | A timeout long enough to read `nbytes` at `MIN_IO_BYTES_PER_SEC` (20 MB/s). | | `default_reencode_args(master)` | Concat-compatible encoder arguments derived from the master. | | `probe_video_params(master)` | Codec, pixel format and colour metadata of the first video stream. | diff --git a/pyproject.toml b/pyproject.toml index f448e6c..dbb75c6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "keycut" -version = "0.1.1" +version = "0.1.2" description = "Lossless multi-range video extraction with ffmpeg stream copy, without the GOP-replay artefact." readme = "README.md" requires-python = ">=3.11" diff --git a/src/keycut/__init__.py b/src/keycut/__init__.py index 7210e95..fd93d1e 100644 --- a/src/keycut/__init__.py +++ b/src/keycut/__init__.py @@ -23,6 +23,7 @@ from .attribution import assign_clips_by_source_window from .extract import ( DEFAULT_CRF, + REENCODE_SEC_PER_SEC, concat_demuxer, default_reencode_args, extract_and_concat_ranges, @@ -34,8 +35,10 @@ from .keyframes import ( DEFAULT_GOP_DURATION_SEC, FAILED_PROBE_RETRY_SEC, + MIN_IO_BYTES_PER_SEC, clear_keyframe_cache, estimate_gop_duration, + io_timeout, keyframe_at_or_before, probe_keyframes, snap_start_to_keyframe, @@ -56,12 +59,14 @@ segments_to_master_ranges, ) -__version__ = "0.1.0" +__version__ = "0.1.2" __all__ = [ "DEFAULT_CRF", "DEFAULT_GOP_DURATION_SEC", "FAILED_PROBE_RETRY_SEC", + "MIN_IO_BYTES_PER_SEC", + "REENCODE_SEC_PER_SEC", "MaskLike", "Range", "Segment", @@ -76,6 +81,7 @@ "estimate_gop_duration", "extract_and_concat_ranges", "extract_and_concat_segments", + "io_timeout", "keyframe_at_or_before", "merge_ranges", "probe_keyframes", diff --git a/src/keycut/extract.py b/src/keycut/extract.py index a7cdfc0..a14a1f2 100644 --- a/src/keycut/extract.py +++ b/src/keycut/extract.py @@ -11,6 +11,7 @@ from .keyframes import ( DEFAULT_GOP_DURATION_SEC, estimate_gop_duration, + io_timeout, probe_keyframes, snap_start_to_keyframe, ) @@ -135,6 +136,12 @@ def default_reencode_args( return args +REENCODE_SEC_PER_SEC = 20.0 +"""Seconds of wall-clock allowed per second of footage re-encoded by +:func:`reencode_range` — 1.5 fps at 30 fps, slower than x265 ``-preset slow`` +manages on 4K with a modern CPU.""" + + def stream_copy_range( master_path: Path | str, start_sec: float, @@ -156,6 +163,9 @@ def stream_copy_range( Pass a keyframe-aligned start (see :func:`keycut.snap_start_to_keyframe`) or the output will contain footage from before the cut. + + ``timeout`` is a floor, raised to the time to read the whole master (see + :func:`keycut.io_timeout`), since no range can be larger than that. """ output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) @@ -172,7 +182,7 @@ def stream_copy_range( "copy", str(output_path), ] - return _run_ffmpeg(cmd, "stream copy", timeout) + return _run_ffmpeg(cmd, "stream copy", io_timeout(timeout, _size(master_path))) def reencode_range( @@ -187,11 +197,20 @@ def reencode_range( ) -> bool: """Decode and re-encode ``[start_sec, end_sec)`` with a frame-accurate seek. - ``-ss``/``-to`` go *after* ``-i``, making them output options: ffmpeg - decodes from the beginning of the file and discards frames until the exact - requested timestamp. That costs a full decode of everything before the cut, - and it re-encodes what it keeps — but it lands on the requested frame - rather than the previous keyframe. + ``-ss`` goes *before* ``-i``: the demuxer seeks to the keyframe at or before + ``start_sec``, and because this transcodes, ffmpeg's ``-accurate_seek`` (on + by default) decodes and drops the frames between that keyframe and the + exact start. The output lands on the requested frame, like an output-side + seek, but decodes at most one GOP before the cut instead of everything + from the start of the file. Timestamps restart at zero after an input + seek, so the length is given with ``-t``. + + Stream-copied audio is not trimmed by the input seek: its packets from the + keyframe on survive at negative timestamps behind an edit list, which the + concat demuxer ignores. An output-side ``-ss 0`` drops them. + + ``timeout`` is a floor, raised to :data:`REENCODE_SEC_PER_SEC` seconds per + second of footage: a long 4K range through a CPU encoder runs for hours. This is the escape hatch for the two cases a stream copy cannot serve: keyframe positions are unknown, or the snap-back region contains footage @@ -199,19 +218,24 @@ def reencode_range( """ output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) + duration = float(end_sec) - float(start_sec) cmd = [ ffmpeg_bin, "-y", + "-ss", + f"{float(start_sec)}", "-i", str(master_path), "-ss", - f"{float(start_sec)}", - "-to", - f"{float(end_sec)}", + "0", + "-t", + f"{duration}", *list(encode_args), str(output_path), ] - return _run_ffmpeg(cmd, "re-encode", timeout) + return _run_ffmpeg( + cmd, "re-encode", max(float(timeout), duration * REENCODE_SEC_PER_SEC) + ) def concat_demuxer( @@ -227,6 +251,10 @@ def concat_demuxer( Writes the ``file ''`` manifest to ``list_file`` first, single-quote- escaped, and passes ``-safe 0`` so absolute paths are accepted. + + ``timeout`` is a floor, raised to the time to move the pieces' bytes (see + :func:`keycut.io_timeout`) — twice over with ``faststart``, which rewrites + the whole output a second time to move the moov atom. """ list_file = Path(list_file) output_path = Path(output_path) @@ -242,7 +270,15 @@ def concat_demuxer( if faststart: cmd += ["-movflags", "+faststart"] cmd.append(str(output_path)) - return _run_ffmpeg(cmd, "concat", timeout) + nbytes = sum(_size(p) for p in parts) * (2 if faststart else 1) + return _run_ffmpeg(cmd, "concat", io_timeout(timeout, nbytes)) + + +def _size(path: Path | str) -> int: + try: + return Path(path).stat().st_size + except OSError: + return 0 def _run_ffmpeg(cmd: list[str], what: str, timeout: float) -> bool: diff --git a/src/keycut/keyframes.py b/src/keycut/keyframes.py index 17fbae4..4efdb57 100644 --- a/src/keycut/keyframes.py +++ b/src/keycut/keyframes.py @@ -36,6 +36,24 @@ most once per window rather than on every call. """ +MIN_IO_BYTES_PER_SEC = 20_000_000 +"""Slowest read/write rate, in bytes per second, a healthy ffmpeg or ffprobe +run is assumed to sustain. + +Every subprocess timeout here is a hang guard, so it must outlast honest work. +Reading a 118 GB day master over a network or 9p mount (WSL's ``/mnt/c``) +runs at roughly 50 MB/s; a flat 600 s or 1800 s window kills that read +part-way through. :func:`io_timeout` sizes each window to the bytes involved +at this floor. +""" + + +def io_timeout(floor_sec: float, nbytes: int) -> float: + """``floor_sec``, or the time to move ``nbytes`` at + :data:`MIN_IO_BYTES_PER_SEC`, whichever is longer.""" + return max(float(floor_sec), nbytes / MIN_IO_BYTES_PER_SEC) + + # Keyframe timestamps keyed by (resolved path, mtime_ns, size) so a file is # probed at most once per process even when several extractions share it. # Each value is ``(keyframes, retry_at)``. A success has ``retry_at=None`` and @@ -70,9 +88,12 @@ def probe_keyframes( join. Because it walks the whole packet index, expect roughly a second per hour - of footage on a local disk. Results are cached per (path, mtime, size); a - failed probe is cached only for :data:`FAILED_PROBE_RETRY_SEC` seconds, so - a transient failure recovers on a later call. + of footage on a local disk; ffprobe still reads every byte, so on a slow + mount it takes as long as reading the file. ``timeout`` is a floor: the + window grows with the file's size (see :func:`io_timeout`). Results are + cached per (path, mtime, size); a failed probe is cached only for + :data:`FAILED_PROBE_RETRY_SEC` seconds, so a transient failure recovers on + a later call. ``None`` means "unknown", not "no keyframes". Callers must treat it as a reason to re-encode rather than a reason to cut blind. @@ -108,7 +129,9 @@ def probe_keyframes( keyframes: list[float] | None = None try: - result = subprocess.run(cmd, capture_output=True, timeout=timeout) + result = subprocess.run( + cmd, capture_output=True, timeout=io_timeout(timeout, st.st_size) + ) if result.returncode == 0: parsed: list[float] = [] for line in (result.stdout or b"").decode("utf-8", errors="replace").splitlines(): diff --git a/tests/conftest.py b/tests/conftest.py index 57ca3fa..3e2b806 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -56,11 +56,13 @@ def __init__(self, *, keyframe_stdout: bytes, stream_stdout: bytes, returncode: self.stream_stdout = stream_stdout self.returncode = returncode self.commands: list[list[str]] = [] + self.timeouts: list[float | None] = [] self.list_files: list[str] = [] def __call__(self, cmd, *args, **kwargs): cmd = list(cmd) self.commands.append(cmd) + self.timeouts.append(kwargs.get("timeout")) joined = " ".join(cmd) if cmd[0].endswith("ffprobe"): if "packet=" in joined: diff --git a/tests/test_extract.py b/tests/test_extract.py index ea8230a..6f33dac 100644 --- a/tests/test_extract.py +++ b/tests/test_extract.py @@ -16,6 +16,7 @@ ) from keycut import ( Segment, + concat_demuxer, default_reencode_args, extract_and_concat_ranges, extract_and_concat_segments, @@ -38,6 +39,16 @@ def _to(cmd): return float(cmd[cmd.index("-to") + 1]) +def _t(cmd): + return float(cmd[cmd.index("-t") + 1]) + + +def _sparse(path, size): + with path.open("wb") as fh: + fh.truncate(size) # no disk is used + return path + + class TestHappyPath: def test_one_extraction_per_range_then_a_concat(self, master, tmp_path): with fake_ffmpeg() as fake: @@ -169,8 +180,17 @@ def test_unknown_keyframes_force_an_accurate_seek_re_encode(self, master, tmp_pa assert result is not None cmd = fake.extract_commands[0] assert is_reencode(cmd) - assert not seek_is_input_side(cmd), "an accurate seek must be output-side" + assert seek_is_input_side(cmd), ( + "an output-side seek decodes everything before the cut; a re-encode " + "with an input-side seek is just as exact (ffmpeg's -accurate_seek)" + ) assert _ss(cmd) == 100.0, "the re-encode lands on the exact requested frame" + assert _t(cmd) == 60.0, "the input seek resets timestamps, so length is -t" + after_input = cmd[cmd.index("-i"):] + assert "-ss" in after_input and float(after_input[after_input.index("-ss") + 1]) == 0.0, ( + "stream-copied audio keeps the packets back to the keyframe at negative " + "timestamps; an output-side -ss 0 drops them" + ) def test_a_clean_cut_still_takes_the_lossless_path(self, master, tmp_path): with fake_ffmpeg() as fake: @@ -376,3 +396,62 @@ def test_the_default_stream_json_fixture_is_what_the_probe_reads(self, master): # Guards the fixture itself: if DEFAULT_STREAM_JSON drifts, every # assertion above silently starts testing something else. assert b"h264" in DEFAULT_STREAM_JSON + + +class TestLargeMasterTimeouts: + """A day master can be 100+ GB on a slow mount. Every ffmpeg step gets a + window sized to the work, so a healthy run is never killed as a hang.""" + + def test_a_stream_copy_window_scales_with_the_master(self, tmp_path): + master = _sparse(tmp_path / "day_master.mp4", 118_000_000_000) + with fake_ffmpeg() as fake: + extract_and_concat_ranges( + master, [(100.0, 160.0)], tmp_path / "out.mp4", tmp_path / "work", + keyframes=[100.0, 200.0], + ) + copy_idx = fake.commands.index(fake.extract_commands[0]) + assert fake.timeouts[copy_idx] >= 118_000_000_000 / 20_000_000 + + def test_a_small_stream_copy_keeps_the_1800s_floor(self, master, tmp_path): + with fake_ffmpeg() as fake: + extract_and_concat_ranges( + master, [(100.0, 160.0)], tmp_path / "out.mp4", tmp_path / "work", + keyframes=[100.0, 200.0], + ) + copy_idx = fake.commands.index(fake.extract_commands[0]) + assert fake.timeouts[copy_idx] == 1800.0 + + def test_a_re_encode_window_scales_with_the_range_length(self, master, tmp_path): + """A 1147 s range of 4K re-encoded with x265 on a CPU runs for hours, + far past the flat 3600 s.""" + with fake_ffmpeg() as fake: + extract_and_concat_ranges( + master, [(3000.0, 4147.0)], tmp_path / "out.mp4", tmp_path / "work", + keyframes=[], reencode_args=["-c:v", "libx265"], + ) + idx = fake.commands.index(fake.extract_commands[0]) + assert fake.timeouts[idx] >= 1147.0 * 20 + + def test_a_short_re_encode_keeps_the_3600s_floor(self, master, tmp_path): + with fake_ffmpeg() as fake: + extract_and_concat_ranges( + master, [(100.0, 102.0)], tmp_path / "out.mp4", tmp_path / "work", + keyframes=[], reencode_args=["-c:v", "libx265"], + ) + idx = fake.commands.index(fake.extract_commands[0]) + assert fake.timeouts[idx] == 3600.0 + + def test_a_concat_window_scales_with_the_pieces(self, tmp_path): + parts = [ + _sparse(tmp_path / "a.mp4", 30_000_000_000), + _sparse(tmp_path / "b.mp4", 21_000_000_000), + ] + with fake_ffmpeg() as fake: + concat_demuxer(parts, tmp_path / "out.mp4", tmp_path / "list.txt", + faststart=False) + concat_demuxer(parts, tmp_path / "out2.mp4", tmp_path / "list2.txt") + plain, faststart = fake.timeouts + assert plain >= 51_000_000_000 / 20_000_000 + assert faststart >= 2 * 51_000_000_000 / 20_000_000, ( + "+faststart rewrites the whole output a second time" + ) diff --git a/tests/test_integration_ffmpeg.py b/tests/test_integration_ffmpeg.py index ca45123..6d94555 100644 --- a/tests/test_integration_ffmpeg.py +++ b/tests/test_integration_ffmpeg.py @@ -266,10 +266,10 @@ def test_an_accurate_seek_re_encode_delivers_the_exact_span(self, source, tmp_pa """``keyframes=[]`` declares them unusable, so every range re-encodes. This is the one path that returns exactly what was asked for — 60 - frames, no more and no fewer — and it pins the semantics keycut relies - on: with ``-ss``/``-to`` placed after ``-i``, both are still read in the - input's timeline, so ``-to 12.5`` means 12.5 in the source and not 12.5 - seconds after the seek. + frames, no more and no fewer. The audio is stream-copied, and after an + input seek it would start at the keyframe (10.0), half a second early, + hidden behind an edit list the concat demuxer ignores. The output-side + ``-ss 0`` drops it, so the file is 2.0 s long in both streams. """ out = extract_and_concat_ranges( source, [(10.5, 12.5)], tmp_path / "out.mp4", tmp_path / "work", @@ -277,6 +277,9 @@ def test_an_accurate_seek_re_encode_delivers_the_exact_span(self, source, tmp_pa ) assert out is not None assert len(video_packets(out)) == int(2.0 * FIXTURE_FPS) + assert probe_duration(out) == pytest.approx(2.0, abs=0.1), ( + "audio from before the cut leaked into the re-encoded range" + ) assert decodes_cleanly(out) diff --git a/tests/test_keyframes.py b/tests/test_keyframes.py index d3581a3..98846f5 100644 --- a/tests/test_keyframes.py +++ b/tests/test_keyframes.py @@ -104,6 +104,31 @@ def _fake_run(cmd, *a, **kw): assert len(calls) == 1, "a second probe of the same file must hit the cache" assert calls[0][0] == "ffprobe" + def test_the_probe_window_scales_with_the_file_size(self, tmp_path): + """ffprobe reads every packet, so the probe takes as long as reading + the whole file. A 110 GB day master on a slow mount needed far more + than the flat 600 s and was killed, which sent every cut down the + re-encode path.""" + big = tmp_path / "day_master.mp4" + with big.open("wb") as fh: + fh.truncate(118_000_000_000) # sparse: no disk is used + small = tmp_path / "short.mp4" + small.write_bytes(b"not really a video") + timeouts = [] + + def _fake_run(cmd, *a, **kw): + timeouts.append(kw.get("timeout")) + return MagicMock(returncode=0, stdout=b"0.000000,K__\n", stderr=b"") + + with patch("keycut.keyframes.subprocess.run", side_effect=_fake_run): + probe_keyframes(small) + probe_keyframes(big) + + assert timeouts[0] == 600.0, "a small file keeps the 600 s floor" + assert timeouts[1] >= 118_000_000_000 / 20_000_000, ( + "a 118 GB file gets at least 20 MB/s worth of time" + ) + def test_missing_file_returns_none(self, tmp_path): assert probe_keyframes(tmp_path / "nope.mp4") is None From 0e8564baca616cba7b1be8c37f2a0ba3611ed0d3 Mon Sep 17 00:00:00 2001 From: Joe Petrucelli Date: Tue, 27 Oct 2026 14:17:26 -0400 Subject: [PATCH 2/2] feat: assembly receipt of what each extraction actually cut (0.2.0) --- CHANGELOG.md | 16 +++ README.md | 60 +++++++++++ pyproject.toml | 2 +- src/keycut/__init__.py | 9 +- src/keycut/extract.py | 166 +++++++++++++++++++++++++++++-- src/keycut/receipt.py | 115 +++++++++++++++++++++ tests/conftest.py | 20 +++- tests/test_integration_ffmpeg.py | 38 ++++++- tests/test_receipt.py | 157 +++++++++++++++++++++++++++++ 9 files changed, 569 insertions(+), 14 deletions(-) create mode 100644 src/keycut/receipt.py create mode 100644 tests/test_receipt.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 50c4cb1..7728ce9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,22 @@ All notable changes to this project are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## 0.2.0 - 2026-10-27 + +### Added + +- Assembly receipt, schema `keycut-assembly-receipt/1`: a record of what an + extraction actually cut. For each merged span it gives the requested ranges + absorbed, the keyframe-snapped source start, copy versus re-encode and the + reason, and where the span landed in the output (parts are probed with + ffprobe; a failed probe falls back to the planned length and says so). + `AssemblyReceipt` and `ReceiptSpan` offer `to_dict()`, `from_dict()` and + `write_json()`. +- `extract_and_concat_ranges_with_receipt` and + `extract_and_concat_segments_with_receipt`, returning `(path, receipt)`, or + `(None, None)` on failure. `extract_and_concat_ranges` and + `extract_and_concat_segments` are unchanged in signature and behaviour. + ## 0.1.2 - 2026-10-27 ### Fixed diff --git a/README.md b/README.md index 9048fa4..bd28f3d 100644 --- a/README.md +++ b/README.md @@ -334,6 +334,9 @@ Cutting: | `extract_and_concat_ranges(master, ranges, output_path, work_dir, **opts)` | The main entry point. Returns the output path, or `None` on failure. | | `extract_and_concat_segments(master, segments, boundaries, output_path, work_dir, **opts)` | Same, for ranges expressed against the master's source clips. | +| `extract_and_concat_ranges_with_receipt(...)` | Same arguments; returns `(path, receipt)`, or `(None, None)` on failure. See [The assembly receipt](#the-assembly-receipt). | +| `extract_and_concat_segments_with_receipt(...)` | The segment form of the above. | + Options: `exclude_mask`, `max_gap_sec`, `reencode_args`, `allow_reencode`, `keyframes`, `ffmpeg_bin`, `ffprobe_bin`, `keep_scratch`, `faststart`. @@ -360,6 +363,63 @@ Optional companion, not part of the cutting path: feeds two independent products, works out which output each derived clip came from, by source footage rather than by dividing one count into another. +## The assembly receipt + +`extract_and_concat_ranges` returns only a path. To learn what was actually cut +— where each cut start snapped, which requested ranges were merged, and whether +a span was copied or re-encoded — call the `_with_receipt` form: + +```python +from keycut import extract_and_concat_ranges_with_receipt + +path, receipt = extract_and_concat_ranges_with_receipt( + "master.mp4", [(3.0, 9.0), (15.0, 21.0)], "out.mp4", "scratch/" +) +if receipt is not None: + receipt.write_json("out.receipt.json") +``` + +The receipt is `None` exactly when the path is. `receipt.to_dict()` returns the +schema below; `AssemblyReceipt.from_dict()` reads it back. + +```json +{ + "schema": "keycut-assembly-receipt/1", + "provenance": "keycut", + "keycut_version": "0.2.0", + "master": {"path": "master.mp4", "size_bytes": 123456}, + "time_base": "seconds", + "max_gap_sec": 2.0, + "requested": [[3.0, 9.0], [15.0, 21.0]], + "spans": [ + {"index": 0, "requested_indices": [0], + "merged_start": 3.0, "merged_end": 9.0, + "source_start": 4.0, "source_end": 9.0, + "method": "copy", "reason": "stream_copy", + "output_start": 0.0, "output_end": 5.0, + "duration_source": "probed"} + ], + "output": {"path": "out.mp4", "duration": 11.0} +} +``` + +All times are seconds, half-open `[start, end)`. + +- `requested_indices`: the input ranges that were merged into this span. +- `merged_start` / `merged_end`: the span after merging, before any alignment. +- `source_start`: where the cut really begins in the master. For `copy` it is + the keyframe the start was snapped forward to; for `reencode` it equals + `merged_start`, because a re-encode seeks exactly. `source_end` is + `merged_end`. +- `method` / `reason`: `copy` with `stream_copy`; or `reencode` with + `keyframes_unknown` or `snap_back_over_excluded`. +- `output_start` / `output_end`: where the span lands in the output, contiguous + from 0 in output order. The length is the part's probed duration + (`duration_source: "probed"`); if ffprobe fails it is + `source_end - source_start` and `duration_source` is `"planned"`. +- `output.duration` is the probed duration of the final file, `null` if the + probe failed. + ## Development ```bash diff --git a/pyproject.toml b/pyproject.toml index dbb75c6..e4a137b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "keycut" -version = "0.1.2" +version = "0.2.0" description = "Lossless multi-range video extraction with ffmpeg stream copy, without the GOP-replay artefact." readme = "README.md" requires-python = ">=3.11" diff --git a/src/keycut/__init__.py b/src/keycut/__init__.py index fd93d1e..fa1d3f9 100644 --- a/src/keycut/__init__.py +++ b/src/keycut/__init__.py @@ -27,7 +27,9 @@ concat_demuxer, default_reencode_args, extract_and_concat_ranges, + extract_and_concat_ranges_with_receipt, extract_and_concat_segments, + extract_and_concat_segments_with_receipt, probe_video_params, reencode_range, stream_copy_range, @@ -51,6 +53,7 @@ merge_ranges, span_has_excluded_second, ) +from .receipt import AssemblyReceipt, ReceiptSpan from .segments import ( Segment, boundaries_from_durations, @@ -59,9 +62,10 @@ segments_to_master_ranges, ) -__version__ = "0.1.2" +__version__ = "0.2.0" __all__ = [ + "AssemblyReceipt", "DEFAULT_CRF", "DEFAULT_GOP_DURATION_SEC", "FAILED_PROBE_RETRY_SEC", @@ -69,6 +73,7 @@ "REENCODE_SEC_PER_SEC", "MaskLike", "Range", + "ReceiptSpan", "Segment", "__version__", "as_mask_lookup", @@ -80,7 +85,9 @@ "default_reencode_args", "estimate_gop_duration", "extract_and_concat_ranges", + "extract_and_concat_ranges_with_receipt", "extract_and_concat_segments", + "extract_and_concat_segments_with_receipt", "io_timeout", "keyframe_at_or_before", "merge_ranges", diff --git a/src/keycut/extract.py b/src/keycut/extract.py index a14a1f2..0404c35 100644 --- a/src/keycut/extract.py +++ b/src/keycut/extract.py @@ -4,6 +4,7 @@ import json import logging +import math import subprocess from collections.abc import Iterable, Sequence from pathlib import Path @@ -22,6 +23,7 @@ copy_would_snap_back_over_excluded, merge_ranges, ) +from .receipt import AssemblyReceipt, ReceiptSpan from .segments import Segment, segments_to_master_ranges logger = logging.getLogger(__name__) @@ -301,6 +303,46 @@ def _run_ffmpeg(cmd: list[str], what: str, timeout: float) -> bool: return True +def _probe_duration( + path: Path | str, ffprobe_bin: str = "ffprobe", timeout: float = 60.0 +) -> float | None: + """The container duration of ``path`` in seconds, or ``None`` if it cannot + be probed. The timeout is a floor, raised to fit the file's size.""" + cmd = [ + ffprobe_bin, "-v", "error", "-show_entries", "format=duration", + "-of", "csv=p=0", str(path), + ] + try: + result = subprocess.run(cmd, capture_output=True, timeout=io_timeout(timeout, _size(path))) + if result.returncode != 0: + return None + value = float((result.stdout or b"").decode("utf-8", errors="replace").strip()) + except Exception as exc: + logger.debug("keycut: duration probe failed for %s: %s", path, exc) + return None + return value if math.isfinite(value) and value >= 0 else None + + +def _requested_membership( + ranges: Sequence[Range], + n_merged: int, + max_gap_sec: float, + mask_lookup: MaskLike | None, +) -> list[tuple[int, ...]]: + """For each merged span, the indices of the requested ranges it absorbed. + + Replays :func:`merge_ranges` over growing prefixes: it only ever extends or + appends to the last span, so range ``k`` belongs to span + ``len(merge(ranges[:k+1])) - 1``. Exact by construction, no second copy of + the merge rule. + """ + members: list[list[int]] = [[] for _ in range(n_merged)] + for k in range(len(ranges)): + count = len(merge_ranges(ranges[: k + 1], max_gap_sec=max_gap_sec, is_excluded=mask_lookup)) + members[count - 1].append(k) + return [tuple(m) for m in members] + + def extract_and_concat_ranges( master_path: Path | str, ranges: Sequence[Range], @@ -319,6 +361,46 @@ def extract_and_concat_ranges( ) -> Path | None: """Cut every range out of ``master_path`` and join them into one file. + Returns ``output_path`` on success, ``None`` on any failure. See + :func:`extract_and_concat_ranges_with_receipt` for the full description of + the algorithm and every argument; this is the same call without the receipt. + """ + path, _ = extract_and_concat_ranges_with_receipt( + master_path, + ranges, + output_path, + work_dir, + exclude_mask=exclude_mask, + max_gap_sec=max_gap_sec, + reencode_args=reencode_args, + allow_reencode=allow_reencode, + keyframes=keyframes, + ffmpeg_bin=ffmpeg_bin, + ffprobe_bin=ffprobe_bin, + keep_scratch=keep_scratch, + faststart=faststart, + ) + return path + + +def extract_and_concat_ranges_with_receipt( + master_path: Path | str, + ranges: Sequence[Range], + output_path: Path | str, + work_dir: Path | str, + *, + exclude_mask: MaskLike | None = None, + max_gap_sec: float | None = None, + reencode_args: Sequence[str] | None = None, + allow_reencode: bool = True, + keyframes: Sequence[float] | None = None, + ffmpeg_bin: str = "ffmpeg", + ffprobe_bin: str = "ffprobe", + keep_scratch: bool = False, + faststart: bool = True, +) -> tuple[Path | None, AssemblyReceipt | None]: + """Cut every range out of ``master_path`` and join them into one file. + The whole algorithm, in order: 1. **Merge** ranges that overlap, touch, or sit closer together than one @@ -358,9 +440,12 @@ def extract_and_concat_ranges( ``work_dir`` for inspection. Returns: - ``output_path`` on success, ``None`` on any failure — missing master, - empty range list, or an ffmpeg command that exited non-zero. Nothing - raises; a failed extraction is a normal outcome to branch on. + ``(output_path, receipt)`` on success: an :class:`AssemblyReceipt` + recording each merged span, its keyframe-snapped source start, whether + it was copied or re-encoded and why, and where it landed in the output. + ``(None, None)`` on any failure — missing master, empty range list, or + an ffmpeg command that exited non-zero. Nothing raises; a failed + extraction is a normal outcome to branch on. """ master_path = Path(master_path) output_path = Path(output_path) @@ -368,11 +453,12 @@ def extract_and_concat_ranges( if not master_path.exists(): logger.error("keycut: master missing: %s", master_path) - return None + return None, None if not ranges: logger.error("keycut: no ranges to extract from %s", master_path) - return None + return None, None + master_size = _size(master_path) work_dir.mkdir(parents=True, exist_ok=True) mask_lookup = as_mask_lookup(exclude_mask) @@ -399,6 +485,7 @@ def extract_and_concat_ranges( reencode_args = default_reencode_args(master_path, ffprobe_bin=ffprobe_bin) parts: list[Path] = [] + plans: list[tuple[float, float, str, str]] = [] # source_start, source_end, method, reason for i, (start, end) in enumerate(merged): part = work_dir / f"keycut_{i:03d}{output_path.suffix or '.mp4'}" aligned = snap_start_to_keyframe(start, end, keyframes) @@ -411,6 +498,7 @@ def extract_and_concat_ranges( ffmpeg_bin=ffmpeg_bin, ) how = "re-encode (keyframes unknown)" + plan = (start, end, "reencode", "keyframes_unknown") elif allow_reencode and copy_would_snap_back_over_excluded( aligned, keyframes, mask_lookup ): @@ -420,6 +508,7 @@ def extract_and_concat_ranges( ffmpeg_bin=ffmpeg_bin, ) how = "re-encode (copy would snap back over excluded footage)" + plan = (start, end, "reencode", "snap_back_over_excluded") else: if aligned != start: logger.debug("keycut: range %d start aligned %.3f -> %.3f", i, start, aligned) @@ -427,20 +516,61 @@ def extract_and_concat_ranges( master_path, aligned, end, part, ffmpeg_bin=ffmpeg_bin ) how = "stream copy" + plan = (aligned, end, "copy", "stream_copy") if not ok: logger.error( "keycut: %s failed on range %d/%d (%.3f -> %.3f) of %s", how, i + 1, len(merged), start, end, master_path, ) - return None + return None, None parts.append(part) + plans.append(plan) + + part_durations = [_probe_duration(p, ffprobe_bin) for p in parts] list_file = work_dir / f"{output_path.stem}_concat.txt" if not concat_demuxer( parts, output_path, list_file, ffmpeg_bin=ffmpeg_bin, faststart=faststart ): - return None + return None, None + + spans: list[ReceiptSpan] = [] + cursor = 0.0 + membership = _requested_membership(ranges, len(merged), max_gap_sec, mask_lookup) + for i, ((m_start, m_end), (src_start, src_end, method, reason)) in enumerate( + zip(merged, plans, strict=True) + ): + probed = part_durations[i] + duration = probed if probed is not None else src_end - src_start + spans.append( + ReceiptSpan( + index=i, + requested_indices=membership[i], + merged_start=float(m_start), + merged_end=float(m_end), + source_start=float(src_start), + source_end=float(src_end), + method=method, + reason=reason, + output_start=cursor, + output_end=cursor + duration, + duration_source="probed" if probed is not None else "planned", + ) + ) + cursor += duration + from . import __version__ + + receipt = AssemblyReceipt( + keycut_version=__version__, + master_path=str(master_path), + master_size_bytes=master_size, + max_gap_sec=float(max_gap_sec), + requested=tuple((float(a), float(b)) for a, b in ranges), + spans=tuple(spans), + output_path=str(output_path), + output_duration=_probe_duration(output_path, ffprobe_bin), + ) if not keep_scratch: for p in [*parts, list_file]: @@ -449,7 +579,7 @@ def extract_and_concat_ranges( except OSError: pass - return output_path + return output_path, receipt def extract_and_concat_segments( @@ -468,11 +598,27 @@ def extract_and_concat_segments( names a part outside ``boundaries``, so one bad reference fails the whole extraction visibly instead of producing a short output nobody notices. """ + path, _ = extract_and_concat_segments_with_receipt( + master_path, segments, boundaries, output_path, work_dir, **kwargs + ) + return path + + +def extract_and_concat_segments_with_receipt( + master_path: Path | str, + segments: Iterable[Segment], + boundaries: Sequence[Range], + output_path: Path | str, + work_dir: Path | str, + **kwargs, +) -> tuple[Path | None, AssemblyReceipt | None]: + """:func:`extract_and_concat_segments` that also returns the + :class:`AssemblyReceipt`. ``(None, None)`` on any failure.""" try: ranges = segments_to_master_ranges(segments, boundaries) except IndexError as exc: logger.error("keycut: segment translation failed: %s", exc) - return None - return extract_and_concat_ranges( + return None, None + return extract_and_concat_ranges_with_receipt( master_path, ranges, output_path, work_dir, **kwargs ) diff --git a/src/keycut/receipt.py b/src/keycut/receipt.py new file mode 100644 index 0000000..8a589d2 --- /dev/null +++ b/src/keycut/receipt.py @@ -0,0 +1,115 @@ +"""The assembly receipt: a record of what an extraction actually cut. + +Schema ``keycut-assembly-receipt/1``. All times are seconds, half-open +``[start, end)``. The receipt is plain data: ``to_dict()`` yields JSON-safe +values and ``from_dict()`` reads them back. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +SCHEMA = "keycut-assembly-receipt/1" +PROVENANCE = "keycut" + + +@dataclass(frozen=True) +class ReceiptSpan: + """One merged extraction and where it landed in the output.""" + + index: int + requested_indices: tuple[int, ...] + merged_start: float + merged_end: float + source_start: float + source_end: float + method: str # "copy" | "reencode" + reason: str # "stream_copy" | "keyframes_unknown" | "snap_back_over_excluded" + output_start: float + output_end: float + duration_source: str # "probed" | "planned" + + def to_dict(self) -> dict[str, Any]: + return { + "index": self.index, + "requested_indices": list(self.requested_indices), + "merged_start": self.merged_start, + "merged_end": self.merged_end, + "source_start": self.source_start, + "source_end": self.source_end, + "method": self.method, + "reason": self.reason, + "output_start": self.output_start, + "output_end": self.output_end, + "duration_source": self.duration_source, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ReceiptSpan: + return cls( + index=int(data["index"]), + requested_indices=tuple(int(i) for i in data["requested_indices"]), + merged_start=float(data["merged_start"]), + merged_end=float(data["merged_end"]), + source_start=float(data["source_start"]), + source_end=float(data["source_end"]), + method=str(data["method"]), + reason=str(data["reason"]), + output_start=float(data["output_start"]), + output_end=float(data["output_end"]), + duration_source=str(data["duration_source"]), + ) + + +@dataclass(frozen=True) +class AssemblyReceipt: + """What :func:`keycut.extract_and_concat_ranges_with_receipt` actually did.""" + + keycut_version: str + master_path: str + master_size_bytes: int | None + max_gap_sec: float + requested: tuple[tuple[float, float], ...] + spans: tuple[ReceiptSpan, ...] + output_path: str + output_duration: float | None + + def to_dict(self) -> dict[str, Any]: + return { + "schema": SCHEMA, + "provenance": PROVENANCE, + "keycut_version": self.keycut_version, + "master": {"path": self.master_path, "size_bytes": self.master_size_bytes}, + "time_base": "seconds", + "max_gap_sec": self.max_gap_sec, + "requested": [[s, e] for s, e in self.requested], + "spans": [span.to_dict() for span in self.spans], + "output": {"path": self.output_path, "duration": self.output_duration}, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> AssemblyReceipt: + if data.get("schema") != SCHEMA: + raise ValueError(f"unsupported receipt schema: {data.get('schema')!r}") + size = data["master"]["size_bytes"] + duration = data["output"]["duration"] + return cls( + keycut_version=str(data["keycut_version"]), + master_path=str(data["master"]["path"]), + master_size_bytes=None if size is None else int(size), + max_gap_sec=float(data["max_gap_sec"]), + requested=tuple((float(s), float(e)) for s, e in data["requested"]), + spans=tuple(ReceiptSpan.from_dict(s) for s in data["spans"]), + output_path=str(data["output"]["path"]), + output_duration=None if duration is None else float(duration), + ) + + def write_json(self, path: Path | str) -> Path: + """Write the receipt as indented JSON; returns the path.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(self.to_dict(), indent=2) + "\n", encoding="utf-8") + return path diff --git a/tests/conftest.py b/tests/conftest.py index 3e2b806..3669ab5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -51,7 +51,17 @@ class FakeFfmpeg: afterwards. """ - def __init__(self, *, keyframe_stdout: bytes, stream_stdout: bytes, returncode: int = 0): + def __init__( + self, + *, + keyframe_stdout: bytes, + stream_stdout: bytes, + returncode: int = 0, + durations: list[float] | None = None, + ): + # ``format=duration`` probes answer from this list in call order; a + # missing entry (or ``None`` list) makes that probe fail. + self.durations = list(durations) if durations is not None else [] self.keyframe_stdout = keyframe_stdout self.stream_stdout = stream_stdout self.returncode = returncode @@ -65,6 +75,12 @@ def __call__(self, cmd, *args, **kwargs): self.timeouts.append(kwargs.get("timeout")) joined = " ".join(cmd) if cmd[0].endswith("ffprobe"): + if "format=duration" in joined: + if not self.durations: + return MagicMock(returncode=1, stdout=b"", stderr=b"no duration") + return MagicMock( + returncode=0, stdout=f"{self.durations.pop(0)}\n".encode(), stderr=b"" + ) if "packet=" in joined: return MagicMock(returncode=0, stdout=self.keyframe_stdout, stderr=b"") return MagicMock(returncode=0, stdout=self.stream_stdout, stderr=b"") @@ -110,6 +126,7 @@ def fake_ffmpeg( keyframe_stdout: bytes | None = None, stream_stdout: bytes | None = None, returncode: int = 0, + durations: list[float] | None = None, ): """Patch both modules' ``subprocess.run`` with one :class:`FakeFfmpeg`. @@ -120,6 +137,7 @@ def fake_ffmpeg( keyframe_stdout=keyframe_csv() if keyframe_stdout is None else keyframe_stdout, stream_stdout=DEFAULT_STREAM_JSON if stream_stdout is None else stream_stdout, returncode=returncode, + durations=durations, ) with patch("keycut.extract.subprocess.run", side_effect=fake), patch( "keycut.keyframes.subprocess.run", side_effect=fake diff --git a/tests/test_integration_ffmpeg.py b/tests/test_integration_ffmpeg.py index 6d94555..4443b24 100644 --- a/tests/test_integration_ffmpeg.py +++ b/tests/test_integration_ffmpeg.py @@ -10,13 +10,19 @@ from __future__ import annotations +import json import subprocess from pathlib import Path import pytest from conftest import requires_ffmpeg -from keycut import extract_and_concat_ranges, probe_keyframes +from keycut import ( + AssemblyReceipt, + extract_and_concat_ranges, + extract_and_concat_ranges_with_receipt, + probe_keyframes, +) from keycut.cli import main pytestmark = requires_ffmpeg @@ -298,3 +304,33 @@ def test_dry_run_reports_the_measured_gop(self, source, tmp_path, capsys): out = capsys.readouterr().out assert "keyframes: 15" in out assert "merge gap: 2.1" in out, "measured from the fixture's real 2.0s GOP" + + +class TestAssemblyReceipt: + def test_receipt_describes_the_real_output(self, source, tmp_path): + # 3.0 and 15.0 sit mid-GOP (keyframes every 2.0 s), so each copy must + # start on the next keyframe: 4.0 and 16.0. + out, receipt = extract_and_concat_ranges_with_receipt( + source, + [(3.0, 9.0), (15.0, 21.0)], + tmp_path / "out.mp4", + tmp_path / "work", + ) + assert out is not None and receipt is not None + assert [s.method for s in receipt.spans] == ["copy", "copy"] + assert [s.requested_indices for s in receipt.spans] == [(0,), (1,)] + assert receipt.spans[0].merged_start == 3.0 + assert receipt.spans[0].source_start == pytest.approx(4.0, abs=1e-3) + assert receipt.spans[1].source_start == pytest.approx(16.0, abs=1e-3) + assert all(s.duration_source == "probed" for s in receipt.spans) + + one_frame = 1.0 / FIXTURE_FPS + assert receipt.spans[0].output_start == 0.0 + assert receipt.spans[1].output_start == receipt.spans[0].output_end + real = probe_duration(out) + assert receipt.spans[-1].output_end == pytest.approx(real, abs=one_frame) + assert receipt.output_duration == pytest.approx(real, abs=1e-3) + + # The receipt survives a JSON round trip unchanged. + path = receipt.write_json(tmp_path / "receipt.json") + assert AssemblyReceipt.from_dict(json.loads(path.read_text())) == receipt diff --git a/tests/test_receipt.py b/tests/test_receipt.py new file mode 100644 index 0000000..1faab64 --- /dev/null +++ b/tests/test_receipt.py @@ -0,0 +1,157 @@ +"""The assembly receipt: what was actually cut, from mocked ffmpeg.""" + +from __future__ import annotations + +import json + +import pytest + +from conftest import fake_ffmpeg +from keycut import ( + AssemblyReceipt, + Segment, + extract_and_concat_ranges, + extract_and_concat_ranges_with_receipt, + extract_and_concat_segments_with_receipt, +) + + +@pytest.fixture +def master(tmp_path): + path = tmp_path / "master.mp4" + path.write_bytes(b"not really a video") + return path + + +def _run(master, tmp_path, ranges, durations=None, **kwargs): + with fake_ffmpeg(durations=durations) as fake: + path, receipt = extract_and_concat_ranges_with_receipt( + master, ranges, tmp_path / "out.mp4", tmp_path / "work", **kwargs + ) + return path, receipt, fake + + +class TestReceiptContent: + def test_header_and_master(self, master, tmp_path): + path, r, _ = _run(master, tmp_path, [(100.0, 160.0)], durations=[60.0, 60.0]) + d = r.to_dict() + assert path == tmp_path / "out.mp4" + assert d["schema"] == "keycut-assembly-receipt/1" + assert d["provenance"] == "keycut" + assert d["time_base"] == "seconds" + assert d["master"] == {"path": str(master), "size_bytes": master.stat().st_size} + assert d["requested"] == [[100.0, 160.0]] + assert d["output"] == {"path": str(tmp_path / "out.mp4"), "duration": 60.0} + + def test_merge_reports_requested_indices(self, master, tmp_path): + _, r, _ = _run( + master, + tmp_path, + [(100.0, 160.0), (160.0, 200.0), (1250.0, 1290.0)], + durations=[100.0, 40.0, 140.0], + ) + assert [s.requested_indices for s in r.spans] == [(0, 1), (2,)] + assert (r.spans[0].merged_start, r.spans[0].merged_end) == (100.0, 200.0) + + def test_out_of_order_ranges_keep_their_own_spans(self, master, tmp_path): + _, r, _ = _run(master, tmp_path, [(500.0, 520.0), (100.0, 120.0)]) + assert [s.requested_indices for s in r.spans] == [(0,), (1,)] + + def test_copy_source_start_is_the_snapped_keyframe(self, master, tmp_path): + _, r, _ = _run(master, tmp_path, [(99.5, 160.0)]) + span = r.spans[0] + assert span.method == "copy" and span.reason == "stream_copy" + assert span.merged_start == 99.5 + assert span.source_start == pytest.approx(100.1, abs=1e-4) + assert span.source_end == 160.0 + + def test_reencode_keyframes_unknown(self, master, tmp_path): + _, r, _ = _run(master, tmp_path, [(99.5, 160.0)], keyframes=[]) + span = r.spans[0] + assert (span.method, span.reason) == ("reencode", "keyframes_unknown") + assert span.source_start == 99.5 + + def test_reencode_snap_back_over_excluded(self, master, tmp_path): + # No keyframe inside 100-160, so a copy would snap back to 0.0; second 50 + # is banned and lies in that region. + mask = [False] * 200 + mask[50] = True + _, r, _ = _run( + master, tmp_path, [(100.0, 160.0)], exclude_mask=mask, keyframes=[0.0, 200.0] + ) + span = r.spans[0] + assert (span.method, span.reason) == ("reencode", "snap_back_over_excluded") + assert span.source_start == 100.0 + + def test_output_spans_are_contiguous_from_zero_using_probed_durations( + self, master, tmp_path + ): + _, r, _ = _run( + master, + tmp_path, + [(100.0, 160.0), (1250.0, 1290.0)], + durations=[59.9, 40.1, 100.0], + ) + a, b = r.spans + assert (a.output_start, a.output_end) == (0.0, 59.9) + assert b.output_start == a.output_end + assert b.output_end == pytest.approx(100.0) + assert a.duration_source == b.duration_source == "probed" + assert r.output_duration == 100.0 + + def test_probe_failure_falls_back_to_planned(self, master, tmp_path): + _, r, _ = _run(master, tmp_path, [(99.5, 160.0), (1250.0, 1290.0)]) + a, b = r.spans + assert a.duration_source == b.duration_source == "planned" + assert a.output_end - a.output_start == pytest.approx(160.0 - a.source_start) + assert (a.output_start, b.output_start) == (0.0, a.output_end) + assert r.output_duration is None + + def test_max_gap_is_recorded(self, master, tmp_path): + _, r, _ = _run(master, tmp_path, [(0.0, 10.0)], max_gap_sec=1.5) + assert r.max_gap_sec == 1.5 + + def test_round_trip_and_write_json(self, master, tmp_path): + _, r, _ = _run(master, tmp_path, [(100.0, 160.0)], durations=[60.0, 60.0]) + assert AssemblyReceipt.from_dict(r.to_dict()) == r + out = r.write_json(tmp_path / "sub" / "receipt.json") + assert AssemblyReceipt.from_dict(json.loads(out.read_text())) == r + + +class TestFailureAndCompat: + def test_failure_gives_none_pair(self, master, tmp_path): + with fake_ffmpeg(returncode=1): + assert extract_and_concat_ranges_with_receipt( + master, [(1.0, 2.0)], tmp_path / "o.mp4", tmp_path / "w" + ) == (None, None) + + def test_missing_master_gives_none_pair(self, tmp_path): + assert extract_and_concat_ranges_with_receipt( + tmp_path / "gone.mp4", [(1.0, 2.0)], tmp_path / "o.mp4", tmp_path / "w" + ) == (None, None) + + def test_plain_function_still_returns_only_the_path(self, master, tmp_path): + with fake_ffmpeg(): + result = extract_and_concat_ranges( + master, [(1.0, 2.0)], tmp_path / "o.mp4", tmp_path / "w" + ) + assert result == tmp_path / "o.mp4" + + +class TestSegments: + def test_segments_with_receipt(self, master, tmp_path): + with fake_ffmpeg(): + path, r = extract_and_concat_segments_with_receipt( + master, + [Segment(0, 5.0, 10.0)], + [(0.0, 100.0), (100.0, 200.0)], + tmp_path / "o.mp4", + tmp_path / "w", + ) + assert path == tmp_path / "o.mp4" + assert r.requested == ((5.0, 10.0),) + + def test_bad_segment_gives_none_pair(self, master, tmp_path): + assert extract_and_concat_segments_with_receipt( + master, [Segment(9, 0.0, 1.0)], [(0.0, 100.0)], tmp_path / "o.mp4", tmp_path / "w" + ) == (None, None)