Commit 6ec3ca09 by PLN (Algolia)

gig-log: upgrade to the better MIDI port while bound, not only when it dies

Answering 'now if i reboot, will gig log be part of any parvagues run?' — yes,
gig-log is enabled and starts at login. But lcxl3-driver is linked, not
enabled, by the on-demand policy rig_units.py owns. So at boot this reader
binds the board's own DAW port and the driver appears later, and the loop
re-resolved only once aseqdump exited. 24:1 does not exit, so a whole set
would have been recorded from the raw board.

Translating the numbers is not enough on its own: with the driver up, rows B
and C are switched to RELATIVE, so the raw port carries encoder DELTAS, and
writing those down as absolute values would be a confident wrong record — in
the one file whose entire job is to be trustworthy after the fact.

So a bound reader now asks, between events and at most every 15s, whether
something better appeared, and rebinds if so. 'Better' is preference_rank
against the authored order, not a judgement, and an unplaceable name ranks
last so we can never downgrade into it. A port pinned with --midi-port is
never second-guessed.
parent fd2e5001
......@@ -526,6 +526,19 @@ SEQ_PREFERENCE: tuple[str, ...] = (
)
def preference_rank(name: str | None) -> int:
"""Where a matched name sits in SEQ_PREFERENCE; lower is better.
Exists so "a better port appeared" is a comparison rather than a judgement.
An unknown name ranks last, so we never downgrade to something we cannot
place.
"""
try:
return SEQ_PREFERENCE.index(name) # type: ignore[arg-type]
except ValueError:
return len(SEQ_PREFERENCE)
def find_seq_port(want: str | None = None) -> tuple[str | None, str | None]:
"""(addr, matched name) for the best MIDI source, re-resolved every connect.
......@@ -576,6 +589,11 @@ class MidiReader(threading.Thread):
self._proc: subprocess.Popen | None = None
self._stop = threading.Event()
# How often a bound reader asks whether something better turned up. One
# `aseqdump -l`, checked only between events, so an idle surface costs
# nothing and a busy one checks no more often than this.
UPGRADE_S = 15.0
@staticmethod
def available() -> bool:
return shutil.which("aseqdump") is not None
......@@ -669,10 +687,30 @@ class MidiReader(threading.Thread):
except OSError:
return
self.alive = True
# UPGRADE WHILE BOUND, not only when the stream dies (2026-09-23).
# Units come up in whatever order systemd likes, and lcxl3-driver is
# `linked` not `enabled` -- so after a reboot this reader binds the
# board's own DAW port and the driver appears LATER. The old loop
# re-resolved only once aseqdump exited, and 24:1 does not exit, so
# a whole set would have been logged from the raw board.
#
# Translation fixes the NUMBERS but not the dialect of the VALUE:
# with the driver up, rows B and C are switched to RELATIVE, so the
# raw port carries encoder DELTAS. Logging those as absolute values
# would be a confident wrong record -- the same class of bug as
# everything else found today, in the one file whose whole job is to
# be trustworthy after the fact.
last_check = time.time()
for line in self._proc.stdout:
if self._stop.is_set():
break
self.feed(line, time.time())
now = time.time()
if not self.fixed and now - last_check >= self.UPGRADE_S:
last_check = now
_addr, better = find_seq_port()
if preference_rank(better) < preference_rank(self.matched):
break # rebind to the better one
self.alive = False
if self._stop.wait(3.0):
return
......
......@@ -135,3 +135,45 @@ def test_the_bind_is_recorded_so_a_future_reader_need_not_guess():
"nothing records WHICH port was bound — and the nine silent sessions "
"happened precisely because nothing recorded that the answer was none")
assert '"xlate"' in src
# ── upgrading to a better port while already bound ─────────────────────────
# PLN, 2026-09-23: "now if i reboot, will gig log be part of any parvagues
# run?" — yes, gig-log is `enabled`. But lcxl3-driver is `linked`, so at boot
# gig-log binds the board and the driver appears later, and the old loop only
# re-resolved once aseqdump EXITED. 24:1 does not exit.
def test_the_rank_makes_better_a_comparison_not_a_feeling():
assert GL.preference_rank("ParVagues LCXL3") < GL.preference_rank("LCXL3 1 DAW")
assert GL.preference_rank("LCXL3 1 DAW") < GL.preference_rank("Launch Control XL")
def test_an_unplaceable_name_ranks_last_so_we_never_downgrade_into_it():
assert GL.preference_rank("Some USB Thing") == len(GL.SEQ_PREFERENCE)
assert GL.preference_rank(None) == len(GL.SEQ_PREFERENCE)
# ...and therefore never beats anything we CAN place.
for known in GL.SEQ_PREFERENCE:
assert GL.preference_rank(known) < GL.preference_rank("Some USB Thing")
def test_the_driver_appearing_later_beats_the_board():
"""The reboot case, stated as the timeline that produces it."""
assert GL.preference_rank("ParVagues LCXL3") < GL.preference_rank("LCXL3 1 DAW")
def test_a_pinned_port_is_never_upgraded_away_from():
"""--midi-port is a human saying 'this one'. Second-guessing that would make
the flag a suggestion."""
src = (TOOLS / "gig-log.py").read_text()
loop = src[src.index("# UPGRADE WHILE BOUND"):src.index("self.alive = False", src.index("# UPGRADE WHILE BOUND"))]
assert "not self.fixed" in loop
def test_the_upgrade_check_is_between_events_not_on_a_timer():
"""Cheap by construction: it runs in the read loop, so an idle surface costs
nothing at all and a busy one is still bounded by UPGRADE_S."""
src = (TOOLS / "gig-log.py").read_text()
loop = src[src.index("# UPGRADE WHILE BOUND"):src.index("self.alive = False", src.index("# UPGRADE WHILE BOUND"))]
assert "for line in self._proc.stdout" in loop
assert "UPGRADE_S" in loop
assert GL.MidiReader.UPGRADE_S >= 5.0, "too eager: this spawns aseqdump -l"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment