"""Streaming-TTS chunk reassembly and the near-miss log. Both are pure — no network, no audio device, no ONNX model.""" import sys from pathlib import Path from unittest.mock import MagicMock, patch import numpy as np import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from bolt_pet import config as tts_config from bolt_pet.audio import tts from bolt_pet.audio.tts import TtsError, chunks_to_int16, voice_for from bolt_pet.audio.wake_word import NearMissLog def _pcm(*values): return np.array(values, dtype=np.int16).tobytes() def test_whole_samples_pass_straight_through(): chunks = list(chunks_to_int16([_pcm(1, 2), _pcm(3, 4)])) assert np.concatenate(chunks).tolist() == [1, 2, 3, 4] def test_a_sample_split_across_two_http_chunks_is_rejoined(): # The killer bug this exists to prevent: an odd byte at a chunk boundary # shifts everything after it by one byte and plays as static. raw = _pcm(100, -200, 300, -400) chunks = list(chunks_to_int16([raw[:3], raw[3:]])) assert np.concatenate(chunks).tolist() == [100, -200, 300, -400] def test_many_odd_boundaries_in_a_row(): raw = _pcm(*range(1, 21)) pieces = [raw[i:i + 3] for i in range(0, len(raw), 3)] # every boundary odd assert np.concatenate(list(chunks_to_int16(pieces))).tolist() == list(range(1, 21)) def test_empty_chunks_are_skipped(): assert list(chunks_to_int16([b"", b""])) == [] def test_a_dangling_byte_at_the_end_is_dropped_not_played(): raw = _pcm(7, 8) + b"\x01" assert np.concatenate(list(chunks_to_int16([raw]))).tolist() == [7, 8] # ── wake-word near misses ─────────────────────────────────────────────────── def test_scores_just_under_the_threshold_are_recorded(): log = NearMissLog(limit=10, margin=0.2) assert log.observe(0.45, threshold=0.5, timestamp=1.0) is True assert log.entries() == [(1.0, 0.45, 0.5)] def test_detections_and_background_noise_are_not_near_misses(): log = NearMissLog(limit=10, margin=0.2) assert log.observe(0.90, threshold=0.5, timestamp=1.0) is False # it fired assert log.observe(0.05, threshold=0.5, timestamp=2.0) is False # just noise assert log.entries() == [] def test_peak_tracks_every_score_not_just_near_misses(): log = NearMissLog(limit=10, margin=0.2) log.observe(0.30, threshold=0.5, timestamp=1.0) log.observe(0.95, threshold=0.5, timestamp=2.0) log.observe(0.10, threshold=0.5, timestamp=3.0) assert log.peak == 0.95 def test_the_log_is_bounded(): log = NearMissLog(limit=3, margin=0.2) for i in range(10): log.observe(0.45, threshold=0.5, timestamp=float(i)) assert len(log.entries()) == 3 assert log.entries()[-1][0] == 9.0 def test_clear_resets_peak_and_entries(): log = NearMissLog(limit=3, margin=0.2) log.observe(0.45, threshold=0.5, timestamp=1.0) log.clear() assert log.entries() == [] and log.peak == 0.0 # ── voice selection (server speak_as) ─────────────────────────────────────── def test_the_override_voice_wins_over_the_configured_one(monkeypatch): monkeypatch.setattr(tts_config, "ELEVENLABS_VOICE_ID", "DEFAULT") assert voice_for("VOICE1") == "VOICE1" assert voice_for("") == "DEFAULT" assert voice_for(None) == "DEFAULT" # ── server-hosted synthesis (/desk/tts) ───────────────────────────────────── # No local ElevenLabs account: both the whole-clip and streaming paths post # to this pet's own server, same as server_client.py's other endpoints. @pytest.fixture(autouse=True) def _configured(monkeypatch): monkeypatch.setattr(tts_config, "SERVER_URL", "http://test-server:5002") monkeypatch.setattr(tts_config, "API_KEY", "test-key") monkeypatch.setattr(tts_config, "SESSION_ID", "pet-test") monkeypatch.setattr(tts_config, "ELEVENLABS_VOICE_ID", "default-voice") def _mock_response(content=b"", ok=True): resp = MagicMock() resp.content = content resp.raise_for_status = MagicMock() if ok else MagicMock(side_effect=Exception("boom")) return resp def test_synthesize_pcm_posts_to_the_servers_tts_endpoint(): pcm_bytes = np.array([1, 2, 3], dtype=np.int16).tobytes() with patch.object(tts.requests, "post") as post: post.return_value = _mock_response(pcm_bytes) pcm, rate = tts.synthesize_pcm("hello there") post.assert_called_once() args, kwargs = post.call_args assert args[0] == "http://test-server:5002/desk/tts" assert kwargs["headers"] == {"X-Desk-Api-Key": "test-key"} assert kwargs["json"] == { "session_id": "pet-test", "text": "hello there", "voice_id": "default-voice", } assert pcm.tolist() == [1, 2, 3] assert rate == tts_config.TTS_SAMPLE_RATE def test_synthesize_pcm_uses_the_override_voice(): with patch.object(tts.requests, "post") as post: post.return_value = _mock_response(b"\x01\x00") tts.synthesize_pcm("hi", voice_id="picked-voice") assert post.call_args.kwargs["json"]["voice_id"] == "picked-voice" def test_synthesize_pcm_raises_when_not_configured(monkeypatch): monkeypatch.setattr(tts_config, "API_KEY", "") with pytest.raises(TtsError, match="BOLT_SERVER_URL"): tts.synthesize_pcm("hi") def test_synthesize_pcm_raises_on_empty_audio(): with patch.object(tts.requests, "post") as post: post.return_value = _mock_response(b"") with pytest.raises(TtsError, match="no audio"): tts.synthesize_pcm("hi") def test_synthesize_pcm_raises_when_the_request_fails(): with patch.object(tts.requests, "post") as post: post.return_value = _mock_response(b"", ok=False) with pytest.raises(TtsError, match="server tts request failed"): tts.synthesize_pcm("hi") def test_stream_pcm_posts_to_the_same_endpoint_with_stream_true(): with patch.object(tts.requests, "post") as post: response = MagicMock() response.raise_for_status = MagicMock() response.iter_content.return_value = [np.array([4, 5], dtype=np.int16).tobytes()] post.return_value = response chunks = list(tts.stream_pcm("hi")) assert post.call_args.kwargs["stream"] is True assert post.call_args[0][0] == "http://test-server:5002/desk/tts" assert np.concatenate(chunks).tolist() == [4, 5] def test_stream_pcm_raises_when_not_configured(monkeypatch): monkeypatch.setattr(tts_config, "SERVER_URL", "") with pytest.raises(TtsError, match="BOLT_SERVER_URL"): list(tts.stream_pcm("hi"))