""" Headless YouTube gateway — render the broadcast page in a browser and push **RTMPS to YouTube**, so the whole live stream runs inside Google Cloud with no desktop/OBS (docs/plans/youtube-live-demo-hr36.md "Run the live stream entirely in Google Cloud"). YouTube live ingest is a *push* protocol (RTMPS) — something must render + encode + push; there is no managed GCP "web → YouTube" service. This is that gateway: on a Compute Engine **GPU** VM it runs the single composited broadcast page ([core/web/index.html] already mixes the 3D scene + boat-cam + commentary into one 1920×1080 page) inside a real Chromium on a virtual X display (Xvfb), captures the display + the page's audio (PulseAudio), and FFmpeg encodes (NVENC when available, else x264) and pushes RTMPS to YouTube. The pieces that matter for correctness — the YouTube URL, the FFmpeg/Chromium command lines, and the encoder choice — are **pure functions** (unit-tested with no system tools). `run()` orchestrates the processes (Xvfb / PulseAudio / Chromium / xdotool / FFmpeg) and only works on the VM; `--dry-run` prints the commands (key masked) so you can inspect them anywhere. VM packages: xvfb pulseaudio chromium-browser ffmpeg xdotool (+ the NVIDIA driver for NVENC/WebGL). """ from __future__ import annotations import shutil import subprocess DEFAULT_RES = "1920x1080" DEFAULT_FPS = 30 DEFAULT_BITRATE_K = 6000 # 6 Mbps — YouTube's recommended 1080p30 bitrate DEFAULT_DISPLAY = ":99" DEFAULT_AUDIO_SINK = "vr_sink" # a PulseAudio null sink; Chromium plays into it, FFmpeg taps .monitor # --- pure command/URL builders (unit-tested) --------------------------------------------------- def youtube_url(stream_key: str, *, secure: bool = True) -> str: """YouTube live ingest URL. RTMPS (TLS, port 443) by default — YouTube's recommended secure ingest; `secure=False` gives plain RTMP.""" if secure: return f"rtmps://a.rtmps.youtube.com/live2/{stream_key}" return f"rtmp://a.rtmp.youtube.com/live2/{stream_key}" def resolve_encoder(prefer: str, *, nvenc_available: bool) -> str: """Resolve `auto|nvenc|x264` to a concrete encoder. `auto` picks NVENC when ffmpeg has it (GPU VM), else x264 (CPU).""" if prefer in ("nvenc", "x264"): return prefer if prefer != "auto": raise ValueError(f"unknown encoder {prefer!r}; valid: auto | nvenc | x264") return "nvenc" if nvenc_available else "x264" def chromium_cmd(*, url: str, res: str = DEFAULT_RES, gpu: bool = True, binary: str = "chromium") -> list[str]: """Chromium on the virtual display: kiosk full-frame, autoplay unblocked, GPU WebGL (EGL) when a GPU is present, else the SwiftShader software path.""" w, h = res.split("x") cmd = [binary, "--kiosk", "--no-first-run", "--no-default-browser-check", "--disable-infobars", "--no-sandbox", # the gateway runs Chrome as root on the VM "--autoplay-policy=no-user-gesture-required", "--disable-features=Translate", "--window-position=0,0", f"--window-size={w},{h}", "--start-fullscreen"] # Software path: modern Chrome dropped bare --use-gl=swiftshader and gates SwiftShader WebGL behind # --enable-unsafe-swiftshader, so a stale flag leaves the WebGL canvas blank. ANGLE→SwiftShader works. cmd += (["--use-gl=egl", "--enable-gpu-rasterization", "--ignore-gpu-blocklist"] if gpu else ["--use-gl=angle", "--use-angle=swiftshader", "--enable-unsafe-swiftshader", "--disable-gpu"]) cmd.append(url) return cmd def ffmpeg_cmd(*, display: str, audio_source: str, url: str, res: str = DEFAULT_RES, fps: int = DEFAULT_FPS, bitrate_k: int = DEFAULT_BITRATE_K, encoder: str = "x264") -> list[str]: """Capture the X display (video) + the PulseAudio monitor (audio) and push FLV/RTMPS to YouTube. CBR, yuv420p, a keyframe every 2 s and AAC audio — YouTube's ingest requirements.""" gop = str(fps * 2) br, buf = f"{bitrate_k}k", f"{bitrate_k * 2}k" if encoder == "nvenc": venc = ["-c:v", "h264_nvenc", "-preset", "p4", "-tune", "ll", "-rc", "cbr", "-profile:v", "high", "-bf", "0"] else: venc = ["-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-profile:v", "high"] return [ "ffmpeg", "-hide_banner", "-y", "-thread_queue_size", "512", "-f", "x11grab", "-framerate", str(fps), "-video_size", res, "-i", display, "-thread_queue_size", "512", "-f", "pulse", "-i", audio_source, "-map", "0:v", "-map", "1:a", *venc, "-b:v", br, "-maxrate", br, "-bufsize", buf, "-g", gop, "-keyint_min", str(fps), # Force constant frame rate + a keyframe every 2 s: an x11grab of a slow (software-WebGL) render is # variable-rate with sparse keyframes, which leaves YouTube stuck on "Preparing stream". CFR + a # 2 s IDR cadence lets it start playback. (`-vsync cfr` for ffmpeg 4.x on the Ubuntu 22.04 VM.) "-vsync", "cfr", "-r", str(fps), "-force_key_frames", "expr:gte(t,n_forced*2)", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "160k", "-ar", "44100", "-ac", "2", "-f", "flv", url, ] def _mask(cmd: list[str], key: str) -> str: """Render a command for logging with the stream key redacted.""" return " ".join(c.replace(key, "***") if key and key in c else c for c in cmd) # --- runner (VM only) -------------------------------------------------------------------------- def nvenc_available() -> bool: """True if this ffmpeg build exposes the NVENC H.264 encoder (i.e. a GPU VM).""" if shutil.which("ffmpeg") is None: return False try: out = subprocess.run(["ffmpeg", "-hide_banner", "-encoders"], # noqa: S603,S607 capture_output=True, text=True, timeout=10).stdout except (OSError, subprocess.SubprocessError): return False return "h264_nvenc" in out def run(*, url: str, key: str, res: str = DEFAULT_RES, fps: int = DEFAULT_FPS, bitrate_k: int = DEFAULT_BITRATE_K, encoder: str = "auto", display: str = DEFAULT_DISPLAY, gpu: bool = True, secure: bool = True, chromium_bin: str = "chromium", audio_sink: str = DEFAULT_AUDIO_SINK, unlock_delay: float = 6.0, dry_run: bool = False) -> None: """Bring up Xvfb + PulseAudio + Chromium (on the scene `url`) + FFmpeg → RTMPS to YouTube. Blocks on FFmpeg; tears everything down on exit. `--dry-run` just prints the resolved commands.""" import os import time yt = youtube_url(key, secure=secure) enc = resolve_encoder(encoder, nvenc_available=(not dry_run and nvenc_available())) monitor = f"{audio_sink}.monitor" xvfb = ["Xvfb", display, "-screen", "0", f"{res}x24", "-nolisten", "tcp"] chrome = chromium_cmd(url=url, res=res, gpu=gpu, binary=chromium_bin) ff = ffmpeg_cmd(display=display, audio_source=monitor, url=yt, res=res, fps=fps, bitrate_k=bitrate_k, encoder=enc) if dry_run: print(f"# encoder: {enc} audio: {monitor} target: {youtube_url('***', secure=secure)}") print("Xvfb :", " ".join(xvfb)) print("chromium:", " ".join(chrome)) print("ffmpeg :", _mask(ff, key)) return env = {**os.environ, "DISPLAY": display, "PULSE_SINK": audio_sink} procs: list[subprocess.Popen] = [] try: procs.append(subprocess.Popen(xvfb)) # noqa: S603 time.sleep(1.5) subprocess.run(["pulseaudio", "--start", "--exit-idle-time=-1"], check=False) # noqa: S603,S607 subprocess.run(["pactl", "load-module", "module-null-sink", # noqa: S603,S607 f"sink_name={audio_sink}"], check=False, env=env) procs.append(subprocess.Popen(chrome, env=env)) # noqa: S603 time.sleep(unlock_delay) # Unlock the page's audio + start (the broadcast page is "frozen until the first click"). subprocess.run(["xdotool", "mousemove", "960", "540", "click", "1"], # noqa: S603,S607 check=False, env=env) print(f"gateway: streaming {res}@{fps} via {enc} → {youtube_url('***', secure=secure)}") subprocess.run(ff, check=False, env=env) # noqa: S603 — blocks finally: for p in reversed(procs): p.terminate() def main() -> None: import argparse import os ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--url", default=None, help="broadcast page URL (the serve_v3_live scene). Unset falls back to " "$VR_PRODUCER_BASE + /core/web/index.html?feed=live, then to the local dev " "default; VR_RACE_DAY=1 refuses a loopback URL") ap.add_argument("--key", default=os.environ.get("YOUTUBE_STREAM_KEY", ""), help="YouTube stream key (or $YOUTUBE_STREAM_KEY)") ap.add_argument("--res", default=DEFAULT_RES) ap.add_argument("--fps", type=int, default=DEFAULT_FPS) ap.add_argument("--bitrate", type=int, default=DEFAULT_BITRATE_K, help="video bitrate (kbps)") ap.add_argument("--encoder", default="auto", choices=["auto", "nvenc", "x264"]) ap.add_argument("--display", default=DEFAULT_DISPLAY) ap.add_argument("--no-gpu", dest="gpu", action="store_false", help="software WebGL (SwiftShader)") ap.add_argument("--rtmp", dest="secure", action="store_false", help="plain RTMP instead of RTMPS") ap.add_argument("--chromium", default="chromium", help="chromium/google-chrome binary") ap.add_argument("--dry-run", action="store_true", help="print the commands (key masked) and exit") args = ap.parse_args() if not args.key and not args.dry_run: raise SystemExit("--key (or $YOUTUBE_STREAM_KEY) is required") from producer_base import RaceDayBaseError, resolve try: args.url, _src = resolve(args.url, default="http://localhost:8765/core/web/index.html?feed=live", path="/core/web/index.html?feed=live") except RaceDayBaseError as e: print(f"[gateway] {e}", flush=True) return 2 print(f"[gateway] broadcast page: {args.url} (from the {_src})", flush=True) run(url=args.url, key=args.key, res=args.res, fps=args.fps, bitrate_k=args.bitrate, encoder=args.encoder, display=args.display, gpu=args.gpu, secure=args.secure, chromium_bin=args.chromium, dry_run=args.dry_run) if __name__ == "__main__": main()