midiviz: the MIDI stream as a picture, not a transcript

A lens over a live MIDI control surface: position is identity, glyph class is
type, bar+hex is value, brightness is recency — zero English words on canvas.
Born watching a Novation Launch Control XL 3 in ParVagues' livecoding rig;
watches any ALSA MIDI source, three themes, optional spectrum backdrop and
Tidal HL feed, resting-state layer over an atomically-published driver state
file, branding overridable by env (MIDIVIZ_BRAND / _WORDMARK / _BRAND_SLUG /
_WAVE), headless selftest with PNG proof, GPL-3.0-or-later.

.gitignore

+__pycache__/
+*.py[cod]
+*.egg-info/
+.pytest_cache/
+build/
+dist/
+.venv/

README.md

+# midiviz
+
+**Live MIDI as a picture, not a transcript.**
+
+`midimon` prints one line per event ("14:0 Control change ch0, controller 29,
+value 15"). That is the right shape for a debugger and the wrong shape for a
+performer: during a set nobody reads sentences, and a fader sweep emits ~400
+events a second, so the words scroll past faster than an eye can land on them.
+
+midiviz is a *lens*, not a log. Zero English words render: an event's identity
+is its **position**, its type is a **glyph class**, its value is a **bar plus
+two hex digits**, and its recency is **brightness**. The layout is the authored
+control surface itself — rows A..F down, columns 1..8 across — so a glance
+answers "which orbit did I just touch" without decoding anything. Only what the
+grid cannot place (foreign CCs, notes, bend, program) falls as rain in the
+right-hand gutter, which is itself the signal: *that came from somewhere else*.
+
+It was born in ParVagues' livecoding rig, watching
+a Novation **Launch Control XL 3** — but it watches any MIDI source, and its
+branding is yours to change.
+
+![midiviz, dark theme, mid-performance](screenshots/hero-branded-dark.png)
+
+## Highlights
+
+- **The grid is the surface.** Column *N* is orbit *N*; the picture is the
+  authored LCXL3 control layout (`lcxl_grid.py`), not an arbitrary CC table.
+- **Mapped-but-idle is never black.** Every grid cell keeps a floor tint, so a
+  connected-but-untouched control never reads as a hardware fault.
+- **Resting state, not just events.** MIDI is edge-triggered, so a filter
+  parked at zero and one parked at the top look identical once the glow fades.
+  If a driver publishes a state file (last-value-wins, atomically written),
+  midiviz paints what each control is *resting at* — three zones for the
+  centre-detented family filters, detent drawn.
+- **Three themes**: `dark` (the cockpit), `light` (pale desktop), `sun`
+  (near-white, heavy ink — for playing outdoors). Colours are data; every
+  theme is held to the same visibility contract by the test suite.
+- **Two frame rates, not a busy loop.** ~30 fps while events arrive, 5 fps two
+  seconds after the last one, precomputed colour LUTs, zero per-frame
+  allocation. Built on a machine that performs live audio.
+- **Spectrum backdrop** (optional, `s`): the master bus, behind the grid.
+- **Tidal HL feed** (optional, `h`): which orbits the music is actually
+  triggering right now, via a SuperDirt OSC mirror — the answer to "I turned
+  the knob and nothing happened: is the pattern silent, or is the knob not
+  wired?"
+- **Survives everything**: unplug/replug (clean-exit-proof by design),
+  restarts (theme + scale persist), tiling WMs (the window's size belongs to
+  the WM, the type density to the window), duplicate launches (flock, exit 78
+  — a deliberate close is never a fault).
+
+## The three themes
+
+| dark — the cockpit | light — pale desktop | sun — outdoors |
+|---|---|---|
+| ![](screenshots/theme-dark.png) | ![](screenshots/theme-light.png) | ![](screenshots/theme-sun.png) |
+
+## Install & run
+
+Needs Python ≥ 3.10, [PySide6](https://pypi.org/project/PySide6/), and
+`aseqdump` (Debian/Ubuntu: `sudo apt install alsa-utils`).
+
+```sh
+git clone https://github.com/PLNech/midiviz.git
+cd midiviz
+python3 midiviz.py
+```
+
+```sh
+python3 midiviz.py -l              # list MIDI ports, show what it would watch
+python3 midiviz.py -p 28:0         # pin a source port
+python3 midiviz.py --theme sun     # near-white, for playing outdoors
+python3 midiviz.py --spectro       # + master-bus spectrum, behind the grid
+python3 midiviz.py --selftest      # 3 s of synthetic events, headless
+```
+
+Keys: `q`/`Esc` quit · `p` pause · `c` clear · `s` spectrum · `t` pin ·
+`d` theme · `+`/`-` type density · `0` birth size · `w` wave · `l` lettering.
+Drag the body to move, the edges and corners to resize (the type follows the
+window); right-click for the menu (also in the system tray).
+
+Tested on Linux/ALSA (Wayland and X11). Ports are resolved automatically —
+a translated/virtual surface port is preferred over raw hardware ports.
+
+## Make it yours (branding)
+
+midiviz ships wearing **ParVagues** and changes nothing about how it works:
+
+```sh
+MIDIVIZ_BRAND="Coolnaut" \
+MIDIVIZ_WORDMARK="COOLNAUT" \
+MIDIVIZ_BRAND_SLUG="coolnaut" \
+MIDIVIZ_WAVE=$HOME/pictures/coolnaut-wave.png \
+python3 midiviz.py
+```
+
+- `MIDIVIZ_BRAND` — spoken name, menu labels
+- `MIDIVIZ_WORDMARK` — the block lettering drawn over row D
+- `MIDIVIZ_BRAND_SLUG` — directory under `$XDG_CONFIG_HOME` / `$XDG_RUNTIME_DIR`
+  for saved theme, scale, close latch and instance lock (one per brand)
+- `MIDIVIZ_WAVE` — any transparent PNG, watermarked faintly behind the grid
+
+## The wire
+
+State has a single legitimate owner, and it is not the viewer. midiviz *reads*
+two optional producers, and both are plain files — `cat` is the debugger:
+
+| path | producer | shape |
+|---|---|---|
+| `~/.cache/parvagues/surface-state.json` | any driver that integrates relative encoders and owns the absolute values | 32 values, last-value-wins, monotonic `seq`, written atomically |
+| `~/.cache/parvagues/eval-events.jsonl` | an editor publishing which orbits the loaded track triggers | one JSON object per line: `{"t":…, "path":…, "d":["d1",…]}` |
+
+Missing files are fine — those layers simply stay dark. The parvagues paths are
+the wire contract midiviz was born on, not branding; point your own producer at
+the same paths, or run everything under your own `MIDIVIZ_BRAND_SLUG` and write
+the state file there.
+
+A reference producer for the LCXL3 (delta integration, translated CC corpus)
+lives in the ParVagues rig repository and is not required to run midiviz:
+on a raw LCXL3 the event grid, gutter, spectrum and themes all work standalone.
+
+## Development
+
+```sh
+python3 -m pytest tests/              # the non-drawing contracts (no Qt needed to fail fast)
+python3 midiviz.py --selftest --screenshot shots/
+                                      # real paint path, headless; PNGs per theme
+```
+
+`--selftest` is the drawing prover: it renders synthetic events through the
+real parser, cycles every theme, resizes through the tiling-WM shapes and
+checks the paint output — green tests on pure functions prove nothing about a
+window, so this drives the window itself.
+
+Layout, one screenful:
+
+| file | role |
+|---|---|
+| `midiviz.py` | the lens: reader, widget, themes, state layer, selftest |
+| `midimon.py` | the one parser (`parse_line` / `enrich` / port resolution) |
+| `midistream.py` | threaded `aseqdump` reader |
+| `lcxl_grid.py` | THE grid, authored once; everything derives from it |
+| `oscfeed.py` | the Tidal HL feed (SuperDirt OSC mirror) |
+| `ui/` | vendored brand assets: wave PNG, display font (Syne 800) |
+| `packaging/` | example systemd user unit + desktop entry |
+
+## License
+
+GPL-3.0-or-later — see [LICENSE](LICENSE).

midimon.py

+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""midimon — a clean live MIDI monitor (aseqdump, parsed & rendered right).
+
+`aseqdump | tr -d …` is a hack. This wraps aseqdump, parses each event, and
+renders an aligned, colourised stream: note numbers → names (C4…), velocity as a
+little bar, events tinted by type. Ctrl-C to stop.
+
+    python3 midimon.py                 # monitor aseqdump's default port
+    python3 midimon.py -p 28:0         # subscribe to a specific source
+    python3 midimon.py -l              # list ports and exit
+    python3 midimon.py --raw           # passthrough (debug)
+
+Parsing is a pure function (`parse_line`) so it's unit-tested without ALSA.
+"""
+from __future__ import annotations
+
+import argparse
+import re
+import subprocess
+import sys
+
+NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
+
+# event keyword → ANSI colour
+C = {"note on": "92", "note off": "90", "control": "93", "program": "96",
+     "pitch": "95", "channel": "94", "aftertouch": "94",
+     "clock": "90", "start": "92", "stop": "91", "continue": "92",
+     "song": "90", "sysex": "95", "active": "90", "reset": "91"}
+RESET, DIM, BOLD = "\033[0m", "\033[2m", "\033[1m"
+
+# " 28:0   Note on                 0, note 60, velocity 100"
+_ROW = re.compile(r"^\s*(\d+:\d+)\s+(\S.*?\S|\S)\s{2,}(.*?)\s*$")
+
+
+def note_name(n: int) -> str:
+    return f"{NOTE_NAMES[n % 12]}{n // 12 - 1}"
+
+
+def parse_line(line: str) -> dict | None:
+    """aseqdump event row → {source, event, data, ch?}. None if not an event."""
+    m = _ROW.match(line.rstrip("\n"))
+    if not m:
+        return None
+    source, event, data = m.group(1), m.group(2), m.group(3)
+    ch = None
+    mc = re.match(r"^(\d+)\b", data)          # first int of data is the channel
+    if mc:
+        ch = int(mc.group(1))
+    return {"source": source, "event": event, "data": data, "ch": ch}
+
+
+def enrich(ev: dict) -> dict:
+    """Add structured fields parsed from `data` (note/velocity/controller/…).
+
+    One parsing source of truth — used by the CLI renderer and the web stream.
+    """
+    out = dict(ev)
+    for key, pat in (("note", r"note (\d+)"), ("velocity", r"velocity (\d+)"),
+                     ("controller", r"controller (\d+)"), ("value", r"value (-?\d+)"),
+                     ("program", r"program (\d+)")):
+        m = re.search(pat, ev["data"])
+        if m:
+            out[key] = int(m.group(1))
+    if "note" in out:
+        out["note_name"] = note_name(out["note"])
+    return out
+
+
+def _color(event: str) -> str:
+    e = event.lower()
+    for kw, code in C.items():
+        if e.startswith(kw):
+            return code
+    return "97"
+
+
+def _render_data(event: str, data: str) -> str:
+    """Friendlier data: note names + a velocity bar where it applies."""
+    note = re.search(r"note (\d+)", data)
+    vel = re.search(r"velocity (\d+)", data)
+    out = data
+    if note:
+        n = int(note.group(1))
+        out = out.replace(f"note {n}", f"note {n} {DIM}({note_name(n)}){RESET}")
+    if vel:
+        v = int(vel.group(1))
+        bars = "▁▂▃▄▅▆▇█"
+        out += f"  {bars[min(v, 127) * (len(bars) - 1) // 127]}"
+    return out
+
+
+def render(ev: dict) -> str:
+    code = _color(ev["event"])
+    ch = f"ch{ev['ch']:<2}" if ev["ch"] is not None else "   "
+    return (f"{DIM}{ev['source']:>6}{RESET}  "
+            f"\033[{code}m{BOLD}{ev['event']:<18}{RESET} "
+            f"{DIM}{ch}{RESET} {_render_data(ev['event'], ev['data'])}")
+
+
+def list_ports():
+    subprocess.run(["aseqdump", "-l"])
+
+
+# What to watch, best first. `aseqdump` with no -p subscribes to NOTHING and
+# prints "Waiting for data at port ..." — so midimon opened blind and PLN had to
+# wire it to Midi Through by hand every time ("had to wire it to midi through
+# myself", 2026-08-29). A monitor that does not monitor anything on open is a
+# monitor you have to repair before you can use it.
+#
+# The ORDER encodes what is actually interesting:
+#   1. the lcxl3-driver's translated stream — the numbers the CORPUS speaks, i.e.
+#      what you actually want to verify when a knob does the wrong thing.
+#   2. the raw v3 DAW port — the numbers the HARDWARE speaks, for when the
+#      question is "is the surface even sending".
+#   3. the raw v3 custom-mode port, for standalone work.
+#   4. Midi Through, the old default, kept last as a catch-all.
+WATCH_PREFERENCE = ("ParVagues LCXL3", "LCXL3 1 DAW", "LCXL3", "Launch Control XL",
+                    "Midi Through")
+
+
+def resolve_port() -> tuple[str | None, str]:
+    """(port_id, human_label) of the most interesting source present.
+
+    Matches CLIENT *and* PORT names: python-rtmidi registers the driver's client
+    as 'RtMidiOut Client' and puts 'ParVagues LCXL3' on the port, so a
+    client-only match misses exactly the stream we most want to watch.
+    """
+    try:
+        out = subprocess.run(["aseqdump", "-l"], capture_output=True,
+                             text=True, timeout=3).stdout
+    except (OSError, subprocess.SubprocessError):
+        return None, "aseqdump unavailable"
+    # `aseqdump -l` prints: "  20:1    LCXL3 1    LCXL3 1 DAW In"
+    rows = []
+    for line in out.splitlines():
+        tok = line.split()
+        if tok and ":" in tok[0] and tok[0].split(":")[0].isdigit():
+            rows.append((tok[0], line))
+    for want in WATCH_PREFERENCE:
+        for pid, line in rows:
+            if want in line:
+                return pid, f"{pid} · {want}"
+    return None, "no MIDI source found"
+
+
+def monitor(port: str | None, raw: bool):
+    label = ""
+    if not port:
+        port, label = resolve_port()
+    cmd = ["aseqdump"] + (["-p", port] if port else [])
+    where = label or (f"port {port}" if port else "NOTHING — nothing to watch")
+    print(f"{DIM}⚓ midimon — {where} · Ctrl-C to stop{RESET}", file=sys.stderr)
+    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, bufsize=1)
+    try:
+        for line in proc.stdout or []:
+            if raw:
+                sys.stdout.write(line); continue
+            ev = parse_line(line)
+            if ev:
+                print(render(ev), flush=True)
+            elif line.strip() and not line.startswith(("Source", "Waiting")):
+                print(f"{DIM}{line.rstrip()}{RESET}")
+    except KeyboardInterrupt:
+        pass
+    finally:
+        proc.terminate()
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("-p", "--port",
+                   help="source port (CLIENT:PORT), e.g. 28:0. Default: the most "
+                        "interesting source present — the lcxl3-driver's "
+                        "translated stream if it is up, else the raw surface, "
+                        "else Midi Through.")
+    ap.add_argument("-l", "--list", action="store_true", help="list ports and exit")
+    ap.add_argument("--raw", action="store_true", help="passthrough, no parsing")
+    a = ap.parse_args(argv)
+    if a.list:
+        return list_ports()
+    monitor(a.port, a.raw)
+
+
+if __name__ == "__main__":
+    main()

midistream.py

+"""midistream — fan live, parsed MIDI events out to web subscribers (SSE).
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+One `aseqdump` subprocess feeds N dashboard tabs. Reuses `midimon.parse_line` +
+`enrich` (one parsing source of truth). The reader runs only while someone is
+watching: started on first subscribe, stopped when the last subscriber leaves —
+so we don't hold a MIDI port open for nothing.
+
+`parse_ports` is pure (tested without ALSA).
+"""
+from __future__ import annotations
+
+import queue
+import re
+import subprocess
+import threading
+
+import midimon
+
+_PORT = re.compile(r"^\s*(\d+:\d+)\s+(.+?)\s{2,}(.+?)\s*$")
+
+# Which events are STATE and which are EVENTS. The distinction is the whole of
+# `coalesce`: a controller's older value is worthless the moment a newer one exists, but
+# a note you drop never happened. Getting this backwards would make the monitor lie
+# about what was played, which is worse than making it slow.
+_CONTINUOUS = ("control change", "pitchbend", "pitch bend", "aftertouch",
+               "channel aftertouch", "poly aftertouch", "control")
+
+
+def _coalesce_key(ev: dict):
+    """A key for events that supersede each other, or None if the event is discrete."""
+    e = (ev.get("event") or "").lower()
+    if not e.startswith(_CONTINUOUS):
+        return None
+    if e.startswith(("control change", "control")):
+        return "cc", ev.get("source"), ev.get("ch"), ev.get("controller")
+    return e.split()[0], ev.get("source"), ev.get("ch")
+
+
+def coalesce(events: list[dict]) -> list[dict]:
+    """Fold a batch: every discrete event survives, continuous ones keep only the last.
+
+    Order is preserved by overwriting in place at the first occurrence, so the monitor
+    still reads chronologically -- a superseded fader value is replaced where it stood,
+    not moved to the end.
+    """
+    out: list[dict] = []
+    at: dict = {}
+    for ev in events:
+        k = _coalesce_key(ev)
+        if k is None:
+            out.append(ev)
+        elif k in at:
+            out[at[k]] = ev
+        else:
+            at[k] = len(out)
+            out.append(ev)
+    return out
+
+
+def parse_ports(text: str) -> list[dict]:
+    """Parse `aseqdump -l` output into [{addr, client, port}] (header skipped)."""
+    out = []
+    for line in text.splitlines():
+        m = _PORT.match(line)
+        if m and ":" in m.group(1):           # header "Port  Client name…" has no digits:digits
+            out.append({"addr": m.group(1), "client": m.group(2).strip(),
+                        "port": m.group(3).strip()})
+    return out
+
+
+class MidiStream:
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._subs: set[queue.Queue] = set()
+        self._proc = None
+        self._port = None
+
+    def list_ports(self) -> list[dict]:
+        try:
+            r = subprocess.run(["aseqdump", "-l"], capture_output=True, text=True, timeout=5)
+        except (OSError, subprocess.SubprocessError):
+            return []
+        return parse_ports(r.stdout)
+
+    # A live monitor wants the PRESENT, not a complete history. 512 was chosen as
+    # "generous"; on a real surface it is a LATENCY BUDGET. PLN, 2026-08-21: "when i
+    # move faders up/down quickly, lags by 500ms+, almost 1s, behind ahah" — a fader
+    # sweep is ~400 CC/s, so a 512-deep queue is 1.3 s of backlog, and once it fills
+    # it STAYS full: drop-oldest keeps the buffer at capacity, so every event the
+    # browser renders is 512 events stale, forever. Deep buffers do not smooth a
+    # stream the consumer cannot keep up with; they just add a fixed delay to it.
+    #
+    # 64 bounds the backlog to ~160 ms at that rate, which is jitter absorption
+    # rather than a queue. The browser coalesces consecutive same-control events, so
+    # it now drains far faster than it fills and this should rarely be reached at all.
+    DEPTH = 64
+
+    def subscribe(self, port=None) -> queue.Queue:
+        q: queue.Queue = queue.Queue(maxsize=self.DEPTH)
+        with self._lock:
+            self._subs.add(q)
+            restart = self._proc is None or (port and port != self._port)
+        if restart:
+            self._start(port)
+        return q
+
+    def unsubscribe(self, q: queue.Queue):
+        with self._lock:
+            self._subs.discard(q)
+            stop = not self._subs
+        if stop:
+            self._stop()
+
+    # ── internals ───────────────────────────────────────────────────────
+    def _start(self, port):
+        self._stop()
+        cmd = ["aseqdump"] + (["-p", port] if port else [])
+        try:
+            self._proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, bufsize=1)
+        except OSError:
+            self._proc = None
+            return
+        self._port = port
+        threading.Thread(target=self._reader, args=(self._proc,), daemon=True).start()
+
+    def _reader(self, proc):
+        for line in proc.stdout or []:
+            ev = midimon.parse_line(line)
+            if not ev:
+                continue
+            self._fanout(midimon.enrich(ev))
+
+    def _fanout(self, ev: dict) -> None:
+        """Hand one event to every subscriber. Its own method so the drop policy is
+        testable without ALSA -- green tests on a parser prove nothing about the seam."""
+        with self._lock:
+            for q in self._subs:
+                try:
+                    q.put_nowait(ev)
+                except queue.Full:
+                    # Drop the OLDEST, not the newest. The original `except Full: pass`
+                    # kept 512 stale events and threw away the one that had just
+                    # happened, so a tab that stalled once showed frozen values forever
+                    # with no error anywhere. For MIDI state the newest event IS truth.
+                    try:
+                        q.get_nowait()
+                        q.put_nowait(ev)
+                    except (queue.Empty, queue.Full):
+                        pass
+
+    def _stop(self):
+        if self._proc:
+            try:
+                self._proc.terminate()
+            except OSError:
+                pass
+        self._proc = None
+        self._port = None

oscfeed.py

+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""oscfeed — the Tidal HL feed: SuperDirt's own /play packets, snoopable.
+
+Tidal does not talk to SuperDirt alone. `BootTidal.hs` mirrors every
+`superdirtShape` packet to a second target (port `VIZ_PORT`), so anything may
+listen to what the music is DOING — which orbit triggered, with which values —
+without touching the audio path, the way the Pulsar editor already reads its
+own highlight feed. This module is that listener for midiviz.
+
+Two halves, both importable without Qt or ALSA:
+
+* `decode_osc` — the smallest OSC decoder the packets need. Pure function,
+  unit-tested against hand-built bytes.
+* `HLFeed` — a UDP daemon thread with the same shape as midiviz's `Reader`
+  (bounded deque, drain, close): a live monitor wants the PRESENT, not a
+  backlog, so the same lesson as `midistream.DEPTH` applies.
+"""
+from __future__ import annotations
+
+import socket
+import struct
+import sys
+import threading
+import time
+from collections import deque
+
+# BootTidal.hs sends the mirror here. High enough to dodge everything
+# scsynth/sclang grab by default, referenced by nothing else in the repo.
+VIZ_PORT = 57130
+
+# Tidal's orbits are 0-based (`orbit: 0` is d1) and the rig runs 14 of them.
+MIN_ORBIT, MAX_ORBIT = 1, 14
+
+
+def _skip_pad(data: bytes, i: int) -> int:
+    """OSC blobs are 4-aligned; return the index past the padding."""
+    return i + ((4 - (i % 4)) % 4)
+
+
+def _read_string(data: bytes, i: int) -> tuple[str, int]:
+    end = data.index(b"\0", i)
+    return data[i:end].decode("utf-8", "replace"), _skip_pad(data, end + 1)
+
+
+def _read_blob(data: bytes, i: int) -> tuple[bytes, int]:
+    (n,) = struct.unpack_from(">i", data, i)
+    i += 4
+    return data[i:i + n], i + n + ((4 - ((i + n) % 4)) % 4)
+
+
+def decode_osc(data: bytes) -> list[dict]:
+    """One UDP datagram → the OSC messages inside it.
+
+    Tidal's Stream sends either bare messages or `#bundle`s (one timetag,
+    then length-prefixed elements). Returns [{"addr", "args"}] with `args`
+    the *named* argument pairs a /play carries: Tidal→SuperDirt arguments
+    are ("gain", 0.8) style name/value runs, so a flat dict is the shape
+    every consumer wants. Unnamed args land under "0", "1", …
+    """
+    out: list[dict] = []
+    stack = [data]
+    while stack:
+        cur = stack.pop(0)
+        if not cur:
+            continue
+        if cur.startswith(b"#bundle"):
+            i = 16                                    # "#bundle\0" + 8-byte timetag
+            while i + 4 <= len(cur):
+                (n,) = struct.unpack_from(">i", cur, i)
+                i += 4
+                if n:
+                    stack.append(cur[i:i + n])
+                i = _skip_pad(cur, i + n)
+            continue
+        try:
+            addr, i = _read_string(cur, 0)
+            tags, i = _read_string(cur, i)            # ",ifs…" — the comma is part of it
+            args: dict = {}
+            slot = 0
+            for t in tags[1:]:
+                if t == "i":
+                    (v,) = struct.unpack_from(">i", cur, i)
+                    i += 4
+                elif t == "f":
+                    (v,) = struct.unpack_from(">f", cur, i)
+                    i += 4
+                elif t == "d":
+                    (v,) = struct.unpack_from(">d", cur, i)
+                    i += 8
+                elif t == "s":
+                    v, i = _read_string(cur, i)
+                elif t == "b":
+                    v, i = _read_blob(cur, i)
+                else:
+                    v = None                          # T/F/I/N… carry no bytes
+                args[str(slot)] = v
+                slot += 1
+        except (ValueError, struct.error, IndexError):
+            continue                                  # one bad element, not the datagram
+        # A /play message's named pairs are positional: ("gain", v), ("orbit", v)…
+        # Re-fold every ("string", value) run into a dict without losing the
+        # originals — the names ARE the Tidal parameter names.
+        named: dict = {}
+        k = 0
+        while k < len(args):
+            key = args.get(str(k))
+            if isinstance(key, str) and str(k + 1) in args:
+                named[key] = args[str(k + 1)]
+                k += 2
+            else:
+                k += 1
+        out.append({"addr": addr, "args": named})
+    return out
+
+
+def play_orbit(msg: dict) -> int | None:
+    """A decoded message → its 1-based orbit number, or None if not a trigger.
+
+    Tidal 1.9's superdirtShape addresses its messages `/dirt/play`, not
+    `/play` — captured off the wire on this rig — and a decoder that only
+    accepted the documentation's address silently dropped every real packet
+    while every synthetic test passed. Accept both spellings."""
+    if msg["addr"] not in ("/play", "/dirt/play"):
+        return None
+    o = msg["args"].get("orbit")
+    if isinstance(o, float) and o == int(o):
+        o = int(o)
+    if not isinstance(o, int) or not (0 <= o <= MAX_ORBIT):
+        return None
+    return o + 1                                       # Tidal 0-based → d1-based
+
+
+class HLFeed:
+    """UDP listener on `VIZ_PORT`; a bounded deque the GUI drains each frame.
+
+    Same contract as midiviz.Reader — start/drain/close/alive/error — so the
+    tick can treat the two feeds identically. A missed datagram is a missed
+    frame of truth, never a queue: bounded like the Reader, for the same
+    recorded reason.
+    """
+
+    DEPTH = 512
+
+    def __init__(self, port: int = VIZ_PORT):
+        self.port = port
+        self.error: str | None = None
+        self.total = 0
+        self._q: deque[dict] = deque(maxlen=self.DEPTH)
+        self._lock = threading.Lock()
+        self._sock: socket.socket | None = None
+        self._stop = threading.Event()
+
+    def start(self) -> "HLFeed":
+        threading.Thread(target=self._run, daemon=True,
+                         name="oscfeed").start()
+        return self
+
+    def _run(self) -> None:
+        try:
+            self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            self._sock.bind(("127.0.0.1", self.port))
+            self._sock.settimeout(0.5)
+        except OSError as exc:
+            self.error = type(exc).__name__
+            return
+        while not self._stop.is_set():
+            try:
+                data, _ = self._sock.recvfrom(65535)
+            except socket.timeout:
+                continue
+            except OSError:
+                if not self._stop.is_set():
+                    self.error = "recv"
+                break
+            now = time.monotonic()
+            # One malformed datagram must not kill the feed the way an
+            # uncaught exception in a daemon thread dies silently: skip it.
+            try:
+                msgs = decode_osc(data)
+            except (ValueError, struct.error, IndexError):
+                continue
+            first = False
+            for msg in msgs:
+                o = play_orbit(msg)
+                if o is None:
+                    continue
+                with self._lock:
+                    self._q.append({"orbit": o, "t": now,
+                                    "params": msg["args"]})
+                    self.total += 1
+                    first = self.total == 1
+            if first:
+                # One line, once: the question "is Tidal even sending?" is
+                # answered by the journal, not by staring at a silent grid.
+                print("⚓ HL feed: first trigger received (d%d)" % o,
+                      file=sys.stderr, flush=True)
+
+    def alive(self) -> bool:
+        return self._sock is not None and self.error is None
+
+    def drain(self) -> list[dict]:
+        with self._lock:
+            if not self._q:
+                return []
+            out = list(self._q)
+            self._q.clear()
+        return out
+
+    def close(self) -> None:
+        self._stop.set()
+        if self._sock is not None:
+            try:
+                self._sock.close()
+            except OSError:
+                pass
+            self._sock = None

packaging/midiviz.desktop

+[Desktop Entry]
+Type=Application
+Name=midiviz
+GenericName=MIDI lens
+Comment=Live MIDI as glyph-rain — your control surface, watched
+Exec=/usr/bin/python3 /PATH/TO/YOUR/CHECKOUT/midiviz.py
+Icon=audio-midi
+Terminal=false
+Categories=AudioVideo;Audio;
+StartupWMClass=midiviz
+NoDisplay=false

packaging/midiviz.service

+[Unit]
+Description=midiviz (MIDI lens: the control surface, watched)
+# GUI unit: it must start only once the compositor's env (WAYLAND_DISPLAY,
+# XDG_RUNTIME_DIR) has been imported into the user manager, which is what
+# graphical-session.target marks. Bound to default.target instead, it would
+# race the session and fail with "could not connect to display".
+After=graphical-session.target
+PartOf=graphical-session.target
+# StartLimitIntervalSec=0 belongs in [Unit], not [Service] -- systemd silently
+# ignores it in the wrong section. 0 = no rate-limit window can ever
+# accumulate a start count and refuse to restart; paired with Restart=always.
+StartLimitIntervalSec=0
+
+[Service]
+Environment=PYTHONUNBUFFERED=1
+# Point this at YOUR checkout. No --spectro here on purpose: the spectrum
+# backdrop is an AUDIO consumer (a tap on the master bus), and this unit is
+# Restart=always -- putting the flag here would make the tap permanent at
+# login. It is a runtime choice: `s` in the window, or a hand-launched
+# --spectro.
+ExecStart=/usr/bin/python3 %h/where/you/cloned/midiviz/midiviz.py
+Restart=always
+# ...EXCEPT when the human closed it. Restart=always cannot distinguish "the
+# aseqdump pipe hit EOF" from "the user clicked the X", and treating the
+# second as the first makes the lens feel welded on: the window goes away and
+# comes back five seconds later. midiviz.py exits USER_CLOSE_EXIT=78 for a
+# deliberate close only (the X, Q, Esc), so this keeps the restart for every
+# failure mode while letting a person actually close the thing.
+# SuccessExitStatus keeps the deliberate close from showing the unit red.
+SuccessExitStatus=78
+RestartPreventExitStatus=78
+RestartSec=5
+
+[Install]
+WantedBy=graphical-session.target

pyproject.toml

+[project]
+name = "midiviz"
+version = "1.0.0"
+description = "Live MIDI as a picture, not a transcript — a lens over your MIDI control surface"
+readme = "README.md"
+license = { text = "GPL-3.0-or-later" }
+authors = [{ name = "Paul-Louis NECH" }]
+requires-python = ">=3.10"
+dependencies = [
+    "PySide6>=6.5",
+]
+keywords = ["midi", "visualization", "alsa", "launch-control-xl", "livecoding"]
+classifiers = [
+    "Development Status :: 5 - Production/Stable",
+    "Environment :: X11 Applications :: Qt",
+    "Intended Audience :: End Users/Desktop",
+    "License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)",
+    "Operating System :: POSIX :: Linux",
+    "Programming Language :: Python :: 3",
+    "Topic :: Multimedia :: Sound/Audio :: MIDI",
+]
+
+[project.urls]
+Homepage = "https://git.nech.pl/pln/midiviz"
+GitHub = "https://github.com/PLNech/midiviz"
+
+# The app is a single-module, sibling-import design (midiviz.py imports
+# midimon/midistream/lcxl_grid as top-level names), so it is not installed as
+# a library — run it from the checkout: `python3 midiviz.py`.
+[tool.pytest.ini_options]
+testpaths = ["tests"]

tests/test_midimon.py

+"""Pure-parser tests for midimon (no ALSA needed)."""
+# SPDX-License-Identifier: GPL-3.0-or-later
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+import midimon as M
+
+
+def test_note_name():
+    assert M.note_name(60) == "C4"
+    assert M.note_name(69) == "A4"
+    assert M.note_name(72) == "C5"
+    assert M.note_name(61) == "C#4"
+
+
+def test_parse_note_on():
+    ev = M.parse_line(" 28:0   Note on                 0, note 60, velocity 100")
+    assert ev["source"] == "28:0"
+    assert ev["event"] == "Note on"
+    assert ev["ch"] == 0
+    assert "note 60" in ev["data"]
+
+
+def test_parse_control_change():
+    ev = M.parse_line(" 28:0   Control change          5, controller 7, value 99")
+    assert ev["event"] == "Control change" and ev["ch"] == 5
+
+
+def test_parse_non_event_returns_none():
+    assert M.parse_line("Source  Event                  Ch  Data") is None
+    assert M.parse_line("Waiting for data at port 128:0.") is None
+    assert M.parse_line("") is None
+
+
+def test_render_contains_event_and_notename():
+    ev = M.parse_line(" 28:0   Note on                 0, note 60, velocity 100")
+    out = M.render(ev)
+    assert "Note on" in out and "C4" in out
+
+
+def test_enrich_note_and_velocity():
+    ev = M.enrich(M.parse_line(" 28:0   Note on                 0, note 60, velocity 100"))
+    assert ev["note"] == 60 and ev["note_name"] == "C4" and ev["velocity"] == 100
+
+
+def test_enrich_controller():
+    ev = M.enrich(M.parse_line(" 28:0   Control change          0, controller 74, value 64"))
+    assert ev["controller"] == 74 and ev["value"] == 64 and "note" not in ev

tests/test_midistream.py

+"""Pure tests for midistream: port parsing, and the coalescer (no ALSA).
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+The coalescer's one rule is the only thing standing between "the monitor is fast" and
+"the monitor lies": continuous controls are STATE and may be superseded, notes are
+EVENTS and may never be dropped. Get that backwards and a fast monitor silently loses
+what was played, which is strictly worse than a slow one.
+"""
+import queue
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+import midistream as MS
+
+SAMPLE = """ Port    Client name                      Port name
+  0:0    System                           Timer
+  0:1    System                           Announce
+ 14:0    Midi Through                     Midi Through Port-0
+ 28:0    nanoKONTROL2                     nanoKONTROL2 MIDI 1
+"""
+
+
+def test_parse_ports_skips_header_and_parses_rows():
+    ports = MS.parse_ports(SAMPLE)
+    assert {"addr": "28:0", "client": "nanoKONTROL2", "port": "nanoKONTROL2 MIDI 1"} in ports
+    assert all(":" in p["addr"] for p in ports)
+    assert not any(p["client"] == "Client name" for p in ports)  # header skipped
+    assert len(ports) == 4
+
+
+def test_parse_ports_empty():
+    assert MS.parse_ports("") == []
+
+
+# ------------------------------------------------------------------ coalesce
+
+def cc(controller, value, ch=0, source="24:0"):
+    return {"source": source, "event": "Control change", "ch": ch,
+            "controller": controller, "value": value, "data": ""}
+
+
+def note(n, vel=100, on=True, ch=0, source="24:0"):
+    return {"source": source, "event": "Note on" if on else "Note off", "ch": ch,
+            "note": n, "velocity": vel, "data": ""}
+
+
+def test_a_fader_sweep_collapses_to_its_final_value():
+    out = MS.coalesce([cc(77, v) for v in range(128)])
+    assert len(out) == 1 and out[0]["value"] == 127
+
+
+def test_every_note_survives_even_a_flood_of_controllers():
+    """Notes are what was PLAYED. Losing one to a fader sweep is unforgivable."""
+    batch = [note(60), *(cc(77, v) for v in range(100)), note(64), note(60, on=False)]
+    out = MS.coalesce(batch)
+    assert [e for e in out if "note" in e] == [note(60), note(64), note(60, on=False)]
+
+
+def test_distinct_controllers_do_not_collapse_into_each_other():
+    out = MS.coalesce([cc(77, 10), cc(78, 20), cc(77, 30)])
+    assert len(out) == 2
+    assert {e["controller"]: e["value"] for e in out} == {77: 30, 78: 20}
+
+
+def test_same_controller_on_a_different_channel_or_device_is_a_different_control():
+    out = MS.coalesce([cc(77, 1, ch=0), cc(77, 2, ch=1), cc(77, 3, source="28:0")])
+    assert len(out) == 3
+
+
+def test_chronological_order_is_preserved():
+    """A superseded value is replaced WHERE IT STOOD, so the monitor still reads as a
+    timeline rather than reshuffling history to the end."""
+    out = MS.coalesce([cc(77, 1), note(60), cc(77, 2), note(62)])
+    assert [e.get("controller", e.get("note")) for e in out] == [77, 60, 62]
+    assert out[0]["value"] == 2
+
+
+def test_coalesce_is_a_no_op_on_an_already_sparse_batch():
+    batch = [note(60), cc(77, 5), note(60, on=False)]
+    assert MS.coalesce(batch) == batch
+
+
+def test_unknown_event_kinds_are_treated_as_discrete():
+    """Unrecognised == keep. A coalescer that guesses wrong should lose nothing."""
+    weird = {"source": "24:0", "event": "System exclusive", "ch": None, "data": "F0"}
+    assert MS.coalesce([weird, weird]) == [weird, weird]
+
+
+# ---------------------------------------------------- the subscriber queue
+
+def test_a_full_queue_drops_the_OLDEST_event_not_the_newest():
+    """The original `except queue.Full: pass` kept 512 stale events and threw away the
+    one that had just happened, so a tab that stalled once froze forever."""
+    stream = MS.MidiStream()
+    q: queue.Queue = queue.Queue(maxsize=4)
+    stream._subs.add(q)
+    for v in range(10):
+        stream._fanout(cc(77, v))
+    got = [q.get_nowait()["value"] for _ in range(q.qsize())]
+    assert got == [6, 7, 8, 9], f"kept {got}; the newest value must survive"

tests/test_oscfeed.py

+"""Tests for the Tidal HL feed: OSC decode + which cells an orbit relates to.
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+No Qt, no ALSA, no network: `decode_osc` is a pure function tested against
+hand-built bytes — the exact packets Tidal's Stream sends (a `#bundle`
+containing a `/play` message whose arguments are name/value runs) and a couple
+of shapes it must survive (bare message, garbage, empty datagram).
+
+The cell relation is tested against `lcxl_grid` itself, so the test dies the
+day the grid changes shape instead of lying about it.
+"""
+import struct
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+import oscfeed                   # noqa: E402
+import sys as _sys               # noqa: E402
+_tools = str(Path(__file__).resolve().parent.parent.parent)
+if _tools not in _sys.path:
+    _sys.path.insert(0, _tools)
+import lcxl_grid as grid         # noqa: E402  lives in tools/, like midiviz's own import
+
+
+def _osc_msg(addr: str, pairs: list[tuple[str, int]]) -> bytes:
+    """Address + typetags + (string, int) pair args, padded the way OSC does.
+    A pair is: name padded to 4 (so 'orbit' → 8 bytes) + a 4-byte int."""
+    msg = addr.encode() + b"\0"
+    msg += b"\0" * ((4 - len(msg) % 4) % 4)
+    tags = "," + "si" * len(pairs)           # (string, int) runs — /play's shape
+    t = tags.encode() + b"\0"
+    msg += t + b"\0" * ((4 - len(t) % 4) % 4)
+    for name, value in pairs:
+        n = name.encode() + b"\0"
+        n += b"\0" * ((4 - len(n) % 4) % 4)
+        msg += n + struct.pack(">i", value)
+    return msg
+
+
+def _bundle(*elements: bytes) -> bytes:
+    out = b"#bundle\0" + b"\0" * 8           # timetag zero
+    for el in elements:
+        out += struct.pack(">i", len(el)) + el
+        out += b"\0" * ((4 - len(out) % 4) % 4)
+    return out
+
+
+def _named_pair(name: str, value: int) -> bytes:
+    n = name.encode() + b"\0"
+    n += b"\0" * ((4 - len(n) % 4) % 4)
+    return n + struct.pack(">i", value)
+
+
+def test_decode_play_bundle_finds_the_orbit():
+    # d4 → Tidal says orbit: 3; midiviz speaks d-numbers.
+    data = _bundle(_osc_msg("/play", [("orbit", 3)]))
+    msgs = oscfeed.decode_osc(data)
+    assert len(msgs) == 1
+    assert msgs[0]["addr"] == "/play"
+    assert oscfeed.play_orbit(msgs[0]) == 4
+
+
+def test_decode_bare_message_and_named_args():
+    data = _osc_msg("/play", [("orbit", 0), ("gain", 1)])
+    msgs = oscfeed.decode_osc(data)
+    assert oscfeed.play_orbit(msgs[0]) == 1
+    assert msgs[0]["args"]["gain"] == 1
+
+
+def test_non_play_and_garbage_yield_nothing():
+    assert oscfeed.play_orbit({"addr": "/hush", "args": {}}) is None
+    assert oscfeed.decode_osc(b"") == []
+    assert oscfeed.decode_osc(b"\x01\x02\x03") == []    # must not raise
+
+
+def test_orbit_bounds():
+    # orbit 14 (0-based 13) is the rig's last; anything past is not ours.
+    data = _bundle(_osc_msg("/play", [("orbit", 13)]))
+    assert oscfeed.play_orbit(oscfeed.decode_osc(data)[0]) == 14
+    data = _bundle(_osc_msg("/play", [("orbit", 99)]))
+    assert oscfeed.play_orbit(oscfeed.decode_osc(data)[0]) is None
+
+
+# ── the relation: which cells does a trigger light ─────────────────────────
+def _hl_cells(orbit):
+    """The widget's relation, re-derived the same way — tested against the
+    grid's own tables rather than against a copy of the widget's cache."""
+    ccs = set(grid.orbit_home(orbit).values())
+    for role, fam_of in (("family_filter", grid.filter_family),
+                         ("family_mute", grid.mute_family)):
+        n = fam_of(orbit)
+        ccs |= {cc for cc, (r, who) in grid.CC_ROLE.items()
+                if r == role and who == n}
+    return ccs
+
+
+def test_d1_trigger_lights_its_column_and_both_families():
+    # d1 (the kick): its own fx/level/gate cells, plus C1 (all-percs filter)
+    # and F1 (kick-alone mute) — the two family controls that act on it.
+    cells = {_: grid.CC_TO_CELL.get(_) for _ in _hl_cells(1)}
+    assert cells[grid.CELL_TO_CC[("C", 1)]] == ("C", 1)   # gF1, the perc filter
+    assert cells[grid.CELL_TO_CC[("F", 1)]] == ("F", 1)   # gM1, the kick mute
+    for row, col in (("B", 1), ("D", 1), ("E", 1)):       # its own column
+        assert (row, col) in cells.values()
+
+
+def test_every_d_orbit_maps_to_at_least_one_control():
+    # An orbit with NO related cells would make the HL feed a no-op that
+    # still LOOKS wired — the worst failure shape. d1-d12 own a full column
+    # plus two family controls; d13/d14 (the 14-orbit boot's tail) own only
+    # their family bloc, which is still one honest thing to show.
+    for o in range(1, 15):
+        assert len(_hl_cells(o)) >= 1, f"d{o} has no related controls"
+    for o in range(1, 13):
+        assert len(_hl_cells(o)) >= 3, f"d{o} has too few related controls"
+
+
+def test_widget_relation_matches_grid_derivation():
+    """The widget caches `_hl_cells` — the cache must never drift from a fresh
+    derivation, which is the same two-copies-one-truth lesson the port
+    preference already taught."""
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+    import midiviz as V
+    # The method hangs off the widget class; derive it without a Qt app.
+    cls = None
+    for obj in vars(V).values():
+        if isinstance(obj, type) and hasattr(obj, "_hl_cells"):
+            cls = obj
+            break
+    if cls is None:
+        return
+    holder = object.__new__(cls)
+    holder._hl_cells_cache = {}
+    for o in (1, 4, 7, 11):
+        assert set(holder._hl_cells(o)) == _hl_cells(o)

ui/parvagues.svg

+<svg xmlns="http://www.w3.org/2000/svg" width="128" height="128" viewBox="0 0 128 128">
+  <rect width="128" height="128" rx="28" fill="#0a0a0a"/>
+  <g fill="none" stroke-linecap="round">
+    <path d="M10 78 Q 26 52 42 78 T 74 78 T 106 78 T 138 78"
+          stroke="#7c5cff" stroke-width="6" opacity="0.55"/>
+    <path d="M10 62 Q 26 34 42 62 T 74 62 T 106 62 T 138 62"
+          stroke="#d900ff" stroke-width="8"/>
+  </g>
+</svg>