diff --git a/forgetest/forgetest/suite/bedsize.py b/forgetest/forgetest/suite/bedsize.py index a4dabc1..00a6d36 100644 --- a/forgetest/forgetest/suite/bedsize.py +++ b/forgetest/forgetest/suite/bedsize.py @@ -17,6 +17,7 @@ import time from ..catalog import test from .cloud import _HOMING_PATH, gfhome_homing +from .homeoff import camera_home_return, judge_whole_motions, session_mark from .setup import read_file, record_path, write_file from .setup_dark import DARK_COVERS, answer, dark, start @@ -173,8 +174,9 @@ def _edge(ctx, fc, ev, axis, edge): ("grblhal-glowforge", "src/driver.c")], requires=["forgectrl.auth", "motion.pacing"], steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The head moves " - "at most about 62 mm out from the camera home in X and in Y."], - description="With cloud mode on, the machine is homed with the camera (the web-service session). " + "at most about 62 mm out from the camera home in X and in Y, and ends where it was found."], + description="With cloud mode on, the machine is homed with the camera (the web-service session), every " + "service motion of it run whole. " "POST /wiz/motion.envelope/start runs the Bed size check; its prompts are jogs with This is the " "end among the options, and are answered as the page would, out 60.1 mm in X and 61.1 mm in Y " "(steps of 10, 1, and 0.1 mm). The check ends complete: the ends it reports are the home plus " @@ -187,7 +189,8 @@ def _edge(ctx, fc, ev, axis, edge): "first run left them and the edges still in force. The keys, the homing mode, and the setup " "record are put back as found, the record under a restart, and the head goes back to the " "home by a Grbl client's jog in machine coordinates; the run fails unless the port reads " - "the head on the home's step before the restart.") + "the head on the home's step before the restart. Then the camera home is dropped and the " + "head goes back to where the test found it, by the travel the session's motions logged.") def check_envelope(ctx): fc = ctx.forgectrl ev = ctx.evidence @@ -196,13 +199,15 @@ def check_envelope(ctx): raw = read_file(record_path()) found = {k: s0.get(k) or "" for k in KEYS} ev["found"] = found - home = back = None + home = back = session_at = None try: st, body = fc.post("/settings", data={"homing_mode": "gfcloud"}) ctx.check(st == 200, "homing_mode=gfcloud -> %s %s", st, body) with ctx.grbl() as g: + session_at = session_mark() gfhome_homing(ctx, ev, g) home = g.status_report().get("MPos") + judge_whole_motions(ctx, ev, session_at) ev["home"] = home ctx.log("the camera home declared %s", home) ctx.check(home and home[0] is not None, "no position after the home: %s", home) @@ -262,12 +267,18 @@ def check_envelope(ctx): ctx.wait_for(lambda: not dark(fc).get("running"), 30, poll=0.5) back = _back_home(ctx, fc, home) _put_back(ctx, fc, found) - # The check records itself when it completes; the record goes back - # as found under a restart, which also ends the envelope it set. - if read_file(record_path()) != raw: - with ctx.takeover(): - write_file(record_path(), raw) - ctx.log("the previous setup record is back under a restart") + try: + # Before the record's restart: the counters since the home say + # where the head stands until a controller start zeroes them. + if session_at is not None: + camera_home_return(ctx, ev, session_at) + finally: + # The check records itself when it completes; the record goes back + # as found under a restart, which also ends the envelope it set. + if read_file(record_path()) != raw: + with ctx.takeover(): + write_file(record_path(), raw) + ctx.log("the previous setup record is back under a restart") # Judged from the reading before the restart: the restart zeroes the # counters where the head stands, so the baseline's position check # after it cannot see a head left out. diff --git a/forgetest/forgetest/suite/cloud.py b/forgetest/forgetest/suite/cloud.py index b2f0847..2a2409d 100644 --- a/forgetest/forgetest/suite/cloud.py +++ b/forgetest/forgetest/suite/cloud.py @@ -254,8 +254,8 @@ def gfhome_homing(ctx, ev, g): "Open the lid the moment you are told, with a hand ready on it: the cloud client's " "first hunt begins a few seconds after its controller starts and must find the lid " "open. Leave it open through the connect and the hunt; close it when told. Nothing " - "else: the switch back and the $H homing run on their own, and the head ends parked " - "at the home corner."], + "else: the switch back and the $H homing run on their own, and the head ends where the " + "test found it."], description="One round trip with the two service-driven motions on it. POST /mode switches " "to the cloud controller with the lid closed (no controller starts with the " "enclosure open): gfcloud comes up under supervision, authenticates and " @@ -268,11 +268,18 @@ def gfhome_homing(ctx, ev, g): "is measured by the airflow gates but not judged (no AIRFLOW, the exhaust row " "reads unjudged); the camera service survives the switch. The lid closed, the " "service's re-hunt is waited out; switching back brings grblHAL up with the Grbl " - "port open and Idle, and the head returns to its start. Then $H with " + "port open and Idle, and the head returns to its start by the travel the " + "client logged for the service's motions. Then $H with " "homing_mode = gfcloud runs gfhome, the web-service homing session with the " "head-accelerometer motion witness: the controller returns to Idle with " - "homed:true and gfhome reports the homing complete with the motion it saw.") + "homed:true, gfhome reports the homing complete with the motion it saw, and " + "every service motion of the session ran whole. The camera home is then " + "dropped and the head goes back to where the test found it, by the travel the " + "session's motions logged.") def mode_switch(ctx): + # The hand-back helpers, imported here so that no other cloud test's + # fingerprint moves. + from .homeoff import camera_home_return, cloud_mode_return, judge_whole_motions, session_mark fc = ctx.forgectrl ev = ctx.evidence st, m0 = fc.get("/mode") @@ -383,10 +390,9 @@ def mode_switch(ctx): st, cam2 = fc.get("/cam/status") ev["cam_after"] = cam2 ctx.check(st == 200, "camera status lost after the switch back") - # cloud mode's connect cleared the kernel counters at the starting position - # and its hunt homed the head: bring it back - ctx.counters_rezeroed() - return_head(ctx) + # The service moved the head in cloud mode, and the controller's start + # zeroed the counters where it left it: the client's record brings it back. + cloud_mode_return(ctx, ev, log_offset) # cloud mode sets its own lid-lamp level (LLvl) and leaves it: hand back the level found lamp1 = hw.sysfs_read("pic/lid_led") ev["lid_lamp"] = {"before": lamp0, "after_cloud": lamp1} @@ -400,9 +406,11 @@ def mode_switch(ctx): st, body = fc.post("/settings", data={"homing_mode": "gfcloud"}) ctx.log("homing_mode=gfcloud for the homing -> %s %s", st, body if isinstance(body, str) else "") ctx.check(st == 200, "homing_mode=gfcloud -> %s %s", st, body) + session_at = session_mark() try: with ctx.grbl() as g: gfhome_homing(ctx, ev, g) + judge_whole_motions(ctx, ev, session_at) finally: if ev["homing_mode"] != "gfcloud": # Back to exactly what the machine had: unset is the empty @@ -413,6 +421,7 @@ def mode_switch(ctx): if not ev["homing_mode"] else fc.post("/settings", data={"homing_mode": ev["homing_mode"]})) ctx.log("restore homing_mode=%r -> %s", ev["homing_mode"], st) + camera_home_return(ctx, ev, session_at) ctx.log("PASS: grbl -> cloud (session, hunt with the lid open, lens homed, airflow unjudged) -> " "grbl (port open, %s), then $H homed in %.1f s", ev["grbl_state"], ev["homing_s"]) diff --git a/forgetest/forgetest/suite/homeoff.py b/forgetest/forgetest/suite/homeoff.py index 8597238..5675d26 100644 --- a/forgetest/forgetest/suite/homeoff.py +++ b/forgetest/forgetest/suite/homeoff.py @@ -6,31 +6,191 @@ """A camera home's offset pair, on the machine. Its own module, so that no other test's fingerprint moves. The homing is -the web-service session cloud.mode-switch runs; the head ends parked at -the home, as it does there. +the web-service session cloud.mode-switch runs. The helpers for a test +that lets the service move the head live here too: the motions ran whole, +and the head goes back to where the test found it, by the travel the +service's motions logged. No counter reading can say where that was: every +service motion zeroes the step counters at its start, and a home and every +controller start zero them again. """ +import re +import sys + +from ..baseline import XY_STEPS_PER_MM, counter_frame, counter_steps_per_mm, read_position from ..catalog import test -from .cloud import _HOMING_PATH, gfhome_homing -from .motion import wait_state +from ..runner import Failed +from . import cloud as logs # the log paths, read at each call +from .cloud import RETURN_MAX_MM, _HOMING_PATH, gfhome_homing, log_lines_since, log_size +from .motion import _drop_reference, clean_slate, machine_idle, wait_idle, wait_state HOME_X, HOME_Y = 4.5, -3.25 +# The client's own record of a motion's end: the kernel counters, which +# the motion zeroed at its start (the "(actual/expected)" line of the same +# motion is not this one). A print's park keeps the counters and drives +# them back, so the line after it is the print's net travel. +MOTION_END = re.compile(r"end positions \((-?\d+), (-?\d+), (-?\d+)\)") + + +def session_mark(path=None): + """Where a client's log ends now (gfhome's by default): the start of + the next session's record.""" + return log_size(path or logs.GFHOME_LOG) + + +def session_travel(offset, path=None): + """The X/Y travel of the service motions logged in a client's log + (gfhome's by default) since offset, in steps, or None when it cannot be + known. Every service motion zeroes the kernel counters at its start and + logs where they ended, so the travel is the sum of those ends. A service + motion runs at x8 or not at all (the client refuses any other mode), so + the steps are x8 steps. A motion refused before it moved logs no end + and moved nothing; a motion with no "end motion" (one that raised after + it moved, or a client killed inside it) leaves the travel unknown, and + so does a log rotated under the run.""" + path = path or logs.GFHOME_LOG + if log_size(path) < offset: + return None + x = y = 0 + inside, end = False, None + for ln in log_lines_since(path, offset): + ln = ln.rstrip() + m = MOTION_END.search(ln) + if ln.endswith(" start motion"): + if inside: + return None + inside, end = True, None + elif m and inside: + end = (int(m.group(1)), int(m.group(2))) + elif ln.endswith(" end motion") and inside: + if end: + x, y = x + end[0], y + end[1] + inside = False + return None if inside else (x, y) + + +def judge_whole_motions(ctx, ev, offset): + """Every service motion of the homing session since offset ran whole. + One that stopped short (the cooling engine's stop, a fault) left the + head short of the camera home, and the position declared there is + false.""" + short = [ln.strip()[:160] for ln in log_lines_since(logs.GFHOME_LOG, offset) + if "run stopped short" in ln + or ("_finish_action motion [" in ln and 'event ":completed"' not in ln)] + ev["short_motions"] = short + ctx.check(not short, "a service motion of the homing did not run whole, so the head stopped short of " + "the camera home and the declared position is false: %s", short) + + +def camera_home_return(ctx, ev, offset): + """Hand the head back to where the test found it after gfhome_homing + ran from gfhome's log offset, whatever the session decided. The head is + the session's logged travel from there, plus, once homed, whatever the + counters read since: the home zeroed them where the session left the + head and wrote the anchor, and a controller start since would have + removed it. Unhomed, the counters still hold the last motion's end, + which the travel already counts. The camera home is dropped first, + since its envelope need not hold the spot the head came from.""" + def back(): + now = [0, 0, 0] + if ev.get("homed"): + now = read_position() if counter_frame() is not None else None + _drop_reference(ctx, ctx.forgectrl) + # Read after the stop: a runner the stop ended has written its last line. + travel = session_travel(offset) + rec = ev["return"] = {"session_travel_steps": travel, "counters_since_home": now} + ctx.check(travel is not None and now is not None, + "the homing session's travel cannot be known (gfhome log %s, counters since the home %s): " + "the head is left where it stands and the hand-back moves nothing", travel, now) + _jog_back(ctx, rec, travel, now) + _hand_back(ctx, ev, "return", back) + + +def cloud_mode_return(ctx, ev, offset): + """Hand the head back to where the test found it after a stay in cloud + mode that began at the cloud client's log offset, on the GRBL + controller the switch back started: that start zeroed the counters + where the service left the head, and they hold whatever moved since. + The client's record of its motions is the rest (a hunt's lens travel is + Z, which a hand-back never touches).""" + def back(): + travel = session_travel(offset, logs.GFCLOUD_LOG) + now = read_position() + rec = ev["cloud_return"] = {"session_travel_steps": travel, "counters_since_start": now} + ctx.check(travel is not None and now is not None, + "the travel in cloud mode cannot be known (gfcloud log %s, counters %s): the head is left " + "where it stands and the hand-back moves nothing", travel, now) + _jog_back(ctx, rec, travel, now) + _hand_back(ctx, ev, "cloud_return", back) + + +def _hand_back(ctx, ev, key, back): + """Run a hand-back from a test's finally. The first failure wins: when + the test has failed already (the exception in flight), a hand-back that + fails as well is logged and recorded, and the test's own failure is the + one reported.""" + first = sys.exc_info()[1] + try: + back() + except Failed as e: + if first is None: + raise + ev.setdefault(key, {})["failed"] = str(e) + ctx.log("the head is not back, and the run has failed already: %s", e) + + +def _jog_back(ctx, rec, travel, since): + """Jog the head back by the service's travel (x8 steps) and what the + counters read since (the controller's own steps), on a controller + without a home. One more start then zeroes the counters where the head + began, which the rest of the catalog counts on.""" + spm = counter_steps_per_mm() + dx = travel[0] / XY_STEPS_PER_MM + since[0] / spm + dy = travel[1] / XY_STEPS_PER_MM + since[1] / spm + rec["mm"] = [round(dx, 3), round(dy, 3)] + ctx.log("the head is %.3f/%.3f mm from where the test found it", dx, dy) + ctx.check(abs(dx) <= RETURN_MAX_MM and abs(dy) <= RETURN_MAX_MM, + "a travel of %.1f/%.1f mm exceeds %.0f mm - not jogging back", dx, dy, RETURN_MAX_MM) + if abs(dx) >= 0.05 or abs(dy) >= 0.05: + with ctx.grbl() as g: + clean_slate(ctx, g) + start = read_position() + r = g.command("$J=G91 G21 X%.3f Y%.3f F2400" % (-dx, -dy)) + ctx.check(not any(x.startswith("error") for x in r), "the return jog was refused: %s", r) + states = wait_idle(ctx, g, timeout=60)[1] + ctx.check(states[-1:] != ["TIMEOUT"], "the return jog did not end: %s", states) + machine_idle(ctx) + end = read_position() + rec["counters_across_jog"] = [start, end] + ctx.check(start is not None and end is not None and abs((end[0] - start[0]) / spm + dx) < 0.1 + and abs((end[1] - start[1]) / spm + dy) < 0.1, + "the return jog did not travel %.3f/%.3f mm: counters %s -> %s", -dx, -dy, start, end) + _drop_reference(ctx, ctx.forgectrl) + ctx.counters_rezeroed() + ctx.log("head returned to where the test found it (%.3f/%.3f mm)", -dx, -dy) @test("homing.cloud-offsets", title="A camera home declares its offset pair, and the envelope reaches back to a " "negative one", - subsystem="homing", kind="auto", mode="grbl", est_min=5, - covers=_HOMING_PATH + [("grblhal-glowforge", "src/glowforge_homing.*")], + subsystem="homing", kind="auto", mode="grbl", est_min=7, + covers=_HOMING_PATH + [("grblhal-glowforge", "src/glowforge_homing.*"), + ("grblhal-glowforge", "src/glowforge_cooling.*")], requires=["forgectrl.auth", "motion.pacing"], - steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The head ends " - "parked at the home, as it does after cloud.mode-switch."], + steps=["Bed clear, lid closed; cloud credentials configured; the machine on the network. The head " + "travels to the camera home and back to where it was."], description="With cloud mode on, homing_mode = gfcloud and the camera-home offsets set to (4.5, -3.25), " - "$H runs the web-service homing session. When it ends, the controller declares the position " - "(4.5, -3.25): a camera-home offset may be negative, and the declared position is snapped to " - "the step grid. The work envelope reaches back to the home on the negative axis, since the head " - "stands there: a jog 2 mm out in +Y and 2 mm back ends at the home, accepted, and the same in " - "X. The homing mode and the offsets are put back as found. That an offset past the axis travel " - "is refused, and the envelope's edge on the grid, are the driver's unit and harness cases.") + "$H runs the web-service homing session. Every service motion of the session runs whole: " + "the runner reports to the cooling engine in the controller's place, and one stopped short " + "(the engine's stop, a fault) leaves the head short of the home and the position declared " + "there false. When the session ends, the controller declares the position (4.5, -3.25): a " + "camera-home offset may be negative, and the declared position is snapped to the step " + "grid. The work envelope reaches back to the home on the negative axis, since the head " + "stands there: a jog 2 mm out in +Y and 2 mm back ends at the home, accepted, and the same " + "in X. The homing mode and the offsets are put back as found, the camera home is dropped, " + "and the head is jogged back to where the test found it by the travel the session's " + "motions logged; a travel that cannot be known fails the test and moves nothing. That an " + "offset past the axis travel is refused, and the envelope's edge on the grid, are the " + "driver's unit and harness cases.") def cloud_offsets(ctx): fc = ctx.forgectrl ev = ctx.evidence @@ -38,12 +198,15 @@ def cloud_offsets(ctx): ctx.check(s0.get("cloud_enabled") == "1", "cloud mode is off: a camera home needs it on") found = {k: s0.get(k) or "" for k in ("homing_mode", "gfcloud_home_x", "gfcloud_home_y")} ev["found"] = found + session_at = None try: for k, v in (("homing_mode", "gfcloud"), ("gfcloud_home_x", str(HOME_X)), ("gfcloud_home_y", str(HOME_Y))): st, body = fc.post("/settings", data={k: v}) ctx.check(st == 200, "%s=%s -> %s %s", k, v, st, body) with ctx.grbl() as g: + session_at = session_mark() gfhome_homing(ctx, ev, g) + judge_whole_motions(ctx, ev, session_at) mpos = g.status_report().get("MPos") or (None, None, None) ev["declared"] = mpos ctx.log("the camera home declared %s", mpos) @@ -73,3 +236,5 @@ def cloud_offsets(ctx): return st_ == 200 or None took = ctx.wait_for(put, 20, poll=0.5) ctx.log("restore %s=%r -> %s", k, v, "taken" if took is not None else "not taken") + if session_at is not None: + camera_home_return(ctx, ev, session_at) diff --git a/forgetest/tests/test_camera_home_return.py b/forgetest/tests/test_camera_home_return.py new file mode 100644 index 0000000..8958702 --- /dev/null +++ b/forgetest/tests/test_camera_home_return.py @@ -0,0 +1,248 @@ +# Copyright 2026 514 LLC d/b/a OpenGlow +# Written by Scott Wiederhold +# https://community.openglow.org +# SPDX-License-Identifier: MIT + +"""A test that lets the service move the head hands it back where it found +it. Every service motion zeroes the step counters at its start, and a home +and every controller start zero them again, so no counter reading says +where the test found the head: the client's own record does. Every motion +logs the counters it ended on, and the travel is their sum (x8 steps, the +one mode a service motion runs at), plus what the counters read since. The +log lines are the machine's own, trimmed to the motion records: a homing +session whose one motion the cooling engine stopped short, and a whole +three-motion one. Runs against scratch logs and fakes for the controller.""" +import os +import shutil +import sys +import tempfile +import unittest + +from forgetest.runner import Failed +from forgetest.suite import cloud, homeoff + +STOPPED = """\ +2026-09-25T15:56:19.641824+00:00 gfhome[7387] INFO gfhome:home service action: motion (ready) +2026-09-25T15:56:19.644041+00:00 gfhome[7387] INFO machine:_motion start motion +2026-09-25T15:56:19.970342+00:00 gfhome[7387] INFO machine:_run_loop starting run +2026-09-25T15:56:20.905040+00:00 gfhome[7387] ERR machine:_run_loop run stopped short: 8130 of 21756 bytes played; the job did not finish +2026-09-25T15:56:20.918652+00:00 gfhome[7387] INFO machine:_feed_and_run end positions (actual/expected): X (4390/12115), Y (2682/7402), Z (0/0) +2026-09-25T15:56:20.937235+00:00 gfhome[7387] INFO machine:_motion_locked end positions (4390, 2682, 0) +2026-09-25T15:56:20.938026+00:00 gfhome[7387] INFO machine:_motion end motion +2026-09-25T15:56:20.938677+00:00 gfhome[7387] INFO basemachine:_finish_action motion [1588842275]: finished with event ":cancelled" +2026-09-25T15:56:21.145580+00:00 gfhome[7387] INFO gfhome:home motion completed +""" + +WHOLE = "".join( + "2026-09-07T20:%s gfhome[28592] INFO machine:_motion start motion\n" + "2026-09-07T20:%s gfhome[28592] INFO machine:_feed_and_run end positions (actual/expected): " + "X (%d/%d), Y (%d/%d), Z (0/0)\n" + "2026-09-07T20:%s gfhome[28592] INFO machine:_motion_locked end positions (%d, %d, 0)\n" + "2026-09-07T20:%s gfhome[28592] INFO machine:_motion end motion\n" + "2026-09-07T20:%s gfhome[28592] INFO basemachine:_finish_action motion [%d]: finished with event " + "\":completed\"\n" % (t, t, x, x, y, y, t, x, y, t, t, i) + for i, (t, x, y) in enumerate((("25:50.1", 4774, 4600), ("25:56.2", -71, -144), ("26:02.9", -13096, -7399)))) + + +class FakeGrbl: + def __init__(self, ctx): + self.ctx = ctx + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def command(self, line, timeout=None): + self.ctx.jogs.append(line) + return ["ok"] + + +class FakeCtx: + def __init__(self): + self.forgectrl = object() + self.jogs = [] + self.lines = [] + self.rezeroed = 0 + + def check(self, cond, msg, *args): + if not cond: + raise Failed(msg % args if args else msg) + + def log(self, msg, *args): + self.lines.append(msg % args if args else msg) + + def grbl(self): + return FakeGrbl(self) + + def counters_rezeroed(self): + self.rezeroed += 1 + + +class HandBackTests(unittest.TestCase): + def setUp(self): + self.dir = tempfile.mkdtemp() + self.log = os.path.join(self.dir, "gfhome.log") + self.cloudlog = os.path.join(self.dir, "gfcloud.log") + self.ctx = FakeCtx() + self.anchor = True + self.counters = [0, 0, 0] + self.restarts = 0 + self.spm = 213.333 # the controller's x32 scale + saved_logs = (cloud.GFHOME_LOG, cloud.GFCLOUD_LOG) + self.addCleanup(lambda: setattr(cloud, "GFHOME_LOG", saved_logs[0])) + self.addCleanup(lambda: setattr(cloud, "GFCLOUD_LOG", saved_logs[1])) + cloud.GFHOME_LOG, cloud.GFCLOUD_LOG = self.log, self.cloudlog + patches = { + "counter_frame": lambda: [1, 2] if self.anchor else None, + "read_position": lambda: list(self.counters), + "counter_steps_per_mm": lambda: self.spm, + "_drop_reference": self._restart, + "clean_slate": lambda ctx, g: None, + "wait_idle": self._jog_ends, + "machine_idle": lambda ctx: None, + } + saved = {k: getattr(homeoff, k) for k in patches} + self.addCleanup(lambda: [setattr(homeoff, k, v) for k, v in saved.items()]) + for k, v in patches.items(): + setattr(homeoff, k, v) + self.addCleanup(shutil.rmtree, self.dir, True) + + def _restart(self, ctx, fc): + # a controller start zeroes the counters where the head stands and + # removes the anchor + self.restarts += 1 + self.counters = [0, 0, 0] + self.anchor = False + + def _jog_ends(self, ctx, g, timeout=30.0): + # the jog played: the counters moved by what it asked, at the scale + words = dict((w[0], float(w[1:])) for w in self.ctx.jogs[-1].split()[2:4]) + self.counters = [self.counters[0] + round(words["X"] * self.spm), + self.counters[1] + round(words["Y"] * self.spm), 0] + return 0.0, ["Jog", "Idle"], {} + + def write(self, text, path=None): + with open(path or self.log, "a", encoding="utf-8", newline="\n") as f: + f.write(text) + + def expect_jog(self, dx, dy): + self.assertEqual(self.ctx.jogs, ["$J=G91 G21 X%.3f Y%.3f F2400" % (-dx, -dy)]) + + # -- the travel ------------------------------------------------------ + def test_the_travel_is_the_sum_of_the_motions_ends(self): + self.write(WHOLE) + self.assertEqual(homeoff.session_travel(0), (4774 - 71 - 13096, 4600 - 144 - 7399)) + + def test_the_travel_starts_at_the_mark(self): + self.write(WHOLE) + mark = homeoff.session_mark() + self.write(STOPPED) + self.assertEqual(homeoff.session_travel(mark), (4390, 2682)) + + def test_the_cloud_clients_log_is_read_the_same_way(self): + self.write(WHOLE.replace("gfhome[", "gfcloud["), self.cloudlog) + self.assertEqual(homeoff.session_travel(0, cloud.GFCLOUD_LOG), (4774 - 71 - 13096, 4600 - 144 - 7399)) + self.assertEqual(homeoff.session_travel(0), (0, 0)) # gfhome's log is its own + + def test_a_motion_with_no_end_leaves_the_travel_unknown(self): + self.write(STOPPED.split("machine:_motion_locked")[0]) + self.assertIsNone(homeoff.session_travel(0)) + + def test_a_log_rotated_under_the_run_leaves_the_travel_unknown(self): + self.write(STOPPED) + self.assertIsNone(homeoff.session_travel(os.path.getsize(self.log) + 1)) + + # -- the stopped-short judge -------------------------------------------- + def test_a_motion_stopped_short_fails_the_judge(self): + self.write(STOPPED) + ev = {} + with self.assertRaises(Failed): + homeoff.judge_whole_motions(self.ctx, ev, 0) + self.assertEqual(len(ev["short_motions"]), 2) + + def test_whole_motions_pass_the_judge(self): + self.write(WHOLE) + ev = {} + homeoff.judge_whole_motions(self.ctx, ev, 0) + self.assertEqual(ev["short_motions"], []) + + # -- after a camera home ------------------------------------------------- + def test_homed_the_head_goes_back_by_the_travel_and_the_counters_since(self): + # the home zeroed the counters; the test's own jogs left a step or so + self.write(STOPPED) + self.counters = [round(1.0 * self.spm), 0, 0] + since = self.counters[0] / self.spm + homeoff.camera_home_return(self.ctx, {"homed": True}, 0) + self.expect_jog(4390 / homeoff.XY_STEPS_PER_MM + since, 2682 / homeoff.XY_STEPS_PER_MM) + # dropped before the jog, and once more with the head back + self.assertEqual(self.restarts, 2) + self.assertEqual(self.counters, [0, 0, 0]) + self.assertEqual(self.ctx.rezeroed, 1) + + def test_unhomed_the_counters_are_not_counted_twice(self): + # a failed session leaves the last motion's end in the counters, + # which the travel already counts + self.write(STOPPED) + self.counters = [4390, 2682, 0] + self.anchor = False + homeoff.camera_home_return(self.ctx, {"homed": False}, 0) + self.expect_jog(4390 / homeoff.XY_STEPS_PER_MM, 2682 / homeoff.XY_STEPS_PER_MM) + + def test_an_unknown_travel_moves_nothing(self): + self.write(STOPPED.split("machine:_motion_locked")[0]) + with self.assertRaises(Failed): + homeoff.camera_home_return(self.ctx, {"homed": True}, 0) + self.assertEqual(self.ctx.jogs, []) + self.assertEqual(self.ctx.rezeroed, 0) + + def test_a_restart_since_the_home_moves_nothing(self): + # the anchor gone: the counters no longer read from the home + self.write(STOPPED) + self.anchor = False + with self.assertRaises(Failed): + homeoff.camera_home_return(self.ctx, {"homed": True}, 0) + self.assertEqual(self.ctx.jogs, []) + + def test_a_head_that_never_moved_is_not_jogged(self): + homeoff.camera_home_return(self.ctx, {}, 0) + self.assertEqual(self.ctx.jogs, []) + self.assertEqual((self.restarts, self.ctx.rezeroed), (1, 1)) + + # -- after cloud mode ---------------------------------------------------- + def test_after_cloud_mode_the_head_goes_back_by_the_clients_record(self): + # the service's re-hunt, cut short by the switch back: to the middle + # of the bed and one correction (the bench reference, 2026-09-25) + self.write(WHOLE[:WHOLE.index("2026-09-07T20:26:02.9")].replace("gfhome[", "gfcloud["), self.cloudlog) + ev = {} + homeoff.cloud_mode_return(self.ctx, ev, 0) + self.expect_jog((4774 - 71) / homeoff.XY_STEPS_PER_MM, (4600 - 144) / homeoff.XY_STEPS_PER_MM) + self.assertEqual(tuple(ev["cloud_return"]["session_travel_steps"]), (4774 - 71, 4600 - 144)) + # no drop before the jog (the switch back started the controller), one after + self.assertEqual((self.restarts, self.ctx.rezeroed), (1, 1)) + + def test_after_cloud_mode_an_unknown_travel_moves_nothing(self): + self.write(STOPPED.replace("gfhome[", "gfcloud[").split("machine:_motion_locked")[0], self.cloudlog) + with self.assertRaises(Failed): + homeoff.cloud_mode_return(self.ctx, {}, 0) + self.assertEqual(self.ctx.jogs, []) + + # -- the first failure wins -------------------------------------------- + def test_a_hand_back_that_fails_after_the_test_failed_is_logged_not_raised(self): + self.write(STOPPED.split("machine:_motion_locked")[0]) + ev = {"homed": True} + with self.assertRaises(Failed) as cm: + try: + raise Failed("the test's own failure") + finally: + homeoff.camera_home_return(self.ctx, ev, 0) + self.assertEqual(str(cm.exception), "the test's own failure") + self.assertIn("cannot be known", ev["return"]["failed"]) + self.assertTrue(any("the head is not back" in l for l in self.ctx.lines), self.ctx.lines) + self.assertEqual(self.ctx.jogs, []) + self.assertIsNone(sys.exc_info()[1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/forgetest/tests/test_cloud_suite.py b/forgetest/tests/test_cloud_suite.py index 1b96fdf..aed7b4f 100644 --- a/forgetest/tests/test_cloud_suite.py +++ b/forgetest/tests/test_cloud_suite.py @@ -28,6 +28,7 @@ import time import unittest import helpers +from forgetest import baseline from forgetest.runner import Context, Failed, Run from forgetest.suite import cloud @@ -566,8 +567,21 @@ class CloudSuiteTests(unittest.TestCase): # -- the mode switch: hunt with the lid open, then $H ----------------------- HUNT_RUN_SAMPLE = {"phase": "run", "verdict": "ok", "armed": False, "fan_gates": {"exhaust": {"state": "unjudged", "reading": 0, "floor": 500}}} + # A homing session's motions as gfhome logs them (the bench reference, + # 2026-09-25): a small correction, then into the camera home. + HOME_SESSION = [ + "2026-09-25T16:43:45.374951+00:00 gfhome[12982] INFO machine:_motion start motion", + "2026-09-25T16:43:45.853976+00:00 gfhome[12982] INFO machine:_motion_locked end positions (-13, -13, 0)", + "2026-09-25T16:43:45.854819+00:00 gfhome[12982] INFO machine:_motion end motion", + "2026-09-25T16:43:45.856076+00:00 gfhome[12982] INFO basemachine:_finish_action motion [1588864529]: " + "finished with event \":completed\"", + "2026-09-25T16:43:51.600734+00:00 gfhome[12982] INFO machine:_motion start motion", + "2026-09-25T16:43:53.605402+00:00 gfhome[12982] INFO machine:_motion_locked end positions (-13095, -7399, 0)", + "2026-09-25T16:43:53.606171+00:00 gfhome[12982] INFO machine:_motion end motion", + "2026-09-25T16:43:53.606916+00:00 gfhome[12982] INFO basemachine:_finish_action motion [1588864567]: " + "finished with event \":completed\""] - def mode_switch_setup(self, hunt_lines=None, home_complete=True, lid_late=False): + def mode_switch_setup(self, hunt_lines=None, home_complete=True, lid_late=False, home_lines=None): """The fakes a mode-switch run needs: grbl to answer $H, the lid lamp attr, homing_mode = gfcloud, the service lines landing on the switch to cloud - the client's start at once, its session and @@ -576,11 +590,16 @@ class CloudSuiteTests(unittest.TestCase): reads as a run to the cooling engine while it lasts) - the re-hunt on the lid close, and gfhome finishing the homing after $H. With lid_late the hunt is requested before the lid opens: the race the - test must call.""" + test must call. The step counters are the machine's: a controller + start zeroes them and removes the homing anchor, the home zeroes + them and writes it, and a jog moves them by what it asked.""" self.grbl = helpers.FakeGrbl().start() os.makedirs(self.sysfs + "pic", exist_ok=True) self._attr("pic/lid_led", "236") self.fc.state["settings"]["homing_mode"] = "gfcloud" + anchor = os.path.join(self.tmp, "grblhal.homed") + self.addCleanup(setattr, baseline, "ANCHOR_PATH", baseline.ANCHOR_PATH) + baseline.ANCHOR_PATH = anchor lines = fixture("huntlid") pre, post = cut(lines, "gfuiservice:__init__ INITIALIZED") hunt_part, close_part = cut(post, "_switch_event lid closed") @@ -588,6 +607,20 @@ class CloudSuiteTests(unittest.TestCase): hunt_part = hunt_lines(hunt_part) fc = self.fc + def controller_start(): + self._pos(0, 0, 3) + if os.path.exists(anchor): + os.remove(anchor) + + def jog(line): + if line.startswith("$J=G91"): + words = dict((w[0], float(w[1:])) for w in line[3:].split() if w[0] in "XY") + with open(self.sysfs + "cnc/position", "rb") as f: + x, y, z = struct.unpack("<3i", f.read(12)) + self._pos(x + round(words.get("X", 0.0) * baseline.XY_STEPS_PER_MM), + y + round(words.get("Y", 0.0) * baseline.XY_STEPS_PER_MM), z) + self.grbl.on_command = jog + def hunt(): self.append(hunt_part, delay=0.0) time.sleep(4.0) # the hunt outlasts the session wait on the bench @@ -608,7 +641,14 @@ class CloudSuiteTests(unittest.TestCase): hunt() threading.Thread(target=land, daemon=True).start() elif path == "/mode" and form.get("controller") == "grbl": + controller_start() self.grbl.state = "Idle" + elif path == "/controller/stop": + fc.state["mode"] = dict(fc.state["mode"], controller="standby", pid=0) + elif path == "/controller/start": + controller_start() + fc.state["mode"] = dict(fc.state["mode"], controller="running", pid=4300, motion="verified") + fc.state["status"]["homed"] = False return None self.fc.on_post = on_post @@ -623,7 +663,11 @@ class CloudSuiteTests(unittest.TestCase): def homing(): self.grbl.state = "Home" + with open(self.homelog, "ab") as f: + f.write(("\n".join(home_lines or self.HOME_SESSION) + "\n").encode()) time.sleep(0.4) + self._pos(0, 0, 3) + open(anchor, "w").close() self.grbl.state = "Idle" self.fc.state["status"]["homed"] = True if home_complete: @@ -658,6 +702,39 @@ class CloudSuiteTests(unittest.TestCase): self.assertEqual([r["state"] for r in ev["actions"]], ["open", "close"]) self.assertEqual(self.script.asked, []) self.assertTrue(any("PASS:" in l for l in run.lines), run.lines[-5:]) + # The head goes back where the test found it, each stretch by its + # client's record: the re-hunt of the excerpt ends 2/7 steps from + # where the cloud mode began, and the homing went into the corner. + self.assertEqual(tuple(ev["cloud_return"]["session_travel_steps"]), (2, 7)) + self.assertEqual(tuple(ev["return"]["session_travel_steps"]), (-13108, -7412)) + spm = baseline.XY_STEPS_PER_MM + self.assertEqual([l for l in self.grbl.sent if l.startswith("$J=")], + ["$J=G91 G21 X%.3f Y%.3f F2400" % (-2 / spm, -7 / spm), + "$J=G91 G21 X%.3f Y%.3f F2400" % (13108 / spm, 7412 / spm)]) + # a start after each jog, and the camera home dropped before the second + self.assertEqual([p for p, f in self.fc.posts if p.startswith("/controller/")], + ["/controller/stop", "/controller/start"] * 3) + with open(self.sysfs + "cnc/position", "rb") as f: + self.assertEqual(struct.unpack("<3i", f.read(12))[:2], (0, 0)) + + def test_mode_switch_a_hand_back_that_fails_too_does_not_hide_the_failure(self): + # gfhome never said it homed, and its last motion has no end on + # record: the travel is unknown, the head is not moved, and the + # failure reported is the test's own + hooks = self.mode_switch_setup(home_complete=False, home_lines=self.HOME_SESSION[:5]) + threading.Thread(target=self.wait_home_command, daemon=True).start() + self.assertFails(cloud.mode_switch, "no 'homing complete' line", hooks=hooks) + run = self.script.run + self.assertTrue(any("the head is not back" in l for l in run.lines), run.lines[-6:]) + self.assertIn("cannot be known", run.evidence["return"]["failed"]) + self.assertEqual(len([l for l in self.grbl.sent if l.startswith("$J=")]), 1) # the cloud stretch's + + def test_mode_switch_fails_when_the_travel_in_cloud_mode_cannot_be_known(self): + # the hunt's motion has no end on record (a client killed inside it) + hooks = self.mode_switch_setup( + hunt_lines=lambda part: [l for l in part if not l.endswith("machine:_motion end motion")]) + self.assertFails(cloud.mode_switch, "the travel in cloud mode cannot be known", hooks=hooks) + self.assertEqual([l for l in self.grbl.sent if l.startswith("$J=")], []) def test_mode_switch_opens_the_lid_behind_the_controller_and_ahead_of_the_hunt(self): # No controller starts with the enclosure open: the switch is made