"""Streaming speech-to-text — transcribing *while* you talk, not after. The one-shot path (`stt.transcribe`) waits for the utterance to finish, then sends the whole clip and waits again. That second wait is dead time between you stopping and the pet reacting, and it grows with the length of what you said — a thirty-second question costs noticeably more than a five-second one. This connects to the Bolt server's own `/desk/stt` — the same websocket relay the Android app uses — which forwards audio to Deepgram and Deepgram's JSON messages back untouched. Frames go up as they are captured, so by the time the VAD decides you have stopped, the transcript is essentially already there. There is no local Deepgram account or API key any more; auth is this pet's own `DESK_API_KEY`, same as every other call to the server. Design constraints that shaped this: - **Failure must be invisible.** No websocket, no network, a mid-utterance disconnect — all of it falls back to the one-shot path, which still has the full audio buffered. Streaming is an optimisation here, never a dependency — `available()`/`open()` returning None/False is an ordinary outcome, not an error. `stt.transcribe()` (the *guaranteed* fallback) talks to the exact same server relay via `connect()` directly, bypassing that opportunistic gate, since there is no second, different backend left to fall back to. - **The VAD still decides when you stopped.** Deepgram has its own endpointing and using it would save more, but it would also move a decision the rest of the pipeline is built around (barge-in, follow-up listening, the grace period) into a remote service. Not worth coupling those to the network on the first pass. - **The socket is per-utterance.** Holding one open across an idle pet would bill for silence and drop on the first network blip; opening one takes ~100ms, which is already inside the time it takes a person to start talking. The websocket client is injectable, so the whole protocol — send frames, read `is_final` transcripts, close, take the result — is tested without a network. """ from __future__ import annotations import json import logging import threading from typing import Callable, Optional import numpy as np from .. import config logger = logging.getLogger("bolt_pet.stt_stream") def available() -> bool: """Whether the streaming (transcribe-while-talking) optimization should be attempted opportunistically. Not a gate on transcription itself — the server relay is the only way to transcribe at all now, so stt.transcribe() connects via connect() directly rather than through this, and isn't affected by STT_STREAMING being off.""" if not config.STT_STREAMING or not config.is_configured(): return False try: import websocket # noqa: F401 (websocket-client) return True except Exception: return False def _connect(sample_rate: int = None): """Open a websocket to the server's `/desk/stt` relay. Raises on any failure — this is the "no fallback left" connector `stt.transcribe()` uses directly, as well as the default for `StreamingTranscriber.open()`. The server chooses the STT model and the endpointing behaviour; this only states the audio format about to be sent, which is fixed by the wake model upstream of it.""" import websocket rate = sample_rate or config.SAMPLE_RATE url = ( config.SERVER_URL.replace("http", "ws", 1) + "/desk/stt" + f"?session_id={config.SESSION_ID}&encoding=linear16&sample_rate={rate}" ) return websocket.create_connection( url, header=[f"X-Desk-Api-Key: {config.API_KEY}"], timeout=10, ) # Public name for external callers (stt.py, tests) — named separately from # the module-private def so StreamingTranscriber.open()'s `connect` parameter # can shadow the bare name locally without losing access to this. connect = _connect class StreamingTranscriber: """One utterance's worth of live transcription. Usage mirrors how the capture loop already works — feed frames as they arrive, then ask what was said: session = StreamingTranscriber.open() ... session.feed(frame) # per mic frame, non-blocking text = session.finish() # after the VAD says you stopped """ def __init__(self, socket, *, sample_rate: int = None): self._socket = socket self._sample_rate = sample_rate or config.SAMPLE_RATE self._transcript: list[str] = [] self._lock = threading.Lock() self._closed = False self._reader = threading.Thread( target=self._read_loop, name="stt-stream-reader", daemon=True) self._reader.start() # -- lifecycle ---------------------------------------------------------- @classmethod def open(cls, *, connect: Optional[Callable] = None, sample_rate: int = None) -> Optional["StreamingTranscriber"]: """Connect, or return None if streaming isn't possible right now. None is a normal outcome, not a failure: the caller keeps the audio and falls back to the one-shot upload.""" if connect is None and not available(): return None rate = sample_rate or config.SAMPLE_RATE try: socket = connect() if connect is not None else _connect(rate) return cls(socket, sample_rate=rate) except Exception as exc: logger.info("Streaming STT unavailable (%s) — using the one-shot path.", exc) return None def feed(self, frame: np.ndarray) -> None: """Send one captured frame. Never raises — a dead socket just means the fallback will do the work.""" if self._closed: return try: self._socket.send_binary(np.asarray(frame, dtype=np.int16).tobytes()) except Exception: logger.debug("Streaming STT send failed; abandoning the stream", exc_info=True) self._closed = True def finish(self, timeout: float = 3.0) -> str: """Close the stream and return whatever was transcribed. Deepgram flushes its final results after the close frame, so this waits briefly for the reader — bounded, because a hung socket must not hold up the reply.""" if not self._closed: try: self._socket.send(json.dumps({"type": "CloseStream"})) except Exception: pass self._closed = True self._reader.join(timeout=timeout) try: self._socket.close() except Exception: pass with self._lock: return " ".join(part for part in self._transcript if part).strip() # -- the reader --------------------------------------------------------- def _read_loop(self) -> None: while not self._closed: try: message = self._socket.recv() except Exception: break if not message: break text, is_final = self._parse(message) if text and is_final: with self._lock: self._transcript.append(text) @staticmethod def _parse(message) -> tuple[str, bool]: """Pull (text, is_final) out of a Deepgram results frame. Tolerant on purpose: anything unrecognised is ignored rather than raising on the reader thread, where an exception would silently kill transcription for the rest of the utterance.""" try: if isinstance(message, bytes): message = message.decode("utf-8", "ignore") data = json.loads(message) except (TypeError, ValueError): return "", False if not isinstance(data, dict): return "", False alternatives = ( ((data.get("channel") or {}).get("alternatives") or []) if data.get("type") in (None, "Results") else [] ) if not alternatives: return "", False text = str((alternatives[0] or {}).get("transcript") or "").strip() return text, bool(data.get("is_final") or data.get("speech_final"))