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/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__": 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__":