From bed46e961eea214884f748ab520478a0ae667bb1 Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Fri, 16 Oct 2026 14:40:53 -0400 Subject: [PATCH] laser.emission-witness: the read rule ends where the sampling stopped, and the rapid is placed inside the window The dark stretch after the M5 runs on past the close, and three of its judgments did not hold with it. The read rule took the last sample served as the trail's last edge, and sample() returns None for a request the daemon does not answer. So an outage after the served close (one request that hangs to the client's 10 s timeout is enough) left a trail that ended at the close with no gap, and the window that latches after the close was never read: the test passed whole with a 30 ms fire at the end of the return rapid. The sampling could also stop as early as about 1.8 s past the close, and with a 0.9 s unread tail allowed, the last reading could come before the latch of the window in flight: the review's tail model (the kernel's ~1 s window latch, the sampler's GET timing, a fire just before the close) finds 4 runs in 20000, 21 with a wider GET spread, where that window goes unread and the judge passes. The placement and the disarm timing searched the whole stretch. A window that closed before the head was back (in the dwell at +20 mm or on the return leg, up to about 2 s early) found the head back after the close and passed "the rapid ran inside the dark stretch", with a negative disarm time. The rapid's sample count in the record ran past the close while its span stopped at the close, so the record read about 8 Hz for a 5 Hz loop. sample_to_disarm moves from emission_witness to module level, ends two windows and EMISSION_READ_GAP_S past the close, and returns the time the sampling stopped beside the samples. read_gaps takes that time as the trail's last edge (t_end), so a tail the daemon served none of is a gap. rapid_place judges the placement over the armed part alone, up to the first disarmed sample, and the disarm time must be 0 or more. The rapid block counts its samples to the close and records how long the sampling ran past it (sampled_past_close_s), and the log line says both. The relit, HV and read judgments still cover the whole stretch. The description says what is judged. tests/test_laser_dwell.py drops relit(), a copy of the judge that no change to laser.py could fail. Every change makes the judge stricter: no energy or motion path changes. Proof: tests/test_laser_dwell.py, the Unit tests step of forgetest-ci.yml (python -B -m unittest discover -s tests), 13 tests OK on Windows with python 3.14 and on Ubuntu 24.04 with python 3.12: test_a_tail_the_daemon_never_served_fails (every sample served read, then nothing for 2.9 s before the sampling stopped: gap 2.9 s, the rule fails; served through, it holds), test_the_sampling_runs_on_past_the_close_and_ends_where_it_stopped (sample_to_disarm on a fake clock, the daemon serving the close and nothing after it: the sampling ends at least 2.9 s past the close and the rule fails), and test_a_window_closed_before_the_head_is_back_does_not_place_the_rapid (a close after the return places out, back and the close; a close in the dwell at +20 mm leaves back None). test_cooling_suite.py (3) and test_setup_burns.py (22) pass. Negative controls on scratch copies: read_gaps without the end edge fails the first two cases; the end at two windows fails the second; sample_to_disarm that returns the samples alone errors in the second; rapid_place over the whole stretch fails the third. emission_witness run whole on a fake clock with only its I/O stubbed: a window closing 0.2, 1.0, 1.9 or 2.5 s before the head is back fails the placement (the code before this change passed the first three), and a normal close passes with 22 samples over 4.07 s to the close in the record (32 before). Against a simulated daemon and kernel latch with that 30 ms fire: an outage after the served close and one request hung to the 10 s timeout fail the read rule (gaps 3.05 s and 10.32 s; both passed before); a tail served through with no fire passes. In the review's tail model, the end edge alone and the longer sampling alone each leave 0 of 20000 runs with the window in flight unread and the judge passing, in both GET spreads. pyflakes reports no undefined name. Acceptance: laser.emission-witness's fingerprint moves (its own text and laser.py's shared text), laser.armed-kill and laser.pause-resume-lid-cancel move through the shared text, and cooling.flow-under-load and cooling.fail-tier-stop move because they import from .laser in their own functions; the other 38 do not move (source_sha compared before and after). Against the campaign this series already needs, that adds no run: 850498f moved all 43 once. laser.emission-witness is always run, and the next campaign judges the rapid with the end edge on the bench. --- forgetest/forgetest/suite/laser.py | 149 ++++++++++++++++------------ forgetest/tests/test_laser_dwell.py | 110 ++++++++++++++++---- 2 files changed, 178 insertions(+), 81 deletions(-) diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 5001393..08c438d 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -291,11 +291,13 @@ def dwell_gap(samples): "hv_enable_dipped": dip is not None, "hv_enable_back_lit": back_lit} -def read_gaps(samples, key): +def read_gaps(samples, key, t_end=None): """Over a sample trail: how many samples carry a reading for `key`, how many do not, and the longest stretch of seconds between two readings, the trail's ends counted as edges (an unread lead or tail is a gap too; a trail with no reading is one gap, its whole span). + The trail ends when its sampling stopped (`t_end`), not at the last + sample served: a tail the daemon did not serve at all is a gap too. A reading the daemon could not serve is a gap, never a dark sample: /status leaves the laser block out when the kernel's count did not read, and the count latches once per ~1 s window, so a window with @@ -303,11 +305,60 @@ def read_gaps(samples, key): if not samples: return {"read": 0, "unread": 0, "gap_max_s": None} ts = [s["t"] for s in samples if s.get(key) is not None] - edges = [samples[0]["t"]] + ts + [samples[-1]["t"]] + last = samples[-1]["t"] if t_end is None else max(t_end, samples[-1]["t"]) + edges = [samples[0]["t"]] + ts + [last] gap = max(b - a for a, b in zip(edges, edges[1:])) return {"read": len(ts), "unread": len(samples) - len(ts), "gap_max_s": round(gap, 3)} +def rapid_place(stretch, x0): + """Where the head's reports place the rapid after the M5, judged over + the armed part of the dark stretch alone: its samples up to the first + disarmed one, that one included. Returns (i_out, i_back, i_closed): + the first sample with the head at least 19.5 mm out, the first after + it with the head back at x0 (within 0.05 mm), and the first disarmed + sample; each None when there is none. A window that closes before + the head is back leaves i_back None: the samples after the close + never place the rapid inside the window.""" + i_closed = next((i for i, s in enumerate(stretch) if not s["armed"]), None) + part = stretch if i_closed is None else stretch[:i_closed + 1] + i_out = next((i for i, s in enumerate(part) if s.get("x") is not None and s["x"] - x0 >= 19.5), None) + i_back = (next((i for i in range(i_out + 1, len(part)) + if part[i].get("x") is not None and abs(part[i]["x"] - x0) < 0.05), None) + if i_out is not None else None) + return i_out, i_back, i_closed + + +def sample_to_disarm(ctx, g, timeout=20.0, sample_hz=8): + """Sample until the armed window closes, or the timeout, with the + controller's state and machine X beside each sample ("x"). A G4 dwell + reports Idle, so Idle cannot end this; the window closing at the + program end does, after two more kernel windows and the read gap + allowed: the LASER_ON count for the window in flight at the close + latches up to one window later and is held through the next, so the + sampling runs on until that count has been read. Returns the samples + served and the time the sampling stopped, the trail's end for the + read rule.""" + samples = [] + period = 1.0 / sample_hz + t0 = time.time() + end = t0 + timeout + t_closed = None + while time.time() < end: + ctx.checkpoint() + smp = sample(ctx) + rep = g.status_report() + if smp: + smp["gstate"] = rep.get("state") + smp["x"] = (rep.get("MPos") or (None,))[0] + samples.append(smp) + if t_closed is None and not smp["armed"]: + t_closed = time.time() + end = t_closed + 2 * EMISSION_WINDOW_S + EMISSION_READ_GAP_S + time.sleep(period) + return samples, time.time() + + def run_and_sample(ctx, g, job, sample_hz=8, overall_timeout=200): """Stream the job; sample forgectrl through arm -> fire -> disarm. Completes on: emission seen then Idle > 3 s; or armed then disarmed @@ -411,13 +462,14 @@ def wait_disarm(ctx, timeout): "second press. The square ends at M5 with the window still open; once the " "LASER_ON count reads 0 (the dark point), a rapid runs 20 mm out over the " "first side and back, then the program ends. From the dark point until the " - "window closes, every sample reads no emission and the HV current at its idle " - "reading, the kernel's LASER_ON count read in every ~1 s window of that stretch " - "and in the window that latches after the close (a reading the daemon could not " - "serve is a gap, never a dark sample), and the head's reports place the rapid " - "inside that stretch: LASER_ON stays off with FIRE clear while the chain can " - "fire. The M2 program end " - "disarms promptly (job-based, not the 60 s idle grace) and the button goes " + "window closes, and on until the count of the window in flight at the close is " + "read, every sample reads no emission and the HV current at its idle reading, " + "the kernel's LASER_ON count read in every ~1 s window of that stretch (a " + "reading the daemon could not serve is a gap, never a dark sample, and so is a " + "tail it served none of), and the head's reports place the rapid inside the " + "window, between the dark point and the close: LASER_ON stays off with FIRE " + "clear while the chain can fire. The M2 program end disarms promptly after the " + "head is back (job-based, not the 60 s idle grace) and the button goes " "dark. The operator confirms all four sides.") def emission_witness(ctx): ev = ctx.evidence @@ -425,33 +477,6 @@ def emission_witness(ctx): # first side and back, then the program end. rapid_tail = ["G0 X20", "G4 P2.5", "G0 X-20", "G4 P0.5", "G90", "M2"] - def sample_to_disarm(g, timeout=20.0, sample_hz=8): - """Sample until the armed window closes, or the timeout, with the - controller's state and machine X beside each sample ("x"). A G4 - dwell reports Idle, so Idle cannot end this; the window closing at - the program end does, after two more kernel windows: the LASER_ON - count for the window in flight at the close latches up to one - window later and is held through the next, so the sampling runs - on until that count has been read.""" - samples = [] - period = 1.0 / sample_hz - t0 = time.time() - end = t0 + timeout - t_closed = None - while time.time() < end: - ctx.checkpoint() - smp = sample(ctx) - rep = g.status_report() - if smp: - smp["gstate"] = rep.get("state") - smp["x"] = (rep.get("MPos") or (None,))[0] - samples.append(smp) - if t_closed is None and not smp["armed"]: - t_closed = time.time() - end = t_closed + 2 * EMISSION_WINDOW_S - time.sleep(period) - return samples - with ctx.grbl() as g, LiveJob(ctx, g): prepare(ctx, g) base = sample(ctx) @@ -490,7 +515,7 @@ def emission_witness(ctx): dark["x"] = (rep.get("MPos") or (None,))[0] ctx.check(dark["x"] is not None, "the controller reports no machine position: %s", rep) stream(g, rapid_tail) - tail = sample_to_disarm(g) + tail, t_end = sample_to_disarm(ctx, g) finally: ctx.clear_notice() samples = cut + [dark] + tail @@ -545,39 +570,40 @@ def emission_witness(ctx): ctx.log("exhaust at the first emission sample: %s rpm (floor %.0f)", ev["exhaust_at_first_fire"], floor) # The rapid after the M5, with the window open and FIRE clear: every - # sample from the dark point until the window closes reads no emission - # and the HV current at its idle reading, and the kernel's LASER_ON - # count was read in every ~1 s window of that stretch and in the one - # that latches after the close (the stretch runs two windows past it). - # A reading the daemon could not serve is a gap, never a dark sample. - # The head's reports place the rapid inside the stretch: out to - # +20 mm, then back at the start. + # sample from the dark point until the sampling stops (two windows and + # the read gap allowed past the close) reads no emission and the HV + # current at its idle reading, and the kernel's LASER_ON count was read + # in every ~1 s window of that stretch, the one that latches after the + # close included: the end of the sampling is the trail's last edge. A + # reading the daemon could not serve is a gap, never a dark sample. + # The head's reports place the rapid inside the window, from the dark + # point to the close: out to +20 mm, then back at the start. stretch = [dark] + tail x0 = dark["x"] - i_out = next((i for i, s in enumerate(stretch) if s.get("x") is not None and s["x"] - x0 >= 19.5), None) - i_back = (next((i for i in range(i_out + 1, len(stretch)) - if stretch[i].get("x") is not None and abs(stretch[i]["x"] - x0) < 0.05), None) - if i_out is not None else None) - closed = next((s for s in stretch if not s["armed"]), None) + i_out, i_back, i_closed = rapid_place(stretch, x0) + closed = stretch[i_closed] if i_closed is not None else None + armed_part = stretch if i_closed is None else stretch[:i_closed + 1] relit = [s for s in stretch if s["emission"]] hv_dark = max((s["hv"] for s in stretch if s["hv"] is not None), default=0) - reads = read_gaps(stretch, "emission") - hv_reads = read_gaps(stretch, "hv") + reads = read_gaps(stretch, "emission", t_end) + hv_reads = read_gaps(stretch, "hv", t_end) dt = (closed["t"] - stretch[i_back]["t"]) if (closed and i_back is not None) else None - ev["rapid"] = {"samples": len(stretch), "span_s": round((closed or stretch[-1])["t"] - dark["t"], 2), - "x_start": x0, "x_max": max((s["x"] for s in stretch if s.get("x") is not None), default=None), + ev["rapid"] = {"samples": len(armed_part), "span_s": round((closed or stretch[-1])["t"] - dark["t"], 2), + "x_start": x0, "x_max": max((s["x"] for s in armed_part if s.get("x") is not None), default=None), "out_at_s": round(stretch[i_out]["t"] - dark["t"], 2) if i_out is not None else None, "back_at_s": round(stretch[i_back]["t"] - dark["t"], 2) if i_back is not None else None, "relit": [(round(s["t"] - dark["t"], 2), s["emission"]) for s in relit][:8], "hv_max": hv_dark, "locked_at_dark": dark["locked"], "locked_at_out": stretch[i_out]["locked"] if i_out is not None else None, + "sampled_past_close_s": round(t_end - closed["t"], 2) if closed else None, "emission_reads": reads, "hv_reads": hv_reads} ev["disarm_after_idle_s"] = round(dt, 1) if dt is not None else None - ctx.log("the rapid after the M5: %d samples over %.1f s from the dark point, out at +%s s, back at +%s s, " - "%d with emission, HV max %s; %d of %d samples carried the LASER_ON count, longest unread " - "stretch %s s; the window closed %s s after the head was back", len(stretch), - ev["rapid"]["span_s"], ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"], len(relit), hv_dark, - reads["read"], len(stretch), reads["gap_max_s"], ev["disarm_after_idle_s"]) + ctx.log("the rapid after the M5: %d samples over %.1f s from the dark point to the close, out at +%s s, " + "back at +%s s; sampled %s s past the close, %d of %d samples with emission, HV max %s, %d of %d " + "carried the LASER_ON count, longest unread stretch %s s; the window closed %s s after the head " + "was back", len(armed_part), ev["rapid"]["span_s"], ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"], + ev["rapid"]["sampled_past_close_s"], len(relit), len(stretch), hv_dark, reads["read"], + len(stretch), reads["gap_max_s"], ev["disarm_after_idle_s"]) ctx.check(armed_seen, "the armed window was never observed (arm refused, or no button press)") ctx.check(peak > 0, "no emission witnessed on the square (emission_samples stayed 0)") first_idle_fire = ev["idle_airflow_fire"][0] if idle_fire else (None, None, None) @@ -592,8 +618,9 @@ def emission_witness(ctx): ctx.check(hv and max(hv) > min(hv), "HV current did not rise during the burn (%s..%s)", ev["hv_min"], ev["hv_max"]) ctx.check(i_out is not None and i_back is not None, - "the rapid did not run inside the dark stretch: out at %s, back at %s (X from %s, max %s)", - ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"], x0, ev["rapid"]["x_max"]) + "the rapid did not run inside the dark stretch with the window open: out at %s, back at %s " + "(X from %s, max %s), the window closed at %s", ev["rapid"]["out_at_s"], ev["rapid"]["back_at_s"], + x0, ev["rapid"]["x_max"], round(closed["t"] - dark["t"], 2) if closed else None) ctx.check(stretch[i_out]["locked"] is False, "the laser latch was locked with the head at +20 mm " "(laser_locked %s): the rapid ran unarmed", stretch[i_out]["locked"]) ctx.check(reads["read"] >= RAPID_MIN_READS and reads["gap_max_s"] is not None @@ -609,7 +636,7 @@ def emission_witness(ctx): hv_reads["read"], len(stretch), RAPID_MIN_READS) ctx.check(hv_dark <= HV_DARK_MAX, "HV current %s after the M5 (idle reads ~0): the discharge ran " "through the rapid", hv_dark) - ctx.check(dt is not None and dt < 10.0, + ctx.check(dt is not None and 0 <= dt < 10.0, "the M2 job did not disarm promptly (%s s after the head was back; the idle grace is ~60 s)", ev["disarm_after_idle_s"]) ctx.check(gap["button_latch_unlocked_samples"] and gap["button_latch_unlocked_max"] == 0, diff --git a/forgetest/tests/test_laser_dwell.py b/forgetest/tests/test_laser_dwell.py index d395d19..48a48f8 100644 --- a/forgetest/tests/test_laser_dwell.py +++ b/forgetest/tests/test_laser_dwell.py @@ -9,8 +9,11 @@ whose readback word shows the laser latch unlocked, both bits from that word, so the relocked tail (where the relock itself sets the latch and the emission counter still reads nonzero) is never judged. And its read-gap verdict over the dark stretch after the M5: a sample whose -/status carried no laser block is a gap, never a dark sample, and a -stretch with a kernel window unread fails the rule.""" +/status carried no laser block is a gap, never a dark sample, a tail the +daemon did not serve before the sampling stopped is a gap too, and a +stretch with a kernel window unread fails the rule; the sampling runs on +past the close; and the head's reports place the rapid inside the armed +window only.""" import unittest from forgetest.suite import laser @@ -71,20 +74,12 @@ def covered(r): and r["gap_max_s"] <= laser.EMISSION_READ_GAP_S) -def relit(trail): - """The samples that read emission: the relit judge alone, which an - unread sample never reaches.""" - return [s for s in trail if s["emission"]] - - class RapidReadTests(unittest.TestCase): def test_every_sample_read_is_inside_the_rule(self): - trail = rapid_trail(25) - r = laser.read_gaps(trail, "emission") + r = laser.read_gaps(rapid_trail(25), "emission") self.assertEqual((r["read"], r["unread"]), (25, 0)) self.assertAlmostEqual(r["gap_max_s"], 0.2) self.assertTrue(covered(r)) - self.assertEqual(relit(trail), []) def test_one_transient_miss_is_inside_the_rule(self): r = laser.read_gaps(rapid_trail(25, holes=(12,)), "emission") @@ -98,28 +93,34 @@ class RapidReadTests(unittest.TestCase): self.assertTrue(covered(r)) def test_a_hole_longer_than_a_window_fails(self): - trail = rapid_trail(25, holes=(10, 11, 12, 13, 14)) - r = laser.read_gaps(trail, "emission") + r = laser.read_gaps(rapid_trail(25, holes=(10, 11, 12, 13, 14)), "emission") self.assertAlmostEqual(r["gap_max_s"], 1.2) self.assertFalse(covered(r)) - self.assertEqual(relit(trail), []) def test_the_block_gone_after_the_dark_point_fails(self): - trail = rapid_trail(25, holes=tuple(range(1, 25))) - r = laser.read_gaps(trail, "emission") + r = laser.read_gaps(rapid_trail(25, holes=tuple(range(1, 25))), "emission") self.assertEqual((r["read"], r["unread"]), (1, 24)) self.assertAlmostEqual(r["gap_max_s"], 4.8) self.assertFalse(covered(r)) - self.assertEqual(relit(trail), []) def test_the_block_gone_for_the_last_window_and_a_half_fails(self): # the tail is an edge: the window that latches after the close # goes unread - trail = rapid_trail(25, holes=tuple(range(17, 25))) - r = laser.read_gaps(trail, "emission") + r = laser.read_gaps(rapid_trail(25, holes=tuple(range(17, 25))), "emission") self.assertAlmostEqual(r["gap_max_s"], 1.6) self.assertFalse(covered(r)) - self.assertEqual(relit(trail), []) + + def test_a_tail_the_daemon_never_served_fails(self): + # every sample served was read, and the served samples end at the + # close: the sampling ran on past it with nothing served, so the + # stretch from the last reading to where the sampling stopped is a gap + trail = rapid_trail(25) + tail = 2 * laser.EMISSION_WINDOW_S + laser.EMISSION_READ_GAP_S + r = laser.read_gaps(trail, "emission", trail[-1]["t"] + tail) + self.assertAlmostEqual(r["gap_max_s"], tail) + self.assertFalse(covered(r)) + # served through: the sampling stops one loop after the last sample + self.assertTrue(covered(laser.read_gaps(trail, "emission", trail[-1]["t"] + 0.125))) def test_a_stretch_of_one_sample_fails(self): r = laser.read_gaps(rapid_trail(1), "emission") @@ -127,5 +128,74 @@ class RapidReadTests(unittest.TestCase): self.assertFalse(covered(r)) +class Clock: + """time.time and time.sleep on a clock the test moves.""" + + def __init__(self, now=1000.0): + self.now = now + + def time(self): + return self.now + + def sleep(self, s): + self.now += s + + +class RapidTailTests(unittest.TestCase): + def test_the_sampling_runs_on_past_the_close_and_ends_where_it_stopped(self): + # The window closes 3 s in, and the daemon serves the closed + # sample and nothing after it: the sampling still runs two kernel + # windows and the read gap allowed past the close, and the time it + # stopped is the trail's end, so the tail it never served fails the + # read rule. + clock = Clock() + close = clock.now + 3.0 + served = [] + + def sample(_ctx): + clock.now += 0.06 # the two GETs + if served and not served[-1]["armed"]: + return None # nothing served after the close + s = {"t": clock.now, "armed": clock.now < close, "emission": 0, "hv": 0} + served.append(s) + return s + + class Ctx: + def checkpoint(self): + pass + + class Grbl: + def status_report(self): + return {"state": "Idle", "MPos": (0.0, 0.0, 0.0)} + saved = laser.time, laser.sample + self.addCleanup(setattr, laser, "time", saved[0]) + self.addCleanup(setattr, laser, "sample", saved[1]) + laser.time, laser.sample = clock, sample + tail, t_end = laser.sample_to_disarm(Ctx(), Grbl()) + closed = tail[-1] + self.assertFalse(closed["armed"]) + self.assertTrue(all(s["armed"] for s in tail[:-1])) + self.assertGreaterEqual(t_end - closed["t"], 2 * laser.EMISSION_WINDOW_S + laser.EMISSION_READ_GAP_S) + self.assertFalse(covered(laser.read_gaps(tail, "emission", t_end))) + + +def place_trail(xs, closes_at): + """A dark stretch with the head at machine X `xs` (one sample each, + 0.2 s apart), the window closing at sample `closes_at`.""" + return [{"t": 100.0 + i * 0.2, "x": float(x), "armed": i < closes_at} for i, x in enumerate(xs)] + + +class RapidPlaceTests(unittest.TestCase): + # out to +20 mm, the dwell there, back at the start, the dwell before the M2 + XS = [0, 5, 20, 20, 20, 10, 0, 0, 0, 0] + + def test_a_window_closed_before_the_head_is_back_does_not_place_the_rapid(self): + # the window closes after the head is back: out at 2, back at 6, closed at 8 + self.assertEqual(laser.rapid_place(place_trail(self.XS, 8), 0.0), (2, 6, 8)) + # the window closes in the dwell at +20 mm and the head comes back + # after the close: the samples past the close do not place it + self.assertEqual(laser.rapid_place(place_trail(self.XS, 4), 0.0), (2, None, 4)) + + if __name__ == "__main__": unittest.main()