Spaces:
Running
Running
Antigravity
Milestone 1 complete: Extract squarecore, prove byte-identity across 3 canonical seeds (#70006, #80021, #50022)
f123ada Download squarecore/encode/ffmpeg_pipe.py from Sarvesh213/auto2.0: direct link, hf CLI and curl.
- Browser
- Download file 3.87 kB
-
https://huggingface.co/spaces/Sarvesh213/auto2.0/resolve/main/squarecore/encode/ffmpeg_pipe.py
- Command line
-
hf download hf://spaces/Sarvesh213/auto2.0/squarecore/encode/ffmpeg_pipe.py
-
curl -L -o ffmpeg_pipe.py https://huggingface.co/spaces/Sarvesh213/auto2.0/resolve/main/squarecore/encode/ffmpeg_pipe.py
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) | |