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)