diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index e66efda..b6d78c0 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -772,247 +772,6 @@ def armed_kill(ctx): "respawned with the latch still locked, button dark", ev["expected"]["zero_at_s"], zero_at) -@test("laser.verdict-cut", title="A blocked verdict mid-cut locks the latch and holds; the clean " - "verdict resumes with no press", - subsystem="laser", kind="live", mode="grbl", est_min=4, - covers=_LASER_COVERS, - requires=["laser.emission-witness"], actions=["button"], - steps=["Scrap under the head with 40 mm of free +X travel; lid closed; exhaust on.", - "Press the physical button when it lights white (the arm). Nothing else: the test " - "takes the cooling verdict away itself and gives it back."], - description="A 40 mm line at constant power (M3 S400/F300, 8 s). About 1.5 s in, the test " - "pauses the machine-services daemon for 3.0 s, so the cooling verdict the " - "controller reads goes stale: the controller's pause tier. The controller must " - "hold the job (grbl Hold) under the still-open armed window with the stream " - "masked dark: the gated LASER_ON output (cnc/laser_on), read about 3300 times a " - "second, reads the laser off in every read that falls inside the hold from 0.3 s " - "after the first Hold:0 (the pause tier's lit deceleration still playing out of " - "the kernel's 200 ms queue) until the cut resumes, at least 500 reads, and the " - "same reads saw the cut lit in the second before the pause. It must not write " - "the latch: a lock sets the hardware button " - "latch, which only a press clears, so both bit 3 (the SoC lock) and bit 2 " - "(the button latch) of interlock_circuit stay clear through the whole run. " - "When the daemon returns (the verdict fresh, clean, resume_ok), the controller " - "must resume the cut lit from where it stopped, with no button press and no " - "prompt; the M2 program end then disarms as usual. The daemon's own dead-man " - "(5 s of silence) is not reached: the controller keeps reporting through the " - "pause. The beam detector witnesses the cut on both sides of the hold.") -def verdict_cut(ctx): - import os as _os - import signal as _signal - ev = ctx.evidence - fc = ctx.forgectrl - # The daemon is frozen long enough for the verdict the controller - # caches to expire (VERDICT_MAX_AGE 2.0 s), so the driver's pause tier - # holds the job - but not so long that the engine's report dead-man - # trips on resume. That dead-man locks the laser latch when the - # controller's report is older than REPORT_TIMEOUT_S (5.0 s) while - # armed, which is exactly the "latch stayed locked" this test refuses. - # The last report is up to 1 s old when the freeze starts, so the - # freeze must stay under 4.0 s; 3.0 s clears the stale floor with room - # to see the hold and stays a full second under the dead-man. - PAUSE_S = 3.0 - # The dark judge's witness is the gated LASER_ON output itself - # (cnc/laser_on, 1 = on), read fast. The kernel's sampled count latches - # once a second over the second before, so it reads 0 only once a whole - # window has closed inside the hold, up to 2 s in, and the hold can be - # shorter than that. About 3300 reads a second on the bench reference; - # each read yields the CPU to anything runnable (the controller). - FAST_S = 0.1 - # The pause tier's first deceleration is lit on purpose (a pause leaves - # no gap in the cut), and the controller reports Hold:0 once it has - # produced the deceleration, not once the kernel has played it: the - # driver's queue depth (200 ms, GFSINK_DEPTH_MS's default) and its - # producer lead (10 ms) are still to play. The dark span starts that - # long after the first Hold:0, with margin. - DRAIN_S = 0.3 - MIN_DARK_READS = 500 - MIN_LIT_READS = 10 - - def fast_reads(seconds, t_ref): - """Reads of cnc/laser_on for `seconds`: [start, end] in seconds - from t_ref, then the reads taken, the reads that saw the laser on, - and the reads that failed.""" - start = time.time() - n = on = bad = 0 - while time.time() - start < seconds: - v = hw.sysfs_int("cnc/laser_on") - n += 1 - if v is None: - bad += 1 - elif v: - on += 1 - _os.sched_yield() - return [round(start - t_ref, 3), round(time.time() - t_ref, 3), n, on, bad] - - def dark_span(rows, drain_s): - """The fast reads that fell wholly inside the hold, drain_s or more - after the first Hold:0 row: a burst counts when the rows either side - of it both read Hold, so the hold held through it. [reads, on, - failed, bursts, from_s, to_s], or None without a Hold:0 row. Pure: - tests/test_laser_verdict.py runs it over synthetic trails.""" - hold0 = next((r["t"] for r in rows if r["gstate"].startswith("Hold:0")), None) - if hold0 is None: - return None - reads = on = bad = bursts = 0 - first = last = None - for a, b in zip(rows, rows[1:]): - f = a.get("fast") - if not f or not a["gstate"].startswith("Hold") or not b["gstate"].startswith("Hold"): - continue - if f[0] < hold0 + drain_s: - continue - reads, on, bad, bursts = reads + f[2], on + f[3], bad + f[4], bursts + 1 - first = f[0] if first is None else first - last = f[1] - return [reads, on, bad, bursts, first, last] - - pids = hw.pidof("forgectrl") - ctx.check(pids, "no forgectrl process found to pause") - ev["daemon_pids"] = pids - state_path = "/run/forgefirm/grbl.state" - - def grbl_state_file(): - try: - with open(state_path) as f: - return f.read() - except OSError: - return "" - - def resume_daemon(): - for p in pids: - try: - _os.kill(p, _signal.SIGCONT) - except OSError: - pass - - trail = [] - text = "" - with ctx.grbl() as g, LiveJob(ctx, g): - prepare(ctx, g) - k0 = kernel_start(ctx) - base = sample(ctx) - ctx.check(base, "forgectrl /status or /cool/status unavailable") - ctx.check(not base["emission"], "emission_samples nonzero before the job (%s)", base["emission"]) - ctx.ready(ARM_CUE % "40 mm +X") - job = ["G91", "G21", "M3", "S400", "G1 X40 F300", "M5", "G0 X-40", "G90", "M2"] - smp = arm_and_fire(ctx, g, room="40 mm +X", job=job) - beams = [(smp.get("beam"), smp.get("beam_d"))] - ctx.log("emission live (%s); pausing the daemon in 1 s", smp["emission"]) - # The witness's own proof: over the lit cut the same reads see it lit. - lit_ctl = fast_reads(1.0, time.time()) - ctx.log("fast reads of cnc/laser_on over the lit cut: %d of %d on, %d failed", lit_ctl[3], - lit_ctl[2], lit_ctl[4]) - g.drain() - t0 = time.time() - try: - for p in pids: - _os.kill(p, _signal.SIGSTOP) - ctx.log("daemon paused (SIGSTOP %s) for %.1f s: the verdict goes stale", pids, PAUSE_S) - # No HTTP while the daemon is paused: the controller's socket, - # sysfs and the controller's state file are the witnesses. - while time.time() - t0 < PAUSE_S: - st = g.status_report()["state"] - text += g.drain() - il = hw.sysfs_int("cnc/interlock_circuit") - trail.append({"t": round(time.time() - t0, 2), "gstate": st, "il": il, - "emission": hw.sysfs_int("cnc/laser_on_sampled"), - "armed": '"armed":true' in grbl_state_file()}) - if st.startswith("Hold"): - trail[-1]["fast"] = fast_reads(FAST_S, t0) - else: - time.sleep(0.12) - finally: - resume_daemon() - ctx.log("daemon resumed (SIGCONT) at +%.2f s", time.time() - t0) - # Through the resume and to the end of the cut, sampling the same - # witnesses (the daemon's own once it answers again). - resumed_at = None - end = time.time() + 30 - while time.time() < end: - ctx.checkpoint() - st = g.status_report()["state"] - text += g.drain() - il = hw.sysfs_int("cnc/interlock_circuit") - row = {"t": round(time.time() - t0, 2), "gstate": st, "il": il, - "emission": hw.sysfs_int("cnc/laser_on_sampled"), - "armed": '"armed":true' in grbl_state_file()} - if time.time() - t0 > PAUSE_S + 1.0: - s = sample(ctx) - if s: - row["beam"] = s.get("beam") - beams.append((s.get("beam"), s.get("beam_d"))) - trail.append(row) - if resumed_at is None and st.startswith("Run"): - resumed_at = time.time() - t0 - if st.startswith("Idle") and resumed_at is not None and time.time() - t0 > resumed_at + 3.0: - break - if resumed_at is None and st.startswith("Hold"): - row["fast"] = fast_reads(FAST_S, t0) - else: - time.sleep(0.12) - for r in trail: - ctx.log(" %s", r) - ev["trail"] = trail - ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:") or ln.startswith("ALARM")] - ctx.log("controller: %s", ev["messages"]) - dt = wait_disarm(ctx, 75) - ev["disarm_after_idle_s"] = round(dt, 1) if dt is not None else None - ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the job") - check_kernel_returned(ctx, ev, k0) - - def locked(r): - return r["il"] is not None and bool(r["il"] & IL_LASER_LATCH) - - def button_latched(r): - return r["il"] is not None and bool(r["il"] & IL_BUTTON_LATCH) - - held = [r for r in trail if r["gstate"].startswith("Hold")] - ev.update({"held_at_s": held[0]["t"] if held else None, - "resumed_at_s": round(resumed_at, 2) if resumed_at else None, - "latch_locked_samples": sum(1 for r in trail if locked(r)), - "button_latch_set_samples": sum(1 for r in trail if button_latched(r))}) - ctx.check(held and held[0]["t"] < PAUSE_S, "the stale verdict did not hold the job during the " - "pause (first Hold at %s)", ev["held_at_s"]) - ctx.check("fire masked, job held" in text, "the controller did not report the pause tier") - ctx.check(ev["latch_locked_samples"] == 0, - "the pause tier wrote the latch (locked in %d samples): a lock sets the hardware " - "button latch and the resume would need a press", ev["latch_locked_samples"]) - ctx.check(ev["button_latch_set_samples"] == 0, - "the hardware button latch read SET in %d samples: the resume needed a press", - ev["button_latch_set_samples"]) - ctx.check(all(r["armed"] for r in trail if r["t"] < (resumed_at or PAUSE_S + 5)), - "the armed window closed during the pause") - ctx.check("press the button" not in text, "the resume asked for a button press") - ctx.check(DISARMED_MSG not in text.split("resuming")[0], "the pause tier disarmed the job") - # Dark through the hold: the gate masked the stream. First the witness: - # the reads that judge the dark saw the cut lit before the pause. - ev["fast_lit_control"] = {"reads": lit_ctl[2], "on": lit_ctl[3], "failed": lit_ctl[4]} - ctx.check(lit_ctl[4] == 0 and lit_ctl[3] >= MIN_LIT_READS, - "the fast reads of cnc/laser_on did not see the lit cut before the pause (%d of %d on, " - "%d failed): they cannot witness the dark", lit_ctl[3], lit_ctl[2], lit_ctl[4]) - dark = dark_span(trail, DRAIN_S) - ev["fast_dark"] = (dict(zip(("reads", "on", "failed", "bursts", "from_s", "to_s"), dark)) - if dark else None) - ctx.check(dark is not None, "no Hold:0 was seen, so the hold cannot be judged dark") - ctx.check(dark[2] == 0, "cnc/laser_on could not be read %d times inside the hold", dark[2]) - ctx.check(dark[1] == 0, "the laser was on in %d of %d reads of cnc/laser_on inside the hold " - "(%s s to %s s, %.1f s or more after the first Hold:0)", dark[1], dark[0], dark[4], dark[5], - DRAIN_S) - ctx.check(dark[0] >= MIN_DARK_READS, "only %d reads of cnc/laser_on fell inside the hold after the " - "drain: too few to judge it dark (at least %d)", dark[0], MIN_DARK_READS) - ctx.check(resumed_at is not None, "the clean verdict did not resume the cut") - ctx.check("resuming" in text, "the controller did not report the resume") - after = [r for r in trail if resumed_at and r["t"] >= resumed_at] - ctx.check(any(r["emission"] for r in after), "the resumed cut ran dark (no emission after the resume)") - ctx.check("ALARM" not in text, "an alarm was raised on the pause tier") - ctx.check(dt is not None and dt < 10.0, "the M2 job did not disarm promptly at Idle (%s s)", dt) - beam_witness(ctx, ev, [{"beam": b, "beam_d": d} for b, d in beams], base) - judge_beam(ctx, ev["beam"], "the cut") - check_button_dark(ctx, ev) - ctx.log("PASS: held at +%s s with the latch untouched, the laser off in all %d reads inside the " - "hold (%s s to %s s), resumed lit at +%s s with no press, disarmed %.1f s after Idle", - ev["held_at_s"], dark[0], dark[4], dark[5], ev["resumed_at_s"], dt) DISARMED_MSG = "laser disarmed - latch locked" diff --git a/forgetest/tests/test_laser_verdict.py b/forgetest/tests/test_laser_verdict.py deleted file mode 100644 index f5ca51c..0000000 --- a/forgetest/tests/test_laser_verdict.py +++ /dev/null @@ -1,105 +0,0 @@ -# Copyright 2026 514 LLC d/b/a OpenGlow -# Written by Scott Wiederhold -# https://community.openglow.org -# SPDX-License-Identifier: MIT - -"""The dark judge of laser.verdict-cut over synthetic trails. The judge is -a function inside the test's own body, so a change to it moves that test's -fingerprint and no other laser test's; it is reached here through the test -function's code object. A trail row is a status poll; its `fast` entry is -the burst of cnc/laser_on reads taken after that poll: [start, end, reads, -reads that saw the laser on, reads that failed].""" -import types -import unittest - -from forgetest.suite import laser - -DRAIN_S = 0.3 - - -def dark_span(): - code = next(c for c in laser.verdict_cut.__code__.co_consts - if isinstance(c, types.CodeType) and c.co_name == "dark_span") - assert code.co_freevars == (), "the judge must not close over the test's variables" - return types.FunctionType(code, vars(laser)) - - -def row(t, gstate, on=None, n=350, bad=0): - r = {"t": t, "gstate": gstate} - if on is not None: - r["fast"] = [round(t + 0.01, 3), round(t + 0.11, 3), n, on, bad] - return r - - -class DarkSpanTests(unittest.TestCase): - def setUp(self): - self.judge = dark_span() - - def hold(self, lit_at=(), tail_lit=0): - """Run, then a hold from 2.0 s (Hold:0) polled every 0.14 s with a - burst after each poll, then the resume at 3.4 s. lit_at: the - poll times whose burst saw the laser on; tail_lit: the burst - after the last Hold poll, which the resume ends.""" - rows = [row(1.72, "Run"), row(1.86, "Run")] - t = 2.0 - while t < 3.3: - rows.append(row(round(t, 2), "Hold:0", on=5 if round(t, 2) in lit_at else 0)) - t += 0.14 - rows[-1]["fast"][3] = tail_lit - rows.append(row(3.4, "Run")) - return rows - - def test_a_dark_hold_is_judged_dark(self): - d = self.judge(self.hold(), DRAIN_S) - reads, on, bad, bursts, first, last = d - self.assertEqual((on, bad), (0, 0)) - self.assertGreater(bursts, 0) - self.assertEqual(reads, bursts * 350) - self.assertGreaterEqual(first, 2.0 + DRAIN_S) - self.assertLessEqual(last, 3.4) - - def test_the_lit_deceleration_inside_the_drain_is_not_judged(self): - # the pause tier's first deceleration plays out of the kernel's - # queue after Hold:0: lit bursts inside the drain are expected - d = self.judge(self.hold(lit_at=(2.0, 2.14)), DRAIN_S) - self.assertEqual(d[1], 0) - - def test_emission_after_the_drain_is_caught(self): - # the negative control: a lit read inside the judged span is seen - d = self.judge(self.hold(lit_at=(2.56,)), DRAIN_S) - self.assertEqual(d[1], 5) - - def test_a_burst_the_resume_ends_is_not_judged(self): - # the poll after it reads Run: the cut may have relit inside it - d = self.judge(self.hold(tail_lit=40), DRAIN_S) - self.assertEqual(d[1], 0) - - def test_failed_reads_are_counted(self): - rows = self.hold() - rows[5]["fast"][4] = 3 - self.assertEqual(self.judge(rows, DRAIN_S)[2], 3) - - def test_the_drain_counts_from_the_first_hold_complete(self): - # Hold:1 (decelerating) rows do not start the span: a lit burst - # after a Hold:1 poll but inside the drain of the first Hold:0 is - # not judged - rows = [row(1.8, "Run"), row(1.94, "Hold:1", on=30), row(2.08, "Hold:0", on=12), - row(2.22, "Hold:0", on=0), row(2.36, "Hold:0", on=0), row(2.5, "Hold:0", on=0), - row(2.64, "Hold:0", on=0), row(2.78, "Run")] - reads, on, bad, bursts, first, last = self.judge(rows, DRAIN_S) - self.assertEqual(on, 0) - self.assertGreaterEqual(first, 2.08 + DRAIN_S) - # 2.51 only: 2.37 is inside the drain, 2.65 is ended by the resume - self.assertEqual(bursts, 1) - - def test_no_hold_complete_cannot_be_judged(self): - self.assertIsNone(self.judge([row(1.0, "Run"), row(1.2, "Hold:1", on=0), row(1.4, "Run")], DRAIN_S)) - - def test_a_hold_too_short_for_the_drain_leaves_nothing_to_judge(self): - # the test then fails on too few reads, never passes on none - rows = [row(1.8, "Run"), row(2.0, "Hold:0", on=0), row(2.14, "Hold:0", on=0), row(2.28, "Run")] - self.assertEqual(self.judge(rows, DRAIN_S)[:4], [0, 0, 0, 0]) - - -if __name__ == "__main__": - unittest.main()