Refactor call classes

voip/ai.py

@@ -11,6 +11,7 @@
 
 import asyncio
 import dataclasses
+import io
 import logging
 from typing import Any
 
@@ -71,54 +72,19 @@ def __post_init__(self) -> None:
         else:
             self.whisper_model = self.model
 
-    def collect_audio(self, audio: np.ndarray, rms: float) -> bool:
-        """Buffer all audio frames (speech and silence) for transcription.
-
-        Args:
-            audio: Decoded float32 PCM frame.
-            rms: Root mean square of *audio*.
-
-        Returns:
-            Always `True` so that intra-utterance silences are preserved.
-        """
-        return True
-
     async def speech_buffer_ready(self, audio: np.ndarray) -> None:
-        """Transcribe the buffered utterance when it meets the minimum length.
-
-        Skips utterances shorter than one second to avoid passing fragments
-        to Whisper that would produce low-quality transcriptions.
-
-        Args:
-            audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz.
-        """
-        if len(audio) < self.RESAMPLING_RATE_HZ:
-            return
         await self.transcribe(audio)
 
     async def transcribe(self, audio: np.ndarray) -> None:
-        """Transcribe decoded audio and deliver non-empty text to the handler.
-
-        Args:
-            audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz.
-        """
         loop = asyncio.get_running_loop()
         raw = await loop.run_in_executor(None, self.run_transcription, audio)
         if text := raw.strip():
             self.transcription_received(text)
 
     def run_transcription(self, audio: np.ndarray) -> str:
-        """Transcribe a float32 PCM array using the Whisper model.
-
-        Args:
-            audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz.
-
-        Returns:
-            Concatenated transcription text from all segments.
-        """
         segments, _ = self.whisper_model.transcribe(audio)
         result = "".join(segment.text for segment in segments)
-        logger.debug("Transcription result: %r", segments)
+        logger.debug("Transcription result: %r", result)
         return result
 
     def transcription_received(self, text: str) -> None:
@@ -166,7 +132,9 @@ class AgentCall(TranscribeCall):
     tts_instance: TTSModel = dataclasses.field(init=False, repr=False)
     voice_state: Any = dataclasses.field(init=False, repr=False)
     messages: list[dict] = dataclasses.field(init=False, repr=False)
-    pending_text: list[str] = dataclasses.field(init=False, repr=False)
+    pending_text: io.StringIO = dataclasses.field(
+        init=False, repr=False, default_factory=io.StringIO
+    )
     response_task: asyncio.Task | None = dataclasses.field(init=False, repr=False)
 
     def __post_init__(self) -> None:
@@ -180,43 +148,32 @@ def __post_init__(self) -> None:
                 + "\n\nYOU MUST NEVER USE NON-VERBAL CHARACTERS IN YOUR RESPONSES!",
             }
         ]
-        self.pending_text = []
         self.response_task = None
 
     def transcription_received(self, text: str) -> None:
-        match text:
-            case "":
-                return
-            case _:
-                self.pending_text.append(text)
-                if self.response_task is not None and not self.response_task.done():
-                    self.response_task.cancel()
-                self.response_task = asyncio.create_task(self.respond())
+        self.pending_text.writelines((text, "\n"))
+        if self.response_task is not None and not self.response_task.done():
+            self.response_task.cancel()
+        self.response_task = asyncio.create_task(self.respond())
 
     async def respond(self) -> None:
         """Fetch an Ollama reply for pending text and stream it as speech via RTP.
 
         On cancellation (human started speaking) the partial user turn is
         removed from the chat history so the history stays consistent.
         """
-        self.messages.append({"role": "user", "content": "\n".join(self.pending_text)})
-        self.pending_text.clear()
-        try:
-            response = await ollama.AsyncClient().chat(
-                model=self.ollama_model,
-                messages=self.messages,
-            )
-            reply = (response.message.content or "").encode("ascii", "ignore").decode()
-            self.messages.append({"role": "assistant", "content": reply})
-            logger.info("Agent reply: %r", reply)
-            await self.send_speech(reply)
-        except asyncio.CancelledError:
-            # Remove the partial user turn so history stays consistent.
-            if self.messages and self.messages[-1]["role"] == "user":
-                self.messages.pop()
-            raise
-        except Exception:
-            logger.exception("Error while generating agent response")
+        self.messages.append({"role": "user", "content": self.pending_text.getvalue()})
+        self.pending_text.seek(0)
+        self.pending_text.truncate(0)
+        response = await ollama.AsyncClient().chat(
+            model=self.ollama_model,
+            messages=self.messages,
+        )
+        reply = (response.message.content or "").encode("ascii", "ignore").decode()
+        self.messages.append({"role": "assistant", "content": reply})
+
+        logger.debug("Agent reply: %r", reply)
+        await self.send_speech(reply)
 
     async def send_speech(self, text: str) -> None:
         """Stream synthesised speech from Pocket TTS and send via RTP.
@@ -228,23 +185,12 @@ async def send_speech(self, text: str) -> None:
         Args:
             text: Text to synthesise and transmit.
         """
