auto2.0 / squarecore /encode /ffmpeg_pipe.py
Antigravity
Milestone 1 complete: Extract squarecore, prove byte-identity across 3 canonical seeds (#70006, #80021, #50022)
f123ada
Raw History Blame Contribute Delete
3.87 kB
"""FFmpeg raw BGRA subprocess pipe encoder producing H.264 High + AAC faststart MP4."""
from __future__ import annotations
import subprocess
from collections.abc import Generator
from pathlib import Path
def encode_silent_mp4(
frame_generator: Generator[bytes, None, None],
out_path: Path,
width: int = 1080,
height: int = 1920,
fps: int = 60,
) -> None:
"""Stream raw BGRA frames directly into FFmpeg stdin to produce a silent H.264 MP4."""
out_path.parent.mkdir(parents=True, exist_ok=True)
cmd = [
"ffmpeg",
"-y",
"-loglevel",
"error",
"-f",
"rawvideo",
"-vcodec",
"rawvideo",
"-s",
f"{width}x{height}",
"-pix_fmt",
"bgra",
"-r",
str(fps),
"-i",
"pipe:0",
"-c:v",
"libx264",
"-profile:v",
"high",
"-pix_fmt",
"yuv420p",
"-preset",
"fast",
"-crf",
"18",
"-movflags",
"+faststart",
str(out_path),
]
log_path = out_path.with_suffix(".ffmpeg.log")
with open(log_path, "wb") as log_file:
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=log_file,
)
try:
assert proc.stdin is not None
for frame_bytes in frame_generator:
proc.stdin.write(frame_bytes)
proc.stdin.flush()
proc.stdin.close()
proc.wait()
if proc.returncode != 0:
err_msg = log_path.read_text(errors="replace")
raise RuntimeError(f"FFmpeg encoding failed with code {proc.returncode}:\n{err_msg}")
except Exception:
proc.kill()
raise
finally:
if log_path.exists() and proc.returncode == 0:
log_path.unlink(missing_ok=True)
def encode_full_mp4(
frame_generator: Generator[bytes, None, None],
wav_path: Path,
out_path: Path,
width: int = 1080,
height: int = 1920,
fps: int = 60,
) -> None:
"""Stream raw BGRA frames directly into FFmpeg stdin and mux with audio to produce final MP4."""
out_path.parent.mkdir(parents=True, exist_ok=True)
cmd = [
"ffmpeg",
"-y",
"-loglevel",
"error",
"-f",
"rawvideo",
"-vcodec",
"rawvideo",
"-s",
f"{width}x{height}",
"-pix_fmt",
"bgra",
"-r",
str(fps),
"-i",
"pipe:0",
"-i",
str(wav_path),
"-c:v",
"libx264",
"-profile:v",
"high",
"-pix_fmt",
"yuv420p",
"-preset",
"fast",
"-crf",
"18",
"-c:a",
"aac",
"-b:a",
"192k",
"-shortest",
"-movflags",
"+faststart",
str(out_path),
]
log_path = out_path.with_suffix(".ffmpeg.log")
with open(log_path, "wb") as log_file:
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=log_file,
)
try:
assert proc.stdin is not None
for frame_bytes in frame_generator:
proc.stdin.write(frame_bytes)
proc.stdin.flush()
proc.stdin.close()
proc.wait()
if proc.returncode != 0:
err_msg = log_path.read_text(errors="replace")
raise RuntimeError(f"FFmpeg AV encoding failed with code {proc.returncode}:\n{err_msg}")
except Exception:
proc.kill()
raise
finally:
if log_path.exists() and proc.returncode == 0:
log_path.unlink(missing_ok=True)