vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-10-16 01:29:55 -07:00
Code Issues Packages Projects Releases Wiki Activity

forgetest: laser.verdict-cut leaves the catalog; the stream rules hold its claim

Browse Source 
laser.verdict-cut held, on the machine, that a stale cooling verdict
mid-cut holds the job dark under the open window with the latch
untouched, and that the clean verdict resumes it lit with no press.
Every part of that is a rule the host now proves:

- dark while held: laser_stream_test rules 24 and 27, check_hold_dark on
  every verdict hold's stationary stretch, and check_fire_gaps (ef15c55);
- the hold, the message, the window kept, the latch never written, the
  resume lit with no press: rules 24 and 27 (latch sideband [unlock,
  lock]), laser_lifecycle_test rule 11, cooling_test.c;
- the physical dwell gap (a kernel-run gap inside the unlocked window,
  the button latch clear, HV_ENABLE back lit): laser.emission-witness,
  unchanged.

- Its own lines go, decorator through body, and nothing else:
  DISARMED_MSG, the module constant only its check read, stays with the
  blank lines around it, so laser.py's shared text is unchanged.
- test_laser_verdict.py goes: every case loaded dark_span out of
  verdict_cut's body.

Proof: the fingerprint parts of laser.py before and after: the shared
text unchanged, the other five tests' own text unchanged. The coverage
lint reports 0 uncovered paths and 0 host-tested row problems over 109
tests (its one failure is setup.check-lens's src/lensfind.* entry, which
the forgectrl pin bump clears); the gate self-check refuses an empty
artifact cleanly; pyflakes adds no finding; the forgetest host suite
passes under Linux (539 tests).

Acceptance: the catalog hash moves; no remaining test's fingerprint
moves.
This commit is contained in:
ScottW514
2026-10-13 17:14:13 -04:00
parent ef15c55577
commit 21f53e4e83
2 changed files with 0 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/suite/laser.py
-241
View File
@@ -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"
forgetest/tests/test_laser_verdict.py
-105
View File
@@ -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()