Skip to content

T
Tidal
Commit d1b21c94 by PLN (Algolia)
Options

feat(gig-log): report where a control is PARKED, not just how far it travelled 

The surface table said cc 51 moved 533 times somewhere between 1 and 127. That
is a biography, not a state, and it cannot answer the only question you ask a
log at 3am: "why is that orbit silent?". Only the LAST value can.

The data was already in the file — MidiReader.feed() has always coalesced to
count/first/last/min/max per control per second, so `v1` is the value the knob
was left at. The report simply never printed it. This is a formatting change to
capture that already happened.

Three additions:

* `report` gains a `left at` column, and for the three DJ filters it prints what
  that value MEANS in hertz. Transcribing BootTidal.hs:376 turned up something
  worth its own note: for v <= 0.5 the gDJF expression reduces to
  lpf = 180 + 39640*v, i.e. LINEAR IN HERTZ. Pitch perception is logarithmic, so
  that knob spends nearly all its travel in the top two octaves and crosses the
  entire audible bottom in the last ~2% — it feels inert, then collapses. That is
  one gesture producing both "it went quiet" and "it sounds lpf'd", which is
  exactly the pair of symptoms #79 was filed with.

* `gig-log.py controls` — the whole surface, by CC, with each value translated:
  filters to Hz, gMute/gMask to "mutes N% of cycles", panic to armed/clear. It
  ends with an explicit list of anything parked somewhere that silences or thins
  the sound, because a table you have to interpret under stage lights is a table
  you will misread.

* `report --from/--to`, wall clock or +M:SS. The recorder ran 10h48m of which
  ~90 minutes was playing and eight hours was tooling; every aggregate therefore
  described the wrong thing. Windowing has one non-obvious requirement, and it
  gets a test of its own: xrun and throttle are RUNNING TOTALS, so a naive slice
  reports the whole session's count inside the window — wrong, and wrong high.
  slice_session() rebases them to the window.

VALIDATION — it immediately paid for itself by killing two hypotheses:
    49  51  gF1  lpf 16098 Hz — open
    50  59  gF2  lpf 18595 Hz — open
    51  80  gF3  lpf 20000 Hz — open
    41   0  gMask   gates 0% of cycles
    73/74/75 0      mutes 0% of cycles
#79's two leading suspects were "a DJ filter parked below centre" and "a mask
left engaged". Both are now dead, from a file, with no rig and nobody's ears.

