From b7dd23d539fbe17ed05d10fedd297475359e879c Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Thu, 24 Sep 2026 14:20:06 -0400 Subject: [PATCH] forgetest: a takeover judges where the head stands forgectrl's start at the end of a takeover is a controller start, which re-zeroes the step counters wherever the head then stands. After it a head left out reads as home, so the baseline's position check after the run could not see a head a test left out before its takeover. setup.check-envelope did exactly that on 2026-09-23 and passed (fixed by 56af6ea). The takeover now judges the head at the two moments it can, and moves nothing: as the controller goes away, against the position the run's baseline expects (when the counters are still in the frame the run began in), and as the takeover ends, before forgectrl comes back, against where it found the head (when the microstep mode has not changed under it). A miss past the dead band is recorded on the run's baseline capture, and the post pass turns it into an unrestorable leftover ("a controller restart re-zeroed the counters there, so the head is not moved"), which fails the run. A frame the test did not declare, cloud mode (its counters are the cloud client's), a bench tool without a run, and a reading that fails are not judged and never keep forgectrl down. ctx.counters_rezeroed() now records the frame in force, so a takeover after a declared re-zero can still judge. The frame is read before the controller is stopped: forgectrl unlinks /run/grblhal.homed when the controller exits (super.c), which the bench's negative control found. Proof: tests/test_takeover_position.py, 12 cases against a fake sysfs tree and a fake anchor file (a head in place, a head left out at the start, the frame read before the stop, the dead band, an undeclared frame, cloud mode, a bench tool, a head left out by a drill at the end, a changed microstep mode, the post pass failing the run and moving nothing, a failed reading, a declared re-zero). forgetest's unit tests pass on the host (474 OK, 4 skipped). On the bench reference, image 20260923232513 with the files bind-mounted: the old bedsize.py (56af6ea^) fails as it must, "position at the takeover start=[2134, 2133, 0] (expected [0, 0, 0])"; then 18 unattended tests that take over the controller PASS in sequence (setup.check-envelope, kernel.k1-k2, kernel.deadman-close, kernel.backtrack-bounds, kernel.fire-line, kernel.resume-lead, setup.gate-blocks-controllers, setup.advisories-rehash, setup.extensions-consent, setup.what-changed, setup.mirror, setup.account-login, setup.first-run-flow, exthost.service, exthost.events, exthost.hold-pause-tier, exthost.motion-job, exthost.motion-jog). A second takeover in one run meets the frame the first one's restart made, which no test declared, so its start is logged and not judged. Acceptance: forgetest is the dev-only harness, outside the catalog's coverage; no test's fingerprint moves (runner.py and baseline.py are not suite modules). --- forgetest/forgetest/baseline.py | 8 + forgetest/forgetest/runner.py | 77 ++++++- forgetest/tests/test_takeover_position.py | 244 ++++++++++++++++++++++ 3 files changed, 327 insertions(+), 2 deletions(-) create mode 100644 forgetest/tests/test_takeover_position.py diff --git a/forgetest/forgetest/baseline.py b/forgetest/forgetest/baseline.py index b7dbde1..d51ddd1 100644 --- a/forgetest/forgetest/baseline.py +++ b/forgetest/forgetest/baseline.py @@ -905,6 +905,14 @@ class Baseline: "not a leftover" % (now, was, act, POSITION_DEADBAND_MM)) else: left.append(Leftover("position", now, was, act)) + # A controller restart during the run (a takeover's) re-zeroed the + # counters wherever the head then stood, so a head left out before + # it reads as home from here. The takeover judged it on the spot + # (runner.Takeover); what it recorded is a leftover, never moved. + for r in captured.get("restart_positions") or []: + left.append(Leftover("position at the %s" % r["where"], r["found"], r["expected"], + "unrestorable: a controller restart re-zeroed the counters there, " + "so the head is not moved")) was = captured.get("settings") if was: st, body = self.fc_get("/settings") diff --git a/forgetest/forgetest/runner.py b/forgetest/forgetest/runner.py index a4f85cd..1ef6968 100644 --- a/forgetest/forgetest/runner.py +++ b/forgetest/forgetest/runner.py @@ -539,7 +539,7 @@ class Context: return hw.Grbl() def takeover(self): - return Takeover(self.run.log, self.test.id) + return Takeover(self.run.log, self.test.id, run=self.run) def counters_rezeroed(self): """Tell the baseline the kernel position counters were re-zeroed @@ -550,6 +550,9 @@ class Context: if cap and cap.get("position") is not None: cap["position"] = [0, 0, 0] cap["rezero_declared"] = True # the test vouches for the new frame + # ... and it is the frame now in force, so a takeover later in + # the run can still judge where the head stands against it + cap["frame"] = _baseline.counter_frame() self.log("position counters re-zeroed at the starting position; the baseline " "expects (0,0,0) at the end") @@ -579,11 +582,15 @@ class Takeover: "cnc/x_mode", "cnc/y_mode", "cnc/x_decay", "cnc/y_decay", "pic/x_step_current", "pic/y_step_current") - def __init__(self, log, who): + def __init__(self, log, who, run=None): self.log = log # callable(str) self.who = who self.marker = marker_path() self.saved = {} + self.run = run # the test's run, whose baseline the head is judged against + self.cloud = False + self.frame = None # the counters' frame, read before the controller is stopped + self.found = None # (counters, microstep mode) once the controller is gone def wait_settled(self): return _baseline.Baseline(self.log).wait_settled() @@ -610,6 +617,14 @@ class Takeover: self.saved[attr] = v if self.saved: log("takeover: preserving %s" % ", ".join("%s=%s" % kv for kv in self.saved.items())) + try: + st, mode = hw.Forgectrl().get("/mode") + self.cloud = st == 200 and isinstance(mode, dict) and mode.get("mode") == "cloud" + except hw.HwError: + pass + # The frame is read while the controller runs: its exit takes the + # anchor with it (forgectrl unlinks it), though not the counters. + self.frame = _baseline.counter_frame() log("takeover: stopping the controller through forgectrl") try: st, body = hw.Forgectrl().post("/controller/stop") @@ -632,9 +647,67 @@ class Takeover: self.__exit__(None, None, None) raise Failed("takeover: processes still alive after stop: %s" % left) log("takeover: pulse device free") + try: + self.position_at_start() + except Exception as e: # noqa: BLE001 - a reading never keeps forgectrl down + log("takeover: WARNING the head's position could not be judged: %s: %s" % (type(e).__name__, e)) return self + # A controller start re-zeroes the step counters where the head then + # stands, and forgectrl's start after a takeover is one, so the + # baseline's check after the run cannot see a head left out before it. + # The takeover judges it at the two moments it can: as the controller + # goes away (against where the run expects the head) and before it + # comes back (against where the takeover found it). A miss is recorded + # for the baseline's post pass, which fails the run; nothing is moved, + # because a jog sized from these counters is only as good as the frame + # they are in. Cloud mode is not judged: its counters are the cloud + # client's, as in the baseline. + + def _record(self, where, found, expected): + cap = self.run.baseline_captured if self.run is not None else None + if cap is not None: + cap.setdefault("restart_positions", []).append( + {"where": where, "found": found, "expected": expected}) + self.log("takeover: the head is not where it belongs at the %s: counters %s, expected %s; " + "recorded as a leftover, nothing moved" % (where, found, expected)) + + def position_at_start(self): + cap = self.run.baseline_captured if self.run is not None else None + if not cap or self.cloud: + return + now = _baseline.read_position() + self.found = (now, hw.sysfs_read("cnc/x_mode")) + was = cap.get("position") + if now is None or was is None: + return + if self.frame is None or cap.get("frame") != self.frame: + self.log("takeover: the counters are not in the frame the run began in (a controller " + "start or a home the test did not declare), so the head is not judged here") + return + if now != was and not _baseline.position_quantized(was, now, _baseline.counter_steps_per_mm()): + self._record("takeover start", now, was) + + def position_at_end(self): + if self.found is None: + return + was, mode = self.found + now = _baseline.read_position() + if was is None or now is None: + return + if hw.sysfs_read("cnc/x_mode") != mode: + self.log("takeover: the microstep mode changed under the takeover, so the counters " + "before and after it are not one scale; the head is not judged here") + return + if now != was and not _baseline.position_quantized(was, now, _baseline.counter_steps_per_mm()): + self._record("takeover end", now, was) + def __exit__(self, exc_type, exc, tb): + try: + self.position_at_end() + except Exception as e: # noqa: BLE001 - a reading never keeps forgectrl down + self.log("takeover: WARNING the head's position could not be judged: %s: %s" + % (type(e).__name__, e)) self.restore_attrs() rc, out = hw.initd("forgectrl", "start") self.log("takeover: forgectrl start -> rc %s" % rc) diff --git a/forgetest/tests/test_takeover_position.py b/forgetest/tests/test_takeover_position.py new file mode 100644 index 0000000..89b1729 --- /dev/null +++ b/forgetest/tests/test_takeover_position.py @@ -0,0 +1,244 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""A takeover judges the head where the baseline cannot. forgectrl's start +at the end of a takeover is a controller start, which re-zeroes the step +counters wherever the head stands, so after it a head left out reads as +home. The takeover reads the counters as the controller goes away (against +where the run expects the head) and before it comes back (against where the +takeover found it); a miss is a leftover that fails the run, and the head is +never moved on it. Runs against a fake sysfs tree and a fake anchor file.""" +import os +import shutil +import struct +import tempfile +import unittest + +from forgetest import baseline, hw, runner + + +class FakeRun: + def __init__(self, cap): + self.baseline_captured = cap + self.lines = [] + + def log(self, s): + self.lines.append(s) + + +class TakeoverPositionTests(unittest.TestCase): + def setUp(self): + # the clean machine test_baseline.py builds, so the post pass + # finds nothing but what a test here leaves + self.tmp = tempfile.mkdtemp(prefix="forgetest-tko-") + self.sysfs = os.path.join(self.tmp, "sysfs") + os.sep + self.leds = os.path.join(self.tmp, "leds") + os.sep + for group in ("cnc", "pic", "head", "thermal"): + os.makedirs(self.sysfs + group) + for name in baseline.BUTTON_LEDS + ("lid_led",): + os.makedirs(self.leds + name) + with open(self.leds + name + "/brightness", "w") as f: + f.write("0") + for attr, val in baseline.fixed_sysfs() + baseline.IDLE_READBACKS: + self._attr(attr, val) + self._attr("cnc/interlock_circuit", "45") + self._attr("pic/lid_led", "0") + self._pos(0, 0, 0) + os.environ["GF_SYSFS_ROOT"] = self.sysfs + os.environ["GF_LEDS_ROOT"] = self.leds + os.environ["FORGECTRL_URL"] = "http://127.0.0.1:1" # nothing listens + baseline.Baseline._unreachable_until = 0.0 + self.anchor = os.path.join(self.tmp, "grblhal.homed") + self._new_anchor(1) + self.old_anchor, baseline.ANCHOR_PATH = baseline.ANCHOR_PATH, self.anchor + self.spm = baseline.counter_steps_per_mm() + + def tearDown(self): + baseline.ANCHOR_PATH = self.old_anchor + shutil.rmtree(self.tmp, ignore_errors=True) + for k in ("GF_SYSFS_ROOT", "GF_LEDS_ROOT", "FORGECTRL_URL"): + os.environ.pop(k, None) + + def _attr(self, attr, val): + with open(self.sysfs + attr, "w") as f: + f.write(str(val)) + + def _pos(self, x, y, z): + with open(self.sysfs + "cnc/position", "wb") as f: + f.write(struct.pack("<5i", x, y, z, 0, 0)) + + def _new_anchor(self, ns): + """A controller start or a home: a new anchor file.""" + with open(self.anchor + ".new", "w") as f: + f.write("0 0 0 4 startup") + os.replace(self.anchor + ".new", self.anchor) + os.utime(self.anchor, ns=(ns, ns)) + + def cap(self): + return {"position": baseline.read_position(), "frame": baseline.counter_frame()} + + def takeover(self, cap): + run = FakeRun(cap) + return runner.Takeover(run.log, "test.id", run=run), run + + def mm(self, mm): + return round(mm * self.spm) + + def start(self, t): + """Takeover.__enter__'s order: the frame is read while the + controller runs, then its exit takes the anchor with it (forgectrl + unlinks it) and leaves the counters, which are then read.""" + t.frame = baseline.counter_frame() + if os.path.exists(self.anchor): + os.unlink(self.anchor) + t.position_at_start() + + def test_a_head_where_the_run_expects_it_is_not_recorded(self): + cap = self.cap() + t, run = self.takeover(cap) + self.start(t) + t.position_at_end() + self.assertNotIn("restart_positions", cap) + + def test_a_head_left_out_at_the_start_is_recorded(self): + cap = self.cap() + self._pos(self.mm(10), self.mm(10), 0) + t, run = self.takeover(cap) + self.start(t) + self.assertEqual(cap["restart_positions"], + [{"where": "takeover start", "found": [self.mm(10), self.mm(10), 0], + "expected": [0, 0, 0]}]) + + def test_the_frame_is_read_before_the_controller_goes(self): + # found on the bench: forgectrl unlinks the anchor when the + # controller exits, so a frame read after the stop never matched + # and the check judged nothing. Read before the stop, it does. + cap = self.cap() + self._pos(self.mm(10), self.mm(10), 0) + late, run = self.takeover(cap) + os.unlink(self.anchor) + late.frame = baseline.counter_frame() + late.position_at_start() + self.assertNotIn("restart_positions", cap) + self._new_anchor(1) + cap = self.cap() # a run that began at the origin in this frame + cap["position"] = [0, 0, 0] + on_time, run = self.takeover(cap) + self.start(on_time) + self.assertEqual([r["where"] for r in cap["restart_positions"]], ["takeover start"]) + + def test_the_dead_band_is_not_a_leftover(self): + cap = self.cap() + self._pos(self.mm(0.05), 0, 0) + t, run = self.takeover(cap) + self.start(t) + self.assertNotIn("restart_positions", cap) + + def test_an_undeclared_frame_change_is_not_judged(self): + cap = self.cap() + self._new_anchor(2) # a controller start the test did not declare + self._pos(self.mm(10), 0, 0) + t, run = self.takeover(cap) + self.start(t) + self.assertNotIn("restart_positions", cap) + self.assertTrue(any("not in the frame the run began in" in l for l in run.lines), run.lines) + + def test_cloud_mode_is_not_judged(self): + cap = self.cap() + self._pos(self.mm(10), 0, 0) + t, run = self.takeover(cap) + t.cloud = True + t.position_at_start() + self._pos(self.mm(20), 0, 0) + t.position_at_end() + self.assertNotIn("restart_positions", cap) + + def test_a_bench_tool_is_not_judged(self): + t = runner.Takeover(lambda s: None, "bench:tool") + self._pos(self.mm(10), 0, 0) + t.position_at_start() + t.position_at_end() # no run: nothing to judge against, nothing raised + + def test_a_drill_that_leaves_the_head_out_is_recorded_at_the_end(self): + cap = self.cap() + t, run = self.takeover(cap) + self.start(t) + self._pos(self.mm(5), 0, 0) # the drill moved the head and did not bring it back + t.position_at_end() + self.assertEqual([r["where"] for r in cap["restart_positions"]], ["takeover end"]) + self.assertEqual(cap["restart_positions"][0]["expected"], [0, 0, 0]) + + def test_a_changed_microstep_mode_is_not_judged_at_the_end(self): + cap = self.cap() + t, run = self.takeover(cap) + self.start(t) + self._attr("cnc/x_mode", "8" if hw.sysfs_read("cnc/x_mode") != "8" else "16") + self._pos(self.mm(5), 0, 0) + t.position_at_end() + self.assertNotIn("restart_positions", cap) + + def test_the_post_pass_fails_the_run_and_moves_nothing(self): + # setup.check-envelope's case: a declared camera home, the head + # left out, the record put back under a takeover, whose restart + # zeroes the counters where the head stands + cap = self.cap() + self._new_anchor(3) # the camera home + cap["position"], cap["rezero_declared"], cap["frame"] = [0, 0, 0], True, baseline.counter_frame() + self._pos(self.mm(10), self.mm(10), 0) + t, run = self.takeover(cap) + self.start(t) + t.position_at_end() + self._new_anchor(4) # forgectrl's start: a controller start + self._pos(0, 0, 0) + lines = [] + left = baseline.Baseline(lines.append).enforce("post", captured=cap) + self.assertEqual([x.item for x in left], ["position at the takeover start"]) + self.assertTrue(left[0].action.startswith("unrestorable"), left[0].action) + self.assertIn("not moved", left[0].action) + self.assertEqual(baseline.read_position(), [0, 0, 0]) + # without the takeover's record the same run reads clean: the gap + del cap["restart_positions"] + self.assertEqual(baseline.Baseline(lines.append).enforce("post", captured=cap), []) + + def test_a_failed_reading_never_keeps_forgectrl_down(self): + # the exit path restores the attributes, relocks the latch and + # starts forgectrl whatever the position reading does + os.environ["FORGETEST_MARKER"] = os.path.join(self.tmp, "marker") + calls = [] + real = (hw.initd, baseline.read_position, runner.Takeover.wait_settled) + hw.initd = lambda service, action, timeout=60: (calls.append((service, action)) or (0, "")) + + def broken(): + raise OSError("unreadable") + baseline.read_position = broken + runner.Takeover.wait_settled = lambda self: None + try: + cap = {"position": [0, 0, 0], "frame": baseline.counter_frame()} + t, run = self.takeover(cap) + t.found = ([0, 0, 0], hw.sysfs_read("cnc/x_mode")) + t.saved = {"cnc/x_mode": hw.sysfs_read("cnc/x_mode")} + t.__exit__(None, None, None) + finally: + hw.initd, baseline.read_position, runner.Takeover.wait_settled = real + os.environ.pop("FORGETEST_MARKER", None) + self.assertEqual(calls, [("forgectrl", "start")]) + with open(self.sysfs + "cnc/laser_latch") as f: + self.assertEqual(f.read().strip(), "1") + self.assertTrue(any("could not be judged" in l for l in run.lines), run.lines) + + def test_a_declared_rezero_records_the_frame_in_force(self): + cap = self.cap() + self._new_anchor(5) # the camera home + stub = type("Ctx", (), {})() + stub.run = FakeRun(cap) + stub.log = stub.run.log + runner.Context.counters_rezeroed(stub) + self.assertEqual(cap["position"], [0, 0, 0]) + self.assertTrue(cap["rezero_declared"]) + self.assertEqual(cap["frame"], baseline.counter_frame()) + + +if __name__ == "__main__": + unittest.main()