"""Streaming speech-to-text — transcribing *while* you talk, not after. The one-shot path (`stt.transcribe`) waits for the utterance to finish, then uploads the whole WAV 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. Deepgram's live endpoint removes it: frames go up as they are captured, so by the time the VAD decides you have stopped, the transcript is essentially already there. Same model, same account, same accuracy — the difference is purely when the work happens. 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, never a dependency, so `open()` returning None is an ordinary outcome rather than an error. - **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") _ENDPOINT = ( "wss://api.deepgram.com/v1/listen" "?encoding=linear16&channels=1&sample_rate={rate}&model={model}" "&language=en&smart_format=true&interim_results=false" ) def available() -> bool: """Whether streaming STT can even be attempted in this install.""" if not config.STT_STREAMING or not config.DEEPGRAM_API_KEY: return False try: import websocket # noqa: F401 (websocket-client) return True except Exception: return False 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: if connect is not None: socket = connect() else: import websocket socket = websocket.create_connection( _ENDPOINT.format(rate=rate, model=config.DEEPGRAM_MODEL), header={"Authorization": f"Token {config.DEEPGRAM_API_KEY}"}, timeout=10, ) 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"))