From f2156312f6db547147ae73a80a6902ed5571751e Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 12 Aug 2026 16:53:30 +0200 Subject: [PATCH] feat(echo): add a robot mode that transforms the audio (#58) echo dropped audio in three places, so trickle looked video-only: the client decoded video=0, the runner returned None for non-video frames, and only a video track was published. A stream with sound came back silent with no explanation. robot ring-modulates the audio and leaves the video alone, which makes the round trip audible. Ring modulation rather than a pitch shift because it preserves the sample count, so audio stays in sync without resampling. The output track is pinned to 48 kHz since opus rejects 44.1, which every consumer source produces. Only robot publishes an audio track: the container opens once every declared track has a first frame, so declaring audio for a silent input would stall the video. Co-Authored-By: Claude Opus 5 (1M context) --- .pre-commit-config.yaml | 12 ++++ echo/README.md | 21 ++++++- echo/client.py | 48 ++++++++++++--- echo/pyproject.toml | 1 + echo/runner.py | 57 +++++++++++++++--- scripts/check_orchestrator_drift.py | 93 +++++++++++++++++++++++++++++ 6 files changed, 216 insertions(+), 16 deletions(-) create mode 100755 scripts/check_orchestrator_drift.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 3a54d6f..45dc2c8 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -24,3 +24,15 @@ repos: - id: end-of-file-fixer - id: check-yaml - id: check-merge-conflict + + - repo: local + hooks: + - id: orchestrator-drift + # api-proxy and vllm restate the shared orchestrator command because they + # add -liveRunnerConfig and `extends` replaces a list. Keep the copies honest. + name: orchestrator command drift + entry: python3 scripts/check_orchestrator_drift.py + language: python + additional_dependencies: [pyyaml] + pass_filenames: false + files: ^(compose\.orchestrator\.yml|compose\.onchain\.yml|[^/]+/compose(\.onchain)?\.yml)$ diff --git a/echo/README.md b/echo/README.md index 7d383ce..f138b13 100644 --- a/echo/README.md +++ b/echo/README.md @@ -1,6 +1,6 @@ # Echo app (trickle realtime video) -A realtime video app on the Livepeer network: it receives a live video stream over **trickle** channels, optionally transforms each frame (gray / invert / blur), and echoes it back. This is the **live/stateful** path — continuous media over trickle, not request/response — so the app embeds the SDK and self-registers (dynamic). +A realtime video app on the Livepeer network: it receives a live video stream over **trickle** channels, optionally transforms each frame (gray / invert / blur) or the audio (robot), and echoes it back. This is the **live/stateful** path — continuous media over trickle, not request/response — so the app embeds the SDK and self-registers (dynamic). | | | | ------------ | ------------------------------------ | @@ -73,10 +73,27 @@ Swap `/dev/video0` for your node. If that size/format isn't supported, list the The `ffplay` low-delay flags (`-fflags nobuffer -flags low_delay -framedrop`) keep the preview close to realtime; drop them and it buffers. -- `--mode` picks the transform: `echo` (passthrough, the default), `gray`, `invert`, or `blur`. Use `--mode blur` on any command above to see the echo visibly transform the stream. +- `--mode` picks the transform: `echo` (passthrough, the default), `gray`, `invert`, `blur`, or `robot`. Use `--mode blur` on any command above to see the echo visibly transform the stream. +- `robot` ring-modulates the audio and leaves the video alone. It is the only mode that publishes an audio track, since a declared track that never gets a frame stalls the stream. - `blur` sweeps the radius `0 -> max -> 0` live (driving `/update`); `--blur-period N` sets the seconds per sweep cycle (default 2; larger is slower). `gray`/`invert` are static. - `--radius N` sets the initial blur strength, `--max-frames N` stops early. +**Hearing `robot`** — every command above is video-only, so `robot` would refuse them. Record yourself with a microphone (`arecord -l` lists capture devices), then play both files: + +```sh +ffmpeg -f v4l2 -input_format mjpeg -video_size 1280x720 -framerate 30 -i /dev/video0 \ + -f alsa -i plughw:1,0 -filter_complex "[1:a]aresample=async=1:first_pts=0[a]" \ + -map 0:v -map "[a]" -fps_mode cfr -t 10 \ + -c:v libx264 -preset ultrafast -pix_fmt yuv420p -g 30 -c:a aac -ar 48000 -f mpegts -y me.ts + +uv run client.py --mode robot --output me-robot.ts me.ts +ffplay -autoexit me.ts && ffplay -autoexit me-robot.ts # you, then you ring-modulated +``` + +To hear it live instead, keep the same capture and swap the tail for `--output - -` piped into `ffplay -fflags nobuffer -i -`. Expect 2 to 4 seconds of lag, since trickle publishes in 2s segments, and wear headphones or the mic re-records the playback. + +The camera and the audio device are separate clocks, so `aresample` and `-fps_mode cfr` align them; without both the publisher fails at the first segment boundary. On a multi-input interface add `-channels 6` and pick one input with `pan=mono|c0=c0`. + Stop the stack with `docker compose down`. ## Run on-chain (paid) diff --git a/echo/client.py b/echo/client.py index 526d2c2..f094b7f 100644 --- a/echo/client.py +++ b/echo/client.py @@ -32,7 +32,12 @@ from livepeer_gateway.errors import LivepeerGatewayError from livepeer_gateway.live_runner import stop_runner_session from livepeer_gateway.media_output import MediaOutput -from livepeer_gateway.media_publish import MediaPublish +from livepeer_gateway.media_publish import ( + AudioOutputConfig, + MediaPublish, + MediaPublishConfig, + VideoOutputConfig, +) from livepeer_gateway.http import post_json from livepeer_gateway.selection import reserve_session @@ -40,7 +45,7 @@ APP_ID = "livepeer-example/echo" DEFAULT_OUTPUT = "echo-out.ts" MAX_BLUR_RADIUS = 100 -MODES = ("echo", "gray", "invert", "blur") +MODES = ("echo", "gray", "invert", "blur", "robot") log = logging.getLogger("echo-client") @@ -79,7 +84,8 @@ def _parse_args() -> argparse.Namespace: choices=MODES, default="echo", help=( - "Transform the runner applies: echo (passthrough), gray, invert, or blur. " + "Transform the runner applies: echo (passthrough), gray, invert, blur, " + "or robot (ring-modulates the audio). " "blur sweeps the radius; the rest are static." ), ) @@ -117,7 +123,20 @@ async def _publish_video( raise LivepeerGatewayError( f"No video stream found in input: {input_source}" ) - publisher = MediaPublish(publish_url) # Livepeer: 2 (publish frames) + # Only robot touches audio, so only robot publishes an audio track: the + # container waits for a first frame on every track it declares. + send_audio = mode == "robot" + if send_audio and not input_.streams.audio: + raise LivepeerGatewayError( + f"robot needs audio, but the input has none: {input_source}" + ) + tracks: list[VideoOutputConfig | AudioOutputConfig] = [VideoOutputConfig()] + if send_audio: + # Pinned: opus rejects 44.1 kHz, so let MediaPublish resample to 48. + tracks.append(AudioOutputConfig(sample_rate=48000)) + publisher = MediaPublish( # Livepeer: 2 (publish frames) + publish_url, config=MediaPublishConfig(tracks=tracks) + ) prev_pts_time: float | None = None prev_wall: float | None = None next_update_pts_time: float | None = None @@ -126,9 +145,18 @@ async def _publish_video( # blur sweeps 0->max->0 (2*MAX steps); spread one full cycle over blur_period. update_interval = blur_period / (2 * MAX_BLUR_RADIUS) + video_index = 0 try: - for index, frame in enumerate(input_.decode(video=0), start=1): - if max_frames > 0 and index > max_frames: + # decode() yields both streams interleaved; without audio, stay on the + # video stream alone. Pacing and the blur sweep run off video frames only. + frames = input_.decode() if send_audio else input_.decode(video=0) + for frame in frames: + if not isinstance(frame, av.VideoFrame): + await publisher.write_frame(frame) + continue + + video_index += 1 + if max_frames > 0 and video_index > max_frames: break current_pts_time = None if frame.pts is not None and frame.time_base is not None: @@ -209,7 +237,13 @@ async def main() -> None: async with session: echo = await post_json( f"{session.app_url.rstrip('/')}/echo", - {"radius": args.radius, "mode": args.mode}, + # robot is the mode that transforms audio, so it is also the one + # that asks the runner for an audio track. + { + "radius": args.radius, + "mode": args.mode, + "audio": args.mode == "robot", + }, ) in_url = _channel_url(echo, "in") out_url = _channel_url(echo, "out") diff --git a/echo/pyproject.toml b/echo/pyproject.toml index a82ff79..80267e7 100644 --- a/echo/pyproject.toml +++ b/echo/pyproject.toml @@ -6,6 +6,7 @@ requires-python = ">=3.12" dependencies = [ "av", # client + runner: decode/encode video frames "opencv-python-headless", # runner: gray/invert/blur transforms + "numpy", # runner: robot ring modulation "aiohttp", "livepeer-gateway", ] diff --git a/echo/runner.py b/echo/runner.py index 066697b..259deb8 100644 --- a/echo/runner.py +++ b/echo/runner.py @@ -25,18 +25,27 @@ from typing import Any import av +import numpy as np from aiohttp import web from livepeer_gateway.live_runner import register_runner from livepeer_gateway.media_decode import AudioDecodedMediaFrame, VideoDecodedMediaFrame from livepeer_gateway.media_output import MediaOutput -from livepeer_gateway.media_publish import MediaPublish +from livepeer_gateway.media_publish import ( + AudioOutputConfig, + MediaPublish, + MediaPublishConfig, + VideoOutputConfig, +) log = logging.getLogger("echo") DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8989 -MODES = frozenset({"echo", "gray", "invert", "blur"}) +MODES = frozenset({"echo", "gray", "invert", "blur", "robot"}) +# "robot" multiplies each sample by a sine at this frequency; the sample count is +# unchanged, so audio stays in sync with video. +ROBOT_HZ = 220.0 state: EchoSession | None = None @@ -121,15 +130,37 @@ def _odd_kernel(radius: int) -> int: return min(kernel, 99) +def _robot_audio(frame: av.AudioFrame) -> av.AudioFrame: + # sample[i] *= sin(2*pi*ROBOT_HZ*t[i]). The carrier phase comes from the frame's + # own timestamp, so it stays continuous across frames (no clicks) without keeping + # state, and |carrier| <= 1 means it cannot clip. + samples = frame.to_ndarray() + t0 = float(frame.pts * frame.time_base) if frame.pts is not None else 0.0 + t = t0 + np.arange(samples.shape[-1], dtype=np.float32) / frame.sample_rate + carrier = np.sin(2.0 * np.pi * ROBOT_HZ * t).astype(np.float32) + out = av.AudioFrame.from_ndarray( + (samples.astype(np.float32) * carrier).astype(samples.dtype), + format=frame.format.name, + layout=frame.layout.name, + ) + out.sample_rate = frame.sample_rate + out.pts = frame.pts + out.time_base = frame.time_base + return out + + def _transform_frame( decoded: AudioDecodedMediaFrame | VideoDecodedMediaFrame, mode: ModeState, -) -> av.VideoFrame | None: +) -> av.VideoFrame | av.AudioFrame | None: + frame = decoded.frame + if decoded.kind == "audio": + # Audio rides along untouched; only "robot" transforms it. + return _robot_audio(frame) if mode.mode == "robot" else frame if decoded.kind != "video": return None - frame = decoded.frame - if mode.mode == "echo": + if mode.mode in ("echo", "robot"): # robot changes audio only return frame import cv2 @@ -175,9 +206,21 @@ async def _handle_echo(request: web.Request) -> web.Response: ) # for production apps, handle errors - mode = _parse_mode(json.loads(await request.read())) + payload = json.loads(await request.read()) + mode = _parse_mode(payload) + # Tracks are declared upfront and the container waits for a first frame on each, + # so only declare audio when the client says it is sending some. + tracks: list[VideoOutputConfig | AudioOutputConfig] = [VideoOutputConfig()] + send_audio = payload.get("audio", False) + if not isinstance(send_audio, bool): + raise web.HTTPBadRequest(text="audio must be a boolean") + if send_audio: + tracks.append(AudioOutputConfig()) # internal_url: runner-reachable address (same as the public url on a shared net). - publisher = MediaPublish(by_name["out"].get("internal_url", by_name["out"]["url"])) + publisher = MediaPublish( + by_name["out"].get("internal_url", by_name["out"]["url"]), + config=MediaPublishConfig(tracks=tracks), + ) async def _on_frame(decoded) -> None: frame = _transform_frame(decoded, mode) diff --git a/scripts/check_orchestrator_drift.py b/scripts/check_orchestrator_drift.py new file mode 100755 index 0000000..a01ea97 --- /dev/null +++ b/scripts/check_orchestrator_drift.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 +"""Fail if an example's restated orchestrator command has drifted from the shared one. + +`compose.orchestrator.yml` and `compose.onchain.yml` define the orchestrator once and +examples pull it in with `extends`. Static runners cannot: they need an extra +`-liveRunnerConfig` flag, and `extends` replaces a command list rather than appending +to it, so they restate the whole thing. This checks those copies still match. + +Run directly, or via pre-commit. Exits non-zero and prints what differs. +""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path + +import yaml + +# Flags a copy may add to the shared command. Anything else is drift. +ALLOWED_EXTRA = {"-liveRunnerConfig"} + +SHARED = { + "compose.yml": Path("compose.orchestrator.yml"), + "compose.onchain.yml": Path("compose.onchain.yml"), +} + + +def _command(path: Path) -> list[str] | None: + doc = yaml.safe_load(path.read_text()) or {} + service = (doc.get("services") or {}).get("orchestrator") or {} + command = service.get("command") + return command if isinstance(command, list) else None + + +def _flag(item: str) -> str: + return str(item).split("=", 1)[0] + + +def main() -> int: + root = Path(__file__).resolve().parent.parent + tracked = subprocess.run( + ["git", "ls-files", "*/compose.yml", "*/compose.onchain.yml"], + cwd=root, + capture_output=True, + text=True, + check=True, + ).stdout.split() + + problems: list[str] = [] + checked = 0 + for rel in sorted(tracked): + path = root / rel + copy = _command(path) + if copy is None: # uses `extends` alone, nothing to drift + continue + shared_path = root / SHARED[Path(rel).name] + shared = _command(shared_path) or [] + checked += 1 + + shared_by_flag = {_flag(i): i for i in shared} + copy_by_flag = {_flag(i): i for i in copy} + + for flag, item in shared_by_flag.items(): + if flag not in copy_by_flag: + problems.append(f"{rel}: missing {item!r} (in {shared_path.name})") + elif copy_by_flag[flag] != item: + problems.append( + f"{rel}: {flag} is {copy_by_flag[flag]!r}, " + f"{shared_path.name} has {item!r}" + ) + for flag, item in copy_by_flag.items(): + if flag not in shared_by_flag and flag not in ALLOWED_EXTRA: + problems.append(f"{rel}: unexpected {item!r} not in {shared_path.name}") + + if problems: + print("orchestrator command drift:\n") + for p in problems: + print(f" {p}") + print( + "\nThese examples restate the shared command because they add " + f"{sorted(ALLOWED_EXTRA)} and `extends` cannot append to a list. " + "Re-copy the shared command, or widen ALLOWED_EXTRA if the new flag " + "is deliberate." + ) + return 1 + + print(f"orchestrator command: {checked} restated copies match the shared files") + return 0 + + +if __name__ == "__main__": + sys.exit(main())