Add transcription example

Author: AHGIJMKLKKZNPJKQR

examples/transcription/.python-version

@@ -0,0 +1 @@
+3.12

examples/transcription/README.md

@@ -0,0 +1,46 @@
+import asyncio
+from contextlib import asynccontextmanager
+from typing import Annotated
+
+from fastapi import Depends, FastAPI
+from transcription.notifier import make_notifier
+from transcription.room import RoomService, fishjam
+from transcription.transcription import transcription_session_factory
+
+from fishjam import PeerOptions, SubscribeOptions
+
+_room_service: RoomService | None = None
+
+
+def get_room_service():
+    if not _room_service:
+        raise RuntimeError("Application skipped lifespan events!")
+    return _room_service
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    async with (
+        transcription_session_factory() as session_factory,
+        asyncio.TaskGroup() as tg,
+    ):
+        global _room_service
+        _room_service = RoomService(session_factory)
+        notifier = make_notifier(_room_service)
+        notifier_task = tg.create_task(notifier.connect())
+
+        yield
+
+        notifier_task.cancel()
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+@app.get("/")
+def get_peer(room_service: Annotated[RoomService, Depends(get_room_service)]):
+    _peer, token = fishjam.create_peer(
+        room_service.get_room().id,
+        PeerOptions(subscribe=SubscribeOptions()),
+    )
+    return token

examples/transcription/main.py

@@ -0,0 +1,13 @@
+[project]
+name = "transcription"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "fastapi[standard]==0.116.0",
+    "fishjam-server-sdk",
+]
+
+[tool.uv.sources]
+fishjam-server-sdk = { workspace = true }

examples/transcription/pyproject.toml

@@ -0,0 +1,61 @@
+import asyncio
+from typing import Callable
+
+from fishjam.agent import Agent, AgentResponseTrackData
+
+from .transcription import TranscriptionSession
+
+
+class TranscriptionAgent:
+    def __init__(
+        self,
+        room_id: str,
+        agent: Agent,
+        on_text: Callable[[str, str], None],
+        session_factory: Callable[[str], TranscriptionSession],
+    ):
+        self._room_id = room_id
+        self._agent = agent
+        self._peers: dict[str, TranscriptionSession] = {}
+        self._on_text = on_text
+        self._session_factory = session_factory
+        self._leave_event = asyncio.Event()
+        self.done = False
+
+        @agent.on_track_data
+        def _(track_data: AgentResponseTrackData):
+            if track_data.peer_id not in self._peers:
+                return
+            self._peers[track_data.peer_id].transcribe(track_data.data)
+
+        print(f"Created agent for room {room_id}")
+        self._task = asyncio.create_task(self._run_agent())
+
+    async def _run_agent(self):
+        print(f"Connecting agent to room {self._room_id}")
+        async with self._agent:
+            print(f"Agent connected to room {self._room_id}")
+            await self._leave_event.wait()
+        print(f"Agent disconnected from room {self._room_id}")
+
+    def on_peer_enter(self, peer_id: str):
+        if peer_id in self._peers:
+            return
+
+        print(f"Starting transcription session for peer {peer_id}")
+
+        self._peers[peer_id] = self._session_factory(peer_id)
+
+    def on_peer_leave(self, peer_id: str):
+        if peer_id not in self._peers:
+            return
+
+        print(f"Ending transcription session for peer {peer_id}")
+
+        session = self._peers[peer_id]
+        session.end()
+        self._peers.pop(peer_id)
+
+        if len(self._peers) == 0:
+            self._leave_event.set()
+            self.done = True

examples/transcription/transcription/__init__.py

@@ -0,0 +1,12 @@
+import os
+
+from google.genai.types import AudioTranscriptionConfig, LiveConnectConfig, Modality
+
+FISHJAM_ID = os.environ["FISHJAM_ID"]
+FISHJAM_TOKEN = os.environ["FISHJAM_MANAGEMENT_TOKEN"]
+FISHJAM_URL = os.getenv("FISHJAM_URL")
+TRANSCRIPTION_MODEL = "gemini-live-2.5-flash-preview"
+TRANSCRIPTION_CONFIG = LiveConnectConfig(
+    response_modalities=[Modality.TEXT],
+    input_audio_transcription=AudioTranscriptionConfig(),
+)

examples/transcription/transcription/agent.py

