Add the translation service to the translation demo

translation-demo/.env.example

@@ -2,3 +2,12 @@
 # the sandbox (the relay host, the Fishjam ID root, and the JWT embedded as `?jwt=`) and connects
 # to it directly. Get it at https://fishjam.io/app/sandbox.
 VITE_SANDBOX_API_URL=https://fishjam.io/api/v1/connect/<your-key>/room-manager
+
+# Translation service (server-side only — never expose these with a VITE_ prefix).
+# The service in ./service fetches a MoQ token from Fishjam with a management token
+# and translates every stream in the account. Get both at https://fishjam.io/app.
+FISHJAM_ID=<your-fishjam-id>
+FISHJAM_MANAGEMENT_TOKEN=<your-management-token>
+
+# Gemini API key for the translation service.
+GEMINI_API_KEY=<your-gemini-api-key>

translation-demo/.gitignore

@@ -9,6 +9,10 @@ dist-ssr
 # Env
 .env
 
+# Python (translation service)
+service/.venv
+__pycache__/
+
 # Yarn (Berry)
 .yarn/*
 !.yarn/patches

translation-demo/README.md

@@ -1,28 +1,57 @@
 # Translation Demo
 
-An example React app that streams over the MoQ protocol with Fishjam, with live audio translation and captions for the viewer.
+Live MoQ streaming with real-time AI audio translation and captions, built on
+Fishjam. The demo has two parts:
 
-## Getting Started
+- **Web app** — a React app for publishing a stream and
+  watching it, with a translation menu and live captions on the watch page.
+- **Translation service** (`service/`) — a Python service that subscribes to
+  every stream and announces AI-translated audio and caption tracks
+  which the web app picks up.
 
-Install dependencies:
+Both are needed for translations: the watch page's translation menu only shows
+languages while the service is running.
+
+## Prerequisites
+
+- A Fishjam account (https://fishjam.io/app) — you'll need the sandbox API URL,
+  your Fishjam ID, and a management token.
+- A Gemini API key for the translation provider.
+- Node.js with Yarn, Python 3.11+, and [uv](https://docs.astral.sh/uv/).
+
+## Setup
+
+Configure credentials once for both parts:
 
 ```bash
-yarn
+cp .env.example .env
+# fill in VITE_SANDBOX_API_URL, FISHJAM_ID, FISHJAM_MANAGEMENT_TOKEN,
+# and GEMINI_API_KEY
 ```
 
-Configure the sandbox API URL (required):
+## Run the translation service
 
 ```bash
-cp .env.example .env
-# then set VITE_SANDBOX_API_URL in .env to your Fishjam sandbox API URL
+cd service
+uv sync
+uv run --env-file ../.env translator
 ```
 
-Start the development server:
+Translation is powered by Google Gemini Live. See `service/README.md` for details.
+
+## Run the web app
 
 ```bash
+yarn
 yarn dev
 ```
 
+Open the printed URL and publish a stream, then open the viewer link shown in
+the publisher panel (the watch page) in another tab and pick a translation
+language.
+
 ## Environment Variables
 
-- `VITE_SANDBOX_API_URL` (required) — Fishjam sandbox API URL used to fetch a MoQ relay connection URL. Get it at https://fishjam.io/app/sandbox.
+- `VITE_SANDBOX_API_URL` (web app) — Fishjam sandbox API URL used to fetch a MoQ relay connection URL. Get it at https://fishjam.io/app/sandbox.
+- `FISHJAM_ID`, `FISHJAM_MANAGEMENT_TOKEN` (translation service) — Fishjam credentials used to mint MoQ tokens. Get them at https://fishjam.io/app.
+- `GEMINI_API_KEY` (translation service) — Google Gemini API key used for translation.

translation-demo/service/.python-version

@@ -0,0 +1 @@
+3.14

translation-demo/service/README.md

@@ -0,0 +1,62 @@
+# Translation Service
+
+Async Python service that announces live AI-translated audio and captions for
+MoQ audio broadcasts, powered by Google Gemini Live.
+
+For each source broadcast announced by the relay, the service reads the Hang
+catalog, finds a supported audio track, and announces a sibling dynamic
+broadcast at:
+
+```text
+<source>/<provider>/translation
+```
+
+Subscribers request a target language by subscribing to a track named with the
+language code, for example `es` or `pt-BR`. Translation starts on demand: the
+provider session opens when the first subscriber requests a language and closes
+shortly after the last one leaves.
+
+While a translated audio track is active, subscribers can request its live
+transcript on a track named `<language>/transcript.json`. Transcript frames are
+UTF-8 JSON replace-state payloads:
+
+```json
+{"segments": [{"text": "Hola", "ts_us": 0, "final": false}]}
+```
+
+## Run
+
+```bash
+uv sync
+uv run --env-file ../.env translator
+```
+
+The service fetches a MoQ token from Fishjam using a management token and
+reconnects with a fresh token before the hourly expiry. A single instance
+covers all streams in the account, including ones published after it starts.
+
+## Environment variables
+
+- `FISHJAM_ID`, `FISHJAM_MANAGEMENT_TOKEN` — Fishjam credentials used to mint
+  MoQ tokens. Get them at https://fishjam.io/app.
+- `GEMINI_API_KEY` — Google Gemini API key (`GOOGLE_API_KEY` is accepted as a
+  fallback).
+
+## Useful flags
+
+- `--prefix` — only watch broadcasts under this path prefix (default: all
+  streams in the account).
+- `--google-model` — override the Gemini Live translation model (default
+  `models/gemini-3.5-live-translate-preview`).
+- `--url` — connect directly to a MoQ relay without Fishjam authentication
+  (local development); combine with `--no-tls-verify` for self-signed relays.
+- `--log-level DEBUG` — verbose logging.
+
+## Notes
+
+- Supported target languages come from the Gemini Live Translate BCP-47
+  language list; unsupported language requests are closed without opening a
+  provider session.
+- Opus and AAC sources are supported. Output mirrors the source codec family.
+- Transcript tracks are scoped to an active audio language track and close when
+  that audio translation stops.

translation-demo/service/pyproject.toml

@@ -0,0 +1,30 @@
+[project]
+name = "moq-live-translation"
+version = "0.1.0"
+description = "Lazy AI MoQ audio translation service"
+readme = "README.md"
+authors = [
+    { name = "Jakub Perżyło", email = "perzylo.jakub@gmail.com" }
+]
+requires-python = ">=3.11"
+dependencies = [
+    "av>=17.0.1",
+    "fishjam-server-sdk>=0.28.1",
+    "moq-net>=0.1.0",
+    "websockets>=15.0",
+]
+
+[tool.uv]
+# fishjam-server-sdk pins the pre-release betterproto==2.0.0b6
+prerelease = "allow"
+
+[project.scripts]
+translator = "translator.cli:main"
+
+[tool.uv.build-backend]
+module-name = "translator"
+module-root = ""
+
+[build-system]
+requires = ["uv_build>=0.9.27,<0.10.0"]
+build-backend = "uv_build"

translation-demo/service/translator/__init__.py

@@ -0,0 +1,5 @@
+"""MoQ live audio translation CLI."""
+
+from .cli import main
+
+__all__ = ["main"]

translation-demo/service/translator/__main__.py

@@ -0,0 +1,5 @@
+from .cli import main
+
+
+if __name__ == "__main__":
+    main()

translation-demo/service/translator/catalog.py

@@ -0,0 +1,147 @@
+"""Helpers for creating output audio tracks from Hang catalog entries."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from urllib.parse import quote
+
+from .moq_compat import moq
+
+
+SAMPLE_RATE_INDEX = {
+    96_000: 0,
+    88_200: 1,
+    64_000: 2,
+    48_000: 3,
+    44_100: 4,
+    32_000: 5,
+    24_000: 6,
+    22_050: 7,
+    16_000: 8,
+    12_000: 9,
+    11_025: 10,
+    8_000: 11,
+    7_350: 12,
+}
+
+
+@dataclass(frozen=True)
+class AudioPublication:
+    """A source audio track that can be represented as a publishable output track."""
+
+    source_name: str
+    audio: moq.Audio
+    format: str
+    init: bytes
+
+
+def translation_path(source_path: str, provider_name: str) -> str:
+    """Return the translated broadcast path for a source broadcast."""
+    provider_segment = quote(provider_name.strip(), safe="")
+    if not provider_segment:
+        raise ValueError("provider name must not be empty")
+
+    return f"{source_path.rstrip('/')}/{provider_segment}/translation"
+
+
+def is_translation_path(path: str) -> bool:
+    return "translation" in path.rstrip("/").split("/")
+
+
+def audio_publication(source_name: str, audio: moq.Audio) -> AudioPublication | None:
+    """Build the publish-media arguments for a supported source audio track."""
+    codec = codec_family(audio.codec)
+    if not 0 < audio.channel_count <= 2:
+        raise ValueError(f"unsupported translation channel count: {audio.channel_count}")
+
+    if codec == "opus":
+        return AudioPublication(
+            source_name=source_name,
+            audio=audio,
+            format="opus",
+            init=opus_head(audio.sample_rate, audio.channel_count),
+        )
+
+    if codec == "aac":
+        return AudioPublication(
+            source_name=source_name,
+            audio=audio,
+            format="aac",
+            init=audio_specific_config(
+                profile=2,
+                sample_rate=audio.sample_rate,
+                channel_count=audio.channel_count,
+            ),
+        )
+
+    return None
+
+
+def codec_family(codec: str) -> str:
+    codec = codec.lower()
+    if codec == "opus":
+        return "opus"
+    if codec == "aac" or codec.startswith("mp4a.40."):
+        return "aac"
+    return codec
+
+
+def opus_head(sample_rate: int, channel_count: int) -> bytes:
+    if not 0 < channel_count <= 255:
+        raise ValueError(f"unsupported Opus channel count: {channel_count}")
+    if not 0 < sample_rate <= 0xFFFFFFFF:
+        raise ValueError(f"unsupported Opus sample rate: {sample_rate}")
+
+    return b"".join(
+        [
+            b"OpusHead",
+            bytes([1, channel_count]),
+            (0).to_bytes(2, "little"),
+            sample_rate.to_bytes(4, "little"),
+            (0).to_bytes(2, "little"),
+            bytes([0]),
+        ]
+    )
+
+
+def aac_init(audio: moq.Audio) -> bytes:
+    if audio.description:
+        return bytes(audio.description)
+
+    return audio_specific_config(
+        profile=aac_profile(audio.codec),
+        sample_rate=audio.sample_rate,
+        channel_count=audio.channel_count,
+    )
+
+
+def aac_profile(codec: str) -> int:
+    codec = codec.lower()
+    if codec.startswith("mp4a.40."):
+        try:
+            profile = int(codec.removeprefix("mp4a.40."))
+        except ValueError as exc:
+            raise ValueError(f"unsupported AAC codec string: {codec}") from exc
+    else:
+        profile = 2
+
+    if not 0 < profile < 31:
+        raise ValueError(f"unsupported AAC profile: {profile}")
+    return profile
+
+
+def audio_specific_config(profile: int, sample_rate: int, channel_count: int) -> bytes:
+    if not 0 < channel_count <= 7:
+        raise ValueError(f"unsupported AAC channel count: {channel_count}")
+
+    frequency_index = SAMPLE_RATE_INDEX.get(sample_rate, 15)
+    first = (profile << 3) | (frequency_index >> 1)
+    second = ((frequency_index & 1) << 7) | (channel_count << 3)
+    config = bytes([first, second])
+
+    if frequency_index == 15:
+        if not 0 < sample_rate <= 0xFFFFFF:
+            raise ValueError(f"unsupported AAC sample rate: {sample_rate}")
+        config += sample_rate.to_bytes(3, "big")
+
+    return config