diff --git a/src/tuimble/audio.py b/src/tuimble/audio.py index 7930d78..1715d71 100644 --- a/src/tuimble/audio.py +++ b/src/tuimble/audio.py @@ -39,6 +39,7 @@ class AudioPipeline: self._input_stream = None self._output_stream = None self._capturing = False + self._deafened = False def start(self): """Open audio streams.""" @@ -84,6 +85,14 @@ class AudioPipeline: def capturing(self, value: bool): self._capturing = value + @property + def deafened(self) -> bool: + return self._deafened + + @deafened.setter + def deafened(self, value: bool): + self._deafened = value + def _capture_callback(self, indata, frames, time_info, status): """Called by sounddevice when input data is available.""" if status: @@ -98,6 +107,9 @@ class AudioPipeline: """Called by sounddevice when output buffer needs data.""" if status: log.warning("playback status: %s", status) + if self._deafened: + outdata[:] = b"\x00" * len(outdata) + return try: pcm = self._playback_queue.get_nowait() n = min(len(pcm), len(outdata)) @@ -116,6 +128,8 @@ class AudioPipeline: def queue_playback(self, pcm_data: bytes): """Queue raw PCM data for playback (16-bit, mono, 48kHz).""" + if self._deafened: + return try: self._playback_queue.put_nowait(pcm_data) except queue.Full: diff --git a/tests/test_audio.py b/tests/test_audio.py index 3d6db05..b93d359 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -10,6 +10,7 @@ def test_default_construction(): assert ap._input_device is None assert ap._output_device is None assert ap.capturing is False + assert ap.deafened is False def test_custom_construction(): @@ -89,6 +90,37 @@ def test_queue_playback_overflow_drops(): assert ap._playback_queue.qsize() == ap._playback_queue.maxsize +def test_deafened_toggle(): + ap = AudioPipeline() + assert ap.deafened is False + ap.deafened = True + assert ap.deafened is True + ap.deafened = False + assert ap.deafened is False + + +def test_queue_playback_discards_when_deafened(): + """Incoming PCM is dropped when deafened.""" + ap = AudioPipeline() + ap.deafened = True + ap.queue_playback(b"\x42" * 100) + assert ap._playback_queue.qsize() == 0 + + +def test_playback_callback_silence_when_deafened(): + """Playback callback writes silence when deafened, even with queued data.""" + ap = AudioPipeline() + frame_bytes = FRAME_SIZE * 2 + # Queue data before deafening + pcm = b"\x42" * frame_bytes + ap.queue_playback(pcm) + ap.deafened = True + + outdata = bytearray(b"\xff" * frame_bytes) + ap._playback_callback(outdata, FRAME_SIZE, None, None) + assert outdata == bytearray(frame_bytes) # all zeros + + def test_stop_without_start(): """Stop on unstarted pipeline should not raise.""" ap = AudioPipeline()