Spaces:
Running
Running
File size: 3,872 Bytes
f123ada | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """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)
|