diff --git a/forgetest/forgetest/bench.py b/forgetest/forgetest/bench.py index aca1cd8..207e993 100644 --- a/forgetest/forgetest/bench.py +++ b/forgetest/forgetest/bench.py @@ -32,10 +32,14 @@ from .log import data_dir, now_ts DEFAULT_TOOL_DIR = "/usr/share/forgetest/bench" -def _arg(name, type="str", default=None, help="", choices=None): +def _arg(name, type="str", default=None, help="", choices=None, flag=None): + """One form field. `flag` names the option the value is passed as + (--feed 600); without it the value is positional, in registry order.""" a = {"name": name, "type": type, "default": default, "help": help} if choices: a["choices"] = list(choices) + if flag: + a["flag"] = flag return a @@ -157,6 +161,21 @@ TOOLS = [ "desc": "Emission witness, disarm grace in Hold, stale-origin refusal, lid-IR characterization cut, armed " "kill on the expected-stop path (+ the separate controller restart). The operator's arm press is " "required for every drill; eye protection, fire watch, extinguisher, exhaust."}, + {"id": "resume-dark-lead", "title": "Pause / resume chain timing (dark lead)", "script": "resume_dark_lead.py", + "safety": "live", "where": "board", "ported": True, + "args": [_arg("run", "choice", "dry", "dry travel, or LIVE FIRE", ["dry", "live"], flag="--run"), + _arg("mode", "choice", "m3", "laser mode for a live run", ["m3", "m4"], flag="--mode"), + _arg("power", "int", 400, "live: S value", flag="--power"), + _arg("feed", "float", 600.0, "feed rate", flag="--feed"), + _arg("len", "float", 60.0, "move length in mm (+X)", flag="--len"), + _arg("passes", "int", 1, "alternating +X/-X moves", flag="--passes"), + _arg("secs", "float", 45.0, "sampling window", flag="--secs"), + _arg("auto", "str", "", "dry only: 'P,R' seconds to send ! and ~ unattended", flag="--auto")], + "desc": "Samples LASER_ON, FIRE, HV_ENABLE and the charge-pump watchdog straight off the SoC pads " + "across a pause and a resume, with motion dated from the kernel counters: how long HV survives " + "a pause, how fast the chain re-arms, and - on a live run - the dark lead between FIRE and " + "LASER_ON that a resumed cut loses. Dry by default; --run live needs the arm press, eye " + "protection, fire watch, extinguisher, exhaust."}, # -- host-side harnesses (CI) ------------------------------------------------------ {"id": "laser-stream-test", "title": "Laser pulse-stream emission harness", "script": "laser_stream_test.py", "safety": "dry", "where": "host", "ported": False, "args": [], @@ -201,8 +220,8 @@ class Bench: for spec in tool.get("args", []): raw = args.get(spec["name"], spec.get("default")) if raw is None or raw == "": - if spec.get("default") is None: - continue # optional and absent + if spec.get("default") in (None, ""): + continue # optional and absent: not passed at all raw = spec["default"] try: if spec["type"] == "int": @@ -219,6 +238,8 @@ class Bench: return False, None, "%s: invalid characters" % spec["name"] except (TypeError, ValueError): return False, None, "%s: invalid %s" % (spec["name"], spec["type"]) + if spec.get("flag"): + argv.append(spec["flag"]) argv.append(val) argv += list(tool.get("argv_fixed_after", [])) return True, argv, None diff --git a/forgetest/forgetest/suite/cloud.py b/forgetest/forgetest/suite/cloud.py index 0c00f7d..0bcd7fc 100644 --- a/forgetest/forgetest/suite/cloud.py +++ b/forgetest/forgetest/suite/cloud.py @@ -702,3 +702,162 @@ def pause_resume(ctx): "resume on the second, and did the job finish and the app show it complete?") settle_cloud(ctx, offset) ctx.log("PASS: button pause/resume mid-print, job completed and parked") + + +@test("cloud.interlock-abort-park", title="The interlock loop aborts a cloud print, and the park ignores " + "the lid", + subsystem="cloud", kind="live", est_min=8, + covers=_CLOUD_COVERS, requires=["cloud.lid-abort"], + steps=[CLOUD_STEP, "The app open in a browser; scrap on the bed and a small engrave/score job ready.", + "Be able to open the remote-interlock loop: unplug the Pro's interlock plug, or pull the " + "jumper at J8 on a Basic/Plus. Restore it at the end.", + "Print from the app and press the button when it lights; open the interlock a few seconds " + "into the run, then open the lid while the head is parking."], + description="In cloud mode the remote-interlock loop is the lid's equal, and the park is immune to " + "both: opening the loop mid-print stops the run and cancels the job, the head returns " + "to the job start, and opening the lid while that park is running does not interrupt " + "it - the print still ends ':cancelled' with the park reported complete.") +def interlock_abort_park(ctx): + ev = ctx.evidence + sw = (ctx.forgectrl.status().get("switches") or {}) + ctx.check(sw.get("interlock_ok"), "the interlock loop already reads open - close it before this test") + offset = enter_cloud(ctx) + ctx.instruct(APP_PRINT_CUE) + got = wait_print_running(ctx, offset, 300) + ctx.check(got, "the print never reached its run within 300 s (not started, or the button not pressed)") + ctx.instruct("The head is moving. Open the INTERLOCK loop now (unplug it / pull the jumper), then " + "click Done.") + sw = (ctx.forgectrl.status().get("switches") or {}) + ev["interlock_ok_after_pull"] = sw.get("interlock_ok") + ctx.check(sw.get("interlock_ok") is False, + "the interlock still reads closed - the loop was not opened (switches: %s)", sw) + stop_line = "interlock opened mid-run; stopping motion" + got = wait_log(ctx, offset, [stop_line, "start return home"], 90) + ctx.check(got[stop_line], "the interlock open did not stop the run") + ctx.check(got["start return home"], "the abort did not start the return home") + # The lid goes up while the park is running: it must change nothing. + ctx.instruct("The head is on its way back. Open the LID now as well, then click Done - leave both " + "open until the head has stopped.") + done = wait_log(ctx, offset, ["return home complete"], 90) + fin = wait_action_finished(ctx, offset, "print", 60) + ev["log"] = {stop_line: message(got[stop_line]), + "start return home": message(got["start return home"]), + "return home complete": message(done["return home complete"]), + "print finished": message(fin)} + for k, v in ev["log"].items(): + ctx.log(" %s: %s", k, "seen" if v else "MISSING") + ctx.check(done["return home complete"], + "the park did not run to completion with the lid opened during it") + sw = (ctx.forgectrl.status().get("switches") or {}) + ev["switches_at_return"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")} + ctx.check(sw.get("lid") is False, + "the lid was not open at the end of the park - the park's immunity was not exercised") + ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the park") + kpos = read_position() + ev["kernel_counters_after_park"] = kpos + ctx.log("kernel counters after the park: %s", kpos) + ctx.check(kpos is not None and abs(kpos[0]) <= 3 and abs(kpos[1]) <= 3, + "the head did not come back to the job start (kernel counters %s)", kpos) + ctx.check(fin and CANCELLED in fin, "the print did not end ':cancelled': %s", + message(fin) or "no finish line") + st, cs = ctx.forgectrl.get("/cool/status") + ev["armed_after"] = cs.get("armed") if isinstance(cs, dict) else None + ev["latch_locked"] = latch_locked() + ctx.check(not ev["armed_after"], "armed window still open after the abort") + ctx.check(ev["latch_locked"], "kernel latch not locked after the abort") + ctx.confirm("Did the head stop when the interlock opened and go back to the corner without the open " + "lid interrupting it, and does the app show the print as cancelled?") + ctx.instruct("Close the lid and restore the interlock loop (plug/jumper back in), then click Done.") + sw = (ctx.forgectrl.status().get("switches") or {}) + ev["restored"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")} + ctx.check(sw.get("interlock_ok"), "the interlock loop is still open - restore it before continuing") + settle_cloud(ctx, offset) + ctx.log("PASS: interlock open -> stop, cancelled, park completed through an open lid") + + +@test("cloud.pause-cancel-paths", title="A paused cloud print is cancelled by the lid, and a running one " + "by the app", + subsystem="cloud", kind="live", est_min=12, + covers=_CLOUD_COVERS, requires=["cloud.pause-resume", "cloud.lid-abort"], + steps=[CLOUD_STEP, "The app open in a browser; scrap on the bed and a small engrave/score job " + "ready - the test runs TWO prints.", + "Print 1: press the button to start, press it again a few seconds in (pause), then open " + "the lid.", + "Print 2: press the button to start, then cancel the print from the app while it runs."], + description="The two ways a print ends other than finishing, each from the state the factory ends " + "it in: a job paused on the button is cancelled by the lid - there is no resume past a " + "lid open - and a running job is cancelled from the app. Both take the same tail: the " + "motion stops, the head parks back at the job start, the latch relocks and the armed " + "window closes, and the print ends ':cancelled'.") +def pause_cancel_paths(ctx): + ev = ctx.evidence + offset = enter_cloud(ctx) + + # -- print 1: paused on the button, then the lid ------------------------ + ctx.instruct(APP_PRINT_CUE) + got = wait_print_running(ctx, offset, 300) + ctx.check(got, "print 1 never reached its run within 300 s (not started, or the button not pressed)") + ctx.instruct("The head is moving. Press the button once NOW (pause), watch it stop and back up a " + "few millimeters, then click Done.") + got = wait_log(ctx, offset, ["button pressed mid-run; pausing", "paused at"], 90) + ev["paused"] = {k: bool(v) for k, v in got.items()} + ctx.check(got["button pressed mid-run; pausing"], "the press did not pause print 1") + ctx.check(got["paused at"], "the pause did not settle (no 'paused at')") + ctx.instruct("The print is paused. Open the lid NOW, then click Done - leave it open until the head " + "has returned to the corner.") + lid_stop = "lid opened mid-run; stopping motion" + got = wait_log(ctx, offset, [lid_stop, "start return home", "return home complete"], 120) + fin1 = wait_action_finished(ctx, offset, "print", 60) + ev["lid_from_pause"] = {k: message(v) for k, v in got.items()} + ev["lid_from_pause"]["print finished"] = message(fin1) + for k, v in ev["lid_from_pause"].items(): + ctx.log(" [print 1] %s: %s", k, "seen" if v else "MISSING") + ctx.check(got[lid_stop], "the lid did not end the paused print") + ctx.check(got["return home complete"], "the paused print did not park to completion") + ctx.check(fin1 and CANCELLED in fin1, "print 1 did not end ':cancelled': %s", + message(fin1) or "no finish line") + ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after print 1's park") + kpos = read_position() + ev["counters_after_print1"] = kpos + ctx.check(kpos is not None and abs(kpos[0]) <= 3 and abs(kpos[1]) <= 3, + "print 1 did not come back to the job start (kernel counters %s)", kpos) + st, cs = ctx.forgectrl.get("/cool/status") + ev["armed_after_print1"] = cs.get("armed") if isinstance(cs, dict) else None + ev["latch_locked_after_print1"] = latch_locked() + ctx.check(not ev["armed_after_print1"], "armed window still open after the paused print was cancelled") + ctx.check(ev["latch_locked_after_print1"], + "kernel latch not locked after the paused print was cancelled") + ctx.instruct("Close the lid, then click Done.") + settle_cloud(ctx, offset) + + # -- print 2: cancelled from the app ------------------------------------ + offset = log_size(GFCLOUD_LOG) + ctx.instruct(APP_PRINT_CUE) + got = wait_print_running(ctx, offset, 300) + ctx.check(got, "print 2 never reached its run within 300 s (not started, or the button not pressed)") + ctx.instruct("The head is moving. Cancel the print from the app now, then click Done.") + svc_stop = "action cancelled mid-run; stopping motion" + got = wait_log(ctx, offset, [svc_stop, "start return home", "return home complete"], 120) + fin2 = wait_action_finished(ctx, offset, "print", 60) + ev["service_cancel"] = {k: message(v) for k, v in got.items()} + ev["service_cancel"]["print finished"] = message(fin2) + for k, v in ev["service_cancel"].items(): + ctx.log(" [print 2] %s: %s", k, "seen" if v else "MISSING") + ctx.check(got[svc_stop], "the app's cancel did not stop the run") + ctx.check(got["return home complete"], "the cancelled print did not park to completion") + ctx.check(fin2 and CANCELLED in fin2, "print 2 did not end ':cancelled': %s", + message(fin2) or "no finish line") + ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after print 2's park") + kpos = read_position() + ev["counters_after_print2"] = kpos + ctx.check(kpos is not None and abs(kpos[0]) <= 3 and abs(kpos[1]) <= 3, + "print 2 did not come back to the job start (kernel counters %s)", kpos) + st, cs = ctx.forgectrl.get("/cool/status") + ev["armed_after_print2"] = cs.get("armed") if isinstance(cs, dict) else None + ev["latch_locked_after_print2"] = latch_locked() + ctx.check(not ev["armed_after_print2"], "armed window still open after the app cancel") + ctx.check(ev["latch_locked_after_print2"], "kernel latch not locked after the app cancel") + ctx.confirm("Did both prints end back at the corner, dark, and does the app show both as cancelled?") + settle_cloud(ctx, offset) + ctx.log("PASS: a paused print cancelled by the lid and a running print cancelled from the app both " + "stopped, parked, relocked and reported ':cancelled'") diff --git a/forgetest/forgetest/suite/laser.py b/forgetest/forgetest/suite/laser.py index 9413be4..2096dff 100644 --- a/forgetest/forgetest/suite/laser.py +++ b/forgetest/forgetest/suite/laser.py @@ -17,7 +17,7 @@ import time from ..catalog import test from .. import hw from ..runner import Failed -from .motion import kernel_xy_mm, check_kernel_returned +from .motion import kernel_xy_mm, check_kernel_returned, wait_state, wait_idle, drain_text _LASER_COVERS = [("grblhal-glowforge", "src/**"), ("kernel-module-glowforge", "**"), ("forgectrl", "src/super.c"), ("forgectrl", "src/cool.c"), @@ -591,3 +591,113 @@ def lid_cancel_mid_fire(ctx): ctx.sleep(1) ctx.log("PASS: lid open mid-burn -> emission 0 at +%s s, cancelled, reset without alarm, returned (drift %.3f mm), " "button latch SET", zero_at, drift) + + +@test("laser.pause-resume-live", title="Button pause and resume mid-burn: emission stops, the window stays " + "open, the cut continues", + subsystem="laser", kind="live", est_min=6, + covers=_LASER_COVERS + [("grblhal-glowforge", "src/glowforge_switches.c"), + ("grblhal-glowforge", "src/glowforge_switch_map.h")], + requires=["laser.emission-witness", "motion.button-hold-resume"], + steps=["Scrap under the head with 60 mm of free +X travel; lid closed; exhaust on.", + "Press the physical button when it lights white (arm). A few seconds into the cut press " + "it once (pause), wait about 3 seconds, and press it once more (resume). Let the job finish.", + "Keep the pause short: the armed window's idle grace closes it after about a minute in a " + "hold, and a job that disarms cannot resume its emission."], + description="The button's pause/resume during a live cut, the factory's behavior on the machine: " + "the first press feed-holds the job and emission stops, the laser latch stays " + "UNLOCKED and the armed window stays open (a pause is not a cancel), and the second " + "press resumes the cut from where it stopped and the job runs to its end. GRBL mode " + "has no backtrack - the kernel refuses one on a live-streamed ring - so the restart " + "picks up where the deceleration ended.") +def pause_resume_live(ctx): + ev = ctx.evidence + with ctx.grbl() as g, LiveJob(ctx, g): + prepare(ctx, g) + start = g.status_report()["MPos"] + ev["start"] = start + ctx.instruct(ARM_CUE % "60 mm +X") + stream(g, ["G91", "G21", "M3", "S400", "G1 X60 F300", "M5", "G90", "M2"]) + t0 = time.time() + smp = None + seen = False + while time.time() - t0 < 240: + ctx.checkpoint() + smp = sample(ctx) + if smp and smp["emission"] and smp["emission"] > 0: + seen = True + break + time.sleep(0.15) + if not seen: + g.realtime(0x18) + raise Failed("no emission seen within 240 s (arm refused, or no button press)") + ev["emission_running"] = smp["emission"] + ctx.log("emission live (%s) - asking the operator to pause", smp["emission"]) + ctx.instruct("The laser is cutting. Press the button ONCE now (pause), then click Done - " + "do not wait long before the next step.") + st = wait_state(ctx, g, "Hold", 8) + ctx.check(st is not None, "the press did not hold the job (state %s)", g.status_report()["state"]) + text = drain_text(g, 0.5) + ev["hold_state"] = st["state"] + ev["pause_message"] = "job paused" in text + # emission has to be gone, and the window has to still be open: a + # pause that relocked would need a fresh arm press to go on. + paused = [] + t1 = time.time() + while time.time() - t1 < 4: + s = sample(ctx) + if s: + paused.append((round(time.time() - t1, 2), s["emission"], s["armed"])) + time.sleep(0.2) + for p in paused: + ctx.log(" paused %s", p) + ev["paused_trail"] = paused + ev["emission_zero_when_paused"] = all(e == 0 for _t, e, _a in paused[-8:]) if paused else None + ev["armed_while_paused"] = paused[-1][2] if paused else None + ilk = hw.sysfs_int("cnc/interlock_circuit") + ev["latch_locked_while_paused"] = ilk is not None and bool(ilk & (1 << 3)) + ctx.log("paused: %s, emission 0 = %s, armed = %s, latch locked = %s", ev["hold_state"], + ev["emission_zero_when_paused"], ev["armed_while_paused"], ev["latch_locked_while_paused"]) + ctx.check(ev["emission_zero_when_paused"], "emission did not stop while the job was paused: %s", paused) + ctx.check(ev["armed_while_paused"], "the armed window closed on the pause (a pause is not a cancel)") + ctx.check(not ev["latch_locked_while_paused"], + "the kernel latch relocked on the pause - the resume could not fire without a new arm press") + ctx.instruct("Press the button once more now (resume), then click Done and let the job finish.") + st = wait_state(ctx, g, "Run", 10) + ev["resumed_state"] = st["state"] if st else g.status_report()["state"] + ctx.check(st is not None, "the second press did not resume the job (state %s)", ev["resumed_state"]) + back = False + t2 = time.time() + trail = [] + while time.time() - t2 < 20: + ctx.checkpoint() + s = sample(ctx) + if s: + trail.append((round(time.time() - t2, 2), s["emission"])) + if s["emission"] and s["emission"] > 0: + back = True + break + time.sleep(0.15) + ev["emission_back_after_s"] = trail[-1][0] if trail else None + ev["resume_trail"] = trail[-6:] + ctx.log("emission after the resume: %s", trail[-6:]) + ctx.check(back, "emission did not return after the resume: %s", trail[-6:]) + # the job runs to its end on its own + peak, states, st = wait_idle(ctx, g, 60) + ev["states_after_resume"] = states + ctx.check("TIMEOUT" not in states, "the resumed job did not finish: %s", states) + moved = st["MPos"][0] - start[0] + ev["moved_mm"] = round(moved, 3) + ctx.log("finished: moved %.3f mm of 60 (states %s)", moved, states) + ctx.check(abs(moved - 60.0) <= 0.1, "the resumed job did not finish its move (%.3f mm of 60)", moved) + ctx.check(ctx.forgectrl.wait_idle(15, abort=ctx.aborted), "machine not idle after the job") + s = sample(ctx) + ev["armed_after_job"] = s["armed"] if s else None + ctx.confirm("Did the burn stop on the first press and start again on the second, continuing the " + "same line to its end? (A small mark where it restarted is expected - GRBL mode does " + "not backtrack.)") + g.command("$J=G91X-60F1200") + wait_idle(ctx, g, 30) + g.command("G90") + ctx.log("PASS: button paused the burn (emission 0, armed kept, latch unlocked), the next press resumed " + "it and the job finished (%.3f mm)", ev["moved_mm"]) diff --git a/forgetest/forgetest/suite/motion.py b/forgetest/forgetest/suite/motion.py index 0ac6a05..e4e8bce 100644 --- a/forgetest/forgetest/suite/motion.py +++ b/forgetest/forgetest/suite/motion.py @@ -339,9 +339,11 @@ def _liveness_masked_restart(ctx, fc, ev): covers=_MOTION_COVERS, requires=["motion.pacing"], steps=["Bed clear; the head needs 40 mm of free +X travel."], description="A jog-cancel (0x85) stops a jog short of its target and returns to Idle with " - "position preserved; a ^X abort mid-move decelerates under control into Alarm " - "with machine position retained, $X recovers to Idle, and a subsequent jog runs " - "(no driver wedge: the rail never cycled).") + "position preserved; a ^X abort mid-move (what the sender's Stop sends) " + "decelerates under control into Alarm with machine position retained, leaves the " + "head where it stopped - no return-to-start, that is the lid policy's move alone - " + "and $X recovers to Idle with a subsequent jog running (no driver wedge: the rail " + "never cycled).") def cancel_abort(ctx): ev = ctx.evidence with ctx.grbl() as g: @@ -374,6 +376,19 @@ def cancel_abort(ctx): ctx.log("$X -> %s", r) st = wait_state(ctx, g, "Idle", 5) ctx.check(st is not None, "$X did not recover to Idle") + # A sender abort is not a lid cancel: it stops where it stopped and + # the head stays there. Only the lid/interlock policy returns to the + # job start, and an unasked-for return move would be a surprise to + # whoever pressed Stop. + text = drain_text(g, 2.0) + ev["abort_returned_home"] = "returned to the job start" in text + p3 = g.status_report()["MPos"] + ev["abort_drift_after_recovery_mm"] = round(abs(p3[0] - p2[0]), 3) + ctx.log("after $X: %s (moved %.3f mm since the abort)", p3, ev["abort_drift_after_recovery_mm"]) + ctx.check(not ev["abort_returned_home"], "a sender abort triggered the return-to-start motion") + ctx.check(ev["abort_drift_after_recovery_mm"] <= 0.05, + "the head moved %.3f mm on its own after a sender abort", + ev["abort_drift_after_recovery_mm"]) # a jog after the abort proves the drivers are alive; return to start back = -(p2[0] - start[0]) r = g.command("$J=G91X%.3fF2400" % back) @@ -566,14 +581,14 @@ def kernel_xy_mm(ctx): return float(pos.get("x", 0.0)), float(pos.get("y", 0.0)) -def check_kernel_returned(ctx, ev, k0, tol_mm=0.1): +def check_kernel_returned(ctx, ev, k0, tol_mm=0.1, tag=""): """After a return-to-start: the kernel counters must be back where the job started too. grbl's own drift can read 0.000 while the head never moved (a run the kernel did not take), which is exactly the failure that must not pass.""" k1 = kernel_xy_mm(ctx) kdrift = max(abs(k1[0] - k0[0]), abs(k1[1] - k0[1])) - ev["kernel_drift_mm"] = round(kdrift, 3) + ev[(tag + "_" if tag else "") + "kernel_drift_mm"] = round(kdrift, 3) ctx.log("kernel counters: start (%.2f, %.2f) -> now (%.2f, %.2f), drift %.3f mm", k0[0], k0[1], k1[0], k1[1], kdrift) ctx.check(kdrift <= tol_mm, @@ -591,6 +606,42 @@ def drain_text(g, seconds): return text +def expect_cancel_and_return(ctx, g, ev, start, k0, why, tag): + """The cancel policy's whole tail, shared by every trigger that ends a + job this way (lid or interlock, from Run or from a hold): the reason is + reported, the controller resets without an alarm and with the position + kept, and the head goes back to where the job started - which the KERNEL + counters have to confirm, not grbl's belief about them.""" + text = drain_text(g, 3.0) + msgs = [ln for ln in text.splitlines() + if ln.startswith("[MSG:") or "help]" in ln or ln.startswith("ALARM")] + ev[tag + "_messages"] = msgs + ctx.log("[%s] controller: %s", tag, msgs) + cancel_msg = "%s - job cancelled" % why + ctx.check(cancel_msg in text, "[%s] the job was not cancelled with %r as the reason", tag, why) + ctx.check("help]" in text, "[%s] no reset banner after the cancel (the sender must see the job end)", tag) + ctx.check("ALARM" not in text, "[%s] an alarm was raised on the cancel (position should be kept)", tag) + t0 = time.time() + returned = "returned to the job start" in text + while not returned and time.time() - t0 < 30: + ctx.checkpoint() + text += g.drain() + returned = "returned to the job start" in text + time.sleep(0.2) + ev[tag + "_returned_message"] = returned + ctx.check(returned, "[%s] the head did not report returning to the job start within 30 s", tag) + st = wait_state(ctx, g, "Idle", 5) + ctx.check(st is not None, "[%s] not Idle after the return (state %s)", tag, + g.status_report()["state"]) + drift = max(abs(st["MPos"][i] - start[i]) for i in range(2)) + ev[tag + "_drift_mm"] = round(drift, 3) + ctx.log("[%s] back at the job start: drift %.3f mm", tag, drift) + ctx.check(drift <= 0.05, "[%s] head not back at the job start (drift %.3f mm)", tag, drift) + machine_idle(ctx, 10) + check_kernel_returned(ctx, ev, k0, tag=tag) + return drift + + @test("motion.button-hold-resume", title="The button pauses and resumes a job", subsystem="motion", kind="operator", est_min=2, covers=_LID_COVERS, requires=["motion.pacing"], @@ -635,16 +686,19 @@ def button_hold_resume(ctx): ctx.log("PASS: button press held the job (%s), the next press resumed it, target reached", ev["held_state"]) -@test("motion.lid-cancel-home", title="Lid open during a job cancels it and returns to the job start", - subsystem="motion", kind="operator", est_min=3, +@test("motion.lid-cancel-home", title="Lid open during a job - running or paused - cancels it and returns " + "to the job start", + subsystem="motion", kind="operator", est_min=5, covers=_LID_COVERS, requires=["motion.pacing", "motion.cancel-abort"], steps=["Bed clear; the head needs 40 mm of free +X travel. No laser is involved.", - "Open the lid when told, and leave it open until the head has come back."], + "Open the lid when told, and leave it open until the head has come back (twice: once " + "with the job running, once with it paused on the button)."], description="A travel job is running when the lid opens: the job parks (planned deceleration), " "the reason is reported, the controller resets (position kept, no alarm - the " "sender's job is over), and the head returns on its own to where the job started " - "with the lid still open; the controller ends Idle at the start position. With " - "lid_policy=cancel (the default).") + "with the lid still open; the controller ends Idle at the start position. The same " + "job paused on the button takes the same path - a lid open from the hold cancels " + "and returns, it never resumes. With lid_policy=cancel (the default).") def lid_cancel_home(ctx): ev = ctx.evidence policy = (ctx.forgectrl.settings() or {}).get("lid_policy") or "cancel" @@ -662,35 +716,7 @@ def lid_cancel_home(ctx): ctx.sleep(0.5) ctx.check(g.status_report()["state"].startswith("Run"), "the move did not start") ctx.instruct("The head is moving. Open the lid NOW and leave it open, then click Done.") - text = drain_text(g, 3.0) - ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:") or "help]" in ln - or ln.startswith("ALARM")] - ctx.log("controller: %s", ev["messages"]) - ctx.check("lid opened - job cancelled" in text, "the lid open was not reported as cancelling the job") - ctx.check("help]" in text, "no reset banner after the cancel (the sender must see the job end)") - ctx.check("ALARM" not in text, "an alarm was raised on the cancel (position should be kept)") - # the head returns on its own; wait for it to report back - t0 = time.time() - returned = False - while time.time() - t0 < 30: - ctx.checkpoint() - text += g.drain() - if "returned to the job start" in text: - returned = True - break - time.sleep(0.2) - ev["returned_message"] = returned - ctx.check(returned, "the head did not report returning to the job start within 30 s") - st = wait_state(ctx, g, "Idle", 5) - ctx.check(st is not None, "not Idle after the return (state %s)", g.status_report()["state"]) - drift = abs(st["MPos"][0] - start[0]) - ev["drift_mm"] = round(drift, 3) - ctx.log("back at the job start: drift %.3f mm (lid still open)", drift) - ctx.check(drift <= 0.05, "head not back at the job start (drift %.3f mm)", drift) - # what the MACHINE did: the kernel counters must agree (the return - # move must have been played, not only planned) - machine_idle(ctx, 10) - check_kernel_returned(ctx, ev, k0) + drift = expect_cancel_and_return(ctx, g, ev, start, k0, "lid opened", "running") sw = (ctx.forgectrl.status().get("switches") or {}) ev["lid_at_return"] = sw.get("lid") ctx.instruct("Close the lid, then click Done.") @@ -701,7 +727,164 @@ def lid_cancel_home(ctx): wait_idle(ctx, g, 15) g.command("$J=G91X-5F1200") wait_idle(ctx, g, 15) + + # -- the same cancel, entered from a hold --------------------------- + # A job paused on the button must not be resumable past a lid open: + # the armed window and the hardware button latch have to agree, so + # the lid ends the job here exactly as it does from Run. + start2 = g.status_report()["MPos"] + k1 = kernel_xy_mm(ctx) + ev["hold_start"] = start2 + g.command("G91") # the reset restored G90 + g.command("G1X40F300", timeout=0.5) + ctx.sleep(0.5) + ctx.check(g.status_report()["state"].startswith("Run"), "the second move did not start") + ctx.instruct("The head is moving again. Press the button once now (pause), then click Done.") + st = wait_state(ctx, g, "Hold", 8) + ctx.check(st is not None, "the press did not hold the job (state %s)", g.status_report()["state"]) + held = drain_text(g, 0.5) + ev["hold_state"] = st["state"] + ev["hold_pause_message"] = "job paused" in held + ctx.log("paused: %s; message seen: %s", st["state"], ev["hold_pause_message"]) + ctx.instruct("The job is paused. Open the lid NOW and leave it open, then click Done.") + hold_drift = expect_cancel_and_return(ctx, g, ev, start2, k1, "lid opened", "hold") + ctx.check(not g.status_report()["state"].startswith("Hold"), + "the controller is still holding after the lid cancelled the paused job") + ctx.instruct("Close the lid, then click Done.") + ctx.sleep(1) + r = g.command("$J=G91X5F1200") + ctx.check(not any(x.startswith("error") for x in r), "jog refused after the paused cancel: %s", r) + wait_idle(ctx, g, 15) + g.command("$J=G91X-5F1200") + wait_idle(ctx, g, 15) g.command("G90") machine_idle(ctx) - ctx.log("PASS: lid open cancelled the job, reset without alarm, head returned to the start (drift %.3f mm)", - ev["drift_mm"]) + ctx.log("PASS: lid open cancelled the job from Run (drift %.3f mm) and from the hold (drift %.3f mm), " + "reset without alarm, head returned to the start both times", drift, hold_drift) + + +@test("motion.interlock-cancel-park", title="The interlock loop cancels a job like the lid, and the return " + "home ignores an open lid", + subsystem="motion", kind="operator", est_min=4, + covers=_LID_COVERS, requires=["motion.lid-cancel-home"], + steps=["Bed clear; the head needs 60 mm of free +X travel. No laser is involved.", + "Be able to open the remote-interlock loop: unplug the Pro's interlock plug, or pull the " + "jumper at J8 on a Basic/Plus. Restore it at the end.", + "Open the interlock when told; then open the lid while the head is on its way back."], + description="The remote-interlock loop is the lid's equal in the cancel policy: opening it mid-job " + "cancels the job with 'interlock open' named as the reason and sends the head back to " + "the job start. Opening the lid during that return changes nothing - the park hides the " + "door for exactly this reason and runs to completion, as the factory's does.") +def interlock_cancel_park(ctx): + ev = ctx.evidence + policy = (ctx.forgectrl.settings() or {}).get("lid_policy") or "cancel" + ev["lid_policy"] = policy + ctx.check(policy == "cancel", "lid_policy is %r; this test needs cancel", policy) + sw = (ctx.forgectrl.status().get("switches") or {}) + ctx.check(sw.get("interlock_ok"), "the interlock loop already reads open - close it before this test") + with ctx.grbl() as g: + clean_slate(ctx, g) + start = g.status_report()["MPos"] + k0 = kernel_xy_mm(ctx) + ev["start"] = start + ev["kernel_start"] = k0 + g.command("M5") + g.command("G91") + g.command("G1X60F300", timeout=0.5) # a 12 s move + ctx.sleep(0.5) + ctx.check(g.status_report()["state"].startswith("Run"), "the move did not start") + ctx.instruct("The head is moving. Open the INTERLOCK loop now (unplug it / pull the jumper) and " + "leave it open, then click Done.") + sw = (ctx.forgectrl.status().get("switches") or {}) + ev["interlock_ok_after_pull"] = sw.get("interlock_ok") + ctx.check(sw.get("interlock_ok") is False, + "the interlock still reads closed - the loop was not opened (switches: %s)", sw) + # The head is now on its way back. The lid goes up during that park: + # nothing may stop it. + ctx.instruct("The job is cancelling and the head is returning. Open the LID now as well, then " + "click Done - leave both open until the head has stopped.") + drift = expect_cancel_and_return(ctx, g, ev, start, k0, "interlock open", "interlock") + sw = (ctx.forgectrl.status().get("switches") or {}) + ev["switches_at_return"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")} + ctx.log("at the end of the park: %s", ev["switches_at_return"]) + ctx.check(sw.get("lid") is False, + "the lid was not open at the end of the park - the park's immunity was not exercised") + ctx.instruct("Close the lid and restore the interlock loop (plug/jumper back in), then click Done.") + ctx.sleep(1) + sw = (ctx.forgectrl.status().get("switches") or {}) + ev["restored"] = {"lid": sw.get("lid"), "interlock_ok": sw.get("interlock_ok")} + ctx.check(sw.get("interlock_ok"), "the interlock loop is still open - restore it before continuing") + r = g.command("$J=G91X5F1200") + ctx.check(not any(x.startswith("error") for x in r), "jog refused after the cancel: %s", r) + wait_idle(ctx, g, 15) + g.command("$J=G91X-5F1200") + wait_idle(ctx, g, 15) + g.command("G90") + machine_idle(ctx) + ctx.log("PASS: interlock open cancelled the job and the head returned to the start (drift %.3f mm) " + "with the lid opened mid-park", drift) + + +@test("motion.lid-policy-hold", title="lid_policy=hold parks the job in Door and a cycle start resumes it", + subsystem="motion", kind="operator", est_min=4, + covers=_LID_COVERS + [("forgectrl", "src/settings.*")], requires=["motion.lid-cancel-home"], + steps=["Bed clear; the head needs 40 mm of free +X travel. No laser is involved.", + "Open the lid when told, then close it when told; the job finishes after that."], + description="The other lid policy, kept for senders that expect stock grblHAL: with " + "lid_policy=hold a lid open parks the job in the door state and holds it there - " + "no cancel, no return home - and once the lid is closed a cycle start finishes the " + "move with its position intact. The setting is restored to cancel at the end.") +def lid_policy_hold(ctx): + ev = ctx.evidence + fc = ctx.forgectrl + was = (fc.settings() or {}).get("lid_policy") or "cancel" + ev["lid_policy_before"] = was + st, _b = fc.post("/settings", data={"lid_policy": "hold"}) + ctx.check(st == 200, "could not set lid_policy=hold (%s)", st) + ctx.check(((fc.settings() or {}).get("lid_policy")) == "hold", "lid_policy did not take") + try: + with ctx.grbl() as g: + clean_slate(ctx, g) + start = g.status_report()["MPos"] + g.command("M5") + g.command("G91") + g.command("G1X40F300", timeout=0.5) # an 8 s move + ctx.sleep(0.5) + ctx.check(g.status_report()["state"].startswith("Run"), "the move did not start") + ctx.instruct("The head is moving. Open the lid NOW and leave it open, then click Done.") + st = wait_state(ctx, g, "Door", 8) + ev["door_state"] = st["state"] if st else g.status_report()["state"] + ctx.check(st is not None, "the lid did not park the job in Door (state %s)", ev["door_state"]) + text = drain_text(g, 1.0) + ev["messages"] = [ln for ln in text.splitlines() if ln.startswith("[MSG:")] + ctx.check("job cancelled" not in text, "the job was cancelled under lid_policy=hold: %s", ev["messages"]) + ctx.check("returned to the job start" not in text, + "the head returned to the job start under lid_policy=hold") + held = g.status_report()["MPos"] + ev["parked_at"] = held + ctx.log("parked in %s at %s", ev["door_state"], held) + ctx.instruct("Close the lid, then click Done.") + ctx.sleep(1) + ev["state_after_close"] = g.status_report()["state"] + ctx.log("after the lid closed: %s (a cycle start is needed)", ev["state_after_close"]) + g.realtime(0x7E) # ~ cycle start + peak, states, st = wait_idle(ctx, g, 30) + ev["states_after_resume"] = states + ctx.check("TIMEOUT" not in states, "the job did not finish after the resume: %s", states) + final = st["MPos"] + moved = final[0] - start[0] + ev["moved_mm"] = round(moved, 3) + ctx.log("finished: moved %.3f mm of 40 (states %s)", moved, states) + ctx.check(abs(moved - 40.0) <= 0.05, + "the resumed job did not finish its move (%.3f mm of 40)", moved) + g.command("$J=G91X-40F1200") + wait_idle(ctx, g, 30) + g.command("G90") + machine_idle(ctx) + finally: + st, _b = fc.post("/settings", data={"lid_policy": was}) + ev["lid_policy_restored"] = (fc.settings() or {}).get("lid_policy") + ctx.log("lid_policy restored to %s", ev["lid_policy_restored"]) + ctx.check(ev["lid_policy_restored"] == was, "lid_policy was not restored to %r", was) + ctx.log("PASS: lid_policy=hold parked the job in Door and the cycle start finished it (%.3f mm)", + ev["moved_mm"]) diff --git a/forgetest/tests/test_bench_registry.py b/forgetest/tests/test_bench_registry.py index f32ed29..bd7eed3 100644 --- a/forgetest/tests/test_bench_registry.py +++ b/forgetest/tests/test_bench_registry.py @@ -51,6 +51,20 @@ class RegistryTests(unittest.TestCase): self.assertTrue(ok, "%s: %s" % (t["id"], err)) self.assertEqual(argv[1], os.path.join(BENCH, t["script"])) + def test_flagged_args_build_option_pairs(self): + tool = next(t for t in bench_mod.TOOLS if t["id"] == "resume-dark-lead") + ok, argv, err = self.bench.command(tool, {"run": "live", "feed": 900}) + self.assertTrue(ok, err) + self.assertIn("--run", argv) + self.assertEqual(argv[argv.index("--run") + 1], "live") + self.assertEqual(argv[argv.index("--feed") + 1], "900.0") + # an empty optional with an empty default is left off entirely + self.assertNotIn("--auto", argv) + # a choice outside the list is refused before anything runs + ok, _argv, err = self.bench.command(tool, {"run": "sideways"}) + self.assertFalse(ok) + self.assertIn("run must be one of", err) + def test_unported_tools_are_host_only(self): # what stays unported is what cannot run on the machine at all for t in bench_mod.TOOLS: diff --git a/forgetest/tests/test_cloud_suite.py b/forgetest/tests/test_cloud_suite.py index f0e9b6e..5c2e1a4 100644 --- a/forgetest/tests/test_cloud_suite.py +++ b/forgetest/tests/test_cloud_suite.py @@ -369,6 +369,141 @@ class CloudSuiteTests(unittest.TestCase): self.assertEqual(self.fc.posts, []) self.assertTrue(any("PASS: lid open at the button prompt" in l for l in run.lines)) + # -- interlock abort, and the park through an open lid -------------------- + def interlock_parts(self): + """The lid-abort excerpt with the interlock as the trigger, split at + the three operator steps: up to the print's run, the stop through + 'start return home', and the park itself.""" + lines = [l.replace("lid opened mid-run; stopping motion", + "interlock opened mid-run; stopping motion") + for l in fixture("lidabort")] + pre, rest = cut(lines, "waiting for button") + run_pre, rest = cut(rest, "machine:_run_loop starting run") + pre, rest = pre + run_pre + [rest[0]], rest[1:] + stop, tail = cut(rest, "start return home") + return pre, stop + [tail[0]], tail[1:] + + def test_interlock_abort_park_on_the_bench_excerpt(self): + self.in_cloud(pid=1522) + self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"]) + pre, stop, park = self.interlock_parts() + + def pull_interlock(): + self.fc.state["status"]["switches"]["interlock_ok"] = False + self.append(stop, delay=0.05) + + def open_lid(): + self.lid(False) + self.append(park, delay=0.05) + + def restore(): + self.lid(True) + self.fc.state["status"]["switches"]["interlock_ok"] = True + run = self.run_test(cloud.interlock_abort_park, + hooks={"Click Done here": lambda: self.append(pre, delay=0.1), + "Open the INTERLOCK loop now": pull_interlock, + "Open the LID now as well": open_lid, + "Close the lid and restore the interlock": restore}, + test_id="cloud.interlock-abort-park") + ev = run.evidence + self.assertFalse(ev["interlock_ok_after_pull"]) + self.assertIn(":cancelled", ev["log"]["print finished"]) + self.assertEqual(ev["switches_at_return"], {"lid": False, "interlock_ok": False}) + self.assertEqual(ev["kernel_counters_after_park"], [0, 0, 3]) + self.assertTrue(ev["latch_locked"]) + self.assertFalse(ev["armed_after"]) + self.assertEqual(ev["restored"], {"lid": True, "interlock_ok": True}) + self.assertEqual(self.fc.posts, []) + self.assertTrue(any("PASS: interlock open" in l for l in run.lines), run.lines) + + def test_interlock_abort_refuses_when_the_loop_is_already_open(self): + self.in_cloud(pid=1522) + self.fc.state["status"]["switches"]["interlock_ok"] = False + self.assertFails(cloud.interlock_abort_park, "already reads open") + + def test_interlock_abort_fails_when_the_park_stops_at_the_lid(self): + # the regression this guards: a park that the lid can interrupt + self.in_cloud(pid=1522) + self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"]) + pre, stop, park = self.interlock_parts() + park = [l for l in park if "return home complete" not in l] + saved = cloud.wait_log + + def fast_wait_log(ctx, offset, needles, timeout, poll=0.5): + return saved(ctx, offset, needles, min(timeout, 1.5), poll=0.1) + cloud.wait_log = fast_wait_log + try: + self.assertFails( + cloud.interlock_abort_park, "did not run to completion", + hooks={"Click Done here": lambda: self.append(pre, delay=0.1), + "Open the INTERLOCK loop now": lambda: ( + self.fc.state["status"]["switches"].__setitem__("interlock_ok", False), + self.append(stop, delay=0.05)), + "Open the LID now as well": lambda: (self.lid(False), + self.append(park, delay=0.05))}) + finally: + cloud.wait_log = saved + + # -- a paused print cancelled by the lid, a running one by the app -------- + def cancel_parts(self): + """(print prologue, the pause lines, the lid stop + park + cancel, + the same tail with the app's cancel as the trigger).""" + lines = fixture("lidabort") + pre, rest = cut(lines, "waiting for button") + run_pre, rest = cut(rest, "machine:_run_loop starting run") + pre, rest = pre + run_pre + [rest[0]], rest[1:] + paused = ["2026-08-17T09:42:25.500000+00:00 gfcloud[1522] INFO machine:_run_loop " + "button pressed mid-run; pausing", + "2026-08-17T09:42:26.100000+00:00 gfcloud[1522] INFO machine:_run_loop " + "paused at Position(x=41.2, y=17.0, z=0.0)"] + app_cancel = [l.replace("lid opened mid-run; stopping motion", + "action cancelled mid-run; stopping motion") for l in rest] + return pre, paused, rest, app_cancel + + def test_pause_cancel_paths_on_the_bench_excerpt(self): + self.in_cloud(pid=1522) + self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"]) + pre, paused, lid_tail, app_tail = self.cancel_parts() + prints = [] + + def next_print(): + prints.append(1) + self.append(pre, delay=0.1) + run = self.run_test(cloud.pause_cancel_paths, + hooks={"Click Done here": next_print, + "Press the button once NOW": lambda: self.append(paused, delay=0.05), + "Open the lid NOW": lambda: (self.lid(False), + self.append(lid_tail, delay=0.05)), + "Close the lid, then click Done": lambda: self.lid(True), + "Cancel the print from the app now": lambda: self.append(app_tail, + delay=0.05)}, + test_id="cloud.pause-cancel-paths") + ev = run.evidence + self.assertEqual(len(prints), 2) # two prints, one cue each + self.assertEqual(ev["paused"], {"button pressed mid-run; pausing": True, "paused at": True}) + self.assertIn(":cancelled", ev["lid_from_pause"]["print finished"]) + self.assertIn(":cancelled", ev["service_cancel"]["print finished"]) + self.assertTrue(ev["lid_from_pause"]["return home complete"]) + self.assertTrue(ev["service_cancel"]["return home complete"]) + self.assertEqual(ev["counters_after_print1"], [0, 0, 3]) + self.assertEqual(ev["counters_after_print2"], [0, 0, 3]) + self.assertTrue(ev["latch_locked_after_print1"] and ev["latch_locked_after_print2"]) + self.assertFalse(ev["armed_after_print1"] or ev["armed_after_print2"]) + self.assertEqual(self.fc.posts, []) + self.assertTrue(any("PASS: a paused print cancelled by the lid" in l for l in run.lines), run.lines) + + def test_pause_cancel_paths_fails_when_the_paused_print_resumes_instead(self): + # a lid that resumed (or was ignored) leaves the print ':completed' + self.in_cloud(pid=1522) + self.append(["2026-08-17T09:41:00.500000+00:00 gfcloud[1522] INFO websocket:_on_open RX-EVENT: ready"]) + pre, paused, lid_tail, _app = self.cancel_parts() + lid_tail = [l.replace(':cancelled"', ':completed"') for l in lid_tail] + self.assertFails( + cloud.pause_cancel_paths, "print 1 did not end ':cancelled'", + hooks={"Click Done here": lambda: self.append(pre, delay=0.1), + "Press the button once NOW": lambda: self.append(paused, delay=0.05), + "Open the lid NOW": lambda: (self.lid(False), self.append(lid_tail, delay=0.05))}) + def test_print_finish_is_the_prints_not_another_actions(self): # a motion that completes before the print must not satisfy the print's finish lines = fixture("lidabort") diff --git a/scripts/bench/README.md b/scripts/bench/README.md index 40c7187..ab3aa5c 100644 --- a/scripts/bench/README.md +++ b/scripts/bench/README.md @@ -52,6 +52,7 @@ page's takeover does that; from a host, stop them first. | `build-feeder.sh` | Cross-compiles `feeder.c` the same way. | | `puls_profile.py` | Decodes factory `.puls` streams (raw or GF1-headered) into velocity/accel profiles: peak speeds, ramp-slope fits, per-move segments, Z cadence. Runs anywhere (stdlib only). Source of the factory-true grblHAL defaults: 700/590 mm/s² accel, 200 mm/s max rate, 28160 Hz travel tick. | | `cp_watchdog_timing.py` | HV charge-pump watchdog one-shot timing (runs on the board): latches every CHG_PUMP feed pulse in GPIO3's edge detector (pin 24 only, IMR untouched, ICR2 restored on exit) and polls the `!Q` (`charge_pump_alive`) and `!HV_ENABLE` (`hv_enable`) pads through /dev/mem in a tight loop while it commands short local jogs; prints per-run t_w (last pulse → Q fall), Q → HV_ENABLE delay, priming latency and the feed period, with the loop's worst gap as the resolution. Motion only, laser locked, no other Grbl client attached. | +| `resume_dark_lead.py` | Pause/resume safety-chain timing (runs on the board, as root): samples LASER_ON, FIRE, HV_ENABLE, the charge-pump watchdog, the button and the doors straight off the SoC pads through /dev/mem at ~2 kHz, with motion dated from the kernel step counters, across a pause and a resume driven by the operator's button presses. Reports how long HV_ENABLE survives the stream stopping, how fast the chain re-arms on the resume, and - on `--run live` - the dark lead between FIRE going back on and LASER_ON following it, in milliseconds and in millimeters at the job's feed. `--run dry` (default) commands no laser at all and `--auto P,R` drives the pause and resume with `!`/`~` for an unattended rehearsal; `--run live` needs the arm press, eye protection, fire watch, extinguisher and exhaust. GRBL mode, no other Grbl client attached. | | `bench_m2.py` | Motion-quality bench, runs against the board over TCP:23: bounded round-trip jogs (sanity, max-rate, diagonal) + feed-hold/resume mid-move, reporting peak feed, state transitions, and position drift. | Data files kept beside the tools: `flow_matrix_results.json` / diff --git a/scripts/bench/resume_dark_lead.py b/scripts/bench/resume_dark_lead.py new file mode 100644 index 0000000..5203764 --- /dev/null +++ b/scripts/bench/resume_dark_lead.py @@ -0,0 +1,437 @@ +#!/usr/bin/env python3 +"""Resume dark-lead characterization (runs ON the board, as root). + +Measures what the safety chain does across a pause and a resume, at pad +resolution, so the GRBL resume dwell can be decided from numbers instead +of from the mark alone. The pause and the resume are the operator's +physical button presses - the same toggle the controller ships. + +Sampled straight from the SoC pads (no kernel change, no scope), one +32-bit read per bank per pass: + + LASER_ON GPIO1_05 gated output of the safety AND gate (active low) + CP_ALIVE GPIO1_08 charge-pump watchdog !Q (0 = alive) + BUTTON_LATCH GPIO1_03 1 = latch SET (fire blocked) + DOORS GPIO1_00 0 = both lid switches closed + FIRE GPIO2_30 laser-enable line, read from the data register + (the SDMA writes it; GDIR says whether it is driven) + BUTTON GPIO4_09 big button (0 = pressed) + HV_ENABLE GPIO4_06 chain output readback, inverted (0 = asserted) + +Levels are read once at idle and every later sample is reported as a +change from that baseline, so no polarity assumption is baked in. + +The headline number is the **dark lead**: FIRE re-asserted (the stream is +commanding emission again) -> LASER_ON asserted (the chain actually lets +the beam through). At the job's feed rate that is also a distance, which +is what shows up in the mark. The pause side reports how long HV_ENABLE +survives the stream stopping, and the charge-pump one-shot period behind it. + +The loop must not hog the CPU: this is a single-core part, only the +shipper thread is SCHED_FIFO, and starving the SCHED_OTHER protocol thread +mid-job would underrun the ring. It samples at ~2 kHz with a positive nice +and reports the worst gap it actually achieved. + +Usage: resume_dark_lead.py [--run dry|live] [--power S] [--feed F] + [--len MM] [--passes N] [--mode m3|m4] + [--secs N] [--auto P,R] [--json FILE] + + --run dry (default) a plain G1 travel of --len at --feed, no laser + command at all. Exercises the button pause/resume and gives + the HV_ENABLE / charge-pump timings, which do not need fire - + the pump runs for any pulse-engine run. Run this first. + --run live LIVE FIRE: M3 (or M4) at --power. Requires the arm press, eye + protection, exhaust, fire watch, extinguisher, scrap under the + head. Adds the LASER_ON edge - the dark lead itself - and the + physical mark to measure. + +Operator sequence in both cases: press to arm (live only), then press once +mid-move to pause, and once more to resume. GRBL mode only, with the +controller idle and no other Grbl client attached (a connection here +displaces the sender). +""" +import argparse +import json +import mmap +import os +import socket +import struct +import sys +import time + +GPIO1, GPIO2, GPIO4 = 0x0209C000, 0x020A0000, 0x020A8000 +DR, GDIR, PSR = 0x00, 0x04, 0x08 + +# name -> (bank, register, bit). Order is the report order. +SIGNALS = [ + ('FIRE', 'g2', DR, 30), + ('LASER_ON', 'g1', PSR, 5), + ('HV_ENABLE', 'g4', PSR, 6), + ('CP_ALIVE', 'g1', PSR, 8), + ('BUTTON', 'g4', PSR, 9), + ('BUTTON_LATCH', 'g1', PSR, 3), + ('DOORS', 'g1', PSR, 0), +] + +# Motion comes from the kernel's step counters, not from the step lines: +# a step pulse is microseconds wide, so a pad sampler at this rate catches +# only the occasional one. cnc/position is 32 binary bytes (X, Y, Z steps +# first) and re-reading it at MOTION_HZ dates the restart closely enough to +# compare against a chain that re-arms in single-digit milliseconds. +POSITION = '/sys/glowforge/cnc/position' +MOTION_HZ = 50.0 +MOTION_GAP_S = 0.10 + +SLEEP_S = 0.0005 +HOST, PORT = '127.0.0.1', 23 + + +class Pads: + """One mmap per bank; a pass reads three words.""" + + def __init__(self): + self.fd = os.open('/dev/mem', os.O_RDWR | os.O_SYNC) + self.m = { + 'g1': mmap.mmap(self.fd, 4096, mmap.MAP_SHARED, + mmap.PROT_READ, offset=GPIO1), + 'g2': mmap.mmap(self.fd, 4096, mmap.MAP_SHARED, + mmap.PROT_READ, offset=GPIO2), + 'g4': mmap.mmap(self.fd, 4096, mmap.MAP_SHARED, + mmap.PROT_READ, offset=GPIO4), + } + + # (bank, register) -> mask of the bits actually watched there. The step + # lines and the PWM share these banks, so an unmasked word compare would + # log a transition on every step. + SOURCES = [] + for _n, _b, _r, _bit in SIGNALS: + for _i, (_sb, _sr, _sm) in enumerate(SOURCES): + if (_sb, _sr) == (_b, _r): + SOURCES[_i] = (_sb, _sr, _sm | (1 << _bit)) + break + else: + SOURCES.append((_b, _r, 1 << _bit)) + + def words(self): + m = self.m + return tuple(struct.unpack_from('> bit) & 1 for n, b, r, bit in SIGNALS} + + def fire_is_driven(self): + gdir = struct.unpack('> 30) & 1) + + def close(self): + for m in self.m.values(): + m.close() + os.close(self.fd) + + +class Grbl: + def __init__(self): + self.s = socket.create_connection((HOST, PORT), timeout=5) + self.s.settimeout(0.2) + time.sleep(0.5) + self.drain() + + def drain(self): + out = b'' + try: + while True: + d = self.s.recv(4096) + if not d: + break + out += d + except socket.timeout: + pass + return out.decode('ascii', 'replace') + + def send(self, line): + self.s.sendall(line.encode() + b'\n') + + def status(self): + self.s.sendall(b'?') + deadline = time.time() + 1.0 + text = '' + while time.time() < deadline: + text += self.drain() + if '>' in text: + break + time.sleep(0.02) + return text[text.rfind('<'):text.rfind('>') + 1] if '<' in text else '' + + def state(self): + st = self.status() + return st[1:].split('|')[0] if st else '' + + +class Counters: + """X/Y/Z step counters, re-read from the same open descriptor.""" + + def __init__(self): + self.f = open(POSITION, 'rb', buffering=0) + + def read(self): + self.f.seek(0) + return struct.unpack_from('= end: + break + gap = now - last + if gap > worst: + worst = gap + last = now + w = pads.words() + if w != prev: + st = pads.decode(w) + changed = {k: v for k, v in st.items() if v != prev_state[k]} + if changed: + events.append((now - t0, changed, st)) + prev, prev_state = w, st + if now >= next_pos_t: + next_pos_t = now + 1.0 / MOTION_HZ + pos = counters.read() + if pos != prev_pos: + prev_pos = pos + if last_move is None or now - last_move > MOTION_GAP_S: + motion.append([now - t0, now - t0]) + else: + motion[-1][1] = now - t0 + last_move = now + if not sent and now - t0 > 1.0: + for ln in job_lines: + grbl.send(ln) + marks.append(('job sent', now - t0)) + sent = True + while auto and now - t0 >= auto[0][0]: + _at, ch, label = auto.pop(0) + grbl.s.sendall(ch) + marks.append((label, now - t0)) + time.sleep(SLEEP_S) + counters.close() + return base, events, worst, motion + + +def edges(events, name, base, to_active): + """Times at which `name` moved to (to_active=True) or away from its + non-baseline level.""" + out = [] + for t, changed, _st in events: + if name in changed: + active = changed[name] != base[name] + if active == to_active: + out.append(t) + return out + + +def report(base, events, worst, args, marks, motion): + print('\nbaseline at idle: %s' % ' '.join( + '%s=%d' % (n, base[n]) for n, _b, _r, _bit in SIGNALS)) + print('worst sampling gap: %.2f ms (%d transitions)' + % (worst * 1000.0, len(events))) + for label, t in marks: + print(' %-10s t=%.3f s' % (label, t)) + + print('\n--- transitions (t in s from sampler start) ---') + for t, changed, _st in events: + desc = ' '.join('%s->%s' % (k, 'ACTIVE' if v != base[k] else 'idle') + for k, v in sorted(changed.items())) + print(' %8.4f %s' % (t, desc)) + + fire_on = edges(events, 'FIRE', base, True) + fire_off = edges(events, 'FIRE', base, False) + lon_on = edges(events, 'LASER_ON', base, True) + lon_off = edges(events, 'LASER_ON', base, False) + hv_on = edges(events, 'HV_ENABLE', base, True) + hv_off = edges(events, 'HV_ENABLE', base, False) + cp_on = edges(events, 'CP_ALIVE', base, True) + cp_off = edges(events, 'CP_ALIVE', base, False) + btn = edges(events, 'BUTTON', base, True) + + print('\n--- motion (step-line activity) ---') + for a, b in motion: + print(' %8.4f -> %8.4f (%.0f ms)' % (a, b, (b - a) * 1000.0)) + + print('\n--- summary ---') + print('button presses: %s' % (', '.join('%.3f' % t for t in btn) or 'none')) + + def after(times, t): + later = [x for x in times if x > t] + return later[0] if later else None + + mm_s = args.feed / 60.0 + starts = [a for a, _b in motion] + stops = [b for _a, b in motion] + + # Each pause and resume is dated from what triggered it - the operator's + # button press, or the ! / ~ the rehearsal sends in its place - so the + # dry and the live run report the same way. + triggers = [(lbl, t) for lbl, t in marks + if lbl.startswith(('pause', 'resume'))] + if btn: + triggers = [('press %d' % (i + 1), t) for i, t in enumerate(btn)] + triggers.sort(key=lambda x: x[1]) + + for lbl, t in triggers: + line = ['%-8s at %7.3f:' % (lbl, t)] + for nm, times in (('motion stops', stops), ('motion starts', starts), + ('FIRE clear', fire_off), ('FIRE set', fire_on), + ('LASER_ON off', lon_off), ('LASER_ON on', lon_on), + ('HV drop', hv_off), ('HV up', hv_on), + ('CP fell', cp_off), ('CP alive', cp_on)): + x = after(times, t) + if x is not None and x - t < 2.0: + line.append('%s +%.0f ms' % (nm, (x - t) * 1000.0)) + print(' ' + ' | '.join(line)) + + # The dark lead is what a resumed cut loses: emission commanded again + # (FIRE) but the chain not yet letting the beam through (LASER_ON). + leads = [] + for t in fire_on: + x = after(lon_on, t) + if x is not None and x - t < 2.0: + leads.append((x - t) * 1000.0) + if leads: + print('\nDARK LEAD (FIRE set -> LASER_ON): %s ms -> %s mm at F%g' + % (', '.join('%.1f' % x for x in leads), + ', '.join('%.2f' % (x / 1000.0 * mm_s) for x in leads), + args.feed)) + elif args.run == 'dry': + print('\n(dry run: FIRE and LASER_ON never assert - the chain ' + 'numbers above are the ones this rehearsal establishes)') + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument('--run', choices=('dry', 'live'), default='dry', + dest='run', + help='dry: travel only, no laser commanded (default). ' + 'live: command the laser - LIVE FIRE.') + ap.add_argument('--mode', choices=('m3', 'm4'), default='m3') + ap.add_argument('--power', type=int, default=400) + ap.add_argument('--feed', type=float, default=1200.0) + ap.add_argument('--len', type=float, default=80.0, dest='length') + ap.add_argument('--passes', type=int, default=1, + help='alternating +X/-X moves (dry: gives a long window ' + 'to press in; live: keep 1 so the mark is one line)') + ap.add_argument('--secs', type=float, default=45.0) + ap.add_argument('--auto', default='', + help='unattended rehearsal (dry only): PAUSE,RESUME in ' + 'seconds from the job start, e.g. --auto 3,6 - sends ' + '! and ~ instead of waiting for button presses') + ap.add_argument('--json', default='') + args = ap.parse_args() + + if os.geteuid() != 0: + sys.exit('must run as root (needs /dev/mem)') + + grbl = Grbl() + st = grbl.state() + if not st.startswith('Idle'): + sys.exit('controller is not Idle (%s) - clear it first' % (st or '?')) + print('controller: %s interlock_circuit=%s faults=%s button_latch=%s' + % (st, sysfs('interlock_circuit'), sysfs('faults'), + sysfs('button_latch'))) + + moves = ['G1 X%g F%g' % (args.length * (1 if i % 2 == 0 else -1), args.feed) + for i in range(max(1, args.passes))] + lines = ['G21', 'G91'] + if args.run == 'live': + lines += ['%s S%d' % (args.mode.upper(), args.power)] + moves \ + + ['M5', 'G90'] + print('\n>>> LIVE FIRE. Eye protection, exhaust running, fire watch,') + print('>>> extinguisher in reach, scrap under the head with %g mm' + % args.length) + print('>>> of clear travel in +X.') + print('>>> 1) press the button to ARM (the stream waits for it)') + print('>>> 2) press once mid-cut to PAUSE') + print('>>> 3) press once more to RESUME') + else: + lines += moves + ['G90'] + print('\n>>> DRY rehearsal - no laser is commanded.') + if not args.auto: + print('>>> 1) press the button mid-move to PAUSE') + print('>>> 2) press once more to RESUME') + print('>>> sampling for %g s from now.\n' % args.secs) + + try: + os.nice(5) + except OSError: + pass + + auto = [] + if args.auto: + if args.run == 'live': + sys.exit('--auto is a dry rehearsal aid; a live run uses the button') + t_p, t_r = (float(x) for x in args.auto.split(',')) + auto = [(1.0 + t_p, b'!', 'pause (!)'), (1.0 + t_r, b'~', 'resume (~)')] + print('unattended: ! at +%g s and ~ at +%g s after the job starts' + % (t_p, t_r)) + + pads = Pads() + print('FIRE line is %s at idle' + % ('driven' if pads.fire_is_driven() else 'high impedance')) + marks = [] + try: + base, events, worst, motion = sample(pads, args.secs, grbl, lines, + marks, auto) + finally: + pads.close() + + report(base, events, worst, args, marks, motion) + print('\nfinal: state=%s kernel=%s interlock_circuit=%s faults=%s' + % (grbl.state(), sysfs('state'), sysfs('interlock_circuit'), + sysfs('faults'))) + + if args.json: + with open(args.json, 'w') as f: + json.dump({'baseline': base, 'worst_gap_s': worst, + 'live': args.run == 'live', 'mode': args.mode, + 'power': args.power, 'feed': args.feed, + 'marks': marks, 'motion': motion, + 'events': [(t, c) for t, c, _s in events]}, f, indent=1) + print('wrote %s' % args.json) + + +if __name__ == '__main__': + main()