bridge

bridge

Bridge browser WebSocket clients to a provider realtime session.

Classes

VoiceLiveBridge

VoiceLiveBridge(
    client_ws: Any,
    session: RealtimeVoiceSession,
    *,
    client_input_rate: int = 16000,
)

Pump audio/control messages between a FastAPI WebSocket and a provider.

Source code in src/diapason/speech/realtime/bridge.py

def __init__(
    self,
    client_ws: Any,
    session: RealtimeVoiceSession,
    *,
    client_input_rate: int = 16000,
) -> None:
    self._client = client_ws
    self._session = session
    self._client_input_rate = client_input_rate
    self._tasks: list[asyncio.Task[None]] = []

Functions:

event_to_client_json

event_to_client_json(event: SessionEvent) -> dict[str, Any]

Serialize a :class:SessionEvent for the browser protocol.

Source code in src/diapason/speech/realtime/bridge.py

def event_to_client_json(event: SessionEvent) -> dict[str, Any]:
    """Serialize a :class:`SessionEvent` for the browser protocol."""
    if event.kind == "ready":
        return {"type": "ready"}
    if event.kind == "audio":
        return {
            "type": "audio",
            "data": event.audio_b64,
            "sample_rate": event.sample_rate,
        }
    if event.kind == "transcript":
        return {
            "type": "transcript",
            "role": event.role,
            "text": event.text,
            "final": event.final,
            "replace": event.replace,
        }
    if event.kind == "interrupted":
        return {"type": "interrupted"}
    if event.kind == "tool":
        return {
            "type": "tool",
            "name": event.tool_name,
            "ok": event.tool_ok,
            "detail": event.detail,
        }
    if event.kind == "error":
        return {"type": "error", "detail": event.detail or "unknown error"}
    if event.kind == "closed":
        return {"type": "closed"}
    return {"type": "error", "detail": f"unknown event {event.kind}"}