-        loop = asyncio.get_running_loop()
-        queue: asyncio.Queue[np.ndarray | None] = asyncio.Queue()
-
-        def generate() -> None:
-            for chunk in self.tts_instance.generate_audio_stream(
-                self.voice_state,
-                text,  # type: ignore[too-many-positional-arguments]
-            ):
-                asyncio.run_coroutine_threadsafe(
-                    queue.put(chunk.numpy()), loop
-                ).result()
-            asyncio.run_coroutine_threadsafe(queue.put(None), loop).result()
-
-        future = loop.run_in_executor(None, generate)
-        while (tts_chunk := await queue.get()) is not None:
-            resampled = self.resample(
-                tts_chunk, self.tts_instance.sample_rate, self.codec.sample_rate_hz
+        audio = self.tts_instance.generate_audio(
+            self.voice_state,
+            text,  # type: ignore[too-many-positional-arguments]
+        )
+        await self.send_rtp_audio(
+            self.resample(
+                audio.numpy(), self.tts_instance.sample_rate, self.codec.sample_rate_hz
             )
-            await self.send_rtp_audio(resampled)
-        await future
+        )

voip/audio.py

@@ -192,37 +192,21 @@ async def emit_audio(self, packet: RTPPacket) -> None:
         Args:
             packet: Parsed RTP packet whose payload will be decoded.
         """
-        loop = asyncio.get_running_loop()
-        audio = await loop.run_in_executor(None, self.decode_payload, packet.payload)
+        audio = self.decode_payload(packet.payload)
         if audio.size > 0:
             self.audio_received(
                 audio=audio, rms=float(np.sqrt(np.mean(np.square(audio))))
             )
 
     def decode_payload(self, payload: bytes) -> np.ndarray:
-        """Decode an RTP payload to float32 PCM at `RESAMPLING_RATE_HZ`.
-
-        Delegates to `payload_decoder`, which is either a
-        [`PerPacketDecoder`][voip.codecs.base.PerPacketDecoder] (for stateless
-        codecs such as PCMA, PCMU, Opus) or a
-        [`G722Decoder`][voip.codecs.g722.G722Decoder] (for G.722, which
-        preserves ADPCM predictor state across consecutive packets).
-
-        Args:
-            payload: Raw RTP payload bytes.
-
-        Returns:
-            Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz.
-        """
         return self.payload_decoder.decode(payload)
 
     def audio_received(self, *, audio: np.ndarray, rms: float) -> None:
         """Handle decoded audio.  Override in subclasses.
 
         Args:
             audio: Float32 mono PCM array at `RESAMPLING_RATE_HZ` Hz.
-            rms: Root mean square of the decoded PCM, as a proxy for signal
-                strength.
+            rms: Root Mean Square of the decoded PCM, as a proxy for signal strength.
         """
 
     async def send_rtp_audio(self, audio: np.ndarray) -> None:
@@ -310,63 +294,51 @@ class VoiceActivityCall(AudioCall):
     """
 
     speech_threshold: float = dataclasses.field(default=0.001)
-    silence_gap: float = dataclasses.field(default=0.5)
+    silence_gap: float = dataclasses.field(default=0.2)
 
     speech_buffer: list[np.ndarray] = dataclasses.field(
         init=False, repr=False, default_factory=list
     )
     silence_handle: asyncio.TimerHandle | None = dataclasses.field(
         init=False, repr=False, default=None
     )
+    transcription_handle: asyncio.TimerHandle | None = dataclasses.field(
+        init=False, repr=False, default=None
+    )
 
     def audio_received(self, *, audio: np.ndarray, rms: float) -> None:
-        if self.collect_audio(audio, rms):
-            self.speech_buffer.append(audio)
+        self.speech_buffer.append(audio)
         if rms > self.speech_threshold:
             self.on_audio_speech()
         else:
             self.on_audio_silence()
 
-    def collect_audio(self, audio: np.ndarray, rms: float) -> bool:
-        """Return whether to buffer this audio frame.
-
-        The default implementation buffers speech frames only (RMS above
-        `speech_threshold`).  Override to change the buffering strategy.
-
-        Args:
-            audio: Decoded float32 PCM frame.
-            rms: Root mean square of *audio*.
-
-        Returns:
-            `True` when the frame should be appended to `speech_buffer`.
-        """
-        return rms > self.speech_threshold
-
     def on_audio_speech(self) -> None:
-        """Cancel any pending silence timer when speech is detected."""
-        if self.silence_handle is not None:
-            self.silence_handle.cancel()
-            self.silence_handle = None
+        if self.transcription_handle is not None:
+            self.transcription_handle.cancel()
+            self.transcription_handle = None
 
     def on_audio_silence(self) -> None:
-        """Arm the silence debounce timer when speech is buffered."""
-        if self.silence_handle is None and self.speech_buffer:
-            loop = asyncio.get_running_loop()
-            self.silence_handle = loop.call_later(
+        if self.transcription_handle is None:
+            loop = asyncio.get_event_loop()
+            self.transcription_handle = loop.call_later(
                 self.silence_gap,
                 self.flush_speech_buffer,
             )
 
     def flush_speech_buffer(self) -> None:
-        """Concatenate buffered audio and schedule [`speech_buffer_ready`][voip.audio.VoiceActivityCall.speech_buffer_ready].
-
-        Resets speech state so the next utterance starts with a clean buffer.
-        """
-        self.silence_handle = None
-        if not self.speech_buffer:
-            return
+        self.transcription_handle = None
+        # Ensure at least one second of audio to avoid cutting words in half.
         audio = np.concatenate(self.speech_buffer)
+        if (
+            sum(len(c) for c in self.speech_buffer)
+            < self.RESAMPLING_RATE_HZ * self.silence_gap
+            or float(np.sqrt(np.mean(np.square(audio)))) < 0.01
+        ):
+            self.speech_buffer.clear()
+            return
         self.speech_buffer.clear()
+
         asyncio.create_task(self.speech_buffer_ready(audio))
 
     async def speech_buffer_ready(self, audio: np.ndarray) -> None: