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()