From 308a033146c97e03f7486410643e5702d349d4ec Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Thu, 24 Sep 2026 14:20:06 -0400 Subject: [PATCH] laser.verdict-cut judges the hold by the gated output; laser.disarm-in-hold presses at the arm Both tests failed for the operator on the bench reference on 2026-09-24, image 20260923232513, from the harness and not the machine. laser.verdict-cut (00:10:52) needed the kernel's sampled LASER_ON count to read 0 within 2.5 s of the hold. That 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 this hold lasted 1.37 s. Every sample of the hold read bit 0 of interlock_circuit as 1: the gated LASER_ON, active low, dark. The dark judge now reads the gated output itself, cnc/laser_on (1 = on), about 3300 times a second on the bench reference, yielding the CPU between reads (the stream threads run SCHED_FIFO). Its witness is the same reads over the second before the pause, which must see the cut lit (10 reads or more, none failed). The hold is dark when every read that falls wholly inside it, from 0.3 s after the first Hold:0 to the last Hold row, reads off, none failed, and there are at least 500. The 0.3 s is the pause tier's first deceleration, lit on purpose, still playing out of the driver's 200 ms queue and its 10 ms lead when the controller reports Hold:0. The daemon's pause is 3.0 s, as the code already had and the description now says. dark_span() is pure, kept inside the test's function, and tests/test_laser_verdict.py runs it through the function's code object over synthetic trails (8 cases: a dark hold, the lit deceleration inside the drain, emission after it caught, a burst the resume ends, failed reads counted, the drain counted from the first Hold:0, no Hold:0, and a hold too short for the drain). laser.disarm-in-hold (00:06:23, 180 s) pressed the button through ctx.act at once, before the job reached its arm wait, so the press was lost and the move never started. It now uses ctx.arm_press(), which presses when the button lights, and waits for Run. Proof: forgetest's unit tests pass on the host (474 OK, 4 skipped). Only the fingerprints of laser.verdict-cut and laser.disarm-in-hold move. Acceptance: the change is the two catalog tests themselves; both are attended (laser emission, the operator present) and a campaign runs both again. --- forgetest/forgetest/suite/laser.py | 116 ++++++++++++++++++++++---- forgetest/tests/test_laser_verdict.py | 105 +++++++++++++++++++++++ 2 files changed, 203 insertions(+), 18 deletions(-) create mode 100644 forgetest/tests/test_laser_verdict.py diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 8d8c4bb..5114b7f 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -671,8 +671,11 @@ def disarm_in_hold(ctx): stream(g, ["G91", "G21", "M4", "S400", "G1 X40 F300"]) ctx.log("armed; waiting for motion to start (arm + your button press)...") w = Watch(g) - ctx.act("button", "press", text="The button is lit white: the press arms the job and the " - "move starts.", until=w.in_state("Run"), timeout=180, fail=False) + # The press waits for the button to light: pressed before the job + # reaches its arm wait, it is lost, and the move never starts. + ctx.arm_press() + ctx.wait_for(w.in_state("Run"), 180) + ctx.clear_notice() st = w.last["state"] if w.last else None ctx.check(st and st.startswith("Run"), "motion never started (state=%s) - arm refused or no press", st) ctx.log("moving under laser: %s; feed-hold in 2 s", st) @@ -850,11 +853,15 @@ def armed_kill(ctx): "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.5 s, so the cooling verdict the " + "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, so the kernel's LASER_ON sample count reads 0 before the cut " - "resumes, and it must not write the latch: a lock sets the hardware button " + "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 " @@ -877,6 +884,61 @@ def verdict_cut(ctx): # 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 @@ -909,7 +971,10 @@ def verdict_cut(ctx): 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"]) - ctx.sleep(1.0) + # 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: @@ -925,7 +990,10 @@ def verdict_cut(ctx): 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()}) - time.sleep(0.12) + 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) @@ -951,7 +1019,10 @@ def verdict_cut(ctx): resumed_at = time.time() - t0 if st.startswith("Idle") and resumed_at is not None and time.time() - t0 > resumed_at + 3.0: break - time.sleep(0.12) + 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 @@ -986,13 +1057,22 @@ def verdict_cut(ctx): "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 from the hold until the resume: the gate masked the stream. - span = [r for r in trail if held and r["t"] >= held[0]["t"] - and (resumed_at is None or r["t"] < resumed_at)] - zero = next((r for r in span if r["emission"] == 0), None) - ev["emission_zero_after_hold_s"] = round(zero["t"] - held[0]["t"], 2) if zero else None - ctx.check(zero is not None and zero["t"] - held[0]["t"] <= 2.5, - "emission did not read 0 within 2.5 s of the hold (before the resume)") + # 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] @@ -1002,9 +1082,9 @@ def verdict_cut(ctx): 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, emission 0 %s s after the hold, resumed " - "lit at +%s s with no press, disarmed %.1f s after Idle", ev["held_at_s"], - ev["emission_zero_after_hold_s"], ev["resumed_at_s"], dt) + 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 new file mode 100644 index 0000000..f5ca51c --- /dev/null +++ b/forgetest/tests/test_laser_verdict.py @@ -0,0 +1,105 @@ +# 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()