diff --git a/bolt_assistant/assistant.py b/bolt_assistant/assistant.py index 861cd2b..0cb79f1 100644 --- a/bolt_assistant/assistant.py +++ b/bolt_assistant/assistant.py @@ -12,6 +12,11 @@ Speaking is half-duplex: the mic is ignored while Bolt talks and for a moment after, so his own voice through the speakers can't wake him (his replies say "Bolt" often enough). +Transcribing is not: waiting for the server's transcript happens on a +background thread, so the mic keeps being segmented meanwhile and whatever +is said next becomes its own utterance and is sent off too. Finished +transcripts are delivered oldest-first between mic frames. + Everything that touches hardware or the network is injected, so the flow is tested with fakes; __main__ wires in the real mic, server and speaker. """ @@ -22,11 +27,12 @@ import platform import queue import threading import time +from collections import deque from typing import Callable, Optional from . import audio, config, server, speech_text from .listener import Segmenter -from .stt import SttError, Transcription +from .stt import Transcription from .tools import Tools from .wake import find_wake @@ -57,6 +63,17 @@ def annotate(text: str, commands: str) -> str: f"{can} Never make up a command's output.]") +class _PendingTranscript: + """One utterance already sent off, its transcript still on the way.""" + + __slots__ = ("thread", "text", "error") + + def __init__(self): + self.thread: Optional[threading.Thread] = None + self.text = "" + self.error: Optional[Exception] = None + + class Assistant: def __init__( self, @@ -89,6 +106,7 @@ class Assistant: self._deadline = 0.0 self._follow_ups = 0 self._stt = None + self._pending: "deque[_PendingTranscript]" = deque() self._mute_until = 0.0 self._voice = "" # a voice the server picked (speak_as) sticks for later replies self._tts_failed = False @@ -109,8 +127,9 @@ class Assistant: frame = self.frames.get(timeout=0.2) except queue.Empty: self._check_deadline() - continue - self.process(frame) + else: + self.process(frame) + self.handle_transcripts() def listening_banner(self) -> str: return 'Listening for "Hey Bolt" (or "Thunderbolt", "Hey Assistant"). Ctrl+C to quit.' @@ -131,18 +150,53 @@ class Assistant: self._stt = None elif event.kind == "end": stt, self._stt = self._stt, None - try: - text = stt.finish(event.audio) if stt is not None else "" - except SttError as exc: - self.log(f"! transcription failed: {exc}") - if self.state != "idle": - self._chime(audio.ERROR_CHIME) - continue - self.heard(text) + if stt is not None: + self._start_finish(stt, event.audio) + + def _start_finish(self, stt, utterance: bytes) -> None: + """finish() waits on the server; done here, that wait would deafen + the loop. Run it on a thread and keep segmenting the mic meanwhile.""" + pending = _PendingTranscript() + + def work() -> None: + try: + pending.text = stt.finish(utterance) + except Exception as exc: # SttError, or the socket misbehaving + pending.error = exc + + pending.thread = threading.Thread(target=work, name="stt-finish", daemon=True) + self._pending.append(pending) + pending.thread.start() + + def handle_transcripts(self, wait: bool = False) -> None: + """Deliver finished transcripts, oldest first. One still in flight — + or an utterance still being spoken — ends the pass rather than + blocking, unless *wait* (tests, shutdown). The head stays queued + while heard() runs so _check_deadline knows speech already happened.""" + while self._pending: + if self.segmenter.in_speech and not wait: + return + head = self._pending[0] + if wait: + head.thread.join() + if head.thread.is_alive(): + return + if head.error is not None: + self._pending.popleft() + self.log(f"! transcription failed: {head.error}") + if self.state != "idle": + self._chime(audio.ERROR_CHIME) + continue + try: + self.heard(head.text) + finally: + self._pending.popleft() def _check_deadline(self) -> None: + # A pending transcript means something *was* said in time — it's + # just still at the server; the window mustn't close on that latency. if self.state in ("command", "follow_up") and self.clock() > self._deadline \ - and not self.segmenter.in_speech: + and not self.segmenter.in_speech and not self._pending: if self.verbose: self.log(" (stopped waiting)") self.state = "idle" diff --git a/tests/test_assistant.py b/tests/test_assistant.py index 19e4b8f..2c45926 100644 --- a/tests/test_assistant.py +++ b/tests/test_assistant.py @@ -111,19 +111,22 @@ def test_long_speech_is_cut_off_at_the_cap(): # ── the flow ──────────────────────────────────────────────────────────────── class FakeTranscription: - """Each utterance returns the next scripted transcript.""" + """Each utterance takes the next scripted transcript. Bound at creation + (the "start" event, on the main thread) because finish() now runs on a + worker thread per utterance and two may be in flight at once.""" script = [] def __init__(self): self.frames = 0 self.cancelled = False + self.transcript = FakeTranscription.script.pop(0) def add(self, frames): self.frames += len(frames) def finish(self, audio): - return FakeTranscription.script.pop(0) + return self.transcript def cancel(self): self.cancelled = True @@ -141,7 +144,7 @@ def utterance(): return [frame(LOUD)] * 8 + [frame(QUIET)] * 12 -def make(replies, commands="read", streaming=False, **kw): +def make(replies, commands="read", streaming=False, transcription=FakeTranscription, **kw): asked, played, logs = [], [], [] def converse(message, on_command, on_say=None): @@ -152,7 +155,7 @@ def make(replies, commands="read", streaming=False, **kw): return server.Reply(reply) clock = Clock() - a = Assistant(transcription=FakeTranscription, converse=converse, + a = Assistant(transcription=transcription, converse=converse, converse_stream=None if not streaming else kw.pop("stream"), tts=lambda text, voice: [text.encode()], play=lambda chunks: played.extend(chunks), tools=Tools(mode=commands, log=logs.append), log=logs.append, clock=clock, **kw) @@ -163,6 +166,7 @@ def say(a, transcript): FakeTranscription.script.append(transcript) for f in utterance(): a.process(f) + a.handle_transcripts(wait=True) # what run() does between frames def spoken(played): @@ -225,6 +229,27 @@ def test_mic_audio_during_speech_is_discarded(): assert not a.segmenter.in_speech +def test_listening_continues_while_a_transcript_is_in_flight(): + import threading + + release = threading.Event() + + class Slow(FakeTranscription): + def finish(self, audio): + release.wait(timeout=5) + return self.transcript + + a, asked, played, logs, _ = make(["One.", "Two."], transcription=Slow) + FakeTranscription.script += ["hey bolt first thing", "hey bolt second thing"] + for f in utterance() + utterance(): # the second said while the first transcribes + a.process(f) + assert asked == [] # both transcripts still at the "server" + release.set() + a.handle_transcripts(wait=True) + assert len(asked) == 2 # nothing said during the wait was lost + assert asked[0].startswith("first thing") and asked[1].startswith("second thing") + + def test_streamed_sentences_are_spoken_as_they_arrive(): def stream(message, on_say, on_command): on_say("First.")