refactor(realtime-transcription): match the other examples, fix three bugs

Review pass over PR #57 before merge.

Consistency with the merged examples:
- port 5005 to 8989, and CMD to ENTRYPOINT so compose stops repeating
  `python runner.py`
- flatten pyproject; the `runner` extra was dead config and broke
  `uv run runner.py`
- drop --device/--compute-type, whose cpu/int8 defaults contradicted the
  GPU-only docs
- cut the compose and Dockerfile headers to the echo template
- client takes its input positionally and inlines the signer, like echo
- hold the session with `async with`, like echo
- shorten the runner docstring and constant comments to the house shape

Fixes:
- eos cancels the worker before its closing transcribe, so a partial can no
  longer land after the final
- _rms no longer raises when a frame splits a sample
- the client only builds an SSL context for wss://

New:
- start/end seconds on every message, matching what Deepgram and OpenAI
  segments carry
- `-` reads raw PCM from stdin, so a microphone works through ffmpeg
- sample.wav is fetched from a public domain NASA clip instead of committed

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: rickstaa

realtime-transcription/.env.example

@@ -1,13 +1,6 @@
 # Copy to .env (gitignored) and fill in. Never commit secrets.
 # Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only.
 
-# Device only: the model is fixed at large-v3-turbo, which needs a GPU to stay
-# realtime. cpu/int8 loads but falls behind a live stream.
-WHISPER_DEVICE=cuda
-WHISPER_COMPUTE=float16
-
-# --- On-chain (paid) only below; offchain ignores these. ---
-
 NETWORK=arbitrum-one-mainnet
 ETH_RPC_URL=https://arb1.arbitrum.io/rpc
 

realtime-transcription/Dockerfile

@@ -1,7 +1,4 @@
-# Realtime-transcription app: an aiohttp WebSocket server wrapping faster-whisper
-# that self-registers as a Live Runner. GPU only: large-v3-turbo is the model that
-# is both accurate and fast enough to keep pace with a live stream, and it needs a
-# CUDA runtime (CTranslate2 wants cuBLAS + cuDNN).
+# Realtime-transcription example app (persistent WebSocket, GPU speech-to-text).
 FROM python:3.12-slim
 
 # Flush stdout/stderr immediately so output isn't block-buffered in `docker logs`.
