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.
This commit is contained in:
ScottW514
2026-09-24 14:20:06 -04:00
parent dc9bede034
commit 308a033146
2 changed files with 203 additions and 18 deletions
+98 -18
View File
@@ -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"
+105
View File
@@ -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()