Bolt-Pet/bolt_pet/speech.py

"""Everything the pet says, and the policy differences between kinds of saying.

There used to be four of these in controller.py — a reply, a holding line, a
streamed sentence, a dialogue scene — each written when its feature was built,
each repeating the same dance: transition state, show the bubble, maybe record
history, reset barge-in, call TTS, reset barge-in again, resume state, decide
whether to keep the mic open. Only the *policy* differed, and the copies had
already started to drift: one forgot to arm barge-in, another logged a
different prefix, a third skipped the follow-up rule.

So the dance lives here once, and the differences are data:

    reply        the answer. Transcript, bubble, follow-up rule, ends IDLE.
    holding      "give me a sec" while a tool runs. No transcript — it is
                 filler, and the transcript should keep the answer. Resumes
                 whatever state it interrupted, because the turn isn't over.
    stream       one sentence of a streamed answer. Transcript and bubble like
                 a reply, but stays TALKING so the sprite doesn't flicker
                 between sentences, and the follow-up rule waits for the last.
    scene        a dialoguectl take. Transcript (the user heard it), resumes
                 mid-turn like a holding line.

The controller keeps the pipeline; this keeps the rules about talking.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, Optional

from . import config, speech_text
from .state import PetState


@dataclass(frozen=True)
class Utterance:
    """One thing to say, and how saying it should behave."""

    text: str
    record: bool = True         # goes in the transcript, or is it filler?
    resume: bool = False        # return to the state it interrupted (mid-turn)
    hold_talking: bool = False  # stay TALKING afterwards (more is coming)
    follow_up: bool = True      # may leave the mic open if it ends on a question
    interruptible: bool = True  # arm barge-in for this one
    log_prefix: str = "Bolt"

    @classmethod
    def reply(cls, text: str) -> "Utterance":
        return cls(text)

    @classmethod
    def holding(cls, text: str) -> "Utterance":
        # Filler: no transcript, no follow-up, and not interruptible — cutting
        # off "give me a sec" would strand the tool that is already running.
        return cls(text, record=False, resume=True, follow_up=False,
                   interruptible=False, log_prefix="Bolt (holding)")

    @classmethod
    def stream_chunk(cls, text: str) -> "Utterance":
        return cls(text, hold_talking=True, follow_up=False)

    @classmethod
    def scene(cls, text: str) -> "Utterance":
        return cls(text, resume=True, follow_up=False)


class Speaker:
    """Says things on behalf of the controller.

    Collaborators are passed in rather than reached for, so the whole of
    speaking is testable without Qt, audio hardware or a state machine: hand it
    fakes and assert on what came out."""

    def __init__(self, *, state, tts, history=None, on_said=None, on_log=None,
                 barge_in: Callable[[], object] = lambda: None,
                 voice_id: Callable[[], str] = lambda: "",
                 detail_of: Callable[[], str] = lambda: "",
                 on_level: Optional[Callable[[float], None]] = None):
        self._state = state
        self._tts = tts
        self._history = history
        self._on_said = on_said or (lambda _text: None)
        self._on_log = on_log or (lambda _msg: None)
        # Read through a callable rather than held: the detector is built after
        # the speaker (it needs the mic stream), replaced when barge-in mode
        # changes, and swapped by tests. Two copies of it drifting apart is a
        # bug nobody notices until the wake model starts hearing the pet.
        self._barge_in_of = barge_in
        self._voice_id = voice_id
        # What actually fired, captured BEFORE the post-playback reset. Read it
        # afterwards and every interruption reports 0.000 at frame 0 — which
        # looks like hard evidence and is nothing of the sort.
        self._detail_of = detail_of
        self.last_detail = ""
        self._on_level = on_level

    @property
    def _barge_in(self):
        try:
            return self._barge_in_of()
        except Exception:
            return None

    def say(self, utterance: Utterance) -> bool:
        """Speak it. Returns False if it was interrupted.

        The one place that knows the order these steps go in — which is the
        point, because getting that order wrong is invisible until the wake
        model starts hearing the pet's own voice."""
        line = speech_text.for_display(utterance.text)
        if not line:
            return True

        resume_state = self._state.state
        if self._state.state != PetState.TALKING:
            self._state.transition(PetState.TALKING)
        self._on_said(line)
        self._on_log(f"{utterance.log_prefix}: {line}")
        if utterance.record and self._history is not None:
            self._history(utterance.text)

        should_stop = None
        if self._barge_in is not None and utterance.interruptible:
            self._barge_in.reset()
            should_stop = self._barge_in.check

        completed = self._tts.speak(
            utterance.text,
            on_error=lambda exc: self._on_log(f"TTS failed: {exc}"),
            should_stop=should_stop,
            voice_id=self._voice_id() or None,
            on_level=self._on_level,
        )
        self.last_detail = self._detail_of()
        if self._barge_in is not None:
            # Playback fed the pet's own voice into the wake model's rolling
            # window. Clear it before the idle listener scores again, or the
            # last sentence is still in there being re-heard.
            self._barge_in.reset()
        if self._on_level is not None:
            self._on_level(0.0)  # mouth closed; nothing is playing now

        if utterance.resume and resume_state in (PetState.THINKING, PetState.IDLE):
            self._state.transition(resume_state)
        return bool(completed)


    def say_pcm(self, utterance: Utterance, *, pcm, sample_rate: int) -> bool:
        """Speak audio that is already synthesized — a dialoguectl scene.

        Same policy, same barge-in handling, same mouth; only the source of
        the samples differs. Sharing this is why a scene can be talked over
        exactly like an ordinary reply."""
        line = speech_text.for_display(utterance.text)
        resume_state = self._state.state
        if self._state.state != PetState.TALKING:
            self._state.transition(PetState.TALKING)
        if line:
            self._on_said(line)
        self._on_log(f"{utterance.log_prefix}: {line}")
        if utterance.record and self._history is not None:
            self._history(utterance.text)

        should_stop = None
        if self._barge_in is not None and utterance.interruptible:
            self._barge_in.reset()
            should_stop = self._barge_in.check
        completed = self._tts.play_pcm(
            pcm, sample_rate, should_stop=should_stop, on_level=self._on_level)
        self.last_detail = self._detail_of()
        if self._barge_in is not None:
            self._barge_in.reset()
        if self._on_level is not None:
            self._on_level(0.0)
        if utterance.resume and resume_state in (PetState.THINKING, PetState.IDLE):
            self._state.transition(resume_state)
        return bool(completed)


def follow_up_decision(text: str, *, completed: bool, follow_ups: int) -> tuple[bool, str]:
    """Whether to keep listening after speaking, and why.

    Split out as a pure function because it is a *rule* with an off-by-one cap
    in it, and rules with counters deserve a test that doesn't need audio."""
    if not completed:
        return True, "interrupted"
    if not speech_text.is_question(text):
        return False, ""
    cap = config.FOLLOW_UP_MAX_TURNS
    if not config.FOLLOW_UP_LISTEN:
        return False, ""
    if cap > 0 and follow_ups >= cap:
        return False, f"follow-up cap ({cap}) reached"
    return True, "question"