Files
human-experience-filter-art/tools/pipeline/run.py
T
2026-06-24 08:09:01 -07:00

68 lines
2.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Thin runner: probe the source duration, then execute the frame -> loop ->
transcode stages with tools.pipeline.ffmpeg_ops. The pure arg builders are
unit-tested; this glue is covered by the opt-in integration test."""
from __future__ import annotations
import shutil
import subprocess
from pathlib import Path
from .ffmpeg_ops import crossfade_loop_args, frame_args, transcode_args
MASTER_MAX_W = 3840
def resolve_ffmpeg() -> str:
"""ffmpeg from PATH, else the bundled imageio-ffmpeg binary (matches
simulator/setup_scales_media.py). The Pi/dev box may lack a system ffmpeg."""
if shutil.which("ffmpeg"):
return "ffmpeg"
import imageio_ffmpeg
return imageio_ffmpeg.get_ffmpeg_exe()
def probe_duration(src) -> float:
"""Clip duration in seconds via cv2 (frame_count / fps) — avoids a system
ffprobe dependency (imageio-ffmpeg ships ffmpeg only; the Pi has no ffprobe)."""
import cv2
cap = cv2.VideoCapture(str(src))
fps = cap.get(cv2.CAP_PROP_FPS) or 0.0
frames = cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0.0
cap.release()
if fps <= 0:
raise ValueError(f"could not read fps from {src}")
return frames / fps
def _run(args: list[str]) -> None:
subprocess.run(args, check=True, capture_output=True)
def process_clip(src, out_dir, *, start: float = 0.0, duration: float | None = None,
overlap: float = 1.0, master_w: int = MASTER_MAX_W, master_h: int = 2160,
ff: str | None = None) -> dict:
"""Run stages 24 for one clip; return {'master','proxy','loop','mezzanine'} paths.
`duration` defaults to the full source length (minus the trim start).
Note: `crossfade_loop_args` uses a fixed fps only to satisfy xfade's
constant-frame-rate requirement; the final output fps is set by `transcode_args`
(both default 30 — keep in sync if overridden)."""
ff = ff or resolve_ffmpeg()
src = Path(src)
out_dir = Path(out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
if duration is None:
duration = probe_duration(src) - start
mezz = out_dir / "mezzanine.mp4"
loop = out_dir / "loop.mp4"
master = out_dir / "master.mp4"
proxy = out_dir / "base.mp4" # the proxy is what the manifest's base_file points at
_run(frame_args(src, mezz, start=start, duration=duration,
width=master_w, height=master_h, ff=ff))
_run(crossfade_loop_args(mezz, loop, duration=duration, overlap=overlap, ff=ff))
_run(transcode_args(loop, master, width=master_w, height=master_h, crf=18, ff=ff))
_run(transcode_args(loop, proxy, width=1920, height=1080, crf=20, ff=ff))
return {"mezzanine": mezz, "loop": loop, "master": master, "proxy": proxy}