@@ -0,0 +1,37 @@
+from fishjam import FishjamNotifier
+from fishjam.events import ServerMessagePeerConnected, ServerMessagePeerDisconnected
+from fishjam.events.allowed_notifications import AllowedNotification
+
+from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+from .room import RoomService
+
+
+def make_notifier(room_service: RoomService):
+    notifier = FishjamNotifier(FISHJAM_ID, FISHJAM_TOKEN, fishjam_url=FISHJAM_URL)
+
+    @notifier.on_server_notification
+    def _(notification: AllowedNotification):
+        print(f"Received notification {notification}")
+        match notification:
+            case ServerMessagePeerConnected(peer_id=peer_id, room_id=room_id):
+                handle_peer_connected(peer_id, room_id)
+
+            case ServerMessagePeerDisconnected(peer_id=peer_id, room_id=room_id):
+                handle_peer_disconnected(peer_id, room_id)
+
+    def handle_peer_connected(peer_id: str, room_id: str):
+        if room_id != room_service.room_id:
+            return
+
+        agent = room_service.create_agent()
+        agent.on_peer_enter(peer_id)
+
+    def handle_peer_disconnected(peer_id: str, room_id: str):
+        if room_id != room_service.room_id:
+            return
+
+        agent = room_service.get_agent()
+        if agent:
+            agent.on_peer_leave(peer_id)
+
+    return notifier

examples/transcription/transcription/config.py

@@ -0,0 +1,63 @@
+from typing import Callable
+
+from fishjam import FishjamClient, FishjamNotifier, Room
+from fishjam.errors import NotFoundError
+
+from .agent import TranscriptionAgent
+from .config import FISHJAM_ID, FISHJAM_TOKEN, FISHJAM_URL
+from .transcription import TranscriptionSession
+
+fishjam = FishjamClient(
+    FISHJAM_ID,
+    FISHJAM_TOKEN,
+    fishjam_url=FISHJAM_URL,
+)
+
+
+class RoomService:
+    def __init__(
+        self,
+        session_factory: Callable[[Callable[[str], None]], TranscriptionSession],
+    ):
+        self.room_id = fishjam.create_room().id
+        self._agent: TranscriptionAgent | None = None
+        self._notifier = FishjamNotifier(
+            FISHJAM_ID,
+            FISHJAM_TOKEN,
+            fishjam_url=FISHJAM_URL,
+        )
+
+        def _make_session(peer_id: str):
+            return session_factory(lambda t: self._handle_transcription(peer_id, t))
+
+        self._session_factory = _make_session
+
+    def get_room(self) -> Room:
+        try:
+            return fishjam.get_room(self.room_id)
+        except NotFoundError:
+            self.clear()
+            room = fishjam.create_room()
+            self.room_id = room.id
+            return room
+
+    def clear(self):
+        fishjam.delete_room(self.room_id)
+
+    def _handle_transcription(self, peer_id: str, text: str):
+        print(f"Peer {peer_id} in room {self.room_id} said: {text}")
+
+    def create_agent(self):
+        if not self._agent or self._agent.done:
+            self._agent = TranscriptionAgent(
+                self.room_id,
+                fishjam.create_agent(self.room_id),
+                self._handle_transcription,
+                self._session_factory,
+            )
+        return self._agent
+
+    def get_agent(self):
+        if self._agent.done:
+            self._agent = None
+        return self._agent

examples/transcription/transcription/notifier.py

@@ -0,0 +1,74 @@
+from asyncio import Event, Queue, Task, TaskGroup
+from contextlib import asynccontextmanager
+from typing import Callable
+
+from google import genai
+from google.genai.live import AsyncSession
+from google.genai.types import Blob
+
+from .config import TRANSCRIPTION_CONFIG, TRANSCRIPTION_MODEL
+
+
+class TranscriptionSession:
+    def __init__(self, on_text: Callable[[str], None]):
+        self._gemini = genai.Client()
+        self._audio_queue = Queue[bytes]()
+        self._end_event = Event()
+        self._model = TRANSCRIPTION_MODEL
+        self._on_text = on_text
+
+    async def start(self):
+        async with self._gemini.aio.live.connect(
+            model=self._model,
+            config=TRANSCRIPTION_CONFIG,
+        ) as session:
+            async with TaskGroup() as tg:
+                send_task = tg.create_task(self._send_loop(session))
+                recv_task = tg.create_task(self._recv_loop(session))
+
+                print("Started transcription session")
+
+                await self._end_event.wait()
+
+                send_task.cancel()
+                recv_task.cancel()
+
+    def transcribe(self, audio: bytes):
+        self._audio_queue.put_nowait(audio)
+
+    def end(self):
+        self._end_event.set()
+
+    async def _send_loop(self, session: AsyncSession):
+        while True:
+            audio_frame = await self._audio_queue.get()
+            await session.send_realtime_input(
+                audio=Blob(data=audio_frame, mime_type="audio/pcm;rate=16000")
+            )
+
+    async def _recv_loop(self, session: AsyncSession):
+        while True:
+            async for res in session.receive():
+                if (
+                    (content := res.server_content)
+                    and (transcription := content.input_transcription)
+                    and (text := transcription.text)
+                ):
+                    self._on_text(text)
+
+
+@asynccontextmanager
+async def transcription_session_factory():
+    async with TaskGroup() as tg:
+        sessions: list[TranscriptionSession] = []
+
+        def make_session(on_text: Callable[[str], None]):
+            session = TranscriptionSession(on_text)
+            sessions.append(session)
+            tg.create_task(session.start())
+            return session
+
+        yield make_session
+
+        for session in sessions:
+            session.end()