audio: add deafened property to suppress playback

When deafened, playback callback writes silence and
queue_playback discards incoming PCM.
This commit is contained in:
Username
2026-02-24 12:58:39 +01:00
parent 31ac90d2c9
commit 3bbe8a96f3
2 changed files with 46 additions and 0 deletions

View File

@@ -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:

View File

@@ -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()