vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-10-15 17:19:55 -07:00
Code Issues Packages Projects Releases Wiki Activity
Files
6594f57362fc3676ab7e4333d222d6f41db64312
forgefirm/scripts/bench/laser_lifecycle_test.py
T
ScottW514 6594f57362 Catalog: motion.port-jog and motion.tray leave; two leftovers of the audit 
Two catalog tests leave for the host proofs that carry them; the catalog
goes from 80 to 78 (78 @test in forgetest/forgetest/suite/).

- motion.port-jog. forgectrl's tests/host_routes.sh (7b2f629 on e69d967)
  runs the controller port's routes against the port's stand-in: a jog
  refused without the token, a jog that is no number 400 with nothing on
  the port, a numeric jog on the port as G91 G21, cancel, release and
  energize each one line, GET /motion/state passing the port's state on
  with state and sender, GET /motion/sender 200. The port beside a
  sender, its polls, the cancel, the client first and the bound are
  grblHAL's ctlport_test.py and forgectrl's grblport_test; motion.release
  keeps the physical currents. The helpers before it (_port_state,
  _port_idle, _words) stay: bedsize, laser and motion.job read them.
- motion.tray. host_routes.sh holds POST /motion/tray (tray=in on the
  port, tray=up 400) and the panel page's switch (id="traybtn",
  trayToggle, /motion/tray); grblHAL's tray_test.py the mode, its offset,
  its persistence and its frame; forgectrl's tray_test the daemon's side.
  suite/tray.py held only it: the module and its import go.

The coverage lint then named forgectrl src/tray.c and src/tray.h, and
nothing else; grblhal-glowforge's src/glowforge_tray.* stays covered by
another catalog test, so it gets no row. One HOST_TESTED row takes
forgectrl's, to forgectrl:tests/tray_test.c on
forgectrl:.github/workflows/build.yml, which the pin c274f19 runs
(./build/tray_test).

Two of the audit's leftovers, in the same change:
- test_manifest.py's ImplementationHashTests sat after the __main__
  guard, so running the file directly skipped it. It is above the guard
  now, byte for byte: python3 -m unittest test_manifest.py and python3
  test_manifest.py (from tests/, where its helpers import resolves) each
  run its 7 cases, and 34 in all; before, the direct run ran none of
  them.
- laser_lifecycle_test.py's test_lid_open_in_wait said an alarm is
  raised; the docstring now says what the test asserts: the lid-open
  cancel is reported and no alarm is raised. The assertions are
  unchanged and the harness was not run.

The coverage lint on the tree manifest finds no uncovered path and no
host-tested row problem across the 78 tests; its one finding is
setup.check-lens's empty src/lensfind.* entry. pyflakes reports nothing
on motion.py, manifest.py, test_manifest.py or laser_lifecycle_test.py.
The full forgetest suite was not run.

Fingerprints, measured against the parent: 17 tests move with
motion.py's shared text: the 11 left in motion.py, and homing.manual,
laser's four and setup.check-envelope, whose modules read it. The
pending pin bumps from forgectrl c274f19 and forgeext 2d0391e already
owe all 17; this deletion adds none to the owed list. The two leftovers
move no fingerprint. releases/ is unchanged.
2026-10-14 18:21:59 -04:00