@@ -11,9 +8,9 @@ RUN apt-get update \
     && apt-get install -y --no-install-recommends git \
     && rm -rf /var/lib/apt/lists/*
 
-# CTranslate2 loads cuBLAS/cuDNN from the nvidia pip wheels, so no CUDA base image
+# livepeer-gateway SDK isn't on PyPI yet; install from Git. CTranslate2 (under
+# 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.
-# livepeer-gateway SDK isn't on PyPI yet; install from Git.
 RUN pip install --no-cache-dir \
     faster-whisper numpy \
     nvidia-cublas-cu12 nvidia-cudnn-cu12 \
@@ -24,6 +21,6 @@ ENV LD_LIBRARY_PATH=/usr/local/lib/python3.12/site-packages/nvidia/cublas/lib:/u
 WORKDIR /app
 COPY runner.py client.py ./
 
-EXPOSE 5005
+EXPOSE 8989
 
-CMD ["python", "runner.py"]
+ENTRYPOINT ["python", "runner.py"]

realtime-transcription/README.md

@@ -9,9 +9,9 @@ Realtime speech-to-text on the Livepeer network over a **WebSocket** — the cli
 | Registration | dynamic (self-registers via the SDK)      |
 | Model        | `large-v3-turbo` (faster-whisper, fixed)  |
 | Transport    | WebSocket (`/transcribe`)                 |
-| Port         | 5005                                      |
+| Port         | 8989                                      |
 
-**Requires an NVIDIA GPU.** The model is fixed at `large-v3-turbo`: it swaps large-v3's 32-layer decoder for 4, so it runs far below realtime on a 3090 while staying near large-v3 quality. On CPU it loads but falls behind a live stream, which is the one thing this example is about. Prerequisites (Docker, `uv`, the not-yet-released SDK) and the shared on-chain/payment setup are in the [repo README](../README.md).
+**Requires an NVIDIA GPU.** The model is fixed at `large-v3-turbo`: it swaps large-v3's 32-layer decoder for 4, so it runs far below realtime on a 3090 while staying near large-v3 quality. The device is pinned with it (`cuda`/`float16`) rather than exposed as a flag: on CPU the model loads but falls behind a live stream, which is the one thing this example is about. Prerequisites (Docker, `uv`, the not-yet-released SDK) and the shared on-chain/payment setup are in the [repo README](../README.md).
 
 ## How it's wired
 
@@ -27,25 +27,43 @@ Wire protocol on `/transcribe`:
 
 - client → server: binary frames of **16 kHz mono PCM (int16)**
 - client → server: text `eos` to finish
-- server → client: JSON `{"text": "...", "final": false|true}`
+- server → client: JSON `{"text": "...", "final": false|true, "start": <sec>, "end": <sec>}`
+
+Cumulative, not incremental: each partial carries the whole utterance so far and may revise earlier words, so a client replaces rather than appends. That matches Deepgram and Vosk, and it is the honest shape for a decoder that re-runs over the buffer. Delta protocols (OpenAI's `transcript.text.delta`) only become correct once decoding is append-only, which is what the LocalAgreement approach below buys you.
 
 ## Audio
 
-Input must be **16 kHz mono WAV**. Convert any file you have, or record a few seconds of yourself talking:
+Input must be **16 kHz mono WAV**. Fetch 21s of NASA podcast speech, public domain under 17 U.S.C. 105:
+
+```sh
+curl -sL https://images-assets.nasa.gov/audio/Ep401_Artemis_II_Launch/Ep401_Artemis_II_Launch~128k.mp3 \
+  | ffmpeg -ss 900 -t 21 -i pipe:0 -ar 16000 -ac 1 sample.wav
+```
+
+Or bring your own, replacing `input.mp3` / picking your capture device:
 
 ```sh
 ffmpeg -i input.mp3 -ar 16000 -ac 1 sample.wav                 # convert
 ffmpeg -f alsa -i default -ar 16000 -ac 1 -t 20 sample.wav     # record (macOS: -f avfoundation -i :0)
 ```
 
-Use a clip with a couple of sentences and a pause between them: the app finalizes on trailing silence, so that is what shows partials turning into finals more than once.
+Use a clip with a couple of sentences and a pause between them: the app finalizes on trailing silence, so that is what shows partials turning into finals more than once. The NASA clip is trimmed to three such sentences.
+
+Or skip the file and talk into a microphone: pass `-` and pipe raw PCM in, which streams until you Ctrl-C.
+
+```sh
+ffmpeg -f alsa -i default -ar 16000 -ac 1 -f s16le - \
+  | uv run client.py --discovery https://localhost:8935/discovery -
+```
+
+(macOS: `-f avfoundation -i :0`. If `default` fails, name the device: `arecord -l` then `-i plughw:1,0`.)
 
 ## Run offchain (free)
 
 ```sh
 docker compose up -d --build      # first run downloads the whisper model
 curl -sk https://localhost:8935/discovery | jq '.[].runners[].app'   # confirm livepeer-example/realtime-transcription registered
-uv run client.py --discovery https://localhost:8935/discovery --file sample.wav
+uv run client.py --discovery https://localhost:8935/discovery sample.wav
 docker compose down
 ```
 
@@ -59,7 +77,7 @@ Layer `compose.onchain.yml` to add a remote signer and run the orchestrator on-c
 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 --discovery https://localhost:8935/discovery \
-  --signer http://localhost:7936 --file sample.wav
+  --signer http://localhost:7936 sample.wav
 docker compose -f compose.yml -f compose.onchain.yml down
 ```
 
@@ -72,5 +90,5 @@ Start an orchestrator built from go-livepeer `v0.9.0` or newer (see [Build from
 ```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 --file sample.wav
+uv run client.py sample.wav
 ```

realtime-transcription/client.py

@@ -10,17 +10,21 @@
   2. ws_connect()           — open the proxied WebSocket to the session URL
   3. stop_runner_session()  — end the session (settles payment on-chain)
 
-Audio must be 16 kHz mono. Convert anything with ffmpeg:
+Audio must be 16 kHz mono. Convert a file, or stream a mic through stdin:
   ffmpeg -i input.mp3 -ar 16000 -ac 1 sample.wav
+  ffmpeg -f alsa -i default -ar 16000 -ac 1 -f s16le - | uv run client.py -
 """
+
 from __future__ import annotations
 
 import argparse
 import asyncio
 import logging
 import ssl
+import sys
 import wave
 from contextlib import suppress
+from pathlib import Path
 
 import aiohttp
 
@@ -40,15 +44,23 @@ def _parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         description="Stream audio to a Whisper Live Runner over WebSocket."
     )
+    parser.add_argument(
+        "input",
+        help=(
+            "16 kHz mono WAV to stream, or - to read raw PCM from stdin "
+            "(e.g. a mic piped from ffmpeg)"
+        ),
+    )
     parser.add_argument("--discovery", default=DEFAULT_DISCOVERY)
-    parser.add_argument("--file", required=True, help="16 kHz mono WAV to stream.")
     parser.add_argument(
         "--signer", default="", help="Remote signer base URL (on-chain/paid path)."
     )
     return parser.parse_args()
 
 
 def _read_pcm(path: str) -> bytes:
+    # The socket carries bare samples with no format header, so the file has to
+    # already match what the runner expects; readframes() drops the WAV header.
     with wave.open(path, "rb") as w:
         if (
             w.getframerate() != SAMPLE_RATE
@@ -62,11 +74,20 @@ def _read_pcm(path: str) -> bytes:
         return w.readframes(w.getnframes())
 
 
-async def _send(ws: aiohttp.ClientWebSocketResponse, pcm: bytes) -> None:
+async def _send(
+    ws: aiohttp.ClientWebSocketResponse, pcm: bytes, *, live: bool = False
+) -> None:
     step = SAMPLE_RATE * 2 * CHUNK_MS // 1000  # bytes per chunk
-    for i in range(0, len(pcm), step):
-        await ws.send_bytes(pcm[i : i + step])
-        await asyncio.sleep(CHUNK_MS / 1000)  # pace at real time
+    if live:
+        # A mic arrives at real time already, so read blocking (off the event
+        # loop) and forward as it comes; the pipe closing ends the stream.
+        loop = asyncio.get_running_loop()
+        while chunk := await loop.run_in_executor(None, sys.stdin.buffer.read, step):
+            await ws.send_bytes(chunk)
+    else:
+        for i in range(0, len(pcm), step):
+            await ws.send_bytes(pcm[i : i + step])
+            await asyncio.sleep(CHUNK_MS / 1000)  # pace at real time
     await ws.send_str("eos")
 
 
@@ -78,22 +99,30 @@ async def _recv(ws: aiohttp.ClientWebSocketResponse) -> None:
             break
         data = msg.json()
         marker = "FINAL" if data.get("final") else "partial"
-        print(f"[{marker}] {data.get('text', '')}")
+        span = f"{data.get('start', 0):6.2f}-{data.get('end', 0):6.2f}s"
+        print(f"[{span}] [{marker}] {data.get('text', '')}")
 
 
 async def main() -> None:
     logging.basicConfig(
         level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s"
     )
     args = _parse_args()
-    pcm = _read_pcm(args.file)
-    signer_url = args.signer.strip() or None
+    input_source = args.input.strip()
+    live = input_source == "-"
+    if not live:
+        input_path = Path(input_source).expanduser()
+        if not input_path.exists():
+            raise SystemExit(f"input file does not exist: {input_path}")
+        input_source = str(input_path)
+    # Read (and validate) the file before reserving a paid session.
+    pcm = b"" if live else _read_pcm(input_source)
     session = None
     try:
         session = await reserve_session(  # Livepeer: 1
             discovery_url=args.discovery,  # omit if the signer does discovery itself
             app=APP_ID,
-            signer_url=signer_url,
+            signer_url=args.signer.strip() or None,
         )
         log.info("session_id=%s app_url=%s", session.session_id, session.app_url)
         ws_url = (
@@ -102,14 +131,21 @@ async def main() -> None:
             .rstrip("/")
             + "/transcribe"
         )
-        ctx = ssl.create_default_context()  # orchestrator serves a self-signed cert
-        ctx.check_hostname = False
-        ctx.verify_mode = ssl.CERT_NONE
-        async with aiohttp.ClientSession() as cs:
-            async with cs.ws_connect(
-                ws_url, ssl=ctx, heartbeat=20
-            ) as ws:  # Livepeer: 2
-                await asyncio.gather(_send(ws, pcm), _recv(ws))
+        # Verification comes off: the orchestrator serves a self-signed cert.
+        ctx: ssl.SSLContext | None = None
+        if ws_url.startswith("wss://"):
+            ctx = ssl.create_default_context()
+            ctx.check_hostname = False
+            ctx.verify_mode = ssl.CERT_NONE
+
+        # The session funds itself while it is held; leaving this block stops that.
+        # Here that block is the socket's lifetime, which is what gets metered.
+        async with session:
+            async with aiohttp.ClientSession() as cs:
+                async with cs.ws_connect(
+                    ws_url, ssl=ctx or True, heartbeat=20
+                ) as ws:  # Livepeer: 2
+                    await asyncio.gather(_send(ws, pcm, live=live), _recv(ws))
     except LivepeerGatewayError as exc:
         raise SystemExit(f"ERROR: {exc}") from exc
     finally:

realtime-transcription/compose.onchain.yml

@@ -1,12 +1,13 @@
-# On-chain payment overlay for the streaming-ASR example. Layer on the base:
+# On-chain payment overlay for realtime-transcription. 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, and
-# registers the app with a price. A WebSocket rides a reserved (persistent,
-# metered) session: reserve_session pays at reserve, the meter runs while the
-# socket is open. Requires a local .env (gitignored). Then pay through the signer:
+# 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 the socket is billed for as long as it
+# stays open. Then pay through the signer:
 #   uv run client.py --discovery https://localhost:8935/discovery \
-#     --signer http://localhost:7936 --file sample.wav
+#     --signer http://localhost:7936 sample.wav
 
 services:
   signer:
@@ -24,12 +25,9 @@ services:
   # Re-declare the command to advertise a price (base file registers free).
   app:
     command:
-      - python
-      - runner.py
       - --host=0.0.0.0
       - --orchestrator=https://orchestrator:8935
       - --orchSecret=abcdef
-      - --runner-url=http://app:5005
-      - --device=${WHISPER_DEVICE:-cuda}
-      - --compute-type=${WHISPER_COMPUTE:-float16}
+      - --runner-url=http://app:8989
+      # Billed per second of session (metered); price cap in .env.example.
       - --price=${PRICE}

realtime-transcription/compose.yml

@@ -1,12 +1,11 @@
-# Offchain demo: orchestrator + a streaming-ASR app that self-registers
-# (dynamic) and serves speech-to-text over a WebSocket.
+# End-to-end offchain demo: orchestrator + realtime-transcription app (WebSocket
+# speech-to-text).
 #
+# The orchestrator service is defined once in ../compose.orchestrator.yml and
+# pulled in with `extends`; this file only adds the app. Needs an NVIDIA GPU: the
+# pinned model only keeps pace on one. Once up, stream a WAV from the host:
 #   docker compose up -d --build
-#   uv run client.py --discovery https://localhost:8935/discovery --file sample.wav
-#
-# The orchestrator is defined once in ../compose.orchestrator.yml and pulled in
-# with `extends`. The app embeds the SDK and registers itself, like hello-world.
-# Requires an NVIDIA GPU: the pinned model only keeps pace on one.
+#   uv run client.py --discovery https://localhost:8935/discovery sample.wav
 
 services:
   orchestrator:
@@ -22,15 +21,10 @@ services:
       orchestrator:
         condition: service_healthy
     command:
-      - python
-      - runner.py
       - --host=0.0.0.0
       - --orchestrator=https://orchestrator:8935
       - --orchSecret=abcdef
-      - --runner-url=http://app:5005
-      - --device=${WHISPER_DEVICE:-cuda}
-      - --compute-type=${WHISPER_COMPUTE:-float16}
-    # large-v3-turbo needs the GPU to stay ahead of a live stream.
+      - --runner-url=http://app:8989
     deploy:
       resources:
         reservations:

realtime-transcription/pyproject.toml

@@ -3,17 +3,11 @@ name = "livepeer-realtime-transcription"
 version = "0.1.0"
 description = "Streaming speech-to-text (WebSocket) example app for the Livepeer network."
 requires-python = ">=3.12"
-# Base deps are what the *client* needs (host `uv run client.py`). The runner's
-# ASR stack (faster-whisper) is the `runner` extra, installed in the Docker image.
 dependencies = [
-    "aiohttp",          # client WebSocket
-    "livepeer-gateway",
-]
-
-[project.optional-dependencies]
-runner = [
     "faster-whisper",   # runner: streaming Whisper ASR
-    "numpy",
+    "numpy",            # runner: PCM buffers + energy VAD
+    "aiohttp",
+    "livepeer-gateway",
 ]
 
 # livepeer-gateway is not on PyPI yet; pull it from the branch.