83 tests (was 71). The new ones pin the BootTidal arithmetic so it cannot drift
from the Haskell, the last-value ordering rule (latest TIMESTAMP wins, not file
order), the cumulative rebase, and midnight-crossing wall-clock parsing.
parent 55a38ba6
Showing with 386 additions and 16 deletions
tools/gig-log.py
......@@ -645,6 +645,145 @@ def hms(seconds: float) -> str:
else f"{s // 60:d}:{s % 60:02d}"
# --------------------------------------------------------------------------
# What a PARKED control means
# --------------------------------------------------------------------------
# A move count and a min-max range cannot answer "why is that orbit silent".
# Only the LAST value can, so the report carries it — and for the controls whose
# low end is dangerous, it carries what that value MEANS in Hz.
#
# BootTidal.hs:376
# gDJF ch = (# lpf (range 180 20000
# (fmap (\v -> 1 - 2 * max 0 (0.5 - v)) (orDef 0.5 ch))))
#
# For v <= 0.5 the inner fmap is just `2*v`, so
# lpf = 180 + 19820 * 2v = 180 + 39640 * v
# which is LINEAR IN HERTZ. That matters more than it looks: pitch perception is
# logarithmic, so a linear-in-Hz sweep spends nearly all its travel in the top
# two octaves and crosses the entire audible bottom in the last ~2% of the knob.
# The knob therefore feels inert for most of its range and then collapses to
# near-silence right at the end — which is exactly one gesture producing BOTH
# "it went quiet" AND "it sounds lpf'd" (#79).
DJF_CC = (49, 50, 51) # gF1 gF2 gF3
MASK_CC = 41 # gMask
MUTE_CC = (73, 74, 75) # gMute1 gMute2 gMute3
PANIC_CC = 93
ARDOUR_CC = range(77, 85) # MIDI-learned to Ardour track gains — never seed these
CONTROL_ROLE = {
**{cc: f"gF{i + 1} DJ filter (lpf)" for i, cc in enumerate(DJF_CC)},
MASK_CC: "gMask someCyclesBy gate",
**{cc: f"gMute{i + 1} someCyclesBy mute" for i, cc in enumerate(MUTE_CC)},
PANIC_CC: "gPanic global gain kill",
**{cc: "Ardour track gain" for cc in ARDOUR_CC},
}
def djf_lpf(value: int) -> float:
"""Hz that gDJF applies for a raw CC value. See the note above."""
v = max(0, min(127, int(value))) / 127.0
x = 1.0 - 2.0 * max(0.0, 0.5 - v)
return 180.0 + (20000.0 - 180.0) * x
def djf_verdict(hz: float) -> str:
"""How a parked DJ filter will SOUND. Thresholds are by ear, not by maths."""
if hz < 400:
return "⚠⚠ NEAR-SILENT"
if hz < 1500:
return "⚠ dark"
if hz < 6000:
return "filtered"
return "open"
def surface_state(events: list[dict]) -> dict[int, dict]:
"""Fold CC records per control, KEEPING the last value.
`v1` (the final value inside each coalescing window) is already in the log —
the report simply never showed it. The last record's `v1` is where the knob
was left, which is the one number that answers "what is this control doing
right now".
"""
agg: dict[int, dict] = {}
for e in events:
if e.get("k") != "cc" or "cc" not in e:
continue
cc = e["cc"]
a = agg.get(cc)
if a is None:
a = agg[cc] = {"n": 0, "lo": 127, "hi": 0, "first": e["t"],
"last": e["t"], "s": 0, "v": e.get("v1"),
"v_t": e["t"]}
a["n"] += e.get("n", 1)
a["s"] += 1
a["lo"] = min(a["lo"], e.get("lo", 0))
a["hi"] = max(a["hi"], e.get("hi", 0))
a["first"] = min(a["first"], e["t"])
a["last"] = max(a["last"], e["t"])
# Guard the ordering rather than trusting file order: a torn or
# out-of-order tail must not silently become "where the knob is".
if e["t"] >= a["v_t"]:
a["v_t"] = e["t"]
a["v"] = e.get("v1")
return agg
# Counters that are RUNNING TOTALS. Windowing the session without rebasing these
# reports the whole session's count inside the window — wrong, and wrong HIGH.
CUMULATIVE = ("xrun", "thr_core", "thr_pkg")
def slice_session(samples: list[dict], events: list[dict], marks: list[dict],
t_from: float | None, t_to: float | None
) -> tuple[list[dict], list[dict], list[dict]]:
"""Window the session AND rebase its cumulative counters to the window."""
def win(rows):
return [r for r in rows
if (t_from is None or r["t"] >= t_from)
and (t_to is None or r["t"] <= t_to)]
sl = win(samples)
if sl and (t_from is not None or t_to is not None):
b = sl[0]
base = {k: (b.get(k) or 0) for k in CUMULATIVE}
base_by = dict(b.get("xrun_by") or {})
out = []
for s in sl:
r = dict(s)
for k in CUMULATIVE:
if s.get(k) is not None:
r[k] = max(0, s[k] - base[k])
if s.get("xrun_by"):
r["xrun_by"] = {k: max(0, v - base_by.get(k, 0))
for k, v in s["xrun_by"].items()}
out.append(r)
sl = out
return sl, win(events), win(marks)
def parse_when(spec: str, t0: float) -> float:
"""`+M:SS` / `+H:MM:SS` = offset from session start; `HH:MM[:SS]` = wall clock.
Wall clock is the useful one, because that is what the Ardour take and PLN's
memory are both stamped in.
"""
s = spec.strip()
if s.startswith("+"):
off = 0.0
for part in s[1:].split(":"):
off = off * 60 + float(part)
return t0 + off
parts = [int(x) for x in s.split(":")]
while len(parts) < 3:
parts.append(0)
base = datetime.fromtimestamp(t0)
want = base.replace(hour=parts[0], minute=parts[1], second=parts[2],
microsecond=0)
ts = want.timestamp()
return ts + 86400 if ts < t0 else ts # the session crossed midnight
def load(path: Path) -> tuple[dict, list[dict], list[dict], list[dict], dict]:
hdr, samples, ccs, marks, end = {}, [], [], [], {}
with path.open() as f:
......@@ -670,12 +809,27 @@ def load(path: Path) -> tuple[dict, list[dict], list[dict], list[dict], dict]:
return hdr, samples, ccs, marks, end
def cmd_report(path: Path) -> int:
def cmd_report(path: Path, since: str | None = None,
until: str | None = None) -> int:
hdr, samples, events, marks, end = load(path)
if not samples:
print(f"gig-log: {path.name} has no samples yet", file=sys.stderr)
return 1
t0 = hdr.get("t") or samples[0]["t"]
win = None
if since or until:
# Measure the SET, not the corpus: a recorder that ran all night mixes
# eight hours of tooling into ninety minutes of playing, and every
# aggregate then describes the wrong thing.
t_from = parse_when(since, t0) if since else None
t_to = parse_when(until, t0) if until else None
samples, events, marks = slice_session(samples, events, marks,
t_from, t_to)
if not samples:
print("gig-log: that window contains no samples", file=sys.stderr)
return 1
win = (t_from, t_to)
t0 = samples[0]["t"]
dur = samples[-1]["t"] - t0
W = 74
p = print
......@@ -683,6 +837,10 @@ def cmd_report(path: Path) -> int:
p(f" GIG LOG {path.name}")
p("=" * W)
p(f" started {hdr.get('iso', '?')} (align the Ardour take to this wall clock)")
if win:
p(f" WINDOW {datetime.fromtimestamp(t0):%H:%M:%S}"
f" → {datetime.fromtimestamp(samples[-1]['t']):%H:%M:%S}"
f" (counters rebased to this window)")
p(f" duration {hms(dur)} in {len(samples)} samples "
f"@ {hdr.get('period', 1)}s")
p(f" perf mode {hdr.get('mode')} gov={hdr.get('gov')} "
......@@ -792,23 +950,19 @@ def cmd_report(path: Path) -> int:
if not ccs and not notes:
p(" no MIDI recorded (controller absent, or aseqdump unavailable)")
else:
agg: dict[int, dict] = {}
for e in ccs:
a = agg.setdefault(e["cc"], {"n": 0, "lo": 127, "hi": 0,
"first": e["t"], "last": e["t"], "s": 0})
a["n"] += e["n"]
a["s"] += 1
a["lo"] = min(a["lo"], e["lo"])
a["hi"] = max(a["hi"], e["hi"])
a["first"] = min(a["first"], e["t"])
a["last"] = max(a["last"], e["t"])
agg = surface_state(ccs)
p(f" {len(agg)} controls touched, {sum(a['n'] for a in agg.values())} raw "
f"events, logged as {len(ccs)} lines "
f"({sum(a['n'] for a in agg.values()) / max(1,len(ccs)):.0f}x coalescing)")
p(f" {'cc':>4} {'moves':>6} {'range':>9} {'first':>7} {'last':>7} active")
p(f" {'cc':>4} {'moves':>6} {'range':>9} {'left at':>7} "
f"{'first':>7} {'last':>7} active")
for cc, a in sorted(agg.items(), key=lambda kv: -kv[1]["n"]):
flag = " ⚠ARDOUR-OWNED" if 77 <= cc <= 84 else ""
p(f" {cc:>4} {a['n']:>6} {a['lo']:>4}-{a['hi']:<4} "
flag = " ⚠ARDOUR-OWNED" if cc in ARDOUR_CC else ""
if cc in DJF_CC and a["v"] is not None:
hz = djf_lpf(a["v"])
flag = f" lpf {round(hz):>5} Hz {djf_verdict(hz)}"
v = "?" if a["v"] is None else str(a["v"])
p(f" {cc:>4} {a['n']:>6} {a['lo']:>4}-{a['hi']:<4} {v:>7} "
f"{hms(a['first']-t0):>7} {hms(a['last']-t0):>7} {a['s']:>5}s{flag}")
if notes:
p(f" {len(notes)} note events "
......@@ -874,6 +1028,70 @@ def cmd_install(enable: bool = True) -> int:
return 0
def cmd_controls(path: Path, since: str | None = None,
until: str | None = None) -> int:
"""Where is every control PARKED — the 3am question when an orbit is silent.
Deliberately not folded into `report`: report answers "how did the session
go", this answers "what is the surface doing right now", and reaching for
the second while debugging should not mean reading the first.
"""
hdr, samples, events, marks, end = load(path)
t0 = hdr.get("t") or (samples[0]["t"] if samples else 0)
if since or until:
samples, events, marks = slice_session(
samples, events, marks,
parse_when(since, t0) if since else None,
parse_when(until, t0) if until else None)
agg = surface_state(events)
if not agg:
print("gig-log: no CC recorded in this log/window", file=sys.stderr)
return 1
W = 74
print("=" * W)
print(f" SURFACE STATE {path.name} ({len(agg)} controls touched)")
print(" the LAST value each control was left at — not its range")
print("=" * W)
print(f" {'cc':>4} {'left at':>7} {'at':>8} {'role':<28} means")
risky = []
for cc in sorted(agg):
a = agg[cc]
v = a["v"]
role = CONTROL_ROLE.get(cc, "")
means = ""
if cc in DJF_CC and v is not None:
hz = djf_lpf(v)
verdict = djf_verdict(hz)
means = f"lpf {round(hz)} Hz — {verdict}"
if "⚠" in verdict:
risky.append((cc, v, means))
elif cc in MUTE_CC and v is not None:
frac = v / 127.0
means = f"mutes {frac:.0%} of cycles"
if frac > 0.5:
risky.append((cc, v, means))
elif cc == MASK_CC and v is not None:
means = f"gates {v / 127.0:.0%} of cycles"
elif cc == PANIC_CC and v is not None:
means = "PANIC ARMED — global gain 0" if v else "clear"
if v:
risky.append((cc, v, means))
elif cc in ARDOUR_CC:
means = "Ardour-owned — never seeded from here"
vs = "?" if v is None else str(v)
print(f" {cc:>4} {vs:>7} {hms(a['last'] - t0):>8} {role:<28} {means}")
print("")
if risky:
print(" ⚠ PARKED SOMEWHERE THAT SILENCES OR THINS THE SOUND:")
for cc, v, means in risky:
print(f" cc {cc} = {v} → {means}")
print(" If an orbit is silent and it routes through one of these, look"
" here FIRST.")
else:
print(" No control is parked in a known-dangerous zone.")
return 0
def cmd_status(d: Path = LOG_DIR) -> int:
r = subprocess.run(["systemctl", "--user", "is-active", "gig-log.service"],
capture_output=True, text=True)
......@@ -1001,6 +1219,15 @@ def main(argv: list[str] | None = None) -> int:
rp = sub.add_parser("report", help="render a log")
rp.add_argument("file", nargs="?", type=Path)
rp.add_argument("--from", dest="since", metavar="WHEN",
help="window start: HH:MM[:SS] wall clock, or +M:SS from start")
rp.add_argument("--to", dest="until", metavar="WHEN", help="window end")
ct = sub.add_parser("controls",
help="where every control is PARKED (and what that means)")
ct.add_argument("file", nargs="?", type=Path)
ct.add_argument("--from", dest="since", metavar="WHEN")
ct.add_argument("--to", dest="until", metavar="WHEN")
sub.add_parser("status", help="is it running, what is it writing")
ins = sub.add_parser("install", help="systemd --user unit")
......@@ -1011,12 +1238,13 @@ def main(argv: list[str] | None = None) -> int:
a = ap.parse_args(argv)
if a.cmd == "mark":
return cmd_mark(" ".join(a.label), a.dir)
if a.cmd == "report":
if a.cmd in ("report", "controls"):
p = a.file or newest_log(a.dir)
if p is None:
print(f"gig-log: no log found in {a.dir}", file=sys.stderr)
return 1
return cmd_report(p)
fn = cmd_report if a.cmd == "report" else cmd_controls
return fn(p, a.since, a.until)
if a.cmd == "status":
return cmd_status(a.dir)
if a.cmd == "install":
......
tools/tests/test_gig_log.py
......@@ -13,6 +13,7 @@ import importlib.util
import json
import os
import time
from datetime import datetime
from pathlib import Path
import pytest
......@@ -558,3 +559,144 @@ def test_the_report_states_zero_xruns_as_a_positive_verdict(tmp_path, capsys):
gl.cmd_report(r.path)
out = capsys.readouterr().out
assert "ZERO" in out and "the audio path held" in out
# ---------------------------------------------------------------------------
# #82 — the LAST value, and what it MEANS
#
# Range and recency cannot answer "why is that orbit silent"; only the parked
# value can. These pin the arithmetic (which is copied from BootTidal.hs and
# must not drift) and the ordering rule.
# ---------------------------------------------------------------------------
def test_djf_lpf_matches_bootTidal_arithmetic():
# gDJF ch = (# lpf (range 180 20000 (fmap (\v -> 1 - 2*max 0 (0.5-v)) ...)))
# Centre and above is fully open; the bottom is linear in HERTZ.
assert round(gl.djf_lpf(0)) == 180 # hard left = 180 Hz = silence
assert round(gl.djf_lpf(127)) == 20000 # hard right = open
assert round(gl.djf_lpf(64)) == 20000 # centre = ALREADY open
assert round(gl.djf_lpf(96)) == 20000 # upper half does nothing
mid = gl.djf_lpf(32) # quarter-turn
assert 9000 < mid < 11000, mid
def test_djf_lpf_is_linear_in_hz_which_is_the_footgun():
# Equal knob steps = equal HERTZ steps, so the whole audible bottom lives in
# the last few percent of travel. If this ever becomes log-spaced the
# verdicts below must be re-derived, so assert the shape explicitly.
a, b, c = gl.djf_lpf(10), gl.djf_lpf(20), gl.djf_lpf(30)
assert abs((b - a) - (c - b)) < 1.0
def test_djf_verdict_flags_only_the_genuinely_inaudible():
assert "NEAR-SILENT" in gl.djf_verdict(gl.djf_lpf(0))
assert gl.djf_verdict(gl.djf_lpf(127)) == "open"
assert gl.djf_verdict(gl.djf_lpf(64)) == "open"
def test_surface_state_keeps_the_last_value_not_the_last_seen_record():
# Out-of-order records must not decide where a knob is parked.
ev = [
{"t": 100.0, "k": "cc", "cc": 49, "n": 3, "v0": 0, "v1": 10, "lo": 0, "hi": 10},
{"t": 130.0, "k": "cc", "cc": 49, "n": 2, "v0": 60, "v1": 64, "lo": 60, "hi": 64},
{"t": 110.0, "k": "cc", "cc": 49, "n": 1, "v0": 5, "v1": 5, "lo": 5, "hi": 5},
]
a = gl.surface_state(ev)[49]
assert a["v"] == 64, "the latest TIMESTAMP wins, not file order"
assert a["n"] == 6 and a["lo"] == 0 and a["hi"] == 64
def test_surface_state_survives_a_log_with_no_v1(tmp_path):
# Logs written before v1 existed must degrade to "?" rather than crash.
a = gl.surface_state([{"t": 1.0, "k": "cc", "cc": 7, "n": 1, "lo": 3, "hi": 3}])[7]
assert a["v"] is None
def test_slice_session_rebases_cumulative_counters():
# THE bug this guards: xrun/throttle are running totals, so a naive window
# reports the WHOLE session's count inside it — wrong, and wrong high.
s = [{"t": float(i), "k": "s", "xrun": i * 10,
"xrun_by": {"ardour": i * 7}, "thr_pkg": i, "thr_core": 0}
for i in range(11)]
sl, _, _ = gl.slice_session(s, [], [], 5.0, 9.0)
assert [r["t"] for r in sl] == [5.0, 6.0, 7.0, 8.0, 9.0]
assert sl[0]["xrun"] == 0, "the window must start at zero"
assert sl[-1]["xrun"] == 40, "5..9 is a delta of 40, not the total of 90"
assert sl[-1]["xrun_by"]["ardour"] == 28
assert sl[-1]["thr_pkg"] == 4
def test_slice_session_without_a_window_does_not_rebase():
s = [{"t": float(i), "k": "s", "xrun": i * 10} for i in range(4)]
sl, _, _ = gl.slice_session(s, [], [], None, None)
assert sl[-1]["xrun"] == 30, "an unwindowed report must be unchanged"
def test_parse_when_accepts_offsets_and_wall_clock():
t0 = datetime(2026, 7, 29, 0, 57, 8).timestamp()
assert gl.parse_when("+1:30", t0) == t0 + 90
assert gl.parse_when("+1:00:00", t0) == t0 + 3600
got = datetime.fromtimestamp(gl.parse_when("09:15", t0))
assert (got.hour, got.minute) == (9, 15)
def test_parse_when_wall_clock_before_start_crosses_midnight():
t0 = datetime(2026, 7, 29, 23, 30, 0).timestamp()
got = datetime.fromtimestamp(gl.parse_when("00:15", t0))
assert got.day == 30 and (got.hour, got.minute) == (0, 15)
def test_controls_report_names_the_dangerous_parking(tmp_path, capsys):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
t = time.time()
r._write({"t": t, "k": "cc", "p": "24:0", "ch": 0, "cc": 51,
"n": 4, "v0": 90, "v1": 0, "lo": 0, "hi": 90}) # parked hard left
r._write({"t": t, "k": "cc", "p": "24:0", "ch": 0, "cc": 74,
"n": 2, "v0": 0, "v1": 127, "lo": 0, "hi": 127}) # mute held on
r._write({"t": t, "k": "cc", "p": "24:0", "ch": 0, "cc": 49,
"n": 2, "v0": 0, "v1": 100, "lo": 0, "hi": 100}) # safe
r.f.close()
assert gl.cmd_controls(r.path) == 0
out = capsys.readouterr().out
assert "PARKED SOMEWHERE THAT SILENCES" in out
assert "cc 51" in out and "NEAR-SILENT" in out
assert "cc 74" in out and "100%" in out
risky = out.split("PARKED SOMEWHERE THAT SILENCES")[1]
assert "cc 49" not in risky, "an open filter must not be flagged"
def test_controls_report_is_quiet_when_the_surface_is_safe(tmp_path, capsys):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
r._write({"t": time.time(), "k": "cc", "p": "24:0", "ch": 0, "cc": 49,
"n": 1, "v0": 100, "v1": 100, "lo": 100, "hi": 100})
r.f.close()
assert gl.cmd_controls(r.path) == 0
assert "No control is parked in a known-dangerous zone" in capsys.readouterr().out
def test_report_window_narrows_the_xrun_total(tmp_path, capsys):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
hdr = json.loads(r.path.read_text().splitlines()[0])
hdr["xruns"] = True
t0 = hdr["t"]
for i in range(20):
r.f.write(json.dumps({"t": t0 + i, "k": "s", "pkg": 50, "core_max": 50,
"fan": [1700], "fmax": 2900, "favg": 2900,
"cpu_max": 10, "cpu_avg": 5, "thr_core": 0,
"thr_pkg": 0, "w": None, "gear": {},
"xrun": i * 100}) + "\n")
r.f.close()
lines = r.path.read_text().splitlines()
lines[0] = json.dumps(hdr)
r.path.write_text("\n".join(lines) + "\n")
gl.cmd_report(r.path)
assert "1900" in capsys.readouterr().out
gl.cmd_report(r.path, since="+0:10", until="+0:15")
out = capsys.readouterr().out
assert "WINDOW" in out and "rebased" in out
assert "500" in out and "1900" not in out
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