diff --git a/docs/screenshots/README.md b/docs/screenshots/README.md index f2dcd333..cece6b47 100644 --- a/docs/screenshots/README.md +++ b/docs/screenshots/README.md @@ -5,7 +5,8 @@ | ![fonts.png](fonts.png) | ![framebuf_simpletest.png](framebuf_simpletest.png) | ![noto_fonts.png](noto_fonts.png) | | ![paint.png](paint.png) | ![proverbs.png](proverbs.png) | ![testris.png](testris.png) | | ![tiny_toasters.gif](tiny_toasters.gif) | ![piano.png](piano.png) | ![google_photos.png](google_photos.png) | -| ![google_photos_view.png](google_photos_view.png) | ![google_photos_connect.png](google_photos_connect.png) | | +| ![google_photos_view.png](google_photos_view.png) | ![google_photos_connect.png](google_photos_connect.png) | ![spectrum_800x480.gif](spectrum_800x480.gif) | +| ![spectrum_800x480.png](spectrum_800x480.png) | ![spectrum_800x480_segmented.png](spectrum_800x480_segmented.png) | ![spectrum_320x170.png](spectrum_320x170.png) | On Android (the PyDevices Runner) and as an installed PWA: diff --git a/docs/screenshots/spectrum_320x170.png b/docs/screenshots/spectrum_320x170.png new file mode 100644 index 00000000..168d1922 Binary files /dev/null and b/docs/screenshots/spectrum_320x170.png differ diff --git a/docs/screenshots/spectrum_800x480.gif b/docs/screenshots/spectrum_800x480.gif new file mode 100644 index 00000000..06302a50 Binary files /dev/null and b/docs/screenshots/spectrum_800x480.gif differ diff --git a/docs/screenshots/spectrum_800x480.png b/docs/screenshots/spectrum_800x480.png new file mode 100644 index 00000000..873d8e4f Binary files /dev/null and b/docs/screenshots/spectrum_800x480.png differ diff --git a/docs/screenshots/spectrum_800x480_segmented.png b/docs/screenshots/spectrum_800x480_segmented.png new file mode 100644 index 00000000..f8485895 Binary files /dev/null and b/docs/screenshots/spectrum_800x480_segmented.png differ diff --git a/docs/screenshots/spectrum_p4_music.png b/docs/screenshots/spectrum_p4_music.png new file mode 100644 index 00000000..b5f0aa98 Binary files /dev/null and b/docs/screenshots/spectrum_p4_music.png differ diff --git a/lib/examples/spectrum/README.md b/lib/examples/spectrum/README.md new file mode 100644 index 00000000..27b39d85 --- /dev/null +++ b/lib/examples/spectrum/README.md @@ -0,0 +1,92 @@ +# spectrum + +A spectrum analyzer that is built to look good rather than to measure +anything. It draws log-spaced bars from 20 Hz to 20 kHz in a cool gradient, +with peak caps that hang and then fall, a faint reflection under the baseline, +and frequency labels along the bottom. On a board running usbif's USB sound +card it shows the audio the PC is playing. The levels are computed in C, in +the sound card's pump. Everywhere else it plays fake music. + +![800x480](../../../docs/screenshots/spectrum_800x480.gif) + +| 800x480, segmented | 320x170 (T-Embed) | +|:--:|:--:| +| ![segmented](../../../docs/screenshots/spectrum_800x480_segmented.png) | ![320x170](../../../docs/screenshots/spectrum_320x170.png) | + +## Run it + +From `lib/`: + +```bash +micropython examples/spectrum/spectrum.py +SPECTRUM_STYLE=segmented micropython examples/spectrum/spectrum.py +PYDEVICES_WIDTH=320 PYDEVICES_HEIGHT=170 PYDEVICES_SCALE=3 micropython examples/spectrum/spectrum.py +``` + +On desktop the window opens at 800x480. On a board, the example takes its size +from the panel. It prints the frame rate and the cost of each part of a frame +every five seconds. + +## What's where + +`spectrum_view.py` does the drawing and knows nothing about audio. Hand +`SpectrumView.update()` one level per band, from 0 to 1, and it handles the +ballistics itself: bars rise fast and fall at a steady rate, and the peak caps +hold for about half a second before dropping with gravity. The palette, band +count, timings and colours are constants at the top of the file. + +`fake_music.py` is the stand-in source: a four-bar loop at 120 BPM with kick, +snare, hats, bass, a pad and a lead, plus a dead stop and a breakdown so you +can watch the fall-off. Its docstring has the arrangement. The loop is seeded, +so every run is the same. + +`capture.py` renders headless under MicroPython and reports timings, and +`make_captures.py` (CPython with Pillow) turns those renders into the images +in `docs/screenshots/`. + +## How it stays cheap + +Every bar is a single `blit` from a gradient column that was built once, so a +tall bar costs the same as a short one. The segment gaps and the reflection's +scanlines use a colour key, so the grid shows through them. The static art +(background, grid, labels) is painted once into a second framebuffer. Each +frame copies back only the row strip the bars could have touched, and that +strip is all that goes to the panel. + +On desktop MicroPython at 800x480 with 48 bands, a frame costs about 0.2 ms to +generate the data, 0.7–1.5 ms to draw and 0.9 ms to blit to the SDL window. +The timer caps it at 50 fps. + +## On the ESP32-P4 panel, beside the sound card + +![The P4 panel's framebuffer, music playing](../../../docs/screenshots/spectrum_p4_music.png) + +This needs a firmware with usbif's meter ([usbif#50](https://github.com/PyDevices/usbif/pull/50)), +which adds `uac_pump_meter()` and `uac_pump_levels()`. Import `spectrum` before +running `soundcard.py`, and the meter draws from a timer while the sound card +runs. `pump_levels.py` is the real source. + +What the measurements found (2026-09-25): + +- **Drawing costs no USB packets, once one setting changes.** A cache + writeback (`esp_cache_msync`) runs with interrupts off, and syncing the + panel held the USB interrupt off long enough to drop delivery to + 98.7-99.4 %, under the 99.5 % gate. With the writeback sliced + (`CONFIG_ESP_MM_CACHE_MSYNC_C2M_CHUNKED_OPS`), meter on and meter off match + to within 0.03 %: 99.75-99.88 % against 99.77-99.86 %. The P4 panel's board + definition carries that setting since micropython-pydevices#23. +- **The sound card's own baseline** settles near 99.8 % a window or two into a + 48 kHz stream, with or without the meter. A 24 kHz wire holds 100 %. +- **The cost in the pump** is 1.05 % of a 360 MHz core for the feed plus + 3.2-3.8 % for the analysis: two FFTs, 60 a second, with the longest single + analysis about 1 ms. +- **The frame rate** is 50 fps in silence, 27-29 with every bar moving and 19 + with loud music. Each bar is two prebuilt columns, only the rows that moved + are copied, and one band is presented a frame (`SpectrumView.render_columns`). +- **The low end, from a full-range track:** the 22, 26 and 30 Hz bars sit near + empty. From 35 Hz up, every bar moves. The top four bars (10-20 kHz) also + hover near the floor. + +[`tools/spectrum/meter_gate.py`](../../../tools/spectrum/meter_gate.py) (board) and +[`tools/spectrum/play_src.py`](../../../tools/spectrum/play_src.py) (Windows Python) +reproduce the numbers. diff --git a/lib/examples/spectrum/__init__.py b/lib/examples/spectrum/__init__.py new file mode 100644 index 00000000..b7f4a217 --- /dev/null +++ b/lib/examples/spectrum/__init__.py @@ -0,0 +1,7 @@ +import sys + +_wd = __file__.replace("\\", "/") +_wd = _wd.rsplit("/", 1)[0] if "/" in _wd else "." +if _wd not in sys.path: + sys.path.insert(0, _wd) +from . import spectrum # noqa: F401 -- gallery/kit entry: import spectrum diff --git a/lib/examples/spectrum/capture.py b/lib/examples/spectrum/capture.py new file mode 100644 index 00000000..c3c1da30 --- /dev/null +++ b/lib/examples/spectrum/capture.py @@ -0,0 +1,79 @@ +""" +capture.py -- render the analyzer headless, for screenshots and timing. + +Runs on MicroPython (or CPython) with no display. Renders ``frames`` frames at +a fixed ``fps`` of fake-music time, prints what each part of a frame costs, +and, given an output path, appends every frame from ``start`` on to it as raw +RGB565 (little-endian, width x height per frame):: + + micropython capture.py WIDTH HEIGHT FRAMES FPS [STYLE] [OUT.raw] [START] + +``make_captures.py`` drives this and turns the raw frames into PNGs and a GIF. +""" + +import sys + +try: + from time import ticks_diff, ticks_us +except ImportError: + from time import perf_counter + + def ticks_us(): + return int(perf_counter() * 1e6) + + def ticks_diff(a, b): + return a - b + + +_here = __file__.replace("\\", "/").rsplit("/", 1)[0] if "/" in __file__ else "." +if _here not in sys.path: + sys.path.insert(0, _here) + +from fake_music import FakeMusic # noqa: E402 +from spectrum_view import SpectrumView # noqa: E402 + +args = sys.argv[1:] +width, height, frames, fps = int(args[0]), int(args[1]), int(args[2]), int(args[3]) +style = args[4] if len(args) > 4 else "smooth" +out = args[5] if len(args) > 5 else None +start = int(args[6]) if len(args) > 6 else 0 + +t0 = ticks_us() +view = SpectrumView(width, height, style=style) +setup_us = ticks_diff(ticks_us(), t0) +music = FakeMusic(view.bands) +dt = 1 / fps +f = open(out, "wb") if out else None +src_us = draw_us = rows = 0 +worst = 0 +for k in range(frames): + t = k * dt + a = ticks_us() + levels = music.levels(t) + b = ticks_us() + view.update(levels, dt) + y, h = view.render() + c = ticks_us() + src_us += ticks_diff(b, a) + draw_us += ticks_diff(c, b) + worst = max(worst, ticks_diff(c, b)) + rows += h + if f and k >= start: + f.write(view.fb.buffer if hasattr(view.fb, "buffer") else view._buf) +if f: + f.close() +print( + "{}x{} {} bands, style {}: setup {:.1f} ms; per frame: fake data {:.2f} ms, " + "draw {:.2f} ms (worst {:.2f}), dirty rows {:.0f} of {}".format( + width, + height, + view.bands, + style, + setup_us / 1000, + src_us / frames / 1000, + draw_us / frames / 1000, + worst / 1000, + rows / frames, + height, + ) +) diff --git a/lib/examples/spectrum/fake_music.py b/lib/examples/spectrum/fake_music.py new file mode 100644 index 00000000..11b9fa11 --- /dev/null +++ b/lib/examples/spectrum/fake_music.py @@ -0,0 +1,197 @@ +""" +fake_music.py -- band levels that move like a band playing, with no audio. + +A stand-in for the real source (band levels published from C where the audio +passes). ``FakeMusic(n).levels(t)`` returns ``n`` levels in 0..1 for time ``t`` +seconds, over log-spaced bands from 20 Hz to 20 kHz, mapped from -54..0 dB. + +What it plays, at 120 BPM in an 8-second loop (four bars): + +* a kick on every beat, a thump around 55 Hz that dies in a tenth of a second; +* a snare on beats 2 and 4, a broad crack centred near 1.5 kHz with some body; +* closed hats on the eighths and an open hat before each bar, fizz above 5 kHz + with a different sparkle on every hit; +* a bass line and a three-note pad following the chords (Am, F, C, G); +* a lead line hopping round a pentatonic scale, with vibrato; +* a pink-ish tilt over the lot, falling about 3 dB an octave above 150 Hz. + +Bar two ends in a dead stop (3.5 s to 4.25 s) so the bars and the peak caps +can be seen falling, then a breakdown of pad, lead and hats until 5.5 s, when +the drums and bass come back in. + +Deterministic: every loop replays the same numbers from a seeded xorshift, so +a capture is repeatable and loops cleanly. +""" + +import math + +from spectrum_view import band_centres + +BEAT = 0.5 # seconds, 120 BPM +LOOP = 8.0 +FLOOR_DB = -54.0 + +# Chord per bar: bass root, pad notes (Hz). +CHORDS = ( + (55.00, (220.0, 261.6, 329.6)), # Am + (43.65, (174.6, 220.0, 261.6)), # F + (65.41, (196.0, 261.6, 329.6)), # C + (49.00, (196.0, 246.9, 293.7)), # G +) +PENTA = (440.0, 523.3, 587.3, 659.3, 784.0, 880.0, 1046.5, 1174.7) + + +class XorShift: + def __init__(self, seed): + self.s = seed & 0xFFFFFFFF or 1 + + def next(self): + s = self.s + s ^= (s << 13) & 0xFFFFFFFF + s ^= s >> 17 + s ^= (s << 5) & 0xFFFFFFFF + self.s = s + return s + + def uniform(self): + return self.next() / 4294967296.0 + + +def _gauss(x): + return math.exp(-x * x) + + +class FakeMusic: + def __init__(self, bands, seed=2026): + self.n = bands + self.seed = seed + fc = band_centres(bands) + self.oct = [math.log(f / 20.0) / math.log(2) for f in fc] # octaves above 20 Hz + self.bw_oct = math.log(1000) / math.log(2) / bands + # Width of a partial's smear across bands, in octaves. + self.sigma = max(0.16, self.bw_oct * 0.6) + self.tilt = [10 ** (-0.3 * max(0.0, math.log(f / 150.0) / math.log(2))) for f in fc] + # Fixed spectral shapes (linear power per band) of the noisy sounds. + self.snare_w = [ + 0.5 * _gauss((o - self._o(1500)) / 1.6) + 0.35 * _gauss((o - self._o(200)) / 0.4) + for o in self.oct + ] + self.hat_w = [ + (1 / (1 + math.exp(-(o - self._o(6000)) * 3.0))) * _gauss((o - self._o(11000)) / 1.4) + for o in self.oct + ] + self.room_w = [ + _gauss((o - self._o(400)) / 3.0) for o in self.oct + ] # reverb wash under everything + self.kick_click_w = [_gauss((o - self._o(3500)) / 0.8) for o in self.oct] + self._p = [0.0] * bands + self._rng = XorShift(seed) + self._loop = -1 + self._sparkle = [1.0] * bands + self._hat_n = -1 + + @staticmethod + def _o(hz): + return math.log(hz / 20.0) / math.log(2) + + def _partial(self, hz, power): + """Add a pitched partial, smeared over the bands within reach of it.""" + if hz <= 20 or hz >= 20000 or power <= 0: + return + o = self._o(hz) + sig = self.sigma + centre = o / self.bw_oct - 0.5 + reach = int(2.5 * sig / self.bw_oct) + 1 + lo = max(0, int(centre) - reach) + hi = min(self.n, int(centre) + reach + 2) + p, oc = self._p, self.oct + for i in range(lo, hi): + d = (oc[i] - o) / sig + p[i] += power * math.exp(-d * d) + + def levels(self, t): + n = self.n + p = self._p + for i in range(n): + p[i] = 0.0 + loop = int(t // LOOP) + tl = t - loop * LOOP + if loop != self._loop: + self._loop = loop + self._rng = XorShift(self.seed) + self._hat_n = -1 + rng = self._rng + + stop = 3.5 <= tl < 4.25 + breakdown = 4.25 <= tl < 5.5 + drums = not stop and not breakdown + bar = int(tl // (4 * BEAT)) % 4 + beat_t = tl % BEAT + beat_i = int(tl // BEAT) % 4 + root, pad = CHORDS[bar] + out = [FLOOR_DB] * n + if stop: + return [0.0] * n + + # Room wash: quiet, keeps the floor alive while the band plays. + room = 0.000006 + 0.000006 * rng.uniform() + for i in range(n): + p[i] += room * self.room_w[i] + + if drums: + k = math.exp(-beat_t / 0.09) + kick_hz = 50 + 70 * math.exp(-beat_t / 0.02) + self._partial(kick_hz, 1.0 * k) + self._partial(kick_hz * 2, 0.25 * k) + click = 0.02 * math.exp(-beat_t / 0.01) + for i in range(n): + p[i] += click * self.kick_click_w[i] + if beat_i in (1, 3): + s = 0.18 * math.exp(-beat_t / 0.11) + for i in range(n): + p[i] += s * self.snare_w[i] + # Bass: eighth notes pumping on the chord root, octave on the offbeat. + e_t = tl % (BEAT / 2) + off = int(tl // (BEAT / 2)) & 1 + b = 0.35 * (0.25 + 0.75 * math.exp(-e_t / 0.18)) + f = root * (2 if off else 1) + for h in range(1, 5): + self._partial(f * h, b / (h * h)) + + # Hats: closed on the eighths, open on the last offbeat of each bar. + e_n = int(tl // (BEAT / 2)) + e_t = tl % (BEAT / 2) + if e_n != self._hat_n: + self._hat_n = e_n + for i in range(n): + self._sparkle[i] = 0.35 + 1.3 * rng.uniform() + open_hat = (e_n % 8) == 7 + acc = 1.0 if (e_n & 1) else 0.55 + hat = 0.5 * acc * math.exp(-e_t / (0.2 if open_hat else 0.035)) + sp = self._sparkle + for i in range(n): + p[i] += hat * self.hat_w[i] * sp[i] + + # Pad: swells in on each bar. + bar_t = tl % (4 * BEAT) + swell = 0.02 * (1 - math.exp(-bar_t / 0.4)) * (1.4 if breakdown else 1.0) + for note in pad: + for h in range(1, 4): + self._partial(note * h, swell / (h * h)) + + # Lead: a new pentatonic note every sixteenth-ish, with vibrato. + step = int(tl / 0.375) + note = PENTA[(step * 5 + bar * 3 + (step >> 2)) % len(PENTA)] + n_t = tl - step * 0.375 + vib = 1 + 0.012 * math.sin(2 * math.pi * 5.5 * tl) + lead = 0.05 * (0.35 + 0.65 * math.exp(-n_t / 0.25)) + for h in range(1, 6): + self._partial(note * vib * h, lead / h) + + # To 0..1 over -60..0 dB, with the tilt applied. + tilt = self.tilt + for i in range(n): + v = p[i] * tilt[i] + if v > 1e-6: + out[i] = 10 * math.log10(v) + return [(d - FLOOR_DB) / -FLOOR_DB for d in out] diff --git a/lib/examples/spectrum/loopback_levels.py b/lib/examples/spectrum/loopback_levels.py new file mode 100644 index 00000000..bc14c20d --- /dev/null +++ b/lib/examples/spectrum/loopback_levels.py @@ -0,0 +1,110 @@ +""" +loopback_levels.py -- band levels from what the computer is playing. + +The desktop counterpart of ``pump_levels.PumpLevels``, behind the same +interface as ``fake_music.FakeMusic``: ``LoopbackLevels(n).levels(t)`` returns +``n`` levels in 0..1. It records the default speaker's output through the +operating system's loopback (WASAPI loopback on Windows, the monitor source +on Linux) and analyses it on a background thread with numpy, so the drawing +timer only reads the latest levels. + +It needs CPython with ``numpy`` and the ``soundcard`` package +(``pip install soundcard``). MicroPython has neither, so there the meter keeps +its other sources. The dB-to-bar mapping is the one ``pump_levels`` uses, so +the desktop and the P4 look alike for the same music. +""" + +import math +import threading +import time + +# The same dB-to-bar mapping as pump_levels (which needs _usbif, so it can't +# be imported on a desktop). +FLOOR_DB = -66.0 +TOP_DB = -6.0 +TILT_DB = 3.0 # per octave +TILT_FROM_HZ = 1000.0 +from spectrum_view import HIGH_HZ, LOW_HZ + +RATE = 48000 +BLOCK = 2048 # samples per analysis: about 43 ms, ~23 Hz resolution +STALE_S = 0.15 + + +def available(): + try: + import numpy # noqa: F401 + import soundcard # noqa: F401 + except ImportError: + return False + return True + + +class LoopbackLevels: + def __init__(self, bands, low_hz=LOW_HZ, high_hz=HIGH_HZ): + import numpy as np + import soundcard as sc + + self.bands = bands + self._np = np + self._out = [0.0] * bands + self._latest = None + self._fresh = 0.0 + self.peak_db = self.rms_db = -100.0 + span = high_hz / low_hz + edges = [low_hz * span ** (i / bands) for i in range(bands + 1)] + freqs = np.fft.rfftfreq(BLOCK, 1.0 / RATE) + # Each band averages the FFT bins inside it; a band narrower than one + # bin (the bottom octaves) takes the nearest bin. + self._bins = [] + for lo, hi in zip(edges[:-1], edges[1:]): + idx = np.nonzero((freqs >= lo) & (freqs < hi))[0] + if len(idx) == 0: + idx = np.array([int(np.argmin(abs(freqs - math.sqrt(lo * hi))))]) + self._bins.append(idx) + centres = [math.sqrt(lo * hi) for lo, hi in zip(edges[:-1], edges[1:])] + self._tilt = np.array( + [TILT_DB * math.log2(c / TILT_FROM_HZ) if c > TILT_FROM_HZ else 0.0 for c in centres] + ) + self._window = np.hanning(BLOCK) + # Full-scale sine through this window and FFT reads 0 dB. + self._ref = self._window.sum() / 2 + speaker = sc.default_speaker() + self.source = speaker.name + self._mic = sc.get_microphone(id=str(speaker.name), include_loopback=True) + self._run = True + self._thread = threading.Thread(target=self._loop, daemon=True) + self._thread.start() + + def _loop(self): + np = self._np + with self._mic.recorder(samplerate=RATE, channels=2, blocksize=BLOCK) as rec: + while self._run: + data = rec.record(numframes=BLOCK) + mono = data.mean(axis=1) if data.ndim > 1 else data + if len(mono) < BLOCK: + continue + mono = mono[-BLOCK:] + mag = np.abs(np.fft.rfft(mono * self._window)) / self._ref + power = mag * mag + band = np.array([power[i].mean() for i in self._bins]) + db = 10.0 * np.log10(band + 1e-12) + self._tilt + lvl = np.clip((db - FLOOR_DB) / (TOP_DB - FLOOR_DB), 0.0, 1.0) + peak = float(np.max(np.abs(mono))) + rms = float(np.sqrt(np.mean(mono * mono))) + self._latest = lvl.tolist() + self.peak_db = 20 * math.log10(peak + 1e-9) + self.rms_db = 20 * math.log10(rms + 1e-9) + self._fresh = time.monotonic() + + def levels(self, t=None): + latest = self._latest + if latest is not None and time.monotonic() - self._fresh < STALE_S: + self._out[:] = latest + else: + self._out[:] = [0.0] * self.bands + self.peak_db = self.rms_db = -100.0 + return self._out + + def close(self): + self._run = False diff --git a/lib/examples/spectrum/make_captures.py b/lib/examples/spectrum/make_captures.py new file mode 100644 index 00000000..1bce1207 --- /dev/null +++ b/lib/examples/spectrum/make_captures.py @@ -0,0 +1,87 @@ +""" +make_captures.py -- regenerate the analyzer's screenshots (CPython + Pillow). + +Runs ``capture.py`` under MicroPython to render raw RGB565 frames, then writes +PNG stills and an animated GIF into ``docs/screenshots/``:: + + python make_captures.py [--micropython PATH] [--out DIR] + +Needs Pillow; the PyDevices toolbox Python has it. +""" + +import argparse +import os +import subprocess +import tempfile + +from PIL import Image + +HERE = os.path.dirname(os.path.abspath(__file__)) +REPO = os.path.abspath(os.path.join(HERE, "..", "..", "..")) +FPS = 30 +LOOP_FRAMES = 8 * FPS # fake_music loops every 8 s + + +def render(mp, w, h, frames, style, start): + fd, raw = tempfile.mkstemp(suffix=".raw") + os.close(fd) + try: + out = subprocess.run( + [mp, "capture.py", str(w), str(h), str(frames), str(FPS), style, raw, str(start)], + cwd=HERE, + check=True, + capture_output=True, + text=True, + ) + print(out.stdout.strip()) + with open(raw, "rb") as f: + data = f.read() + finally: + os.remove(raw) + size = w * h * 2 + return [to_image(data[i : i + size], w, h) for i in range(0, len(data), size)] + + +def to_image(buf, w, h): + rgb = bytearray(w * h * 3) + for i in range(w * h): + c = buf[2 * i] | (buf[2 * i + 1] << 8) + r, g, b = (c >> 11) & 0x1F, (c >> 5) & 0x3F, c & 0x1F + rgb[3 * i] = (r << 3) | (r >> 2) + rgb[3 * i + 1] = (g << 2) | (g >> 4) + rgb[3 * i + 2] = (b << 3) | (b >> 2) + return Image.frombytes("RGB", (w, h), bytes(rgb)) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--micropython", default=os.path.expanduser("~/gh/pydevices/bin/micropython")) + ap.add_argument("--out", default=os.path.join(REPO, "docs", "screenshots")) + ap.add_argument("--still-at", type=float, default=2.26, help="seconds into the loop") + args = ap.parse_args() + mp, out = args.micropython, args.out + still = LOOP_FRAMES + int(args.still_at * FPS) + + # Stills: warm up one loop so the peak caps are in play, keep one frame. + for w, h, style, name in ( + (800, 480, "smooth", "spectrum_800x480.png"), + (320, 170, "smooth", "spectrum_320x170.png"), + (800, 480, "segmented", "spectrum_800x480_segmented.png"), + ): + (img,) = render(mp, w, h, still + 1, style, still) + img.save(os.path.join(out, name)) + print("wrote", name) + + # The loop: second pass through the 8 s loop, so it starts warm and wraps. + frames = render(mp, 800, 480, 2 * LOOP_FRAMES, "smooth", LOOP_FRAMES) + pal = frames[len(frames) // 3].quantize(colors=255, method=Image.Quantize.MEDIANCUT) + q = [f.quantize(palette=pal, dither=Image.Dither.NONE) for f in frames] + path = os.path.join(out, "spectrum_800x480.gif") + q[0].save( + path, save_all=True, append_images=q[1:], duration=1000 // FPS, loop=0, optimize=False + ) + print("wrote spectrum_800x480.gif,", len(q), "frames") + + +if __name__ == "__main__": + main() diff --git a/lib/examples/spectrum/pump_levels.py b/lib/examples/spectrum/pump_levels.py new file mode 100644 index 00000000..33b8bf57 --- /dev/null +++ b/lib/examples/spectrum/pump_levels.py @@ -0,0 +1,99 @@ +""" +pump_levels.py -- band levels from the sound card's C pump. + +The real source behind the same interface as ``fake_music.FakeMusic``: +``PumpLevels(n).levels(t)`` returns ``n`` levels in 0..1. The work happens in +C, in usbif's UAC pump task, where the audio from the PC already passes. This +file only asks for the bands once and reads the latest levels each frame. + +It needs a firmware whose ``_usbif`` has ``uac_pump_levels`` (usbif#50) and something else running the sound card's pump, +such as ``soundcard.py``. With no audio flowing the levels stop advancing, and +after ``STALE_MS`` this returns zeros so the bars fall. + +The mapping from dB to bar height is here, not in C, because it's a matter of +taste: ``FLOOR_DB`` is an empty bar, ``TOP_DB`` a full one, and ``TILT_DB`` +lifts the top octaves a little per octave above ``TILT_FROM_HZ``, since music +has far less energy up there than a bar chart would like. +""" + +from time import ticks_diff, ticks_ms + +import _usbif + +from spectrum_view import HIGH_HZ, LOW_HZ, band_centres + +FLOOR_DB = -66.0 +TOP_DB = -6.0 +TILT_DB = 3.0 # per octave +TILT_FROM_HZ = 1000.0 +STALE_MS = 150 + + +def available(): + return hasattr(_usbif, "uac_pump_levels") + + +class PumpLevels: + def __init__(self, bands, low_hz=LOW_HZ, high_hz=HIGH_HZ): + self.bands = bands + _usbif.uac_pump_meter(bands, low_hz, high_hz) + self._buf = bytearray(bands) + self._out = [0.0] * bands + self._seq = -1 + self._fresh = ticks_ms() + self.peak_db = -100.0 + self.rms_db = -100.0 + self.track = None # set to track_start() to record per-band extremes + # Per band: offset (tilt) in half-dB units, folded into one scale. + import math + + span = 2.0 * (TOP_DB - FLOOR_DB) + self._scale = 1.0 / span + self._offset = [] + for hz in band_centres(bands): + tilt = TILT_DB * math.log(hz / TILT_FROM_HZ) / math.log(2) if hz > TILT_FROM_HZ else 0 + # level = (byte + 2*tilt - 2*(FLOOR_DB + 100)) / span + self._offset.append(2.0 * tilt - 2.0 * (FLOOR_DB + 100.0)) + + def raw(self): + """(seq, levels bytes, peak, rms) straight from C.""" + return _usbif.uac_pump_levels(self._buf) + + def levels(self, t=None): + seq, buf, pk, rms = _usbif.uac_pump_levels(self._buf) + out = self._out + now = ticks_ms() + if seq != self._seq: + self._seq = seq + self._fresh = now + sc, off = self._scale, self._offset + for i in range(self.bands): + out[i] = (buf[i] + off[i]) * sc + tr = self.track + if tr and buf[0] | buf[self.bands // 2] | buf[-1]: + hi, lo, tot = tr[0], tr[1], tr[2] + for i in range(self.bands): + b = buf[i] + if b > hi[i]: + hi[i] = b + if b < lo[i]: + lo[i] = b + tot[i] += b + tr[3] += 1 + self.peak_db = pk / 2 - 100 + self.rms_db = rms / 2 - 100 + elif ticks_diff(now, self._fresh) > STALE_MS: + for i in range(self.bands): + out[i] = 0.0 + self.peak_db = self.rms_db = -100.0 + return out + + def track_start(self): + """Record each band's loudest, quietest and summed raw level (half-dB + bytes) over every fresh analysis from now on, while audio flows.""" + n = self.bands + self.track = [bytearray(n), bytearray(b"\xff" * n), [0] * n, 0] + return self.track + + def close(self): + _usbif.uac_pump_meter(0) diff --git a/lib/examples/spectrum/spectrum.py b/lib/examples/spectrum/spectrum.py new file mode 100644 index 00000000..dea5e7ee --- /dev/null +++ b/lib/examples/spectrum/spectrum.py @@ -0,0 +1,226 @@ +# deps: pygraphics +# gallery: skip +""" +spectrum.py -- an audio spectrum analyzer. + +Log-spaced bars from 20 Hz to 20 kHz in a cool gradient, with peak-hold caps, +a reflection under the baseline, and frequency labels at their true log +positions. Drawn with ``pygraphics``: the static art once, then each bar as +its own packed column, so a frame copies and sends only the rows that moved. + +The layout comes from the display's size, so the same file runs on a desktop +window, the ESP32-P4 panel (800x480) and the T-Embed (320x170). On desktop it +opens at 800x480 unless ``PYDEVICES_WIDTH``/``PYDEVICES_HEIGHT`` say otherwise. +Set ``SPECTRUM_STYLE=segmented`` for LED-style bars. + +The levels come from the sound card's C pump when the firmware has one +(``pump_levels.PumpLevels``: run this beside usbif's ``soundcard.py``), from +what the computer is playing on a CPython desktop with numpy and +``soundcard`` installed (``loopback_levels.LoopbackLevels``), and from +``fake_music.FakeMusic`` otherwise. ``SPECTRUM_SOURCE=fake|pump|loopback`` +forces one. Either only has to hand ``SpectrumView.update`` one 0..1 level per band. + +``capture(path)`` writes the frame on screen to a file as raw RGB565, for a +screenshot of a real panel. +""" + +import sys + +from displaydev import env_get, env_set + +# A desktop board_config reads these; a real board's display ignores them. +if env_get("PYDEVICES_WIDTH") is None: + env_set("PYDEVICES_WIDTH", 800) + env_set("PYDEVICES_HEIGHT", 480) +if env_get("PYDEVICES_SCALE") is None: + env_set("PYDEVICES_SCALE", 1.0) + +_here = __file__.replace("\\", "/").rsplit("/", 1)[0] if "/" in __file__ else "." +if _here not in sys.path: + sys.path.insert(0, _here) + +import board_config # noqa: E402 +import appdev # noqa: E402 +from board_config import display_drv # noqa: E402 +from multimer import ticks_diff, ticks_ms # noqa: E402 + +from fake_music import FakeMusic # noqa: E402 +from spectrum_view import SpectrumView, band_count_for # noqa: E402 +from pygraphics import RGB565, FrameBuffer # noqa: E402 + +try: + from time import ticks_us +except ImportError: # CPython + + def ticks_us(): + return ticks_ms() * 1000 + + +FRAME_MS = 16 +REPORT_S = 5 + +# A panel that needs presenting (the P4's DSI panel samples its framebuffer +# only when told to) gets just the changed rows presented after each frame, +# rather than appdev's whole-frame refresh every 33 ms, which costs 22 ms a +# time on the P4 and keeps PSRAM busy for two thirds of every second. +_present_rows = bool(getattr(display_drv, "needs_refresh", False)) and hasattr( + getattr(display_drv, "_raw_buffer", None), "refresh_rect" +) +app = None +view = None +music = None +_send = None +_y = 0 # the meter's top row on the panel +timer = None + +def _source(): + """The sound card's C pump on a board, what the computer is playing on a + desktop (``loopback_levels``), else the fake. ``SPECTRUM_SOURCE`` forces + one: ``fake``, ``pump`` or ``loopback``.""" + want = env_get("SPECTRUM_SOURCE") + if want != "fake": + if want in (None, "pump"): + try: + import pump_levels + + if pump_levels.available(): + return pump_levels.PumpLevels(view.bands) + except ImportError: + pass + if want in (None, "loopback"): + try: + import loopback_levels + + if loopback_levels.available(): + return loopback_levels.LoopbackLevels(view.bands) + except Exception as error: # no loopback device, no audio stack + print("spectrum: no loopback source (%r)" % (error,)) + return FakeMusic(view.bands) + + +def _panel_blit(): + """How a run of bar rows reaches the panel. + + Where the driver shares a packed framebuffer that has to be presented + (the P4's DSI panel), rows are copied straight into it with no cache sync, + and the frame's dirty band is synced and presented once, after the bars. + The panel's own ``blit`` syncs the cache for every call, and a cache sync + runs with interrupts off: forty-odd of them a frame kept the USB + interrupt waiting long enough to lose sound-card packets. Otherwise, the + driver's ``blit_rect``.""" + if _present_rows and getattr(display_drv, "share_framebuffer", False): + buf, _, n, stride = display_drv.framebuffers() + w, h = display_drv.width, display_drv.height + if n == w * h * 2 and stride == w * 2: + return FrameBuffer(buf, w, h, RGB565).blit_rect + return display_drv.blit_rect + + +_dirty = [0, 0] # rows touched this frame, top and bottom + + +def blit(buf, x, y, w, h): + y += _y + _send(buf, x, y, w, h) + d = _dirty + if y < d[0]: + d[0] = y + if y + h > d[1]: + d[1] = y + h + + +last_report = "" +_t0 = 0 +_last = 0 +_stats = [0, 0, 0, 0, 0] # frames, data us, draw us, blit us, report start + + +def _tick(_=None): + global _last, last_report + now = ticks_ms() + dt = ticks_diff(now, _last) / 1000 + _last = now + a = ticks_us() + levels = music.levels(ticks_diff(now, _t0) / 1000) + b = ticks_us() + view.update(levels, min(dt, 0.1)) + c = ticks_us() + _dirty[0], _dirty[1] = _y + view.height, 0 + view.render_columns(blit) + if _present_rows and _dirty[1] > _dirty[0]: + display_drv.flush_rect(0, _dirty[0], view.width, _dirty[1] - _dirty[0]) + d = ticks_us() + s = _stats + s[0] += 1 + s[1] += b - a + s[2] += c - b + s[3] += d - c + span = ticks_diff(now, s[4]) + if span >= REPORT_S * 1000: + n = s[0] + last_report = "{:.1f} fps; per frame: data {:.2f} ms, update {:.2f} ms, draw+send {:.2f} ms".format( + n * 1000 / span, s[1] / n / 1000, s[2] / n / 1000, s[3] / n / 1000 + ) + print(last_report) + s[0] = s[1] = s[2] = s[3] = 0 + s[4] = now + + + +def capture(path): + """Write the meter's rows on screen to ``path`` as raw little-endian + RGB565: from the panel's own framebuffer when the driver shares it, else a + rebuild.""" + frame = None + fbs = getattr(display_drv, "framebuffers", None) + if fbs is not None and getattr(display_drv, "share_framebuffer", False): + buf, _, n, stride = fbs() + if stride == view.width * 2 and n >= (_y + view.height) * stride: + frame = memoryview(buf)[_y * stride : (_y + view.height) * stride] + with open(path, "wb") as f: + f.write(frame if frame is not None else view.compose()) + return view.width, view.height + + +def start(bands=None, height=None, y=None, style=None): + """Start the meter. Every argument is optional: + + - ``bands``: how many bars (x resolution). Default: half of what + ``band_count_for`` gives for the panel's width. + - ``height``: the meter's height in rows (y resolution). Default: half + the panel. + - ``y``: the meter's top row. Default: the meter sits at the bottom of + the panel. + - ``style``: ``"smooth"`` or ``"segmented"``; default ``SPECTRUM_STYLE`` + or smooth. + + Halving both was Brad's trade for frame rate under loud music + (2026-09-26): 18-19 fps full size, 35-45 at half by half on the P4. + """ + global app, view, music, _send, _y, timer, _t0, _last, _stats + W, H = display_drv.width, display_drv.height + if height is None: + height = H // 2 + if bands is None: + bands = band_count_for(W) // 2 + _y = H - height if y is None else y + app = appdev.App(board_config, refresh_period=0 if _present_rows else None) + view = SpectrumView(W, height, bands=bands, style=style or env_get("SPECTRUM_STYLE") or "smooth") + music = _source() + _send = _panel_blit() + fill = getattr(display_drv, "fill_rect", None) + if fill is not None and height < H: + fill(0, 0, W, H, 0) # clear whatever the panel showed before + display_drv.blit_rect(view.strip(0, view.height), 0, _y, view.width, view.height) + if _present_rows: + display_drv.show() + view._build_bar_columns() # a second on the P4; not on the first frame + _t0 = ticks_ms() + _last = _t0 + _stats = [0, 0, 0, 0, _t0] + timer = app.every(_tick, period=FRAME_MS) + return view + + +if __name__ == "__main__": + start() diff --git a/lib/examples/spectrum/spectrum_view.py b/lib/examples/spectrum/spectrum_view.py new file mode 100644 index 00000000..64bc0b23 --- /dev/null +++ b/lib/examples/spectrum/spectrum_view.py @@ -0,0 +1,397 @@ +""" +spectrum_view.py -- draws a log-frequency spectrum analyzer with pygraphics. + +``SpectrumView`` owns two RGB565 framebuffers the size of the screen: a static +background (gradient, grid, frequency labels) painted once, and the frame that +gets shown. Each frame it copies the background back over only the rows the +bars could have touched, blits the bars on top, and returns that row strip so +the caller can push just those rows to the panel. + +A bar is one ``blit`` from a precomputed gradient column, so its cost does not +grow with its height. The reflection is a second, shorter column blitted under +the baseline, and the gaps of the segmented style are a colour key in the +column, so the grid shows through them. + +Levels come in as 0..1 per band (0 = the floor, 1 = full scale); the view +applies the rise and fall ballistics and the peak-hold caps itself, so a +source only has to say how loud each band is right now. +""" + +import math + +from pygraphics import RGB565, Font, FrameBuffer + +# Bottom-to-top colour stops of the bars: deep blue, blue, cyan, teal, violet. +PALETTE = ( + (0.00, (10, 24, 110)), + (0.28, (0, 84, 220)), + (0.52, (0, 180, 245)), + (0.70, (30, 240, 225)), + (0.86, (120, 150, 255)), + (1.00, (200, 110, 255)), +) +BG_TOP = (1, 3, 10) # background at the top of the plot +BG_BASE = (4, 10, 26) # background at the baseline +BG_FLOOR = (2, 5, 14) # under the baseline (reflection and labels) +GRID = (20, 36, 64) +GRID_MINOR = (13, 24, 46) +BASELINE = (30, 70, 120) +LABEL = (96, 136, 180) +LABEL_DECADE = (150, 205, 240) + +LOW_HZ = 20.0 +HIGH_HZ = 20000.0 +LABELS = (50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000) + +ATTACK_S = 0.025 # rise time constant +FALL_PER_S = 1.6 # bars fall this much of full scale per second +PEAK_HOLD_S = 0.55 # peak caps hang this long before falling +PEAK_GRAVITY = 2.6 # then accelerate down at this (full scale per s^2) + +KEY = 0x0020 # colour key for transparent gradient rows; never a drawn colour + + +def rgb565(rgb): + r, g, b = rgb + c = ((int(r) & 0xF8) << 8) | ((int(g) & 0xFC) << 3) | (int(b) >> 3) + return c + 1 if c == KEY else c + + +def mix(a, b, t): + return (a[0] + (b[0] - a[0]) * t, a[1] + (b[1] - a[1]) * t, a[2] + (b[2] - a[2]) * t) + + +def palette_at(frac): + if frac <= PALETTE[0][0]: + return PALETTE[0][1] + for i in range(1, len(PALETTE)): + f1, c1 = PALETTE[i] + if frac <= f1: + f0, c0 = PALETTE[i - 1] + return mix(c0, c1, (frac - f0) / (f1 - f0)) + return PALETTE[-1][1] + + +def band_count_for(plot_w): + """About one band per 12-16 px, clamped to 16..48.""" + return max(16, min(48, plot_w // (12 if plot_w < 480 else 16))) + + +def label_text(hz): + return "{}k".format(hz // 1000) if hz >= 1000 else str(hz) + + +def band_centres(n): + """Centre frequency of each of ``n`` log-spaced bands between 20 Hz and 20 kHz.""" + span = HIGH_HZ / LOW_HZ + return [LOW_HZ * span ** ((i + 0.5) / n) for i in range(n)] + + +class SpectrumView: + def __init__(self, width, height, bands=None, style="smooth"): + self.width = width + self.height = height + self.style = style + self._buf = bytearray(width * height * 2) + self._bgbuf = bytearray(width * height * 2) + self.fb = FrameBuffer(self._buf, width, height, RGB565) + self._bg = FrameBuffer(self._bgbuf, width, height, RGB565) + self._mv = memoryview(self._buf) + self._bgmv = memoryview(self._bgbuf) + + # Layout, all from the screen size. + big = height >= 300 + self.font = Font(height=16 if big else 8) + fh = self.font.height + pad = max(3, min(width, height) // 40) + side = max(pad, (3 * 8) // 2 + 2) # room for a centred "20k" at the right edge + plot_w = width - 2 * side + n = bands or band_count_for(plot_w) + pitch = plot_w // n + self.bands = n + self.pitch = pitch + self.gap = max(1, pitch // 5) + self.bar_w = pitch - self.gap + self.x0 = side + (plot_w - n * pitch) // 2 + self.gap // 2 + self.span_x0 = self.x0 - self.gap // 2 # left edge of the band span + self.span_w = n * pitch + self.label_y = height - fh - max(2, pad // 2) + free = self.label_y - pad - 3 + self.refl_h = max(4, free // 8) + self.top = pad + self.baseline = self.label_y - 3 - self.refl_h - 1 # first row below the bars + self.plot_h = self.baseline - self.top + self.refl_y = self.baseline + 1 + self.cap_h = max(2, self.plot_h // 120 + 1) + self.hi_h = max(1, self.plot_h // 180) + seg_pitch = max(4, self.plot_h // 44) + self.seg_pitch = seg_pitch if style == "segmented" else 0 + + self._paint_background() + self._build_columns() + + # Ballistics state: bar level, peak level, peak hold timer, peak velocity. + self.level = [0.0] * n + self.peak = [0.0] * n + self.hold = [0.0] * n + self.pvel = [0.0] * n + self._last_top = self.baseline + self.fb.blit(self._bg, 0, 0) + + # --- static art ------------------------------------------------------- + + def _bg_row_color(self, y): + if y >= self.baseline: + return BG_FLOOR + t = (y - self.top) / max(1, self.plot_h) + t = max(0.0, min(1.0, t)) + return mix(BG_TOP, BG_BASE, t * t) + + def _paint_background(self): + bg = self._bg + w = self.width + # Vertical gradient, one hline per row (a one-off cost). + for y in range(self.height): + bg.hline(0, y, w, rgb565(self._bg_row_color(y))) + x0, x1 = self.span_x0, self.span_x0 + self.span_w + # Horizontal grid: quarters of full scale. + for q in (1, 2, 3): + y = self.baseline - (self.plot_h * q) // 4 + bg.hline(x0, y, x1 - x0, rgb565(GRID_MINOR if q != 2 else GRID)) + bg.hline(x0, self.top, x1 - x0, rgb565(GRID_MINOR)) + # Vertical grid and labels at the true log position of each frequency. + fh = self.font.height + for hz in LABELS: + x = self.freq_x(hz) + decade = hz in (100, 1000, 10000) + if x < x1 - 1: + bg.vline(x, self.top, self.plot_h, rgb565(GRID if decade else GRID_MINOR)) + s = label_text(hz) + tw = self.font.text_width(s) + tx = max(0, min(self.width - tw, x - tw // 2)) + self.font.text(bg, s, tx, self.label_y, rgb565(LABEL_DECADE if decade else LABEL)) + bg.vline(x, self.label_y - 3, 2, rgb565(LABEL)) + if fh >= 16: + self.font.text(bg, "Hz", 2, self.label_y, rgb565(LABEL)) + bg.hline(x0, self.baseline, x1 - x0, rgb565(BASELINE)) + + def freq_x(self, hz): + u = math.log(hz / LOW_HZ) / math.log(HIGH_HZ / LOW_HZ) + return self.span_x0 + int(u * self.span_w + 0.5) + + def _build_columns(self): + bw, ph, rh = self.bar_w, self.plot_h, self.refl_h + # Bar column: row 0 is the top of the plot. + col = bytearray(bw * ph * 2) + cfb = FrameBuffer(col, bw, ph, RGB565) + self._hilite = [0] * ph + self._capcol = [0] * ph + sp = self.seg_pitch + for r in range(ph): + h = ph - r # height of this row above the baseline, 1..ph + c = palette_at(h / ph) + gap = sp and (h % sp == 0) + cfb.hline(0, r, bw, KEY if gap else rgb565(c)) + self._hilite[r] = rgb565(mix(c, (255, 255, 255), 0.45)) + self._capcol[r] = rgb565(mix(c, (235, 240, 255), 0.7)) + self._col = col + # Reflection column: row 0 sits just under the baseline and mirrors the + # bar's bottom row; it fades into the floor and skips every other line. + rcol = bytearray(bw * rh * 2) + rfb = FrameBuffer(rcol, bw, rh, RGB565) + for i in range(rh): + h = i + 1 + c = palette_at(h / ph) + fade = 0.75 * (1 - i / rh) ** 1.2 + gap = (i & 1) or (sp and (h % sp == 0)) + rfb.hline(0, i, bw, KEY if gap else rgb565(mix(BG_FLOOR, c, fade))) + self._rcol = rcol + + # --- per frame ---------------------------------------------------------- + + def update(self, targets, dt): + """Move the bars toward ``targets`` (0..1 per band) over ``dt`` seconds.""" + a = 1 - math.exp(-dt / ATTACK_S) + fall = FALL_PER_S * dt + level, peak, hold, pvel = self.level, self.peak, self.hold, self.pvel + for i in range(self.bands): + t = targets[i] + t = 0.0 if t < 0 else (1.0 if t > 1 else t) + v = level[i] + if t > v: + v += (t - v) * a + else: + v = max(t, v - fall) + level[i] = v + if v >= peak[i]: + peak[i] = v + hold[i] = PEAK_HOLD_S + pvel[i] = 0.0 + elif hold[i] > 0: + hold[i] -= dt + else: + pvel[i] += PEAK_GRAVITY * dt + peak[i] = max(v, peak[i] - pvel[i] * dt) + + def render(self): + """Draw this frame; return ``(y, h)``, the row strip that changed.""" + fb = self.fb + ph, bw, base = self.plot_h, self.bar_w, self.baseline + cap_h, hi_h = self.cap_h, self.hi_h + # The strip to redraw: from the highest thing on screen, this frame or + # last, down to the bottom of the reflection. + top = base + for p in self.peak: + y = base - int(p * ph) - cap_h - 1 + if y < top: + top = y + top = max(self.top, top) + y0 = min(top, self._last_top) + self._last_top = top + y1 = self.refl_y + self.refl_h + w2 = self.width * 2 + self._mv[y0 * w2 : y1 * w2] = self._bgmv[y0 * w2 : y1 * w2] + + col, rcol, rh = self._col, self._rcol, self.refl_h + hilite, capcol = self._hilite, self._capcol + x = self.x0 + pitch = self.pitch + sp = self.seg_pitch + for i in range(self.bands): + h = int(self.level[i] * ph) + if sp: + h -= h % sp + if h > 0: + r = ph - h + fb.blit((memoryview(col)[r * bw * 2 :], bw, h, RGB565), x, base - h, KEY) + if not sp: + fb.fill_rect(x, base - h, bw, min(hi_h, h), hilite[r]) + rr = h if h < rh else rh + fb.blit((rcol, bw, rr, RGB565), x, self.refl_y, KEY) + ph_ = int(self.peak[i] * ph) + if ph_ > 0: + yc = base - ph_ - cap_h - 1 + if yc >= self.top: + fb.fill_rect(x, yc, bw, cap_h, capcol[ph - ph_]) + x += pitch + return y0, y1 - y0 + + # --- per column, for panels where every byte through PSRAM counts ------- + + def _build_bar_columns(self): + """Per bar, two packed columns (bar_w wide, top of plot to bottom of + reflection): the background under it, and the same with the bar fully + lit. A frame then sends row runs of one or the other straight to the + panel, plus the highlight and the cap from small prebuilt rows, so no + pixel is copied twice.""" + bw, top = self.bar_w, self.top + self._ch = self.refl_y + self.refl_h - top # column height in rows + ch, bw2 = self._ch, bw * 2 + w2 = self.width * 2 + bgmv = self._bgmv + col, rcol, ph, rh = self._col, self._rcol, self.plot_h, self.refl_h + self._cbg, self._clit = [], [] + x = self.x0 + for i in range(self.bands): + bg = bytearray(ch * bw2) + for r in range(ch): + o = (top + r) * w2 + x * 2 + bg[r * bw2 : (r + 1) * bw2] = bgmv[o : o + bw2] + lit = bytearray(bg) + lfb = FrameBuffer(lit, bw, ch, RGB565) + lfb.blit((col, bw, ph, RGB565), 0, 0, KEY) + lfb.blit((rcol, bw, rh, RGB565), 0, self.refl_y - top, KEY) + self._cbg.append(memoryview(bg)) + self._clit.append(memoryview(lit)) + x += self.pitch + # One highlight run and one cap per row they can sit on. + hi_h, cap_h = min(self.hi_h, ph), self.cap_h + self._hirow = [c.to_bytes(2, "little") * (bw * hi_h) for c in self._hilite] + self._caprow = [c.to_bytes(2, "little") * (bw * cap_h) for c in self._capcol] + n = self.bands + self._shown_h = [0] * n + self._shown_cap = [-1] * n + self._shown_rr = [0] * n + + def render_columns(self, blit): + """Draw this frame bar by bar, calling ``blit(buf, x, y, w, h)`` for + each run of rows that changed. Only what moved is sent.""" + if not hasattr(self, "_cbg"): + self._build_bar_columns() + ph, bw, top = self.plot_h, self.bar_w, self.top + bw2 = bw * 2 + cap_h, hi_h = self.cap_h, self.hi_h + sp = self.seg_pitch + roff = self.refl_y - top + rh = self.refl_h + hirow, caprow = self._hirow, self._caprow + shown_h, shown_cap, shown_rr = self._shown_h, self._shown_cap, self._shown_rr + level, peak = self.level, self.peak + x = self.x0 + pitch = self.pitch + for i in range(self.bands): + h = int(level[i] * ph) + if sp: + h -= h % sp + p = int(peak[i] * ph) + cy = ph - p - cap_h - 1 if p > 0 else -1 # cap row in column coords + oh, ocy = shown_h[i], shown_cap[i] + if h != oh or cy != ocy: + bt, obt = ph - h, ph - oh + r0 = bt if bt < obt else obt + if 0 <= cy < r0: + r0 = cy + if 0 <= ocy < r0: + r0 = ocy + r1 = (bt if bt > obt else obt) + (0 if sp else hi_h) + if ocy >= 0 and ocy + cap_h > r1: + r1 = ocy + cap_h + if r1 > ph: + r1 = ph + if r0 < bt: + blit(self._cbg[i][r0 * bw2 : bt * bw2], x, top + r0, bw, bt - r0) + if bt < r1: + blit(self._clit[i][bt * bw2 : r1 * bw2], x, top + bt, bw, r1 - bt) + if h > 0 and not sp: + hh = hi_h if hi_h < h else h + blit(hirow[bt] if hh == hi_h else hirow[bt][: bw2 * hh], x, top + bt, bw, hh) + if cy >= 0: + blit(caprow[ph - p], x, top + cy, bw, cap_h) + shown_h[i] = h + shown_cap[i] = cy + rr = h if h < rh else rh + orr = shown_rr[i] + if rr != orr: + if rr > orr: + blit(self._clit[i][(roff + orr) * bw2 : (roff + rr) * bw2], x, top + roff + orr, bw, rr - orr) + else: + blit(self._cbg[i][(roff + rr) * bw2 : (roff + orr) * bw2], x, top + roff + rr, bw, orr - rr) + shown_rr[i] = rr + x += pitch + + def compose(self): + """The whole frame as the panel shows it, rebuilt into ``self.fb`` + from the background and the shown state (for a screenshot).""" + self.fb.blit(self._bg, 0, 0) + if hasattr(self, "_cbg"): + fb = self.fb + + def put(buf, x, y, w, h): + fb.blit_rect(buf, x, y, w, h) + + n = self.bands + h, c, r = self._shown_h, self._shown_cap, self._shown_rr + self._shown_h, self._shown_cap, self._shown_rr = [0] * n, [-1] * n, [0] * n + lv, pk = self.level, self.peak + ph = self.plot_h + self.level = [(x + 0.5) / ph for x in h] + self.peak = [(ph - y - self.cap_h - 1 + 0.5) / ph if y >= 0 else 0 for y in c] + self.render_columns(put) + self.level, self.peak = lv, pk + self._shown_h, self._shown_cap, self._shown_rr = h, c, r + return self._buf + + def strip(self, y, h): + """The bytes of rows ``y..y+h`` of the frame, for ``blit_rect``.""" + w2 = self.width * 2 + return self._mv[y * w2 : (y + h) * w2] diff --git a/packages/examples.json b/packages/examples.json index f93fc0db..e30ad887 100644 --- a/packages/examples.json +++ b/packages/examples.json @@ -520,6 +520,38 @@ "roku_remote/roku_widgets.py", "github:PyDevices/pydevices-examples/lib/examples/roku_remote/roku_widgets.py" ], + [ + "spectrum/__init__.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/__init__.py" + ], + [ + "spectrum/capture.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/capture.py" + ], + [ + "spectrum/fake_music.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/fake_music.py" + ], + [ + "spectrum/loopback_levels.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/loopback_levels.py" + ], + [ + "spectrum/make_captures.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/make_captures.py" + ], + [ + "spectrum/pump_levels.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/pump_levels.py" + ], + [ + "spectrum/spectrum.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/spectrum.py" + ], + [ + "spectrum/spectrum_view.py", + "github:PyDevices/pydevices-examples/lib/examples/spectrum/spectrum_view.py" + ], [ "tiny_toasters/__init__.py", "github:PyDevices/pydevices-examples/lib/examples/tiny_toasters/__init__.py" diff --git a/packages/spectrum.json b/packages/spectrum.json new file mode 100644 index 00000000..ba757df2 --- /dev/null +++ b/packages/spectrum.json @@ -0,0 +1,37 @@ +{ + "urls": [ + [ + "spectrum/__init__.py", + "../lib/examples/spectrum/__init__.py" + ], + [ + "spectrum/capture.py", + "../lib/examples/spectrum/capture.py" + ], + [ + "spectrum/fake_music.py", + "../lib/examples/spectrum/fake_music.py" + ], + [ + "spectrum/loopback_levels.py", + "../lib/examples/spectrum/loopback_levels.py" + ], + [ + "spectrum/make_captures.py", + "../lib/examples/spectrum/make_captures.py" + ], + [ + "spectrum/pump_levels.py", + "../lib/examples/spectrum/pump_levels.py" + ], + [ + "spectrum/spectrum.py", + "../lib/examples/spectrum/spectrum.py" + ], + [ + "spectrum/spectrum_view.py", + "../lib/examples/spectrum/spectrum_view.py" + ] + ], + "version": "0.0.1" +} diff --git a/pydevices-examples.toml b/pydevices-examples.toml index 4824d5ed..73c6c636 100644 --- a/pydevices-examples.toml +++ b/pydevices-examples.toml @@ -124,6 +124,14 @@ "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/scroll.py" = "/examples/scroll.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/scroll_touch_test.py" = "/examples/scroll_touch_test.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/simon.py" = "/examples/simon.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/__init__.py" = "/examples/spectrum/__init__.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/capture.py" = "/examples/spectrum/capture.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/fake_music.py" = "/examples/spectrum/fake_music.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/loopback_levels.py" = "/examples/spectrum/loopback_levels.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/make_captures.py" = "/examples/spectrum/make_captures.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/pump_levels.py" = "/examples/spectrum/pump_levels.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/spectrum.py" = "/examples/spectrum/spectrum.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/spectrum/spectrum_view.py" = "/examples/spectrum/spectrum_view.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/stt_groq.py" = "/examples/stt_groq.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/testris.py" = "/examples/testris.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/tiny_hello.py" = "/examples/tiny_hello.py" diff --git a/tests/test_browser_url.py b/tests/test_browser_url.py index 7884cfcd..de4d1d05 100644 --- a/tests/test_browser_url.py +++ b/tests/test_browser_url.py @@ -5,9 +5,14 @@ import unittest _ROOT = Path(__file__).resolve().parents[1] +# The sibling pydevices checkout: next to this repo, inside it (CI), or next to +# an ancestor when this is a worktree under .worktrees/. _TOOLS = next( path - for path in (_ROOT.parent / "pydevices" / "tools", _ROOT / "pydevices" / "tools") + for path in ( + _ROOT / "pydevices" / "tools", + *(parent / "pydevices" / "tools" for parent in _ROOT.parents), + ) if (path / "_browser_url.py").is_file() ) sys.path.insert(0, str(_TOOLS)) diff --git a/tools/example_test_manifest.toml b/tools/example_test_manifest.toml index 44735ee0..3ea3241b 100644 --- a/tools/example_test_manifest.toml +++ b/tools/example_test_manifest.toml @@ -107,6 +107,13 @@ quit = "poll" deps = ["core"] timeout_s = 45 +[examples.spectrum] +import = "spectrum" +script = "examples/spectrum/spectrum.py" +kind = "async" +quit = "poll" +deps = ["core"] + [examples.appdev_simpletest] import = "appdev_simpletest" script = "examples/appdev_simpletest.py" diff --git a/tools/spectrum/meter_gate.py b/tools/spectrum/meter_gate.py new file mode 100644 index 00000000..44820932 --- /dev/null +++ b/tools/spectrum/meter_gate.py @@ -0,0 +1,122 @@ +"""Audio meter gate, board half. Mode set by the host before upload. + +MODE 'off' : sound card pump only (soundcard.py's configuration). +MODE 'c' : plus the C meter analysing, nothing drawn. +MODE 'draw' : plus the spectrum drawing on the panel at frame rate. + +Every 10 s: packets/s against 8000, pump% against the wire rate, the meter's +CPU cost and worst analysis time, and the drawing's fps. Logged to /gate.txt. +""" + +import sys +import time + +import _usbif +import board_peripherals as bp +import usbif.auto + +MODE = "draw" +VOLUME = 20 +WINDOWS = 8 +TRACK = 0 +CAPTURE_AT = 0 +WIRE = 48000 + +_LOG = "/gate.txt" +try: + import os + + os.remove(_LOG) +except OSError: + pass + + +def log(*a): + line = " ".join(str(x) for x in a) + with open(_LOG, "a") as f: + f.write(line + "\n") + print(line) + + +dev = usbif.auto.device() +w = bp.AUDIO_OUT.wire +bp.audio_power(True, volume=VOLUME) +dev.functions("cdc", "uac") +kw = {"rate": WIRE, "bits": bp.AUDIO_OUT.default.bits, "channels": bp.AUDIO_OUT.default.channels} +if w.mck is not None and w.mck >= 0: + kw["mclk"] = w.mck + kw["mclk_multiple"] = w.mck_fs +dev.uac_pump_stop() +_usbif.uac_pump_meter(0) +dev.uac_pump_start(w.sck, w.ws, w.sd, **kw) + +spectrum = None +if MODE == "c": + _usbif.uac_pump_meter(44, 20, 20000) +elif MODE == "draw": + sys.path.insert(0, "/spectrum") + import spectrum # noqa: F811 + + spectrum.start() + + if TRACK: + spectrum.music.track_start() +log("MEASURE start mode=%s volume=%d" % (MODE, VOLUME)) + +prev = None +k = 0 +while k < WINDOWS: + time.sleep_ms(10000) + s = dev.uac_pump_stats() + c = _usbif.uac_pump_clock() + mt = _usbif.uac_pump_meter() + host, wire, _ = dev.uac_pump_rate() + cur = (s[1], c[0], c[4], s[3], mt[2], mt[3], mt[4], mt[6], c[2]) + if prev: + k += 1 + dt = (cur[2] - prev[2]) / 1e6 + pump = (cur[0] - prev[0]) / dt + pk = (cur[1] - prev[1]) / dt + el = (cur[7] - prev[7]) or 1 + feed = 100 * (cur[5] - prev[5]) / el + fft = 100 * (cur[6] - prev[6]) / el + an = (cur[4] - prev[4]) / dt + fps = spectrum.last_report if spectrum else "" + dma = (cur[8] - prev[8]) / dt / 2 + log( + "MEASURE host=%d wire=%d pkts/s=%.1f pump%%=%.3f timeouts=%d " + "dma_rate=%.1f meter: %.1f/s feed=%.2f%% fft=%.2f%% worst=%dus | %s" + % ( + host, + wire, + pk, + 100 * pump / (2 * wire) if wire else 0, + cur[3] - prev[3], + dma, + an, + feed, + fft, + mt[5], + fps, + ) + ) + if CAPTURE_AT and k == CAPTURE_AT and spectrum: + spectrum.capture("/spectrum.raw") + log("CAPTURED /spectrum.raw %dx%d" % (spectrum.view.width, spectrum.view.height)) + prev = cur + +if spectrum and TRACK: + tr = spectrum.music.track + n = tr[3] or 1 + from spectrum_view import band_centres + + for i, hz in enumerate(band_centres(spectrum.view.bands)): + log( + "BAND %2d %7.1f Hz max=%6.1f mean=%6.1f min=%6.1f dB" + % (i, hz, tr[0][i] / 2 - 100, tr[2][i] / n / 2 - 100, tr[1][i] / 2 - 100) + ) + log("BANDS analyses=%d" % tr[3]) +log("MEASURE done") +if spectrum: + spectrum.timer.cancel() if hasattr(spectrum.timer, "cancel") else None +dev.uac_pump_stop() diff --git a/tools/spectrum/play_src.py b/tools/spectrum/play_src.py new file mode 100644 index 00000000..11529f1b --- /dev/null +++ b/tools/spectrum/play_src.py @@ -0,0 +1,102 @@ +"""Play test material to the Espressif sound card over WASAPI (Windows Python). + + python.exe play_src.py SECONDS SOURCE [GAIN] [excl|shared] + +SOURCE: tone | pink | sweep | mix (10 s pink, 10 s log sweep 20 Hz-20 kHz, +alternating) | a path to a 48 kHz wav. Loops. GAIN is linear (default 0.25). +Prints the frame rate the host delivered and underflows, like uac_play_tone. +""" + +import sys +import time + +import numpy as np +import sounddevice as sd + +RATE = 48000 +secs = float(sys.argv[1]) +src = sys.argv[2] +gain = float(sys.argv[3]) if len(sys.argv) > 3 else 0.25 +excl = (sys.argv[4] if len(sys.argv) > 4 else "shared") == "excl" + + +def pink(n, seed=1): + rng = np.random.default_rng(seed) + w = rng.standard_normal(n) + f = np.fft.rfft(w) + k = np.arange(len(f)) + k[0] = 1 + f /= np.sqrt(k) + x = np.fft.irfft(f, n) + return x / np.max(np.abs(x)) + + +def sweep(n): + t = np.arange(n) / RATE + T = n / RATE + f0, f1 = 20.0, 20000.0 + k = np.log(f1 / f0) + ph = 2 * np.pi * f0 * T / k * (np.exp(t / T * k) - 1) + return 0.7 * np.sin(ph) + + +if src == "tone": + n = RATE + mono = 0.7 * np.sin(2 * np.pi * 440 * np.arange(n) / RATE) + data = np.stack([mono, mono], 1) +elif src == "pink": + mono = pink(10 * RATE) + data = np.stack([mono, mono], 1) +elif src == "sweep": + mono = sweep(10 * RATE) + data = np.stack([mono, mono], 1) +elif src == "mix": + mono = np.concatenate([pink(10 * RATE), sweep(10 * RATE)]) + data = np.stack([mono, mono], 1) +else: + import soundfile as sf + + data, r = sf.read(src, dtype="float64", always_2d=True) + assert r == RATE, r + if data.shape[1] == 1: + data = np.repeat(data, 2, 1) + data = data[:, :2] / max(1e-9, np.max(np.abs(data))) +pcm = np.clip(data * gain * 32767, -32768, 32767).astype(np.int16) + +dev = next( + i + for i, d in enumerate(sd.query_devices()) + if "Espressif" in d["name"] + and d["max_output_channels"] + and sd.query_hostapis(d["hostapi"])["name"] == "Windows WASAPI" +) +state = {"pos": 0, "frames": 0, "under": 0} + + +def cb(out, frames, t, status): + if status.output_underflow: + state["under"] += 1 + p = state["pos"] + n = len(pcm) + idx = (np.arange(frames) + p) % n + out[:] = pcm[idx] + state["pos"] = (p + frames) % n + state["frames"] += frames + + +with sd.OutputStream( + device=dev, + samplerate=RATE, + channels=2, + dtype="int16", + callback=cb, + extra_settings=sd.WasapiSettings(exclusive=excl), + latency="high", +): + t0 = time.perf_counter() + time.sleep(secs) + el = time.perf_counter() - t0 +print( + "HOST src=%s gain=%.3f exclusive=%s frames/s=%.1f underflows=%d" + % (src, gain, excl, state["frames"] / el, state["under"]) +)