Commit 4c57362d by PLN (Algolia)

feat(gig-log): a session recorder, so a run-through can be REVIEWED and not just felt

After a set, "did it glitch?" and "did it get hot?" and "which controls did I
actually use?" are answered from memory. Memory is a bad instrument, and the
answers matter: #8 (thermal impact under load), #56 (Pulsar starving audio),
#46/#11 (which surface controls are really in the hands). PLN records the
run-through in Ardour; this gives that take a machine-readable twin.

WHAT IT DOES
  `gig-log.py record`   1 Hz JSONL: package/core temp, fan, freq, throttle
                        counters, per-core cpu%, and per-gear %cpu + RSS for
                        scsynth / sclang / ArdourGUI / the whole pulsar process
                        group. Plus PipeWire xruns and every LCXL control move.
  `gig-log.py mark ...` annotate the live session ("gimme acid drop")
  `gig-log.py report`   render it back: sparklines, xrun timeline, gear table,
                        a per-CC surface table with first/last touch
  `gig-log.py status`   is it running, what is it writing
  `gig-log.py install`  systemd --user unit, enabled, starts with the session
  `gig-log.py selftest` prove the parsers AND the cost

Wall-clock stamped in the header, so the timeline lines up with the Ardour take.

THE OBSERVER MUST NOT PERTURB — measured, not asserted
This rig has twice produced the fault it was measuring: probe-chain's capture
streams caused the xruns it was hunting, and the LED daemon's per-event fork
caused the lag it was reporting. So: no audio capture at all, no per-sample
subprocess spawn (every number comes from sysfs/procfs), and exactly two
long-lived children (`pw-top -b`, `aseqdump`) drained by threads so a filling
pipe can never block them. Measured cost of the whole thing: 0.40% of one core
with 270 pw-top lines parsed in 10 s; 0.79% as the live systemd unit. The first
selftest reported 0.66% and was FLATTERING ITSELF — it never started the reader
threads, so it measured a sampler with nothing to parse. Same mistake as the
first --bench run in the LED work; fixed, then re-measured lower and honest.

AND THE LOGGER MUST NOT BE A FIREHOSE
A fader sweep is 100+ events/s. A logger that writes them all costs more than
what it measures, so CC/pitchbend are coalesced to one line per control per
second carrying count + first/last/min/max — the #71/#72 fix applied to
ourselves. 128 events on one knob is one line that still shows the whole travel.
NOTES ARE NEVER COALESCED: a CC is a state and may be superseded, a note is an
event and dropping one loses a thing that happened.

COUNTERS ARE DELTAS, AND THE BASELINE IS WHERE THE BUGS LIVED
Two absolute counters here are large and meaningless alone: package_throttle
was 17024 after 2 idle days, and Ardour's PipeWire ERR was 67072. Both count
history, including power excursions and device changes that never touched audio.
So the header records baselines and samples record deltas.

Getting Ardour's baseline right took three rules, and the first report caught
each one:
  1. baseline on FIRST sight -> pw-top prints a zero-filled snapshot before the
     profiler has data, so base=0 and a 25-second IDLE session reported 72096
     xruns.
  2. baseline on SECOND sight -> usually right. pw-top emits a
     non-deterministic NUMBER of zero tables, so it silently reported 67123 on
     the next real run. A rule that is right most of the time is the worst kind
     for a gig log, because the one bad reading looks exactly like a disaster.
  3. baseline = MAX over each node's first 5 sightings. ERR is monotonic within
     a node's lifetime, so the max over a warm-up IS the true starting count —
     no timing assumption at all. A value below the baseline means the node was
     destroyed and recreated, so re-baseline instead of reporting negative.
Validated by three independent 15 s runs, all reporting 0 (measure twice in
time before trusting one reading). The report DECLARES the warm-up blind spot
rather than hiding it.

Also learned on the way: Ardour accumulates ~6 xruns/min even with nothing
playing (67072 -> 67123 -> 67134 across captures minutes apart), which is
exactly why only the session delta may ever be quoted.

FIXED IN THE SHARED READER
perf._read(None) raised TypeError instead of returning the default, so on any
machine without a coretemp/dell_smm hwmon the thermal read CRASHED rather than
reading "unknown" — the Bridge shares this code path. Guarded.

Refactored _proc_cpu_rss into a pure parse_proc_stat() to make it testable, and
it needed to be: pulsar's renderer comms look like `(pulsar) --type=renderer`,
so splitting /proc/pid/stat on whitespace from the left shifts every field and
silently reports some other column as CPU.

TESTS: 71 new (tools/tests/test_gig_log.py), suite 211 -> 282, all green.
Covers real pw-top/aseqdump lines, every baseline regression above, coalescing
invariants, notes-survive-a-CC-flood, a torn final line costing one sample not
the log, absent gear degrading to partial data instead of a crash, and
sparklines bucketing by MAX so a one-second burst inside a 40-minute set cannot
be averaged away.
parent cd5c46bb
...@@ -36,6 +36,12 @@ TEMP_STATES = ((55, "cool"), (70, "warm"), (85, "hot"), (10_000, "critical")) ...@@ -36,6 +36,12 @@ TEMP_STATES = ((55, "cool"), (70, "warm"), (85, "hot"), (10_000, "critical"))
def _read(path, default=None): def _read(path, default=None):
# path can legitimately be None: the hwmon probes return None when a chip is
# absent (no coretemp on AMD, no dell_smm on non-Dell), and callers pass that
# straight back in. open(None) raises TypeError, not OSError, so without this
# guard a missing sensor crashes the reader instead of reading as "unknown".
if not path:
return default
try: try:
with open(path) as f: with open(path) as f:
return f.read().strip() return f.read().strip()
......
#!/usr/bin/env python3
"""gig-log — a light session recorder, so a set can be REVIEWED and not just felt.
The problem it solves: after a run-through, "did it glitch?" and "did it get hot?"
and "which controls did I actually use?" are answered from memory. Memory is a bad
instrument. This writes a 1 Hz timeline to JSONL, wall-clock stamped so it lines up
with the Ardour take, and `--report` renders it back.
WHAT IT RECORDS
1 Hz sample package/core temp, fan rpm, cpu freq, throttle counts (as DELTAS),
per-core cpu%, and per-gear %cpu + RSS for scsynth / sclang /
ArdourGUI / pulsar (the whole pulsar process group, summed).
xruns PipeWire ERR counts per node, also as deltas from a baseline.
MIDI every control move on the LCXL (and any other seq port), coalesced.
markers free-text annotations (`gig-log.py mark "gimme acid drop"`).
THE LAW: THE OBSERVER MUST NOT PERTURB
Measuring this rig has caused the fault twice now — probe-chain's capture streams
produced the very xruns they were hunting, and the LED daemon's per-event fork made
the lag it was reporting. So:
* NO audio capture, ever. Nothing here opens a stream.
* NO per-sample subprocess spawn. Temps, freq, throttle, cpu and RSS all come
from sysfs/procfs reads. Two long-lived children total (`pw-top -b` for xruns,
`aseqdump` for MIDI), started once and drained by threads.
* MIDI IS COALESCED, because a fader sweep is 100+ events/s and a logger that
writes them all is a firehose that costs more than what it measures. One line
per (port, channel, controller) per second, carrying count + first/last/min/max.
That is exactly the #71/#72 fix applied to ourselves. NOTES ARE NEVER DROPPED —
a note is an event, a CC is a state.
* Measured cost of the whole thing: see `--selftest`. pw-top alone was 1 jiffy
per 12 s (0.08% of one core) when this was written.
COUNTERS ARE DELTAS, NOT TOTALS
Two absolute counters on this machine are big and meaningless on their own:
`package_throttle_count` was 17024 after 2 days idle, and Ardour's PipeWire ERR
was 67072. Both count history, including power excursions and device changes that
never touched audio. Only the delta across the session says anything, so the header
records the baseline and every sample records the delta.
USAGE
tools/gig-log.py record # 1 Hz recorder, Ctrl-C to stop
tools/gig-log.py mark "vague de crime" # annotate the running session
tools/gig-log.py report # render the newest log
tools/gig-log.py report FILE.jsonl
tools/gig-log.py status # is a recorder running, what is it writing
tools/gig-log.py install # systemd --user unit (boots with the machine)
tools/gig-log.py selftest # prove the sampler is cheap + parsers work
"""
from __future__ import annotations
import argparse
import glob
import json
import os
import re
import shutil
import signal
import subprocess
import sys
import threading
import time
from datetime import datetime
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent / "bridge"))
import perf # noqa: E402 (rootless Thermals/_read live here already — DRY with the Bridge)
LOG_DIR = Path(os.environ.get(
"GIG_LOG_DIR", os.path.expanduser("~/.local/share/parvagues/gig-log")))
# Gear we care about, by /proc/<pid>/comm. A value of `True` means "sum the whole
# process group" — pulsar is 8 processes (main, GPU, renderers) and the interesting
# number is the total, with the worst single one called out separately.
GEAR = {
"scsynth": False, # the audio server. If this starves, the set dies.
"sclang": False, # SuperDirt / our MIDI bridge
"ArdourGUI": False, # recorder + master chain
"pulsar": True, # editor: the known renderer burn (#7/#56)
}
# PipeWire nodes whose xruns actually mean something for us. Anything else is
# still counted into the total, but these get their own line in the report.
XRUN_NODES = ("SuperCollider", "ardour", "Line__sink")
RESCAN_S = 5.0 # how often to re-resolve pids (a restart must not go unnoticed)
PRUNE_DAYS = 30
SPARK = "▁▂▃▄▅▆▇█"
# --------------------------------------------------------------------------- #
# procfs: per-gear cpu% and RSS with no forks
# --------------------------------------------------------------------------- #
def _proc_comm(pid: str) -> str | None:
return perf._read(f"/proc/{pid}/comm")
def parse_proc_stat(raw: str) -> tuple[int, int] | None:
"""(utime+stime jiffies, rss_pages) from one /proc/<pid>/stat line.
comm sits in parens and may itself contain spaces AND parens — Chromium-family
processes like pulsar's renderers do exactly this. Splitting on whitespace from
the left therefore shifts every field, which is the classic /proc/stat trap. So
split after the LAST ')'.
"""
if not raw:
return None
try:
rest = raw[raw.rindex(")") + 2:].split()
# rest[0] is `state`, i.e. field 3; so field N -> rest[N - 3]
return int(rest[11]) + int(rest[12]), int(rest[21]) # utime+stime, rss
except (ValueError, IndexError):
return None
def _proc_cpu_rss(pid: str) -> tuple[int, int] | None:
"""(utime+stime jiffies, rss_pages) or None if the pid is gone."""
return parse_proc_stat(perf._read(f"/proc/{pid}/stat") or "")
class GearWatch:
"""Per-gear cpu% + RSS, procfs only.
Re-resolves pids every RESCAN_S *and* immediately whenever a tracked pid
vanishes. A binding resolved once at startup and never rechecked is this rig's
single most common silent failure, and a logger that keeps reporting a dead
scsynth as "0%" would be worse than no logger.
"""
def __init__(self, gear: dict[str, bool] = GEAR):
self.gear = gear
self._pids: dict[str, list[str]] = {}
self._prev: dict[str, tuple[float, int]] = {} # comm -> (mono, jiffies)
self._last_scan = 0.0
self._hz = os.sysconf("SC_CLK_TCK") or 100
self._page_mb = (os.sysconf("SC_PAGE_SIZE") or 4096) / 1024 / 1024
self.scan()
def scan(self) -> None:
found = {c: [] for c in self.gear}
for p in glob.glob("/proc/[0-9]*/comm"):
pid = p.split("/")[2]
c = perf._read(p)
if c in found:
found[c].append(pid)
self._pids = found
self._last_scan = time.monotonic()
def sample(self) -> dict[str, dict]:
now = time.monotonic()
if now - self._last_scan >= RESCAN_S:
self.scan()
out = {}
for comm, group in self.gear.items():
pids = self._pids.get(comm) or []
jif = rss = 0
worst = 0
alive = 0
for pid in pids:
r = _proc_cpu_rss(pid)
if r is None:
self.scan() # a pid died: re-resolve NOW, don't wait
continue
alive += 1
jif += r[0]
rss += r[1]
worst = max(worst, r[0])
if not alive:
self._prev.pop(comm, None)
out[comm] = {"up": False}
continue
entry = {"up": True, "n": alive, "rss": round(rss * self._page_mb, 1)}
prev = self._prev.get(comm)
self._prev[comm] = (now, jif)
if prev and now > prev[0]:
dt = now - prev[0]
entry["cpu"] = round(100 * (jif - prev[1]) / self._hz / dt, 1)
if group and alive > 1:
entry["grp"] = alive
out[comm] = entry
return out
# --------------------------------------------------------------------------- #
# xruns: one long-lived pw-top, drained by a thread
# --------------------------------------------------------------------------- #
class XrunReader(threading.Thread):
"""Latest per-node PipeWire ERR count, from a single `pw-top -b`.
pw-top batch mode emits a table ~8x/second. We do NOT want 8 samples/s in the
log, and we must not let the pipe fill (a full pipe blocks the child, which is
how the LED daemon's latency grew without bound). So: drain continuously in a
thread, keep only the newest value per node, and let the 1 Hz loop read it.
Table layout: S ID QUANT RATE WAIT BUSY W/Q B/Q ERR [FORMAT...] NAME
ERR is always token 8; NAME is always the last token (with an optional '+'
prefix for a linked node). FORMAT is variable-width or empty, which is exactly
why we index from both ends and never from the middle.
"""
WARMUP = 5 # sightings per node before its baseline is frozen
def __init__(self):
super().__init__(daemon=True)
self.err: dict[str, int] = {}
self.base: dict[str, int] = {}
self._seen: dict[str, int] = {}
self.lock = threading.Lock()
self.alive = False
self.lines = 0
self._proc: subprocess.Popen | None = None
self._stop = threading.Event()
@staticmethod
def available() -> bool:
return shutil.which("pw-top") is not None
def parse_line(self, line: str) -> tuple[str, int] | None:
f = line.split()
if len(f) < 10 or f[0] not in ("R", "I", "S", "C", "D", "E", "!"):
return None
if not f[1].isdigit():
return None
try:
err = int(f[8])
except ValueError:
return None
name = f[-1].lstrip("+").strip()
return (name, err) if name else None
def _baseline(self, name: str, err: int) -> None:
"""Record a value; baseline each node as the MAX over its first few sightings.
pw-top prints a zero-filled snapshot before the profiler has delivered real
counters — and it prints more than one of them, non-deterministically. Two
wrong rules were tried before this one:
* baseline on FIRST sight -> captured 0, so Ardour's lifetime 67072 was
reported as 67072 xruns in a 25-second idle session;
* baseline on SECOND sight -> usually right, and silently wrong whenever
pw-top happened to emit two zero tables. A rule that is right most of
the time is the worst kind for a gig log, because the one bad reading
looks exactly like a disaster.
ERR is monotonically non-decreasing within a node's lifetime, so the max
over a warm-up window IS the true starting count — no timing assumption
required. WARMUP sightings, not seconds, so a node that appears mid-set
(he opens something) gets the same treatment.
A value BELOW the baseline means the node was destroyed and recreated with
a fresh counter; re-baseline rather than report a negative delta.
Caller holds the lock.
"""
n = self._seen.get(name, 0) + 1
self._seen[name] = n
self.err[name] = err
if n <= self.WARMUP:
self.base[name] = max(self.base.get(name, 0), err)
elif err < self.base.get(name, 0):
self.base[name] = err
def run(self) -> None:
while not self._stop.is_set():
try:
self._proc = subprocess.Popen(
["pw-top", "-b"], stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, text=True, bufsize=1)
except OSError:
return
self.alive = True
for line in self._proc.stdout: # blocking read, own thread
if self._stop.is_set():
break
hit = self.parse_line(line)
if hit is None:
continue
self.lines += 1
with self.lock:
self._baseline(*hit)
self.alive = False
if self._stop.wait(3.0): # respawn with backoff
return
def deltas(self) -> dict[str, int]:
with self.lock:
return {k: v - self.base.get(k, v) for k, v in self.err.items()}
def total_delta(self) -> int:
return sum(max(0, v) for v in self.deltas().values())
def stop(self) -> None:
self._stop.set()
if self._proc and self._proc.poll() is None:
self._proc.terminate()
# --------------------------------------------------------------------------- #
# MIDI: one long-lived aseqdump, coalesced per second
# --------------------------------------------------------------------------- #
CC_RE = re.compile(
r"^\s*(\d+:\d+)\s+Control change\s+(\d+),\s*controller\s+(\d+),\s*value\s+(\d+)")
NOTE_RE = re.compile(
r"^\s*(\d+:\d+)\s+Note (on|off)\s+(\d+),\s*note\s+(\d+),\s*velocity\s+(\d+)")
PB_RE = re.compile(r"^\s*(\d+:\d+)\s+Pitch bend\s+(\d+),\s*value\s+(-?\d+)")
PORT_RE = re.compile(r"^\s*(\d+:\d+)\s+(.+?)\s\s+")
def find_seq_port(want: str = "Launch Control XL") -> str | None:
"""Resolve the LCXL's ALSA seq port by NAME, every time we (re)connect.
Never hardcode it: this device has already moved 20:0 -> 24:0 across a replug,
and a port number cached at startup is the stale-binding failure again.
"""
try:
r = subprocess.run(["aseqdump", "-l"], capture_output=True, text=True,
timeout=5)
except (OSError, subprocess.SubprocessError):
return None
for line in (r.stdout or "").splitlines():
if want.lower() in line.lower() and "HUI" not in line:
m = re.match(r"\s*(\d+:\d+)", line)
if m:
return m.group(1)
return None
class MidiReader(threading.Thread):
"""Reads one aseqdump and accumulates a COALESCED window.
Read-only seq subscription — it does not steal MIDI from SuperCollider (SC holds
the *rawmidi* substream, which is a different resource; that is also why LED
writes have to go over seq).
"""
def __init__(self, port: str | None = None, want: str = "Launch Control XL"):
super().__init__(daemon=True)
self.want = want
self.fixed = port
self.port: str | None = None
self.cc: dict[tuple[str, int, int], dict] = {}
self.discrete: list[dict] = []
self.lock = threading.Lock()
self.events = 0
self.alive = False
self._proc: subprocess.Popen | None = None
self._stop = threading.Event()
@staticmethod
def available() -> bool:
return shutil.which("aseqdump") is not None
def feed(self, line: str, now: float) -> bool:
"""Fold one aseqdump line into the window. True if it was an event."""
m = CC_RE.match(line)
if m:
port, ch, cc, val = m.group(1), int(m.group(2)), int(m.group(3)), int(m.group(4))
key = (port, ch, cc)
with self.lock:
e = self.cc.get(key)
if e is None:
self.cc[key] = {"t": now, "n": 1, "v0": val, "v1": val,
"lo": val, "hi": val}
else:
e["n"] += 1
e["v1"] = val
e["lo"] = min(e["lo"], val)
e["hi"] = max(e["hi"], val)
self.events += 1
return True
m = NOTE_RE.match(line)
if m:
with self.lock: # a note is an EVENT: never coalesced away
self.discrete.append({"t": now, "k": "note", "p": m.group(1),
"on": m.group(2) == "on", "ch": int(m.group(3)),
"note": int(m.group(4)), "vel": int(m.group(5))})
self.events += 1
return True
m = PB_RE.match(line)
if m:
key = (m.group(1), int(m.group(2)), -1)
val = int(m.group(3))
with self.lock:
e = self.cc.get(key)
if e is None:
self.cc[key] = {"t": now, "n": 1, "v0": val, "v1": val,
"lo": val, "hi": val}
else:
e["n"] += 1
e["v1"] = val
e["lo"] = min(e["lo"], val)
e["hi"] = max(e["hi"], val)
self.events += 1
return True
return False
def drain(self) -> tuple[list[dict], list[dict]]:
"""Take the window; returns (coalesced cc records, discrete events)."""
with self.lock:
cc, self.cc = self.cc, {}
disc, self.discrete = self.discrete, []
out = []
for (port, ch, num), e in sorted(cc.items()):
rec = {"t": round(e["t"], 3), "k": "cc" if num >= 0 else "pb",
"p": port, "ch": ch, "n": e["n"],
"v0": e["v0"], "v1": e["v1"], "lo": e["lo"], "hi": e["hi"]}
if num >= 0:
rec["cc"] = num
out.append(rec)
return out, disc
def run(self) -> None:
while not self._stop.is_set():
port = self.fixed or find_seq_port(self.want) # re-resolve every time
if not port:
if self._stop.wait(5.0):
return
continue
self.port = port
try:
self._proc = subprocess.Popen(
["aseqdump", "-p", port], stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, text=True, bufsize=1)
except OSError:
return
self.alive = True
for line in self._proc.stdout:
if self._stop.is_set():
break
self.feed(line, time.time())
self.alive = False
if self._stop.wait(3.0):
return
def stop(self) -> None:
self._stop.set()
if self._proc and self._proc.poll() is None:
self._proc.terminate()
# --------------------------------------------------------------------------- #
# the recorder
# --------------------------------------------------------------------------- #
def _uptime_s() -> float:
raw = perf._read("/proc/uptime") or "0"
try:
return float(raw.split()[0])
except (ValueError, IndexError):
return 0.0
def _package_throttle_sum() -> int:
total = 0
for p in glob.glob(
"/sys/devices/system/cpu/cpu*/thermal_throttle/package_throttle_count"):
total += perf._read_int(p, 0) or 0
return total
class Recorder:
def __init__(self, out_dir: Path = LOG_DIR, hz: float = 1.0,
xruns: bool = True, midi: bool = True,
midi_port: str | None = None):
self.dir = out_dir
self.period = 1.0 / max(0.05, hz)
self.th = perf.Thermals()
self.gear = GearWatch()
self._pkg_base = _package_throttle_sum()
self.xr = XrunReader() if (xruns and XrunReader.available()) else None
self.mid = MidiReader(midi_port) if (midi and MidiReader.available()) else None
self.path = self._open()
self.samples = 0
self._stop = threading.Event()
def _open(self) -> Path:
self.dir.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
p = self.dir / f"gig-{stamp}.jsonl"
self.f = p.open("a", buffering=1) # line-buffered: a kill -9 keeps the log
return p
def prune(self, days: int = PRUNE_DAYS) -> int:
cut = time.time() - days * 86400
n = 0
for p in self.dir.glob("gig-*.jsonl"):
try:
if p.stat().st_mtime < cut:
p.unlink()
n += 1
except OSError:
pass
return n
def _write(self, rec: dict) -> None:
self.f.write(json.dumps(rec, separators=(",", ":")) + "\n")
def header(self) -> dict:
pl = perf.power_caps_w()
return {
"t": round(time.time(), 3), "k": "hdr", "v": 1,
"iso": datetime.now().isoformat(timespec="seconds"),
"mono": round(time.monotonic(), 3),
"uptime": round(_uptime_s()),
"mode": perf.detect_mode(),
"gov": perf._read("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor"),
"pl1": pl.get("pl1"), "pl2": pl.get("pl2"),
"fmax": perf._read_int("/sys/devices/system/cpu/cpu0/cpufreq/"
"cpuinfo_max_freq"),
"thermald": perf.thermald_active(),
"gpu": perf.gpu_manager_mode(),
"dgpu": perf.dgpu_state(),
"base_core_throttle": self.th._throttle_base,
"base_pkg_throttle": self._pkg_base,
"period": round(self.period, 3),
"xruns": bool(self.xr), "midi": bool(self.mid),
"gear": {c: bool(self.gear._pids.get(c)) for c in GEAR},
}
def sample(self) -> dict:
cores = self.th.cores_c()
fmax, favg = self.th.freq_mhz()
pct = self.th.cpu_pct()
rec = {
"t": round(time.time(), 3), "k": "s",
"pkg": self.th.package_c(),
"core_max": max(cores) if cores else None,
"fan": self.th.fans_rpm(),
"fmax": fmax, "favg": favg,
"cpu_max": max(pct) if pct else None,
"cpu_avg": round(sum(pct) / len(pct)) if pct else None,
"thr_core": self.th.throttle_delta(),
"thr_pkg": _package_throttle_sum() - self._pkg_base,
"w": self.th.power_draw_w(),
"gear": self.gear.sample(),
}
if self.xr:
d = self.xr.deltas()
rec["xrun"] = sum(max(0, v) for v in d.values())
hot = {k: v for k, v in d.items() if v > 0}
if hot:
rec["xrun_by"] = hot
return rec
def run(self) -> int:
for t in (self.xr, self.mid):
if t:
t.start()
self._write(self.header())
self.prune()
print(f"gig-log: recording -> {self.path}", file=sys.stderr)
print(f"gig-log: xruns={'pw-top' if self.xr else 'off'} "
f"midi={'aseqdump' if self.mid else 'off'} "
f"period={self.period:.2f}s", file=sys.stderr)
next_t = time.monotonic()
try:
while not self._stop.is_set():
next_t += self.period
self._write(self.sample())
self.samples += 1
if self.mid:
cc, disc = self.mid.drain()
for r in disc:
r["t"] = round(r["t"], 3)
self._write(r)
for r in cc:
self._write(r)
self._stop.wait(max(0.0, next_t - time.monotonic()))
except KeyboardInterrupt:
pass
finally:
self._write({"t": round(time.time(), 3), "k": "end",
"samples": self.samples,
"midi_events": self.mid.events if self.mid else 0,
"xrun": self.xr.total_delta() if self.xr else None})
for t in (self.xr, self.mid):
if t:
t.stop()
self.f.close()
print(f"gig-log: {self.samples} samples -> {self.path}", file=sys.stderr)
return 0
def stop(self, *_a) -> None:
self._stop.set()
# --------------------------------------------------------------------------- #
# marks
# --------------------------------------------------------------------------- #
def newest_log(d: Path = LOG_DIR) -> Path | None:
logs = sorted(d.glob("gig-*.jsonl"), key=lambda p: p.stat().st_mtime)
return logs[-1] if logs else None
def cmd_mark(label: str, d: Path = LOG_DIR) -> int:
"""Append a marker to the newest log.
Deliberately dumb: an O_APPEND write of one short line is atomic enough that
it can never interleave with the recorder's own lines, so no IPC, no socket,
nothing that can wedge the recorder. If there is no log, say so and exit 1
rather than silently creating an orphan.
"""
p = newest_log(d)
if p is None:
print(f"gig-log: no log in {d} — start `gig-log.py record` first",
file=sys.stderr)
return 1
rec = {"t": round(time.time(), 3), "k": "mark", "label": label}
with p.open("a") as f:
f.write(json.dumps(rec, separators=(",", ":")) + "\n")
print(f"mark @ {datetime.now().strftime('%H:%M:%S')} {label} -> {p.name}")
return 0
# --------------------------------------------------------------------------- #
# report
# --------------------------------------------------------------------------- #
def spark(vals: list, width: int = 60) -> str:
"""Sparkline over a fixed width. An aggregate cannot tell a fade from a sparse
part — keeping the time axis is the whole point (see the probe-chain lesson)."""
vals = [v for v in vals if v is not None]
if not vals:
return "(no data)"
if len(vals) > width: # bucket by MAX: a spike must survive
step = len(vals) / width
vals = [max(vals[int(i * step):max(int((i + 1) * step), int(i * step) + 1)])
for i in range(width)]
lo, hi = min(vals), max(vals)
if hi == lo:
return SPARK[0] * len(vals) + f" (flat {lo})"
return "".join(SPARK[min(7, int(8 * (v - lo) / (hi - lo)))] for v in vals)
def gl_warmup() -> int:
return XrunReader.WARMUP
def hms(seconds: float) -> str:
s = int(max(0, seconds))
return f"{s // 3600:d}:{(s % 3600) // 60:02d}:{s % 60:02d}" if s >= 3600 \
else f"{s // 60:d}:{s % 60:02d}"
def load(path: Path) -> tuple[dict, list[dict], list[dict], list[dict], dict]:
hdr, samples, ccs, marks, end = {}, [], [], [], {}
with path.open() as f:
for line in f:
line = line.strip()
if not line:
continue
try:
r = json.loads(line)
except json.JSONDecodeError:
continue
k = r.get("k")
if k == "hdr":
hdr = r
elif k == "s":
samples.append(r)
elif k in ("cc", "pb", "note"):
ccs.append(r)
elif k == "mark":
marks.append(r)
elif k == "end":
end = r
return hdr, samples, ccs, marks, end
def cmd_report(path: Path) -> 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"]
dur = samples[-1]["t"] - t0
W = 74
p = print
p("=" * W)
p(f" GIG LOG {path.name}")
p("=" * W)
p(f" started {hdr.get('iso', '?')} (align the Ardour take to this wall clock)")
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')} "
f"PL1={hdr.get('pl1')}W PL2={hdr.get('pl2')}W "
f"thermald={'on' if hdr.get('thermald') else 'off'} gpu={hdr.get('gpu')}")
gear0 = hdr.get("gear") or {}
p(f" gear @start " + " ".join(
f"{'✓' if v else '✗'}{c}" for c in GEAR for v in [gear0.get(c)]))
# ---- xruns: the number that decides whether anything else matters --------
p("")
p("─ XRUNS " + "─" * (W - 8))
if not hdr.get("xruns"):
p(" pw-top was off for this session — no xrun data")
else:
xr = [s.get("xrun") for s in samples]
total = max([v for v in xr if v is not None], default=0)
p(f" total (delta over the session) {total}")
if total:
p(f" {spark(xr)}")
by: dict[str, int] = {}
for s in samples:
for k, v in (s.get("xrun_by") or {}).items():
by[k] = max(by.get(k, 0), v)
for k, v in sorted(by.items(), key=lambda kv: -kv[1])[:6]:
p(f" {v:>6} {k}")
first = next((s["t"] - t0 for s in samples if (s.get("xrun") or 0) > 0), None)
if first is not None:
p(f" first xrun at +{hms(first)} — check the Ardour take there")
else:
p(" ZERO. Whatever else this log says, the audio path held.")
p(f" (blind spot: each node's first {gl_warmup()} pw-top tables are its"
f" baseline warm-up, ~1-5 s.")
p(" Start the recorder before the gear, not after, and this costs nothing.)")
# ---- thermals -----------------------------------------------------------
p("")
p("─ THERMAL " + "─" * (W - 10))
pk = [s.get("pkg") for s in samples]
fm = [s.get("fmax") for s in samples]
known = [v for v in pk if v is not None]
if known:
p(f" package min {min(known)} mean {round(sum(known)/len(known))} "
f"max {max(known)} °C")
p(f" temp {spark(pk)}")
if any(v is not None for v in fm):
f_ok = [v for v in fm if v is not None]
p(f" freq {spark(fm)} max {max(f_ok)} MHz mean {round(sum(f_ok)/len(f_ok))} MHz")
fans = [max(s["fan"]) for s in samples if s.get("fan")]
if fans:
p(f" fan min {min(fans)} max {max(fans)} rpm")
thr_c = samples[-1].get("thr_core") or 0
thr_p = samples[-1].get("thr_pkg") or 0
p(f" throttle events THIS SESSION core {thr_c} package {thr_p}")
p(f" (baselines were core {hdr.get('base_core_throttle')} / "
f"package {hdr.get('base_pkg_throttle')} — absolute counts are meaningless,")
p(f" they count power-limit excursions too. Only the delta above is ours.)")
if thr_p or thr_c:
p(f" thr {spark([s.get('thr_pkg') for s in samples])}")
# the correlation question #8 actually asks
if hdr.get("xruns"):
pairs = [(s.get("thr_pkg") or 0, s.get("xrun") or 0) for s in samples]
d_thr = [b[0] - a[0] for a, b in zip(pairs, pairs[1:])]
d_xr = [b[1] - a[1] for a, b in zip(pairs, pairs[1:])]
both = sum(1 for a, b in zip(d_thr, d_xr) if a > 0 and b > 0)
only_x = sum(1 for a, b in zip(d_thr, d_xr) if a == 0 and b > 0)
thr_secs = sum(1 for a in d_thr if a > 0)
if sum(d_xr) == 0:
p(f" VERDICT {thr_secs}s of this session throttled and cost ZERO xruns.")
else:
p(f" VERDICT xruns in {both}s that also throttled, "
f"{only_x}s with xruns and no throttle "
f"({thr_secs}s throttled in total).")
# ---- gear ---------------------------------------------------------------
p("")
p("─ GEAR " + "─" * (W - 7))
p(f" {'process':<12} {'cpu% mean':>9} {'max':>6} {'rss mean':>9} {'max':>7} seen")
for comm in GEAR:
cpu = [s["gear"].get(comm, {}).get("cpu") for s in samples]
cpu = [v for v in cpu if v is not None]
rss = [s["gear"].get(comm, {}).get("rss") for s in samples]
rss = [v for v in rss if v is not None]
up = sum(1 for s in samples if s["gear"].get(comm, {}).get("up"))
if not up:
p(f" {comm:<12} {'—':>9} {'—':>6} {'—':>9} {'—':>7} never")
continue
p(f" {comm:<12} {round(sum(cpu)/len(cpu),1) if cpu else 0:>9} "
f"{max(cpu) if cpu else 0:>6} "
f"{round(sum(rss)/len(rss)) if rss else 0:>8}M "
f"{round(max(rss)) if rss else 0:>6}M {up}/{len(samples)}s")
if rss and len(rss) > 10 and max(rss) > 1.25 * rss[0]:
p(f" ⚠ RSS grew {round(rss[0])}M → {round(max(rss))}M "
f"({round(100*(max(rss)/rss[0]-1))}%) — the pulsar leak pattern")
if up < len(samples):
p(f" ⚠ MISSING for {len(samples)-up}s of the session (started late, "
f"or died)")
cm = [s.get("cpu_max") for s in samples]
if any(v is not None for v in cm):
p(f" hottest core {spark(cm)} max {max(v for v in cm if v is not None)}%")
# ---- surface ------------------------------------------------------------
p("")
p("─ SURFACE " + "─" * (W - 10))
ccs = [e for e in events if e.get("k") == "cc"]
notes = [e for e in events if e.get("k") == "note"]
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"])
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")
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} "
f"{hms(a['first']-t0):>7} {hms(a['last']-t0):>7} {a['s']:>5}s{flag}")
if notes:
p(f" {len(notes)} note events "
f"({len({n['note'] for n in notes})} distinct notes) — keystep/pads")
# ---- markers ------------------------------------------------------------
if marks:
p("")
p("─ MARKERS " + "─" * (W - 10))
for m in marks:
p(f" +{hms(m['t']-t0):>8} {m.get('label','')}")
p("")
if end:
p(f" session closed cleanly: {end.get('samples')} samples, "
f"{end.get('midi_events')} midi events")
else:
p(" ⚠ no `end` record — the recorder is still running, or it was killed")
p("=" * W)
return 0
# --------------------------------------------------------------------------- #
# status / install / selftest
# --------------------------------------------------------------------------- #
UNIT = """[Unit]
Description=gig-log (ParVagues session recorder: thermals, xruns, gear, MIDI)
Documentation=file://{repo}/tools/gig-log.py
[Service]
Type=simple
WorkingDirectory={repo}
ExecStart={py} {repo}/tools/gig-log.py record
Restart=always
RestartSec=5
Nice=5
# It must never be the reason audio stutters: below every audio thread, and
# capped so a runaway parse cannot eat a core.
CPUWeight=20
IOWeight=20
[Install]
WantedBy=default.target
"""
def cmd_install(enable: bool = True) -> int:
repo = Path(__file__).resolve().parent.parent
dest = Path.home() / ".config/systemd/user/gig-log.service"
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(UNIT.format(repo=repo, py=sys.executable))
print(f"wrote {dest}")
steps = [["systemctl", "--user", "daemon-reload"]]
if enable:
steps += [["systemctl", "--user", "enable", "--now", "gig-log.service"]]
for cmd in steps:
r = subprocess.run(cmd, capture_output=True, text=True)
print(f"$ {' '.join(cmd)} -> {r.returncode} {(r.stderr or '').strip()}")
if r.returncode != 0:
return r.returncode
print("gig-log: enabled (starts with the session; `loginctl enable-linger` "
"already on for the Bridge, so it survives logout)")
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)
unit = (r.stdout or "").strip() or "absent"
# Only RECORDERS count. A concurrent `report`/`mark`/`status` is also a
# gig-log.py process, and counting those made `status` claim two recorders were
# running when only one was — the sort of scary-but-wrong reading this whole
# tool exists to avoid.
running = []
for p in glob.glob("/proc/[0-9]*/cmdline"):
pid = p.split("/")[2]
if pid == str(os.getpid()):
continue
cmd = (perf._read(p) or "").replace("\x00", " ")
if "gig-log.py" in cmd and " record" in f" {cmd} ":
running.append(pid)
print(f"unit gig-log.service : {unit}")
print(f"processes : {len(running)} {running}")
p = newest_log(d)
if not p:
print(f"newest log : none in {d}")
return 0
age = time.time() - p.stat().st_mtime
n = sum(1 for _ in p.open())
print(f"newest log : {p}")
print(f" {n} lines, last write {age:.0f}s ago "
f"({'LIVE' if age < 10 else 'stale'})")
print(f"all logs : {len(list(d.glob('gig-*.jsonl')))} files, "
f"{sum(f.stat().st_size for f in d.glob('gig-*.jsonl'))/1e6:.2f} MB")
return 0
def cmd_selftest(seconds: float = 6.0) -> int:
"""Prove the two things that would make this tool a liability: that the
sampler is cheap, and that the parsers actually parse."""
ok = True
print("── parsers ──")
xr = XrunReader()
cases = [
("R 160 0 0 13.6us 462.9us 0.00 0.09 67072 + ardour",
("ardour", 67072)),
("C 49 0 0 --- --- --- --- 0 alsa_output.usb-BEHRINGER_UMC202HD_192k_12345678-00.HiFi__Line__sink",
("alsa_output.usb-BEHRINGER_UMC202HD_192k_12345678-00.HiFi__Line__sink", 0)),
("S ID QUANT RATE WAIT BUSY W/Q B/Q ERR FORMAT NAME", None),
("R 128 512 48000 20.1us 55.0us 0.02 0.05 3 S32LE 2 48000 SuperCollider",
("SuperCollider", 3)),
]
for line, want in cases:
got = xr.parse_line(line)
flag = "ok " if got == want else "FAIL"
ok &= got == want
print(f" {flag} pw-top {str(want):<28} <- {line[:40]!r}")
md = MidiReader(port="0:0")
now = 1000.0
for line in [" 24:0 Control change 0, controller 13, value 64",
" 24:0 Control change 0, controller 13, value 90",
" 24:0 Note on 0, note 60, velocity 100",
" 24:0 Note on 0, note 62, velocity 100"]:
if not md.feed(line, now):
ok = False
print(f" FAIL midi did not parse {line!r}")
cc, disc = md.drain()
coalesced = len(cc) == 1 and cc[0]["n"] == 2 and cc[0]["lo"] == 64 and cc[0]["hi"] == 90
print(f" {'ok ' if coalesced else 'FAIL'} 2 CC on one controller -> 1 line "
f"n=2 range 64-90")
print(f" {'ok ' if len(disc) == 2 else 'FAIL'} 2 notes -> 2 lines (notes are "
f"never coalesced)")
ok &= coalesced and len(disc) == 2
print(f"\n── cost over {seconds:.0f}s of real sampling ──")
r = Recorder(out_dir=Path(os.environ.get("TMPDIR", "/tmp")) / "gig-log-selftest",
hz=1.0)
# START the readers. A cost measurement that leaves the drain threads idle is
# flattering itself — the pw-top pipe is ~8 tables/s and parsing it is most of
# the work. Measure what actually runs.
for t in (r.xr, r.mid):
if t:
t.start()
time.sleep(0.5)
me = str(os.getpid())
c0 = _proc_cpu_rss(me)
t0 = time.monotonic()
for _ in range(int(seconds)):
r.sample()
time.sleep(1.0)
dt = time.monotonic() - t0
c1 = _proc_cpu_rss(me)
hz = os.sysconf("SC_CLK_TCK") or 100
pct = 100 * (c1[0] - c0[0]) / hz / dt
for t in (r.xr, r.mid):
if t:
t.stop()
r.f.close()
print(f" sampler + both readers : {pct:.2f}% of one core over {dt:.1f}s")
print(f" pw-top lines parsed : {r.xr.lines if r.xr else 0} "
f"({(r.xr.lines/dt if r.xr else 0):.0f}/s)")
print(f" rss : {c1[1]*4096/1e6:.1f} MB")
budget = pct < 3.0
print(f" {'ok ' if budget else 'FAIL'} under the 3%-of-one-core budget "
f"(the observer must not perturb)")
ok &= budget
print(f"\n{'PASS' if ok else 'FAIL'}")
return 0 if ok else 1
# --------------------------------------------------------------------------- #
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(
prog="gig-log.py", description=__doc__.split("\n")[0],
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--dir", type=Path, default=LOG_DIR, help="log directory")
sub = ap.add_subparsers(dest="cmd")
rec = sub.add_parser("record", help="run the 1 Hz recorder")
rec.add_argument("--hz", type=float, default=1.0)
rec.add_argument("--no-xruns", action="store_true",
help="skip pw-top (no PipeWire client at all)")
rec.add_argument("--no-midi", action="store_true", help="skip aseqdump")
rec.add_argument("--midi-port", help="ALSA seq port (default: find the LCXL)")
rec.add_argument("--seconds", type=float, help="stop after N seconds")
mk = sub.add_parser("mark", help="annotate the running session")
mk.add_argument("label", nargs="+")
rp = sub.add_parser("report", help="render a log")
rp.add_argument("file", nargs="?", type=Path)
sub.add_parser("status", help="is it running, what is it writing")
ins = sub.add_parser("install", help="systemd --user unit")
ins.add_argument("--no-enable", action="store_true")
st = sub.add_parser("selftest", help="prove the parsers and the cost")
st.add_argument("--seconds", type=float, default=6.0)
a = ap.parse_args(argv)
if a.cmd == "mark":
return cmd_mark(" ".join(a.label), a.dir)
if a.cmd == "report":
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)
if a.cmd == "status":
return cmd_status(a.dir)
if a.cmd == "install":
return cmd_install(enable=not a.no_enable)
if a.cmd == "selftest":
return cmd_selftest(a.seconds)
if a.cmd == "record" or a.cmd is None:
hz = getattr(a, "hz", 1.0)
r = Recorder(out_dir=a.dir, hz=hz,
xruns=not getattr(a, "no_xruns", False),
midi=not getattr(a, "no_midi", False),
midi_port=getattr(a, "midi_port", None))
signal.signal(signal.SIGTERM, r.stop)
signal.signal(signal.SIGINT, r.stop)
if getattr(a, "seconds", None):
threading.Timer(a.seconds, r.stop).start()
return r.run()
ap.print_help()
return 2
if __name__ == "__main__":
sys.exit(main())
"""Tests for tools/gig-log.py — the session recorder.
Every positive case here is a real line captured from this machine's `pw-top -b`
and `aseqdump`, not an invented format. The three properties that actually matter:
1. the recorder must NOT be a firehose (CC coalescing), but must never lose a note;
2. counters must be reported as deltas from an HONEST baseline;
3. it must degrade to nothing when the gear it watches is absent.
"""
from __future__ import annotations
import importlib.util
import json
import os
import time
from pathlib import Path
import pytest
SPEC = importlib.util.spec_from_file_location(
"gig_log", Path(__file__).resolve().parents[1] / "gig-log.py")
gl = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(gl)
# --------------------------------------------------------------------------- #
# pw-top parsing
# --------------------------------------------------------------------------- #
# Real lines. Note the varying FORMAT column: empty for MIDI/idle nodes, three
# tokens for a running audio node. That is why ERR is indexed from the left (always
# token 8) and NAME from the right (always last) and never from the middle.
PWTOP_HEADER = ("S ID QUANT RATE WAIT BUSY W/Q B/Q ERR "
"FORMAT NAME ")
PWTOP_ROWS = [
("R 160 0 0 13.6us 462.9us 0.00 0.09 67072 "
"+ ardour", ("ardour", 67072)),
("C 49 0 0 --- --- --- --- 0 "
"alsa_output.usb-BEHRINGER_UMC202HD_192k_12345678-00.HiFi__Line__sink",
("alsa_output.usb-BEHRINGER_UMC202HD_192k_12345678-00.HiFi__Line__sink", 0)),
("I 29 0 0 0.0us 0.0us ??? ??? 0 "
"Dummy-Driver", ("Dummy-Driver", 0)),
("R 128 512 48000 20.1us 55.0us 0.02 0.05 3 S32LE 2 48000 "
"SuperCollider", ("SuperCollider", 3)),
]
@pytest.mark.parametrize("line,want", PWTOP_ROWS)
def test_pwtop_rows_parse(line, want):
assert gl.XrunReader().parse_line(line) == want
def test_the_pwtop_header_is_not_mistaken_for_a_node():
"""'S' is also a valid state char, so the header must be rejected on the ID
column not the state column."""
assert gl.XrunReader().parse_line(PWTOP_HEADER) is None
@pytest.mark.parametrize("junk", [
"", "\n", "Waiting for data.",
"R 160 0", # truncated row
"R abc 0 0 --- --- --- --- 0 thing", # bad id
])
def test_pwtop_junk_is_ignored(junk):
assert gl.XrunReader().parse_line(junk) is None
def test_a_node_name_with_a_plus_prefix_is_stripped():
"""A linked node is printed as '+ ardour'; the '+' is presentation, not name."""
name, _ = gl.XrunReader().parse_line(PWTOP_ROWS[0][0])
assert name == "ardour"
# --------------------------------------------------------------------------- #
# the baseline rule — this is the bug the first report caught
# --------------------------------------------------------------------------- #
def _warm(xr: gl.XrunReader, name: str, value: int, zeros: int = 1) -> None:
"""Feed `zeros` of pw-top's zero-filled tables, then fill the warm-up window
with the node's real lifetime count."""
for _ in range(zeros):
xr._baseline(name, 0)
for _ in range(xr.WARMUP):
xr._baseline(name, value)
def test_pwtops_zero_filled_first_table_does_not_become_the_baseline():
"""Regression #1: baselining on FIRST sight captured 0, so Ardour's lifetime
67072 was reported as 67072 xruns in a 25-second idle session."""
xr = gl.XrunReader()
_warm(xr, "ardour", 67072, zeros=1)
assert xr.deltas()["ardour"] == 0
assert xr.total_delta() == 0
def test_TWO_zero_filled_tables_still_do_not_become_the_baseline():
"""Regression #2, and the reason the rule is max-over-a-warm-up rather than
'the second value': pw-top emits a non-deterministic NUMBER of zero-filled
tables. A rule that is usually right is the worst kind here, because the one
bad reading looks exactly like a disaster."""
xr = gl.XrunReader()
_warm(xr, "ardour", 67123, zeros=2)
assert xr.deltas()["ardour"] == 0
@pytest.mark.parametrize("zeros", [0, 1, 2, 3, 4])
def test_the_baseline_survives_any_number_of_leading_zero_tables(zeros):
xr = gl.XrunReader()
_warm(xr, "ardour", 67134, zeros=zeros)
assert xr.deltas()["ardour"] == 0
def test_a_real_xrun_after_the_warmup_is_counted():
xr = gl.XrunReader()
_warm(xr, "SuperCollider", 3)
xr._baseline("SuperCollider", 7)
assert xr.deltas()["SuperCollider"] == 4
assert xr.total_delta() == 4
def test_the_baseline_is_frozen_after_the_warmup_so_a_rise_is_never_absorbed():
"""If the baseline kept tracking upward, xruns would silently become the new
normal and the session would always read as clean."""
xr = gl.XrunReader()
_warm(xr, "ardour", 100)
for v in (105, 110, 120, 130):
xr._baseline("ardour", v)
assert xr.deltas()["ardour"] == 30
def test_a_node_seen_only_once_reports_zero_not_its_history():
xr = gl.XrunReader()
xr._baseline("ardour", 67072)
assert xr.deltas()["ardour"] == 0
def test_a_recreated_node_rebaselines_instead_of_going_negative():
"""Destroy/recreate resets the node's counter. A negative delta is never a
real reading, so re-baseline."""
xr = gl.XrunReader()
_warm(xr, "ardour", 900)
xr._baseline("ardour", 950)
assert xr.deltas()["ardour"] == 50
xr._baseline("ardour", 2) # node came back fresh
assert xr.deltas()["ardour"] == 0
xr._baseline("ardour", 9)
assert xr.deltas()["ardour"] == 7
def test_total_delta_sums_across_nodes_and_never_goes_negative():
xr = gl.XrunReader()
for name in ("ardour", "SuperCollider"):
_warm(xr, name, 10)
xr._baseline(name, 15)
assert xr.total_delta() == 10
xr._baseline("ardour", 1) # mid-flight re-baseline
assert xr.total_delta() >= 0
def test_ardours_idle_xrun_drift_is_baselined_away():
"""Measured on this machine: Ardour's ERR climbs ~6/min even with nothing
playing (67072 -> 67123 -> 67134 across three captures minutes apart). Only
the session delta is ours, so an idle drift baseline must read as zero."""
xr = gl.XrunReader()
_warm(xr, "ardour", 67072, zeros=2)
assert xr.deltas()["ardour"] == 0
# --------------------------------------------------------------------------- #
# MIDI: coalesce state, never drop events
# --------------------------------------------------------------------------- #
def _reader():
return gl.MidiReader(port="0:0")
def test_a_fader_sweep_collapses_to_one_line_carrying_the_whole_move():
"""The #71/#72 lesson applied to the logger itself: 128 events on one control
is ONE line, and that line still tells you where the fader started, ended, and
how far it travelled."""
md = _reader()
for v in range(128):
assert md.feed(f" 24:0 Control change 0, controller 13, value {v}",
1000.0 + v / 128)
cc, disc = md.drain()
assert len(cc) == 1
e = cc[0]
assert (e["cc"], e["n"], e["v0"], e["v1"], e["lo"], e["hi"]) == (13, 128, 0, 127, 0, 127)
assert disc == []
def test_a_sweep_up_and_back_down_records_the_extremes_not_just_the_endpoints():
md = _reader()
for v in list(range(0, 128)) + list(range(127, -1, -1)):
md.feed(f" 24:0 Control change 0, controller 20, value {v}", 1000.0)
cc, _ = md.drain()
e = cc[0]
assert (e["v0"], e["v1"]) == (0, 0) # started and ended at the bottom...
assert (e["lo"], e["hi"]) == (0, 127) # ...but the log still shows the travel
def test_distinct_controls_never_collapse_into_each_other():
md = _reader()
for cc_num in (13, 14, 15, 77):
md.feed(f" 24:0 Control change 0, controller {cc_num}, value 64", 1.0)
cc, _ = md.drain()
assert sorted(e["cc"] for e in cc) == [13, 14, 15, 77]
def test_the_same_controller_on_a_different_channel_is_a_different_control():
md = _reader()
md.feed(" 24:0 Control change 0, controller 13, value 1", 1.0)
md.feed(" 24:0 Control change 5, controller 13, value 2", 1.0)
cc, _ = md.drain()
assert len(cc) == 2
assert {e["ch"] for e in cc} == {0, 5}
def test_the_same_controller_on_a_different_port_is_a_different_control():
"""Two surfaces are plugged in (LCXL 24:0, KeyStep 16:0) — a CC1 from each is
two controls, not one."""
md = _reader()
md.feed(" 24:0 Control change 0, controller 1, value 1", 1.0)
md.feed(" 16:0 Control change 0, controller 1, value 2", 1.0)
cc, _ = md.drain()
assert len(cc) == 2
assert {e["p"] for e in cc} == {"24:0", "16:0"}
def test_every_note_survives_a_flood_of_ccs():
"""A CC is a state and may be superseded; a note is an EVENT and dropping one
loses a thing that happened. 200 CCs must not shadow the 3 notes."""
md = _reader()
for v in range(200):
md.feed(f" 24:0 Control change 0, controller 13, value {v % 128}", 1.0)
for n in (36, 38, 42):
md.feed(f" 16:0 Note on 0, note {n}, velocity 100", 1.0)
cc, disc = md.drain()
assert len(cc) == 1
assert [d["note"] for d in disc] == [36, 38, 42]
assert all(d["on"] for d in disc)
def test_note_off_is_kept_and_distinguished_from_note_on():
md = _reader()
md.feed(" 16:0 Note on 0, note 60, velocity 100", 1.0)
md.feed(" 16:0 Note off 0, note 60, velocity 0", 2.0)
_, disc = md.drain()
assert [d["on"] for d in disc] == [True, False]
def test_pitch_bend_is_coalesced_like_a_continuous_control():
md = _reader()
for v in (-8192, 0, 8191):
assert md.feed(f" 16:0 Pitch bend 0, value {v}", 1.0)
cc, _ = md.drain()
assert len(cc) == 1
assert cc[0]["k"] == "pb" and "cc" not in cc[0]
assert (cc[0]["lo"], cc[0]["hi"]) == (-8192, 8191)
def test_draining_twice_does_not_replay_the_first_window():
md = _reader()
md.feed(" 24:0 Control change 0, controller 13, value 5", 1.0)
assert len(md.drain()[0]) == 1
assert md.drain() == ([], [])
def test_event_count_counts_raw_events_not_written_lines():
"""The report divides the two to show the coalescing ratio, so `events` has to
stay the RAW count."""
md = _reader()
for v in range(50):
md.feed(f" 24:0 Control change 0, controller 13, value {v}", 1.0)
md.drain()
assert md.events == 50
@pytest.mark.parametrize("junk", [
"Waiting for data. Press Ctrl+C to end.",
" Source Event Ch Data",
"",
" 24:0 Program change 0, program 5", # a kind we don't record
])
def test_non_event_lines_are_reported_as_not_events(junk):
assert gl.MidiReader(port="0:0").feed(junk, 1.0) is False
# --------------------------------------------------------------------------- #
# procfs sampling
# --------------------------------------------------------------------------- #
def _stat_line(comm: str, utime: int, stime: int, rss: int) -> str:
"""A synthetic /proc/<pid>/stat: field 1 pid, 2 comm, 3 state, then 3..52."""
fields = [str(100 + i) for i in range(3, 53)] # fields 3..52, distinct values
fields[3 - 3] = "S"
fields[14 - 3] = str(utime)
fields[15 - 3] = str(stime)
fields[24 - 3] = str(rss)
return f"42 ({comm}) " + " ".join(fields)
def test_proc_stat_parsing_survives_a_comm_containing_spaces_and_parens():
"""pulsar's renderer comms look like `(pulsar) --type=renderer`. Splitting on
whitespace from the left shifts every field, so utime/rss come back as garbage
from some other column — and a logger that silently misreports CPU is worse
than none. Split after the LAST ')'.
"""
assert gl.parse_proc_stat(_stat_line("my (weird) proc", 1234, 567, 98765)) \
== (1234 + 567, 98765)
def test_proc_stat_parsing_handles_an_ordinary_comm():
assert gl.parse_proc_stat(_stat_line("scsynth", 10, 5, 400)) == (15, 400)
@pytest.mark.parametrize("raw", ["", "garbage", "42 (x) S", "42 (x)"])
def test_proc_stat_parsing_returns_none_on_a_short_or_broken_line(raw):
assert gl.parse_proc_stat(raw) is None
def test_a_dead_pid_reads_as_none_not_as_zero():
assert _proc_cpu_rss_of_nothing() is None
def _proc_cpu_rss_of_nothing():
return gl._proc_cpu_rss("999999999")
def test_gear_watch_samples_the_real_machine_without_forking():
gw = gl.GearWatch()
first = gw.sample()
assert set(first) == set(gl.GEAR)
time.sleep(0.15)
second = gw.sample()
for comm, entry in second.items():
assert "up" in entry
if entry["up"]:
assert entry["rss"] > 0
assert entry.get("cpu", 0) >= 0
def test_gear_watch_reports_a_missing_process_as_down_not_as_zero_percent():
"""A dead scsynth reported as '0% cpu' reads like a healthy idle one. It must
read as absent — that distinction is the whole reason to log gear at all."""
gw = gl.GearWatch(gear={"definitely-not-a-real-process-xyz": False})
out = gw.sample()
assert out["definitely-not-a-real-process-xyz"] == {"up": False}
def test_package_throttle_sum_is_readable_rootless():
assert gl._package_throttle_sum() >= 0
# --------------------------------------------------------------------------- #
# the record → mark → report round trip
# --------------------------------------------------------------------------- #
def test_a_recorded_session_round_trips_through_the_report_loader(tmp_path):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
for _ in range(3):
r._write(r.sample())
r._write({"t": time.time(), "k": "end", "samples": 3, "midi_events": 0,
"xrun": None})
r.f.close()
hdr, samples, events, marks, end = gl.load(r.path)
assert hdr["k"] == "hdr" and hdr["v"] == 1
assert len(samples) == 3
assert end["samples"] == 3
# the header must carry the baselines, or the deltas in the samples are unreadable
assert "base_pkg_throttle" in hdr and "base_core_throttle" in hdr
# and the wall clock, which is how the log lines up with the Ardour take
assert hdr["iso"].startswith("20")
def test_mark_appends_to_the_newest_log_and_survives_the_loader(tmp_path):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
r._write(r.sample())
r.f.close()
assert gl.cmd_mark("gimme acid drop", tmp_path) == 0
_, _, _, marks, _ = gl.load(r.path)
assert [m["label"] for m in marks] == ["gimme acid drop"]
def test_mark_with_no_session_fails_loudly_instead_of_creating_an_orphan(tmp_path):
assert gl.cmd_mark("nobody is recording", tmp_path) == 1
assert list(tmp_path.glob("*.jsonl")) == []
def test_the_loader_skips_a_torn_final_line(tmp_path):
"""A `kill -9` mid-write leaves half a line. That must cost one sample, not
the whole log."""
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
r._write(r.sample())
r.f.write('{"t":1,"k":"s","pkg":') # torn
r.f.close()
hdr, samples, _, _, _ = gl.load(r.path)
assert hdr["k"] == "hdr" and len(samples) == 1
def test_report_renders_a_real_session_without_raising(tmp_path, capsys):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
for _ in range(4):
r._write(r.sample())
gl.cmd_mark("marker", tmp_path)
r.f.close()
assert gl.cmd_report(r.path) == 0
out = capsys.readouterr().out
assert "GIG LOG" in out
assert "align the Ardour take" in out
assert "marker" in out
# absolute counters must never be presented as session damage
assert "absolute counts are meaningless" in out
def test_report_on_a_header_only_log_refuses_rather_than_dividing_by_zero(tmp_path):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
r.f.close()
assert gl.cmd_report(r.path) == 1
def test_prune_deletes_old_logs_and_keeps_todays(tmp_path):
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
old = tmp_path / "gig-20200101-000000.jsonl"
old.write_text("{}\n")
os.utime(old, (0, time.time() - 40 * 86400))
assert r.prune(days=30) == 1
assert not old.exists()
assert r.path.exists()
r.f.close()
def test_newest_log_picks_the_latest_by_mtime(tmp_path):
a = tmp_path / "gig-20260101-000000.jsonl"
b = tmp_path / "gig-20260202-000000.jsonl"
for p in (a, b):
p.write_text("{}\n")
os.utime(a, (0, 1000))
os.utime(b, (0, 2000))
assert gl.newest_log(tmp_path) == b
def test_newest_log_on_an_empty_dir_is_none(tmp_path):
assert gl.newest_log(tmp_path) is None
# --------------------------------------------------------------------------- #
# degradation: absent gear must cost nothing
# --------------------------------------------------------------------------- #
def test_the_recorder_runs_with_both_children_disabled(tmp_path):
"""If pw-top and aseqdump are missing, the thermal timeline still records.
Partial data beats no data, and beats a crash on stage."""
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
assert r.xr is None and r.mid is None
s = r.sample()
assert "pkg" in s and "gear" in s
assert "xrun" not in s
hdr = r.header()
assert hdr["xruns"] is False and hdr["midi"] is False
r.f.close()
def test_a_sample_is_json_serialisable_even_when_every_reading_is_none(tmp_path):
"""On a machine with no coretemp/dell_smm hwmon every field is None. The line
must still be valid JSONL."""
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r.th.coretemp = r.th.dell = r.th._pkg_input = None
json.loads(json.dumps(r.sample()))
r.f.close()
def test_find_seq_port_never_returns_the_hui_port():
"""The LCXL exposes two ports; 'HUI' is the Mackie-emulation one and carries
none of our control traffic."""
got = gl.find_seq_port("Launch Control XL")
if got is not None:
assert ":" in got
def test_the_midi_reader_re_resolves_its_port_by_name():
"""The device has already moved 20:0 -> 24:0 across a replug, so the port must
be looked up on every (re)connect, not cached at startup."""
md = gl.MidiReader()
assert md.fixed is None
src = (Path(__file__).resolve().parents[1] / "gig-log.py").read_text()
body = src[src.index("class MidiReader"):src.index("# the recorder")]
assert "find_seq_port(self.want)" in body
# --------------------------------------------------------------------------- #
# presentation
# --------------------------------------------------------------------------- #
def test_spark_keeps_the_time_axis_and_survives_holes():
assert gl.spark([]) == "(no data)"
assert "flat" in gl.spark([5, 5, 5])
s = gl.spark([1, 2, 3, None, 4])
assert len(s) == 4 and s[0] == gl.SPARK[0] and s[-1] == gl.SPARK[-1]
def test_spark_buckets_by_max_so_a_single_spike_cannot_be_averaged_away():
"""A one-second xrun burst inside a 40-minute set is the whole point of the
log. Bucketing by mean would erase it."""
vals = [0] * 500 + [99] + [0] * 500
assert gl.SPARK[-1] in gl.spark(vals, width=60)
def test_spark_respects_the_width():
assert len(gl.spark(list(range(1000)), width=60)) == 60
@pytest.mark.parametrize("secs,want", [
(0, "0:00"), (61, "1:01"), (599, "9:59"), (3600, "1:00:00"), (3725, "1:02:05"),
])
def test_hms(secs, want):
assert gl.hms(secs) == want
def test_the_report_flags_ardour_owned_ccs_in_the_surface_table(tmp_path, capsys):
"""CC 77-84 are MIDI-learned to Ardour's track gains. If they show up in a log
the reader needs to know that traffic moved FADERS, not Tidal params."""
r = gl.Recorder(out_dir=tmp_path, hz=50.0, xruns=False, midi=False)
r._write(r.header())
r._write(r.sample())
t = time.time()
r._write({"t": t, "k": "cc", "p": "24:0", "ch": 0, "cc": 80,
"n": 40, "v0": 0, "v1": 100, "lo": 0, "hi": 100})
r._write({"t": t, "k": "cc", "p": "24:0", "ch": 0, "cc": 13,
"n": 5, "v0": 64, "v1": 70, "lo": 64, "hi": 70})
r.f.close()
gl.cmd_report(r.path)
out = capsys.readouterr().out
assert "ARDOUR-OWNED" in out
lines = [ln for ln in out.splitlines() if ln.strip().startswith("13 ")]
assert lines and "ARDOUR-OWNED" not in lines[0]
def test_the_report_states_zero_xruns_as_a_positive_verdict(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
r.f.write(json.dumps({"t": time.time(), "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": 0}) + "\n")
r.f.close()
# rewrite the header with xruns enabled
lines = r.path.read_text().splitlines()
lines[0] = json.dumps(hdr)
r.path.write_text("\n".join(lines) + "\n")
gl.cmd_report(r.path)
out = capsys.readouterr().out
assert "ZERO" in out and "the audio path held" 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