From be47aede0278e59e425e4c42d44fd00dddc206dd Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Tue, 20 Oct 2026 10:08:35 +0300 Subject: [PATCH] feat: Add script for trimming still sections off of a video --- ffmpeg_remove_still.py | 192 ++++++++++++++++++++++++++++++++++++ ffmpeg_remove_still_test.py | 107 ++++++++++++++++++++ 2 files changed, 299 insertions(+) create mode 100755 ffmpeg_remove_still.py create mode 100755 ffmpeg_remove_still_test.py diff --git a/ffmpeg_remove_still.py b/ffmpeg_remove_still.py new file mode 100755 index 0000000..748bf81 --- /dev/null +++ b/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() diff --git a/ffmpeg_remove_still_test.py b/ffmpeg_remove_still_test.py new file mode 100755 index 0000000..fb4aea4 --- /dev/null +++ b/ffmpeg_remove_still_test.py @@ -0,0 +1,107 @@ +#!/usr/bin/env -S uv run --script +# /// script +# dependencies = ["pytest"] +# /// + +import json +import shutil +import subprocess +import sys +from pathlib import Path + +import pytest + +import ffmpeg_remove_still as script + + +@pytest.mark.parametrize("log, expected", [ + ("", []), + ("lavfi.freezedetect.freeze_start: 0", [(0.0, 10.0)]), + ("lavfi.freezedetect.freeze_start: 2\nlavfi.freezedetect.freeze_end: 5", [(2.0, 5.0)]), + ("lavfi.freezedetect.freeze_start: 2\nlavfi.freezedetect.freeze_end: 5\n" + "lavfi.freezedetect.freeze_start: 8", [(2.0, 5.0), (8.0, 10.0)]), +]) +def test_parse_freezes(log: str, expected: list[tuple[float, float]]) -> None: + assert script.parse_freezes(log=log, duration=10.0) == expected + + +@pytest.mark.parametrize("freezes, keep_still, keyframes, expected", [ + ([], 1.0, [0.0], [(0.0, 10.0)]), + ([(0.0, 10.0)], 1.0, [0.0, 0.5, 9.5], [(0.0, 0.5), (9.5, 10.0)]), + ([(2.0, 6.0)], 2.0, [0.0, 3.0, 5.0], [(0.0, 3.0), (5.0, 10.0)]), + ([(2.0, 3.0)], 2.0, [0.0, 2.5], [(0.0, 10.0)]), + ([(0.0, 4.0), (6.0, 10.0)], 1.0, [0.0, 0.5, 3.5, 6.5, 9.5], + [(0.0, 0.5), (3.5, 6.5), (9.5, 10.0)]), + ([(2.0, 6.0)], 1.0, [0.0, 3.0, 5.0, 7.0], [(0.0, 3.0), (5.0, 10.0)]), + ([(2.0, 6.0)], 1.0, [0.0, 4.0, 8.0], [(0.0, 10.0)]), + ([(2.0, 6.0)], 1.0, [0.0, 8.0], [(0.0, 10.0)]), +]) +def test_plan_segments(freezes: list[tuple[float, float]], keep_still: float, + keyframes: list[float], expected: list[tuple[float, float]]) -> None: + assert script.plan_segments(duration=10.0, freezes=freezes, keep_still=keep_still, keyframes=keyframes) == expected + + +@pytest.mark.skipif(not shutil.which(cmd="ffmpeg") or not shutil.which(cmd="ffprobe"), reason="Requires FFmpeg") +@pytest.mark.parametrize("has_audio, keep_still, keyframe_interval, expected_duration", [ + (False, 0.5, 5, 4.0), (True, 0.5, 5, 4.0), (True, 1.5, 5, 6.0), + (True, 0.5, 100, 8.0), +]) +def test_copies_packets_without_encoding(tmp_path: Path, has_audio: bool, keep_still: float, + keyframe_interval: int, expected_duration: float) -> None: + source = tmp_path / "demo's recording.mp4" + output = tmp_path / "short.mp4" + command = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-nostdin"] + for pattern in ["testsrc2=size=64x64:rate=10:duration=1", "color=red:size=64x64:rate=10:duration=3", + "testsrc2=size=64x64:rate=10:duration=1", "color=blue:size=64x64:rate=10:duration=3"]: + command.extend(["-f", "lavfi", "-i", pattern]) + if has_audio: + command.extend(["-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=8"]) + command.extend(["-filter_complex", "[0:v][1:v][2:v][3:v]concat=n=4:v=1:a=0[v]", "-map", "[v]"]) + if has_audio: + command.extend(["-map", "4:a", "-c:a", "aac"]) + command.extend(["-c:v", "libx264", "-g", str(keyframe_interval), "-keyint_min", str(keyframe_interval), + "-sc_threshold", "0", "-bf", "2", "-flags", "+cgop", str(source)]) + subprocess.run(args=command, check=True, capture_output=True) + copied = subprocess.run( + args=[sys.executable, str(Path(script.__file__).resolve()), str(source), "--output", str(output), + "--min-still", "2", "--keep-still", str(keep_still)], + check=True, capture_output=True, text=True, + ) + assert "non-monoton" not in copied.stderr.lower() + result = subprocess.run( + args=["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(output)], + 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") + assert float(video["duration"]) == pytest.approx(expected_duration, abs=0.15) + audio = [stream for stream in metadata["streams"] if stream["codec_type"] == "audio"] + assert bool(audio) == has_audio + if has_audio: + assert float(audio[0]["duration"]) == pytest.approx(float(video["duration"]), abs=0.1) + packet_hashes = [] + for path in (source, output): + packets = subprocess.run( + args=["ffprobe", "-v", "error", "-show_packets", "-show_data_hash", "sha256", + "-show_entries", "packet=stream_index,data_hash", "-of", "json", str(path)], + check=True, capture_output=True, text=True, + ) + packet_hashes.append({(packet["stream_index"], packet["data_hash"]) + for packet in json.loads(packets.stdout)["packets"]}) + assert packet_hashes[1] <= packet_hashes[0] + decoded = subprocess.run( + args=["ffmpeg", "-v", "error", "-i", str(output), "-f", "null", "-"], + check=True, capture_output=True, text=True, + ) + assert decoded.stderr == "" + original = source.read_bytes() + refused = subprocess.run( + args=[sys.executable, str(Path(script.__file__).resolve()), str(source), "--output", str(source), "--overwrite"], + capture_output=True, text=True, + ) + assert refused.returncode == 1 + assert source.read_bytes() == original + + +if __name__ == "__main__": + sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))