diff --git a/hermes_cli/web_routers/audio.py b/hermes_cli/web_routers/audio.py index 992ba1a69a..340114d75e 100644 --- a/hermes_cli/web_routers/audio.py +++ b/hermes_cli/web_routers/audio.py @@ -376,6 +376,100 @@ async def tts_lease(payload: TTSLeaseRequest, profile: Optional[str] = None): return {"ok": True, "lease": lease, "active": payload.active, **result} +class _SyncSentencePCMStreamer: + """Speak one sentence through the sync TTS tool and yield int16 mono PCM. + + Not a second provider: ``text_to_speech_tool`` is the same stack the CLI + speaker uses for edge and every other non-chunked provider. The desktop + socket only plays PCM, so the written file is decoded before it is sent. + ``sample_rate`` is updated before the first yield, matching the chunked + streamers whose rate is only final once synthesis has answered. + """ + + sample_rate = 24000 + channels = 1 + + def stream(self, text: str): + pcm, rate = _sync_sentence_to_pcm(text) + if rate: + self.sample_rate = rate + if pcm: + yield pcm + + +def _sync_sentence_to_pcm(text: str) -> tuple: + """Synthesize *text* with the configured sync provider and return PCM.""" + from tools.tts_tool import text_to_speech_tool + + fd, tmp_path = tempfile.mkstemp(suffix=".mp3") + os.close(fd) + extra = None + try: + raw = text_to_speech_tool(text=text, output_path=tmp_path) + try: + payload = json.loads(raw) if isinstance(raw, str) else raw + except (TypeError, ValueError) as exc: + raise RuntimeError("Invalid TTS response") from exc + if not isinstance(payload, dict) or not payload.get("success"): + detail = str(payload.get("error") or "") if isinstance(payload, dict) else "" + raise RuntimeError(detail or "Speech synthesis failed") + written = payload.get("file_path") or tmp_path + if not isinstance(written, str) or not os.path.isfile(written) or os.path.getsize(written) <= 0: + raise RuntimeError("Audio file missing") + if os.path.abspath(written) != os.path.abspath(tmp_path): + extra = written + pcm, rate = _audio_file_to_pcm(written) + if not pcm: + raise RuntimeError("TTS audio decoded to silence") + return pcm, rate + finally: + _unlink_quietly(tmp_path) + if extra: + _unlink_quietly(extra) + + +def _audio_file_to_pcm(path: str) -> tuple: + """Decode *path* to int16 mono PCM. WAV via the stdlib; anything else via ffmpeg.""" + with open(path, "rb") as fh: + head = fh.read(12) + if len(head) >= 12 and head.startswith(b"RIFF") and head[8:12] == b"WAVE": + pcm, rate = _wav_s16le_mono(path) + if pcm: + return pcm, rate + return _ffmpeg_s16le_mono(path) + + +def _wav_s16le_mono(path: str) -> tuple: + import wave + + try: + with wave.open(path, "rb") as wf: + if wf.getsampwidth() != 2 or wf.getnchannels() != 1 or wf.getframerate() <= 0: + return b"", 0 + return wf.readframes(wf.getnframes()), int(wf.getframerate()) + except (wave.Error, EOFError, OSError): + return b"", 0 + + +def _ffmpeg_s16le_mono(path: str) -> tuple: + import shutil + + from tools.tts_tool_delivery import _ffmpeg_run + + ffmpeg = shutil.which("ffmpeg") + if not ffmpeg: + raise RuntimeError("ffmpeg is required to decode TTS audio for speak-stream") + result = _ffmpeg_run( + ffmpeg, + ["-i", path, "-f", "s16le", "-ac", "1", "-ar", "24000", "-loglevel", "error", "pipe:1"], + timeout=60, + ) + if result.returncode != 0 or not result.stdout: + stderr = (result.stderr or b"").decode("utf-8", "replace")[:200] + raise RuntimeError(f"TTS audio decode failed: {stderr}") + return result.stdout, 24000 + + @router.websocket("/api/audio/speak-stream") async def speak_stream_ws(ws: "WebSocket") -> None: """Streaming TTS for the desktop: text in, raw int16 PCM frames out. @@ -392,8 +486,10 @@ async def speak_stream_ws(ws: "WebSocket") -> None: server → ``{"type": "start", "sample_rate": N, "channels": 1}`` (sent with the first PCM frame, once the provider's rate is final), binary PCM frames, then ``{"type": "end"}`` - server → ``{"type": "fallback"}`` when the configured provider has no - chunked API — the client uses the POST endpoint instead. + server → ``{"type": "fallback"}`` only when sentence synthesis produced + no audio. Providers with no chunked API (edge, the default) + still speak per sentence via ``text_to_speech_tool`` and stream + that PCM — the client POST is the last resort, not the Edge path. """ if not _ws_auth_ok(ws): await ws.close(code=4401) @@ -416,19 +512,19 @@ async def speak_stream_ws(ws: "WebSocket") -> None: with _config_profile_scope(profile): cfg = _load_tts_config() streamer = resolve_streaming_provider(cfg) - cap = _resolve_max_text_length(_get_provider(cfg), cfg) if streamer else 0 + cap = _resolve_max_text_length(_get_provider(cfg), cfg) return streamer, cap, cfg try: streamer, cap, cfg = await loop.run_in_executor(None, _resolve) except Exception: _log.exception("speak-stream provider resolution failed") - streamer, cap = None, 0 + streamer, cap, cfg = None, 0, {} if streamer is None: - with contextlib.suppress(Exception): - await ws.send_json({"type": "fallback"}) - await ws.close() - return + # Edge (the default) and every other non-chunked provider still have a + # documented per-sentence path. type=fallback here is what makes Desktop + # wait for the whole reply and POST it to /api/audio/speak. + streamer = _SyncSentencePCMStreamer() # The start frame is deferred until the first PCM chunk (or end-of-speech): # the OpenAI-compatible streamer only learns the endpoint's real rate from @@ -446,6 +542,8 @@ async def speak_stream_ws(ws: "WebSocket") -> None: ) stop = threading.Event() + produced_audio = False + synthesis_failed = False text_q: queue.Queue = queue.Queue() # str deltas; None = end-of-text chunks: asyncio.Queue = asyncio.Queue() # PCM out; None = synthesis done @@ -457,6 +555,7 @@ async def speak_stream_ws(ws: "WebSocket") -> None: _synthesize() def _synthesize(): + nonlocal produced_audio, synthesis_failed from tools.tts_streaming import SentenceChunker from tools.tts_text_normalize import _strip_markdown_for_tts @@ -499,9 +598,11 @@ async def speak_stream_ws(ws: "WebSocket") -> None: for chunk in streamer.stream(piece): if stop.is_set(): return + produced_audio = True loop.call_soon_threadsafe(chunks.put_nowait, chunk) except Exception as exc: _log.warning("speak-stream synthesis failed: %s", exc) + synthesis_failed = True finally: loop.call_soon_threadsafe(chunks.put_nowait, None) @@ -533,8 +634,13 @@ async def speak_stream_ws(ws: "WebSocket") -> None: await _send_start() await ws.send_bytes(chunk) if not stop.is_set(): - await _send_start() - await ws.send_json({"type": "end"}) + # Fallback is the last resort: sentence synthesis was asked for and + # produced nothing. A normal edge reply has already streamed PCM. + if synthesis_failed and not produced_audio: + await ws.send_json({"type": "fallback"}) + else: + await _send_start() + await ws.send_json({"type": "end"}) except (WebSocketDisconnect, RuntimeError): pass finally: diff --git a/tests/hermes_cli/test_web_server_speak_stream.py b/tests/hermes_cli/test_web_server_speak_stream.py index 7c30d3c3fa..c6589352b9 100644 --- a/tests/hermes_cli/test_web_server_speak_stream.py +++ b/tests/hermes_cli/test_web_server_speak_stream.py @@ -158,3 +158,68 @@ def test_split_text_respects_cap_and_preserves_content(): assert word in joined +def test_edge_speak_stream_speaks_each_sentence_instead_of_whole_text_fallback( + stream_client, monkeypatch, tmp_path +): + """Edge has no chunked API, but the documented path is per-sentence sync + synthesis. speak-stream must stream that PCM — not ``type: fallback``, + which makes Desktop POST the whole reply and wait for it. + """ + first = "The first spoken sentence is long enough to stand alone." + second = "The second spoken sentence is also long enough to stand alone." + reply = f"{first} {second}" + pcm = [b"\x11\x00\x22\x00", b"\x33\x00\x44\x00"] + calls: list[str] = [] + pcm_for: dict[str, bytes] = {} + + def fake_tts(*args, **kwargs): + raw = kwargs.get("text") if "text" in kwargs else (args[0] if args else "") + text = raw if isinstance(raw, str) else "" + calls.append(text) + path = kwargs.get("output_path") or str(tmp_path / f"sent-{len(calls)}.mp3") + # Edge's sync tool writes MP3 bytes, not a WAV container. + with open(path, "wb") as fh: + fh.write(b"ID3not-a-wav") + pcm_for[path] = pcm[len(calls) - 1] + return json.dumps({"success": True, "file_path": path, "file_paths": [path]}) + + def fake_ffmpeg(_ffmpeg, args, **_kwargs): + import subprocess + + path = args[args.index("-i") + 1] + assert "s16le" in args and "pipe:1" in args + return subprocess.CompletedProcess(args, 0, stdout=pcm_for[path], stderr=b"") + + monkeypatch.setattr("tools.tts_streaming.resolve_streaming_provider", lambda cfg: None) + monkeypatch.setattr( + "tools.tts_tool._load_tts_config", + lambda: {"provider": "edge", "streaming": {"min_len": 6}}, + ) + monkeypatch.setattr("tools.tts_tool._get_provider", lambda cfg: "edge") + monkeypatch.setattr("tools.tts_tool._resolve_max_text_length", lambda provider, cfg: 4000) + monkeypatch.setattr("tools.tts_tool.text_to_speech_tool", fake_tts) + monkeypatch.setattr("tools.tts_tool_delivery._ffmpeg_run", fake_ffmpeg) + monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/ffmpeg" if name == "ffmpeg" else None) + + with stream_client.websocket_connect(_url()) as conn: + conn.send_text(json.dumps({"text": reply, "done": True})) + frames = [] + while True: + message = conn.receive() + if message.get("bytes") is not None: + frames.append(message["bytes"]) + continue + frames.append(json.loads(message["text"])) + if frames[-1].get("type") in {"end", "fallback"}: + break + + assert {"type": "fallback"} not in frames + assert frames[0] == {"type": "start", "sample_rate": 24000, "channels": 1} + assert frames[-1] == {"type": "end"} + assert frames[1:-1] == pcm + assert len(calls) == 2 + assert first in calls[0] and second not in calls[0] + assert second in calls[1] and first not in calls[1] + assert reply not in calls + +