diff --git a/backend/services/clip_generator.py b/backend/services/clip_generator.py index 1f7c613..40d7f13 100644 --- a/backend/services/clip_generator.py +++ b/backend/services/clip_generator.py @@ -16,6 +16,7 @@ from utils.proc import run as proc_run, ProcError from utils.text import safe_filename +from utils.log import timed from config.paths import paths sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -816,10 +817,11 @@ def generate_clip( progress_callback(10, msg) segment_path = os.path.join(work_dir, "segment.mp4") - if keep_segments and len(keep_segments) > 1: - cut_multi_segment(video_path, segment_path, keep_segments) - else: - cut_segment(video_path, segment_path, start_second, end_second) + with timed("render", "cut", segments=len(keep_segments) if keep_segments else 1): + if keep_segments and len(keep_segments) > 1: + cut_multi_segment(video_path, segment_path, keep_segments) + else: + cut_segment(video_path, segment_path, start_second, end_second) # Remap transcript words for multi-segment clips. # Needed before crop (speaker detection) and captions. @@ -861,18 +863,19 @@ def generate_clip( progress_callback(30, f"Resizing for {spec.name} format (2/{total_steps})") cropped_path = os.path.join(work_dir, "cropped.mp4") - if spec.reframe: - crop_to_vertical( - segment_path, cropped_path, - strategy=crop_strategy, - transcript_words=crop_words, - clip_start=crop_clip_start, - face_map=face_map, - crop_keyframes=crop_keyframes, - target_dims=spec.dims, - ) - else: - fit_to_frame(segment_path, cropped_path, target_dims=spec.dims) + with timed("render", "crop", strategy=crop_strategy if spec.reframe else "fit"): + if spec.reframe: + crop_to_vertical( + segment_path, cropped_path, + strategy=crop_strategy, + transcript_words=crop_words, + clip_start=crop_clip_start, + face_map=face_map, + crop_keyframes=crop_keyframes, + target_dims=spec.dims, + ) + else: + fit_to_frame(segment_path, cropped_path, target_dims=spec.dims) # Step 3: Render captions (Remotion-first; ASS fallback optional) if transcript_words: diff --git a/backend/services/local_reframe.py b/backend/services/local_reframe.py index 9a66728..ee04caa 100644 --- a/backend/services/local_reframe.py +++ b/backend/services/local_reframe.py @@ -29,7 +29,12 @@ def count_scene_cuts(video_path: str, threshold: float = 0.35) -> int: [ "ffmpeg", "-i", str(video_path), - "-filter:v", f"select='gt(scene,{threshold})',showinfo", + # Audio and subtitles are irrelevant to scene detection, and the + # scene score is a whole-frame statistic — computing it on a + # 320px-wide copy gives the same cuts for a fraction of the + # decode. Downscale before select so the filter sees small frames. + "-an", "-sn", + "-filter:v", f"scale=320:-2,select='gt(scene,{threshold})',showinfo", "-f", "null", "-", ], timeout=180, diff --git a/backend/services/video_processor.py b/backend/services/video_processor.py index 7256662..65be64b 100644 --- a/backend/services/video_processor.py +++ b/backend/services/video_processor.py @@ -9,11 +9,13 @@ import subprocess import json import math +import time +from concurrent.futures import ThreadPoolExecutor from typing import Optional from services.encoder import get_video_encode_flags from utils.proc import run as proc_run, ProcError -from utils.log import log_event +from utils.log import log_event, timed from services import media_probe from services.media_probe import ( CPU_FLAGS, @@ -980,6 +982,118 @@ def _face_sample_indices(total_frames: int, fps: float) -> list[int]: return indices +# Memory ceiling for one batch of decoded frames awaiting detection. Sized so +# a 1080p clip batches ~40 frames and a 4K clip ~10 — enough to keep the pool +# busy without holding a whole clip of decoded frames in RAM. +_FACE_BATCH_BYTES = 256 * 1024 * 1024 + + +# Each extra worker needs its own cv2.FaceDetectorYN, which costs ~21ms to +# construct against ~4.5ms per detection. Below this many frames per worker the +# construction never pays for itself and the pool makes short clips *slower*, +# so the pool is sized by workload rather than by core count alone. +_FACE_FRAMES_PER_WORKER = 32 + + +def _face_detect_workers(sample_count: int = 0) -> int: + """Thread count for YuNet inference during face sampling. + + PODCLI_FACE_WORKERS overrides and is taken literally; 1 disables the pool + entirely, which is the first thing to try when bisecting a framing + regression. Otherwise the count is capped both by cores and by how many + frames there are to share out — a 2s clip stays serial. + """ + raw = os.environ.get("PODCLI_FACE_WORKERS", "").strip() + if raw: + try: + return max(1, int(raw)) + except ValueError: + pass + by_cores = min(12, os.cpu_count() or 4) + by_work = sample_count // _FACE_FRAMES_PER_WORKER + return max(1, min(by_cores, by_work)) + + +def _detect_batch(batch: list, detectors: list, width: int, height: int, executor) -> list: + """Detect faces across a batch of (time, frame) pairs, input order preserved. + + Each worker owns one detector and walks its own stride of the batch, so no + cv2.FaceDetectorYN instance is ever touched by two threads — they carry + per-instance input-size state and are not thread-safe. Striding (rather + than a shared work queue) also makes the assignment deterministic. + """ + # Imported here, like every other cv2 dependency in this module, so the + # module stays importable without OpenCV installed. + from services.face_detector import detect_faces + + if executor is None or len(detectors) < 2 or len(batch) < 2: + det = detectors[0] + return [(t, detect_faces(det, frame, width, height)) for t, frame in batch] + + n = len(detectors) + + def _stride(k: int) -> list: + det = detectors[k] + return [ + (i, (batch[i][0], detect_faces(det, batch[i][1], width, height))) + for i in range(k, len(batch), n) + ] + + merged: dict = {} + for part in executor.map(_stride, range(n)): + merged.update(part) + return [merged[i] for i in range(len(batch))] + + +def _dump_crop_path( + *, + input_path: str, + keyframes_x: list, + crop_w: int, + crop_h: int, + crop_y: int, + width: int, + height: int, + detections: list, + segment_tracks: list, + has_any_split: bool, +) -> None: + """Write the computed camera path to PODCLI_CROP_DUMP/.json. + + No-op unless the env var is set, and never raises: this is diagnostics and + must not be able to fail a render. Frame-level detections are summarised + rather than dumped whole — the camera path is what a regression would move. + """ + dump_dir = os.environ.get("PODCLI_CROP_DUMP") + if not dump_dir: + return + try: + os.makedirs(dump_dir, exist_ok=True) + stem = os.path.splitext(os.path.basename(input_path))[0] + payload = { + "source": os.path.basename(input_path), + "source_dims": [width, height], + "crop": {"w": crop_w, "h": crop_h, "y": crop_y}, + "has_any_split": bool(has_any_split), + "detection_frames": len(detections), + "frames_with_faces": sum(1 for _, faces in detections if faces), + "segment_tracks": [ + # (start, end, speaker, track_id, ...) — keep the framing- + # relevant fields, drop anything unhashable/verbose. + {"start": round(s[0], 3), "end": round(s[1], 3), + "speaker": s[2], "track_id": s[3]} + for s in segment_tracks + ], + "keyframes_x": [[t, x] for t, x in keyframes_x], + } + out = os.path.join(dump_dir, f"{stem}.crop.json") + with open(out, "w", encoding="utf-8") as f: + json.dump(payload, f, indent=2, sort_keys=True) + log_event("crop", "dumped-path", file=out, keyframes=len(keyframes_x)) + except Exception as exc: # diagnostics must never break a render + log_event("crop", "dump-failed", level="warn", error=str(exc)) + + def _track_and_crop( input_path: str, output_path: str, @@ -1032,21 +1146,78 @@ def _track_and_crop( sample_indices = _face_sample_indices(total_frames, fps) detections = [] # [(time, faces), ...] + # This loop is inference-bound, not decode-bound: on a 60s 1080p clip YuNet + # is ~83% of the time against ~17% for decode. So frames are decoded + # serially (cheap, and it keeps frame indices exact) and detected on a small + # pool. Detections are independent per frame and OpenCV releases the GIL + # inside YuNet, so results are bit-identical; measured 3281ms -> 1179ms for + # the stage, with detection itself going 2718ms -> 594ms. + # + # Batches are bounded by bytes rather than frame count so a 4K source does + # not hold an unbounded number of decoded frames in memory at once. + frame_bytes = max(1, width * height * 3) + batch_limit = max(1, int(_FACE_BATCH_BYTES / frame_bytes)) + + decode_ns = 0 + detect_ns = 0 next_pos = 0 frame_idx = -1 - while next_pos < len(sample_indices): - if not cap.grab(): - break - frame_idx += 1 - if frame_idx < sample_indices[next_pos]: - continue - next_pos += 1 - ret, frame = cap.retrieve() - if not ret: - continue - t = frame_idx / fps - faces = detect_faces(detector, frame, width, height) - detections.append((t, faces)) + batch: list = [] + + # Detector construction is inside the timed block on purpose: at ~21ms per + # instance it is a real cost, and timing only the loop would hide it. + with timed("crop", "face_sampling", frames=total_frames) as t_fields: + workers = _face_detect_workers(len(sample_indices)) + detectors = [detector] + if workers > 1: + detectors += [ + d for d in (create_detector(width, height) for _ in range(workers - 1)) + if d is not None + ] + + executor = None + try: + if len(detectors) > 1: + executor = ThreadPoolExecutor(max_workers=len(detectors)) + + def _flush() -> None: + nonlocal detect_ns, batch + if not batch: + return + _t = time.perf_counter_ns() + detections.extend( + _detect_batch(batch, detectors, width, height, executor) + ) + detect_ns += time.perf_counter_ns() - _t + batch = [] + + while next_pos < len(sample_indices): + _t0 = time.perf_counter_ns() + grabbed = cap.grab() + decode_ns += time.perf_counter_ns() - _t0 + if not grabbed: + break + frame_idx += 1 + if frame_idx < sample_indices[next_pos]: + continue + next_pos += 1 + _t0 = time.perf_counter_ns() + ret, frame = cap.retrieve() + decode_ns += time.perf_counter_ns() - _t0 + if not ret: + continue + batch.append((frame_idx / fps, frame)) + if len(batch) >= batch_limit: + _flush() + _flush() + finally: + if executor is not None: + executor.shutdown(wait=True) + + t_fields["samples"] = len(detections) + t_fields["workers"] = len(detectors) + t_fields["decode_ms"] = decode_ns // 1_000_000 + t_fields["detect_ms"] = detect_ns // 1_000_000 cap.release() @@ -1515,6 +1686,22 @@ def _nearest_face_cx(t_target: float, window: float = 1.5) -> float | None: validated.append((kf_t, kf_x)) keyframes_x = validated + # ── Optional crop-path dump (regression harness) ───────────── + # Crop decisions are the least testable part of the pipeline: unit tests + # cover the helper math, and the e2e render uses a synthetic video with no + # faces in it. With PODCLI_CROP_DUMP set to a directory, every crop writes + # its computed camera path as JSON, so a change that was meant to be purely + # a speed optimization can be proven not to have moved the camera. + _dump_crop_path( + input_path=input_path, + keyframes_x=keyframes_x, + crop_w=crop_w, crop_h=crop_h, crop_y=crop_y, + width=width, height=height, + detections=detections, + segment_tracks=segment_tracks, + has_any_split=has_any_split, + ) + # ── Build FFmpeg filter ────────────────────────────────────── if not keyframes_x: crop_x = max(0, (width - crop_w) // 2) diff --git a/backend/utils/log.py b/backend/utils/log.py index 7635f25..56111c6 100644 --- a/backend/utils/log.py +++ b/backend/utils/log.py @@ -9,6 +9,8 @@ import os import sys +import time +from contextlib import contextmanager _VERBOSE = os.environ.get("PODCLI_LOG_VERBOSE", "").lower() in ("1", "true", "yes") @@ -40,3 +42,24 @@ def warn(category: str, message: str, **fields) -> None: def debug(category: str, message: str, **fields) -> None: log_event(category, message, level="debug", **fields) + + +@contextmanager +def timed(category: str, stage: str, **fields): + """Time a block and emit `[category] timing stage=... ms=...` on exit. + + Debug level, so stage timings stay silent unless PODCLI_LOG_VERBOSE is set. + Emits on failure too — a stage that blows up after 40s is exactly the one + worth seeing. Yields a dict the caller can add fields to before the line is + written, for counts that are only known once the block has run. + """ + extra: dict = {} + start = time.perf_counter() + try: + yield extra + finally: + elapsed_ms = int((time.perf_counter() - start) * 1000) + log_event( + category, "timing", level="debug", + stage=stage, ms=elapsed_ms, **{**fields, **extra}, + ) diff --git a/docs/configuration.md b/docs/configuration.md index ba10292..f72a629 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -18,6 +18,8 @@ Copy `.env.example` to `.env`, or export these in your shell. `setup.sh` copies | `PODCLI_BACKEND` | resolved | Override the Python backend directory | | `PODCLI_PYTHON` | resolved | Override the Python interpreter. The launcher exports the resolved path as `PYTHON_PATH` for internal use | | `PODCLI_TRANSITION_AUTOFIX_PASSES` | auto | Transition QA/autofix passes. Runs only on reframes that can produce hard cuts. Set a number to force it, `0` disables; the renderer caps it at 2 | +| `PODCLI_FACE_WORKERS` | auto | Threads used for face detection during reframing. Auto sizes by core count (max 12) and by clip length, so short clips stay serial rather than paying to build detectors they cannot amortize. Set `1` to force serial — the first thing to try when bisecting a framing regression, since it changes speed only, never the result | +| `PODCLI_CROP_DUMP` | unset | Directory to write each clip's computed camera path to as `.crop.json`. Diagnostics only; use it to diff framing decisions before and after a change | | `FFMPEG_PATH` / `FFPROBE_PATH` | `ffmpeg` / `ffprobe` | Override the FFmpeg binaries | Installed builds provision their own Python, Node, FFmpeg, and whisper.cpp, so the diff --git a/tests/test_crop_path_golden.py b/tests/test_crop_path_golden.py new file mode 100644 index 0000000..7472f2b --- /dev/null +++ b/tests/test_crop_path_golden.py @@ -0,0 +1,262 @@ +"""Golden tests for the crop camera path and the crop-path dump. + +The helper tests next door check each piece of the tracking math in isolation. +What they cannot catch is a change that leaves every helper correct but moves +the camera anyway — a different sampling rate, a reordered pipeline, a changed +default. Those are exactly the changes a "pure speed optimization" makes. + +So these lock the composed output: fixed detections in, exact keyframes out. +No cv2, ffmpeg, or video files — the fixtures stand in for decoded frames. +""" + +import json +import os +import sys +import tempfile +import unittest +from concurrent.futures import ThreadPoolExecutor +from unittest import mock + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +BACKEND_ROOT = os.path.join(ROOT, "backend") +if BACKEND_ROOT not in sys.path: + sys.path.insert(0, BACKEND_ROOT) + +from services import video_processor as vp + +WIDTH, HEIGHT = 1920, 1080 +CROP_W = int(HEIGHT * (1080 / 1920)) # 607 — full-height 9:16 window + + +def _face(cx, fw=180): + return {"cx": float(cx), "cy": 400.0, "fw": fw, "fh": fw, "confidence": 0.9} + + +def _two_speaker_detections(n=40, step=0.1): + """Split-screen: a stable face at x=520 and another at x=1400.""" + return [(round(i * step, 3), [_face(520), _face(1400)]) for i in range(n)] + + +class FaceSampleIndicesGoldenTests(unittest.TestCase): + """The sampling schedule is the input to everything downstream. If a speed + change alters it, every keyframe below shifts — so pin the exact schedule.""" + + def test_30fps_schedule_is_exact(self): + idx = vp._face_sample_indices(90, 30.0) + # Every frame to 0.5s (0-14), every other to 1.0s (15-29), then ~10fps. + self.assertEqual(idx[:15], list(range(15))) + self.assertEqual(idx[15:19], [15, 16, 17, 18]) + steady = [b - a for a, b in zip(idx, idx[1:]) if a >= 30] + self.assertEqual(set(steady), {3}) + self.assertEqual(idx[-1], 87) # last sample before the 90-frame end + + def test_sampling_rate_is_ten_fps_in_steady_state(self): + for fps in (24.0, 25.0, 30.0, 50.0, 60.0): + idx = vp._face_sample_indices(int(fps * 10), fps) + steady = [b - a for a, b in zip(idx, idx[1:]) if a >= fps] + self.assertTrue(steady, f"no steady-state samples at {fps}fps") + self.assertEqual( + set(steady), {max(1, int(fps / 10))}, + f"steady-state step drifted off 10fps at {fps}fps", + ) + + +class AssignFaceTracksGoldenTests(unittest.TestCase): + def test_split_screen_yields_two_stable_tracks(self): + tracked = vp._assign_face_tracks(_two_speaker_detections(), WIDTH) + self.assertEqual(len(tracked), 40) + ids_per_frame = [{f["track_id"] for f in faces} for _, faces in tracked] + # Two identities, and they stay the same for the whole clip. + self.assertTrue(all(len(s) == 2 for s in ids_per_frame)) + self.assertEqual(len(set().union(*ids_per_frame)), 2) + + def test_track_ids_follow_position_not_list_order(self): + # Same two faces, but the detector returns them in flipped order + # halfway through. Identity must stay pinned to position. + dets = [] + for i in range(20): + faces = [_face(520), _face(1400)] + dets.append((round(i * 0.1, 3), faces if i < 10 else faces[::-1])) + tracked = vp._assign_face_tracks(dets, WIDTH) + left_ids = { + min(faces, key=lambda f: f["cx"])["track_id"] for _, faces in tracked + } + self.assertEqual(len(left_ids), 1, "left face changed identity mid-clip") + + +class TripodCameraGoldenTests(unittest.TestCase): + """The camera is what the viewer actually sees. Pin its behaviour.""" + + def test_force_snap_centres_exactly(self): + cam = vp._update_tripod_camera( + current_center_x=100.0, target_center_x=960.0, + crop_w=CROP_W, video_width=WIDTH, dt=0.0, force_snap=True, + ) + self.assertEqual(cam, 960.0) + + def test_camera_holds_still_for_small_drift(self): + start = 960.0 + cam = start + # A face jittering by a few px must not move the camera at all. + for target in (964.0, 957.0, 962.0, 959.0): + cam = vp._update_tripod_camera( + current_center_x=cam, target_center_x=target, + crop_w=CROP_W, video_width=WIDTH, dt=0.1, + ) + self.assertEqual(cam, start, "tripod drifted on sub-threshold jitter") + + def test_camera_never_leaves_frame(self): + for target in (-500.0, 0.0, 99999.0): + cam = vp._update_tripod_camera( + current_center_x=960.0, target_center_x=target, + crop_w=CROP_W, video_width=WIDTH, dt=1.0, force_snap=True, + ) + self.assertGreaterEqual(cam, CROP_W / 2) + self.assertLessEqual(cam, WIDTH - CROP_W / 2) + + +class FaceDetectWorkersTests(unittest.TestCase): + """Pool size is a correctness-adjacent concern: each worker needs its own + ~21ms detector, so over-sizing the pool on a short clip is slower than + staying serial. Measured before this cap: a 2s clip went 216ms -> 313ms.""" + + def test_env_override_wins_and_is_taken_literally(self): + with mock.patch.dict(os.environ, {"PODCLI_FACE_WORKERS": "3"}): + self.assertEqual(vp._face_detect_workers(10_000), 3) + # Even when the workload would not justify it. + self.assertEqual(vp._face_detect_workers(1), 3) + + def test_one_worker_is_allowed_to_disable_the_pool(self): + with mock.patch.dict(os.environ, {"PODCLI_FACE_WORKERS": "1"}): + self.assertEqual(vp._face_detect_workers(10_000), 1) + + def test_garbage_and_zero_fall_back_to_a_sane_count(self): + for bad in ("garbage", "0", "-4"): + with mock.patch.dict(os.environ, {"PODCLI_FACE_WORKERS": bad}): + self.assertGreaterEqual(vp._face_detect_workers(10_000), 1) + + def test_short_clips_stay_serial(self): + # A ~2s clip samples ~40 frames — not enough to repay a second detector. + with mock.patch.dict(os.environ, {"PODCLI_FACE_WORKERS": ""}): + self.assertEqual(vp._face_detect_workers(0), 1) + self.assertEqual(vp._face_detect_workers(40), 1) + + def test_pool_grows_with_the_workload(self): + with mock.patch.object(os, "cpu_count", return_value=128): + with mock.patch.dict(os.environ, {"PODCLI_FACE_WORKERS": ""}): + self.assertEqual(vp._face_detect_workers(70), 2) + self.assertEqual(vp._face_detect_workers(170), 5) + # Capped at 12 no matter how long the clip or how many cores. + self.assertEqual(vp._face_detect_workers(100_000), 12) + + def test_never_exceeds_core_count(self): + with mock.patch.object(os, "cpu_count", return_value=2): + with mock.patch.dict(os.environ, {"PODCLI_FACE_WORKERS": ""}): + self.assertEqual(vp._face_detect_workers(100_000), 2) + + +class DetectBatchTests(unittest.TestCase): + """Parallel inference must be indistinguishable from serial. Verified on a + real 620-frame 1080p clip (5.5x faster, identical output); these pin the + ordering and detector-isolation properties that make that true.""" + + @staticmethod + def _fake_detect(det, frame, w, h): + # Encodes which detector handled the frame, so sharing is detectable. + return [{"cx": float(frame), "detector": det}] + + def _run(self, n_frames, n_detectors, use_executor=True): + batch = [(i * 0.1, i) for i in range(n_frames)] + detectors = [f"det{k}" for k in range(n_detectors)] + executor = ( + ThreadPoolExecutor(max_workers=n_detectors) + if use_executor and n_detectors > 1 + else None + ) + try: + with mock.patch( + "services.face_detector.detect_faces", side_effect=self._fake_detect + ): + return vp._detect_batch(batch, detectors, 1920, 1080, executor) + finally: + if executor: + executor.shutdown(wait=True) + + def test_parallel_matches_serial_exactly(self): + serial = self._run(50, 1, use_executor=False) + parallel = self._run(50, 4) + self.assertEqual( + [(t, f[0]["cx"]) for t, f in parallel], + [(t, f[0]["cx"]) for t, f in serial], + ) + + def test_input_order_is_preserved(self): + out = self._run(37, 5) + self.assertEqual([t for t, _ in out], [i * 0.1 for i in range(37)]) + self.assertEqual([f[0]["cx"] for _, f in out], [float(i) for i in range(37)]) + + def test_each_detector_is_used_by_exactly_one_stride(self): + out = self._run(20, 4) + # Frame i must be handled by detector i % 4 — deterministic striding, + # so no detector is ever touched by two threads. + for i, (_, faces) in enumerate(out): + self.assertEqual(faces[0]["detector"], f"det{i % 4}") + + def test_falls_back_to_serial_without_executor(self): + out = self._run(10, 4, use_executor=False) + self.assertTrue(all(f[0]["detector"] == "det0" for _, f in out)) + + def test_handles_empty_and_single_frame_batches(self): + self.assertEqual(self._run(0, 4), []) + self.assertEqual(len(self._run(1, 4)), 1) + + +class DumpCropPathTests(unittest.TestCase): + """The dump is the harness used to prove a speed change moved nothing. + If it is silently a no-op or throws, that proof is worthless.""" + + KWARGS = dict( + input_path="/tmp/clip_001.mp4", + keyframes_x=[(0.0, 300), (1.5, 900)], + crop_w=CROP_W, crop_h=HEIGHT, crop_y=0, + width=WIDTH, height=HEIGHT, + detections=_two_speaker_detections(n=3), + segment_tracks=[(0.0, 1.5, "SPEAKER_00", 1, None)], + has_any_split=True, + ) + + def test_writes_expected_payload(self): + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.dict(os.environ, {"PODCLI_CROP_DUMP": tmp}): + vp._dump_crop_path(**self.KWARGS) + out = os.path.join(tmp, "clip_001.crop.json") + self.assertTrue(os.path.exists(out), "no dump written") + with open(out, encoding="utf-8") as f: + payload = json.load(f) + + self.assertEqual(payload["keyframes_x"], [[0.0, 300], [1.5, 900]]) + self.assertEqual(payload["crop"], {"w": CROP_W, "h": HEIGHT, "y": 0}) + self.assertEqual(payload["source_dims"], [WIDTH, HEIGHT]) + self.assertEqual(payload["detection_frames"], 3) + self.assertEqual(payload["frames_with_faces"], 3) + self.assertEqual( + payload["segment_tracks"], + [{"start": 0.0, "end": 1.5, "speaker": "SPEAKER_00", "track_id": 1}], + ) + + def test_noop_without_env_var(self): + with tempfile.TemporaryDirectory() as tmp: + env = {k: v for k, v in os.environ.items() if k != "PODCLI_CROP_DUMP"} + with mock.patch.dict(os.environ, env, clear=True): + vp._dump_crop_path(**self.KWARGS) + self.assertEqual(os.listdir(tmp), []) + + def test_never_raises_on_bad_destination(self): + # An unwritable dump dir must not take a render down with it. + bad = "/dev/null/nope" + with mock.patch.dict(os.environ, {"PODCLI_CROP_DUMP": bad}): + vp._dump_crop_path(**self.KWARGS) # must not raise + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_local_reframe.py b/tests/test_local_reframe.py new file mode 100644 index 0000000..fab73d8 --- /dev/null +++ b/tests/test_local_reframe.py @@ -0,0 +1,95 @@ +"""Tests for local_reframe scene-cut detection. + +count_scene_cuts runs a full decode of every split-screen clip, so it is worth +keeping cheap. It downscales to 320px and drops audio before the scene filter: +the scene score is a whole-frame statistic, so a small copy yields the same +cuts. Measured on a 1080p source, that is ~2x faster with scene scores within +0.005 of the full-resolution values. + +These tests pin the flags that make it cheap, so the optimization cannot be +quietly dropped, and cover the parsing/failure contract callers rely on. +""" + +import os +import sys +import unittest +from unittest import mock + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +BACKEND_ROOT = os.path.join(ROOT, "backend") +if BACKEND_ROOT not in sys.path: + sys.path.insert(0, BACKEND_ROOT) + +from services import local_reframe +from utils.proc import ProcError + + +def _result(returncode=0, stderr=""): + return mock.Mock(returncode=returncode, stdout="", stderr=stderr) + + +# Two cuts, in the showinfo format the parser scrapes. +_TWO_CUTS = ( + "[Parsed_showinfo_1 @ 0x1] n:0 pts:61440 pts_time:4 pos:1 fmt:yuv420p\n" + "[Parsed_showinfo_1 @ 0x1] n:1 pts:122880 pts_time:8 pos:2 fmt:yuv420p\n" +) + + +class CountSceneCutsCommandTests(unittest.TestCase): + def _captured_cmd(self, **kwargs): + with mock.patch.object( + local_reframe, "proc_run", return_value=_result(stderr=_TWO_CUTS) + ) as run: + local_reframe.count_scene_cuts("/tmp/clip.mp4", **kwargs) + return run.call_args[0][0] + + def test_downscales_before_scene_filter(self): + cmd = self._captured_cmd() + vf = cmd[cmd.index("-filter:v") + 1] + # Order matters: scale must come first so select sees small frames. + self.assertTrue( + vf.startswith("scale=320:-2,"), + f"scene filter is not running on a downscaled copy: {vf}", + ) + self.assertIn("select=", vf) + self.assertIn("showinfo", vf) + + def test_skips_audio_and_subtitle_decode(self): + cmd = self._captured_cmd() + self.assertIn("-an", cmd) + self.assertIn("-sn", cmd) + + def test_threshold_is_passed_through(self): + vf = self._captured_cmd(threshold=0.5)[ + self._captured_cmd(threshold=0.5).index("-filter:v") + 1 + ] + self.assertIn("gt(scene,0.5)", vf) + + +class CountSceneCutsResultTests(unittest.TestCase): + def test_counts_showinfo_lines(self): + with mock.patch.object( + local_reframe, "proc_run", return_value=_result(stderr=_TWO_CUTS) + ): + self.assertEqual(local_reframe.count_scene_cuts("/tmp/clip.mp4"), 2) + + def test_no_cuts_returns_zero(self): + with mock.patch.object(local_reframe, "proc_run", return_value=_result()): + self.assertEqual(local_reframe.count_scene_cuts("/tmp/clip.mp4"), 0) + + def test_ffmpeg_failure_returns_zero_not_raise(self): + # Callers treat 0 as "no info" and proceed with their default plan; + # a scene-detect failure must never take a render down. + with mock.patch.object( + local_reframe, "proc_run", return_value=_result(returncode=1) + ): + self.assertEqual(local_reframe.count_scene_cuts("/tmp/clip.mp4"), 0) + + def test_proc_error_returns_zero_not_raise(self): + err = ProcError(["ffmpeg"], returncode=-9, stderr="timed out", duration=180.0) + with mock.patch.object(local_reframe, "proc_run", side_effect=err): + self.assertEqual(local_reframe.count_scene_cuts("/tmp/clip.mp4"), 0) + + +if __name__ == "__main__": + unittest.main()