tts speaks MP3: a seventh of WAV's bytes over a slow link to the server

A sentence's length for `append` comes from the Xing header LAME writes into
the first frame. The cache key carries the format, so no WAV is reused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ljdk7nDaguh9Uhi3zhcSjn

backend/fluksio/ai/client.py

@@ -189,7 +189,11 @@ async def speak(
     model: str = "tts-1",
     timeout: float = 30.0,
 ) -> bytes:
-    """``POST {api_base}/audio/speech``: ``text`` spoken, as WAV."""
+    """``POST {api_base}/audio/speech``: ``text`` spoken, as MP3.
+
+    MP3 rather than WAV: a seventh of the bytes, which over a slow link to
+    the server is most of the wait for a sentence.
+    """
     async with httpx.AsyncClient(timeout=timeout) as http:
         response = await http.post(
             api_base.rstrip("/") + "/audio/speech",
@@ -197,7 +201,7 @@ async def speak(
                 "model": model,
                 "input": text,
                 "voice": voice,
-                "response_format": "wav",
+                "response_format": "mp3",
             },
         )
     if response.is_error:

backend/fluksio/flow/nodes/speech.py

@@ -10,12 +10,10 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
-import io
 import logging
 import re
 import threading
 import time
-import wave
 from collections.abc import Coroutine, Iterable
 from concurrent.futures import Future
 from typing import TYPE_CHECKING, Any
@@ -68,6 +66,23 @@ def sentences(text: str) -> list[str]:
     return [s for s in (p.strip() for p in re.split(r"(?<=[.!?])\s+", text)) if s]
 
 
+def mp3_seconds(data: bytes) -> float:
+    """An MP3's length, from the frame count LAME writes into its first frame."""
+    header = int.from_bytes(data[:4], "big")
+    version = (header >> 19) & 3  # 3 is MPEG-1, 2 MPEG-2, 0 MPEG-2.5
+    rates = {
+        3: (44100, 48000, 32000),
+        2: (22050, 24000, 16000),
+        0: (11025, 12000, 8000),
+    }
+    rate = rates[version][(header >> 10) & 3]
+    at = max(data.find(b"Xing", 0, 64), data.find(b"Info", 0, 64))
+    if at < 0:
+        raise ValueError("the MP3 has no Xing or Info header to read its length from")
+    frames = int.from_bytes(data[at + 8 : at + 12], "big")
+    return round(frames * (1152 if version == 3 else 576) / rate, 3)
+
+
 class _Speech(Node):
     """What both nodes share: the url, the artifact store and their health."""
 
@@ -334,19 +349,17 @@ class TtsNode(_Speech):
 
     def _spoken(self, url: str, sentence: str) -> dict[str, Any]:
         """The reference to ``sentence`` spoken, from the cache or the server."""
-        key = hashlib.sha256(f"{self.cfg.voice}\n{sentence}".encode()).hexdigest()
+        key = hashlib.sha256(f"mp3\n{self.cfg.voice}\n{sentence}".encode()).hexdigest()
         # Kept in state, so the sweep keeps the bytes it refers to.
         spoken = dict(self.recall("spoken") or {})
         reference = spoken.get(key)
         if reference is not None and self._artifacts.has(reference["digest"]):
             return dict(reference)
         data = asyncio.run(client.speak(url + "/v1", sentence, self.cfg.voice))
-        with wave.open(io.BytesIO(data)) as clip:
-            seconds = round(clip.getnframes() / clip.getframerate(), 3)
         stored: dict[str, Any] = self._artifacts.put(
-            [data], name="speech.wav", media_type="audio/wav"
+            [data], name="speech.mp3", media_type="audio/mpeg"
         )
-        reference = stored | {"meta": {"seconds": seconds}}
+        reference = stored | {"meta": {"seconds": mp3_seconds(data)}}
         spoken.pop(key, None)
         spoken[key] = reference
         self.remember("spoken", dict(list(spoken.items())[-SPOKEN:]))

backend/tests/flow/test_speech.py

@@ -2,8 +2,6 @@
 
 import asyncio
 import hashlib
-import io
-import wave
 from typing import Any
 
 from fluksio.ai import client
@@ -105,14 +103,11 @@ def test_an_utterance_is_one_session_committed_when_listening_ends(monkeypatch):
     asyncio.run(scenario())
 
 
-def _wav(seconds: float) -> bytes:
-    out = io.BytesIO()
-    with wave.open(out, "wb") as clip:
-        clip.setnchannels(1)
-        clip.setsampwidth(2)
-        clip.setframerate(24000)
-        clip.writeframes(b"\0\0" * int(24000 * seconds))
-    return out.getvalue()
+def _mp3(frames: int) -> bytes:
+    """The first frame LAME writes: MPEG-2 layer III at 24 kHz, mono, and the
+    Xing header counting the frames, 576 samples each."""
+    xing = b"Xing" + (15).to_bytes(4, "big") + frames.to_bytes(4, "big")
+    return bytes.fromhex("fff384c4") + bytes(9) + xing + bytes(100)
 
 
 def test_a_sentence_is_spoken_once_and_the_rest_of_a_reply_queues(monkeypatch):
@@ -121,7 +116,7 @@ def test_a_sentence_is_spoken_once_and_the_rest_of_a_reply_queues(monkeypatch):
 
     async def speak(api_base, text, voice):
         spoken.append(text)
-        return _wav(len(spoken))
+        return _mp3(25 * len(spoken))
 
     monkeypatch.setattr(client, "speak", speak)
     monkeypatch.setattr(TtsNode, "_say", lambda self, speech: published.append(speech))
@@ -144,4 +139,5 @@ def test_a_sentence_is_spoken_once_and_the_rest_of_a_reply_queues(monkeypatch):
     assert spoken == ["Kitchen light on.", "Heating off."]
     assert ["append" in speech for speech in published] == [False, True, False]
     assert published[2] == published[0]
-    assert published[1]["meta"] == {"seconds": 2.0}
+    assert published[1]["meta"] == {"seconds": 1.2}
+    assert published[1]["media_type"] == "audio/mpeg"

docs/reference/node-types.md

@@ -453,7 +453,7 @@ such as one that sends whole clips.
 
 | Output | Type | Notes |
 |---|---|---|
-| `speech` | `audio` | one WAV reference per sentence |
+| `speech` | `audio` | one MP3 reference per sentence |
 
 | Setting | Default | Notes |
 |---|---|---|