diff --git a/src/owner_voice_pet/assistant_pipeline.py b/src/owner_voice_pet/assistant_pipeline.py index 4036674..efa5367 100644 --- a/src/owner_voice_pet/assistant_pipeline.py +++ b/src/owner_voice_pet/assistant_pipeline.py @@ -489,24 +489,22 @@ class TurnController: ) def _play_interruptible(self, segment: AudioSegment, *, turn_id: int) -> bool: - elapsed_ms = 0 guard_cleared = False self.vad_recorder.provider.reset() realtime_session = self._start_realtime_transcript() speech_ms = 0 partial_seen = False pending_frames: list[AudioFrame] = [] - for chunk in _audio_chunks(segment, chunk_ms=100): - playback = self.transport.play_pcm(chunk) - if playback.error: - raise playback.error - elapsed_ms += chunk.duration_ms + interrupted = False + + def after_chunk(_chunk: AudioSegment, elapsed_ms: int) -> bool: + nonlocal guard_cleared, speech_ms, partial_seen, interrupted if elapsed_ms < self.config.barge_in_echo_guard_ms: - continue + return False if not guard_cleared: self.transport.flush_input() guard_cleared = True - continue + return False detected, speech_ms, partial_seen, new_frames = self._detect_barge_in( realtime_session, turn_id=turn_id, @@ -518,12 +516,16 @@ class TurnController: self._pending_capture_frames.extend(pending_frames) self._event(BARGE_IN_DETECTED, PipelineState.INTERRUPTED, "检测到用户打断", turn_id=turn_id) self._event(PLAYBACK_INTERRUPTED, PipelineState.INTERRUPTED, "播报已打断", turn_id=turn_id) - if realtime_session is not None: - realtime_session.finish() + interrupted = True return True + return False + + playback = self.transport.play_pcm_chunks(segment, chunk_ms=100, after_chunk=after_chunk) + if playback.error: + raise playback.error if realtime_session is not None: realtime_session.finish() - return False + return interrupted def _detect_barge_in( self, @@ -692,30 +694,3 @@ class VoiceAssistantPipeline: handler(event) return dispatch_pipeline_event(self.reporter, event) - - -def _audio_chunks(segment: AudioSegment, *, chunk_ms: int) -> list[AudioSegment]: - if chunk_ms <= 0 or segment.duration_ms <= chunk_ms: - return [segment] - bytes_per_ms = max(1, int(segment.sample_rate * segment.channels * 2 / 1000)) - chunk_bytes = max(2 * segment.channels, bytes_per_ms * chunk_ms) - chunk_bytes -= chunk_bytes % (2 * segment.channels) - chunks: list[AudioSegment] = [] - offset = 0 - start_ms = segment.start_time_ms - while offset < len(segment.pcm): - data = segment.pcm[offset : offset + chunk_bytes] - duration_ms = max(1, int(len(data) / bytes_per_ms)) - chunks.append( - AudioSegment( - data, - segment.sample_rate, - segment.channels, - start_ms, - start_ms + duration_ms, - dict(segment.metadata), - ) - ) - offset += len(data) - start_ms += duration_ms - return chunks diff --git a/src/owner_voice_pet/protocols.py b/src/owner_voice_pet/protocols.py index ac9cdf1..eba1888 100644 --- a/src/owner_voice_pet/protocols.py +++ b/src/owner_voice_pet/protocols.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Iterable, Sequence +from collections.abc import Callable, Iterable, Sequence from typing import Protocol from .models import ( @@ -28,6 +28,15 @@ class AudioTransport(Protocol): def play_pcm(self, segment: AudioSegment, interrupt: bool = False) -> PlaybackResult: ... + def play_pcm_chunks( + self, + segment: AudioSegment, + *, + chunk_ms: int, + after_chunk: Callable[[AudioSegment, int], bool] | None = None, + ) -> PlaybackResult: + ... + def flush_input(self) -> int: ... diff --git a/src/owner_voice_pet/transport.py b/src/owner_voice_pet/transport.py index 1a2950d..862b8f7 100644 --- a/src/owner_voice_pet/transport.py +++ b/src/owner_voice_pet/transport.py @@ -6,6 +6,7 @@ import shutil import subprocess import tempfile import time +from collections.abc import Callable from collections import deque from pathlib import Path from typing import Any @@ -106,6 +107,23 @@ class MemoryAudioTransport: self.played_segments.append(segment) return PlaybackResult(True, segment.duration_ms) + def play_pcm_chunks( + self, + segment: AudioSegment, + *, + chunk_ms: int, + after_chunk: Callable[[AudioSegment, int], bool] | None = None, + ) -> PlaybackResult: + elapsed_ms = 0 + for chunk in _raw_pcm_chunks(segment, chunk_ms=chunk_ms): + playback = self.play_pcm(chunk) + if playback.error: + return playback + elapsed_ms += chunk.duration_ms + if after_chunk is not None and after_chunk(chunk, elapsed_ms): + return PlaybackResult(True, elapsed_ms) + return PlaybackResult(True, elapsed_ms) + def flush_input(self) -> int: self.flush_count += 1 if not self._flush_clears_input: @@ -288,6 +306,59 @@ class SoundDeviceAudioTransport: ) return PlaybackResult(True, segment.duration_ms) + def play_pcm_chunks( + self, + segment: AudioSegment, + *, + chunk_ms: int, + after_chunk: Callable[[AudioSegment, int], bool] | None = None, + ) -> PlaybackResult: + if self._sd is None or segment.metadata.get("format") in {"aiff", "wav", "mp3", "m4a", "aac"}: + playback = self.play_pcm(segment) + if playback.error: + return playback + if after_chunk is not None: + after_chunk(segment, segment.duration_ms) + return playback + if not segment.pcm: + return PlaybackResult( + False, + 0, + ProviderError( + ErrorCode.AUDIO_STREAM_UNDERRUN, + "cannot play empty audio segment", + True, + "sounddevice-transport", + "transport", + ), + ) + elapsed_ms = 0 + try: + with self._sd.RawOutputStream( + samplerate=segment.sample_rate, + channels=segment.channels, + dtype="int16", + device=_coerce_device_id(self._output_device), + ) as stream: + for chunk in _raw_pcm_chunks(segment, chunk_ms=chunk_ms): + stream.write(chunk.pcm) + elapsed_ms += chunk.duration_ms + if after_chunk is not None and after_chunk(chunk, elapsed_ms): + return PlaybackResult(True, elapsed_ms) + except Exception as exc: + return PlaybackResult( + False, + elapsed_ms, + ProviderError( + ErrorCode.AUDIO_OUTPUT_DEVICE_MISSING, + f"cannot play through sounddevice output stream: {exc}", + False, + "sounddevice-transport", + "transport", + ), + ) + return PlaybackResult(True, elapsed_ms) + def stop(self) -> None: if self._stream is None: return None @@ -418,3 +489,30 @@ def _play_file_bytes_with_afplay(segment: AudioSegment) -> PlaybackResult: ), ) return PlaybackResult(True, segment.duration_ms) + + +def _raw_pcm_chunks(segment: AudioSegment, *, chunk_ms: int) -> list[AudioSegment]: + if chunk_ms <= 0 or segment.duration_ms <= chunk_ms: + return [segment] + bytes_per_ms = max(1, int(segment.sample_rate * segment.channels * 2 / 1000)) + chunk_bytes = max(2 * segment.channels, bytes_per_ms * chunk_ms) + chunk_bytes -= chunk_bytes % (2 * segment.channels) + chunks: list[AudioSegment] = [] + offset = 0 + start_ms = segment.start_time_ms + while offset < len(segment.pcm): + data = segment.pcm[offset : offset + chunk_bytes] + duration_ms = max(1, int(len(data) / bytes_per_ms)) + chunks.append( + AudioSegment( + data, + segment.sample_rate, + segment.channels, + start_ms, + start_ms + duration_ms, + dict(segment.metadata), + ) + ) + offset += len(data) + start_ms += duration_ms + return chunks diff --git a/src/owner_voice_pet/tts.py b/src/owner_voice_pet/tts.py index 432ff4d..4b3853c 100644 --- a/src/owner_voice_pet/tts.py +++ b/src/owner_voice_pet/tts.py @@ -123,7 +123,7 @@ class MacSayTtsProvider: ) from exc try: subprocess.run( - ["afconvert", "-f", "WAVE", "-d", "LEI16@16000", "-c", "1", str(aiff_output), str(wav_output)], + ["afconvert", "-f", "WAVE", "-d", "LEI16", "-c", "1", str(aiff_output), str(wav_output)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, diff --git a/tests/test_transport.py b/tests/test_transport.py index 1e07bc7..d033b8c 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -114,6 +114,24 @@ class TransportTests(unittest.TestCase): self.assertTrue(result.played) self.assertEqual(fake.output_writes, [b"\x00\x00\x01\x00"]) + self.assertEqual(fake.output_stream_open_count, 1) + + def test_sounddevice_chunk_playback_keeps_one_output_stream_open(self) -> None: + fake = FakeSoundDevice() + transport = SoundDeviceAudioTransport(sounddevice_module=fake) + segment = AudioSegment(b"\x00\x00" * 1600, 16000, 1, 0, 100) + callbacks: list[int] = [] + + result = transport.play_pcm_chunks( + segment, + chunk_ms=20, + after_chunk=lambda _chunk, elapsed_ms: callbacks.append(elapsed_ms) is not None and False, + ) + + self.assertTrue(result.played) + self.assertEqual(fake.output_stream_open_count, 1) + self.assertGreater(len(fake.output_writes), 1) + self.assertEqual(callbacks[-1], result.duration_ms) def test_sounddevice_device_report_uses_query_devices(self) -> None: fake = FakeSoundDevice() @@ -162,6 +180,7 @@ class FakeSoundDevice: def __init__(self, callback_payloads: list[bytes] | None = None) -> None: self.callback_payloads = callback_payloads self.output_writes: list[bytes] = [] + self.output_stream_open_count = 0 def RawInputStream(self, **kwargs): if self.callback_payloads is not None: @@ -169,6 +188,7 @@ class FakeSoundDevice: return FakeInputStream(**kwargs) def RawOutputStream(self, **kwargs): + self.output_stream_open_count += 1 return FakeOutputStream(self, **kwargs) def query_devices(self):