From 8b6b37dd57f30f3e911b21afda251603869d8bd2 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 15 Oct 2026 02:18:33 -0500 Subject: [PATCH] audiodev.worklet_audio: AudioWorklet output for Python in a Pyodide Web Worker --- docs/audio.md | 10 + lib/audiodev/README.md | 38 ++++ lib/audiodev/worklet_audio.py | 377 ++++++++++++++++++++++++++++++++++ mip-split.toml | 3 +- pydevices-desktop.toml | 1 + tests/test_auto.py | 1 + tests/test_worklet_audio.py | 332 ++++++++++++++++++++++++++++++ 7 files changed, 761 insertions(+), 1 deletion(-) create mode 100644 lib/audiodev/worklet_audio.py create mode 100644 tests/test_worklet_audio.py diff --git a/docs/audio.md b/docs/audio.md index ce886989..7bf34885 100644 --- a/docs/audio.md +++ b/docs/audio.md @@ -306,6 +306,16 @@ neutral PCM buffers and queue operations. It supports mono/stereo 8-, 16-, and 32-bit signed or unsigned samples in either byte order at caller-selected rates. A host page must explicitly enable audio or microphone access first. +`audiodev.worklet_audio` is the worker half of an AudioWorklet output, for +Python running in a Pyodide Web Worker. The page owns the `AudioContext` and +the worklet and gives your worker a `MessagePort`; the worklet asks for blocks +of 256 frames and this module answers, either from a render callback you set +(`WorkletOutput(port).source = render`) or from the queue behind an ordinary +`pcm_out(format, port=port)`. It tells you the worklet's frame offset, so you +can place an event at an exact sample. Import it explicitly: automatic +selection never picks it. The protocol is in +[lib/audiodev/README.md](../lib/audiodev/README.md#worklet_audiopy). + `audiodev.i2s_audio` adapts `machine.I2S` for MCU boards; boards construct it directly with board-specific pins/codec wiring, then wrap it in `AudioOut`. diff --git a/lib/audiodev/README.md b/lib/audiodev/README.md index 3e361abc..cfd517cc 100644 --- a/lib/audiodev/README.md +++ b/lib/audiodev/README.md @@ -20,6 +20,7 @@ still directly usable for raw `write()`/`readinto()`. | `audiodev.pygame_audio` | CPython desktop with pygame-ce installed | Queued PCM transport on pygame-ce's bundled SDL | | `audiodev.win_audio` | Windows | WASAPI shared-mode queued PCM transport through `uwin32` | | `audiodev.web_audio` | PyScript / browser | Web Audio playback and `getUserMedia` capture | +| `audiodev.worklet_audio` | Pyodide Web Worker | AudioWorklet output over a `MessagePort`; import it explicitly | | `audiodev.i2s_audio` | MCU | `machine.I2S` transport adapter | | `audiodev.pwm_tone` | MCU | PWM / buzzer adapter (`ToneOutput`, unrelated to `AudioOut`) | | `audiodev.emulated_audio` | CI / no hardware | WAV, generator, loopback, discard transports | @@ -624,6 +625,42 @@ duration so a later gesture can still play them. There is no host device to wedge here, so none of the SDL stall machinery exists. +## `worklet_audio.py` + +For Python running in a Pyodide Web Worker. The page owns the `AudioContext` +and an `AudioWorkletProcessor`, and hands your worker one end of a +`MessageChannel`. This module answers the worklet's requests on that port. +`audiodev.auto` can't tell a worker from a page, so you import it yourself. + +Pull mode suits an engine that renders on demand: + +```python +from audiodev import worklet_audio + +out = worklet_audio.WorkletOutput(port, channels=2, sample_rate=48000) +out.source = render # render(blocks) returns blocks * 256 * channels int16 samples +``` + +`pcm_out(format, port=port, latency=None, queue_ms=None)` gives you an ordinary +`PCMOutput` instead, for an `AudioOut` pump or anything else that calls +`write()`. Requests drain its queue and are padded with silence when it runs +short, counted in `starved`. The worker can't block waiting for the worklet, +because the requests arrive on the same thread: `write()` drops the oldest audio +when the queue is full (counted in `dropped`), `try_write()` takes only what +fits, and `awrite()`/`adrain()` wait by yielding to the event loop. + +The protocol, in blocks of 256 frames of int16, interleaved for stereo: + +| Direction | Message | Meaning | +|-----------|---------|---------| +| worklet to worker | `{need, asked}` | send `need` blocks; one reply per request | +| worker to worklet | `{pcm, asked}` | an `Int16Array` of `need * 256 * channels` samples, its buffer transferred, `asked` echoed | +| worklet to worker | `{offset}` | rendered frame `k` is heard at context frame `k + offset`; read it as `out.offset` | +| worklet to worker | `{underrun}` | underruns since the last report; summed in `out.underruns` | + +Each reply is one bulk copy into JS and one `postMessage`, with no per-sample +Python. The format's rate must match the `AudioContext`'s. + ## `emulated_audio.py` CI / no-hardware provider. Never selected by `audiodev.auto`. @@ -667,6 +704,7 @@ the import is a no-op. transport + a fake audiosample), [`tests/test_sdl2_audio.py`](../../tests/test_sdl2_audio.py), [`tests/test_pygame_audio.py`](../../tests/test_pygame_audio.py), [`tests/test_win_audio.py`](../../tests/test_win_audio.py), [`tests/test_web_audio.py`](../../tests/test_web_audio.py) (the PCM16 conversion, its JS run under node when installed), +[`tests/test_worklet_audio.py`](../../tests/test_worklet_audio.py) (against a fake `MessagePort`), [`tests/test_emulated_audio.py`](../../tests/test_emulated_audio.py) and [`tests/test_auto.py`](../../tests/test_auto.py) cover these modules against SDL's `dummy` driver, and [`tests/test_audiodev_latency.py`](../../tests/test_audiodev_latency.py) covers the profile vocabulary on its own. [`tests/test_audio_playback_golden.py`](../../tests/test_audio_playback_golden.py) runs a diff --git a/lib/audiodev/worklet_audio.py b/lib/audiodev/worklet_audio.py new file mode 100644 index 00000000..45ebeabd --- /dev/null +++ b/lib/audiodev/worklet_audio.py @@ -0,0 +1,377 @@ +"""AudioWorklet output for Python running in a Pyodide Web Worker. + +Web Audio lives on the page, but your Python can run in a Web Worker so audio +work never waits on the UI. The page creates an ``AudioContext`` and an +``AudioWorkletProcessor``, connects the worklet to your worker with a +``MessageChannel``, and hands the worker one end of it. This module is the +worker's side of that port. It answers the worklet's requests for audio with +PCM, either rendered on demand or drained from a queue you write into. + +Two ways to use it: + +Pull, for an engine that renders on demand:: + + from audiodev import worklet_audio + + out = worklet_audio.WorkletOutput(port, channels=2, sample_rate=48000) + out.source = render # render(blocks) -> int16 bytes + +Push, for code written against :class:`audiodev.PCMOutput` (an +``AudioOut`` pump, a TTS writer):: + + pcm = worklet_audio.pcm_out(AudioFormat(48000, 2, 16), port=port) + pcm.write(chunk) + +The wire protocol, all messages plain objects over the ``MessagePort``: + +- Audio moves in blocks of 256 frames. Samples are signed 16-bit, + interleaved when there are two channels. +- worklet to worker: ``{need: blocks, asked: frame}`` asks for ``blocks`` + blocks. The worker replies exactly once per request. +- worker to worklet: ``{pcm: Int16Array, asked: frame}``, where the + ``Int16Array`` holds ``blocks * 256 * channels`` samples and ``asked`` + echoes the request. Its buffer goes in the transfer list, so nothing is + copied on the way. +- worklet to worker: ``{offset: n}`` says rendered frame ``k`` is heard at + ``AudioContext`` frame ``k + n``. :attr:`WorkletOutput.offset` holds the + latest value, so you can schedule an event at an exact sample. +- worklet to worker: ``{underrun: n}`` reports ``n`` underruns since its + last report; :attr:`WorkletOutput.underruns` adds them up. + +The format's rate must match the page's ``AudioContext`` rate: the worklet +plays samples as they come and does not resample. +""" + +try: + import asyncio +except ImportError: # pragma: no cover + try: + import uasyncio as asyncio + except ImportError: + asyncio = None + +import time + +from audiodev import AudioFormat, PCMOutput, check_latency, queue_bytes +from audiodev.bytequeue import ByteQueue + +try: + from js import Array, Int16Array, Object +except ImportError: # pragma: no cover - non-browser import for tooling + Array = None + Int16Array = None + Object = None + +try: + from pyodide.ffi import create_proxy, to_js +except ImportError: # pragma: no cover - non-browser import for tooling + try: + from pyscript.ffi import create_proxy, to_js + except ImportError: + create_proxy = None + to_js = None + +#: Frames in one block of the wire protocol. +BLOCK_FRAMES = 256 + +DEFAULT_FORMAT = AudioFormat(48000, 1, 16) + +# Buffered profile depth for pcm_out's queue, in milliseconds. +_BUFFERED_QUEUE_MS = 500 +# Never smaller than this many blocks: a queue shorter than one request +# starves on every request. +_MIN_QUEUE_BLOCKS = 8 + + +def _field(data, name): + """Read ``name`` from a message: a JS object, or a dict in tests.""" + if isinstance(data, dict): + return data.get(name) + try: + return getattr(data, name, None) + except Exception: + return None + + +def _as_bytes(data): + view = memoryview(data) + if getattr(view, "format", "B") not in ("B", "b", "c"): + view = view.cast("B") + return view + + +class WorkletOutput: + """The worker's end of an AudioWorklet port. + + ``port`` is the ``MessagePort`` the page transferred to this worker. + ``channels`` is 1 or 2. Set :attr:`source` to a callable taking a block + count and returning ``blocks * 256 * channels`` int16 samples as any + bytes-like object; until you do, the worklet gets silence. + + By default this sets ``port.onmessage``. If your worker shares the port + with other messages, pass ``listen=False`` and call :meth:`handle` with + each message's ``data`` yourself. + """ + + def __init__(self, port, channels=1, sample_rate=48000, *, listen=True): + if channels not in (1, 2): + raise ValueError("worklet_audio supports 1 or 2 channels") + self.port = port + self.channels = int(channels) + self.sample_rate = int(sample_rate) + self.source = None + #: Latest ``offset`` from the worklet, or None before the first one: + #: rendered frame ``k`` is heard at context frame ``k + offset``. + self.offset = None + #: Underruns the worklet has reported. + self.underruns = 0 + #: Replies padded with silence because the source came up short. + self.starved = 0 + #: Frames sent so far: the rendered frame index of the next reply. + self.frames = 0 + #: ``asked`` of the latest request. + self.asked = None + self._proxy = None + if listen: + self.listen() + + def block_bytes(self, blocks=1): + """Bytes in ``blocks`` blocks at this channel count.""" + return int(blocks) * BLOCK_FRAMES * self.channels * 2 + + def listen(self): + """Take ``port.onmessage``. Called for you unless ``listen=False``.""" + if self._proxy is not None: + return + handler = self._on_message + self._proxy = create_proxy(handler) if create_proxy is not None else handler + self.port.onmessage = self._proxy + + def close(self): + """Stop answering: release ``port.onmessage`` and its proxy.""" + if self._proxy is None: + return + try: + self.port.onmessage = None + except Exception: + pass + destroy = getattr(self._proxy, "destroy", None) + if destroy is not None: + try: + destroy() + except Exception: + pass + self._proxy = None + + def _on_message(self, event): + self.handle(event.data) + + def handle(self, data): + """Act on one message from the worklet. Returns True if it was ours.""" + need = _field(data, "need") + if need is not None: + self._answer(int(need), _field(data, "asked")) + return True + offset = _field(data, "offset") + if offset is not None: + self.offset = int(offset) + return True + underrun = _field(data, "underrun") + if underrun is not None: + self.underruns += int(underrun) + return True + return False + + def _render(self, blocks): + """PCM for ``blocks`` blocks, exactly ``block_bytes(blocks)`` long.""" + size = self.block_bytes(blocks) + source = self.source + if source is None: + return bytearray(size) + data = source(blocks) + if data is None: + self.starved += 1 + return bytearray(size) + view = _as_bytes(data) + if len(view) == size: + return view + if len(view) > size: + return view[:size] + self.starved += 1 + out = bytearray(size) + out[: len(view)] = view + return out + + def _answer(self, blocks, asked): + self.asked = asked + if blocks <= 0: + pcm = bytearray(0) + else: + try: + pcm = self._render(blocks) + except Exception: + # Reply anyway: an unanswered request stalls the worklet. + self._post(bytearray(self.block_bytes(blocks)), asked) + self.frames += blocks * BLOCK_FRAMES + raise + self._post(pcm, asked) + self.frames += max(0, blocks) * BLOCK_FRAMES + + def _post(self, pcm, asked): + # One bulk copy into JS, then a view of that fresh buffer as int16. + u8 = to_js(_as_bytes(pcm)) + samples = Int16Array.new(u8.buffer, u8.byteOffset, u8.byteLength // 2) + message = Object.new() + message.pcm = samples + message.asked = asked + transfer = Array.new() + transfer.push(samples.buffer) + self.port.postMessage(message, transfer) + + +class WorkletPCMOutput(PCMOutput): + """:class:`audiodev.PCMOutput` over an AudioWorklet port. + + ``write()`` fills a queue and each request from the worklet drains it, + padding with silence when the queue runs short (counted in + :attr:`starved`). The worker can't block while it waits for the worklet, + since the worklet's requests arrive on the same thread, so: + + - ``write()`` never waits. When the queue is full it drops the oldest + audio to make room and counts the bytes in :attr:`dropped`. + - ``try_write()`` takes only what fits and leaves you the rest. + - ``awrite()`` and ``adrain()`` wait properly, yielding to the event loop + so requests keep being answered. + """ + + def __init__(self, fmt, *, port, latency=None, queue_ms=None, poll_ms=4, listen=True, **kwargs): + if fmt.bits != 16 or not fmt.signed or fmt.byteorder != "little": + raise ValueError("worklet_audio requires signed 16-bit little-endian PCM") + if fmt.channels not in (1, 2): + raise ValueError("worklet_audio supports 1 or 2 channels") + check_latency(latency) + super().__init__(fmt, **kwargs) + self.latency = latency + self.poll_ms = int(poll_ms) + minimum = _MIN_QUEUE_BLOCKS * BLOCK_FRAMES * fmt.frame_size + limit = queue_bytes(fmt, latency, queue_ms, minimum=minimum) + if limit is None: + limit = max(minimum, fmt.rate * fmt.frame_size * _BUFFERED_QUEUE_MS // 1000) + limit -= limit % fmt.frame_size + self._queue = ByteQueue(limit) + #: Bytes ``write()`` dropped because the queue was full. + self.dropped = 0 + self.worklet = WorkletOutput(port, fmt.channels, fmt.rate, listen=listen) + self.worklet.source = self._drain_blocks + + # The worklet's numbers, here so you needn't reach through ``worklet``. + @property + def offset(self): + return self.worklet.offset + + @property + def underruns(self): + return self.worklet.underruns + + @property + def starved(self): + return self.worklet.starved + + @property + def frames(self): + return self.worklet.frames + + def handle(self, data): + """Pass one worklet message in, when you created this with ``listen=False``.""" + return self.worklet.handle(data) + + def _drain_blocks(self, blocks): + size = self.worklet.block_bytes(blocks) + out = bytearray(size) + got = self._queue.readinto(out) + if got < size and not self.is_open: + # Nothing has been written yet: silence, not starvation. + return out + return memoryview(out)[:got] if got < size else out + + def space(self): + """Bytes the queue can take now, in whole frames.""" + free = self._queue.free() + return max(0, free - free % self.format.frame_size) + + def queued_size(self): + return len(self._queue) + + def is_active(self): + return len(self._queue) > 0 + + def clear(self): + self._queue.clear() + + def _write(self, buf): + view = memoryview(buf) + over = len(view) - self._queue.free() + if over > 0: + self.dropped += over + self._queue.push(view) + return len(view) + + def try_write(self, buf): + self.open() + source = self._prepare(buf) + count = min(len(source), self.space()) + if count <= 0: + return 0 + return self._queue.write(source[:count]) + + async def _awrite(self, buf): + view = memoryview(buf) + while True: + count = min(len(view), self.space()) + if count > 0: + return self._queue.write(view[:count]) + await _asleep_ms(self.poll_ms) + + def _drain(self): + # Waiting here would block the very requests that drain the queue. + return None + + async def _adrain(self): + frames = len(self._queue) // self.format.frame_size + deadline = time.time() + frames / float(self.format.rate) + 1.0 + while len(self._queue) and time.time() < deadline: + await _asleep_ms(self.poll_ms) + + def _close(self): + self._queue.clear() + + def deinit(self): + """Close and stop answering the worklet.""" + self.close() + self.worklet.close() + + +async def _asleep_ms(milliseconds): + if hasattr(asyncio, "sleep_ms"): + await asyncio.sleep_ms(milliseconds) + else: + await asyncio.sleep(milliseconds / 1000) + + +def pcm_out(format=None, *, port, latency=None, queue_ms=None, poll_ms=4, listen=True): + """Create a :class:`WorkletPCMOutput` on ``port`` (a Pyodide worker's MessagePort). + + ``latency`` sizes the queue: the default holds 500 ms, ``"low"`` holds + 100 ms. ``queue_ms`` sets it directly. Either way it is at least eight + blocks. ``format`` defaults to 48 kHz mono and must be 16-bit signed + little-endian with one or two channels, at the ``AudioContext``'s rate. + """ + fmt = format or DEFAULT_FORMAT + return WorkletPCMOutput( + fmt, + port=port, + latency=latency, + queue_ms=queue_ms, + poll_ms=poll_ms, + listen=listen, + ) diff --git a/mip-split.toml b/mip-split.toml index eaa040e1..76787852 100644 --- a/mip-split.toml +++ b/mip-split.toml @@ -44,7 +44,7 @@ host-only = [ ] [audiodev] -# Host mixers and the two browser backends. `i2s_audio`, `pwm_tone`, +# Host mixers and the three browser backends. `i2s_audio`, `pwm_tone`, # `sample_out`, `accel` and `pump` are the MCU's, and `emulated_audio` stays # with them: it is never selected by `auto`, is imported explicitly, and builds # on audiodev's own PCM classes with no host dependency. @@ -55,6 +55,7 @@ host-only = [ "wasm_audio", "web_audio", "win_audio", + "worklet_audio", ] [multimer] diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml index c9668160..ad9ea30c 100644 --- a/pydevices-desktop.toml +++ b/pydevices-desktop.toml @@ -20,6 +20,7 @@ "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/audiodev/wasm_audio.py" = "/lib/audiodev/wasm_audio.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/audiodev/web_audio.py" = "/lib/audiodev/web_audio.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/audiodev/win_audio.py" = "/lib/audiodev/win_audio.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/audiodev/worklet_audio.py" = "/lib/audiodev/worklet_audio.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/__init__.py" = "/lib/bledev/__init__.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/auto.py" = "/lib/bledev/auto.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/bledev/bleak.py" = "/lib/bledev/bleak.py" diff --git a/tests/test_auto.py b/tests/test_auto.py index 3e48ea03..050b6f4c 100644 --- a/tests/test_auto.py +++ b/tests/test_auto.py @@ -89,6 +89,7 @@ def test_backends_do_not_import_auto(self): "android_audio.py", "emulated_audio.py", "win_audio.py", + "worklet_audio.py", "__init__.py", ): tree = ast.parse((root / name).read_text()) diff --git a/tests/test_worklet_audio.py b/tests/test_worklet_audio.py new file mode 100644 index 00000000..54d48d45 --- /dev/null +++ b/tests/test_worklet_audio.py @@ -0,0 +1,332 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""``audiodev.worklet_audio`` against a fake MessagePort and fake JS types.""" + +import asyncio +import struct +import unittest +from array import array +from types import SimpleNamespace +from unittest import mock + +import _env # noqa: F401 + +from audiodev import AudioFormat, PCMOutput, worklet_audio + +BLOCK = worklet_audio.BLOCK_FRAMES + + +class FakeBuffer: + def __init__(self, data): + self.data = bytes(data) + self.byteLength = len(self.data) + + +class FakeUint8Array: + def __init__(self, data): + self.buffer = FakeBuffer(data) + self.byteOffset = 0 + self.byteLength = len(self.buffer.data) + + +class FakeInt16Array: + @classmethod + def new(cls, buffer, offset, length): + obj = cls() + obj.buffer = buffer + obj.samples = list(array("h", buffer.data[offset : offset + 2 * length])) + return obj + + +class FakeObject: + @staticmethod + def new(): + return SimpleNamespace() + + +class FakeArray(list): + @classmethod + def new(cls): + return cls() + + def push(self, item): + self.append(item) + + +class FakePort: + def __init__(self): + self.onmessage = None + self.posted = [] + + def postMessage(self, message, transfer): + self.posted.append((message, transfer)) + + def send(self, **data): + self.onmessage(SimpleNamespace(data=data)) + + def ask(self, need, asked=0): + self.send(need=need, asked=asked) + message, transfer = self.posted[-1] + self._check(message, transfer) + return message.pcm.samples, message.asked + + def _check(self, message, transfer): + # The reply's own buffer is the one transferred, so nothing is copied. + assert list(transfer) == [message.pcm.buffer], transfer + + +def _pcm(values): + return struct.pack(f"<{len(values)}h", *values) + + +class WorkletTestCase(unittest.TestCase): + def setUp(self): + patcher = mock.patch.multiple( + worklet_audio, + to_js=FakeUint8Array, + Int16Array=FakeInt16Array, + Object=FakeObject, + Array=FakeArray, + create_proxy=None, + ) + patcher.start() + self.addCleanup(patcher.stop) + self.port = FakePort() + + +class TestPullMode(WorkletTestCase): + def test_listens_on_the_port(self): + out = worklet_audio.WorkletOutput(self.port) + self.assertIsNotNone(self.port.onmessage) + out.close() + self.assertIsNone(self.port.onmessage) + + def test_silence_without_a_source(self): + worklet_audio.WorkletOutput(self.port) + samples, asked = self.port.ask(2, asked=512) + self.assertEqual([0] * (2 * BLOCK), samples) + self.assertEqual(512, asked) + + def test_source_called_once_per_request_with_block_count(self): + out = worklet_audio.WorkletOutput(self.port, channels=1) + calls = [] + + def render(blocks): + calls.append(blocks) + return _pcm([n % 1000 for n in range(blocks * BLOCK)]) + + out.source = render + samples, asked = self.port.ask(3, asked=7) + self.assertEqual([3], calls) + self.assertEqual([n % 1000 for n in range(3 * BLOCK)], samples) + self.assertEqual(7, asked) + self.assertEqual(1, len(self.port.posted)) + self.assertEqual(3 * BLOCK, out.frames) + self.assertEqual(7, out.asked) + self.assertEqual(0, out.starved) + + def test_stereo_stays_interleaved(self): + out = worklet_audio.WorkletOutput(self.port, channels=2) + frames = [(n, -n - 1) for n in range(BLOCK)] + out.source = lambda blocks: _pcm([s for frame in frames for s in frame]) + samples, _ = self.port.ask(1) + self.assertEqual(2 * BLOCK, len(samples)) + self.assertEqual(list(range(BLOCK)), samples[0::2]) + self.assertEqual([-n - 1 for n in range(BLOCK)], samples[1::2]) + + def test_typed_memoryview_source(self): + out = worklet_audio.WorkletOutput(self.port) + out.source = lambda blocks: memoryview(array("h", range(blocks * BLOCK))) + samples, _ = self.port.ask(1) + self.assertEqual(list(range(BLOCK)), samples) + + def test_short_source_padded_and_counted(self): + out = worklet_audio.WorkletOutput(self.port) + out.source = lambda blocks: _pcm([5] * 10) + samples, _ = self.port.ask(1) + self.assertEqual([5] * 10 + [0] * (BLOCK - 10), samples) + self.assertEqual(1, out.starved) + + def test_long_source_truncated(self): + out = worklet_audio.WorkletOutput(self.port) + out.source = lambda blocks: _pcm([1] * (BLOCK + 50)) + samples, _ = self.port.ask(1) + self.assertEqual([1] * BLOCK, samples) + + def test_failing_source_still_replies(self): + out = worklet_audio.WorkletOutput(self.port) + + def broken(blocks): + raise RuntimeError("engine fault") + + out.source = broken + with self.assertRaises(RuntimeError): + self.port.send(need=1, asked=3) + message, _ = self.port.posted[-1] + self.assertEqual([0] * BLOCK, message.pcm.samples) + self.assertEqual(3, message.asked) + + def test_offset_and_underruns(self): + out = worklet_audio.WorkletOutput(self.port) + self.assertIsNone(out.offset) + self.port.send(offset=1234) + self.assertEqual(1234, out.offset) + self.port.send(offset=1500) + self.assertEqual(1500, out.offset) + self.port.send(underrun=2) + self.port.send(underrun=1) + self.assertEqual(3, out.underruns) + self.assertEqual([], self.port.posted) + + def test_handle_without_listening(self): + out = worklet_audio.WorkletOutput(self.port, listen=False) + self.assertIsNone(self.port.onmessage) + self.assertFalse(out.handle({"other": 1})) + self.assertTrue(out.handle({"offset": 9})) + self.assertEqual(9, out.offset) + + def test_rejects_three_channels(self): + with self.assertRaises(ValueError): + worklet_audio.WorkletOutput(self.port, channels=3) + + +class TestPCMOut(WorkletTestCase): + def make(self, channels=1, **kwargs): + return worklet_audio.pcm_out(AudioFormat(48000, channels, 16), port=self.port, **kwargs) + + def test_is_a_pcm_output(self): + pcm = self.make() + self.assertIsInstance(pcm, PCMOutput) + self.assertEqual(AudioFormat(48000, 1, 16), worklet_audio.pcm_out(port=FakePort()).format) + + def test_format_checks(self): + with self.assertRaises(ValueError): + worklet_audio.pcm_out(AudioFormat(48000, 1, 8), port=self.port) + with self.assertRaises(ValueError): + worklet_audio.pcm_out(AudioFormat(48000, 1, 16, byteorder="big"), port=self.port) + with self.assertRaises(ValueError): + worklet_audio.pcm_out(AudioFormat(48000, 4, 16), port=self.port) + with self.assertRaises(ValueError): + self.make(latency="fast") + with self.assertRaises(ValueError): + self.make().write(b"\x00") # half a frame + + def test_write_then_need_drains_in_order(self): + pcm = self.make() + pcm.write(_pcm(range(2 * BLOCK))) + self.assertEqual(2 * BLOCK * 2, pcm.queued_size()) + self.assertTrue(pcm.is_active()) + first, _ = self.port.ask(1) + second, _ = self.port.ask(1) + self.assertEqual(list(range(BLOCK)), first) + self.assertEqual(list(range(BLOCK, 2 * BLOCK)), second) + self.assertEqual(0, pcm.queued_size()) + self.assertEqual(0, pcm.starved) + self.assertEqual(2 * BLOCK, pcm.frames) + + def test_starvation_padded_with_silence(self): + pcm = self.make() + pcm.write(_pcm([9] * 100)) + samples, _ = self.port.ask(1) + self.assertEqual([9] * 100 + [0] * (BLOCK - 100), samples) + self.assertEqual(1, pcm.starved) + samples, _ = self.port.ask(1) + self.assertEqual([0] * BLOCK, samples) + self.assertEqual(2, pcm.starved) + + def test_silence_before_first_write_is_not_starvation(self): + pcm = self.make() + samples, _ = self.port.ask(2) + self.assertEqual([0] * (2 * BLOCK), samples) + self.assertEqual(0, pcm.starved) + + def test_stereo_ring(self): + pcm = self.make(channels=2) + values = [s for n in range(BLOCK) for s in (n, -n)] + pcm.write(_pcm(values)) + samples, _ = self.port.ask(1) + self.assertEqual(values, samples) + + def test_volume_applies(self): + pcm = self.make() + pcm.set_volume(50) + pcm.write(_pcm([1000] * BLOCK)) + samples, _ = self.port.ask(1) + self.assertTrue(all(0 < s < 1000 for s in samples), samples[:4]) + + def test_queue_depth_follows_latency(self): + frame = 2 + self.assertEqual(48000 * frame // 2, self.make().space()) + self.assertEqual(48000 * frame // 10, self.make(latency="low").space()) + self.assertEqual(48000 * frame * 200 // 1000, self.make(queue_ms=200).space()) + # Never below eight blocks. + self.assertEqual(8 * BLOCK * frame, self.make(queue_ms=1).space()) + + def test_try_write_takes_only_what_fits(self): + pcm = self.make(queue_ms=1) # 8 blocks + room = pcm.space() + data = _pcm([3] * (room // 2 + 100)) + self.assertEqual(room, pcm.try_write(data)) + self.assertEqual(0, pcm.space()) + self.assertEqual(0, pcm.try_write(_pcm([1]))) + self.assertEqual(0, pcm.dropped) + + def test_write_when_full_drops_oldest(self): + pcm = self.make(queue_ms=1) + room = pcm.space() // 2 + pcm.write(_pcm([1] * room)) + pcm.write(_pcm([2] * 10)) + self.assertEqual(20, pcm.dropped) + samples, _ = self.port.ask(8) + self.assertEqual([1] * (room - 10) + [2] * 10, samples) + + def test_awrite_waits_for_room(self): + pcm = self.make(queue_ms=1) + room = pcm.space() // 2 + + async def run(): + task = asyncio.ensure_future(pcm.awrite(_pcm([4] * (room + BLOCK)))) + await asyncio.sleep(0.02) + self.assertFalse(task.done()) + self.port.ask(1) + await asyncio.wait_for(task, 1.0) + + asyncio.run(run()) + self.assertEqual(0, pcm.dropped) + self.assertEqual(room * 2, pcm.queued_size()) + + def test_adrain_returns_once_drained(self): + pcm = self.make() + pcm.write(_pcm([1] * BLOCK)) + + async def run(): + task = asyncio.ensure_future(pcm.adrain()) + await asyncio.sleep(0.02) + self.assertFalse(task.done()) + self.port.ask(1) + await asyncio.wait_for(task, 1.0) + + asyncio.run(run()) + + def test_clear_and_close(self): + pcm = self.make() + pcm.write(_pcm([1] * BLOCK)) + pcm.clear() + self.assertEqual(0, pcm.queued_size()) + pcm.write(_pcm([1] * BLOCK)) + pcm.close() + self.assertEqual(0, pcm.queued_size()) + pcm.deinit() + self.assertIsNone(self.port.onmessage) + + def test_offset_and_underruns_forwarded(self): + pcm = self.make() + self.port.send(offset=4096) + self.port.send(underrun=1) + self.assertEqual(4096, pcm.offset) + self.assertEqual(1, pcm.underruns) + + +if __name__ == "__main__": + unittest.main()