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())