963 lines
43 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# Copyright 2026 514 LLC d/b/a OpenGlow
# Written by Scott Wiederhold
# https://community.openglow.org
# SPDX-License-Identifier: MIT
"""Host-side verification of the operator-armed window lifecycle.
Runs the native grblHAL_glowforge binary in null-sink mode (no
hardware, no root; arming is automatic without the physical button)
and drives jobs over TCP, asserting the window's state machine on the
reported messages:
1. the first laser-on of a job arms exactly once; the window's
persistence across S changes and M5/M3 toggles inside the job is
laser_arm_test.c case I's
2. program end (M2) closing the window, and the next job arming
afresh: laser_stream_test session K
3. a sender change closes the window: a reconnected client must
re-arm - the displaced session's consent does not carry over
4. the spindle-off grace counts down in HOLD, not only in Idle - a
job abandoned in feed hold disarms after laser_disarm_s
5. arming refused while the cooling verdict blocks fire, or before
the engine acknowledges the session: laser_arm_test.c cases C, A2
and A3, and cooling_test.c's arm-time gate; the wait for the fans
is laser_arm_test.c case A4
6. with a switch source present (GF_SWITCH_FILE), the arm blocks on the
button: a press with the lid closed arms, and the lid or the
interlock loop opening during the wait cancels the job (a clean
soft reset, no alarm; latch relocked, never armed)
7. outside the arm wait the button is the pause/resume toggle: a press
while running feed-holds, a press while held resumes; the arming
press itself is not a pause press; a travel job (no laser, no arm
wait) takes the same toggle, and its resumed move lands on its target
8. the lid or the interlock loop opening mid-job cancels the job (the
default lid_policy): reason reported, armed window closed, a soft
reset with the position kept (no alarm), the head returns to the
job start on its own; with lid_policy = hold the stock door hold and
cycle-start resume apply
9. the controller publishes its state for the daemon: grbl.settings
(the $$ view, the derived floor in force from boot) and grbl.state
(JSON with ts_mono, machine state, sender session, laser window
and dose model, modals) appear under GF_STATE_DIR, follow the
sender connection and the armed window, and carry a fresh
ts_mono
10. the job start survives a pause: a job paused and resumed by the
button, then canceled by the lid, returns to where the job began,
not to where it was paused (the core restarts a held cycle through
Idle); a job abandoned in a hold and reset ends there, so the next
job's start is captured afresh where it begins
11. the cooling verdict's pause tier holding the job under the open
window without a latch write, and the clean verdict resuming it with
no press and no second arm: laser_stream_test rule 24 and
laser_arm_test.c case V1
12. the verdict's fail tier ending the job (disarmed, ALARM:3, nothing
resumed): laser_stream_test rule 25 and laser_arm_test.c case V2
13. a sender change during a re-arm cancels it: the press that follows
resumes nothing, the job stays held, and the new sender's own ~
prompts afresh
14. a jog not holding the armed window open: laser_arm_test.c case S10
15. a press counts only after the button has been seen up: a press that
began before the wait is not consent (the harness's own presses
follow a fresh prompt for the same reason)
The disarm grace is shortened via a temp config (GFHOME_CONF), the
cooling verdict is published hermetically (GF_VERDICT_FILE), the same
overrides the emission harness uses, and the switches are driven through
the file-backed test source (GF_SWITCH_FILE: the EV_SW word as an integer).
Usage: laser_lifecycle_test.py [path-to-binary]
(default ./build-native/grblHAL_glowforge)
"""
import os
import re
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
BIN = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "build-native/grblHAL_glowforge")
PORT = 2398
ARMED = "laser armed"
DISARMED = "laser disarmed - latch locked"
BLOCKED = "laser fire blocked"
PROMPT = "press the button to start the laser job"
RESUME_PROMPT = "press the button to resume the laser job"
LID_CANCEL = "lid opened during arm - job canceled"
LOOP_CANCEL = "interlock open during arm - job canceled"
# EV_SW words for the file-backed switch source: bit 2 button, bit 3 doors
# (set = closed), bit 5 interlock loop (set = OPEN).
SW_CLOSED = 1 << 3
SW_PRESSED = SW_CLOSED | (1 << 2)
SW_LID_OPEN = 0
SW_LOOP_OPEN = SW_CLOSED | (1 << 5)
def fail(msg):
print("FAIL: %s" % msg)
sys.exit(1)
def read_avail(sock, log, timeout, until=None):
end = time.time() + timeout
buf = b""
while time.time() < end:
sock.settimeout(max(0.05, end - time.time()))
try:
data = sock.recv(4096)
except socket.timeout:
data = b""
except OSError:
return None
if data:
buf += data
log.append(data.decode(errors="replace"))
if until:
text = buf.decode(errors="replace")
for token in until:
if token in text or re.search(r"^%s\b" % re.escape(token), text, re.M):
return token
elif until is None:
return None
return None
def send_line(sock, line, log, tolerate_error=False):
sock.sendall((line + "\n").encode())
r = read_avail(sock, log, 5.0, until=("ok", "error"))
if r is None:
fail("no ok/error for %r" % line)
if r == "error" and not tolerate_error:
fail("error response to %r" % line)
return r
def wait_for(log, needle, timeout, sock=None):
"""Wait until needle appears in the accumulated log (draining the
socket while waiting when one is given)."""
end = time.time() + timeout
while time.time() < end:
if needle in "".join(log):
return True
if sock is not None:
read_avail(sock, log, 0.2)
else:
time.sleep(0.1)
return False
def wait_idle(sock, log):
for _ in range(100):
sock.sendall(b"?")
read_avail(sock, log, 0.3)
if re.search(r"<Idle", "".join(log[-3:])):
return
time.sleep(0.2)
fail("controller never returned to Idle")
def publish_verdicts(path, stop, verdict):
"""Stand in for the cooling engine. `verdict` is the session's live
dict: fire_ok, hold, resume_ok, name, armed_ack, ack_after_s. "armed"
is the engine's acknowledgment that it has taken the controller's
armed window and applied the run airflow; the controller refuses to
fire on a verdict that lacks it, so an engine that never
acknowledges (armed_ack False) must produce a refused arm and no
emission. ack_after_s withholds the acknowledgment for that long
first, which is the real engine's case: it answers on its next
tick, and the controller has to see the refreshed verdict to get
past the arm."""
t0 = time.monotonic()
while not stop.is_set():
v = dict(verdict)
acked = v["armed_ack"] and time.monotonic() - t0 >= v["ack_after_s"]
body = ('{"ts_mono":%.3f,"fire_ok":%s,"verdict":"%s","hold":%s,'
'"resume_ok":%s,"armed":%s,"reason":"%s"}'
% (time.clock_gettime(time.CLOCK_MONOTONIC),
"true" if v["fire_ok"] else "false", v["name"],
"true" if v["hold"] else "false",
"true" if v["resume_ok"] else "false",
"true" if acked else "false",
"" if v["fire_ok"] else "harness: %s" % v["name"]))
tmp = path + ".tmp"
with open(tmp, "w") as f:
f.write(body)
os.replace(tmp, path)
stop.wait(0.5)
def wait_for_new(log, needle, n_before, timeout, sock=None):
"""Wait until needle appears in the log more times than n_before: a
fresh occurrence, never a stale match from an earlier job."""
end = time.time() + timeout
while time.time() < end:
if "".join(log).count(needle) > n_before:
return True
if sock is not None:
read_avail(sock, log, 0.2)
else:
time.sleep(0.1)
return False
class Session:
"""One controller process with the lifecycle overrides applied."""
def __init__(self, name, fire_ok=True, disarm_s=2, switches=None, conf_extra="",
armed_ack=True, ack_after_s=0.0):
self.name = name
self.workdir = tempfile.mkdtemp(prefix="laser-lifecycle-")
conf = os.path.join(self.workdir, "forgefirm.conf")
with open(conf, "w") as f:
# The fan grace at 0 keeps the arm's acknowledgment budget at its
# 5 s floor, so the refusal cases stay short; a case that wants
# the real budget sets cool_fan_grace_s in conf_extra.
f.write("laser_disarm_s = %d\ncool_fan_grace_s = 0\n%s" % (disarm_s, conf_extra))
verdict = os.path.join(self.workdir, "cooling.state")
env = dict(os.environ, GF_VERDICT_FILE=verdict, GFHOME_CONF=conf,
GF_STATE_DIR=self.workdir, FFLOG_STDERR="1")
env.pop("GFSINK", None)
env.pop("GF_SWITCH_FILE", None)
self.switch_file = None
if switches is not None:
self.switch_file = os.path.join(self.workdir, "switches")
self.set_switches(switches)
env["GF_SWITCH_FILE"] = self.switch_file
self.stop = threading.Event()
self.verdict = {"fire_ok": fire_ok, "hold": False, "resume_ok": True, "name": "OK",
"armed_ack": armed_ack, "ack_after_s": ack_after_s}
self.pub = threading.Thread(target=publish_verdicts,
args=(verdict, self.stop, self.verdict),
daemon=True)
self.pub.start()
self.proc = subprocess.Popen([BIN, "-p", str(PORT)],
cwd=self.workdir, env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE)
self.log = []
self.sock = self.connect()
def connect(self):
for _ in range(50):
try:
s = socket.create_connection(("127.0.0.1", PORT), timeout=1)
read_avail(s, self.log, 0.5) # banner
return s
except OSError:
time.sleep(0.1)
err = b""
if self.proc.poll() is not None:
err = self.proc.stderr.read() or b""
fail("[%s] cannot connect to the controller (exit=%s)\n%s"
% (self.name, self.proc.poll(), err.decode(errors="replace")))
def set_switches(self, word):
tmp = self.switch_file + ".tmp"
with open(tmp, "w") as f:
f.write("%d\n" % word)
os.replace(tmp, self.switch_file)
def set_verdict(self, name="OK", fire_ok=True, hold=False, resume_ok=True):
"""The engine's next verdicts: a pause tier (OVERTEMP: fire
blocked, hold, no resume), a fail tier (AIRFLOW: the same under a
name the client ends the job on), or clean."""
self.verdict.update({"name": name, "fire_ok": fire_ok, "hold": hold,
"resume_ok": resume_ok})
def count(self, needle):
return "".join(self.log).count(needle)
def send_raw(self, line):
"""Send a line without waiting for ok/error (the arm wait blocks
the gcode stream, so the ok only comes once the button is pressed)."""
self.sock.sendall((line + "\n").encode())
def press_button(self, hold_s=0.15):
"""A momentary press: word with the button bit, then without."""
base = SW_CLOSED if self.switch_word_closed() else SW_LID_OPEN
self.set_switches(base | (1 << 2))
time.sleep(hold_s)
self.set_switches(base)
def switch_word_closed(self):
try:
with open(self.switch_file) as f:
return int(f.read().strip() or "0", 0) & SW_CLOSED
except (OSError, ValueError):
return True
def state(self):
"""One '?' report's state word (e.g. 'Run', 'Hold:0', 'Door:1', 'Idle')."""
self.sock.sendall(b"?")
read_avail(self.sock, self.log, 0.3)
m = re.findall(r"<([A-Za-z]+(?::\d)?)", "".join(self.log[-4:]))
return m[-1] if m else ""
def wait_state(self, prefix, timeout):
end = time.time() + timeout
while time.time() < end:
st = self.state()
if st.startswith(prefix):
return st
time.sleep(0.1)
return None
def mpos_x(self):
self.sock.sendall(b"?")
read_avail(self.sock, self.log, 0.3)
m = re.findall(r"MPos:(-?[\d.]+)", "".join(self.log[-4:]))
return float(m[-1]) if m else None
def armed_count(self):
return "".join(self.log).count(ARMED)
def disarmed_count(self):
return "".join(self.log).count(DISARMED)
def close(self):
try:
self.sock.close()
except OSError:
pass
self.proc.send_signal(signal.SIGINT)
try:
self.proc.wait(5)
except subprocess.TimeoutExpired:
self.proc.kill()
self.stop.set()
self.pub.join(2)
shutil.rmtree(self.workdir, ignore_errors=True)
def read_state(sess, key=None, timeout=8.0):
"""The published grbl.state as text, polled until `key` appears (the
publisher runs on the protocol thread's pace; a '?' keeps it ticking)."""
path = os.path.join(sess.workdir, "grbl.state")
deadline = time.time() + timeout
text = ""
while time.time() < deadline:
sess.sock.sendall(b"?")
read_avail(sess.sock, sess.log, 0.2)
try:
with open(path) as f:
text = f.read()
except OSError:
text = ""
if text.endswith("\n") and (key is None or key in text):
return text
time.sleep(0.2)
return text
def test_status_files():
print("state files: settings and state published, following the edges")
s = Session("state-files", disarm_s=30)
try:
st = read_state(s, '"connected":true')
assert '"state":"Idle"' in st, "state file lacks Idle: %r" % st
assert '"connected":true' in st, "sender not reported connected: %r" % st
assert '"armed":false' in st, "armed before any job: %r" % st
assert '"model":"density"' in st and '"floor_pct":10' in st, \
"model/floor missing (the floor must be derived from boot): %r" % st
assert '"curve":"bench-default"' in st, \
"the curve in force is not published: %r" % st
assert '"modals":"[GC:' in st, "modal report missing: %r" % st
ts0 = float(st.split('"ts_mono":')[1].split(',')[0])
cfg = open(os.path.join(s.workdir, "grbl.settings")).read()
assert "$35=" in cfg and "$30=" in cfg, "settings file lacks $ lines"
# The derived floor is in force from boot, before any arm.
assert "$35=10" in cfg, "the derived floor is not in $$ from boot: %r" % [
l for l in cfg.splitlines() if l.startswith("$35")]
# The armed window reaches the file.
send_line(s.sock, "G91", s.log)
send_line(s.sock, "M3 S100", s.log)
st = read_state(s, '"armed":true')
assert '"armed":true' in st, "armed window not published: %r" % st
send_line(s.sock, "M5", s.log)
send_line(s.sock, "M2", s.log)
# A reconnect bumps the generation and stays connected.
gen0 = int(st.split('"generation":')[1].split(',')[0])
s.sock.close()
s.sock = s.connect()
st = read_state(s, '"connected":true')
gen1 = int(st.split('"generation":')[1].split(',')[0])
assert gen1 > gen0, "generation did not advance on reconnect (%d -> %d)" % (gen0, gen1)
ts1 = float(st.split('"ts_mono":')[1].split(',')[0])
assert ts1 > ts0, "ts_mono did not advance"
print("PASS [state-files]: published, armed/switch/revert followed, "
"generation %d -> %d" % (gen0, gen1))
finally:
s.close()
def test_laser_keys_reload():
"""M102 reloads the corner gamma, the curve, and the floor inside an
armed job, synchronized, and the state file follows them."""
print("laser keys reload: M102 takes rewritten laser keys mid-job")
s = Session("keys-reload", disarm_s=30, conf_extra="laser_corner_gamma = 1.5\n")
try:
st = read_state(s, '"gamma":1.50')
assert '"gamma":1.50' in st, "the gamma from the conf is not in force at boot: %r" % st
send_line(s.sock, "G91", s.log)
send_line(s.sock, "M4 S100", s.log)
st = read_state(s, '"armed":true')
assert '"armed":true' in st, "armed window not published: %r" % st
send_line(s.sock, "G1 X2 F600", s.log)
conf = os.path.join(s.workdir, "forgefirm.conf")
with open(conf, "w") as f:
f.write("laser_disarm_s = 30\nlaser_corner_gamma = 1.25\nlaser_dose_curve = off\n"
"laser_floor_density = 0\n")
send_line(s.sock, "M102", s.log)
assert wait_for(s.log, "laser keys reloaded: corner gamma 1.25, curve off, floor 0 %", 5, s.sock), \
"M102 did not report the reload: %r" % s.log[-6:]
st = read_state(s, '"gamma":1.25')
assert '"gamma":1.25' in st and '"curve":"off"' in st and '"floor_pct":0' in st, \
"the reloaded keys did not reach the state file: %r" % st
assert '"armed":true' in st, "the reload closed the armed window: %r" % st
send_line(s.sock, "G1 X2 F600", s.log)
send_line(s.sock, "M5", s.log)
send_line(s.sock, "M2", s.log)
print("PASS [keys-reload]: gamma 1.50 -> 1.25, the curve off, the floor 0, inside the armed job")
finally:
s.close()
def test_sender_change_holds_and_rearms():
"""Rule 3 on a machine with a button: a sender change mid-move, spindle
on, holds the job where the cut stopped; the held job's resume from
the new sender prompts for the press (collected from the poll, never
inside the core's held state), and the press re-arms and resumes."""
s = Session("sender-change-rearm", disarm_s=60, switches=SW_CLOSED)
try:
s.send_raw("M4 S100")
if not wait_for(s.log, PROMPT, 5, s.sock):
fail("[sender-change-rearm] no arm prompt")
s.press_button()
if not wait_for(s.log, ARMED, 5, s.sock):
fail("[sender-change-rearm] the press did not arm")
send_line(s.sock, "G1 X5 F60", s.log) # 5 s of motion
time.sleep(1.0)
s.sock.close() # mid-move, spindle on
time.sleep(0.5)
s.sock = s.connect()
if s.wait_state("Hold", 5) is None:
fail("[sender-change-rearm] the job was not held on the sender change")
x_held = s.mpos_x()
if x_held is None or not 0.5 < x_held < 4.5:
fail("[sender-change-rearm] held outside the move (MPos X %s)" % x_held)
before = s.armed_count()
s.sock.sendall(b"~")
if not wait_for(s.log, RESUME_PROMPT, 5, s.sock):
fail("[sender-change-rearm] the resume did not prompt for the button")
if s.wait_state("Hold", 2) is None:
fail("[sender-change-rearm] the job left Hold before the press")
s.press_button()
end = time.time() + 10
while time.time() < end and s.armed_count() != before + 1:
read_avail(s.sock, s.log, 0.2)
if s.armed_count() != before + 1:
fail("[sender-change-rearm] the press did not re-arm the held job")
wait_idle(s.sock, s.log)
x_end = s.mpos_x()
if x_end is None or abs(x_end - 5.0) > 0.01:
fail("[sender-change-rearm] the resumed job did not finish the move "
"(MPos X %s)" % x_end)
send_line(s.sock, "M5", s.log)
print("PASS [sender-change-rearm]: held at X=%.3f, the held job prompted on ~, the "
"press re-armed it, and it finished at X=%.3f" % (x_held, x_end))
finally:
s.close()
def test_resume_after_grace():
"""Rule 4, the other half: a job whose window closed while it sat in
Hold can still be resumed. The sender's ~ prompts for the button, the
press re-arms, and the cut finishes lit (the controller must not sit
in the arm wait inside the held state, where a press is never read)."""
s = Session("resume-after-grace", disarm_s=2, switches=SW_CLOSED)
try:
s.send_raw("M4 S100")
if not wait_for(s.log, PROMPT, 5, s.sock):
fail("[resume-after-grace] no arm prompt")
s.press_button()
if not wait_for(s.log, ARMED, 5, s.sock):
fail("[resume-after-grace] the press did not arm")
send_line(s.sock, "G1 X5 F60", s.log) # 5 s of motion
time.sleep(1.0)
s.sock.sendall(b"!") # feed hold
if not wait_for(s.log, DISARMED, 10, s.sock):
fail("[resume-after-grace] the window did not close in the hold")
before = s.armed_count()
s.sock.sendall(b"~")
if not wait_for(s.log, RESUME_PROMPT, 5, s.sock):
fail("[resume-after-grace] the resume did not prompt for the button")
if s.wait_state("Hold", 2) is None:
fail("[resume-after-grace] the job left Hold before the press")
s.press_button()
end = time.time() + 10
while time.time() < end and s.armed_count() != before + 1:
read_avail(s.sock, s.log, 0.2)
if s.armed_count() != before + 1:
fail("[resume-after-grace] the press did not re-arm the held job")
wait_idle(s.sock, s.log)
x_end = s.mpos_x()
if x_end is None or abs(x_end - 5.0) > 0.01:
fail("[resume-after-grace] the resumed job did not finish the move "
"(MPos X %s)" % x_end)
send_line(s.sock, "M5", s.log)
print("PASS [resume-after-grace]: ~ prompted, the press re-armed the held job, "
"and it finished at X=%.3f" % x_end)
finally:
s.close()
def test_rx_overrun_aborts():
"""A sender that writes past the RX ring (Bf: is the contract) loses
lines, and a job with lines missing is not the job the sender wrote:
the controller reports the overrun, stops the way a ^X stops it
(alarm, window closed) and never runs on with a hole in it. A clean
job after the reset arms again."""
s = Session("rx-overrun", disarm_s=60)
try:
send_line(s.sock, "M4 S100", s.log)
send_line(s.sock, "G1 X1 F600", s.log)
if not wait_for(s.log, ARMED, 5, s.sock):
fail("[rx-overrun] first laser-on did not arm")
job = "".join("G1 X%d F60\n" % (2 + (i % 2)) for i in range(120)) # ~1.3 KB at once
s.sock.sendall(job.encode())
if not wait_for(s.log, "RX overrun", 10, s.sock):
fail("[rx-overrun] the overrun was not reported to the sender")
if not s.wait_state("Alarm", 5):
fail("[rx-overrun] the overrun did not stop the job (state %s)" % s.state())
if s.disarmed_count() < 1:
read_avail(s.sock, s.log, 1.0)
if s.disarmed_count() < 1:
fail("[rx-overrun] the overrun did not close the armed window")
send_line(s.sock, "$X", s.log)
before = s.armed_count()
send_line(s.sock, "M4 S100", s.log)
send_line(s.sock, "G1 X0 F600", s.log)
end = time.time() + 5
while time.time() < end and s.armed_count() != before + 1:
read_avail(s.sock, s.log, 0.2)
if s.armed_count() != before + 1:
fail("[rx-overrun] a clean job after the overrun did not arm")
wait_idle(s.sock, s.log)
print("PASS [rx-overrun]: the overrun was reported, the job stopped in alarm with "
"the window closed, and the next clean job armed")
finally:
s.close()
def test_sigterm_mid_job():
"""Rule 5: a termination signal during an armed job is a STOP, not a
"finish the job first". The supervisor's expected-stop path (mode
switch, diagnostics, the emergency lever) delivers SIGTERM; the
controller must bring motion to a controlled stop, close the armed
window (latch relocked) and exit promptly - long before the job
would have ended and inside the supervisor's SIGKILL grace."""
s = Session("sigterm-mid-job", disarm_s=60)
try:
send_line(s.sock, "M4 S100", s.log)
send_line(s.sock, "G1 X300 F60", s.log) # a 5-minute move
if not wait_for(s.log, ARMED, 5, s.sock):
fail("[sigterm-mid-job] the job did not arm")
# Let the run get under way.
deadline = time.time() + 5
running = False
while time.time() < deadline and not running:
s.sock.sendall(b"?")
read_avail(s.sock, s.log, 0.3)
running = "<Run" in "".join(s.log[-5:])
if not running:
fail("[sigterm-mid-job] the job never reported Run")
t0 = time.time()
s.proc.send_signal(signal.SIGTERM)
# Keep draining the sender socket so the disarm report is captured.
exited = False
while time.time() - t0 < 10:
read_avail(s.sock, s.log, 0.2)
if s.proc.poll() is not None:
exited = True
break
dt = time.time() - t0
if not exited:
fail("[sigterm-mid-job] the controller did not exit within 10 s of "
"SIGTERM during a job (it must stop, not finish the job)")
if dt > 3.0:
fail("[sigterm-mid-job] exit took %.1f s after SIGTERM (want < 3 s)" % dt)
if s.disarmed_count() < 1:
fail("[sigterm-mid-job] no disarm (latch relock) reported before exit")
if s.proc.returncode != 0:
fail("[sigterm-mid-job] exit status %s (want a clean 0)" % s.proc.returncode)
print("PASS [sigterm-mid-job]: SIGTERM during a job stopped it, relocked "
"the latch and exited in %.2f s" % dt)
finally:
s.close()
def test_button_wait_arms():
"""Rule 6a: with a switch source the arm blocks on the button; a press
with the lid closed arms; the window is not opened before it."""
s = Session("button-wait", disarm_s=60, switches=SW_CLOSED)
try:
s.send_raw("M4 S100")
s.send_raw("G1 X1 F600")
if not wait_for(s.log, PROMPT, 5, s.sock):
fail("[button-wait] no button prompt with a switch source present")
read_avail(s.sock, s.log, 1.0)
if ARMED in "".join(s.log):
fail("[button-wait] armed before the button was pressed")
s.set_switches(SW_PRESSED)
if not wait_for(s.log, ARMED, 5, s.sock):
fail("[button-wait] the button press did not arm")
s.set_switches(SW_CLOSED)
wait_idle(s.sock, s.log)
print("PASS [button-wait]: the arm blocked on the button and the press armed")
finally:
s.close()
def test_lid_open_in_wait():
"""Rule 6b: the lid opening during the button wait cancels the job:
the lid-open cancel is reported, no alarm is raised, the window never
opens. A press with the lid still open must not arm either."""
s = Session("lid-in-wait", disarm_s=60, switches=SW_CLOSED)
try:
s.send_raw("M4 S100")
s.send_raw("G1 X1 F600")
if not wait_for(s.log, PROMPT, 5, s.sock):
fail("[lid-in-wait] no button prompt")
s.set_switches(SW_LID_OPEN)
if not wait_for(s.log, LID_CANCEL, 5, s.sock):
fail("[lid-in-wait] lid open during the wait was not reported as a cancel")
if not wait_for(s.log, "for help]", 5, s.sock):
fail("[lid-in-wait] no reset banner after the lid-open cancel (a clean cancel, not an alarm)")
if "ALARM" in "".join(s.log):
fail("[lid-in-wait] an alarm was raised on the lid-open cancel")
# Press with the lid still open: nothing may arm.
s.set_switches(1 << 2)
read_avail(s.sock, s.log, 1.0)
if ARMED in "".join(s.log):
fail("[lid-in-wait] armed after a lid-open cancel")
# Interlock variant on a fresh session, so the alarm state does not
# mask the check.
s.close()
s = Session("loop-in-wait", disarm_s=60, switches=SW_CLOSED)
s.send_raw("M4 S100")
s.send_raw("G1 X1 F600")
if not wait_for(s.log, PROMPT, 5, s.sock):
fail("[loop-in-wait] no button prompt")
s.set_switches(SW_LOOP_OPEN)
if not wait_for(s.log, LOOP_CANCEL, 5, s.sock):
fail("[loop-in-wait] interlock open during the wait was not reported as a cancel")
if ARMED in "".join(s.log):
fail("[loop-in-wait] armed despite the open interlock loop")
print("PASS [lid-in-wait]: lid or interlock open during the arm wait "
"canceled the job and never armed")
finally:
s.close()
def start_armed_move(s, tag, gcode="G1 X30 F60"):
"""Arm through the button and get a long move under way; returns once
the controller reports Run. The prompt and the armed message are
waited for as fresh occurrences: a press that lands before the
wait has begun is not consent (rule 15), so the press must follow
THIS job's prompt, not an earlier job's."""
prompts, armed = s.count(PROMPT), s.count(ARMED)
s.send_raw("M4 S100")
s.send_raw(gcode)
if not wait_for_new(s.log, PROMPT, prompts, 5, s.sock):
fail("[%s] no button prompt" % tag)
s.press_button()
if not wait_for_new(s.log, ARMED, armed, 5, s.sock):
fail("[%s] the button press did not arm" % tag)
if not s.wait_state("Run", 5):
fail("[%s] the job never reported Run" % tag)
def test_button_pause_resume():
"""Rule 7: outside the arm wait the button is the pause/resume toggle -
a press while running is a feed hold, a press while held is a cycle
start; the arming press itself is not a pause press. A travel job takes
the same toggle, and its resumed move lands on its target."""
s = Session("button-toggle", disarm_s=60, switches=SW_CLOSED)
try:
start_armed_move(s, "button-toggle")
time.sleep(0.5)
if s.state().startswith("Hold"):
fail("[button-toggle] the arming press was taken as a pause press")
s.press_button()
if not s.wait_state("Hold", 3):
fail("[button-toggle] a press while running did not feed-hold (state %s)" % s.state())
if not wait_for(s.log, "button pressed - job paused", 2, s.sock):
fail("[button-toggle] no pause message")
s.press_button()
if not s.wait_state("Run", 3):
fail("[button-toggle] a press while held did not resume (state %s)" % s.state())
if not wait_for(s.log, "button pressed - job resumed", 2, s.sock):
fail("[button-toggle] no resume message")
s.sock.sendall(b"\x18")
print("PASS [button-toggle]: press paused (Hold), press resumed (Run); the arming press did not")
finally:
s.close()
# The same toggle on a travel job: no laser, so no arm wait, and the
# move resumed from the hold lands where it was programmed to.
s = Session("button-toggle-travel", disarm_s=60, switches=SW_CLOSED)
try:
x0 = s.mpos_x()
if x0 is None:
fail("[button-toggle-travel] no position in the status report")
s.send_raw("M5")
s.send_raw("G91")
s.send_raw("G1 X40 F300") # 8 s
if not s.wait_state("Run", 5):
fail("[button-toggle-travel] the travel job never reported Run")
if s.count(PROMPT):
fail("[button-toggle-travel] a job with no laser asked for the arm press")
time.sleep(1.0)
s.press_button()
if not s.wait_state("Hold", 3):
fail("[button-toggle-travel] a press while running did not feed-hold (state %s)" % s.state())
held = s.mpos_x()
if held is None or not x0 < held < x0 + 40.0:
fail("[button-toggle-travel] held at X %s, not part-way from %s" % (held, x0))
time.sleep(0.5)
s.press_button()
if not s.wait_state("Run", 3):
fail("[button-toggle-travel] a press while held did not resume (state %s)" % s.state())
if not s.wait_state("Idle", 15):
fail("[button-toggle-travel] the resumed move did not complete (state %s)" % s.state())
x1 = s.mpos_x()
if x1 is None or abs(x1 - x0 - 40.0) > 0.05:
fail("[button-toggle-travel] the resumed move ended at X %s, not %.3f" % (x1, x0 + 40.0))
print("PASS [button-toggle-travel]: a travel job held at %.3f mm and resumed to its target, "
"%.3f mm" % (held - x0, x1 - x0))
finally:
s.close()
def test_lid_cancels_and_returns():
"""Rule 8 (lid_policy = cancel, the default): the lid opening mid-job
parks the job and cancels it - reason reported, armed window closed,
a soft reset (no alarm: position kept), then the head returns on its
own to the job start; the sender sees the reset banner. The interlock
loop is the same event with its own reason."""
s = Session("lid-cancel", disarm_s=60, switches=SW_CLOSED)
try:
start_armed_move(s, "lid-cancel", gcode="G1 X20 F600")
time.sleep(0.5)
x_mid = s.mpos_x()
s.set_switches(SW_LID_OPEN)
if not wait_for(s.log, "lid opened - job canceled", 5, s.sock):
fail("[lid-cancel] the lid open did not cancel the job")
if not wait_for(s.log, DISARMED, 5, s.sock):
fail("[lid-cancel] the cancel did not close the armed window")
if not wait_for(s.log, "for help]", 5, s.sock):
fail("[lid-cancel] no reset banner after the cancel")
if not wait_for(s.log, "returned to the job start", 15, s.sock):
fail("[lid-cancel] the head did not report returning to the job start")
st = s.wait_state("Idle", 5)
if not st:
fail("[lid-cancel] not Idle after the return (state %s)" % s.state())
if "ALARM" in "".join(s.log):
fail("[lid-cancel] an alarm was raised on the cancel (position should be kept)")
x_end = s.mpos_x()
if x_end is None or abs(x_end) > 0.05:
fail("[lid-cancel] head not back at the job start: MPos X=%s (was %s mid-job)" % (x_end, x_mid))
print("PASS [lid-cancel]: lid open mid-job -> canceled, disarmed, reset banner, "
"returned to X=%.3f (mid-job X=%.3f), Idle, no alarm" % (x_end, x_mid))
s.close()
# Interlock variant.
s = Session("loop-cancel", disarm_s=60, switches=SW_CLOSED)
start_armed_move(s, "loop-cancel", gcode="G1 X20 F600")
time.sleep(0.5)
s.set_switches(SW_LOOP_OPEN)
if not wait_for(s.log, "interlock open - job canceled", 5, s.sock):
fail("[loop-cancel] the interlock open did not cancel the job")
if not wait_for(s.log, "returned to the job start", 15, s.sock):
fail("[loop-cancel] no return to the job start after the interlock cancel")
print("PASS [loop-cancel]: interlock open mid-job -> canceled and returned")
finally:
s.close()
def test_pause_then_lid_cancel_returns_to_the_job_start():
"""Rule 9: the button pause + resume does not move the job start. The
lid cancel after a resume returns the head to where the job began;
the pause point is not a job start. Then the other side: a job left
in a hold and soft-reset is over, and the next job from that spot
returns to that spot, not to the earlier job's start."""
s = Session("pause-lid-cancel", disarm_s=60, switches=SW_CLOSED)
try:
start_armed_move(s, "pause-lid-cancel", gcode="G1 X30 F600")
time.sleep(0.6)
s.press_button()
if not s.wait_state("Hold", 3):
fail("[pause-lid-cancel] the press did not feed-hold (state %s)" % s.state())
time.sleep(0.5)
x_hold = s.mpos_x()
if x_hold is None or x_hold < 1.0:
fail("[pause-lid-cancel] the hold landed too close to the start to tell (X=%s)" % x_hold)
s.press_button()
if not s.wait_state("Run", 3):
fail("[pause-lid-cancel] the second press did not resume (state %s)" % s.state())
time.sleep(0.4)
s.set_switches(SW_LID_OPEN)
if not wait_for(s.log, "lid opened - job canceled", 5, s.sock):
fail("[pause-lid-cancel] the lid open did not cancel the job")
if not wait_for(s.log, "returned to the job start", 15, s.sock):
fail("[pause-lid-cancel] the head did not report returning to the job start")
if not s.wait_state("Idle", 5):
fail("[pause-lid-cancel] not Idle after the return (state %s)" % s.state())
x_end = s.mpos_x()
if x_end is None or abs(x_end) > 0.05:
fail("[pause-lid-cancel] the head returned to X=%s, not to the job start X=0 "
"(paused at X=%.3f: the resume was taken as a new job start)" % (x_end, x_hold))
print("PASS [pause-lid-cancel]: paused at X=%.3f, resumed, lid cancel returned to X=%.3f"
% (x_hold, x_end))
s.close()
# A job abandoned in a hold and reset is over: the next job starts
# where it starts.
s = Session("hold-reset-restart", disarm_s=60, switches=SW_CLOSED)
start_armed_move(s, "hold-reset-restart", gcode="G1 X30 F600")
time.sleep(0.6)
s.press_button()
if not s.wait_state("Hold", 3):
fail("[hold-reset-restart] the press did not feed-hold (state %s)" % s.state())
time.sleep(0.5)
s.sock.sendall(b"\x18")
if not wait_for(s.log, "for help]", 5, s.sock):
fail("[hold-reset-restart] no reset banner")
if not s.wait_state("Idle", 5):
fail("[hold-reset-restart] not Idle after the reset (state %s)" % s.state())
x_new_start = s.mpos_x()
if x_new_start is None or x_new_start < 1.0:
fail("[hold-reset-restart] the reset did not keep the position (X=%s)" % x_new_start)
read_avail(s.sock, s.log, 0.3)
start_armed_move(s, "hold-reset-restart", gcode="G1 X60 F600") # absolute: past the old start
time.sleep(0.5)
s.set_switches(SW_LID_OPEN)
if not wait_for(s.log, "returned to the job start", 15, s.sock):
fail("[hold-reset-restart] no return to the job start after the lid cancel")
if not s.wait_state("Idle", 5):
fail("[hold-reset-restart] not Idle after the return (state %s)" % s.state())
x_end = s.mpos_x()
if x_end is None or abs(x_end - x_new_start) > 0.05:
fail("[hold-reset-restart] the head returned to X=%s, not to this job's start X=%.3f "
"(the earlier job's start survived the reset)" % (x_end, x_new_start))
print("PASS [hold-reset-restart]: a job reset from a hold ended there; the next job "
"returned to its own start X=%.3f" % x_end)
finally:
s.close()
def test_lid_policy_hold():
"""lid_policy = hold keeps stock grblHAL behavior: the lid parks the job
in Door and a cycle start resumes it once closed."""
s = Session("lid-hold", disarm_s=60, switches=SW_CLOSED, conf_extra="lid_policy = hold\n")
try:
start_armed_move(s, "lid-hold", gcode="G1 X20 F60")
time.sleep(0.5)
s.set_switches(SW_LID_OPEN)
if not s.wait_state("Door", 3):
fail("[lid-hold] the lid open did not park the job in Door (state %s)" % s.state())
read_avail(s.sock, s.log, 1.0)
if "job canceled" in "".join(s.log):
fail("[lid-hold] the hold policy canceled the job")
s.set_switches(SW_CLOSED)
time.sleep(0.5)
s.sock.sendall(b"~")
if not s.wait_state("Run", 3):
fail("[lid-hold] cycle start did not resume the held job (state %s)" % s.state())
s.sock.sendall(b"\x18")
print("PASS [lid-hold]: with lid_policy=hold the lid parked the job in Door and ~ resumed it")
finally:
s.close()
def test_sender_change_during_rearm_cancels():
"""Rule 13: a sender change while a re-arm waits for the press
cancels the re-arm; the press resumes nothing and the job stays
held; the new sender's ~ prompts afresh."""
s = Session("rearm-sender-change", disarm_s=60, switches=SW_CLOSED)
try:
start_armed_move(s, "rearm-sender-change", gcode="G1 X30 F120")
time.sleep(1.0)
s.sock.close() # mid-move: held, disarmed
time.sleep(0.5)
s.sock = s.connect()
if s.wait_state("Hold", 5) is None:
fail("[rearm-sender-change] the job was not held on the sender change")
s.sock.sendall(b"~")
if not wait_for(s.log, RESUME_PROMPT, 5, s.sock):
fail("[rearm-sender-change] the resume did not prompt for the button")
# A second sender change, inside the re-arm wait. The cancel is
# reported at the disconnect, to nobody (output with no client is
# discarded), so the evidence is what the press does next: nothing.
s.sock.close()
time.sleep(0.5)
s.sock = s.connect()
st = read_state(s, '"connected":true')
if '"arming":false' not in st:
fail("[rearm-sender-change] the re-arm wait survived the sender change: %r" % st)
# A press now is the machine's resume button for the new sender:
# it resumes nothing and arms nothing, it opens a fresh re-arm
# prompt of its own (the displaced consent is gone), and only a
# second press, against that prompt, re-arms the held job.
before = s.armed_count()
prompts = s.count(RESUME_PROMPT)
s.press_button()
read_avail(s.sock, s.log, 1.0)
if s.armed_count() != before or not s.state().startswith("Hold"):
fail("[rearm-sender-change] a press after the canceled re-arm resumed the job "
"(armed %d -> %d, state %s)" % (before, s.armed_count(), s.state()))
if not wait_for_new(s.log, RESUME_PROMPT, prompts, 3, s.sock):
fail("[rearm-sender-change] the press after the cancel did not open a fresh re-arm prompt")
s.press_button()
end = time.time() + 10
while time.time() < end and s.armed_count() != before + 1:
read_avail(s.sock, s.log, 0.2)
if s.armed_count() != before + 1:
fail("[rearm-sender-change] the second press did not re-arm the held job")
s.sock.sendall(b"\x18")
print("PASS [rearm-sender-change]: the sender change canceled the re-arm; the next press "
"resumed nothing and prompted afresh, and the press after it re-armed")
finally:
s.close()
def main():
if not os.path.isfile(BIN):
fail("controller binary not found at %s" % BIN)
test_sender_change_during_rearm_cancels()
test_status_files()
test_laser_keys_reload()
test_sender_change_holds_and_rearms()
test_rx_overrun_aborts()
test_resume_after_grace()
test_button_wait_arms()
test_lid_open_in_wait()
test_button_pause_resume()
test_lid_cancels_and_returns()
test_pause_then_lid_cancel_returns_to_the_job_start()
test_lid_policy_hold()
test_sigterm_mid_job()
print("PASS: the armed-window lifecycle holds")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink