Commit ec722018 by PLN (Algolia)

rig health: with the rig up, nothing recording is a failure

recording_check reads the newest .opus segment mtime (never systemd): rig up and
no segment landing in 30 s is FAIL, naming the cause (unit not installed, no
segment, stale segment). The tray offers a one-click start of the recorder unit
only while that check fails.
parent 9d16dcdd
......@@ -524,6 +524,13 @@ class PerfTray:
lambda: QProcess.startDetached("systemctl", ["--user", "start", "lcxl3-driver.service"]))
self.fix_driver_action.setVisible(False)
self.menu.addAction(self.fix_driver_action)
# Same rule for the black box: with the rig up and nothing landing on disk, one click
# starts the recorder unit. Visible only while the recording check is failing.
self.fix_recorder_action = QAction("● start the recorder (black box)", self.menu)
self.fix_recorder_action.triggered.connect(
lambda: QProcess.startDetached("systemctl", ["--user", "start", "gig-record.service"]))
self.fix_recorder_action.setVisible(False)
self.menu.addAction(self.fix_recorder_action)
self.menu.aboutToShow.connect(self.refresh)
self.menu.addSeparator()
......@@ -963,6 +970,7 @@ class PerfTray:
a.setText(txt)
self.health_action.setText(hs["row"])
self.fix_driver_action.setVisible("start_driver" in hs["fixes"])
self.fix_recorder_action.setVisible("start_recorder" in hs["fixes"])
# The Qt-drawn popup gets the Ship's Bridge look (armada/DESIGN.md): tonal dark
......
......@@ -198,6 +198,48 @@ def surface_check(cards: str | None = None, seq: str | None = None) -> tuple[str
return FAIL, "LCXL3 on the bus but lcxl3-driver publishes no port: no faders reach Ardour"
def recordings_dir() -> Path:
return Path(os.environ.get("GIG_RECORD_DIR") or Path(__file__).resolve().parent.parent / "recordings")
UNIT_PATH = Path(os.path.expanduser("~/.config/systemd/user/gig-record.service"))
RECORDING_STALE_S = 30.0 # opus segments are flushed in ~13 s steps at 160k: older = not landing
def recording_check(gear: dict | None, rec_dir: Path | None = None, now=time.time,
unit_path: Path | None = None) -> tuple[str, str]:
"""With the rig up, the black box (gig-record) must be landing audio on disk.
A missed gig recording cannot be recovered, and an unneeded one costs 72 MB an
hour, so the asymmetry decides: rig up and nothing growing is FAIL, loud, with
the cause named. 2026-10-11: the unit was not even installed, two hours of
playing were recorded nowhere, and the old gate showed it as one ignorable WARN.
"Landing" is read off the file (newest segment mtime), never off systemd: an
active unit that records silence or nothing is exactly the failure."""
gear = gear or {}
if not (gear.get("sclang", {}).get("up") and gear.get("scsynth", {}).get("up")):
return NA, "SuperDirt not running: nothing to record"
rec_dir = rec_dir or recordings_dir()
newest = 0.0
try:
with os.scandir(rec_dir) as it:
for e in it:
if e.name.endswith(".opus"):
newest = max(newest, e.stat().st_mtime)
except OSError:
pass
age = now() - newest if newest else None
if age is not None and age <= RECORDING_STALE_S:
return OK, f"black box capturing (newest segment {age:.0f}s old)"
if not (unit_path or UNIT_PATH).exists():
why = "gig-record.service is NOT INSTALLED (tools/rig-install.sh links it)"
elif age is None:
why = "no recording segment exists"
else:
why = f"newest segment is {age:.0f}s old"
return FAIL, f"the rig is up and nothing is recording the master: {why}"
def superdirt_check(gear: dict | None) -> tuple[str, str]:
"""sclang alive with scsynth dead is the 2026-08-01 failure (systemd stays
active because sclang is the MainPID)."""
......@@ -281,7 +323,8 @@ class Health:
# -- the checks that feed one tick ------------------------------------ #
def probe(self, gear: dict | None) -> dict[str, tuple[str, str]]:
checks = {"surface": surface_check(), "superdirt": superdirt_check(gear)}
checks = {"surface": surface_check(), "superdirt": superdirt_check(gear),
"recording": recording_check(gear)}
if self.clock is not None:
checks["audio_clock"] = self.clock.verdict()
return checks
......@@ -435,7 +478,8 @@ def tray_summary(h: dict) -> dict:
if overall == "down":
return {"badge": "down", "row": "rig: DOWN · " + ", ".join(n for n, _ in bad),
"tip": [f"rig: {n}: {c['detail']}" for n, c in bad],
"fixes": ["start_driver"] if any(n == "surface" for n, _ in bad) else []}
"fixes": (["start_driver"] if any(n == "surface" for n, _ in bad) else [])
+ (["start_recorder"] if any(n == "recording" for n, _ in bad) else [])}
if overall == "stale":
return {"badge": "stale", "row": "rig: observer stale",
"tip": [h.get("detail", "health.json is stale")], "fixes": []}
......
......@@ -329,3 +329,74 @@ def test_observer_alive_is_forkless_and_checks_the_command():
assert rh.observer_alive(10**9) is False # no such pid
assert rh.observer_alive(os.getpid()) is False # alive, but not a gig-log
assert rh.observer_alive("not-a-pid") is False
# --------------------------------------------------------------------------- #
# recording: with the rig up, the black box must be landing audio
# --------------------------------------------------------------------------- #
RIG_UP = {"sclang": {"up": True}, "scsynth": {"up": True}}
def _seg(tmp_path, age_s, name="2026-10-11_150000.opus"):
import os, time
p = tmp_path / name
p.write_bytes(b"x" * 100)
t = time.time() - age_s
os.utime(p, (t, t))
return p
def test_a_cold_rig_has_nothing_to_record(tmp_path):
assert rh.recording_check({"sclang": {"up": False}, "scsynth": {"up": False}}, tmp_path)[0] == rh.NA
assert rh.recording_check(None, tmp_path)[0] == rh.NA
def test_a_fresh_segment_means_the_black_box_is_landing_audio(tmp_path):
_seg(tmp_path, 4)
state, detail = rh.recording_check(RIG_UP, tmp_path)
assert state == rh.OK and "capturing" in detail
def test_rig_up_and_no_segment_is_a_failure_and_names_the_missing_unit(tmp_path):
"""2026-10-11: the unit was not installed and two hours of playing were recorded
nowhere. The detail must say THAT, not just 'no segment'."""
state, detail = rh.recording_check(RIG_UP, tmp_path, unit_path=tmp_path / "absent.service")
assert state == rh.FAIL and "NOT INSTALLED" in detail and "nothing is recording" in detail
def test_rig_up_unit_installed_but_no_segment_says_so(tmp_path):
unit = tmp_path / "gig-record.service"
unit.write_text("")
state, detail = rh.recording_check(RIG_UP, tmp_path, unit_path=unit)
assert state == rh.FAIL and "no recording segment exists" in detail
def test_a_stale_segment_is_a_failure_even_though_the_unit_is_active(tmp_path):
"""Landing is read off the file, never off systemd."""
unit = tmp_path / "u.service"
unit.write_text("")
_seg(tmp_path, 300)
state, detail = rh.recording_check(RIG_UP, tmp_path, unit_path=unit)
import re
assert state == rh.FAIL and re.search(r"segment is 30\ds old", detail), detail
def test_only_opus_segments_count_as_a_recording(tmp_path):
(tmp_path / "recorder.log").write_text("fresh log, no audio")
unit = tmp_path / "u.service"
unit.write_text("")
assert rh.recording_check(RIG_UP, tmp_path, unit_path=unit)[0] == rh.FAIL
def test_the_health_probe_includes_recording(tmp_path, monkeypatch):
monkeypatch.setenv("GIG_RECORD_DIR", str(tmp_path))
_seg(tmp_path, 2)
h = rh.Health(path=tmp_path / "h.json", async_evidence=False)
assert h.probe(RIG_UP)["recording"][0] == rh.OK
def test_a_recording_failure_offers_to_start_the_recorder_and_nothing_else():
s = rh.tray_summary(_doc("down", recording=(rh.FAIL, "nothing is recording")))
assert s["badge"] == "down" and s["fixes"] == ["start_recorder"] and "recording" in s["row"]
s2 = rh.tray_summary(_doc("down", surface=(rh.FAIL, "x"), recording=(rh.FAIL, "y")))
assert s2["fixes"] == ["start_driver", "start_recorder"]
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