| from __future__ import annotations |
|
|
| import json |
| import shutil |
| import subprocess |
| import time |
| from pathlib import Path |
|
|
|
|
| def _run(args: list[str]) -> subprocess.CompletedProcess[str]: |
| return subprocess.run(args, capture_output=True, text=True, check=False) |
|
|
|
|
| def probe(path: str | Path) -> dict: |
| ffprobe = shutil.which("ffprobe") |
| if not ffprobe: |
| return {"error": "ffprobe is not available on this host."} |
| result = _run([ffprobe, "-v", "error", "-of", "json", "-show_format", "-show_streams", str(path)]) |
| if result.returncode: |
| return {"error": result.stderr.strip() or "Unable to inspect video."} |
| data = json.loads(result.stdout) |
| video = next((s for s in data.get("streams", []) if s.get("codec_type") == "video"), {}) |
| fmt = data.get("format", {}) |
| return { |
| "width": video.get("width"), |
| "height": video.get("height"), |
| "codec": video.get("codec_name"), |
| "fps": video.get("r_frame_rate"), |
| "duration_s": float(fmt.get("duration", 0) or 0), |
| "size_bytes": int(fmt.get("size", Path(path).stat().st_size) or 0), |
| } |
|
|
|
|
| def encode(path: str | Path, codec: str, workdir: str | Path) -> dict: |
| ffmpeg = shutil.which("ffmpeg") |
| if not ffmpeg: |
| return {"codec": codec, "error": "ffmpeg is not available on this host."} |
| output = Path(workdir) / f"simam_codec_{codec}.mp4" |
| encoders = {"h264": ("libx264", ["-crf", "28"]), "hevc": ("libx265", ["-crf", "30"]), "av1": ("libaom-av1", ["-crf", "35", "-cpu-used", "8"])} |
| encoder, options = encoders[codec] |
| started = time.perf_counter() |
| result = _run([ffmpeg, "-y", "-v", "error", "-i", str(path), "-an", "-c:v", encoder, *options, "-movflags", "+faststart", str(output)]) |
| elapsed = time.perf_counter() - started |
| if result.returncode or not output.exists(): |
| return {"codec": codec, "error": result.stderr.strip() or f"{encoder} is unavailable."} |
| info = probe(output) |
| duration = float(info.get("duration_s") or 0) |
| return {"codec": codec, "size_bytes": output.stat().st_size, "duration_s": duration, "encode_s": round(elapsed, 2), "realtime": round(duration / elapsed, 2) if elapsed else None, "source": str(output), "source_info": info} |
|
|
|
|
| def run_benchmark(path: str | Path) -> tuple[str, list[str]]: |
| source = probe(path) |
| if "error" in source: |
| return f"**Error:** {source['error']}", [] |
| workdir = Path(path).parent |
| results = [encode(path, codec, workdir) for codec in ("h264", "hevc", "av1")] |
| rows = ["| Codec | Size | Bitrate | Encode time | Real-time factor |", "|---|---:|---:|---:|---:|"] |
| for item in results: |
| if "error" in item: |
| rows.append(f"| {item['codec']} | unavailable | — | — | {item['error']} |") |
| continue |
| bitrate = (item["size_bytes"] * 8 / item["duration_s"] / 1000) if item["duration_s"] else 0 |
| rows.append(f"| {item['codec'].upper()} | {item['size_bytes'] / 1024:.1f} KB | {bitrate:.1f} kb/s | {item['encode_s']} s | {item['realtime']}x |") |
| report = "\n".join([f"### Source\n`{source['width']}×{source['height']}` · `{source['codec']}` · `{source['duration_s']:.2f}s` · `{source['size_bytes'] / 1024:.1f} KB`", "", "### Encode comparison", "", *rows, "", "> Quality note: this first pass reports bitrate and runtime. Add frame-level PSNR/VMAF before drawing quality conclusions."]) |
| return report, [x["source"] for x in results if "source" in x] |
|
|
