diff --git a/.github/workflows/images.yml b/.github/workflows/images.yml index 84f4708..598313d 100644 --- a/.github/workflows/images.yml +++ b/.github/workflows/images.yml @@ -21,6 +21,7 @@ on: - "hello-world/**" - "echo/**" - "tiles/**" + - "notepad/**" - "realtime-transcription/**" - ".github/workflows/images.yml" pull_request: @@ -41,7 +42,7 @@ jobs: strategy: fail-fast: false matrix: - example: [hello-world, echo, tiles, realtime-transcription] + example: [hello-world, echo, tiles, notepad, realtime-transcription] steps: - uses: actions/checkout@v7 diff --git a/README.md b/README.md index 3726779..c1c395d 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ flowchart LR The orchestrator is a **transparent reverse proxy**: every endpoint you expose is passed through to your app unchanged, so you write an ordinary service and it runs on the network as-is. The transports supported today: -- **HTTP** request/response — the common case. (`hello-world`, `tiles`, `api-proxy`) +- **HTTP** request/response — the common case. (`hello-world`, `tiles`, `api-proxy`, `notepad`) - **HTTP + SSE** — streamed / token responses. (`vllm`) - **Trickle** — continuous realtime video in/out. (`echo`) - **WebSocket** — long-lived bidirectional sessions. (`realtime-transcription`) @@ -51,6 +51,7 @@ Need a transport that isn't here? [Open an issue](https://github.com/livepeer/ru | [`tiles`](./tiles) | Capacity fan-out — one call per tile | dynamic | single-shot | HTTP (base64 PNG) | fixed | | [`api-proxy`](./api-proxy) | Pass calls through to hosted APIs — the operator holds the key, one capability per model | static | single-shot | HTTP (JPEG bytes) | fixed | | [`echo`](./echo) | Realtime video, transformed and echoed back | dynamic | persistent | trickle | hour | +| [`notepad`](./notepad) | Held-open HTTP session with process-local state | dynamic | persistent | HTTP (JSON) | hour | | [`vllm`](./vllm) | Drop-in OpenAI API; the client stays unmodified | static | single-shot | HTTP + SSE | hour | | [`realtime-transcription`](./realtime-transcription) | Audio up, transcripts back, on one socket | dynamic | persistent | WebSocket | hour | @@ -62,7 +63,7 @@ This set stays **minimal and curated**: it covers each value of the axes above ( How the app attaches to the orchestrator: -- **Dynamic** — the app self-registers via the SDK (`register_runner`) and heartbeats; the orchestrator drops it when heartbeats stop. Best for apps that come and go. (`hello-world`, `echo`, `realtime-transcription`) +- **Dynamic** — the app self-registers via the SDK (`register_runner`) and heartbeats; the orchestrator drops it when heartbeats stop. Best for apps that come and go. (`hello-world`, `echo`, `notepad`, `realtime-transcription`) - **Static** — the orchestrator is configured with the app's URL in a `runners.json` and health-polls it; the app needs no SDK. Best for fixed, long-running deployments. (`vllm`, `api-proxy`) The arrow flips: dynamic, the app announces itself; static, the orchestrator is told about a passive app: @@ -92,7 +93,7 @@ Clients read it off the discovered runner, whose `raw` holds that runner's disco Chosen _at_ registration (above); **defaults to `persistent`**, set on both `register_runner(...)` and in `runners.json`. The examples set it explicitly. -- **Persistent** — a held-open session the client reserves and releases, billed per second of wall-clock (or once, with fixed pricing). Best for realtime / streaming. (`echo`, `realtime-transcription`) +- **Persistent** — a held-open session the client reserves and releases, billed per second of wall-clock (or once, with fixed pricing). Best for realtime / streaming. (`echo`, `notepad`, `realtime-transcription`) - **Single-shot** — one request in, one response out; the orchestrator reserves a session per call and releases it when the response returns, so the client manages no session at all. Best for batch / request-response. With metered pricing the call pays for as long as it runs, so the work need not be short. (`hello-world`, `tiles`, `api-proxy`, `vllm`) ## Calling your app @@ -100,7 +101,7 @@ Chosen _at_ registration (above); **defaults to `persistent`**, set on both `reg The client side depends on the runner's mode: - **Single-shot** — **discover → call**: find the app via `runner_selector`, then one `call_runner`. The orchestrator reserves a session for the call and releases it when the response returns; on the paid path `call_runner` answers the 402 payment challenge inline. (`hello-world`, `tiles`, `api-proxy`, `vllm`) -- **Persistent** — **discover → reserve → call → release**: reserve a session (`reserve_session`), call it — `call_runner`, streamed frames, or a WebSocket, depending on transport — then release it (`stop_runner_session`), which settles payment on-chain. (`echo`, `realtime-transcription`) +- **Persistent** — **discover → reserve → call → release**: reserve a session (`reserve_session`), call it — `call_runner`, streamed frames, or a WebSocket, depending on transport — then release it (`stop_runner_session`), which settles payment on-chain. (`echo`, `notepad`, `realtime-transcription`) Each example's `client.py` shows its exact calls: grep `# Livepeer:` to find them. diff --git a/notepad/.env.example b/notepad/.env.example new file mode 100644 index 0000000..8c588bc --- /dev/null +++ b/notepad/.env.example @@ -0,0 +1,25 @@ +# Copy to .env (gitignored) and fill in. Never commit secrets. +# Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only. + +NETWORK=arbitrum-one-mainnet +ETH_RPC_URL=https://arb1.arbitrum.io/rpc + +# Signer (payer): needs an on-chain deposit + reserve. +SIGNER_KEYSTORE_DIR=/absolute/path/to/signer-keystore +SIGNER_ETH_ACCT=0xYourSignerAddress +SIGNER_ETH_PASSWORD=your-signer-keystore-password + +# Orchestrator operating key (split-key): needs ETH for gas to redeem tickets. +ORCH_KEYSTORE_DIR=/absolute/path/to/operator-keystore +ORCH_ETH_ACCT=0xYourOperatorAddress +ORCH_ETH_PASSWORD=your-operator-keystore-password +# Registered orch = ticket recipient (-ethOrchAddr); empty = use the operating key. +ORCH_ONCHAIN_ADDR=0xYourRegisteredOrchestrator + +# Runner price (on-chain): USD per hour, metered per second while the client +# holds the session. Keep under ~0.67: the signer signs at most 100 tickets +# per payment, and the demo orchestrator runs -ticketEV=1e10 (fee / ticketEV). +PRICE=0.10 +# Signer's max-price cap (payer side), per billing unit. Metered here, so the +# unit is one second and must exceed PRICE / 3600 (0.000111USD is ~0.40/hour). +MAX_PRICE_PER_UNIT=0.000111USD diff --git a/notepad/Dockerfile b/notepad/Dockerfile new file mode 100644 index 0000000..fcfe15d --- /dev/null +++ b/notepad/Dockerfile @@ -0,0 +1,15 @@ +# Notepad example app (persistent HTTP). +FROM python:3.12-slim + +# Flush stdout/stderr immediately so output isn't block-buffered in `docker logs`. +ENV PYTHONUNBUFFERED=1 + +RUN pip install --no-cache-dir \ + "livepeer-gateway>=1.0.0" + +WORKDIR /app +COPY runner.py client.py ./ + +EXPOSE 8989 + +ENTRYPOINT ["python", "runner.py"] diff --git a/notepad/README.md b/notepad/README.md new file mode 100644 index 0000000..48b1870 --- /dev/null +++ b/notepad/README.md @@ -0,0 +1,66 @@ +# Notepad app (persistent HTTP) + +A held-open **HTTP** session that keeps one string in memory. `POST /set` writes it; `POST /get` reads it back. This is the missing cell in the examples table: **persistent + HTTP**, with no WebSocket and no trickle. It is also the regression target for Console `run_capability` with `endpoint` — and the reason that tool cannot reuse session state: each `runInference` reserves, calls once, and stops. + +| | | +| ------------ | --------------------------------------------- | +| App id | `livepeer-example/notepad` | +| Runner mode | persistent (held-open HTTP session) | +| Registration | dynamic (self-registers via the SDK) | +| Transport | HTTP (JSON request/response, two round-trips) | +| Pricing | hour (metered while the session is held) | +| Port | 8989 | + +Prerequisites (Docker, `uv`, and the [`livepeer-gateway` SDK](https://pypi.org/project/livepeer-gateway/)) and the shared on-chain/payment setup live in the [repo README](../README.md). + +## How it's wired + +The app is **dynamically registered** with `mode="persistent"` ([runner.py](runner.py)). Process-local `_note` **is** session state, because the orchestrator pins this runner to the reserved session. The client calls it with `reserve_session` → `POST /set` → `POST /get` → `stop_runner_session` ([client.py](client.py)). Grep `# Livepeer:` in either file to see the exact calls. + +`run_capability` on Console (gateway-web `runInference`) can hit `/set` **or** `/get` with `endpoint`, but not both on the same session. Use this client when you need the two-call proof. + +## Run offchain (free) + +> [!TIP] +> Built locally by the compose file below, or run the published [`ghcr.io/livepeer/runner-example-notepad`](https://github.com/livepeer/runner-app-examples/pkgs/container/runner-example-notepad) with `docker compose up -d --pull always` — see [Images](../README.md#images). + +```sh +docker compose up -d --build +curl -sk https://localhost:8935/discovery | jq '.[].runners[] | {app, mode}' +uv run client.py --text "hello from a held session" +# {'set': {'text': 'hello from a held session', 'revision': 1}, 'get': {'text': 'hello from a held session', 'revision': 1}} +docker compose down +``` + +## Run on-chain (paid) + +```sh +cp .env.example .env # fill in RPC, network, keystore paths, accounts, pricing +docker compose -f compose.yml -f compose.onchain.yml up -d --build +uv run client.py --text "hello" \ + --discovery https://localhost:8935/discovery \ + --signer http://localhost:7936 +docker compose -f compose.yml -f compose.onchain.yml down +``` + +The session is **metered** (`unit` defaults to `hour`) for as long as the client holds it. Keep the demo short. + +## Run without Docker + +```sh +./livepeer -orchestrator -useLiveRunners -serviceAddr localhost:8935 -orchSecret abcdef -v 6 +uv run runner.py --orchestrator https://localhost:8935 --orchSecret abcdef +uv run client.py --text "hello from a held session" +``` + +## Console MCP + +``` +run_capability({ + capability: "livepeer-example/notepad", + endpoint: "/set", + inputs: { text: "hello" } +}) +``` + +That pays a full reserve for one POST and then stops. A second `run_capability` to `/get` is a **new** session and returns empty text. That is expected. diff --git a/notepad/client.py b/notepad/client.py new file mode 100644 index 0000000..d3415ef --- /dev/null +++ b/notepad/client.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +"""notepad client: reserve a session, write then read, settle up. + +Proves persistent HTTP: two POSTs against the same reserved session share +process-local state. gateway-web `runInference` cannot do this — it reserves, +calls once, and stops. + +Livepeer integration (grep `# Livepeer:`): + 1. reserve_session() — discover the runner, reserve a session + 2. post_json(app_url/set) — write + 3. post_json(app_url/get) — read (same session) + 4. stop_runner_session() — end the session +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +from contextlib import suppress + +from livepeer_gateway.errors import LivepeerGatewayError +from livepeer_gateway.http import post_json +from livepeer_gateway.live_runner import stop_runner_session +from livepeer_gateway.selection import reserve_session + +DEFAULT_DISCOVERY = "https://localhost:8935/discovery" +APP_ID = "livepeer-example/notepad" + +log = logging.getLogger("notepad-client") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Write then read a notepad Live Runner session." + ) + parser.add_argument("--discovery", default=DEFAULT_DISCOVERY) + parser.add_argument("--text", default="hello from a held session") + parser.add_argument( + "--signer", default="", help="Remote signer base URL (on-chain/paid path)." + ) + return parser.parse_args() + + +async def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + session = None + try: + session = await reserve_session( # Livepeer: 1 + discovery_url=args.discovery, + app=APP_ID, + signer_url=args.signer.strip() or None, + ) + log.info("session_id=%s app_url=%s", session.session_id, session.app_url) + async with session: + app_url = session.app_url.rstrip("/") + written = await post_json( # Livepeer: 2 + f"{app_url}/set", + {"text": args.text}, + ) + read = await post_json(f"{app_url}/get", {}) # Livepeer: 3 + print({"set": written, "get": read}) + if read.get("text") != args.text: + raise SystemExit(f"ERROR: session did not keep state: {read}") + except LivepeerGatewayError as exc: + raise SystemExit(f"ERROR: {exc}") from exc + finally: + if session is not None: + with suppress(Exception): + await stop_runner_session(session) # Livepeer: 4 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/notepad/compose.onchain.yml b/notepad/compose.onchain.yml new file mode 100644 index 0000000..d916a7b --- /dev/null +++ b/notepad/compose.onchain.yml @@ -0,0 +1,34 @@ +# On-chain payment overlay for notepad. Layer it on the offchain base: +# docker compose -f compose.yml -f compose.onchain.yml up -d --build +# +# Adds the shared remote signer, re-points the orchestrator on-chain (see +# ../compose.onchain.yml), and registers the app with a price so the orchestrator +# issues a payment challenge. Requires a local .env (gitignored); copy .env.example +# and fill it in. The session is metered, so it is billed for as long as the +# client holds it. Then pay through the signer: +# uv run client.py --text "hello" \ +# --discovery https://localhost:8935/discovery \ +# --signer http://localhost:7936 + +services: + signer: + extends: + file: ../compose.onchain.yml + service: signer + ports: + - "7936:7936" + + orchestrator: + extends: + file: ../compose.onchain.yml + service: orchestrator + + # Re-declare the command to advertise a price (base file registers free). + app: + command: + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:8989 + # Billed per second of session (metered); price cap in .env.example. + - --price=${PRICE} diff --git a/notepad/compose.yml b/notepad/compose.yml new file mode 100644 index 0000000..3c9a2e1 --- /dev/null +++ b/notepad/compose.yml @@ -0,0 +1,29 @@ +# End-to-end offchain demo: orchestrator + persistent-HTTP notepad. +# +# The orchestrator service is defined once in ../compose.orchestrator.yml and +# pulled in with `extends`; this file only adds the app. Once up, call it from +# the host with the SDK: +# docker compose up -d --build +# uv run client.py --text "hello from a held session" + +services: + orchestrator: + extends: + file: ../compose.orchestrator.yml + service: orchestrator + + app: + # `up` always builds; `--pull always` runs the published image instead. + image: ghcr.io/livepeer/runner-example-notepad:latest + pull_policy: build + build: . + container_name: example_apps_notepad + # Wait for the orchestrator's healthcheck so registration doesn't race its boot. + depends_on: + orchestrator: + condition: service_healthy + command: + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:8989 diff --git a/notepad/pyproject.toml b/notepad/pyproject.toml new file mode 100644 index 0000000..199fffc --- /dev/null +++ b/notepad/pyproject.toml @@ -0,0 +1,8 @@ +[project] +name = "livepeer-notepad" +version = "0.1.0" +description = "Persistent HTTP notepad example app for the Livepeer network." +requires-python = ">=3.12" +dependencies = [ + "livepeer-gateway>=1.0.0", +] diff --git a/notepad/runner.py b/notepad/runner.py new file mode 100644 index 0000000..9c35fc4 --- /dev/null +++ b/notepad/runner.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""notepad app: persistent HTTP with per-session state, no streaming transport. + +A held-open session keeps one string in memory. POST /set writes it, POST /get +reads it. That is the gap the other examples leave: persistent + ordinary HTTP. + +Livepeer integration (grep `# Livepeer:`): + 1. register_runner() — announce the app (mode=persistent) + 2. registration.close() — deregister (cleanup) + +/set and /get are ordinary HTTP handlers. The orchestrator pins this process to +the reserved session, so process-local state *is* session state. +""" + +from __future__ import annotations + +import argparse +import logging +from contextlib import suppress + +from aiohttp import web + +from livepeer_gateway.live_runner import register_runner + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 8989 +APP_ID = "livepeer-example/notepad" + +log = logging.getLogger("notepad") + +_note = "" +_revision = 0 + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Live Runner persistent-HTTP notepad demo." + ) + parser.add_argument("--orchestrator", default="https://localhost:8935") + parser.add_argument("--orchSecret", default="abcdef") + parser.add_argument("--runner-url", default=f"http://{DEFAULT_HOST}:{DEFAULT_PORT}") + parser.add_argument( + "--host", default=DEFAULT_HOST, help="Bind address (use 0.0.0.0 in containers)." + ) + parser.add_argument( + "--price", + type=float, + default=0, + help="Runner price in USD per hour (0 = free, the offchain default).", + ) + return parser.parse_args() + + +async def _handle_set(request: web.Request) -> web.Response: + global _note, _revision + try: + payload = await request.json() + except Exception: + payload = None + if not isinstance(payload, dict): + raise web.HTTPBadRequest(text="body must be a JSON object") + _note = str(payload.get("text", "")) + _revision += 1 + return web.json_response({"text": _note, "revision": _revision}) + + +async def _handle_get(_request: web.Request) -> web.Response: + return web.json_response({"text": _note, "revision": _revision}) + + +def main() -> None: + logging.basicConfig( + 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=APP_ID, + mode="persistent", + price=args.price, # USD per hour, metered while the session is held + ) + log.info( + "registered runner_id=%s orchestrator=%s", + app["registration"].runner_id, + app["registration"].orchestrator_url, + ) + + async def _on_cleanup(app: web.Application) -> None: + with suppress(Exception): + await app["registration"].close() # Livepeer: 2 + + app = web.Application() + app.router.add_post("/set", _handle_set) + app.router.add_post("/get", _handle_get) + app.on_startup.append(_on_startup) + app.on_cleanup.append(_on_cleanup) + web.run_app(app, host=args.host, port=DEFAULT_PORT) + + +if __name__ == "__main__": + main() diff --git a/realtime-transcription/README.md b/realtime-transcription/README.md index 349034e..7fadfa1 100644 --- a/realtime-transcription/README.md +++ b/realtime-transcription/README.md @@ -95,3 +95,13 @@ Start an orchestrator built from go-livepeer `v0.9.1` or newer (see [Build from uv run runner.py --orchestrator https://localhost:8935 --orchSecret abcdef uv run client.py sample.wav ``` + +## Session handoff (spike) + +[`handoff_client.py`](handoff_client.py) is a proof-of-concept for the Console → client streaming design: it does **not** call `reserve_session`. It reads a JSON envelope (`session_id`, `app_url`, `control_url`, `endpoint`, payment snapshot, `signer_url`, `signer_token`), runs the 3-second funding loop itself, opens `wss://` from `app_url`, and `POST`s `control_url/stop` on the way out. + +This is not a production MCP tool. A signer JWT that can call `generate-live-payment` is unscoped today — see `gateway-web/docs/stream-session-handoff.md`. Use it only against a local envelope you minted yourself: + +```sh +uv run handoff_client.py envelope.json sample.wav +``` diff --git a/realtime-transcription/handoff_client.py b/realtime-transcription/handoff_client.py new file mode 100644 index 0000000..32c0a91 --- /dev/null +++ b/realtime-transcription/handoff_client.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 +"""Consume a Console session-handoff envelope and stream to /transcribe. + +Spike / proof-of-concept. Console would reserve with startFunding=false, return +session URLs plus a LivePaymentSession snapshot plus a signer JWT. This client +drives the WebSocket and the 3-second funding loop. Do not treat a signer JWT +as production-safe until it is scoped to one manifest — see +gateway-web/docs/stream-session-handoff.md. + +Envelope (JSON file or stdin): + + { + "session_id": "…", + "app_url": "https://orch/…/app", + "control_url": "https://orch/…/control/…", + "endpoint": "/transcribe", + "payment": { + "type": "live", + "challenge": { + "paymentParams": "…", + "manifestId": "…", + "paymentUrl": "https://orch/…/pay" + }, + "app": "livepeer-example/realtime-transcription", + "maxPrice": {"price": 0.1, "currency": "usd", "unit": "hour"}, + "state": {"n": 1} + }, + "signer_url": "https://signer…", + "signer_token": "eyJ…" + } + +Usage: + uv run handoff_client.py envelope.json sample.wav + uv run handoff_client.py envelope.json - +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import ssl +import sys +from contextlib import suppress +from pathlib import Path +from typing import Any + +import aiohttp + +from client import _read_pcm, _recv, _send + +PAYMENT_INTERVAL_S = 3.0 + +log = logging.getLogger("handoff-client") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Stream audio using a reserved session handoff envelope." + ) + parser.add_argument( + "envelope", + help="Path to the handoff JSON, or - to read it from stdin.", + ) + parser.add_argument( + "input", + help=( + "16 kHz mono WAV to stream, or - to read raw PCM from stdin " + "(cannot combine with envelope on stdin)." + ), + ) + return parser.parse_args() + + +def _load_envelope(path: str) -> dict[str, Any]: + if path.strip() == "-": + raw = sys.stdin.read() + else: + raw = Path(path).expanduser().read_text(encoding="utf-8") + data = json.loads(raw) + if not isinstance(data, dict): + raise SystemExit("ERROR: envelope must be a JSON object") + return data + + +def _challenge_field(challenge: dict[str, Any], camel: str, snake: str) -> str: + value = challenge.get(camel, challenge.get(snake, "")) + return value.strip() if isinstance(value, str) else "" + + +class PaymentLoop: + """Rehydrate a gateway-web LivePaymentSession snapshot over HTTP.""" + + def __init__( + self, envelope: dict[str, Any], session: aiohttp.ClientSession + ) -> None: + payment = envelope.get("payment") + if not isinstance(payment, dict): + raise SystemExit("ERROR: envelope missing payment snapshot") + challenge = payment.get("challenge") + if not isinstance(challenge, dict): + raise SystemExit("ERROR: envelope.payment.challenge must be an object") + signer_url = str(envelope.get("signer_url") or "").rstrip("/") + token = str(envelope.get("signer_token") or "").strip() + if not signer_url or not token: + raise SystemExit("ERROR: envelope needs signer_url and signer_token") + self._http = session + self._signer_url = signer_url + self._headers = {"Authorization": f"Bearer {token}"} + self._type = str(payment.get("type") or "live") + self._app = payment.get("app") + self._max_price = payment.get("maxPrice", payment.get("max_price")) + self._state = payment.get("state") + self._payment_params = _challenge_field( + challenge, "paymentParams", "payment_params" + ) + self._manifest_id = _challenge_field(challenge, "manifestId", "manifest_id") + self._payment_url = _challenge_field(challenge, "paymentUrl", "payment_url") + if not self._payment_params or not self._manifest_id or not self._payment_url: + raise SystemExit( + "ERROR: challenge missing paymentParams/manifestId/paymentUrl" + ) + + async def run(self, stop: asyncio.Event) -> None: + # First interval payment waits — reserve already paid the 402. + while not stop.is_set(): + try: + await asyncio.wait_for(stop.wait(), timeout=PAYMENT_INTERVAL_S) + return + except TimeoutError: + pass + try: + await self._send_payment() + except Exception: + log.exception("funding cycle failed") + + async def _send_payment(self) -> None: + payload: dict[str, Any] = { + "orchestrator": self._payment_params, + "type": self._type, + "ManifestID": self._manifest_id, + } + if self._app: + payload["app"] = self._app + if self._max_price is not None: + payload["maxPrice"] = self._max_price + if self._state is not None: + payload["state"] = self._state + async with self._http.post( + f"{self._signer_url}/generate-live-payment", + json=payload, + headers=self._headers, + ) as resp: + body = await resp.json(content_type=None) + if resp.status != 200: + raise RuntimeError(f"generate-live-payment HTTP {resp.status}: {body}") + payment = body.get("payment") + seg = body.get("segCreds") + state = body.get("state") + if not isinstance(payment, str) or not payment: + raise RuntimeError("generate-live-payment missing payment") + if not isinstance(seg, str) or not seg: + raise RuntimeError("generate-live-payment missing segCreds") + if isinstance(state, dict): + self._state = state + headers = {"Livepeer-Payment": payment, "Livepeer-Segment": seg} + async with self._http.post( + self._payment_url, headers=headers, ssl=_orch_ssl() + ) as pay: + if pay.status >= 400: + text = await pay.text() + raise RuntimeError(f"payment POST HTTP {pay.status}: {text[:200]}") + + +def _orch_ssl() -> ssl.SSLContext | bool: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + return ctx + + +def _ws_url(app_url: str, endpoint: str) -> str: + base = app_url.replace("https://", "wss://").replace("http://", "ws://").rstrip("/") + path = endpoint if endpoint.startswith("/") else f"/{endpoint}" + return base + path + + +async def _stop_session(http: aiohttp.ClientSession, control_url: str) -> None: + url = control_url.rstrip("/") + "/stop" + with suppress(Exception): + async with http.post(url, ssl=_orch_ssl()) as resp: + log.info("stop %s -> %s", url, resp.status) + + +async def main() -> None: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + args = _parse_args() + if args.envelope.strip() == "-" and args.input.strip() == "-": + raise SystemExit("ERROR: envelope and audio cannot both be stdin") + envelope = _load_envelope(args.envelope) + app_url = str(envelope.get("app_url") or "").strip() + control_url = str(envelope.get("control_url") or "").strip() + endpoint = str(envelope.get("endpoint") or "/transcribe").strip() + if not app_url or not control_url: + raise SystemExit("ERROR: envelope needs app_url and control_url") + + live = args.input.strip() == "-" + pcm = b"" if live else _read_pcm(str(Path(args.input).expanduser())) + ws_url = _ws_url(app_url, endpoint) + log.info("session_id=%s ws=%s", envelope.get("session_id"), ws_url) + + stop = asyncio.Event() + ssl_ctx: ssl.SSLContext | bool = ( + _orch_ssl() if ws_url.startswith("wss://") else True + ) + async with aiohttp.ClientSession() as http: + funding = PaymentLoop(envelope, http) + fund_task = asyncio.create_task(funding.run(stop)) + try: + async with http.ws_connect(ws_url, ssl=ssl_ctx, heartbeat=20) as ws: + await asyncio.gather(_send(ws, pcm, live=live), _recv(ws)) + finally: + stop.set() + fund_task.cancel() + with suppress(asyncio.CancelledError): + await fund_task + await _stop_session(http, control_url) + + +if __name__ == "__main__": + asyncio.run(main())