diff --git a/midiports/README.md b/midiports/README.md index 645f563..a64511d 100644 --- a/midiports/README.md +++ b/midiports/README.md @@ -29,6 +29,8 @@ To use another transport, change the line that opens the ports: | A UART (5-pin DIN, TRS, board to board) | `uart.ports(1, tx=17, rx=18)` | | Windows' MIDI ports (a keyboard, loopMIDI, a DAW) | `windows.ports("loopMIDI")` | | Linux raw MIDI | `linux.ports("DMK25")` | +| Web MIDI, in a browser page (Pyodide) | `await webmidi.open("DMK25")` | +| Web MIDI, in a Web Worker, relayed by the page | `webmidi.relay_ports(port)` | Import each from `midiports`. BLE MIDI ports come from [`bledev`](../docs/bledev.md), which moves onto this package's ports next. @@ -71,12 +73,57 @@ Input ports hand out whole messages. A 5-pin cable sends running status middle of a note, and `adafruit_midi` reads neither. So the UART and Linux ports wrap their input in `midiports.WholeMessages`, which fixes both. Wrap any other byte stream the same way. USB-MIDI and Windows already deliver -whole messages. +whole messages, and so does Web MIDI. A port that polls beside audio should call `readinto` with a buffer of its own. `read` makes a new `bytes` for every message that arrives, though an empty poll allocates nothing. +## In a browser + +Python running in Pyodide reaches MIDI devices through Web MIDI. The browser +asks the person for permission the first time. + +On the page (Pyodide on the main thread, as PyScript runs it), `open` is +awaitable: + +```python +from midiports import webmidi + +midi_in, midi_out = await webmidi.open("DMK25", sysex=False) +midi = adafruit_midi.MIDI(midi_in=midi_in, midi_out=midi_out) +``` + +The name matches part of a port's name, ignoring case. Leave it out to merge +every input and send to the first output. A device plugged in later, or +unplugged and plugged back in, joins while the ports are open. If you +already hold a `MIDIAccess`, `webmidi.ports(access, "DMK25")` does the same +without awaiting. Sending SysEx needs `sysex=True`. + +Web MIDI doesn't exist in a Web Worker, so a worker gets its MIDI from the +page over a `MessagePort`, and a few lines of JavaScript on the page relay +it. Open the worker's end with `webmidi.relay_ports(port)`. The relay's +messages are plain objects: + +| Direction | Message | +|---|---| +| page to worker | `{data: Uint8Array, timeStamp: number}`: one whole MIDI message per post, `timeStamp` from its `MIDIMessageEvent` | +| worker to page | `{data: Uint8Array}`: whole messages to `send()` to the page's output | + +Every Web MIDI input keeps the time each message arrived, in milliseconds on +the `performance.now()` clock. Call `midi_in.take_timestamp()` once for each +message `receive()` gives you and you get that message's time, or `None` when +there is none: + +```python +msg = midi.receive() +if msg is not None: + when = midi_in.take_timestamp() +``` + +That works because Web MIDI hands over whole messages, one timestamp each. +The [module docstring](midiports/webmidi.py) has the details. + ## Limits - Windows: no SysEx yet. Writing one raises `NotImplementedError` rather than diff --git a/midiports/midiports/__init__.py b/midiports/midiports/__init__.py index 0716993..451d3e4 100644 --- a/midiports/midiports/__init__.py +++ b/midiports/midiports/__init__.py @@ -47,7 +47,7 @@ message. ``adafruit_midi`` needs both. It reads no running status and turns a clock byte inside a note into a ``MIDIBadEvent``, and a 5-pin cable sends both. A transport whose bytes can arrive that way (a UART, Linux raw MIDI) -wraps its input in :class:`WholeMessages`; USB-MIDI and Windows already +wraps its input in :class:`WholeMessages`; USB-MIDI, Windows and Web MIDI already deliver whole messages. A port that holds an OS handle (the desktop ones) also has ``close()`` and diff --git a/midiports/midiports/webmidi.py b/midiports/midiports/webmidi.py new file mode 100644 index 0000000..e9691b8 --- /dev/null +++ b/midiports/midiports/webmidi.py @@ -0,0 +1,474 @@ +"""MIDI in a browser, through Web MIDI, for Python running in Pyodide. + +Web MIDI (``navigator.requestMIDIAccess``) exists only on a page's main +thread, never in a Web Worker, so there are two ways in. + +On the page, with Pyodide on the main thread (PyScript, for one):: + + from midiports import webmidi + + midi_in, midi_out = await webmidi.open("keystation") + +``name`` matches part of a port's name, ignoring case. The default, ``""``, +merges every input and sends to the first output. The browser asks the +person for permission the first time. A device plugged in later, or +unplugged and plugged back in, is picked up while the ports are open. If +your code already holds a ``MIDIAccess``, ``ports(access, name)`` does the +same without awaiting. + +In a worker, a few lines of JavaScript on the page relay MIDI over a +``MessagePort``, and the worker opens ports on it:: + + midi_in, midi_out = webmidi.relay_ports(port) + +The relay's messages are plain objects: + +- page to worker: ``{data: Uint8Array, timeStamp: number}``, one whole MIDI + message per post, with ``timeStamp`` taken from the ``MIDIMessageEvent``. +- worker to page: ``{data: Uint8Array}``, whole messages for the page to + ``send()`` to its output. + +Timestamps +---------- +Each input keeps the time every message arrived, in milliseconds on the +``performance.now()`` clock, as Web MIDI stamps it. ``take_timestamp()`` +pops the oldest one not yet taken, or returns ``None`` when there is none. +Call it once for each message ``adafruit_midi.MIDI.receive()`` hands you and +you get that message's time:: + + msg = midi.receive() + if msg is not None: + when = midi_in.take_timestamp() + +That pairing holds because Web MIDI delivers whole messages, one timestamp +each. A message dropped because the input was full takes its timestamp with +it. If you never call ``take_timestamp()``, old timestamps are discarded and +nothing else changes. + +Received bytes wait in a ring (``ring`` bytes, 1024 by default). If you don't +read for long enough to fill it, whole messages are dropped and counted in +``midi_in.overflows``. Writes are sent a whole message at a time, as Web MIDI +requires; with no output connected they are dropped and counted in +``midi_out.dropped``. Sending SysEx on the page needs ``open(sysex=True)``. + +Inputs don't wrap themselves in ``WholeMessages``: Web MIDI already hands +over whole messages, never running status and never a clock byte inside a +note. +""" + +from array import array +from math import isnan + +from . import PortIn, PortOut, _count, message_length + +#: Longest unfinished message a write holds back, waiting for its end. +PENDING_LIMIT = 4096 + +_NAN = float("nan") + + +# -- the browser's FFI, imported only when it's there ------------------------------ + + +def _proxy(fn): + """A JavaScript callable for ``fn`` that outlives this call.""" + try: + from pyodide.ffi import create_proxy + except ImportError: + return fn + return create_proxy(fn) + + +def _destroy(proxy): + destroy = getattr(proxy, "destroy", None) + if destroy is not None: + destroy() + + +def _uint8(data): + """``data`` as a JavaScript ``Uint8Array`` (plain ``bytes`` outside a browser).""" + try: + from js import Uint8Array + except ImportError: + return bytes(data) + out = Uint8Array.new(len(data)) + out.assign(data) + return out + + +def _object(**fields): + """A plain JavaScript object (a ``dict`` outside a browser).""" + try: + from js import Object + from pyodide.ffi import to_js + except ImportError: + return fields + return to_js(fields, dict_converter=Object.fromEntries) + + +def _bytes(data): + to_bytes = getattr(data, "to_bytes", None) + return to_bytes() if to_bytes is not None else bytes(data) + + +def _ms(stamp): + try: + return float(stamp) + except (TypeError, ValueError): + return None + + +def _matches(port, name): + return not name or name.lower() in str(port.name).lower() + + +def _values(maplike): + values = maplike.values() + return list(values) + + +# -- input: a byte ring with a timestamp per message --------------------------------- + + +class _RingIn(PortIn): + """Whole messages pushed in by the browser, read out like ``usb_midi``.""" + + def __init__(self, ring=1024): + PortIn.__init__(self) + self._ring = bytearray(ring) + self._head = 0 + self._tail = 0 + # One timestamp per message, oldest first. Room for one per byte of + # the ring, plus some for messages already read whose timestamp + # hasn't been taken yet. + self._times = array("d", [0.0] * (ring + 64)) + self._thead = 0 + self._ttail = 0 + self.overflows = 0 + self._closed = False + + def push(self, data, timestamp=None): + """Queue one whole message that arrived at ``timestamp`` (ms).""" + if self._closed: + return + n = len(data) + if not n: + return + ring = self._ring + size = len(ring) + head = self._head + if (self._tail - head - 1) % size < n: + self.overflows += 1 + return + for i in range(n): + ring[head] = data[i] + head = (head + 1) % size + self._head = head + times = self._times + tsize = len(times) + thead = self._thead + times[thead] = _NAN if timestamp is None else timestamp + thead = (thead + 1) % tsize + if thead == self._ttail: + # Full of timestamps nobody took: forget the oldest. + self._ttail = (self._ttail + 1) % tsize + self._thead = thead + + def take_timestamp(self): + """The arrival time (ms) of the next message, in order, or ``None``.""" + tail = self._ttail + if tail == self._thead: + return None + stamp = self._times[tail] + self._ttail = (tail + 1) % len(self._times) + return None if isnan(stamp) else stamp + + def any(self): + if self._closed: + raise OSError("MIDI input is closed") + return (self._head - self._tail) % len(self._ring) + + def readinto(self, buf, nbytes=None): + if self._closed: + raise OSError("MIDI input is closed") + nbytes = _count(buf, nbytes) + ring = self._ring + size = len(ring) + tail = self._tail + head = self._head + i = 0 + while i < nbytes and tail != head: + buf[i] = ring[tail] + tail = (tail + 1) % size + i += 1 + self._tail = tail + return i or None + + def close(self): + self._closed = True + + deinit = close + + +# -- output: whole messages only ----------------------------------------------------- + + +class _WholeOut(PortOut): + """Sends whole messages; a partial one waits for the rest of it. + + Web MIDI's ``send()`` refuses part of a message, so bytes that end + mid-message are held until the write that finishes it. A data byte with + no status before it can't be sent and is dropped. + """ + + def __init__(self): + self._pending = bytearray() + self.dropped = 0 + self._closed = False + + def _deliver(self, data): + raise NotImplementedError + + def write(self, buf, nbytes=None): + if self._closed: + raise OSError("MIDI output is closed") + n = _count(buf, nbytes) + p = self._pending + p.extend(memoryview(buf)[:n]) + size = len(p) + out = bytearray() + i = 0 + done = 0 + while i < size: + b = p[i] + if b < 0x80 or b == 0xF7: + i += 1 # no status to belong to + elif b >= 0xF8: + out.append(b) + i += 1 + elif b == 0xF0: + j = i + 1 + while j < size and p[j] != 0xF7: + j += 1 + if j == size: + break + out.extend(p[i:j + 1]) + i = j + 1 + else: + end = i + 1 + message_length(b) + if end > size: + break + out.extend(p[i:end]) + i = end + done = i + if done: + self._pending = p[done:] + if len(self._pending) > PENDING_LIMIT: + self._pending = bytearray() + self.dropped += 1 + if out: + self._deliver(out) + return n + + def close(self): + self._closed = True + + deinit = close + + +# -- on the page: a MIDIAccess ------------------------------------------------------- + + +class _PageIn(_RingIn): + def __init__(self, session, ring): + _RingIn.__init__(self, ring) + self._session = session + + @property + def names(self): + """The names of the inputs feeding this port right now.""" + return [port.name for port, _ in self._session.inputs.values()] + + def close(self): + if not self._closed: + _RingIn.close(self) + self._session.closed_one() + + deinit = close + + +class _PageOut(_WholeOut): + def __init__(self, session): + _WholeOut.__init__(self) + self._session = session + + @property + def name(self): + """The name of the output this port sends to, or ``None``.""" + output = self._session.output + return None if output is None else output.name + + def _deliver(self, data): + output = self._session.output + if output is None: + self.dropped += 1 + return + output.send(_uint8(data)) + + def close(self): + if not self._closed: + _WholeOut.close(self) + self._session.closed_one() + + deinit = close + + +class _Session: + """The inputs and output one ``ports()`` call holds, kept current.""" + + def __init__(self, access, name, ring): + self.access = access + self.name = name + self.inputs = {} # id -> (MIDIInput, handler proxy) + self.output = None + self.open = 2 + self.midi_in = _PageIn(self, ring) + self.midi_out = _PageOut(self) + for port in _values(access.inputs): + self.attach(port) + for port in _values(access.outputs): + self.adopt(port) + self._on_state = _proxy(self.on_state) + access.addEventListener("statechange", self._on_state) + + def attach(self, port): + if port.id in self.inputs or not _matches(port, self.name): + return + midi_in = self.midi_in + + def on_message(event): + midi_in.push(_bytes(event.data), _ms(event.timeStamp)) + + handler = _proxy(on_message) + self.inputs[port.id] = (port, handler) + port.onmidimessage = handler # also opens the port + + def detach(self, port_id): + entry = self.inputs.pop(port_id, None) + if entry is not None: + port, handler = entry + port.onmidimessage = None + _destroy(handler) + + def adopt(self, port): + if self.output is None and _matches(port, self.name): + self.output = port + + def on_state(self, event): + port = event.port + connected = port.state == "connected" + if port.type == "input": + if self.midi_in._closed: + return + if connected: + self.attach(port) + else: + self.detach(port.id) + elif port.type == "output": + if self.midi_out._closed: + return + if connected: + self.adopt(port) + elif self.output is not None and self.output.id == port.id: + self.output = None + + def closed_one(self): + if self.midi_in._closed: + for port_id in list(self.inputs): + self.detach(port_id) + if self.midi_out._closed: + self.output = None + self.open -= 1 + if not self.open: + self.access.removeEventListener("statechange", self._on_state) + _destroy(self._on_state) + + +def ports(access, name="", ring=1024): + """``(midi_in, midi_out)`` over a ``MIDIAccess`` you already hold. + + ``name`` matches part of a port's name, ignoring case; ``""`` merges every + input and sends to the first output. Ports that match later, plugged in + or plugged back in, join while these are open. Closing both lets go of + the browser's ports. + """ + session = _Session(access, name, ring) + return session.midi_in, session.midi_out + + +async def _request_access(sysex): + from js import navigator + + if not hasattr(navigator, "requestMIDIAccess"): + raise OSError( + "this browser has no Web MIDI here (a worker never does: relay it from the page)" + ) + return await navigator.requestMIDIAccess(_object(sysex=bool(sysex))) + + +async def open(name="", sysex=False, ring=1024): + """Ask the browser for MIDI, then ``ports()`` on it. Awaitable. + + ``sysex=True`` asks for SysEx too, which browsers grant separately. + Raises ``OSError`` where there is no Web MIDI, which includes any worker. + """ + access = await _request_access(sysex) + return ports(access, name, ring) + + +# -- in a worker: the page relays over a MessagePort ------------------------------ + + +class _RelayIn(_RingIn): + def __init__(self, port, ring): + _RingIn.__init__(self, ring) + self._port = port + self._handler = _proxy(self._on_message) + port.addEventListener("message", self._handler) + start = getattr(port, "start", None) + if start is not None: + start() + + def _on_message(self, event): + message = event.data + data = getattr(message, "data", None) + if data is None: + return + self.push(_bytes(data), _ms(getattr(message, "timeStamp", None))) + + def close(self): + if not self._closed: + _RingIn.close(self) + self._port.removeEventListener("message", self._handler) + _destroy(self._handler) + + deinit = close + + +class _RelayOut(_WholeOut): + def __init__(self, port): + _WholeOut.__init__(self) + self._port = port + + def _deliver(self, data): + self._port.postMessage(_object(data=_uint8(data))) + + +def relay_ports(port, ring=1024): + """``(midi_in, midi_out)`` over a ``MessagePort`` the page relays MIDI on. + + For a worker, where Web MIDI doesn't exist. The page posts + ``{data, timeStamp}`` for each message it receives and sends the + ``{data}`` this posts back. Closing the ports leaves ``port`` open. + """ + return _RelayIn(port, ring), _RelayOut(port) diff --git a/midiports/package.json b/midiports/package.json index faf75da..8e6f1e1 100644 --- a/midiports/package.json +++ b/midiports/package.json @@ -24,6 +24,10 @@ "midiports/linux.py", "github:PyDevices/pydevices/midiports/midiports/linux.py" ], + [ + "midiports/webmidi.py", + "github:PyDevices/pydevices/midiports/midiports/webmidi.py" + ], [ "adafruit_midi/__init__.py", "https://raw.githubusercontent.com/adafruit/Adafruit_CircuitPython_MIDI/1.6.0/adafruit_midi/__init__.py" diff --git a/tests/midiports_contract.py b/tests/midiports_contract.py index dcd9016..e9f8ea7 100644 --- a/tests/midiports_contract.py +++ b/tests/midiports_contract.py @@ -14,8 +14,8 @@ ``tests/test_midiports.py`` fetches it there and runs this every way. Each transport is checked with a fake underneath: a fake ``machine.UART``, -a fake ``usb.device.midi.MIDIInterface``, a fake usbif ``MidiPort`` and a -fake ``uwin32``. The Linux port needs real files and is checked in +a fake ``usb.device.midi.MIDIInterface``, a fake usbif ``MidiPort``, a +fake ``uwin32`` and, for Web MIDI, a fake ``MIDIAccess`` and ``MessagePort``. The Linux port needs real files and is checked in ``test_midiports.py`` on CPython. It prints one PASS/FAIL line per check and exits non-zero on any failure. @@ -654,6 +654,197 @@ def c_windows(): sys.modules.pop("midiports.windows", None) +# -- Web MIDI, over a fake MIDIAccess and a fake MessagePort --------------------------------- + + +def split_messages(data): + """What Web MIDI does to a send(): one event per whole message.""" + data = bytes(data) + out = [] + i = 0 + while i < len(data): + b = data[i] + if b == 0xF0: + end = data.index(b"\xf7", i) + 1 if b"\xf7" in data[i:] else len(data) + else: + end = i + 1 + midiports.message_length(b) + out.append(data[i:end]) + i = end + return out + + +class _Event: + def __init__(self, **fields): + for k, v in fields.items(): + setattr(self, k, v) + + +class FakeMIDIPort: + def __init__(self, id, name, type): + self.id = id + self.name = name + self.type = type + self.state = "connected" + self.onmidimessage = None + self.sent = [] + + def send(self, data): + self.sent.append(bytes(data)) + + +class _Map: + def __init__(self, ports): + self.ports = ports + + def values(self): + return iter(self.ports) + + +class FakeMIDIAccess: + """A MIDIAccess with a loopback device: what goes out comes back in.""" + + def __init__(self): + self.clock = 1000.0 + self.loop_in = FakeMIDIPort("i1", "Loopback Port", "input") + self.loop_out = FakeMIDIPort("o1", "Loopback Port", "output") + self.inputs = _Map([self.loop_in]) + self.outputs = _Map([self.loop_out]) + self.listeners = [] + self.loop_out.send = self._loop + + def _loop(self, data): + for msg in split_messages(data): + self.fire(self.loop_in, msg) + + def fire(self, port, msg): + self.clock += 0.5 + if port.onmidimessage is not None: + port.onmidimessage(_Event(data=msg, timeStamp=self.clock)) + + def addEventListener(self, kind, fn): + self.listeners.append((kind, fn)) + + def removeEventListener(self, kind, fn): + self.listeners.remove((kind, fn)) + + def change(self, port, state): + port.state = state + for kind, fn in self.listeners: + if kind == "statechange": + fn(_Event(port=port)) + + +class FakeMessagePort: + """A worker's end of the relay; the page loops every post back.""" + + def __init__(self): + self.clock = 0.0 + self.listeners = [] + self.posted = [] + self.started = False + self.loop = True + + def addEventListener(self, kind, fn): + self.listeners.append((kind, fn)) + + def removeEventListener(self, kind, fn): + self.listeners.remove((kind, fn)) + + def start(self): + self.started = True + + def deliver(self, msg, stamp=None): + self.clock += 1.0 + payload = _Event(data=msg, timeStamp=self.clock if stamp is None else stamp) + for kind, fn in self.listeners: + if kind == "message": + fn(_Event(data=payload)) + + def postMessage(self, message): + self.posted.append(bytes(message["data"])) + if self.loop: + for msg in split_messages(message["data"]): + self.deliver(msg) + + +def stamps_follow_receive(midi_in, feed): + """receive() one message at a time; take_timestamp() names each one's time.""" + rx = adafruit_midi.MIDI(midi_in=midi_in) + feed() + got = [] + while True: + msg = rx.receive() + if msg is None: + break + got.append((msg, midi_in.take_timestamp())) + return got + + +def c_webmidi_page(): + from midiports import webmidi + + access = FakeMIDIAccess() + midi_in, midi_out = webmidi.ports(access, "loopback") + expect(midi_in.names == ["Loopback Port"] and midi_out.name == "Loopback Port", "attached") + expect(midi_in.any() == 0, "nothing waiting") + quiet_any_allocates_nothing(midi_in) + got = round_trip(midi_in, midi_out) + expect(got == 200, "%d of 200 through a page MIDIAccess" % got) + got = any_first(midi_in, midi_out) + expect(got == 200, "%d of 200 through a page MIDIAccess, any() first" % got) + while midi_in.take_timestamp() is not None: + pass + + def feed(): + for msg in (b"\x90\x3c\x64", b"\xf8", b"\xb0\x07\x7f"): + access.fire(access.loop_in, msg) + + start = access.clock + got = stamps_follow_receive(midi_in, feed) + expect([type(m) for m, _ in got] == [NoteOn, TimingClock, ControlChange], "got %r" % (got,)) + expect([t for _, t in got] == [start + 0.5, start + 1.0, start + 1.5], + "timestamps %r" % ([t for _, t in got],)) + expect(midi_in.take_timestamp() is None, "no more timestamps") + midi_in.close() + midi_out.close() + expect(access.listeners == [] and access.loop_in.onmidimessage is None, "closed cleanly") + + +def c_webmidi_relay(): + from midiports import webmidi + + port = FakeMessagePort() + midi_in, midi_out = webmidi.relay_ports(port) + expect(port.started, "the port is started") + expect(midi_in.any() == 0, "nothing waiting") + quiet_any_allocates_nothing(midi_in) + got = round_trip(midi_in, midi_out) + expect(got == 200, "%d of 200 through the relay" % got) + got = any_first(midi_in, midi_out) + expect(got == 200, "%d of 200 through the relay, any() first" % got) + while midi_in.take_timestamp() is not None: + pass + port.loop = False + port.posted[:] = [] + midi_out.write(b"\x90\x3c") + expect(port.posted == [], "half a message waits") + midi_out.write(b"\x64\xf0\x7d\x01") + expect(port.posted == [b"\x90\x3c\x64"], "posted %r" % (port.posted,)) + midi_out.write(b"\xf7") + expect(port.posted[-1] == b"\xf0\x7d\x01\xf7", "sysex posted whole") + + def feed(): + port.deliver(b"\x90\x3c\x64", 12.5) + port.deliver(b"\x80\x3c\x00", 20.0) + + got = stamps_follow_receive(midi_in, feed) + expect(len(got) == 2 and isinstance(got[0][0], NoteOn) and isinstance(got[1][0], NoteOff), + "got %r" % (got,)) + expect(got[0][1] == 12.5 and got[1][1] == 20.0, "timestamps %r" % (got,)) + midi_in.close() + expect(port.listeners == [], "close removes the listener") + + def any_first(midi_in, midi_out, notes=200, pump=None): """The loop a rack beside audio runs: receive() only when any() says so.""" tx = adafruit_midi.MIDI(midi_out=midi_out) @@ -757,6 +948,8 @@ def c_garbage_per_note(): ("usbif_shim: 200 of 200, read(n) over read(buf)", c_usbif_shim), ("windows: ports by name, short messages, drained input", c_windows), ("any(): counts without taking, allocates nothing, drives a loop", c_any), + ("webmidi: a page MIDIAccess, 200 of 200, timestamps follow receive()", c_webmidi_page), + ("webmidi: a worker relay, 200 of 200, whole messages posted", c_webmidi_relay), ("garbage per received note", c_garbage_per_note), ) diff --git a/tests/test_midiports.py b/tests/test_midiports.py index 0e2a6d9..34d0570 100644 --- a/tests/test_midiports.py +++ b/tests/test_midiports.py @@ -1,8 +1,9 @@ # SPDX-FileCopyrightText: 2026 Brad Barnett # # SPDX-License-Identifier: MIT -"""midiports: its package file, its contract on CPython and MicroPython, and -the Linux port against a synthetic ALSA tree. +"""midiports: its package file, its contract on CPython and MicroPython, the +Linux port against a synthetic ALSA tree, and the Web MIDI port against fake +browser objects. The contract (``midiports_contract.py``) runs with Adafruit's ``adafruit_midi``, fetched from the URLs ``midiports/package.json`` pins, so @@ -13,6 +14,7 @@ """ from pathlib import Path +import asyncio import json import os import shutil @@ -32,7 +34,7 @@ sys.path.insert(0, str(PACKAGE)) import midiports # noqa: E402 -from midiports import IN, OUT, WholeMessages, linux # noqa: E402 +from midiports import IN, OUT, WholeMessages, linux, webmidi # noqa: E402 def _package_json(): @@ -218,5 +220,216 @@ def test_no_matching_port_says_so(self): self.assertRaises(OSError, linux.ports, "nothing", proc=self.proc, dev=self.dev) +class _Obj: + def __init__(self, **fields): + self.__dict__.update(fields) + + +class _Port: + def __init__(self, id, name, type): + self.id, self.name, self.type, self.state = id, name, type, "connected" + self.onmidimessage = None + self.sent = [] + + def send(self, data): + self.sent.append(bytes(data)) + + def play(self, data, stamp): + self.onmidimessage(_Obj(data=data, timeStamp=stamp)) + + +class _Map: + def __init__(self, ports): + self.ports = ports + + def values(self): + return iter(self.ports) + + +class _Access: + def __init__(self, inputs=(), outputs=()): + self.inputs = _Map(list(inputs)) + self.outputs = _Map(list(outputs)) + self.listeners = [] + + def addEventListener(self, kind, fn): + self.listeners.append((kind, fn)) + + def removeEventListener(self, kind, fn): + self.listeners.remove((kind, fn)) + + def change(self, port, state): + port.state = state + for _, fn in list(self.listeners): + fn(_Obj(port=port)) + + +class _MessagePort: + def __init__(self): + self.listeners = [] + self.posted = [] + + def addEventListener(self, kind, fn): + self.listeners.append((kind, fn)) + + def removeEventListener(self, kind, fn): + self.listeners.remove((kind, fn)) + + def postMessage(self, message): + self.posted.append(message) + + def deliver(self, payload): + for _, fn in list(self.listeners): + fn(_Obj(data=payload)) + + +class WebMidiPort(unittest.TestCase): + """Web MIDI, against fake MIDIAccess, MIDIPort and MessagePort objects.""" + + def test_the_ring_keeps_whole_messages_and_drops_whole_ones_when_full(self): + midi_in = webmidi._RingIn(ring=8) + self.assertIsNone(midi_in.read(16)) + self.assertEqual(midi_in.any(), 0) + for note in range(4): + midi_in.push(bytes((0x90, note, 0x40)), float(note)) + self.assertEqual(midi_in.overflows, 2) # 7 bytes fit: two notes, never a part + self.assertEqual(midi_in.any(), 6) + self.assertEqual(midi_in.read(4), b"\x90\x00\x40\x90") + self.assertEqual(midi_in.any(), 2) + buf = bytearray(8) + self.assertEqual(midi_in.readinto(buf, 8), 2) + self.assertEqual(bytes(buf[:2]), b"\x01\x40") + self.assertEqual([midi_in.take_timestamp() for _ in range(3)], [0.0, 1.0, None]) + for _ in range(5): # wraps round the ring + midi_in.push(b"\xb0\x07\x7f") + self.assertEqual(midi_in.read(16), b"\xb0\x07\x7f") + midi_in.close() + self.assertRaises(OSError, midi_in.read, 4) + self.assertRaises(OSError, midi_in.any) + + def test_timestamps_pair_with_messages_and_stale_ones_are_forgotten(self): + midi_in = webmidi._RingIn(ring=16) + midi_in.push(b"\x90\x3c\x64", 5.25) + midi_in.push(b"\xf8") # no stamp + midi_in.push(b"\x80\x3c\x00", 9.0) + self.assertEqual(midi_in.take_timestamp(), 5.25) + self.assertIsNone(midi_in.take_timestamp()) + self.assertEqual(midi_in.take_timestamp(), 9.0) + self.assertIsNone(midi_in.take_timestamp()) + self.assertEqual(midi_in.read(16), b"\x90\x3c\x64\xf8\x80\x3c\x00") + # A reader that never takes timestamps loses nothing else. + for i in range(500): + midi_in.push(b"\xf8", float(i)) + self.assertEqual(midi_in.read(4), b"\xf8") + self.assertEqual(midi_in.overflows, 0) + oldest = midi_in.take_timestamp() + self.assertEqual(oldest, 500.0 - (len(midi_in._times) - 1)) + + def test_ports_attach_by_name_and_merge_inputs(self): + keys = _Port("i1", "Keystation 49", "input") + pads = _Port("i2", "Pad Controller", "input") + synth = _Port("o1", "Some Synth", "output") + access = _Access([keys, pads], [synth]) + midi_in, midi_out = webmidi.ports(access, "KEYSTATION") + self.assertEqual(midi_in.names, ["Keystation 49"]) + self.assertIsNone(pads.onmidimessage) + self.assertIsNone(midi_out.name) + midi_out.write(b"\x90\x3c\x64") + self.assertEqual(midi_out.dropped, 1) + both_in, both_out = webmidi.ports(access) + keys.play(b"\x90\x3c\x64", 1.0) + pads.play(b"\x99\x24\x7f", 2.0) + self.assertEqual(both_in.read(16), b"\x90\x3c\x64\x99\x24\x7f") + self.assertEqual(both_out.name, "Some Synth") + both_out.write(b"\xb0\x07\x7f\xc0\x05") + self.assertEqual(synth.sent, [b"\xb0\x07\x7f\xc0\x05"]) + + def test_statechange_reattaches_a_port_that_comes_back(self): + keys = _Port("i1", "Keys", "input") + synth = _Port("o1", "Keys", "output") + access = _Access([keys], [synth]) + midi_in, midi_out = webmidi.ports(access, "keys") + access.change(keys, "disconnected") + access.change(synth, "disconnected") + self.assertIsNone(keys.onmidimessage) + self.assertEqual(midi_in.names, []) + self.assertIsNone(midi_out.name) + access.change(keys, "connected") + access.change(keys, "connected") # opening fires another; still one handler + access.change(synth, "connected") + self.assertEqual(midi_in.names, ["Keys"]) + keys.play(b"\x90\x3c\x64", 3.0) + self.assertEqual(midi_in.read(16), b"\x90\x3c\x64") + self.assertEqual(midi_in.take_timestamp(), 3.0) + midi_out.write(b"\xfa") + self.assertEqual(synth.sent, [b"\xfa"]) + late = _Port("i9", "Keys 2", "input") + access.change(late, "connected") + self.assertEqual(sorted(midi_in.names), ["Keys", "Keys 2"]) + other = _Port("i8", "Drums", "input") + access.change(other, "connected") + self.assertIsNone(other.onmidimessage) + + def test_closing_both_lets_go_of_every_port(self): + keys = _Port("i1", "Keys", "input") + access = _Access([keys], [_Port("o1", "Keys", "output")]) + midi_in, midi_out = webmidi.ports(access) + midi_in.close() + self.assertIsNone(keys.onmidimessage) + self.assertEqual(len(access.listeners), 1) + access.change(keys, "connected") # a closed input attaches nothing + self.assertIsNone(keys.onmidimessage) + midi_out.close() + midi_out.close() + self.assertEqual(access.listeners, []) + self.assertRaises(OSError, midi_out.write, b"\xfa") + + def test_open_requests_access_and_says_so_without_web_midi(self): + access = _Access([_Port("i1", "Keys", "input")], []) + asked = [] + + async def fake_request(sysex): + asked.append(sysex) + return access + + real = webmidi._request_access + webmidi._request_access = fake_request + try: + midi_in, _ = asyncio.run(webmidi.open("keys", sysex=True)) + finally: + webmidi._request_access = real + self.assertEqual(asked, [True]) + self.assertEqual(midi_in.names, ["Keys"]) + # Outside a browser there's no js module at all. + self.assertRaises(ImportError, asyncio.run, webmidi.open()) + + def test_relay_input_takes_page_messages_with_their_timestamps(self): + port = _MessagePort() + midi_in, _ = webmidi.relay_ports(port) + port.deliver(_Obj(data=b"\x90\x3c\x64", timeStamp=41.5)) + port.deliver(_Obj(data=b"\xf8")) # no timeStamp + port.deliver(_Obj(other=1)) # not MIDI: ignored + self.assertEqual(midi_in.any(), 4) + self.assertEqual(midi_in.read(16), b"\x90\x3c\x64\xf8") + self.assertEqual(midi_in.take_timestamp(), 41.5) + self.assertIsNone(midi_in.take_timestamp()) + midi_in.close() + self.assertEqual(port.listeners, []) + + def test_relay_output_posts_whole_messages(self): + port = _MessagePort() + _, midi_out = webmidi.relay_ports(port) + self.assertEqual(midi_out.write(b"\x3c\x90\x3c", 3), 3) # a stray data byte, then half a note + self.assertEqual(port.posted, []) + midi_out.write(b"\x64\xf8\xf0\x7d") + self.assertEqual(port.posted, [{"data": b"\x90\x3c\x64\xf8"}]) + midi_out.write(bytearray(b"\x01\x02\xf7\xb0\x07\x7f")) + self.assertEqual(port.posted[-1], {"data": b"\xf0\x7d\x01\x02\xf7\xb0\x07\x7f"}) + midi_out.write(b"\xf0" + bytes(webmidi.PENDING_LIMIT)) # a SysEx that never ends + self.assertEqual(midi_out.dropped, 1) + midi_out.write(b"\xfa") + self.assertEqual(port.posted[-1], {"data": b"\xfa"}) + + if __name__ == "__main__": unittest.main()