diff --git a/lib/programkit/__init__.py b/lib/programkit/__init__.py new file mode 100644 index 0000000..0e65d60 --- /dev/null +++ b/lib/programkit/__init__.py @@ -0,0 +1,27 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Reach a board's display, keys and touch from a program that runs without Home. + +A program in ``programs/`` runs on any board with the hardware it needs, so it +never names a pin. `Board` finds what the board's ``board_config`` and +``board_peripherals`` offer, `Rows` puts lines of text on whatever display is +there, and `Press` hears a press of any key, touch or button:: + + from programkit import Board, Press, Rows + + board = Board() + rows = Rows(board.display) if board.display else None + press = Press(board) + if board.has("audio_out"): + out = board.get("audio_out")(AudioFormat(44100, 2, 16)) + +A board with no display has no ``board_config``; `Board` then reads +``board_peripherals`` alone, which is how a headless board describes itself. +""" + +from .board import Board +from .press import Press +from .rows import Rows + +__all__ = ["Board", "Press", "Rows"] diff --git a/lib/programkit/board.py b/lib/programkit/board.py new file mode 100644 index 0000000..fc2706a --- /dev/null +++ b/lib/programkit/board.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""What the board offers: its ``board_config`` if it has one, and its lazy roles.""" + + +def _optional(name): + """Import a module that may be absent, but not one that's broken.""" + try: + return __import__(name) + except ImportError as e: + if name in str(e): + return None + raise + + +class Board: + """The board this program runs on. + + Importing ``board_config`` starts the board's display, so a program that + makes a `Board` gets one. A board without a display has only + ``board_peripherals``. + + Attributes: + config: the ``board_config`` module, or None. + peripherals: the ``board_peripherals`` module, or None. + display: ``board_config.display_drv``, or None. + """ + + def __init__(self): + self.config = _optional("board_config") + self.peripherals = _optional("board_peripherals") + self.display = getattr(self.config, "display_drv", None) + self._roles = getattr(self.peripherals, "PERIPHERALS", frozenset()) + self._factories = getattr(self.peripherals, "FACTORY_ROLES", frozenset()) + + def has(self, role): + """True when the board lists ``role`` (``"audio_out"``, ``"sdcard"``, ...).""" + return role in self._roles + + def get(self, role): + """The role's object, or its factory for the audio roles; None if absent. + + Each call through ``board_peripherals`` alone makes a new object, so + keep what this returns. + """ + if role not in self._roles: + return None + if self.config is not None: + return getattr(self.config, role) + factory = getattr(self.peripherals, role) + return factory if role in self._factories else factory() + + def eager(self, name): + """A ``board_config`` name such as ``touch_read`` or ``keypad_read``, or None.""" + return getattr(self.config, name, None) diff --git a/lib/programkit/press.py b/lib/programkit/press.py new file mode 100644 index 0000000..0a312f4 --- /dev/null +++ b/lib/programkit/press.py @@ -0,0 +1,63 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""A press of any key, touch or button the board has.""" + +import time + + +class Press: + """Hears the board's touch screen, its keys and its Boot button as one button. + + Call it as often as the program loops. It reads every source each time, + so a board with none of them simply never reports a press. + + Args: + board: a `Board`. + + Attributes: + sources: the names of what it listens to, for a program to report. + """ + + def __init__(self, board): + self._reads = [] + self.sources = [] + touch = board.eager("touch_read") + if touch is not None: + self._reads.append(lambda: bool(touch())) + self.sources.append("touch") + keys = board.eager("keypad_read") + if keys is not None: + self._reads.append(lambda: bool(keys())) + self.sources.append("keys") + button = board.get("boot_button") + if button is not None: + self._reads.append(lambda: button.value() == 0) + self.sources.append("Boot button") + self._down_at = None + + def down(self): + """True while anything is pressed.""" + for read in self._reads: + try: + if read(): + return True + except OSError: + pass # a touch controller that missed one read + return False + + def __call__(self): + """True once for each new press.""" + if self.down(): + if self._down_at is None: + self._down_at = time.ticks_ms() + return True + return False + self._down_at = None + return False + + def held_ms(self): + """How long the current press has lasted, in ms, or None between presses.""" + if self._down_at is None: + return None + return time.ticks_diff(time.ticks_ms(), self._down_at) diff --git a/lib/programkit/rows.py b/lib/programkit/rows.py new file mode 100644 index 0000000..0de2dc1 --- /dev/null +++ b/lib/programkit/rows.py @@ -0,0 +1,149 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Lines of text on a board's display, sent one row at a time. + +A one-bit OLED (SH1106, SSD1305, SSD1306) is drawn with ``framebuf`` straight +into its driver's copy of the panel, and each changed row goes out as one +page: drawing a pixel at a time through ``blit_rect`` costs about 90 ms a +line on a classic ESP32, too long to sit beside audio. That leans on two +driver internals, ``_buffer`` and ``_flush``, at rotation 0. Every other +display gets each row composed in a strip with pygraphics and sent in one +``blit_rect``. +""" + +WHITE = 0xFFFF +BLACK = 0 + + +def clean(text): + """Printable ASCII only: the fonts have nothing else.""" + return "".join(c if " " <= c <= "~" else "?" for c in text) + + +def wrap(text, width): + """Split text at the last space that fits in width, or at width if none does.""" + if len(text) <= width: + return text, "" + cut = text.rfind(" ", 0, width + 1) + if cut <= 0: + return text[:width], text[width:] + return text[:cut], text[cut + 1 :] + + +def _one_bit(display): + buf = getattr(display, "_buffer", None) + if buf is None or not hasattr(display, "_flush"): + return False + if getattr(display, "rotation", 0) not in (0, None): + return False + return len(buf) == display.width * ((display.height + 7) // 8) + + +class Rows: + """Rows of text, each redrawn only when it changes. + + `line` changes a row in memory and `send` puts one changed row on the + panel, so a program that must keep audio fed can draw a row a turn. + + Args: + display: the board's ``display_drv``. + + Attributes: + count: how many rows fit. + cols: characters in a full row. + height: a row's height in pixels. + width: the display's width in pixels. + """ + + def __init__(self, display): + self.display = display + self.width = display.width + self._dirty = [] + if _one_bit(display): + import framebuf + + self._fb = framebuf.FrameBuffer(display._buffer, display.width, display.height, framebuf.MONO_VLSB) + self._mono = True + self.height = 8 + self._scale = 1 + self._fg = 1 + self._fb.fill(0) + display._flush(0, 0, display.width - 1, display.height - 1) + else: + try: + import pygraphics + except ImportError: + raise ImportError("this display needs pygraphics to draw text: mip.install(\"pygraphics\")") + depth = getattr(display, "color_depth", 16) or 16 + fmt = {1: pygraphics.MONO_HMSB, 8: pygraphics.GS8, 16: pygraphics.RGB565, 24: pygraphics.RGB888}[depth] + self._mono = False + self._scale = max(1, display.width // 320) + self.height = 16 * self._scale + self._fg = WHITE if depth == 16 else (1 if depth == 1 else 0xFFFFFF if depth == 24 else 0xFF) + self._depth = depth + self._fmt = fmt + self._pg = pygraphics + self._strips = {} # width: (buffer, FrameBuffer), one per row width in use + display.fill_rect(0, 0, display.width, display.height, BLACK) + self.count = display.height // self.height + self.cols = display.width // (8 * self._scale) + self._text = [None] * self.count + self._widths = [self.width] * self.count + + def line(self, row, text, width=None): + """Put text on a row, replacing what was there; nothing happens if it's unchanged. + + Args: + row: 0 is the top. + text: cut to fit. + width: pixels from the left edge that the text owns; the rest of + the row is left alone, for a program's own drawing. + """ + if not 0 <= row < self.count: + return + width = self.width if width is None else width + text = clean(text)[: width // (8 * self._scale)] + if self._text[row] == text and self._widths[row] == width: + return + self._text[row] = text + self._widths[row] = width + if row not in self._dirty: + self._dirty.append(row) + + def lines(self, first_row, texts): + """Put each of ``texts`` on its own row, starting at ``first_row``.""" + for i, text in enumerate(texts): + self.line(first_row + i, text) + + def send(self): + """Put one changed row on the panel. Returns False when none was waiting.""" + if not self._dirty: + return False + row = self._dirty.pop(0) + text = self._text[row] + width = self._widths[row] + y = row * self.height + if self._mono: + self._fb.fill_rect(0, y, width, 8, 0) + self._fb.text(text, 0, y, 1) + self.display._flush(0, y, width - 1, y + 7) + else: + strip = self._strips.get(width) + if strip is None: + buf = bytearray((width * self._depth + 7) // 8 * self.height) + strip = self._strips[width] = (buf, self._pg.FrameBuffer(buf, width, self.height, self._fmt)) + buf, fb = strip + fb.fill(BLACK) + fb.text16(text, 0, 0, self._fg, self._scale) + self.display.blit_rect(buf, 0, y, width, self.height) + return True + + def send_all(self): + """Put every changed row on the panel.""" + while self.send(): + pass + + def rect(self, x, y, w, h, on=True): + """Fill a rectangle in white, or black when ``on`` is False, straight to the panel.""" + self.display.fill_rect(x, y, w, h, WHITE if on else BLACK) diff --git a/programs/README.md b/programs/README.md index d7a2c80..62b8adc 100644 --- a/programs/README.md +++ b/programs/README.md @@ -6,6 +6,9 @@ runs it from the home screen. | Program | What it does | |---|---| +| [Bluetooth speaker](bt-speaker) | A speaker your phone or computer pairs with, on a board with Bluetooth Classic. | +| [Internet radio](internet-radio) | Plays MP3 streams over Wi-Fi; a press changes station. | +| [MIDI module](midi-module) | A General MIDI sound module: MIDI from a DIN socket plays a SoundFont. | | [Ostinato](ostinato) | A pattern sequencer and song editor, in a browser or a desktop window. | | [Vamp](vamp) | Plays Ostinato's songs on one MicroPython, with the screens elsewhere. | diff --git a/programs/bt-speaker/README.md b/programs/bt-speaker/README.md new file mode 100644 index 0000000..0310834 --- /dev/null +++ b/programs/bt-speaker/README.md @@ -0,0 +1,40 @@ +# Bluetooth speaker + +Bluetooth speaker turns a board into a speaker your phone or computer pairs +with and plays through. With a display, it shows the connection, the title and +artist, and on a display tall enough, the sender's name and a left and right +level meter. Holding a key, the touch screen or the Boot button for a second +drops the sender, so another device can find the speaker. + +Runs on: a MicroPython board with Bluetooth Classic and audio out (`pcm_out` +in its board package): ESP32 boards, and the Pico W and Pico 2 W family. +Tested on the QT Py ESP32 Pico with the Audio BFF and a 128x64 SH1106 OLED, +and the Pimoroni Pico Plus 2 W with a Pico-Audio board and a Pico-OLED-2.23. + +## Put it on a board + +The firmware needs `btclassic`. Install your board's package from +[pydevices](https://github.com/PyDevices/pydevices) first (it brings +`board_config` and `board_peripherals`), then, from the apps repo's root: + +```bash +mpftp put lib/programkit /lib/programkit +mpftp put programs/bt-speaker/bt_speaker.py /bt_speaker.py +``` + +and make `main.py` say `import bt_speaker`. Reset the board, pair with its +name and play something. + +## Settings + +All optional, in `/bt_speaker.json` on the board: + +```json +{"name": "Kitchen Speaker", "phone": true} +``` + +`name` is what senders see; without it, the speaker uses the board's hostname. +With `phone` on, the speaker also shows a phone's battery and calls (the call +itself stays on the phone). A computer may then list the speaker as a headset +as well. The rest of the settings are at the top of +[`bt_speaker.py`](bt_speaker.py). diff --git a/programs/bt-speaker/bt_speaker.py b/programs/bt-speaker/bt_speaker.py new file mode 100644 index 0000000..9be4e9b --- /dev/null +++ b/programs/bt-speaker/bt_speaker.py @@ -0,0 +1,246 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Bluetooth speaker: pair a phone or computer with the board and play. + +The board's ``pcm_out`` plays what the sender streams. With a display, it +shows the connection, the title and artist, and on a display tall enough, the +sender's name and a left and right level meter. With ``"phone": true`` it +also follows a phone's battery and calls; the call itself stays on the phone. +Holding any key, the touch screen or the Boot button for a second drops the +sender, so another device can find the speaker. + +Settings go in ``/bt_speaker.json`` on the board, all optional:: + + {"name": "Kitchen Speaker", "phone": true} + +Put this on a board with Bluetooth Classic and have ``main.py`` import it. +""" + +import json +import math +import time + +import btclassic +from audiodev import AudioFormat +from programkit import Board, Press, Rows + +# Change these to match your setup, or put them in /bt_speaker.json. +NAME = None # the name senders see, or None for the board's hostname +PHONE = False # follow a phone's battery and calls (a computer may then treat the speaker as a headset) +AHEAD_MS = 60 # audio kept queued ahead of the speaker +HOLD_MS = 1000 # holding a key this long drops the sender +METER_MS = 50 # meter refresh +FLOOR_DB = 48 # the meter's left end, in dB below full scale +FALL = 4 # pixels a meter bar falls per refresh + +_CALLS = { + "incoming": "Incoming call", + "dialing": "Calling", + "alerting": "Ringing", + "active": "On a call", + "held": "Call on hold", +} + +try: + with open("/bt_speaker.json") as f: + _settings = json.load(f) + NAME = _settings.get("name", NAME) + PHONE = _settings.get("phone", PHONE) +except OSError: + pass + +# Wi-Fi and Bluetooth share the radio on these boards, and with both on the +# audio arrives in bursts and stutters. +try: + import network + + network.WLAN(network.STA_IF).active(False) +except (ImportError, OSError): + pass + +board = Board() +if not board.has("pcm_out"): + raise SystemExit("bt_speaker: this board has no pcm_out in its board_peripherals") +pcm_out = board.get("pcm_out") +press = Press(board) +rows = Rows(board.display) if board.display is not None else None + +bt = btclassic.Bluetooth() +if NAME: + bt.config(name=NAME) +name = bt.config("name") +receiver = btclassic.AudioSink(bt) +phone = btclassic.HandsFree(bt) if PHONE else None +print("Pair with %s and play something." % name) + +# Where things go: a 128x64 OLED has eight rows, a 128x32 one has four. +if rows is not None and rows.count >= 8: + SENDER, TITLE, ARTIST, METER = 1, 2, 5, 6 +elif rows is not None: + SENDER, TITLE, ARTIST, METER = None, 1, 3, None +BATTERY_W = 24 # the battery icon's width at the right of the top row + + +def show_status(): + battery = phone is not None and phone.connected and phone.battery is not None + width = rows.width - BATTERY_W - 4 if battery else None + if not receiver.connected: + rows.line(0, "Pair with", width) + else: + state = "Playing" if receiver.playing else "Paused" + rows.line(0, "%-8s%3d%%" % (state, receiver.volume * 100 // 127), width) + if SENDER is not None: + rows.line(SENDER, "From " + receiver.peer_name if receiver.connected else "") + + +def show_battery(): + """The phone's battery, 0 to 5 bars, as an icon at the top right.""" + x = rows.width - BATTERY_W + h = rows.height + rows.rect(x, 0, BATTERY_W, h, False) + show_status() + if phone is None or not phone.connected or phone.battery is None: + return + rows.rect(x, 0, BATTERY_W - 3, h, True) + rows.rect(x + 1, 1, BATTERY_W - 5, h - 2, False) + rows.rect(x + BATTERY_W - 3, h // 4, 2, h // 2, True) + step = (BATTERY_W - 7) // 5 + for i in range(max(0, min(5, phone.battery))): + rows.rect(x + 2 + step * i, 2, step - 1, h - 4, True) + + +def call_lines(): + """The call's three lines while there is one, or None.""" + if phone is None: + return None + label = _CALLS.get(phone.call) + if label is None: + return None + if phone.call == "active" and call_started is not None: + seconds = time.ticks_diff(time.ticks_ms(), call_started) // 1000 + label = "Call %d:%02d" % (seconds // 60, seconds % 60) + return label, phone.caller or "", "" + + +def show_middle(): + """The call while there is one, the track while connected, else the name to pair with.""" + lines = call_lines() + if lines is None and not receiver.connected: + lines = (name, "", "") + elif lines is None: + first, rest = wrap(receiver.title, rows.cols) + lines = (first, wrap(rest, rows.cols)[0], receiver.artist) + rows.line(TITLE, lines[0]) + rows.line(TITLE + 1, lines[1]) + rows.line(ARTIST, lines[2]) + + +def wrap(text, width): + """Split text at the last space that fits in width, or at width if none does.""" + if len(text) <= width: + return text, "" + cut = text.rfind(" ", 0, width + 1) + if cut <= 0: + return text[:width], text[width:] + return text[:cut], text[cut + 1 :] + + +def bar_width(peak, full): + if peak <= 0: + return 0 + db = 20 * math.log10(peak / 32767) + return max(0, min(full, int((db + FLOOR_DB) * full / FLOOR_DB))) + + +call_started = None +bars = [0, 0] +if rows is not None: + show_status() + show_middle() + if METER is not None: + meter_x = rows.width // rows.cols + 2 # right of the L and R labels + meter_w = rows.width - meter_x + rows.line(METER, "L", meter_x - 2) + rows.line(METER + 1, "R", meter_x - 2) + rows.send_all() + +buf = bytearray(512 * 4) +speaker = None +rate = 0 +ahead = 0 +last_meter = last_tick = time.ticks_ms() + +while True: + if receiver.sample_rate and receiver.sample_rate != rate: + rate = receiver.sample_rate + if speaker: + speaker.close() + speaker = pcm_out(AudioFormat(rate, 2, 16)) + ahead = rate * 4 * AHEAD_MS // 1000 + + # The speaker's clock sets the pace: keep AHEAD_MS queued, and draw + # while it plays. + if speaker and speaker.queued_size() < ahead: + receiver.readinto(buf) + speaker.write(buf) + continue + + press() + held = press.held_ms() + if held is not None and held >= HOLD_MS and receiver.connected: + print("Held: dropping", receiver.peer_name or "the sender", "so another device can pair") + receiver.disconnect() + + changed = receiver.changes() + if changed & btclassic.TRACK: + print("Playing:", receiver.title, "-", receiver.artist) + if rows is not None: + if changed & (btclassic.CONNECTION | btclassic.PEER | btclassic.PLAYING | btclassic.VOLUME): + show_status() + if changed & (btclassic.CONNECTION | btclassic.TRACK): + show_middle() + + if phone is not None: + changed = phone.changes() + if phone.audio: + phone.disconnect_audio() # no microphone here: the call stays on the phone + if changed & btclassic.CALL: + if phone.call == "active": + if call_started is None: + call_started = time.ticks_ms() + else: + call_started = None + print("Call:", phone.call, phone.caller) + if rows is not None: + if changed & (btclassic.CONNECTION | btclassic.PHONE): + show_battery() + if changed & btclassic.CALL: + show_middle() + + if rows is None: + time.sleep_ms(5) + continue + + now = time.ticks_ms() + if call_started is not None and time.ticks_diff(now, last_tick) >= 1000: + last_tick = now + show_middle() + + # One row a turn, so the audio never waits for the whole screen. Text + # goes before the meter: while music streams the loop gets here only a + # few times a second, and the meter would otherwise take every turn. + if rows.send(): + continue + if METER is not None and time.ticks_diff(now, last_meter) >= METER_MS: + last_meter = now + for i, peak in enumerate(receiver.levels()): + new = max(bar_width(peak, meter_w), bars[i] - FALL) + y = (METER + i) * rows.height + 1 + if new > bars[i]: + rows.rect(meter_x + bars[i], y, new - bars[i], rows.height - 2, True) + elif new < bars[i]: + rows.rect(meter_x + new, y, bars[i] - new, rows.height - 2, False) + bars[i] = new + else: + time.sleep_ms(5) diff --git a/programs/internet-radio/README.md b/programs/internet-radio/README.md new file mode 100644 index 0000000..5814f1c --- /dev/null +++ b/programs/internet-radio/README.md @@ -0,0 +1,39 @@ +# Internet radio + +Internet radio plays MP3 streams over Wi-Fi. A press of a key, the touch screen +or the Boot button moves to the next station, and a display, if the board has +one, shows which station is playing. The board holds a few seconds of the +stream in RAM, so a late packet isn't heard, and a stream that stops sending +is opened again. + +Runs on: a MicroPython board with Wi-Fi and audio out (`audio_out` in its board +package) and `audiomp3` in its firmware. Tested on the QT Py ESP32 Pico with +the Audio BFF and a 128x64 SH1106 OLED, and the Waveshare +ESP32-P4-WIFI6-Touch-LCD-4B. + +## Put it on a board + +Install your board's package from +[pydevices](https://github.com/PyDevices/pydevices) first (it brings +`board_config`, `board_peripherals` and `wifi_manager`), and put your network +in `secrets.py` as `WIFI_SSID` and `WIFI_PASSWORD`. Then, from the apps repo's +root: + +```bash +mpftp put lib/programkit /lib/programkit +mpftp put programs/internet-radio/internet_radio.py /internet_radio.py +``` + +and make `main.py` say `import internet_radio`. + +## Settings + +All optional, in `/internet_radio.json` on the board: + +```json +{"stations": ["http://ice1.somafm.com/groovesalad-128-mp3"], "volume": 70, "hostname": "radio"} +``` + +A station is an `http://` MP3 stream URL. Without the file, three SomaFM +stations play. The buffer and stall settings are at the top of +[`internet_radio.py`](internet_radio.py). diff --git a/programs/internet-radio/internet_radio.py b/programs/internet-radio/internet_radio.py new file mode 100644 index 0000000..ebe6695 --- /dev/null +++ b/programs/internet-radio/internet_radio.py @@ -0,0 +1,214 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Internet radio: play MP3 streams over Wi-Fi, and step through stations. + +Joins Wi-Fi, opens an MP3 stream over HTTP and plays it through the board's +``audio_out``. A press of any key, the touch screen or the Boot button moves +to the next station, and a display, if the board has one, shows which. The +board holds a few seconds of the stream in RAM, so a late packet isn't +heard, and a stream that stops sending is opened again. + +The stations are a list of ``http://`` MP3 stream URLs in +``/internet_radio.json`` on the board, or the defaults below:: + + {"stations": ["http://ice1.somafm.com/groovesalad-128-mp3"], "volume": 70} + +Put this on the board and have ``main.py`` import it. +""" + +import io +import json +import socket +import time + +import audiomp3 +import wifi_manager +from audiodev import AudioFormat +from programkit import Board, Press, Rows + +# Change these to match your setup, or put them in /internet_radio.json. +STATIONS = [ + "http://ice1.somafm.com/groovesalad-128-mp3", + "http://ice1.somafm.com/dronezone-128-mp3", + "http://ice1.somafm.com/secretagent-128-mp3", +] +VOLUME = 85 # 0 is silent, 100 the loudest +HOSTNAME = "radio" # the board's name on the network +STALL_MS = 8000 # a stream that sends nothing this long is opened again +BUFFER = 64 * 1024 # bytes of the stream held in RAM; 64 KB is 4 s at 128 kbit/s +START = BUFFER // 2 # bytes to hold before playing starts +REPORT_MS = 10000 # how often the console hears how full the buffer is + +try: + with open("/internet_radio.json") as f: + _settings = json.load(f) + STATIONS = _settings.get("stations", STATIONS) + VOLUME = _settings.get("volume", VOLUME) + HOSTNAME = _settings.get("hostname", HOSTNAME) +except OSError: + pass + +board = Board() +if not board.has("audio_out"): + raise SystemExit("internet_radio: this board has no audio_out in its board_peripherals") +audio_out = board.get("audio_out") +press = Press(board) +rows = Rows(board.display) if board.display is not None else None + + +def show(row, text): + """A line on the display, if there is one, sent at once.""" + if rows is not None: + rows.line(row, text) + rows.send_all() + + +show(0, "Internet radio") +show(1, "Joining Wi-Fi") +print("internet_radio: joined Wi-Fi at", wifi_manager.connect(hostname=HOSTNAME)) +print("internet_radio: next station on", ", ".join(press.sources) or "nothing: this board has no key, touch or Boot button") + + +def open_stream(url, hops=3): + """A socket at the stream's first byte, following redirects, and the station's name.""" + host, _, path = url.split("//", 1)[1].partition("/") + host, _, port = host.partition(":") + addr = socket.getaddrinfo(host, int(port or 80))[0][-1] + s = socket.socket() + s.settimeout(10) + s.connect(addr) + s.write(b"GET /%s HTTP/1.0\r\nHost: %s\r\nUser-Agent: PyDevices-radio\r\n\r\n" % (path, host)) + status = s.readline() + location = None + title = host + while True: + line = s.readline() + if not line or line == b"\r\n": + break + lower = line.lower() + if lower.startswith(b"location:"): + location = line[9:].strip().decode() + elif lower.startswith(b"icy-name:"): + title = line[9:].strip().decode() or host + code = int(status.split()[1]) + if code in (301, 302, 303, 307, 308) and location and hops: + s.close() + return open_stream(location, hops - 1) + if code != 200: + s.close() + raise OSError("%s answered %d" % (url, code)) + return s, title + + +class StreamBuffer(io.IOBase): + """Seconds of the stream held in RAM, between the socket and the decoder. + + The decoder takes only a few milliseconds of sound ahead of the speaker, + so a TCP segment that comes late would starve it: a blocking socket + holds the decoder up and the speaker plays gaps, and a non-blocking one + has it decode silence into the music. Fed from this, the decoder reads + from memory as it would from a card, and only a stall longer than the + buffer is heard. + """ + + def __init__(self, sock, size): + self.sock = sock + self.buf = bytearray(size) + self.view = memoryview(self.buf) + self.read_at = 0 + self.level = 0 + self.received = 0 + + def fill(self): + """Move what the socket has into the buffer. Call it often.""" + size = len(self.buf) + while self.level < size: + write_at = (self.read_at + self.level) % size + room = min(size - self.level, size - write_at) + try: + n = self.sock.readinto(self.view[write_at : write_at + room]) + except OSError: + n = None # nothing yet on a non-blocking socket + if not n: + if n == 0: + raise OSError("the station closed the stream") + return + self.level += n + self.received += n + + def readinto(self, buf): + self.fill() + size = len(self.buf) + n = min(len(buf), self.level, size - self.read_at) + buf[:n] = self.view[self.read_at : self.read_at + n] + self.read_at = (self.read_at + n) % size + self.level -= n + return n if n else None # None: nothing now, but not the end + + def ioctl(self, request, arg): + if request == 3: # poll: readable while anything is held + return arg & 1 if self.level else 0 + return 0 + + +if board.has("audio_power"): + board.get("audio_power")(True) + +station = 0 +while True: + url = STATIONS[station % len(STATIONS)] + print("internet_radio: tuning to", url) + show(0, "Station %d of %d" % (station % len(STATIONS) + 1, len(STATIONS))) + show(1, url.split("//", 1)[-1]) + show(2, "Tuning") + speaker = stream = None + try: + stream, title = open_stream(url) + show(1, title) + stream.setblocking(False) + held = StreamBuffer(stream, BUFFER) + # Fill the buffer halfway before the decoder reads the first frame. + moved = time.ticks_ms() + while held.level < START: + held.fill() + if time.ticks_diff(time.ticks_ms(), moved) > STALL_MS: + raise OSError("the station sent too little") + time.sleep_ms(5) + decoder = audiomp3.MP3Decoder(held) + print("internet_radio: %s, %d Hz, %d channels" % (title, decoder.sample_rate, decoder.channel_count)) + speaker = audio_out(AudioFormat(decoder.sample_rate, decoder.channel_count, 16)) + speaker.set_volume(VOLUME) # opening the output sets it to 100 + speaker.play(decoder) + show(2, "Playing %.1f kHz" % (decoder.sample_rate / 1000)) + # Keep the decoder fed until a press or a stalled stream. + received = held.received + moved = report = time.ticks_ms() + lowest = held.level + while not press(): + held.fill() + speaker.service() + if held.received != received: + received = held.received + moved = time.ticks_ms() + elif time.ticks_diff(time.ticks_ms(), moved) > STALL_MS: + print("internet_radio: the stream stalled") + show(2, "Stalled: tuning again") + break + lowest = min(lowest, held.level) + if time.ticks_diff(time.ticks_ms(), report) >= REPORT_MS: + report = time.ticks_ms() + print("internet_radio: %d KB held, at least %d KB in the last %d s" % (held.level // 1024, lowest // 1024, REPORT_MS // 1000)) + lowest = held.level + time.sleep_ms(5) + else: + station += 1 + except OSError as e: + print("internet_radio:", e) + show(2, str(e)) + time.sleep(3) + if speaker: + speaker.stop() + speaker.close() + if stream: + stream.close() diff --git a/programs/midi-module/README.md b/programs/midi-module/README.md new file mode 100644 index 0000000..884a014 --- /dev/null +++ b/programs/midi-module/README.md @@ -0,0 +1,75 @@ +# MIDI module + +MIDI module turns a board into a General MIDI sound module: MIDI from a 5-pin +DIN socket plays a SoundFont out of the board's audio out. It uses the sound +module engine in [`lib/soundmodule`](../../lib/soundmodule): sixteen parts, +Bank Select and Program Change pick the sound, and the parts share one pool of +sampler voices. A press of a key, the touch screen or the Boot button silences +every note, and a display, if the board has one, shows what's playing. + +Runs on: a MicroPython board with audio out (`audio_out` in its board package), +a free UART input pin, and `audiodsp` in its firmware. Tested on the Waveshare +ESP32-P4-WIFI6-Touch-LCD-4B and the QT Py ESP32 Pico with the Audio BFF, each +with the self-test below in place of a DIN socket. + +## Wire the MIDI input + +A MIDI input always goes through an optocoupler, so the board never shares a +ground with the instrument. An H11L1 or a PC900V works at 3.3 V: + +| DIN socket | Optocoupler | Board | +|---|---|---| +| pin 4 | through 220 Ω to the LED's anode | | +| pin 5 | the LED's cathode | | +| | a 1N4148 across the LED, cathode to the anode, for reverse voltage | | +| | the output (open collector) | the RX pin, with 1 kΩ to 3V3 | +| | VCC and GND | 3V3 and GND | + +DIN pin 2 (the shield) stays unconnected on an input. Pins 4 and 5 are the +ones either side of the middle pin. + +## Put it on a board + +Install your board's package from +[pydevices](https://github.com/PyDevices/pydevices) first (it brings +`board_config` and `board_peripherals`), and `midiports`: + +```bash +mpftp mip install --index https://PyDevices.github.io/mip midiports +``` + +The engine needs audioinstruments' SoundFont player: +`audioinstruments/__init__`, `_names`, `soundfont`, `soundfont_drums`, and +under `_rack`, `__init__`, `_component`, `_gm`, `_lazy`, `_sf2`, `adapter`, +`soundfont`, `soundfont_drums`, `modules/__init__` and `core/`. Then, from the +apps repo's root: + +```bash +mpftp put lib/programkit /lib/programkit +mpftp put lib/soundmodule /lib/soundmodule +mpftp put programs/midi-module/midi_module.py /midi_module.py +``` + +Put a General MIDI SoundFont (`.sf2`) in `/soundfonts` on the board, or in +`/soundfonts` on its SD card, which is looked at first. The first by name is +used. Then say which pin the optocoupler drives, in `/midi_module.json`: + +```json +{"uart": 1, "rx": 32} +``` + +and make `main.py` say `import midi_module`. The other settings (`tx`, `rate`, +`voices`, `effects`, `master`) are at the top of +[`midi_module.py`](midi_module.py). Sixteen voices with no effects is what a +classic ESP32 holds at 44.1 kHz. + +## Check it without a MIDI cable + +[`midi_module_selftest.py`](midi_module_selftest.py) plays a four-bar loop +(piano, strings, bass and drums, from +[`midi_module_song.py`](midi_module_song.py)) into the module's own RX pin +from a second UART inside the chip, so the bytes cross a real UART at 31250 +baud. Leave the RX pin unwired, put both files on the board, and make +`main.py` say `import midi_module_selftest`. The loop starts 15 seconds after +boot. `"selftest_uart"` in `/midi_module.json` picks the sending UART (2 by +default). diff --git a/programs/midi-module/midi_module.py b/programs/midi-module/midi_module.py new file mode 100644 index 0000000..2d4f311 --- /dev/null +++ b/programs/midi-module/midi_module.py @@ -0,0 +1,167 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""General MIDI sound module: MIDI from a 5-pin DIN socket plays a SoundFont. + +MIDI bytes arriving on a UART play General MIDI through the sound module's +engine (sixteen parts sharing one pool of sampler voices), out of the board's +``audio_out``. The audio pump renders it in C on its own task. A press of any +key, the touch screen or the Boot button silences every note, and a display, +if the board has one, shows what's playing. + +The SoundFont is the first ``.sf2`` by name in ``/sd/soundfonts`` on the +board's SD card, or else in ``/soundfonts`` on its flash. Settings go in +``/midi_module.json``; ``rx``, the pin the optocoupler drives, is the one +you must give:: + + {"uart": 1, "rx": 13} + +Put this on the board with its libraries and have ``main.py`` import it. +""" + +import gc +import json +import os +import sys +import time + +from audiodev import AudioFormat +from programkit import Board, Press, Rows + +# Change these to match your setup, or put them in /midi_module.json. +MIDI_UART = 1 # the UART the DIN input is read on +MIDI_RX = None # the pin the optocoupler's output drives; required +MIDI_TX = -1 # a MIDI out pin, or -1 for none +RATE = 44100 # Hz +VOICES = 16 # the sampler's voices; each costs about 4% of a classic ESP32's core at 44.1 kHz +EFFECTS = False # the reverb alone takes most of a classic ESP32's core +MASTER = 80 # the output's volume, 0 to 100; MIDI's master volume changes it +STATUS_MS = 10000 + +try: + with open("/midi_module.json") as f: + _settings = json.load(f) + MIDI_UART = _settings.get("uart", MIDI_UART) + MIDI_RX = _settings.get("rx", MIDI_RX) + MIDI_TX = _settings.get("tx", MIDI_TX) + RATE = _settings.get("rate", RATE) + VOICES = _settings.get("voices", VOICES) + EFFECTS = _settings.get("effects", EFFECTS) + MASTER = _settings.get("master", MASTER) +except OSError: + pass +if MIDI_RX is None: + raise SystemExit('midi_module: say which pin the MIDI input is wired to: {"rx": 13} in /midi_module.json') + +board = Board() +if not board.has("audio_out"): + raise SystemExit("midi_module: this board has no audio_out in its board_peripherals") +press = Press(board) +rows = Rows(board.display) if board.display is not None else None + + +def show(row, text): + """A line on the display, if there is one, sent at once.""" + if rows is not None: + rows.line(row, text) + rows.send_all() + + +show(0, "MIDI module") + +# Wi-Fi off: on most of these boards it shares a radio or a core with the audio. +try: + import network + + network.WLAN(network.STA_IF).active(False) +except (ImportError, OSError): + pass + +# The card holds the SoundFont, and the libraries when flash has no room. +folders = ["/soundfonts"] +if board.has("sdcard"): + try: + os.listdir("/sd") + except OSError: + try: + os.mount(board.get("sdcard"), "/sd") + except OSError as e: + print("midi_module: no SD card:", e) + folders.insert(0, "/sd/soundfonts") + sys.path.insert(0, "/sd/lib") +from midiports import uart +from soundmodule import engine + +bank = None +for folder in folders: + try: + banks = sorted(n for n in os.listdir(folder) if n.lower().endswith(".sf2")) + except OSError: + continue + if banks: + bank = folder + "/" + banks[0] + break +if bank is None: + show(1, "No SoundFont") + raise SystemExit("midi_module: no SoundFont: put a .sf2 in %s" % " or ".join(folders)) +name = bank.rsplit("/", 1)[1] +show(1, name) + +# The UART, at MIDI's 31250 baud. The engine reads raw MIDI itself, running +# status and all, so it takes the port's bytes as they come. There is no +# MIDI out, so the engine's replies (an Identity Reply) go nowhere. +# Both pins are given where the port allows it: on some ESP32 modules a +# UART's default TX pin is a flash or PSRAM pin, and -1 leaves it alone. +try: + _, port = uart.ports(MIDI_UART, tx=MIDI_TX, rx=MIDI_RX, rxbuf=1024) +except (ValueError, TypeError): + _, port = uart.ports(MIDI_UART, rx=MIDI_RX, rxbuf=1024) + +# The engine, with the SoundFont's presets read before anything plays. +module = engine.Module(RATE, 2, bank=bank, voices=VOICES, effects=EFFECTS, send=None) +started = time.ticks_ms() +if module.prepare() is None: + show(2, "Can't open it") + raise SystemExit("midi_module: can't open %s: %s" % (bank, module.error)) +print("midi_module: %s open in %d ms, %d voices, %d KB free" + % (name, time.ticks_diff(time.ticks_ms(), started), module.voice_count, gc.mem_free() // 1024)) + +# The output. The pump pulls the engine's mix from here on. +if board.has("audio_power"): + board.get("audio_power")(True) +out = board.get("audio_out")(AudioFormat(RATE, 2, 16)) +module.on_master = out.set_volume +module.set_master(MASTER) +out.play(module.output) + +print("midi_module: listening on UART %d, RX pin %s" % (MIDI_UART, MIDI_RX)) +print("midi_module: all notes off on", ", ".join(press.sources) or "nothing: this board has no key, touch or Boot button") +show(2, "MIDI in: UART %d, pin %s" % (MIDI_UART, MIDI_RX)) + +buf = bytearray(256) +received = 0 +next_status = time.ticks_add(time.ticks_ms(), STATUS_MS) +while True: + # Hand whatever arrived to the engine. Its voice writes are queued and + # reach the pump together, so a chord waits for one block, not three. + n = port.readinto(buf) + if n: + with engine.deferred(module): + module.feed(buf, n) + received += n + module.fill() + + if press(): + module.panic() + print("midi_module: all notes off") + show(3, "All notes off") + + # Now and then, what's playing. + if time.ticks_diff(time.ticks_ms(), next_status) >= 0: + next_status = time.ticks_add(next_status, STATUS_MS) + pool = module.voice_pool + sounding = pool.sounding() if pool else 0 + parts = ["%d:%s" % (p.channel + 1, module.name(p.channel)) for p in module.parts if p.inst is not None] + print("midi_module: %d bytes, %d voices sounding, %s" % (received, sounding, ", ".join(parts))) + show(3, "%d voices, %d bytes" % (sounding, received)) + time.sleep_ms(2) diff --git a/programs/midi-module/midi_module_selftest.py b/programs/midi-module/midi_module_selftest.py new file mode 100644 index 0000000..19f1f55 --- /dev/null +++ b/programs/midi-module/midi_module_selftest.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Check the MIDI module with no MIDI cable: the board sends itself a song. + +A second UART drives the module's RX pin from inside the chip (an ESP32's +pins can carry one UART's TX and another's RX at once), and a timer sends a +short General MIDI piece through it at the right times: a piano, strings, +bass and drums. So the bytes cross a real UART at 31250 baud into the same +loop a DIN socket feeds. Have ``main.py`` import this instead of +``midi_module``. + +It reads ``rx`` from ``/midi_module.json`` as the module does, and +``selftest_uart`` there names the sending UART (2 unless you say). +""" + +import json + +import machine +import midi_module_song as song + +STEP_MS = 125 # one sixteenth at 120 BPM +START_MS = 15000 # after the SoundFont has opened + +with open("/midi_module.json") as f: + _settings = json.load(f) +RX_PIN = _settings["rx"] +TX_UART = _settings.get("selftest_uart", 2) + +state = {"tx": None, "step": -START_MS // STEP_MS, "sent": 0} + + +def tick(timer): + step = state["step"] + state["step"] = step + 1 + if step < 0: + return + if state["tx"] is None: + # Set up after the module's UART, so its RX pin keeps its input. + # rx=-1 leaves this UART's default RX pin alone: on some ESP32 + # modules that's a flash pin. + try: + state["tx"] = machine.UART(TX_UART, baudrate=31250, tx=RX_PIN, rx=-1) + except (ValueError, TypeError): + state["tx"] = machine.UART(TX_UART, baudrate=31250, tx=RX_PIN) + state["tx"].write(song.SETUP) + data = song.step(step) + if data: + state["tx"].write(data) + state["sent"] += len(data) + if step % 64 == 0: + print("selftest: bar %d, %d bytes sent" % (step // 16 + 1, state["sent"])) + + +machine.Timer(0).init(period=STEP_MS, callback=tick) +import midi_module # noqa: F401 (runs the module, forever) diff --git a/programs/midi-module/midi_module_song.py b/programs/midi-module/midi_module_song.py new file mode 100644 index 0000000..743716d --- /dev/null +++ b/programs/midi-module/midi_module_song.py @@ -0,0 +1,55 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""A short General MIDI loop as raw bytes, for the MIDI module's self-test. + +Four bars (C, A minor, F, G) on a piano, strings, a fingered bass and the +drum channel, one sixteenth note per step. step(n) returns what to send at +step n, note-offs first. +""" + +PIANO, STRINGS, BASS, DRUMS = 0, 1, 2, 9 # channels 1, 2, 3 and 10 +CHORDS = ((60, 64, 67), (57, 60, 64), (53, 57, 60), (55, 59, 62)) +KICK, SNARE, HAT = 36, 38, 42 + +# Program changes: Acoustic Grand Piano, String Ensemble 1, Electric Bass +# (finger); some reverb on each. +SETUP = bytes((0xC0 | PIANO, 0, 0xC0 | STRINGS, 48, 0xC0 | BASS, 33, + 0xB0 | PIANO, 91, 50, 0xB0 | STRINGS, 91, 70, 0xB0 | STRINGS, 7, 80)) + +_held = [] # (channel, note) sounding now + + +def _on(out, channel, note, velocity): + out.extend((0x90 | channel, note, velocity)) + _held.append((channel, note)) + + +def step(n): + """The bytes for sixteenth ``n`` of the loop.""" + out = bytearray() + bar, pos = divmod(n % 64, 16) + chord = CHORDS[bar] + # Let go of the notes whose time is up: the piano, bass and drums every + # eighth, the strings at the bar. + keep = [] + for channel, note in _held: + if channel in (PIANO, BASS, DRUMS) and pos % 2 == 0 or pos == 0: + out.extend((0x80 | channel, note, 0)) + else: + keep.append((channel, note)) + _held[:] = keep + if pos == 0: + for note in chord: + _on(out, STRINGS, note, 70) + if pos % 4 == 0: + for note in chord: + _on(out, PIANO, note + 12, 90) + if pos % 2 == 0: + _on(out, BASS, chord[0] - 24, 100) + _on(out, DRUMS, HAT, 60) + if pos in (0, 6, 8): + _on(out, DRUMS, KICK, 110) + if pos in (4, 12): + _on(out, DRUMS, SNARE, 100) + return bytes(out) diff --git a/pyproject.toml b/pyproject.toml index a46c5a2..1b27674 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "omnipython-applibs" dynamic = ["version"] -description = "The libraries apps share: appkit, controlkit, deskkit, sayso, songmodel, soundmodule, spectrum and touchkit" +description = "The libraries apps share: appkit, controlkit, deskkit, programkit, sayso, songmodel, soundmodule, spectrum and touchkit" readme = "README.md" authors = [{ name = "Brad Barnett" }] license = { text = "MIT" }