Commit d1947eef by PLN (Algolia)

feat(lcxl3): the OLED footer becomes a REC clock while Ardour records

Third line of the home screen: 'ParVagues' at rest; 'REC mm:ss' while the
gig-log k:'rec' stream says audio is landing on disk (tail-read of the newest
gig jsonl, same style as last_eval); '?REC' when the files and OSC lenses
disagree — a timer the performer must not trust blindly is marked, not hidden.
Stale or absent signal degrades to the plain footer, never crashes a redraw.
Pure decision function rec_line() + tests.
parent abad7515
......@@ -124,6 +124,14 @@ REPO = pathlib.Path(__file__).resolve().parent.parent
TRACK_FILE = pathlib.Path(os.path.expanduser("~/.cache/parvagues/current-track"))
EVAL_FILE = pathlib.Path(os.path.expanduser("~/.cache/parvagues/eval-events.jsonl"))
# Same env override + default gig-log.py itself uses (its LOG_DIR) — the home
# screen's REC timer must agree with `gig-log.py status` about which file is
# "the running session", not run its own second discovery down a fork.
GIG_LOG_DIR = pathlib.Path(os.environ.get(
"GIG_LOG_DIR", os.path.expanduser("~/.local/share/parvagues/gig-log")))
REC_STALE_S = 3.0 # a `k:"rec"` line older than this is a dead heartbeat,
# not a recording — gig-log down must read as "not recording"
def read_current_track() -> str | None:
try:
......@@ -182,6 +190,77 @@ def last_eval(path: pathlib.Path) -> float:
return 0.0
def newest_gig_log(d: pathlib.Path = GIG_LOG_DIR) -> pathlib.Path | None:
"""The gig-log file a live `record` session is writing right now, or None.
Mirrors gig-log.py's own `newest_log()` (newest `gig-*.jsonl` by mtime) so
the two tools can never disagree about which file "the session" means —
same doctrine as `resolve_track` agreeing with lcxl-leds.py. A missing dir,
or a file vanishing mid-stat (rotation), is not an error: the REC line just
falls back to the everyday drip.
"""
try:
logs = sorted(d.glob("gig-*.jsonl"), key=lambda p: p.stat().st_mtime)
except OSError:
return None
return logs[-1] if logs else None
def last_rec(path: pathlib.Path) -> dict | None:
"""Most recent `k:"rec"` record from the tail of a gig-log file, or None.
Same shape as last_eval: append-only, all-night log; only the last "rec"
line is live. 4 KB of tail comfortably covers gig-log's densest tick mix
(s/cc/rec at 1 Hz) without ever reading more than the newest few lines.
"""
try:
with path.open("rb") as fh:
fh.seek(0, os.SEEK_END)
back = min(fh.tell(), 4096)
fh.seek(-back, os.SEEK_END)
lines = fh.read().decode("utf-8", "replace").splitlines()
except OSError:
return None
for line in reversed(lines):
try:
ev = json.loads(line)
except ValueError:
continue
if isinstance(ev, dict) and ev.get("k") == "rec":
return ev
return None
def rec_line(rec: dict | None, now: float,
stale_s: float = REC_STALE_S) -> str | None:
"""The home screen's third line while Ardour is actually recording, else
None (meaning: keep the everyday "ParVagues" drip).
Pure: (most-recent `k:"rec"` dict off the gig-log tail, wall-clock now) ->
optional display string. `rec` is None whenever the lens is down for any
reason (no log dir, no file yet, empty/malformed tail) — that reads as
"not recording", never as a guess, per spec: gig-log down counts as not
recording. A record older than `stale_s` is a dead heartbeat, same
treatment. `el` (elapsed seconds, the files lens's own count) renders
mm:ss, or h:mm:ss past one hour. `conflict: true` — the two rec lenses
(files vs. Ardour's OSC) disagree — prefixes "?": the performer must not
trust the timer blindly.
"""
if not isinstance(rec, dict) or not rec.get("rec"):
return None
t = rec.get("t")
if not isinstance(t, (int, float)) or abs(now - t) > stale_s:
return None
el = rec.get("el")
if not isinstance(el, (int, float)) or el < 0:
return None
total = int(el)
h, rem = divmod(total, 3600)
m, s = divmod(rem, 60)
stamp = f"{h}:{m:02d}:{s:02d}" if h else f"{m:02d}:{s:02d}"
return f"{'?' if rec.get('conflict') else ''}REC {stamp}"
def parse_tempo_knob(text: str) -> tuple[int, float, float] | None:
"""(v2 cc, lo BPM, hi BPM) when the track's tempo IS a knob.
......@@ -959,8 +1038,25 @@ class Driver:
el = int(time.time() - self.track_started)
return f"{self.current_bpm():.0f}BPM {el // 60}:{el % 60:02d}"
def _home_footer(self) -> str:
"""Third home-screen line: "ParVagues", or "REC h:mm:ss" while the
gig-log's tail confirms Ardour is actually recording.
Every failure mode here — no GIG_LOG_DIR, no log file yet, a stale or
malformed tail — must degrade silently to the everyday drip. A broken
lens freezing a false REC on screen would be worse than no lens at all,
so this can never raise into the redraw tick.
"""
try:
path = newest_gig_log()
rec = last_rec(path) if path is not None else None
return rec_line(rec, time.time()) or "ParVagues"
except Exception:
return "ParVagues"
def oled_home(self) -> None:
t, pmt, val = self.lang.home_lines(self.track_name, self._home_detail())
t, pmt, val = self.lang.home_lines(self.track_name, self._home_detail(),
self._home_footer())
self.surface.configure_display(L.TGT_STATIONARY, 2)
for field, line in enumerate((t, pmt, val)):
self.surface.set_text(L.TGT_STATIONARY, field, line)
......@@ -1069,7 +1165,8 @@ class Driver:
def ghost_write(self) -> None:
try:
age = time.time() - self._oled_up_at
t, pmt, val = self.lang.home_lines(self.track_name, self._home_detail())
t, pmt, val = self.lang.home_lines(self.track_name, self._home_detail(),
self._home_footer())
home = (t, pmt, val)
cc = getattr(self, "_last_cc", None)
if age >= 1.2:
......@@ -1275,9 +1372,11 @@ class Driver:
self.follow_tick()
last_follow = now
if self.lang and now - last_home > 2.0:
# tick the lap clock on home row 2; a field write on the
# stationary target needs no re-summon
# tick the lap clock on home row 2, and the REC timer (if any)
# on row 3; a field write on the stationary target needs no
# re-summon
self.surface.set_text(L.TGT_STATIONARY, 1, self._home_detail())
self.surface.set_text(L.TGT_STATIONARY, 2, self._home_footer())
self.ghost_write()
last_home = now
if self.lang and now - last_labels > 20.0:
......
......@@ -333,11 +333,14 @@ def djf_readout(value: int) -> str:
return s
def home_lines(track: str, detail: str = "") -> tuple[str, str, str]:
def home_lines(track: str, detail: str = "",
footer: str = "ParVagues") -> tuple[str, str, str]:
"""The stationary track HUD, rows by PLN's chair ergonomics: the name, then
the row that matters (tempo + time played — the overstretch check), and the
ParVagues drip at the bottom where the bezel eats it anyway."""
return (track.upper()[:16], detail[:16], "ParVagues")
the row that matters (tempo + time played — the overstretch check), and
normally the ParVagues drip at the bottom where the bezel eats it anyway.
While gig-log confirms Ardour is recording, the caller passes a "REC
h:mm:ss" footer instead — the one moment that corner earns its keep."""
return (track.upper()[:16], detail[:16], footer[:16])
# ---------------------------------------------------------------- mapping ----
......
......@@ -161,6 +161,53 @@ check(all(r[3] for r in rows if r not in djf),
rows = drv.label_table(m, {}, set(m.values()), lang, grid, overlay_only=True)
check(not any(r[3] for r in rows), "--oled-overlay suppresses every native display")
# ---- home_lines: the third row defaults to the drip, footer overrides it --
check(lang.home_lines("rose_rouge", "120BPM 3:45") == ("ROSE_ROUGE", "120BPM 3:45", "ParVagues"),
"no footer arg -> unchanged behaviour, the everyday drip")
check(lang.home_lines("x", "y", "REC 01:23") == ("X", "y", "REC 01:23"),
"an explicit footer replaces the drip verbatim")
check(len(lang.home_lines("x", "y", "?REC 1:02:03")[2]) <= 16,
"a footer must be clipped to the 16-char panel like the other two rows")
# ---- rec_line: pure (rec dict, now) -> optional "REC h:mm:ss" ---------------
NOW = 1_000_000.0
check(drv.rec_line(None, NOW) is None, "no rec record at all -> keep the drip")
check(drv.rec_line({}, NOW) is None, "an empty/malformed record -> keep the drip")
check(drv.rec_line({"k": "rec", "rec": False, "t": NOW, "el": 12}, NOW) is None,
"rec:false -> keep the drip even with a fresh timestamp")
check(drv.rec_line({"k": "rec", "rec": True, "t": NOW, "el": 83}, NOW) == "REC 01:23",
"83s elapsed -> mm:ss, zero-padded")
check(drv.rec_line({"k": "rec", "rec": True, "t": NOW, "el": 3723}, NOW) == "REC 1:02:03",
"past one hour -> h:mm:ss")
check(drv.rec_line({"k": "rec", "rec": True, "t": NOW, "el": 5, "conflict": True}, NOW)
== "?REC 00:05",
"conflict:true prefixes '?' — the two lenses disagree, don't trust blindly")
check(drv.rec_line({"k": "rec", "rec": True, "t": NOW - 10, "el": 5}, NOW) is None,
"a 10s-old rec line is a dead heartbeat -> keep the drip, never freeze a REC")
check(drv.rec_line({"k": "rec", "rec": True, "t": NOW, "el": -1}, NOW) is None,
"a negative elapsed is malformed, not a countdown -> keep the drip")
check(drv.rec_line({"k": "rec", "rec": True, "el": 5}, NOW) is None,
"a record with no timestamp at all can never be judged fresh -> keep the drip")
# ---- last_rec: tail-reads the newest k:"rec" line, ignores everything else --
import tempfile
with tempfile.TemporaryDirectory() as td:
p = pathlib.Path(td) / "gig-fake.jsonl"
p.write_text(
'{"k": "s", "t": 1.0}\n'
'{"k": "rec", "t": 2.0, "rec": true, "el": 1}\n'
'not even json\n'
'{"k": "cc", "t": 3.0}\n'
'{"k": "rec", "t": 4.0, "rec": true, "el": 3}\n'
)
got = drv.last_rec(p)
check(got is not None and got.get("t") == 4.0,
f"must return the LAST rec line, not the first or a non-rec one: {got!r}")
check(drv.last_rec(pathlib.Path(td) / "gone.jsonl") is None,
"a missing file must degrade to None, never raise")
check(drv.newest_gig_log(pathlib.Path(td) / "no-such-dir") is None,
"a missing gig-log directory must degrade to None, never raise")
if FAILED:
print(f"{len(FAILED)} FAILED:")
for f in FAILED:
......
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