feat: Add script for trimming still sections off of a video

ffmpeg_remove_still.py

@@ -0,0 +1,192 @@
+#!/usr/bin/env python3
+# /// script
+# dependencies = []
+# ///
+
+import argparse
+from bisect import bisect_left, bisect_right
+import json
+import logging
+import math
+import re
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Shorten still sections using FFmpeg stream copying, without re-encoding.",
+        epilog="Cuts follow video keyframes and audio packets. Still sections may remain longer than requested. "
+               "Audio during removed time is also removed, including any narration.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    parser.add_argument("input", type=Path, help="Input video")
+    parser.add_argument("--output", type=Path, help="Output video (default: INPUT.short with the original extension)")
+    parser.add_argument("--min-still", type=float, default=2.0,
+                        help="Minimum still section duration to detect, in seconds")
+    parser.add_argument("--keep-still", type=float, default=0.5,
+                        help="Target seconds to retain per still section; keyframe alignment may retain more")
+    parser.add_argument("--noise", type=float, default=0.001,
+                        help="Frame difference tolerance, from 0 to 1; higher values ignore more movement")
+    parser.add_argument("--dry-run", action="store_true", help="Report cuts without writing output")
+    parser.add_argument("--overwrite", action="store_true", help="Replace an existing output file")
+    arguments = parser.parse_args()
+    if not math.isfinite(arguments.min_still) or arguments.min_still <= 0:
+        parser.error("--min-still must be positive and finite")
+    if not math.isfinite(arguments.keep_still) or arguments.keep_still <= 0:
+        parser.error("--keep-still must be positive and finite")
+    if not 0 <= arguments.noise <= 1:
+        parser.error("--noise must be between 0 and 1")
+    return arguments
+
+
+def probe_video(path: Path) -> tuple[float, float]:
+    result = subprocess.run(
+        args=["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)],
+        check=True, capture_output=True, text=True,
+    )
+    metadata = json.loads(result.stdout)
+    video = next((stream for stream in metadata["streams"] if stream["codec_type"] == "video"), None)
+    if video is None:
+        raise ValueError("Input has no video stream")
+    duration = float(video.get("duration") or metadata["format"].get("duration", 0))
+    if not math.isfinite(duration) or duration <= 0:
+        raise ValueError("Cannot determine video duration")
+    start_time = float(video.get("start_time") or metadata["format"].get("start_time", 0))
+    return duration, start_time
+
+
+def probe_keyframes(path: Path, start_time: float) -> list[tuple[float, float]]:
+    result = subprocess.run(
+        args=["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_packets",
+              "-show_entries", "packet=pts_time,dts_time,flags", "-of", "csv=p=0", str(path)],
+        check=True, capture_output=True, text=True,
+    )
+    keyframes = []
+    for line in result.stdout.splitlines():
+        fields = line.split(",")
+        if len(fields) >= 3 and "K" in fields[2] and fields[0] != "N/A":
+            presentation_time = float(fields[0]) - start_time
+            decode_time = float(fields[1]) - start_time if fields[1] != "N/A" else presentation_time
+            keyframes.append((max(0.0, presentation_time), decode_time))
+    if not keyframes:
+        raise ValueError("No video keyframes found; cannot safely trim by stream copying")
+    return sorted(keyframes)
+
+
+def parse_freezes(log: str, duration: float) -> list[tuple[float, float]]:
+    freezes = []
+    start = None
+    for kind, timestamp in re.findall(r"lavfi\.freezedetect\.freeze_(start|end):\s*([-+\d.eE]+)", log):
+        timestamp = min(duration, max(0.0, float(timestamp)))
+        if kind == "start":
+            start = timestamp
+        elif start is not None:
+            if timestamp > start:
+                freezes.append((start, timestamp))
+            start = None
+    if start is not None and duration > start:
+        freezes.append((start, duration))
+    return freezes
+
+
+def detect_freezes(path: Path, duration: float, min_still: float, noise: float) -> list[tuple[float, float]]:
+    logging.info("Detecting still sections...")
+    result = subprocess.run(
+        args=[
+            "ffmpeg", "-hide_banner", "-nostdin", "-nostats", "-i", str(path),
+            "-map", "0:v:0", "-vf", f"setpts=PTS-STARTPTS,freezedetect=noise={noise}:duration={min_still}",
+            "-an", "-f", "null", "-",
+        ],
+        check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True,
+    )
+    return parse_freezes(log=result.stderr, duration=duration)
+
+
+def plan_segments(duration: float, freezes: list[tuple[float, float]], keep_still: float,
+                  keyframes: list[float]) -> list[tuple[float, float]]:
+    segments = []
+    cursor = 0.0
+    for start, end in freezes:
+        if end - start <= keep_still:
+            continue
+        first = bisect_left(keyframes, start + keep_still / 2)
+        last = bisect_right(keyframes, end - keep_still / 2) - 1
+        if first >= len(keyframes) or last < 0 or keyframes[first] >= keyframes[last]:
+            logging.info(f"Still section {start:.2f}–{end:.2f}s: no removable span between keyframes")
+            continue
+        cut_start = keyframes[first]
+        cut_end = keyframes[last]
+        logging.info(f"Still section {start:.2f}–{end:.2f}s: retain {end - start - (cut_end - cut_start):.2f}s")
+        if cut_start > cursor:
+            segments.append((cursor, cut_start))
+        cursor = cut_end
+    if cursor < duration:
+        segments.append((cursor, duration))
+    return segments
+
+
+def copy_video(source: Path, output: Path, segments: list[tuple[float, float]], duration: float,
+               start_time: float, keyframes: list[tuple[float, float]], overwrite: bool) -> None:
+    decode_times = dict(keyframes)
+    with tempfile.TemporaryDirectory() as temporary_directory:
+        playlist = Path(temporary_directory) / "segments.ffconcat"
+        local_source = Path(temporary_directory) / f"source{source.suffix}"
+        local_source.symlink_to(target=source)
+        entries = ["ffconcat version 1.0"]
+        for start, end in segments:
+            entries.extend([f"file '{local_source.name}'", f"inpoint {start + start_time:.6f}"])
+            if end < duration:
+                # The concat demuxer's outpoint uses DTS, while inpoint uses PTS.
+                entries.append(f"outpoint {decode_times[end] + start_time:.6f}")
+            entries.append(f"duration {end - start:.6f}")
+        playlist.write_text(data="\n".join(entries) + "\n")
+        command = [
+            "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "warning", "-stats",
+            "-y" if overwrite else "-n", "-f", "concat", "-safe", "0", "-auto_convert", "0", "-i", str(playlist),
+            "-map", "0:v:0", "-map", "0:a?", "-c", "copy",
+        ]
+        if output.suffix.lower() in {".mp4", ".mov"}:
+            command.extend(["-movflags", "+faststart"])
+        command.append(str(output))
+        subprocess.run(args=command, check=True)
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
+    arguments = parse_args()
+    try:
+        for executable in ("ffmpeg", "ffprobe"):
+            if shutil.which(cmd=executable) is None:
+                raise ValueError(f"{executable} must be installed and available on PATH")
+        source = arguments.input.resolve(strict=True)
+        output = (arguments.output or source.with_name(f"{source.stem}.short{source.suffix or '.mkv'}")).resolve()
+        if source == output:
+            raise ValueError("Input and output must be different files")
+        if output.exists() and source.samefile(output):
+            raise ValueError("Input and output must be different files")
+        if not arguments.dry_run and output.exists() and not arguments.overwrite:
+            raise ValueError(f"Output exists: {output}; use --overwrite to replace it")
+        duration, start_time = probe_video(path=source)
+        freezes = detect_freezes(path=source, duration=duration, min_still=arguments.min_still, noise=arguments.noise)
+        keyframes = probe_keyframes(path=source, start_time=start_time)
+        segments = plan_segments(duration=duration, freezes=freezes, keep_still=arguments.keep_still,
+                                 keyframes=[time for time, _ in keyframes])
+        retained = sum(end - start for start, end in segments)
+        logging.info(f"{duration:.2f}s → {retained:.2f}s; removing {duration - retained:.2f}s ({(1 - retained / duration):.1%})")
+        if arguments.dry_run:
+            return
+        copy_video(source=source, output=output, segments=segments, duration=duration,
+                   start_time=start_time, keyframes=keyframes, overwrite=arguments.overwrite)
+        logging.info(f"Saved {output}")
+    except (OSError, ValueError, subprocess.CalledProcessError) as error:
+        logging.error(str(error))
+        if isinstance(error, subprocess.CalledProcessError) and error.stderr:
+            logging.error(error.stderr)
+        raise SystemExit(1) from error
+
+
+if __name__ == "__main__":
+    main()