From 4b858f38d67d27f265b646508b4b7ace7896bae7 Mon Sep 17 00:00:00 2001 From: Elite Encoder Date: Tue, 18 Aug 2026 13:14:46 -0400 Subject: [PATCH] feat(comfystream): overlay the published image with Livepeer integration Keep ComfyStream as a pulled package/image and move register_runner, trickle, and the smoke client into this example so the app can be redeployed without forking the ComfyStream repo. --- README.md | 9 +- comfystream/.env.example | 33 ++ comfystream/Dockerfile | 23 + comfystream/README.md | 98 ++++ comfystream/client.py | 195 +++++++ comfystream/compose.existing.yml | 34 ++ comfystream/compose.onchain.yml | 33 ++ comfystream/compose.yml | 36 ++ comfystream/entrypoint.sh | 10 + comfystream/pyproject.toml | 10 + comfystream/runner.py | 552 ++++++++++++++++++++ comfystream/workflows/analyze-stub-api.json | 15 + 12 files changed, 1044 insertions(+), 4 deletions(-) create mode 100644 comfystream/.env.example create mode 100644 comfystream/Dockerfile create mode 100644 comfystream/README.md create mode 100644 comfystream/client.py create mode 100644 comfystream/compose.existing.yml create mode 100644 comfystream/compose.onchain.yml create mode 100644 comfystream/compose.yml create mode 100644 comfystream/entrypoint.sh create mode 100644 comfystream/pyproject.toml create mode 100644 comfystream/runner.py create mode 100644 comfystream/workflows/analyze-stub-api.json diff --git a/README.md b/README.md index 981e0c3..72ca82b 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ The orchestrator is a **transparent reverse proxy**: every endpoint you expose i - **HTTP** request/response — the common case. (`hello-world`, `tiles`, `api-proxy`) - **HTTP + SSE** — streamed / token responses. (`vllm`) -- **Trickle** — continuous realtime video in/out. (`echo`) +- **Trickle** — continuous realtime video in/out. (`echo`, `comfystream`) - **WebSocket** — long-lived bidirectional sessions. (`realtime-transcription`) Need a transport that isn't here? [Open an issue](https://github.com/livepeer/runner-app-examples/issues). @@ -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 | +| [`comfystream`](./comfystream) | ComfyUI live video + analyze; consumes the published `livepeer/comfystream` image | dynamic | persistent | trickle + HTTP | 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`, `comfystream`, `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`, `comfystream`, `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`, `comfystream`, `realtime-transcription`) Each example's `client.py` shows its exact calls: grep `# Livepeer:` to find them. diff --git a/comfystream/.env.example b/comfystream/.env.example new file mode 100644 index 0000000..a6c1d30 --- /dev/null +++ b/comfystream/.env.example @@ -0,0 +1,33 @@ +# 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 + +# Optional: pin a GPU and host model/storage dirs when using compose.existing.yml. +# COMFYSTREAM_GPU_UUID=GPU-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx +# COMFYSTREAM_MODELS_DIR=/livepeer/ai/data/models +# COMFYSTREAM_STORAGE_DIR=/livepeer/ai/data +# LIVEPEER_ORCH_URL=https://127.0.0.1:8935 +# LIVEPEER_ORCH_SECRET=abcdef +# LIVEPEER_RUNNER_URL=http://127.0.0.1:8991 diff --git a/comfystream/Dockerfile b/comfystream/Dockerfile new file mode 100644 index 0000000..904fba7 --- /dev/null +++ b/comfystream/Dockerfile @@ -0,0 +1,23 @@ +# Overlay: published ComfyStream image + this folder's live-runner integration. +# The comfystream package and ComfyUI workspace already live in the base image. +ARG BASE_IMAGE=livepeer/comfystream:latest +FROM ${BASE_IMAGE} + +WORKDIR /app + +# livepeer-gateway's generated pb2 needs protobuf >= 6.31.1. ComfyUI pins +# protobuf<5; the live-runner path still runs with protobuf 6. +RUN /bin/bash -lc 'source /workspace/miniconda3/etc/profile.d/conda.sh \ + && conda activate comfystream \ + && pip install --no-cache-dir \ + "protobuf>=6.31.1" \ + "livepeer-gateway>=1.0.0" \ + av' + +COPY runner.py /app/runner.py +COPY entrypoint.sh /app/entrypoint.sh +RUN chmod +x /app/entrypoint.sh + +EXPOSE 8991 + +ENTRYPOINT ["/app/entrypoint.sh"] diff --git a/comfystream/README.md b/comfystream/README.md new file mode 100644 index 0000000..ff17620 --- /dev/null +++ b/comfystream/README.md @@ -0,0 +1,98 @@ +# ComfyStream app (workflow-driven video + analyze) + +A ComfyUI live-video app on the Livepeer network. It consumes the published +[`livepeer/comfystream`](https://hub.docker.com/r/livepeer/comfystream) image — +the `comfystream` package and ComfyUI workspace stay in that image. This folder +is only the Livepeer integration: dynamic registration, trickle channels, and +the analyze / start_stream HTTP surface. + +| | | +| ------------ | ------------------------------------ | +| App id | `livepeer-example/comfystream` | +| Runner mode | persistent (held-open session) | +| Registration | dynamic (self-registers via the SDK) | +| Transport | trickle + HTTP (`/analyze`) | +| Pricing | hour (metered per second) | +| Port | 8991 | + +Prerequisites (Docker, NVIDIA GPU, `uv`, the +[`livepeer-gateway` SDK](https://pypi.org/project/livepeer-gateway/)) and the +shared setup are in the [repo README](../README.md). You also need the +`livepeer/comfystream` image (pulled as the Docker build base). + +## How it's wired + +ComfyStream is **installed, not vendored**. The Dockerfile starts `FROM +livepeer/comfystream:latest` and adds `livepeer-gateway` into that image's conda +env. `runner.py` imports `comfystream.Pipeline` from the package already in the +image and registers with the orchestrator. + +The app is **dynamically registered**: it self-registers via `register_runner` +([runner.py](runner.py)) and exposes: + +- `POST /analyze` — video-in → text-out (trickle `in` + JSONL `text`) +- `POST /start_stream` — live trickle video (optional text) +- `POST /update_stream` — mid-session workflow / resolution change +- `GET /text` — buffered text for the active session +- `GET /healthz` + +The client calls it with `reserve_session` → `post_json` / `MediaPublish` → +`stop_runner_session` ([client.py](client.py)). Grep `# Livepeer:` in either +file to see the exact calls. + +This is the same persistent + trickle shape as [`echo`](../echo), plus an HTTP +analyze surface driven by a ComfyUI API-format workflow JSON. + +## Run offchain (free) + +Start the stack and confirm the runner registered. The first build pulls the +ComfyStream image and can take a while. + +```sh +docker compose up -d --build +curl -sk https://localhost:8935/discovery | jq '.[].runners[].app' # confirm livepeer-example/comfystream registered +``` + +Any video works; this makes a short one if you need it: + +```sh +ffmpeg -f lavfi -i testsrc=size=512x512:rate=30 -t 5 -c:v libx264 -preset ultrafast -pix_fmt yuv420p sample.mp4 +uv run client.py sample.mp4 --discovery https://localhost:8935/discovery +``` + +`client.py` defaults to [workflows/analyze-stub-api.json](workflows/analyze-stub-api.json) +(video-in → a fixed text token). Pass `--workflow path/to/workflow.json` for a +real ComfyUI graph, and `--mode stream` for `start_stream`. + +Stop the stack with `docker compose down`. + +## Attach to an existing orchestrator + +Same image and runner, no local go-livepeer. Point at the orch you already run +(env vars in `.env` / `.env.example`): + +```sh +docker compose -f compose.existing.yml up -d --build +uv run client.py sample.mp4 --discovery "$LIVEPEER_ORCH_URL/discovery" +``` + +## Run on-chain (paid) + +Layer `compose.onchain.yml` to run the orchestrator on-chain with a remote +signer paying for the session. This example uses **metered pricing**: `PRICE` +is USD per hour, billed per second for as long as the client holds the session. +For the required RPC and wallets see +[On-chain (paid) setup](../README.md#on-chain-paid-setup) in the repo README. + +```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 sample.mp4 \ + --discovery https://localhost:8935/discovery \ + --signer http://localhost:7936 +docker compose -f compose.yml -f compose.onchain.yml down +``` + +A metered session pays **more than once**. Use a clip of at least a few tens of +seconds if you want to see repeated payments in +`docker compose logs -f orchestrator`. diff --git a/comfystream/client.py b/comfystream/client.py new file mode 100644 index 0000000..9647788 --- /dev/null +++ b/comfystream/client.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +"""comfystream client: reserve a session, drive analyze / start_stream, release. + +Publishes video frames into the runner's trickle `in` channel. Analyze collects +text via GET /text; start_stream can also swap the workflow mid-session. + +Livepeer integration (grep `# Livepeer:`): + 1. reserve_session() + 2. post_json / MediaPublish through session.app_url (orch injects session headers) + 3. stop_runner_session() +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +from contextlib import suppress +from pathlib import Path +from typing import Any + +import av +from livepeer_gateway.errors import LivepeerGatewayError +from livepeer_gateway.http import get_json, post_json +from livepeer_gateway.live_runner import stop_runner_session +from livepeer_gateway.media_publish import MediaPublish +from livepeer_gateway.selection import reserve_session + +APP_ID = "livepeer-example/comfystream" +DEFAULT_DISCOVERY = "https://localhost:8935/discovery" +DEFAULT_WORKFLOW = ( + Path(__file__).resolve().parent / "workflows" / "analyze-stub-api.json" +) +log = logging.getLogger("comfystream-client") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run the proxied ComfyStream Live Runner demo." + ) + parser.add_argument("input", help="Input video file.") + parser.add_argument("--discovery", default=DEFAULT_DISCOVERY) + parser.add_argument( + "--signer", + default="", + help="Remote signer URL for on-chain path.", + ) + parser.add_argument( + "--workflow", + default=str(DEFAULT_WORKFLOW), + help="ComfyUI API-format workflow JSON (default: analyze stub).", + ) + parser.add_argument( + "--mode", + choices=("analyze", "stream", "both"), + default="analyze", + help="Which surface to exercise (default: analyze).", + ) + parser.add_argument("--max-frames", type=int, default=30) + parser.add_argument("--width", type=int, default=512) + parser.add_argument("--height", type=int, default=512) + parser.add_argument( + "--update-workflow", + default="", + help="Optional second workflow JSON for update_stream.", + ) + return parser.parse_args() + + +def _load_workflow(path: str) -> Any: + return json.loads(Path(path).read_text(encoding="utf-8")) + + +async def _publish_frames( + publish_url: str, + input_path: str, + *, + max_frames: int, +) -> None: + publisher = MediaPublish(publish_url) + try: + container = av.open(input_path) + sent = 0 + for frame in container.decode(video=0): + await publisher.write_frame(frame) + sent += 1 + if max_frames and sent >= max_frames: + break + container.close() + log.info("published %d frames to %s", sent, publish_url) + finally: + await publisher.close() + + +async def _run_analyze(args: argparse.Namespace, workflow: Any) -> None: + session = await reserve_session( # Livepeer: 1 + discovery_url=args.discovery, + app=APP_ID, + signer_url=args.signer.strip() or None, + ) + try: + async with session: + data = await post_json( # Livepeer: 2 + f"{session.app_url.rstrip('/')}/analyze", + { + "prompts": workflow, + "width": args.width, + "height": args.height, + }, + timeout=120.0, + ) + log.info("analyze started: %s", data) + await _publish_frames(data["in"], args.input, max_frames=args.max_frames) + await asyncio.sleep(2.0) + texts = await get_json( + f"{session.app_url.rstrip('/')}/text", + timeout=30.0, + ) + log.info("analyze texts: %s", texts) + except LivepeerGatewayError as exc: + raise SystemExit(f"ERROR: {exc}") from exc + finally: + with suppress(Exception): + await stop_runner_session(session) # Livepeer: 3 + + +async def _run_stream(args: argparse.Namespace, workflow: Any) -> None: + session = await reserve_session( # Livepeer: 1 + discovery_url=args.discovery, + app=APP_ID, + signer_url=args.signer.strip() or None, + ) + try: + async with session: + data = await post_json( # Livepeer: 2 + f"{session.app_url.rstrip('/')}/start_stream", + { + "prompts": workflow, + "width": args.width, + "height": args.height, + }, + timeout=120.0, + ) + log.info("stream started: %s", data) + + publish = asyncio.create_task( + _publish_frames(data["in"], args.input, max_frames=args.max_frames) + ) + if args.update_workflow: + await asyncio.sleep(1.0) + update = _load_workflow(args.update_workflow) + updated = await post_json( + f"{session.app_url.rstrip('/')}/update_stream", + {"prompts": update}, + timeout=60.0, + ) + log.info("update_stream: %s", updated) + await publish + await asyncio.sleep(1.0) + except LivepeerGatewayError as exc: + raise SystemExit(f"ERROR: {exc}") from exc + finally: + with suppress(Exception): + await stop_runner_session(session) # Livepeer: 3 + log.info("stream session stopped") + + +async def _amain() -> int: + args = _parse_args() + input_path = Path(args.input).expanduser() + if not input_path.exists(): + raise SystemExit(f"input file does not exist: {input_path}") + args.input = str(input_path) + workflow = _load_workflow(args.workflow) + if args.mode in ("analyze", "both"): + await _run_analyze(args, workflow) + if args.mode in ("stream", "both"): + await _run_stream(args, workflow) + return 0 + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + ) + try: + raise SystemExit(asyncio.run(_amain())) + except KeyboardInterrupt: + raise SystemExit(130) from None + + +if __name__ == "__main__": + main() diff --git a/comfystream/compose.existing.yml b/comfystream/compose.existing.yml new file mode 100644 index 0000000..fac3b9b --- /dev/null +++ b/comfystream/compose.existing.yml @@ -0,0 +1,34 @@ +# Attach to an already-running orchestrator. Does not start go-livepeer. +# +# docker compose -f compose.existing.yml up -d --build +# +# Override URLs, secret, GPU, and host model/storage dirs via `.env`. + +services: + app: + build: . + container_name: example_apps_comfystream + restart: unless-stopped + network_mode: host + environment: + - NVIDIA_VISIBLE_DEVICES=${COMFYSTREAM_GPU_UUID:-all} + - COMFYUI_CWD=/workspace/ComfyUI + - PYTHONUNBUFFERED=1 + volumes: + - ${COMFYSTREAM_MODELS_DIR:-/livepeer/ai/data/models}:/workspace/ComfyUI/models + - ${COMFYSTREAM_STORAGE_DIR:-/livepeer/ai/data}:/app/storage + command: + - --host=0.0.0.0 + - --orchestrator=${LIVEPEER_ORCH_URL:-https://127.0.0.1:8935} + - --orchSecret=${LIVEPEER_ORCH_SECRET:-abcdef} + - --runner-url=${LIVEPEER_RUNNER_URL:-http://127.0.0.1:8991} + - --workspace=/workspace/ComfyUI + - --capacity=1 + - --price=${PRICE:-0} + deploy: + resources: + reservations: + devices: + - driver: nvidia + count: 1 + capabilities: [gpu] diff --git a/comfystream/compose.onchain.yml b/comfystream/compose.onchain.yml new file mode 100644 index 0000000..f2d345e --- /dev/null +++ b/comfystream/compose.onchain.yml @@ -0,0 +1,33 @@ +# On-chain payment overlay for comfystream. 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 a stream is billed for as long as it +# runs. Then pay through the signer: +# uv run client.py sample.mp4 \ +# --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 + + app: + command: + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:8991 + - --workspace=/workspace/ComfyUI + - --price=${PRICE} diff --git a/comfystream/compose.yml b/comfystream/compose.yml new file mode 100644 index 0000000..a1fe1c0 --- /dev/null +++ b/comfystream/compose.yml @@ -0,0 +1,36 @@ +# End-to-end offchain demo: orchestrator + ComfyStream live-runner. +# +# ComfyStream is consumed as the published `livepeer/comfystream` image; this +# folder only adds the Livepeer integration (runner.py). Requires an NVIDIA GPU. +# docker compose up -d --build +# uv run client.py sample.mp4 + +services: + orchestrator: + extends: + file: ../compose.orchestrator.yml + service: orchestrator + + app: + build: . + container_name: example_apps_comfystream + depends_on: + orchestrator: + condition: service_healthy + environment: + - NVIDIA_VISIBLE_DEVICES=${NVIDIA_VISIBLE_DEVICES:-all} + - COMFYUI_CWD=/workspace/ComfyUI + - PYTHONUNBUFFERED=1 + command: + - --host=0.0.0.0 + - --orchestrator=https://orchestrator:8935 + - --orchSecret=abcdef + - --runner-url=http://app:8991 + - --workspace=/workspace/ComfyUI + deploy: + resources: + reservations: + devices: + - driver: nvidia + count: 1 + capabilities: [gpu] diff --git a/comfystream/entrypoint.sh b/comfystream/entrypoint.sh new file mode 100644 index 0000000..2142b07 --- /dev/null +++ b/comfystream/entrypoint.sh @@ -0,0 +1,10 @@ +#!/bin/bash +# Activate the ComfyStream image's conda env, then run the integration runner. +set -euo pipefail + +# shellcheck disable=SC1091 +source /workspace/miniconda3/etc/profile.d/conda.sh +conda activate comfystream + +export COMFYUI_CWD="${COMFYUI_CWD:-/workspace/ComfyUI}" +exec python /app/runner.py "$@" diff --git a/comfystream/pyproject.toml b/comfystream/pyproject.toml new file mode 100644 index 0000000..1cfa492 --- /dev/null +++ b/comfystream/pyproject.toml @@ -0,0 +1,10 @@ +[project] +name = "livepeer-comfystream" +version = "0.1.0" +description = "ComfyStream (workflow-driven video + analyze) example app for the Livepeer network." +requires-python = ">=3.12" +dependencies = [ + "av", + "aiohttp", + "livepeer-gateway>=1.0.0", +] diff --git a/comfystream/runner.py b/comfystream/runner.py new file mode 100644 index 0000000..ce58922 --- /dev/null +++ b/comfystream/runner.py @@ -0,0 +1,552 @@ +#!/usr/bin/env python3 +"""comfystream app: workflow-driven analyze + live stream on the Livepeer network. + +Consumes the published ComfyStream package (Pipeline) already installed in the +`livepeer/comfystream` image. This file is only the Livepeer integration. + +Agent surface: + POST /analyze video-in → text-out + POST /start_stream live video (and optional text) trickle session + POST /update_stream mid-session prompt / resolution update + GET /text buffered text outputs for the active session + GET /healthz + +Livepeer integration (grep `# Livepeer:`): + 1. register_runner() + 2. create_trickle_channels() + 3. registration.close() / on_session_release cleanup +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import os +from contextlib import suppress +from dataclasses import dataclass, field +from typing import Any, Optional + +from aiohttp import web +from comfystream.modalities import WorkflowModality +from comfystream.pipeline import Pipeline +from comfystream.utils import convert_prompt +from livepeer_gateway.channel_writer import JSONLWriter +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 + +log = logging.getLogger("comfystream") + +APP_ID = "livepeer-example/comfystream" +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 8991 +CHANNEL_MIME_VIDEO = "video/mp2t" +CHANNEL_MIME_JSONL = "application/jsonl" +TEXT_POLL_INTERVAL = 0.25 + + +@dataclass +class RunnerSession: + session_id: str + kind: str # "analyze" | "stream" + io: WorkflowModality + in_url: str + out_url: str | None = None + text_url: str | None = None + media_in: MediaOutput | None = None + video_out: MediaPublish | None = None + text_out: JSONLWriter | None = None + text_task: asyncio.Task | None = None + collected_text: list[str] = field(default_factory=list) + prompts: Any = None + + def to_json(self) -> dict[str, Any]: + data: dict[str, Any] = { + "session": self.session_id, + "kind": self.kind, + "in": self.in_url, + "modalities": self.io, + } + if self.out_url: + data["out"] = self.out_url + if self.text_url: + data["text"] = self.text_url + return data + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="ComfyStream example app for the Livepeer network." + ) + parser.add_argument( + "--orchestrator", + default=os.environ.get("LIVEPEER_ORCH_URL", "https://localhost:8935"), + ) + parser.add_argument( + "--orchSecret", + default=os.environ.get( + "LIVEPEER_ORCH_SECRET", os.environ.get("ORCH_SECRET", "abcdef") + ), + ) + parser.add_argument( + "--runner-url", + default=os.environ.get( + "LIVEPEER_RUNNER_URL", f"http://{DEFAULT_HOST}:{DEFAULT_PORT}" + ), + ) + parser.add_argument( + "--host", + default=os.environ.get("LIVEPEER_RUNNER_HOST", DEFAULT_HOST), + ) + parser.add_argument( + "--port", + type=int, + default=int(os.environ.get("LIVEPEER_RUNNER_PORT", str(DEFAULT_PORT))), + ) + parser.add_argument( + "--workspace", + default=os.environ.get("COMFYUI_CWD", os.environ.get("COMFYUI_WORKSPACE", "")), + help="ComfyUI workspace directory (COMFYUI_CWD).", + ) + parser.add_argument("--width", type=int, default=512) + parser.add_argument("--height", type=int, default=512) + parser.add_argument( + "--price", + type=float, + default=float(os.environ.get("LIVEPEER_RUNNER_PRICE", "0")), + help="USD per hour (metered). 0 = free, the offchain default.", + ) + parser.add_argument( + "--capacity", + type=int, + default=int(os.environ.get("LIVEPEER_RUNNER_CAPACITY", "1")), + ) + parser.add_argument( + "--skip-bootstrap", + action="store_true", + help="Skip Pipeline default-workflow bootstrap.", + ) + return parser.parse_args() + + +def _session_id(request: web.Request) -> str: + session_id = request.headers.get("Livepeer-Session-Id", "").strip() + if not session_id: + raise web.HTTPBadRequest(text="missing Livepeer-Session-Id header") + return session_id + + +def _channel_url(channel: dict[str, Any], *, internal: bool = False) -> str: + if internal: + return str(channel.get("internal_url") or channel["url"]) + return str(channel["url"]) + + +def _extract_prompts(payload: dict[str, Any]) -> Any: + prompts = payload.get("prompts", payload.get("prompt")) + if prompts is None: + raise web.HTTPBadRequest(text="missing prompts/prompt in body") + if isinstance(prompts, str): + prompts = json.loads(prompts) + return prompts + + +def _convert_prompts(prompts: Any) -> list[dict[str, Any]]: + if isinstance(prompts, list): + return [convert_prompt(p, return_dict=True) for p in prompts] + return [convert_prompt(prompts, return_dict=True)] + + +def _require_analyze_io(io: WorkflowModality) -> None: + if not io["video"]["input"]: + raise web.HTTPBadRequest(text="analyze requires a workflow with video input") + if not io["text"]["output"]: + raise web.HTTPBadRequest(text="analyze requires a workflow with text output") + + +def _require_stream_io(io: WorkflowModality) -> None: + if not (io["video"]["input"] or io["audio"]["input"] or io["video"]["output"]): + raise web.HTTPBadRequest(text="start_stream requires a workflow with media I/O") + + +async def _close_session(app: web.Application, *, stop_prompts: bool = True) -> None: + session: RunnerSession | None = app.get("session") + if session is None: + return + app["session"] = None + + if session.text_task is not None and not session.text_task.done(): + session.text_task.cancel() + with suppress(asyncio.CancelledError, Exception): + await session.text_task + + with suppress(Exception): + if session.media_in is not None: + await session.media_in.close() + with suppress(Exception): + if session.video_out is not None: + await session.video_out.close() + with suppress(Exception): + if session.text_out is not None: + await session.text_out.close() + + pipeline: Pipeline | None = app.get("pipeline") + if stop_prompts and pipeline is not None: + with suppress(Exception): + await pipeline.stop_prompts(cleanup=True) + + +async def _on_session_release(app: web.Application, event: Any) -> None: + session_id = getattr(event, "session_id", "") or "" + session: RunnerSession | None = app.get("session") + if session is None: + return + if session_id and session.session_id != session_id: + return + log.info("orchestrator released session %s; cleaning up", session.session_id) + await _close_session(app) + + +async def _text_forward_loop(app: web.Application, session: RunnerSession) -> None: + pipeline: Pipeline = app["pipeline"] + while True: + try: + text = await pipeline.get_text_output() + if text is None or str(text).strip() == "": + await asyncio.sleep(TEXT_POLL_INTERVAL) + continue + text_str = str(text) + session.collected_text.append(text_str) + if session.text_out is not None: + await session.text_out.write({"type": "text", "text": text_str}) + except asyncio.CancelledError: + raise + except Exception: + log.exception("text forwarder error") + await asyncio.sleep(TEXT_POLL_INTERVAL) + + +async def _handle_video_frame( + app: web.Application, + session: RunnerSession, + decoded: AudioDecodedMediaFrame | VideoDecodedMediaFrame, +) -> None: + if decoded.kind != "video": + return + pipeline: Pipeline = app["pipeline"] + frame = decoded.frame + await pipeline.put_video_frame(frame) + if pipeline.produces_video_output(): + out = await pipeline.get_processed_video_frame() + if session.video_out is not None: + await session.video_out.write_frame(out) + else: + # Video-in / text-out: drain the sync queue without waiting for a video tensor. + await pipeline.video_incoming_frames.get() + + +async def _apply_workflow( + pipeline: Pipeline, + prompts: Any, + *, + width: Optional[int], + height: Optional[int], + skip_warmup: bool = False, +) -> WorkflowModality: + converted = _convert_prompts(prompts) + if width and width > 0: + pipeline.width = int(width) + if height and height > 0: + pipeline.height = int(height) + await pipeline.apply_prompts(converted, skip_warmup=skip_warmup) + if not skip_warmup: + await pipeline.ensure_warmup(pipeline.width, pipeline.height) + if pipeline.state_manager.can_stream(): + await pipeline.start_streaming() + return pipeline.get_workflow_io_capabilities() + + +async def _handle_analyze(request: web.Request) -> web.Response: + app = request.app + session_id = _session_id(request) + existing: RunnerSession | None = app.get("session") + if existing is not None: + if existing.session_id != session_id: + raise web.HTTPConflict(text="runner already has an active session") + return web.json_response(existing.to_json()) + + payload = json.loads(await request.read() or b"{}") + if not isinstance(payload, dict): + raise web.HTTPBadRequest(text="body must be a JSON object") + prompts = _extract_prompts(payload) + width = payload.get("width") + height = payload.get("height") + + pipeline: Pipeline = app["pipeline"] + try: + io = await _apply_workflow( + pipeline, + prompts, + width=int(width) if width else None, + height=int(height) if height else None, + ) + except web.HTTPException: + raise + except Exception as exc: + log.exception("failed to apply analyze workflow") + raise web.HTTPBadRequest(text=f"invalid workflow: {exc}") from exc + + _require_analyze_io(io) + + channels = await app["registration"].create_trickle_channels( # Livepeer: 2 + request, + [ + {"name": "in", "mime_type": CHANNEL_MIME_VIDEO}, + {"name": "text", "mime_type": CHANNEL_MIME_JSONL}, + ], + ) + by_name = {c["name"]: c for c in channels} + if "in" not in by_name or "text" not in by_name: + raise web.HTTPInternalServerError( + text="orchestrator did not return in/text channels" + ) + + session = RunnerSession( + session_id=session_id, + kind="analyze", + io=io, + in_url=_channel_url(by_name["in"]), + text_url=_channel_url(by_name["text"]), + text_out=JSONLWriter(_channel_url(by_name["text"], internal=True)), + prompts=prompts, + ) + + async def _on_frame(decoded) -> None: + await _handle_video_frame(app, session, decoded) + + session.media_in = MediaOutput( + _channel_url(by_name["in"], internal=True), + on_frame=_on_frame, + ) + session.text_task = asyncio.create_task(_text_forward_loop(app, session)) + app["session"] = session + + for task in session.media_in.callback_tasks(): + task.add_done_callback(lambda _t: asyncio.create_task(_close_session(app))) + + log.info("started analyze session %s", session_id) + return web.json_response(session.to_json()) + + +async def _handle_start_stream(request: web.Request) -> web.Response: + app = request.app + session_id = _session_id(request) + existing: RunnerSession | None = app.get("session") + if existing is not None: + if existing.session_id != session_id: + raise web.HTTPConflict(text="runner already has an active session") + return web.json_response(existing.to_json()) + + payload = json.loads(await request.read() or b"{}") + if not isinstance(payload, dict): + raise web.HTTPBadRequest(text="body must be a JSON object") + prompts = _extract_prompts(payload) + width = payload.get("width") + height = payload.get("height") + + pipeline: Pipeline = app["pipeline"] + try: + io = await _apply_workflow( + pipeline, + prompts, + width=int(width) if width else None, + height=int(height) if height else None, + ) + except Exception as exc: + log.exception("failed to apply stream workflow") + raise web.HTTPBadRequest(text=f"invalid workflow: {exc}") from exc + + _require_stream_io(io) + + channel_reqs: list[dict[str, str]] = [] + if io["video"]["input"] or io["audio"]["input"]: + channel_reqs.append({"name": "in", "mime_type": CHANNEL_MIME_VIDEO}) + if io["video"]["output"]: + channel_reqs.append({"name": "out", "mime_type": CHANNEL_MIME_VIDEO}) + if io["text"]["output"]: + channel_reqs.append({"name": "text", "mime_type": CHANNEL_MIME_JSONL}) + if not channel_reqs: + raise web.HTTPBadRequest(text="workflow produced no trickle channels") + + channels = await app["registration"].create_trickle_channels( # Livepeer: 2 + request, + channel_reqs, + ) + by_name = {c["name"]: c for c in channels} + + session = RunnerSession( + session_id=session_id, + kind="stream", + io=io, + in_url=_channel_url(by_name["in"]) if "in" in by_name else "", + out_url=_channel_url(by_name["out"]) if "out" in by_name else None, + text_url=_channel_url(by_name["text"]) if "text" in by_name else None, + prompts=prompts, + ) + if "out" in by_name: + session.video_out = MediaPublish(_channel_url(by_name["out"], internal=True)) + if "text" in by_name: + session.text_out = JSONLWriter(_channel_url(by_name["text"], internal=True)) + session.text_task = asyncio.create_task(_text_forward_loop(app, session)) + + if "in" in by_name: + + async def _on_frame(decoded) -> None: + await _handle_video_frame(app, session, decoded) + + session.media_in = MediaOutput( + _channel_url(by_name["in"], internal=True), + on_frame=_on_frame, + ) + for task in session.media_in.callback_tasks(): + task.add_done_callback(lambda _t: asyncio.create_task(_close_session(app))) + + app["session"] = session + log.info("started stream session %s", session_id) + return web.json_response(session.to_json()) + + +async def _handle_update_stream(request: web.Request) -> web.Response: + app = request.app + session_id = _session_id(request) + session: RunnerSession | None = app.get("session") + if session is None: + raise web.HTTPNotFound(text="no active session") + if session.session_id != session_id: + raise web.HTTPConflict(text="runner has a different active session") + + payload = json.loads(await request.read() or b"{}") + if not isinstance(payload, dict): + raise web.HTTPBadRequest(text="body must be a JSON object") + + pipeline: Pipeline = app["pipeline"] + width = payload.get("width") + height = payload.get("height") + if width: + pipeline.width = int(width) + if height: + pipeline.height = int(height) + + if "prompts" in payload or "prompt" in payload: + prompts = _extract_prompts(payload) + try: + io = await _apply_workflow( + pipeline, + prompts, + width=int(width) if width else None, + height=int(height) if height else None, + skip_warmup=True, + ) + except Exception as exc: + log.exception("failed to update stream workflow") + raise web.HTTPBadRequest(text=f"invalid workflow update: {exc}") from exc + session.io = io + session.prompts = prompts + if io["text"]["output"] and session.text_task is None: + if session.text_out is None and session.text_url: + session.text_out = JSONLWriter(session.text_url) + if session.text_out is not None: + session.text_task = asyncio.create_task( + _text_forward_loop(app, session) + ) + + return web.json_response(session.to_json()) + + +async def _handle_text(request: web.Request) -> web.Response: + session: RunnerSession | None = request.app.get("session") + if session is None: + raise web.HTTPNotFound(text="no active session") + session_id = request.headers.get("Livepeer-Session-Id", "").strip() + if session_id and session_id != session.session_id: + raise web.HTTPConflict(text="runner has a different active session") + return web.json_response( + { + "session": session.session_id, + "texts": list(session.collected_text), + } + ) + + +async def _handle_healthz(_request: web.Request) -> web.Response: + return web.json_response({"ok": True, "app": APP_ID}) + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + ) + args = _parse_args() + if not args.workspace: + raise SystemExit("--workspace / COMFYUI_CWD is required") + + async def _on_startup(app: web.Application) -> None: + pipeline = Pipeline( + width=args.width, + height=args.height, + cwd=args.workspace, + disable_cuda_malloc=True, + gpu_only=True, + preview_method="none", + blacklist_custom_nodes=["ComfyUI-Manager"], + bootstrap_default_prompt=not args.skip_bootstrap, + ) + await pipeline.initialize() + app["pipeline"] = pipeline + app["session"] = None + + async def _release(event: Any) -> None: + await _on_session_release(app, event) + + app["registration"] = await register_runner( # Livepeer: 1 + args.orchestrator, + secret=args.orchSecret, + runner_url=args.runner_url, + app=APP_ID, + mode="persistent", + capacity=args.capacity, + price=args.price, + currency="usd", + unit="hour", + on_session_release=_release, + ) + log.info( + "registered app=%s runner_id=%s orchestrator=%s runner_url=%s", + APP_ID, + app["registration"].runner_id, + app["registration"].orchestrator_url, + args.runner_url, + ) + + async def _on_cleanup(app: web.Application) -> None: + await _close_session(app) + with suppress(Exception): + await app["registration"].close() # Livepeer: 3 + + app = web.Application() + app.router.add_post("/analyze", _handle_analyze) + app.router.add_post("/start_stream", _handle_start_stream) + app.router.add_post("/update_stream", _handle_update_stream) + app.router.add_get("/text", _handle_text) + app.router.add_get("/healthz", _handle_healthz) + app.on_startup.append(_on_startup) + app.on_cleanup.append(_on_cleanup) + web.run_app(app, host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/comfystream/workflows/analyze-stub-api.json b/comfystream/workflows/analyze-stub-api.json new file mode 100644 index 0000000..0ce11cb --- /dev/null +++ b/comfystream/workflows/analyze-stub-api.json @@ -0,0 +1,15 @@ +{ + "1": { + "inputs": {}, + "class_type": "LoadTensor", + "_meta": { "title": "LoadTensor" } + }, + "2": { + "inputs": { + "data": "comfystream-analyze-ok", + "remove_linebreaks": true + }, + "class_type": "SaveTextTensor", + "_meta": { "title": "SaveTextTensor" } + } +}