From 1e64211d671168111e56ab0842c5a2ec31029df3 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Fri, 9 Oct 2026 03:33:57 -0500 Subject: [PATCH] P4 boards record while they play: pcm_in reads the audio pump's RX half On the ESP32-P4 boards the speaker codec and the microphone share I2S0, and machine.I2S takes one direction per port, so pcm_in and the speaker refused each other (#23, #109). On firmware whose pump driver can read the RX half of its own channel (_audioif.rx_open), they no longer need to. - audiodev.pump.PumpPCMInput: a PCMInput over the pump's RX half. It opens the channel pair or joins the one audio_out opened, before the codec comes up; readinto() fills the buffer, areadinto() waits without blocking, stats() reports captured and dropped frames. - audiodev.pump.duplex() and clock_running(); BusioDriver passes the wire's new sd_in to I2SOut as data_in=, so audio_out opens the pair. - I2SWire takes sd_in=, the microphone's pin on a shared wire. - Both P4 board configs publish sd_in, build a PumpPCMInput when the firmware can, and never start the PWM MCLK bootstrap while the pump's channel drives the pin. pcm_out is still machine.I2S and still refuses alongside a recording; without the pump nothing changes. --- .../board_peripherals.py | 79 +++++- .../board_peripherals.py | 68 ++++- docs/audio.md | 30 ++- lib/audiodev/__init__.py | 11 +- lib/audiodev/pump.py | 203 +++++++++++++++ tests/_duplex_fake.py | 88 +++++++ tests/test_esp32_p4_audio.py | 136 +++++++++- tests/test_pump_capture.py | 233 ++++++++++++++++++ 8 files changed, 823 insertions(+), 25 deletions(-) create mode 100644 tests/_duplex_fake.py create mode 100644 tests/test_pump_capture.py diff --git a/board_configs/fbdisplay/esp32-p4-wifi6-touch-lcd-4b/board_peripherals.py b/board_configs/fbdisplay/esp32-p4-wifi6-touch-lcd-4b/board_peripherals.py index 9e55f5e9..bf90be09 100644 --- a/board_configs/fbdisplay/esp32-p4-wifi6-touch-lcd-4b/board_peripherals.py +++ b/board_configs/fbdisplay/esp32-p4-wifi6-touch-lcd-4b/board_peripherals.py @@ -78,13 +78,19 @@ channels=(1, 2), native_channels=_MAX_CHANNELS, bits=(16,), - wire=I2SWire(0, sck=_SCLK, ws=_LRCK, sd=_DSDIN, mck=_MCLK, mck_fs=_MCLK_FS), + # sd_in is the ES7210's data pin. Publishing it is what lets audio_out and + # pcm_in run at once on firmware whose audio pump can read the RX half of + # its own channel: the port opens as one TX/RX pair on one clock. + wire=I2SWire( + 0, sck=_SCLK, ws=_LRCK, sd=_DSDIN, mck=_MCLK, mck_fs=_MCLK_FS, sd_in=_ASDOUT + ), ) -# Capture is the ES7210 on the SAME I2S port and the same clocks as playback, -# with AudioSession(duplex=False) below. That is why this board cannot run the -# acoustic loopback in tools/audio_rig/ and the T-Embed can: there, output and -# input are separate peripherals on separate pins. +# Capture is the ES7210 on the SAME I2S port and the same clocks as playback. +# With the audio pump (audiodev.pump.duplex()), pcm_in reads the RX half of +# the pump's channel, so it records while audio_out plays. Without it, pcm_in +# is machine.I2S, which takes one direction per port, so it and the speaker +# refuse each other (see _refuse_if_held). AUDIO_IN = AudioCapability( _FORMAT, rates=None, @@ -135,13 +141,29 @@ def _i2c(): return _i2c_own +def _pump_clock_running(): + try: + from audiodev import pump + except ImportError: + return False + return pump.clock_running() + + def _ensure_mclk(multiplier=_MCLK_FS, rate=None): - """PWM-bootstrap the shared codec MCLK on GPIO13 until I2S takes the pin.""" + """PWM-bootstrap the shared codec MCLK on GPIO13 until I2S takes the pin. + + Never while the audio pump's channel is running: it is already driving + GPIO13, and a PWM started now would take the pin from it, so both codecs + would lose their clock mid-stream. The idle PWM is released instead. + """ global _mclk, _mclk_rate from machine import PWM, Pin if rate is not None: _mclk_rate = int(rate) + if _pump_clock_running(): + _stop_mclk() + return None freq = _mclk_rate * multiplier if _mclk is None: _mclk = PWM(Pin(_MCLK), freq=freq, duty_u16=32768) @@ -208,11 +230,17 @@ def _refuse_if_held(session, wanted, holder): during playback silenced the speaker with no error anywhere, and ``playing`` stayed True (pydevices#23, heard on this board). So the second direction is refused here, by name, instead. + + This still applies with the audio pump, to pcm_out: it is machine.I2S + whatever the firmware, so it and the pump's channel cannot share the port. + audio_out plays through the pump and records alongside pcm_in. """ if session._owners: raise OSError( - "I2S(0) is in use by %s: this board's speaker and microphone " - "share one I2S port, so close the %s before opening the %s" + "I2S(0) is in use by the %s: this board's speaker and microphone " + "share one I2S port, and pcm_out and machine.I2S take the whole " + "port, so close the %s before opening the %s. audio_out can play " + "while pcm_in records, at the recording's rate" % (holder, holder, wanted) ) @@ -314,9 +342,17 @@ def _pcm_out(format=None, *, latency=None, queue_ms=None): def _pcm_in(format=None, *, latency=None, queue_ms=None): """ES7210 capture: a raw ``PCMInput`` with hardware ADC gain. - Shares I2S(0) and its clocks with playback, so this board cannot record - itself: opening it while the speaker is open raises OSError, and so does - the reverse. + Shares I2S(0) and its clocks with playback. On firmware with the audio + pump, this reads the RX half of the pump's channel, so it records while + ``audio_out`` plays (an acoustic loopback through the panel's own speaker + and microphones works), in either order. The recording must then be at + the rate the speaker is clocked at. The two microphones are averaged into + the one channel. + + Without the pump it is ``machine.I2S``, which has one direction per port: + opening it while the speaker is open raises OSError, and so does the + reverse. ``pcm_out`` is always ``machine.I2S``, so it refuses alongside + a recording either way. """ global _mclk_rate @@ -334,6 +370,18 @@ def _pcm_in(format=None, *, latency=None, queue_ms=None): ) _mclk_rate = wire.rate ibuf = queue_bytes(wire, latency, queue_ms, default=_IBUF, minimum=_MIN_IBUF) + pump = _duplex_pump() + if pump is not None: + return pump.PumpPCMInput( + wire, + wire=AUDIO_OUT.wire, + ring_bytes=ibuf, + take="mix", + before_open=lambda: _refuse_if_held(_SESSION, "microphone", "speaker"), + session=_INPUT_SESSION, + set_hardware_gain=lambda value: _INPUT_SESSION.get_codec().set_gain(value), + power=lambda enable: _input_power(enable, wire.rate), + ) return I2SPCMInput( lambda: _input_stream(ibuf, wire), wire, @@ -343,6 +391,15 @@ def _pcm_in(format=None, *, latency=None, queue_ms=None): ) +def _duplex_pump(): + """``audiodev.pump`` when this firmware records while it plays, else None.""" + try: + from audiodev import pump + except ImportError: + return None + return pump if pump.duplex() else None + + def _audio_out(format=None, *, latency=None, queue_ms=None, **kwargs): """ES8311 sample player: ``play(sample, loop=)``/``stop()``/``pause()``/ ``resume()``/``playing`` over any audiosample (``synthio.Synthesizer``, diff --git a/board_configs/nodisplay/esp32-p4-wifi6-dev-kit/board_peripherals.py b/board_configs/nodisplay/esp32-p4-wifi6-dev-kit/board_peripherals.py index 400c362d..db4c72a5 100644 --- a/board_configs/nodisplay/esp32-p4-wifi6-dev-kit/board_peripherals.py +++ b/board_configs/nodisplay/esp32-p4-wifi6-dev-kit/board_peripherals.py @@ -107,10 +107,18 @@ channels=(1, 2), native_channels=1, bits=(16,), - wire=I2SWire(0, sck=_SCLK, ws=_LRCK, sd=_DSDIN, mck=_MCLK, mck_fs=_MCLK_FS), + # sd_in is the ES8311's ADC output. Publishing it is what lets audio_out + # and pcm_in run at once on firmware whose audio pump can read the RX + # half of its own channel: the port opens as one TX/RX pair on one clock. + wire=I2SWire( + 0, sck=_SCLK, ws=_LRCK, sd=_DSDIN, mck=_MCLK, mck_fs=_MCLK_FS, sd_in=_ASDOUT + ), ) -# The microphone is the ES8311's ADC on the same I2S port and clocks. +# The microphone is the ES8311's ADC on the same I2S port and clocks. With the +# audio pump (audiodev.pump.duplex()), pcm_in reads the RX half of the pump's +# channel, so it records while audio_out plays. Without it, pcm_in is +# machine.I2S, and it and the speaker refuse each other. AUDIO_IN = AudioCapability( _FORMAT, rates=None, @@ -154,12 +162,37 @@ def i2c(): return _i2c_own +def _pump_clock_running(): + try: + from audiodev import pump + except ImportError: + return False + return pump.clock_running() + + +def _duplex_pump(): + """``audiodev.pump`` when this firmware records while it plays, else None.""" + try: + from audiodev import pump + except ImportError: + return None + return pump if pump.duplex() else None + + def _ensure_mclk(multiplier=_MCLK_FS, rate=None): + """PWM-bootstrap MCLK on GPIO13 until I2S takes the pin. + + Never while the audio pump's channel is running: it already drives GPIO13, + and a PWM started now would take the pin from it mid-stream. + """ global _mclk, _mclk_rate from machine import PWM, Pin if rate is not None: _mclk_rate = int(rate) + if _pump_clock_running(): + _stop_mclk() + return None freq = _mclk_rate * multiplier if _mclk is None: _mclk = PWM(Pin(_MCLK), freq=freq, duty_u16=32768) @@ -233,12 +266,16 @@ def _refuse_if_held(session, wanted, holder): Constructing machine.I2S(0) while it is in use doesn't raise on esp32: it deinitialises the live instance and hands the same object back in the - new direction, so the speaker would just go quiet. + new direction, so the speaker would just go quiet. pcm_out is always + machine.I2S, so this holds with the audio pump too; audio_out plays + through the pump and records alongside pcm_in. """ if session._owners: raise OSError( - "I2S(0) is in use by %s: this board's speaker and microphone " - "share one I2S port, so close the %s before opening the %s" + "I2S(0) is in use by the %s: this board's speaker and microphone " + "share one I2S port, and pcm_out and machine.I2S take the whole " + "port, so close the %s before opening the %s. audio_out can play " + "while pcm_in records, at the recording's rate" % (holder, holder, wanted) ) @@ -312,8 +349,12 @@ def _pcm_out(format=None, *, latency=None, queue_ms=None): def _pcm_in(format=None, *, latency=None, queue_ms=None): """The onboard microphone through the ES8311's ADC: a raw ``PCMInput``. - Shares I2S(0) and its clocks with playback, so opening it while the - speaker is open raises OSError, and so does the reverse. + Shares I2S(0) and its clocks with playback. On firmware with the audio + pump, this reads the RX half of the pump's channel, so it records while + ``audio_out`` plays, in either order, at the rate the speaker is clocked + at. Without the pump it is ``machine.I2S``: opening it while the speaker + is open raises OSError, and so does the reverse. ``pcm_out`` is always + ``machine.I2S``, so it refuses alongside a recording either way. """ global _mclk_rate @@ -326,6 +367,19 @@ def _pcm_in(format=None, *, latency=None, queue_ms=None): ) _mclk_rate = wire.rate ibuf = queue_bytes(wire, latency, queue_ms, default=_IBUF, minimum=_MIN_IBUF) + pump = _duplex_pump() + if pump is not None: + # The ES8311 has one ADC channel; this records the left slot. + return pump.PumpPCMInput( + wire, + wire=AUDIO_OUT.wire, + ring_bytes=ibuf, + take="left", + before_open=lambda: _refuse_if_held(_SESSION, "microphone", "speaker"), + session=_INPUT_SESSION, + set_hardware_gain=lambda value: _INPUT_SESSION.get_codec().set_gain(value), + power=lambda enable: _INPUT_SESSION.get_codec().enable_input(enable), + ) return I2SPCMInput( lambda: _input_stream(ibuf, wire), wire, diff --git a/docs/audio.md b/docs/audio.md index b77f20ae..2d540a4f 100644 --- a/docs/audio.md +++ b/docs/audio.md @@ -266,6 +266,31 @@ Codec-specific features remain reachable through `device.codec`. Shared half-duplex hardware uses `AudioSession`; opening the opposite direction while one direction owns the session raises `OSError`. +### Recording while you play, on one I2S port + +Some boards put the speaker codec and the microphone on one I2S port (the +ESP32-P4 boards do). `machine.I2S` opens one direction per port, so there you +can't record and play at the same time through it. On firmware with the audio +pump you can: `audio_out` opens the port as a TX/RX pair, and `pcm_in` returns +an `audiodev.pump.PumpPCMInput` that reads the RX half of the same channel. +Open them in either order; closing one leaves the other running. + +```python +out = board_peripherals.audio_out() +mic = board_peripherals.pcm_in() +out.play(sample, loop=True) +mic.open() +buf = bytearray(48000) # one second at 24 kHz mono +mic.readinto(buf) # the speaker keeps playing +print(mic.stats()) # {'captured': ..., 'dropped': 0, 'waiting': ...} +``` + +The two share one clock, so they share one rate. A recorder at another rate +than the speaker is refused, and so is a sample at another rate while a +recorder is open. `pcm_out` is `machine.I2S`, so it still can't open +alongside a recording. A board joins in by publishing `sd_in=`, the +microphone's pin, on its `I2SWire`. + ## Host backends (transports under `AudioOut`, or usable raw) `audiodev.sdl2_audio` is the reference playback and real-microphone backend for @@ -322,7 +347,8 @@ needed. See [`tests/test_audiodev.py`](../tests/test_audiodev.py). ## ESP32-P4 status -The Waveshare ESP32-P4 configuration uses one half-duplex session for its I2S -peripheral and ES8311. Playback exposes hardware DAC volume/mute and controls +The Waveshare ESP32-P4 configuration shares one I2S port between the ES8311 +and the ES7210; with the audio pump, `audio_out` and `pcm_in` run at once (see +"Recording while you play" above), and without it the two refuse each other. Playback exposes hardware DAC volume/mute and controls the speaker amplifier through `AudioOut`; capture exposes hardware ADC gain. Register, stream, session, and GPIO behavior is covered by host simulations. diff --git a/lib/audiodev/__init__.py b/lib/audiodev/__init__.py index c8c2f3ad..1fe75e94 100644 --- a/lib/audiodev/__init__.py +++ b/lib/audiodev/__init__.py @@ -135,18 +135,23 @@ class I2SWire: such a consumer stop reaching into a board module's private names. """ - def __init__(self, port, *, sck, ws, sd, mck=None, mck_fs=256): + def __init__(self, port, *, sck, ws, sd, mck=None, mck_fs=256, sd_in=None): self.port = int(port) self.sck = int(sck) self.ws = int(ws) self.sd = int(sd) self.mck = None if mck is None else int(mck) self.mck_fs = int(mck_fs) + # The microphone's data pin, when the microphone shares this wire's + # clocks. A board that publishes it can record while it plays: the + # output opens the port as a TX/RX pair and the input reads the RX + # half (``audiodev.pump.PumpPCMInput``). + self.sd_in = None if sd_in is None else int(sd_in) def __repr__(self): return ( - "I2SWire(port=%d, sck=%d, ws=%d, sd=%d, mck=%r, mck_fs=%d)" - % (self.port, self.sck, self.ws, self.sd, self.mck, self.mck_fs) + "I2SWire(port=%d, sck=%d, ws=%d, sd=%d, mck=%r, mck_fs=%d, sd_in=%r)" + % (self.port, self.sck, self.ws, self.sd, self.mck, self.mck_fs, self.sd_in) ) diff --git a/lib/audiodev/pump.py b/lib/audiodev/pump.py index ab0583d1..6502c7d1 100644 --- a/lib/audiodev/pump.py +++ b/lib/audiodev/pump.py @@ -74,6 +74,29 @@ _FIRST_BLOCK_WAIT_US = 20000 _FIRST_BLOCK_STEP_US = 100 +from audiodev import PCMInput + +try: + from time import sleep_ms as _sleep_ms +except ImportError: # CPython, CircuitPython + import time as _time_ms + + def _sleep_ms(ms): + _time_ms.sleep(ms / 1000) + +try: + import asyncio as _asyncio +except ImportError: # pragma: no cover + _asyncio = None + + +async def _async_sleep_ms(ms): + if hasattr(_asyncio, "sleep_ms"): + await _asyncio.sleep_ms(ms) + else: + await _asyncio.sleep(ms / 1000) + + try: from time import sleep_us as _sleep_us from time import ticks_diff as _ticks_diff @@ -269,6 +292,38 @@ def on_board(): return mod is not None and hasattr(mod, "i2s_start") +def duplex(): + """True when this firmware can record while it plays, on one I2S port. + + The driver opens the board's port as a TX/RX channel pair whose RX half + Python can read (``_audioif.rx_open``), so a speaker and a microphone + that share a port's clocks -- the ESP32-P4 boards' ES8311 and ES7210, or + the ES8311's own ADC -- are two halves of one channel rather than two + owners fighting over it. A board asks this to decide which ``PCMInput`` + to build: :class:`PumpPCMInput` when True, ``machine.I2S`` when not. + """ + mod = driver() + return mod is not None and hasattr(mod, "rx_open") + + +def clock_running(): + """True while the pump's driver has an I2S channel open on this board. + + A board whose codecs need MCLK before I2S exists bootstraps it with PWM + on the same pin. Starting that PWM while the channel is running takes the + pin away from it, so both codecs lose their clock mid-stream; a board + asks this first. False on any firmware that cannot answer. + """ + mod = driver() + state = getattr(mod, "i2s_state", None) if mod is not None else None + if state is None: + return False + try: + return state() is not None + except Exception: # noqa: BLE001 - a question must not raise + return False + + def threaded(): """True where spawn() puts the pull loop on a thread of its own. @@ -991,6 +1046,14 @@ def open(self): sink = getattr(w, "sink", None) if sink is not None: kwargs["sink"] = sink + # The microphone's pin, where it shares this port's clocks. The output + # then opens the port as a TX/RX pair, so a PumpPCMInput opened later + # joins this channel instead of being refused -- and one opened + # earlier is joined by this output. Only on a driver that can read the + # RX half; an older I2SOut has no `data_in` to take. + sd_in = getattr(w, "sd_in", None) + if sd_in is not None and duplex(): + kwargs["data_in"] = int(sd_in) try: self.out = mod.I2SOut(w.sck, w.ws, w.sd, main_clock=w.mck, **kwargs) @@ -1053,6 +1116,146 @@ def close(self): self._power(False) +_TAKES = {"left": 0, "right": 1, "mix": 2} + + +class PumpPCMInput(PCMInput): + """Capture from the RX half of the pump's own I2S channel. + + On a board whose speaker and microphone share one I2S port, + ``machine.I2S`` can have only one of them: it opens one direction per + port, and opening the second silently takes the port from the first. + The pump's driver opens the port as a TX/RX pair instead, and this + reads the RX half -- so an ``audio_out`` can play while this records, + in either order, and closing one leaves the other running. + + It is a ``PCMInput`` like any other: ``readinto()`` fills the buffer + and blocks until it has, ``areadinto()`` waits without blocking the + loop, and gain and mute are the board's codec. What is different is + underneath: the driver's interrupt copies every finished DMA block + into a ring of ``ring_bytes``, so a reader that falls behind by less + than that loses nothing. ``stats()["dropped"]`` counts the frames a + slower reader lost. + + The channel is opened (or joined) BEFORE the codec is brought up, so + the codec's setup runs with the channel's MCLK already on its pin; a + board must not start a PWM clock on that pin while + :func:`clock_running` says the channel is up. + + ``take`` picks what a one-channel recording hears from a two-slot + wire: ``"left"``, ``"right"``, or ``"mix"``, their average, which is + the right answer for a codec with a microphone in each slot. + """ + + def __init__(self, format, *, wire, ring_bytes=20000, take="mix", + timeout_ms=1000, before_open=None, **kwargs): + super().__init__(format, **kwargs) + if format.bits != 16 or format.channels not in (1, 2): + raise ValueError( + "the pump's I2S capture is signed 16-bit, one or two " + "channels") + if getattr(wire, "sd_in", None) is None: + raise ValueError("this wire has no microphone pin (sd_in)") + if format.channels == 2: + self._take = 3 + elif take in _TAKES: + self._take = _TAKES[take] + else: + raise ValueError("take must be 'left', 'right' or 'mix'") + self.wire = wire + self.ring_bytes = int(ring_bytes) + self.timeout_ms = int(timeout_ms) + self._before_open = before_open + self._held = False + + def open(self): + if self.is_open: + return self + if self._before_open is not None: + self._before_open() + mod = driver() + if mod is None or not hasattr(mod, "rx_open"): + raise OSError("this firmware cannot record from the pump's " + "I2S channel") + w = self.wire + mod.rx_open(w.port, w.sck, w.ws, w.sd, w.sd_in, self.format.rate, + mclk=-1 if w.mck is None else w.mck, + mclk_fs=w.mck_fs, ring=self.ring_bytes, + take=self._take) + self._held = True + try: + super().open() + except Exception: + self._release() + raise + return self + + def _release(self): + if self._held: + self._held = False + mod = driver() + if mod is not None: + mod.rx_close() + + def _close(self): + self._release() + + def stats(self): + """``captured`` and ``dropped`` frames, and ``waiting`` bytes.""" + mod = driver() + if mod is None or not hasattr(mod, "rx_stats"): + return {"captured": 0, "dropped": 0, "waiting": 0} + captured, dropped, waiting = mod.rx_stats() + return {"captured": captured, "dropped": dropped, + "waiting": waiting} + + def _read_some(self, view): + return driver().rx_read(view) + + def _readinto(self, buf): + view = memoryview(buf) + got = 0 + idle_ms = 0 + while got < len(view): + n = self._read_some(view[got:]) + if n: + got += n + idle_ms = 0 + continue + # A DMA block lands every 5 ms; anything near the timeout + # means the channel is not clocking at all. + if idle_ms >= self.timeout_ms: + raise OSError("the microphone delivered nothing for %d ms" + % self.timeout_ms) + _sleep_ms(2) + idle_ms += 2 + return got + + async def _areadinto(self, buf): + view = memoryview(buf) + got = 0 + idle_ms = 0 + while got < len(view): + n = self._read_some(view[got:]) + if n: + got += n + idle_ms = 0 + continue + if idle_ms >= self.timeout_ms: + raise OSError("the microphone delivered nothing for %d ms" + % self.timeout_ms) + await asyncio.sleep_ms(5) if hasattr(asyncio, "sleep_ms") \ + else asyncio.sleep(0.005) + idle_ms += 5 + return got + + + +def pcm_input(format, **kwargs): + """Build a :class:`PumpPCMInput` (see its docstring for the keywords).""" + return PumpPCMInput(format, **kwargs) + + class PumpOutput: """A push stream on the pump: ``PCMOutput``'s ``_write``, over a Ring. diff --git a/tests/_duplex_fake.py b/tests/_duplex_fake.py new file mode 100644 index 00000000..04eb14fe --- /dev/null +++ b/tests/_duplex_fake.py @@ -0,0 +1,88 @@ +"""A stand-in for ``_audioif`` on firmware that records while it plays. + +The real one is C: the output and the capture reader are two holders of one +I2S channel pair, and the RX half's interrupt copies every DMA block into a +ring that ``rx_read()`` empties. What ``audiodev`` depends on is the contract, +and that is what this models: who holds the channel, at what rate, whether a +join with a different wire is refused, and a ring the test fills by hand. +""" + + +class FakeDuplexDriver: + def __init__(self): + self.holds = 0 # 1 the output, 2 the capture reader + self.live = None # (port, bclk, ws, dout, din, rate) + self.ring = bytearray() # what the interrupt would have copied + self.dropped = 0 + self.captured = 0 + self.calls = [] # every rx_open/rx_close, in order + self.opened_with = None + self.reads = 0 + + # --- what a board asks ------------------------------------------------ + + def i2s_state(self): + if self.live is None: + return None + return (self.live[5], 2, True, self.holds) + + # --- the capture reader ----------------------------------------------- + + def rx_open(self, port, bclk, ws, dout, din, rate, *, mclk=-1, + mclk_fs=256, ring=20000, take=2): + if self.holds & 2: + raise ValueError("capture already open") + want = (port, bclk, ws, dout, din, rate) + if self.live is not None and self.live[:5] != want[:5]: + raise RuntimeError("Peripheral in use") + if self.live is not None and self.live[5] != rate: + raise ValueError("the I2S channel is clocked at %d Hz and this " + "asks for %d Hz" % (self.live[5], rate)) + self.live = want + self.holds |= 2 + self.opened_with = dict(port=port, bclk=bclk, ws=ws, dout=dout, + din=din, rate=rate, mclk=mclk, + mclk_fs=mclk_fs, ring=ring, take=take) + self.calls.append("rx_open") + return ring + + def rx_read(self, buf): + if not self.holds & 2: + raise OSError(19) + self.reads += 1 + n = min(len(buf), len(self.ring)) + n -= n % 2 + buf[:n] = self.ring[:n] + del self.ring[:n] + return n + + def rx_close(self): + self.calls.append("rx_close") + if not self.holds & 2: + return + self.holds &= ~2 + if not self.holds: + self.live = None + + def rx_stats(self): + return (self.captured, self.dropped, len(self.ring)) + + # --- the output, as audiobusio.I2SOut(data_in=) would hold it ---------- + + def output_opens(self, port=0, bclk=12, ws=10, dout=9, din=11, rate=24000): + want = (port, bclk, ws, dout, din, rate) + if self.live is not None and self.live != want: + raise ValueError("another wire") + self.live = want + self.holds |= 1 + + def output_closes(self): + self.holds &= ~1 + if not self.holds: + self.live = None + + # --- the interrupt ------------------------------------------------------ + + def deliver(self, data): + self.ring += data + self.captured += len(data) // 2 diff --git a/tests/test_esp32_p4_audio.py b/tests/test_esp32_p4_audio.py index 693c9e2e..73083369 100644 --- a/tests/test_esp32_p4_audio.py +++ b/tests/test_esp32_p4_audio.py @@ -13,6 +13,7 @@ ROOT = _env.ROOT BOARD = ROOT / "board_configs" / "fbdisplay" / "esp32-p4-wifi6-touch-lcd-4b" +DEV_KIT = ROOT / "board_configs" / "nodisplay" / "esp32-p4-wifi6-dev-kit" class FakeI2C: @@ -91,7 +92,11 @@ def deinit(self): self.closed = True -class ESP32P4AudioTests(unittest.TestCase): +class P4BoardFixture(unittest.TestCase): + """Loads the panel's board_peripherals against fake machine and codecs.""" + + BOARD_DIR = BOARD + @classmethod def setUpClass(cls): cls.saved_modules = { @@ -104,7 +109,9 @@ def setUpClass(cls): sys.modules["machine"] = types.SimpleNamespace( I2S=FakeI2S, Pin=FakePin, PWM=FakePWM, I2C=FakeI2C ) - spec = importlib.util.spec_from_file_location("p4_board_peripherals", BOARD / "board_peripherals.py") + spec = importlib.util.spec_from_file_location( + "p4_board_peripherals_" + cls.BOARD_DIR.name, cls.BOARD_DIR / "board_peripherals.py" + ) cls.board = importlib.util.module_from_spec(spec) spec.loader.exec_module(cls.board) @@ -135,6 +142,8 @@ def tearDown(self): self.board._mclk_rate = self.board._RATE self.board._i2c_own = None + +class ESP32P4AudioTests(P4BoardFixture): def test_output_format_is_24khz_mono_pcm(self): from audiodev import AudioFormat @@ -346,5 +355,128 @@ def test_headless_factory_opens_panel_i2c_without_board_config(self): sys.modules["board_config"] = saved +class PanelRecordsWhilePlaying(P4BoardFixture): + """With the audio pump, pcm_in reads the RX half of the pump's channel. + + The ES8311 and the ES7210 share I2S0's clocks, so machine.I2S can have one + of them at a time (pydevices#23). The pump's driver opens the port as a + TX/RX pair instead; the board publishes the microphone's pin on its wire + and builds a PumpPCMInput, so audio_out plays while pcm_in records. + """ + + def setUp(self): + from _duplex_fake import FakeDuplexDriver + from audiodev import pump + + self.pump = pump + self._saved_driver = (pump._driver, pump._driver_looked) + self.driver = FakeDuplexDriver() + pump._driver = self.driver + pump._driver_looked = True + + def tearDown(self): + self.pump._driver, self.pump._driver_looked = self._saved_driver + super().tearDown() + + def test_the_wire_publishes_the_microphone_pin(self): + self.assertEqual(11, self.board.AUDIO_OUT.wire.sd_in) + + def test_pcm_in_reads_the_pump_channel(self): + capture = self.board.pcm_in() + self.assertIsInstance(capture, self.pump.PumpPCMInput) + capture.set_gain(35) + capture.open() + opened = self.driver.opened_with + self.assertEqual((0, 12, 10, 9, 11, 24000, 13, 256), + (opened["port"], opened["bclk"], opened["ws"], + opened["dout"], opened["din"], opened["rate"], + opened["mclk"], opened["mclk_fs"])) + self.assertEqual(2, opened["take"], "the two microphones are averaged") + self.assertEqual(20000, opened["ring"]) + self.assertEqual(35, capture.codec.gain) + self.assertEqual([], FakeI2S.constructed, "machine.I2S was opened") + capture.close() + self.assertEqual(["rx_open", "rx_close"], self.driver.calls) + self.assertFalse(capture.codec.enabled) + + def test_no_pwm_clock_while_the_channel_drives_gpio13(self): + """A PWM started on GPIO13 while the channel runs takes the pin from + it, and both codecs lose their clock mid-stream.""" + self.driver.output_opens() + capture = self.board.pcm_in() + capture.open() + self.board.audio_power(True) + self.assertEqual([], FakePWM.instances) + capture.close() + + def test_a_running_pwm_is_released_when_the_channel_is_up(self): + self.board._ensure_mclk() + pwm = FakePWM.instances[-1] + self.driver.output_opens() + self.board._ensure_mclk() + self.assertTrue(pwm.closed) + self.assertIsNone(self.board._mclk) + + def test_pcm_out_still_refuses_alongside_a_recording(self): + """pcm_out is machine.I2S whatever the firmware.""" + capture = self.board.pcm_in() + capture.open() + output = self.board.pcm_out() + with self.assertRaises(OSError) as ctx: + output.open() + self.assertIn("audio_out", str(ctx.exception)) + self.assertEqual([], FakeI2S.constructed) + capture.close() + + def test_a_recording_is_refused_while_pcm_out_holds_the_port(self): + output = self.board.pcm_out() + output.open() + capture = self.board.pcm_in() + with self.assertRaises(OSError): + capture.open() + self.assertEqual([], self.driver.calls, "the channel was opened anyway") + output.close() + + +class DevKitRecordsWhilePlaying(P4BoardFixture): + """The DEV-KIT's microphone is the ES8311's own ADC on the same port.""" + + BOARD_DIR = DEV_KIT + + def setUp(self): + from _duplex_fake import FakeDuplexDriver + from audiodev import pump + + self.pump = pump + self._saved_driver = (pump._driver, pump._driver_looked) + self.driver = FakeDuplexDriver() + pump._driver = self.driver + pump._driver_looked = True + + def tearDown(self): + self.pump._driver, self.pump._driver_looked = self._saved_driver + self.board._es8311 = None + super().tearDown() + + def test_pcm_in_reads_the_pump_channel_one_slot(self): + self.assertEqual(11, self.board.AUDIO_OUT.wire.sd_in) + capture = self.board.pcm_in() + self.assertIsInstance(capture, self.pump.PumpPCMInput) + capture.open() + opened = self.driver.opened_with + self.assertEqual((9, 11, 0), (opened["dout"], opened["din"], opened["take"])) + self.assertEqual([], FakeI2S.constructed) + self.assertEqual([], FakePWM.instances, "a PWM clock started on the running channel") + capture.close() + self.assertEqual(["rx_open", "rx_close"], self.driver.calls) + + def test_pcm_out_still_refuses_alongside_a_recording(self): + capture = self.board.pcm_in() + capture.open() + with self.assertRaises(OSError): + self.board.pcm_out().open() + capture.close() + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_pump_capture.py b/tests/test_pump_capture.py new file mode 100644 index 00000000..20a224df --- /dev/null +++ b/tests/test_pump_capture.py @@ -0,0 +1,233 @@ +"""Recording while playing: ``audiodev.pump.PumpPCMInput`` over the pump's RX. + +On a board whose speaker and microphone share one I2S port, ``machine.I2S`` +can only have one of them. The pump's driver opens the port as a TX/RX pair +and lets Python read the RX half; these tests hold ``PumpPCMInput`` and +``BusioDriver`` to that contract against a stand-in driver. +""" + +from pathlib import Path +import sys +import unittest + +_TESTS = Path(__file__).resolve().parent +if str(_TESTS) not in sys.path: + sys.path.insert(0, str(_TESTS)) +import _env # noqa: E402, F401 +from _duplex_fake import FakeDuplexDriver # noqa: E402 + +from audiodev import AudioFormat, AudioSession, I2SWire # noqa: E402 +from audiodev import pump as pump_mod # noqa: E402 + +FMT = AudioFormat(24000, 1, 16) +WIRE = I2SWire(0, sck=12, ws=10, sd=9, mck=13, mck_fs=256, sd_in=11) + + +class DriverFixture(unittest.TestCase): + def setUp(self): + self._saved = (pump_mod._driver, pump_mod._driver_looked) + self.driver = FakeDuplexDriver() + pump_mod._driver = self.driver + pump_mod._driver_looked = True + self.addCleanup(self._restore) + + def _restore(self): + pump_mod._driver, pump_mod._driver_looked = self._saved + + +class TheFirmwareSaysWhetherItCanRecordWhilePlaying(DriverFixture): + def test_a_driver_with_rx_open_is_duplex(self): + self.assertTrue(pump_mod.duplex()) + + def test_a_driver_without_it_is_not(self): + class Old: + def i2s_start(self, *args, **kwargs): + pass + + pump_mod._driver = Old() + self.assertFalse(pump_mod.duplex()) + self.assertFalse(pump_mod.clock_running()) + + def test_no_driver_at_all_is_not(self): + pump_mod._driver = None + self.assertFalse(pump_mod.duplex()) + self.assertFalse(pump_mod.clock_running()) + + def test_clock_running_follows_the_channel(self): + self.assertFalse(pump_mod.clock_running()) + self.driver.output_opens() + self.assertTrue(pump_mod.clock_running()) + self.driver.output_closes() + self.assertFalse(pump_mod.clock_running()) + + +class ReadingTheRxHalf(DriverFixture): + def make(self, **kwargs): + mic = pump_mod.PumpPCMInput(FMT, wire=WIRE, **kwargs) + self.addCleanup(mic.close) + return mic + + def test_open_names_the_speaker_pin_and_the_microphone_pin(self): + """The pair is allocated together, so a recorder that never plays + still names the speaker's pin: an output arriving later has to find + its half already wired.""" + mic = self.make(ring_bytes=9600) + mic.open() + self.assertEqual( + dict(port=0, bclk=12, ws=10, dout=9, din=11, rate=24000, mclk=13, + mclk_fs=256, ring=9600, take=2), + self.driver.opened_with) + + def test_readinto_fills_the_whole_buffer_across_blocks(self): + mic = self.make() + mic.open() + delivered = bytes(range(200)) * 4 + # The ring has some of it now; the rest arrives while readinto waits, + # as DMA blocks do every 5 ms. + self.driver.deliver(delivered[:300]) + later = [delivered[300:500], delivered[500:]] + real_read = self.driver.rx_read + + def read(buf): + n = real_read(buf) + if n == 0 and later: + self.driver.deliver(later.pop(0)) + return n + + self.driver.rx_read = read + buf = bytearray(len(delivered)) + self.assertEqual(len(delivered), mic.readinto(buf)) + self.assertEqual(delivered, bytes(buf)) + + def test_a_channel_that_never_clocks_is_an_error_not_a_hang(self): + mic = self.make(timeout_ms=20) + mic.open() + with self.assertRaises(OSError) as ctx: + mic.readinto(bytearray(64)) + self.assertIn("nothing", str(ctx.exception)) + + def test_the_channel_opens_before_the_codec_is_brought_up(self): + """The codec's setup must run with the channel's MCLK on its pin; a + board that bootstrapped MCLK with PWM while the channel ran would + take the pin from it.""" + order = [] + + def codec(): + order.append("codec, holds=%d" % self.driver.holds) + return object() + + mic = self.make(session=AudioSession(codec_factory=codec)) + mic.open() + self.assertEqual(["codec, holds=2"], order) + + def test_a_codec_that_fails_gives_the_channel_back(self): + def codec(): + raise OSError("no ES7210 on the bus") + + mic = self.make(session=AudioSession(codec_factory=codec)) + with self.assertRaises(OSError): + mic.open() + self.assertEqual(["rx_open", "rx_close"], self.driver.calls) + self.assertFalse(mic.is_open) + self.assertIsNone(self.driver.live) + + def test_close_gives_the_channel_back_once(self): + mic = self.make() + mic.open() + mic.close() + mic.close() + self.assertEqual(["rx_open", "rx_close"], self.driver.calls) + + def test_before_open_can_refuse_and_nothing_is_opened(self): + def refuse(): + raise OSError("I2S(0) is in use by speaker") + + mic = self.make(before_open=refuse) + with self.assertRaises(OSError): + mic.open() + self.assertEqual([], self.driver.calls) + + def test_a_recorder_joins_the_output_at_its_rate(self): + self.driver.output_opens(rate=24000) + mic = self.make() + mic.open() + self.assertEqual(3, self.driver.holds) + mic.close() + self.assertEqual(1, self.driver.holds, + "closing the recorder took the speaker's channel") + + def test_a_recorder_at_another_rate_is_refused_by_name(self): + self.driver.output_opens(rate=44100) + mic = self.make() + with self.assertRaises(ValueError) as ctx: + mic.open() + self.assertIn("44100", str(ctx.exception)) + self.assertFalse(mic.is_open) + + def test_stats_report_what_the_driver_counted(self): + mic = self.make() + mic.open() + self.driver.deliver(bytes(10)) + self.driver.dropped = 3 + self.assertEqual({"captured": 5, "dropped": 3, "waiting": 10}, + mic.stats()) + + def test_take(self): + self.assertEqual(0, self.make(take="left")._take) + self.assertEqual(1, self.make(take="right")._take) + self.assertEqual(2, self.make()._take) + stereo = pump_mod.PumpPCMInput(AudioFormat(24000, 2, 16), wire=WIRE) + self.assertEqual(3, stereo._take) + with self.assertRaises(ValueError): + self.make(take="both") + + def test_a_wire_without_a_microphone_pin_is_refused(self): + with self.assertRaises(ValueError): + pump_mod.PumpPCMInput(FMT, wire=I2SWire(0, sck=12, ws=10, sd=9)) + + def test_only_signed_16_bit(self): + with self.assertRaises(ValueError): + pump_mod.PumpPCMInput(AudioFormat(24000, 1, 32), wire=WIRE) + + +class FakeI2SOut: + built = [] + + def __init__(self, *pins, **kwargs): + self.pins = pins + self.kwargs = kwargs + FakeI2SOut.built.append(self) + + +class TheOutputOpensThePairWhenTheBoardPublishesAMicPin(DriverFixture): + def setUp(self): + super().setUp() + self._saved_busio = (pump_mod._busio, pump_mod._busio_looked) + pump_mod._busio = type("Busio", (), {"I2SOut": FakeI2SOut})() + pump_mod._busio_looked = True + FakeI2SOut.built = [] + self.addCleanup(self._put_back) + + def _put_back(self): + pump_mod._busio, pump_mod._busio_looked = self._saved_busio + + def test_data_in_reaches_the_constructor(self): + pump_mod.BusioDriver(WIRE, FMT).open() + self.assertEqual(11, FakeI2SOut.built[-1].kwargs.get("data_in")) + + def test_not_on_firmware_that_cannot_read_rx(self): + class Old: + def i2s_start(self, *args, **kwargs): + pass + + pump_mod._driver = Old() + pump_mod.BusioDriver(WIRE, FMT).open() + self.assertNotIn("data_in", FakeI2SOut.built[-1].kwargs) + + def test_not_for_a_wire_without_a_microphone(self): + pump_mod.BusioDriver(I2SWire(0, sck=12, ws=10, sd=9), FMT).open() + self.assertNotIn("data_in", FakeI2SOut.built[-1].kwargs) + + +if __name__ == "__main__": + unittest.main()