From 3c32206a2b2ad43dd875f960f4fe08c394134859 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 12 Aug 2026 18:38:45 +0200 Subject: [PATCH 1/2] feat(hello-world): serve a generated OpenAPI spec via FastAPI Discovery gives a caller an app id and a URL and nothing about the interface, so the only way to learn what an app accepts is to read its source. FastAPI derives the schema from the request and response models and serves it at /openapi.json, which the orchestrator proxies like any other endpoint, so a caller can fetch it from the discovered url. Exploratory: one example only, to size the change before deciding whether the convention is worth adopting across the repo. Co-Authored-By: Claude Opus 5 (1M context) --- hello-world/Dockerfile | 3 +- hello-world/pyproject.toml | 2 ++ hello-world/runner.py | 65 ++++++++++++++++++++------------------ 3 files changed, 39 insertions(+), 31 deletions(-) diff --git a/hello-world/Dockerfile b/hello-world/Dockerfile index 6907f43..040344a 100644 --- a/hello-world/Dockerfile +++ b/hello-world/Dockerfile @@ -1,4 +1,4 @@ -# Hello-world example app (http server). +# Hello-world example app (FastAPI http server). FROM python:3.12-slim # Flush stdout/stderr immediately so output isn't block-buffered in `docker logs`. @@ -10,6 +10,7 @@ RUN apt-get update \ # livepeer-gateway SDK isn't on PyPI yet; install from Git. RUN pip install --no-cache-dir \ + fastapi uvicorn \ "livepeer-gateway @ git+https://github.com/livepeer/livepeer-python-gateway@main" WORKDIR /app diff --git a/hello-world/pyproject.toml b/hello-world/pyproject.toml index f508ba0..16f28ea 100644 --- a/hello-world/pyproject.toml +++ b/hello-world/pyproject.toml @@ -4,6 +4,8 @@ version = "0.1.0" description = "Hello-world example app for the Livepeer network." requires-python = ">=3.12" dependencies = [ + "fastapi", # runner: routes + generated OpenAPI schema + "uvicorn", # runner: ASGI server "livepeer-gateway", ] diff --git a/hello-world/runner.py b/hello-world/runner.py index 4bad5f2..1735d45 100644 --- a/hello-world/runner.py +++ b/hello-world/runner.py @@ -1,21 +1,24 @@ #!/usr/bin/env python3 -"""hello-world app: a normal aiohttp service, made callable on the Livepeer network. +"""hello-world app: a normal FastAPI service, made callable on the Livepeer network. Livepeer integration (grep `# Livepeer:`): 1. register_runner() — announce the app to the orchestrator (startup) 2. registration.close() — deregister (cleanup) /hello is an ordinary HTTP handler; being on the network doesn't change how you write -it. +it. FastAPI derives the request/response schema from the models below and serves it at +/openapi.json, so callers can discover the interface without reading this file. """ from __future__ import annotations import argparse import logging -from contextlib import suppress +from contextlib import asynccontextmanager, suppress -from aiohttp import web +import uvicorn +from fastapi import FastAPI +from pydantic import BaseModel, Field from livepeer_gateway.live_runner import register_runner @@ -43,25 +46,18 @@ def _parse_args() -> argparse.Namespace: return parser.parse_args() -async def _handle_hello(request: web.Request) -> web.Response: - try: - payload = await request.json() - except Exception: - payload = None - if not isinstance(payload, dict): - raise web.HTTPBadRequest(text="body must be a JSON object") - name = str(payload.get("name", "world")) - return web.json_response({"message": f"Hello, {name}!"}) +class HelloRequest(BaseModel): + name: str = Field("world", description="Who to greet.") -def main() -> None: - logging.basicConfig( - level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" - ) - args = _parse_args() +class HelloResponse(BaseModel): + message: str + - async def _on_startup(app: web.Application) -> None: - app["registration"] = await register_runner( # Livepeer: 1 +def build_app(args: argparse.Namespace) -> FastAPI: + @asynccontextmanager + async def _lifespan(app: FastAPI): + app.state.registration = await register_runner( # Livepeer: 1 args.orchestrator, secret=args.orchSecret, runner_url=args.runner_url, @@ -73,19 +69,28 @@ async def _on_startup(app: web.Application) -> None: ) log.info( "registered runner_id=%s orchestrator=%s", - app["registration"].runner_id, - app["registration"].orchestrator_url, + app.state.registration.runner_id, + app.state.registration.orchestrator_url, ) - - async def _on_cleanup(app: web.Application) -> None: + yield with suppress(Exception): - await app["registration"].close() # Livepeer: 2 + await app.state.registration.close() # Livepeer: 2 + + app = FastAPI(title=APP_ID, version="0.1.0", lifespan=_lifespan) - app = web.Application() - app.router.add_post("/hello", _handle_hello) - app.on_startup.append(_on_startup) - app.on_cleanup.append(_on_cleanup) - web.run_app(app, host=args.host, port=DEFAULT_PORT) + @app.post("/hello", response_model=HelloResponse) + async def hello(body: HelloRequest) -> HelloResponse: + return HelloResponse(message=f"Hello, {body.name}!") + + return app + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + uvicorn.run(build_app(args), host=args.host, port=DEFAULT_PORT, access_log=False) if __name__ == "__main__": From 92bbf4ae609d0d7a087e1d2ccc64618a1bc9dd7e Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 12 Aug 2026 18:47:56 +0200 Subject: [PATCH 2/2] feat: serve a generated OpenAPI spec from every Python runner Discovery gives a caller an app id and a URL and nothing about the interface, so the only way to learn what an app accepts is to read its source. FastAPI derives the schema from the request and response models and serves it at /openapi.json, which the orchestrator proxies like any other endpoint. Converts the four examples that run a server of ours. vllm and api-proxy are static runners with no code of ours, so they cannot participate. Exploratory, to size the change rather than to merge as is. Co-Authored-By: Claude Opus 5 (1M context) --- echo/Dockerfile | 1 + echo/pyproject.toml | 2 + echo/runner.py | 238 +++++++++++++------------- realtime-transcription/Dockerfile | 1 + realtime-transcription/pyproject.toml | 3 + realtime-transcription/runner.py | 74 ++++---- tiles/Dockerfile | 1 + tiles/pyproject.toml | 2 + tiles/runner.py | 84 ++++----- 9 files changed, 215 insertions(+), 191 deletions(-) diff --git a/echo/Dockerfile b/echo/Dockerfile index a220b45..9246a25 100644 --- a/echo/Dockerfile +++ b/echo/Dockerfile @@ -11,6 +11,7 @@ RUN apt-get update \ # livepeer-gateway SDK isn't on PyPI yet; install from Git. av (PyAV) ships its # own ffmpeg; opencv-python-headless needs no system GUI libs. RUN pip install --no-cache-dir \ + fastapi uvicorn \ "livepeer-gateway @ git+https://github.com/livepeer/livepeer-python-gateway@main" \ av opencv-python-headless aiohttp diff --git a/echo/pyproject.toml b/echo/pyproject.toml index 80267e7..9a00746 100644 --- a/echo/pyproject.toml +++ b/echo/pyproject.toml @@ -4,6 +4,8 @@ version = "0.1.0" description = "Echo (trickle realtime video) example app for the Livepeer network." requires-python = ">=3.12" dependencies = [ + "fastapi", # runner: routes + generated OpenAPI schema + "uvicorn", # runner: ASGI server "av", # client + runner: decode/encode video frames "opencv-python-headless", # runner: gray/invert/blur transforms "numpy", # runner: robot ring modulation diff --git a/echo/runner.py b/echo/runner.py index 259deb8..e5574e3 100644 --- a/echo/runner.py +++ b/echo/runner.py @@ -18,15 +18,16 @@ import argparse import asyncio -import json import logging -from contextlib import suppress +from contextlib import asynccontextmanager, suppress from dataclasses import dataclass -from typing import Any +from typing import Any, Literal import av import numpy as np -from aiohttp import web +import uvicorn +from fastapi import FastAPI, HTTPException, Request +from pydantic import BaseModel, Field from livepeer_gateway.live_runner import register_runner from livepeer_gateway.media_decode import AudioDecodedMediaFrame, VideoDecodedMediaFrame @@ -42,7 +43,6 @@ DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8989 -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 @@ -104,23 +104,35 @@ def _parse_args() -> argparse.Namespace: return parser.parse_args() -def _session_id(request: web.Request) -> str: +def _session_id(request: Request) -> str: session_id = request.headers.get("Livepeer-Session-Id", "").strip() if not session_id: - raise web.HTTPBadRequest(text="missing Livepeer-Session-Id header") + raise HTTPException( + status_code=400, detail="missing Livepeer-Session-Id header" + ) return session_id -def _parse_mode(payload: dict[str, Any]) -> ModeState: - mode = str(payload.get("mode", "echo")).strip().lower() - if mode not in MODES: - raise web.HTTPBadRequest(text=f"mode must be one of {sorted(MODES)}") - radius = payload.get("radius", 7) - try: - radius_int = int(radius) - except (TypeError, ValueError) as exc: - raise web.HTTPBadRequest(text="radius must be an integer") from exc - return ModeState(mode=mode, radius=max(1, min(99, radius_int))) +class EchoRequest(BaseModel): + mode: Literal["echo", "gray", "invert", "blur", "robot"] = "echo" + # The client sweeps 0..100; clamped rather than rejected, as before. + radius: int = Field(7, description="Blur strength; blur mode only.") + audio: bool = Field(False, description="Publish an audio track (robot needs one).") + + +class UpdateRequest(BaseModel): + mode: Literal["echo", "gray", "invert", "blur", "robot"] = "echo" + radius: int = 7 + + +class SessionResponse(BaseModel): + session: str + in_: str = Field(..., alias="in") + out: str + mode: str + radius: int | None = None + + model_config = {"populate_by_name": True} def _odd_kernel(radius: int) -> int: @@ -181,83 +193,101 @@ def _transform_frame( return out -async def _handle_echo(request: web.Request) -> web.Response: - global state - session_id = _session_id(request) - - if state is not None: - if state.session_id != session_id: - raise web.HTTPConflict(text="echo runner already has an active session") - return web.json_response(state.to_json()) - - # Pass the request so the SDK opens channels using the orchestrator's - # Session-Control header, whose URLs are reachable from the runner's network. - channels = await request.app["registration"].create_trickle_channels( # Livepeer: 2 - request, - [ - {"name": "in", "mime_type": "video/mp2t"}, - {"name": "out", "mime_type": "video/mp2t"}, - ], - ) - by_name = {channel["name"]: channel for channel in channels} - if "in" not in by_name or "out" not in by_name: - raise web.HTTPInternalServerError( - text="orchestrator did not return in/out channels" +def build_app(args: argparse.Namespace) -> FastAPI: + @asynccontextmanager + async def _lifespan(app: FastAPI): + app.state.registration = await register_runner( # Livepeer: 1 + args.orchestrator, + secret=args.orchSecret, + runner_url=args.runner_url, + app="livepeer-example/echo", + mode="persistent", + price=args.price, + ) + log.info( + "registered runner_id=%s orchestrator=%s", + app.state.registration.runner_id, + app.state.registration.orchestrator_url, + ) + yield + await _close_pipeline() + with suppress(Exception): + await app.state.registration.close() # Livepeer: 3 + + app = FastAPI(title="livepeer-example/echo", version="0.1.0", lifespan=_lifespan) + + @app.post("/echo", response_model=SessionResponse, response_model_by_alias=True) + async def echo(body: EchoRequest, request: Request) -> dict[str, Any]: + global state + session_id = _session_id(request) + + if state is not None: + if state.session_id != session_id: + raise HTTPException(409, "echo runner already has an active session") + return state.to_json() + + # Pass the request so the SDK opens channels using the orchestrator's + # Session-Control header, whose URLs are reachable from the runner's network. + channels = ( + await request.app.state.registration.create_trickle_channels( # Livepeer: 2 + request, + [ + {"name": "in", "mime_type": "video/mp2t"}, + {"name": "out", "mime_type": "video/mp2t"}, + ], + ) + ) + by_name = {channel["name"]: channel for channel in channels} + if "in" not in by_name or "out" not in by_name: + raise HTTPException(500, "orchestrator did not return in/out channels") + + mode = ModeState(mode=body.mode, radius=max(1, min(99, body.radius))) + # 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()] + if body.audio: + tracks.append(AudioOutputConfig()) + # internal_url: runner-reachable address (same as public url on a shared net). + publisher = MediaPublish( + by_name["out"].get("internal_url", by_name["out"]["url"]), + config=MediaPublishConfig(tracks=tracks), ) - # for production apps, handle errors - 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"]), - config=MediaPublishConfig(tracks=tracks), - ) - - async def _on_frame(decoded) -> None: - frame = _transform_frame(decoded, mode) - if frame is not None: - await publisher.write_frame(frame) - - output = MediaOutput( - by_name["in"].get("internal_url", by_name["in"]["url"]), on_frame=_on_frame - ) - - # Hand public channel urls to the client, so it can send/receive media. - state = EchoSession( - session_id=session_id, - in_url=by_name["in"]["url"], - out_url=by_name["out"]["url"], - mode=mode, - output=output, - publisher=publisher, - ) - for task in output.callback_tasks(): - task.add_done_callback(lambda _task: asyncio.create_task(_close_pipeline())) - log.info("started echo session %s", session_id) - return web.json_response(state.to_json()) + async def _on_frame(decoded) -> None: + frame = _transform_frame(decoded, mode) + if frame is not None: + await publisher.write_frame(frame) + output = MediaOutput( + by_name["in"].get("internal_url", by_name["in"]["url"]), on_frame=_on_frame + ) -async def _handle_update(request: web.Request) -> web.Response: - session_id = _session_id(request) - if state is None: - raise web.HTTPNotFound(text="echo session not started") - if state.session_id != session_id: - raise web.HTTPConflict(text="echo runner has a different active session") + # Hand public channel urls to the client, so it can send/receive media. + state = EchoSession( + session_id=session_id, + in_url=by_name["in"]["url"], + out_url=by_name["out"]["url"], + mode=mode, + output=output, + publisher=publisher, + ) + for task in output.callback_tasks(): + task.add_done_callback(lambda _t: asyncio.create_task(_close_pipeline())) + log.info("started echo session %s", session_id) + return state.to_json() + + @app.post("/update", response_model=SessionResponse, response_model_by_alias=True) + async def update(body: UpdateRequest, request: Request) -> dict[str, Any]: + session_id = _session_id(request) + if state is None: + raise HTTPException(404, "echo session not started") + if state.session_id != session_id: + raise HTTPException(409, "echo runner has a different active session") + state.mode.mode = body.mode + state.mode.radius = max(1, min(99, body.radius)) + return state.to_json() - # for production apps, handle errors - mode = _parse_mode(json.loads(await request.read())) - state.mode.mode = mode.mode - state.mode.radius = mode.radius - return web.json_response(state.to_json()) + return app def main() -> None: @@ -265,35 +295,7 @@ def main() -> None: level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" ) args = _parse_args() - - async def _on_startup(app: web.Application) -> None: - app["registration"] = await register_runner( # Livepeer: 1 - args.orchestrator, - secret=args.orchSecret, - runner_url=args.runner_url, - app="livepeer-example/echo", - mode="persistent", # realtime trickle streaming is a held-open session - # Metered: the session is billed per second of wall-clock for as long - # as the client holds it, which is what a live stream costs. - price=args.price, # USD per hour - ) - log.info( - "registered runner_id=%s orchestrator=%s", - app["registration"].runner_id, - app["registration"].orchestrator_url, - ) - - async def _on_cleanup(app: web.Application) -> None: - await _close_pipeline() - with suppress(Exception): - await app["registration"].close() # Livepeer: 3 - - app = web.Application() - app.router.add_post("/echo", _handle_echo) - app.router.add_post("/update", _handle_update) - app.on_startup.append(_on_startup) - app.on_cleanup.append(_on_cleanup) - web.run_app(app, host=args.host, port=DEFAULT_PORT) + uvicorn.run(build_app(args), host=args.host, port=DEFAULT_PORT, access_log=False) if __name__ == "__main__": diff --git a/realtime-transcription/Dockerfile b/realtime-transcription/Dockerfile index 06feaca..169978d 100644 --- a/realtime-transcription/Dockerfile +++ b/realtime-transcription/Dockerfile @@ -12,6 +12,7 @@ RUN apt-get update \ # faster-whisper) loads cuBLAS/cuDNN from the nvidia wheels, so no CUDA base image # is needed -- the host driver comes in via the compose `deploy` reservation. RUN pip install --no-cache-dir \ + fastapi uvicorn websockets \ faster-whisper numpy \ nvidia-cublas-cu12 nvidia-cudnn-cu12 \ "livepeer-gateway @ git+https://github.com/livepeer/livepeer-python-gateway@main" diff --git a/realtime-transcription/pyproject.toml b/realtime-transcription/pyproject.toml index 406540d..1446751 100644 --- a/realtime-transcription/pyproject.toml +++ b/realtime-transcription/pyproject.toml @@ -4,6 +4,9 @@ version = "0.1.0" description = "Streaming speech-to-text (WebSocket) example app for the Livepeer network." requires-python = ">=3.12" dependencies = [ + "fastapi", # runner: routes + generated OpenAPI schema + "uvicorn", # runner: ASGI server + "websockets", # runner: uvicorn WebSocket support "faster-whisper", # runner: streaming Whisper ASR "numpy", # runner: PCM buffers + energy VAD "aiohttp", diff --git a/realtime-transcription/runner.py b/realtime-transcription/runner.py index 8573249..b127e79 100644 --- a/realtime-transcription/runner.py +++ b/realtime-transcription/runner.py @@ -23,10 +23,11 @@ import argparse import asyncio import logging -from contextlib import suppress +from contextlib import asynccontextmanager, suppress import numpy as np -from aiohttp import web +import uvicorn +from fastapi import FastAPI, WebSocket, WebSocketDisconnect from livepeer_gateway.live_runner import register_runner @@ -94,9 +95,8 @@ def _transcribe(pcm: bytes) -> str: log = logging.getLogger("realtime-transcription") -async def _handle_transcribe(request: web.Request) -> web.WebSocketResponse: - ws = web.WebSocketResponse(heartbeat=20) - await ws.prepare(request) +async def _transcribe_socket(ws: WebSocket) -> None: + await ws.accept() log.info("transcription socket opened") seg = bytearray() # current utterance PCM; the worker trims finalized audio @@ -139,12 +139,17 @@ async def _worker() -> None: worker = asyncio.create_task(_worker()) try: - async for msg in ws: - if msg.type == web.WSMsgType.BINARY: - seg.extend(msg.data) - if _rms(msg.data) >= SILENCE_RMS: + while True: + message = await ws.receive() + if message["type"] == "websocket.disconnect": + break + if (data := message.get("bytes")) is not None: + seg.extend(data) + if _rms(data) >= SILENCE_RMS: spoke = True - elif msg.type == web.WSMsgType.TEXT and msg.data.strip() == "eos": + elif (text_in := message.get("text")) is not None: + if text_in.strip() != "eos": + continue # Stop the worker first, or a partial it is mid-way through can # land after the closing final and read as the last transcript. worker.cancel() @@ -154,15 +159,15 @@ async def _worker() -> None: if text: await ws.send_json(_message(text, len(seg), final=True)) break - elif msg.type == web.WSMsgType.ERROR: - log.warning("ws error: %s", ws.exception()) - break + except WebSocketDisconnect: + pass finally: worker.cancel() with suppress(asyncio.CancelledError, Exception): await worker + with suppress(Exception): + await ws.close() log.info("transcription socket closed") - return ws def _parse_args() -> argparse.Namespace: @@ -184,15 +189,10 @@ def _parse_args() -> argparse.Namespace: return parser.parse_args() -def main() -> None: - logging.basicConfig( - level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" - ) - args = _parse_args() - _load_model() # fail fast if the model or the GPU is missing - - async def _on_startup(app: web.Application) -> None: - app["registration"] = await register_runner( # Livepeer: 1 +def build_app(args: argparse.Namespace) -> FastAPI: + @asynccontextmanager + async def _lifespan(app: FastAPI): + app.state.registration = await register_runner( # Livepeer: 1 args.orchestrator, secret=args.orchSecret, runner_url=args.runner_url, @@ -201,18 +201,28 @@ async def _on_startup(app: web.Application) -> None: price=args.price, # decimal USD/hour ) log.info( - "registered runner_id=%s app=%s", app["registration"].runner_id, APP_ID + "registered runner_id=%s app=%s", app.state.registration.runner_id, APP_ID ) - - async def _on_cleanup(app: web.Application) -> None: + yield with suppress(Exception): - await app["registration"].close() # Livepeer: 2 + await app.state.registration.close() # Livepeer: 2 + + app = FastAPI(title=APP_ID, version="0.1.0", lifespan=_lifespan) + + @app.websocket("/transcribe") + async def transcribe(websocket: WebSocket) -> None: + await _transcribe_socket(websocket) - app = web.Application() - app.router.add_get("/transcribe", _handle_transcribe) - app.on_startup.append(_on_startup) - app.on_cleanup.append(_on_cleanup) - web.run_app(app, host=args.host, port=DEFAULT_PORT) + return app + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + _load_model() # fail fast if the model or the GPU is missing + uvicorn.run(build_app(args), host=args.host, port=DEFAULT_PORT, access_log=False) if __name__ == "__main__": diff --git a/tiles/Dockerfile b/tiles/Dockerfile index 09ffa97..76a634d 100644 --- a/tiles/Dockerfile +++ b/tiles/Dockerfile @@ -11,6 +11,7 @@ RUN apt-get update \ # livepeer-gateway SDK isn't on PyPI yet; install from Git. # opencv-python-headless needs no system GUI libs. RUN pip install --no-cache-dir \ + fastapi uvicorn \ "livepeer-gateway @ git+https://github.com/livepeer/livepeer-python-gateway@main" \ opencv-python-headless numpy aiohttp diff --git a/tiles/pyproject.toml b/tiles/pyproject.toml index 8eb1caf..2d242c1 100644 --- a/tiles/pyproject.toml +++ b/tiles/pyproject.toml @@ -4,6 +4,8 @@ version = "0.1.0" description = "Tiles (capacity fan-out) example app for the Livepeer network." requires-python = ">=3.12" dependencies = [ + "fastapi", # runner: routes + generated OpenAPI schema + "uvicorn", # runner: ASGI server "opencv-python-headless", # runner: stylize tiles; client: split/stitch "numpy", "aiohttp", diff --git a/tiles/runner.py b/tiles/runner.py index 0ff4b23..e87d96a 100644 --- a/tiles/runner.py +++ b/tiles/runner.py @@ -11,6 +11,7 @@ 2. registration.close() — deregister (cleanup) /tile is an ordinary HTTP handler; its CPU work runs in a thread (parallel tiles). +FastAPI derives the schema from the models below and serves it at /openapi.json. """ from __future__ import annotations @@ -19,20 +20,19 @@ import asyncio import base64 import logging -from contextlib import suppress +from contextlib import asynccontextmanager, suppress import cv2 import numpy as np -from aiohttp import web +import uvicorn +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel, Field from livepeer_gateway.live_runner import register_runner DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8989 APP_ID = "livepeer-example/tiles" -# Tiles are base64 PNGs in JSON; a photographic tile can exceed aiohttp's 1 MB default. -MAX_REQUEST_BYTES = 32 * 1024 * 1024 - log = logging.getLogger("tiles") @@ -94,33 +94,18 @@ def _process(png: bytes, work: int) -> bytes: return out.tobytes() -async def _handle_tile(request: web.Request) -> web.Response: - try: - payload = await request.json() - except Exception: - payload = None - if not isinstance(payload, dict): - raise web.HTTPBadRequest(text="body must be a JSON object") - b64 = payload.get("tile") - if not isinstance(b64, str) or not b64: - raise web.HTTPBadRequest(text="missing 'tile' (base64 PNG)") - # Offload the CPU-bound work to a thread so `capacity` concurrent tiles run in - # parallel (cv2 releases the GIL); on the event loop they would serialize. - loop = asyncio.get_running_loop() - out = await loop.run_in_executor( - None, _process, base64.b64decode(b64), request.app["work"] - ) - return web.json_response({"tile": base64.b64encode(out).decode()}) +class TileRequest(BaseModel): + tile: str = Field(..., min_length=1, description="Base64-encoded PNG tile.") -def main() -> None: - logging.basicConfig( - level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" - ) - args = _parse_args() +class TileResponse(BaseModel): + tile: str = Field(..., description="Base64-encoded stylized PNG tile.") - async def _on_startup(app: web.Application) -> None: - app["registration"] = await register_runner( # Livepeer: 1 + +def build_app(args: argparse.Namespace) -> FastAPI: + @asynccontextmanager + async def _lifespan(app: FastAPI): + app.state.registration = await register_runner( # Livepeer: 1 args.orchestrator, secret=args.orchSecret, runner_url=args.runner_url, @@ -133,21 +118,38 @@ async def _on_startup(app: web.Application) -> None: ) log.info( "registered runner_id=%s capacity=%d orchestrator=%s", - app["registration"].runner_id, + app.state.registration.runner_id, args.capacity, - app["registration"].orchestrator_url, + app.state.registration.orchestrator_url, ) - - async def _on_cleanup(app: web.Application) -> None: + yield with suppress(Exception): - await app["registration"].close() # Livepeer: 2 - - app = web.Application(client_max_size=MAX_REQUEST_BYTES) - app["work"] = args.work - app.router.add_post("/tile", _handle_tile) - app.on_startup.append(_on_startup) - app.on_cleanup.append(_on_cleanup) - web.run_app(app, host=args.host, port=DEFAULT_PORT) + await app.state.registration.close() # Livepeer: 2 + + app = FastAPI(title=APP_ID, version="0.1.0", lifespan=_lifespan) + + @app.post("/tile", response_model=TileResponse) + async def tile(body: TileRequest) -> TileResponse: + # Offload the CPU-bound work to a thread so `capacity` concurrent tiles run in + # parallel (cv2 releases the GIL); on the event loop they would serialize. + loop = asyncio.get_running_loop() + try: + out = await loop.run_in_executor( + None, _process, base64.b64decode(body.tile), args.work + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return TileResponse(tile=base64.b64encode(out).decode()) + + return app + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + uvicorn.run(build_app(args), host=args.host, port=DEFAULT_PORT, access_log=False) if __name__ == "__main__":