laser: feed hold and resume in GRBL mode, the sender-change hold

Stream harness rule 21: a feed hold leaves no dark ground in either mode (lit into the hold, dark while held, lit from the first step out), with realtime and wait-state steps in the session runner. Lifecycle harness: the hold a sender change puts a running job into, the resume that re-arms a held job from the sender and from the button, a reset from a held job, and the resume after the grace closed the window in Hold. Live-fire drills: holdres (the pause as a corner in time, the re-arm after the grace); senderchg follows the hold. BRINGUP: the gapless-pause and sender-change items close, the facts bank gains the measured hold and resume behavior; CAMPAIGN-LOG records the proof and the bench runs. Acceptance: the pause-resume-lid-cancel text follows the behavior; the driver stays covered by src/**.
This commit is contained in:
ScottW514
2026-09-01 18:03:45 -04:00
parent 6a48cd3969
commit 8a9c6a9062
6 changed files with 530 additions and 103 deletions
+140 -15
View File
@@ -66,6 +66,7 @@ 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 cancelled"
LOOP_CANCEL = "interlock open during arm - job cancelled"
@@ -363,12 +364,11 @@ def test_job_window():
def test_sender_change_mid_job():
"""Rule 3, the hard case: the sender changes while the job is still
running with the spindle on, so the core never turns the spindle off
between the two sessions. The next laser-on from the new sender must
still prompt: the arm decision reads the window, not the spindle-state
record (a job whose M5 was lost leaves that record on the same way).
M3 after M4 is a state change the core always pushes through
set_state, planner-synced, so it lands once the first job's move ends."""
running with the spindle on. The window closes and the job is HELD
where the cut stopped, so the next sender finds it in Hold rather than
at the end of a dark pass. Its cycle start goes through the resume
gate, which re-arms first: with no button on a host build that is
immediate, and the job then runs to its end lit."""
s = Session("sender-change-mid-job", disarm_s=60)
try:
send_line(s.sock, "M4 S100", s.log)
@@ -378,22 +378,144 @@ def test_sender_change_mid_job():
time.sleep(1.0)
s.sock.close() # mid-move, spindle on
time.sleep(0.5)
# The disarm message is written while no client is connected, and
# output with no client is discarded, so the new session cannot
# see it; the re-arm below is the evidence the window closed.
# The disarm and hold messages are written while no client is
# connected, and output with no client is discarded, so the new
# session cannot see them; the Hold state and the re-arm below are
# the evidence.
s.sock = s.connect()
st = s.wait_state("Hold", 5)
if st is None:
fail("[sender-change-mid-job] the job was not held on the sender change "
"(state %r)" % s.state())
x_held = s.mpos_x()
if x_held is None or not 0.5 < x_held < 4.5:
fail("[sender-change-mid-job] held outside the move (MPos X %s)" % x_held)
before = s.armed_count()
s.send_raw("M3 S100") # laser-on against a closed window
s.sock.sendall(b"~") # the new sender resumes
end = time.time() + 15
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-mid-job] a laser-on after a mid-job sender change did not "
"re-arm: the arm read the stale spindle state instead of the window")
send_line(s.sock, "M5", s.log)
fail("[sender-change-mid-job] the resume after a mid-job sender change did "
"not re-arm")
wait_idle(s.sock, s.log)
print("PASS [sender-change-mid-job]: a laser-on against a window closed mid-job "
"prompted and re-armed")
x_end = s.mpos_x()
if x_end is None or abs(x_end - 5.0) > 0.01:
fail("[sender-change-mid-job] the resumed job did not finish the move "
"(MPos X %s)" % x_end)
send_line(s.sock, "M5", s.log)
print("PASS [sender-change-mid-job]: the job held at X=%.3f on the sender change, "
"re-armed on the resume and finished at X=%.3f" % (x_held, x_end))
finally:
s.close()
def test_sender_change_holds_and_rearms():
"""Rule 3 on a machine with a button: 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")
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]: the held job prompted on ~, the press re-armed "
"it, and it finished at X=%.3f" % x_end)
finally:
s.close()
def test_sender_change_then_reset():
"""A new sender that does not want the held job resets it: a soft
reset from the hold ends the job with the position kept and no alarm."""
s = Session("sender-change-reset", disarm_s=60)
try:
send_line(s.sock, "M4 S100", s.log)
send_line(s.sock, "G1 X5 F60", s.log)
if not wait_for(s.log, ARMED, 5, s.sock):
fail("[sender-change-reset] first laser-on did not arm")
time.sleep(1.0)
s.sock.close()
time.sleep(0.5)
s.sock = s.connect()
if s.wait_state("Hold", 5) is None:
fail("[sender-change-reset] the job was not held on the sender change")
s.sock.sendall(b"\x18")
st = s.wait_state("Idle", 5)
if st is None:
fail("[sender-change-reset] a reset from the held job did not end in Idle "
"(state %r)" % s.state())
print("PASS [sender-change-reset]: a reset from the held job ends in Idle, no alarm")
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()
@@ -797,8 +919,11 @@ def main():
test_status_files()
test_sender_change()
test_sender_change_mid_job()
test_sender_change_holds_and_rearms()
test_sender_change_then_reset()
test_rx_overrun_aborts()
test_hold_grace()
test_resume_after_grace()
test_button_wait_arms()
test_lid_open_in_wait()
test_button_pause_resume()
+67
View File
@@ -353,6 +353,19 @@ def wait_idle(sock, log):
fail("controller never returned to Idle")
def wait_state(sock, log, prefix, timeout=5.0):
"""Poll '?' until the state word starts with prefix (e.g. 'Hold:0')."""
end = time.time() + timeout
while time.time() < end:
sock.sendall(b"?")
read_avail(sock, log, 0.3)
m = re.findall(r"<([A-Za-z]+(?::\d)?)", "".join(log[-3:]))
if m and m[-1].startswith(prefix):
return
time.sleep(0.1)
fail("controller never reached %s" % prefix)
def publish_verdicts(path, stop):
"""Publish a fresh, clean cooling verdict every 0.5 s (the arm flow
refuses without one; freshness window is 2 s). Same-host monotonic
@@ -418,6 +431,10 @@ def run_session(name, steps, conf=None, workdir=None, keep=False,
wait_idle(sock, log)
elif isinstance(step, tuple) and step[0] == "sleep":
time.sleep(step[1])
elif isinstance(step, tuple) and step[0] == "rt":
sock.sendall(step[1]) # a realtime character: no ok follows
elif isinstance(step, tuple) and step[0] == "wait_state":
wait_state(sock, log, step[1])
else:
send_line(sock, step, log)
@@ -929,6 +946,56 @@ def main():
print("PASS [curve]: S %s -> densities %s through the bench-default curve "
"(floored at %.3f)" % (list(CURVE_S), [round(g, 3) for g in got], floor_frac))
# --- rule 21: a feed hold leaves no dark ground, in either mode -----
# One long line at 100 mm/s, held mid-move and resumed. The planned
# deceleration runs lit (M4 velocity-scaled, M3 constant), the
# stationary stretch is dark, and the acceleration out of the hold is
# lit from its first step: a pause is a sharp corner in time. The
# first-window check is what pins the M3 resume, which once ran dark
# for a segment buffer because the segments prepped while held
# carried no spindle update.
HOLD_WIN = 704 # 25 ms at 28160 Hz
HOLD_EDGE = 120 # one pulse straddles each edge
for mode, job in (("m4", ["G90", "G21", "M4 S0", "G1 X150 F6000 S500"]),
("m3", ["G90", "G21", "M3 S500", "G1 X150 F6000"])):
steps = job + [("sleep", 0.7), ("rt", b"!"), ("wait_state", "Hold:0"),
("sleep", 0.5), ("rt", b"~"), WAIT_IDLE, "M5"]
data = run_session("hold-" + mode, steps, conf=DENSITY_CONF_FLOORED)
check_fire_gaps("hold-" + mode, data)
ticks = tick_bytes(data)
step = [1 if t & 0x05 else 0 for t in ticks]
fire = [1 if t & 0x10 else 0 for t in ticks]
first = step.index(1)
last = len(step) - 1 - step[::-1].index(1)
best, run = (0, 0), 0
for i in range(first, last + 1):
if step[i]:
if run > best[0]:
best = (run, i - run)
run = 0
else:
run += 1
dlen, dstart = best
dend = dstart + dlen
if dlen < 2000:
fail("[hold-%s] no hold in the stream (longest stepless run %d ticks)"
% (mode, dlen))
decel = sum(fire[dstart - HOLD_WIN:dstart])
dwell = sum(fire[dstart + HOLD_EDGE:dend - HOLD_EDGE])
accel = sum(fire[dend:dend + HOLD_WIN])
if decel < 50:
fail("[hold-%s] the deceleration into the hold ran dark (%d fire ticks in "
"its last 25 ms)" % (mode, decel))
if dwell:
fail("[hold-%s] FIRE while held: %d fire ticks in the stationary stretch"
% (mode, dwell))
if accel < 50:
fail("[hold-%s] the resume ran dark (%d fire ticks in the 25 ms after the "
"first step)" % (mode, accel))
print("PASS [hold-%s]: lit into the hold (%d fire ticks), dark while held "
"(%d ticks), lit from the first step out (%d fire ticks)"
% (mode, decel, dlen, accel))
print("PASS: all stream emission rules hold")
+161 -29
View File
@@ -147,11 +147,17 @@ Drills (pass a name):
and fits rise against dose. JSON records like dpatch.
flowload t1 | flowload t2 <secs> [pct] | flowload fit
senderchg A sender change mid-job: a 20 mm line at F60 lit on the
press, the connection dropped five seconds in, a reconnect,
the move finishing dark, then a fresh M3 and a 5 mm line that
must prompt for the button again. PASS: two prompts, two arms,
no lit sample between the drop and the second press, the
second line marks. Two presses; JSON record.
press, the connection dropped five seconds in, the job held
where it stopped, a reconnect, then ~ from the new session,
which must light the button and wait. PASS: the hold, the
resume prompt, no lit sample between the drop and the second
press, the rest of the line marks. Two presses; JSON record.
holdres Feed hold and resume, the pause as a corner in time: a 30 mm
M4 line held near 10 mm and resumed, then a 90 degree corner
to compare the marks; the same hold under M3 on the return
line (no dark lead on the resume); then a hold past the
disarm grace, resumed with ~ through the re-arm prompt. Two
presses: the arm, and the re-arm after the long hold.
overrun An RX overrun mid-job: a 20 mm line at F60 lit on the press,
a 93-line fill written at once three seconds in (about 1270
bytes against the 1023-byte ring). PASS: the controller
@@ -2389,7 +2395,6 @@ def drill_flowload(g):
SENDERCHG_S = 400
SENDERCHG_LINE1 = 'G1 X20 F60' # 20 s of motion
SENDERCHG_DROP_S = 5.0 # lit seconds before the drop
SENDERCHG_LINE2 = 'G1 X5 F600'
def senderchg_wait_reply(g, needle, timeout):
@@ -2420,9 +2425,9 @@ def drill_senderchg(g):
poller.start()
arm_cue()
print('>>> Scrap with 25 mm of free +X travel from the head. TWO presses: one')
print('>>> for the first line, and a second one after the reconnect, when the')
print('>>> prompt comes again. The line goes dark at the drop and the head')
print('>>> finishes the move without a sender; that is expected.\n')
print('>>> for the line, and a second one after the reconnect, when the button')
print('>>> lights again. The line goes dark at the drop and the head HOLDS')
print('>>> where it is; the new session resumes it with your second press.\n')
print('G91/G21: %s / %s' % (g.cmd('G91'), g.cmd('G21')))
t_m3 = time.time()
for ln in ('M3 S%d' % SENDERCHG_S, SENDERCHG_LINE1, 'M5'):
@@ -2445,31 +2450,33 @@ def drill_senderchg(g):
time.sleep(1.0)
g2 = Grbl(HOST, PORT) # a new session
t_reconnect = time.time()
st = g2.wait_state('Idle', 40)
t_idle1 = time.time()
print(' reconnected; the first move ended dark %.1f s after the drop (state %s)'
% (t_idle1 - t_drop, st))
st = g2.wait_state('Hold', 10)
print(' reconnected; the job is held %.1f s after the drop (state %s)'
% (time.time() - t_drop, st))
cs = sample_forgectrl() or {}
print(' engine after the drop: armed=%s verdict=%s' % (cs.get('armed'), cs.get('verdict')))
outcome = 'done'
t_run2 = None
try:
for ln in ('M3 S%d' % SENDERCHG_S, SENDERCHG_LINE2, 'M5'):
g2.s.sendall(ln.encode() + b'\n')
prompt2 = senderchg_wait_reply(g2, 'press the button', 10)
print(' second prompt after the reconnect: %s' % prompt2)
if not prompt2:
outcome = 'no-prompt'
if not st.startswith('Hold'):
outcome = 'not-held'
g2.rt(b'\x18')
else:
st = g2.wait_state('Run', FLOWLOAD_PRESS_WAIT_S)
if not st.startswith('Run'):
outcome = 'no-run'
g2.rt(b'~') # the new sender resumes the held job
prompt2 = senderchg_wait_reply(g2, 'press the button to resume', 10)
print(' resume prompt after the reconnect: %s' % prompt2)
if not prompt2:
outcome = 'no-prompt'
g2.rt(b'\x18')
else:
t_run2 = time.time()
g2.wait_state('Idle', 30)
print(' M2: %s' % g2.cmd('G90\nM2', timeout=10))
st = g2.wait_state('Run', FLOWLOAD_PRESS_WAIT_S)
if not st.startswith('Run'):
outcome = 'no-run'
g2.rt(b'\x18')
else:
t_run2 = time.time()
g2.wait_state('Idle', 60)
print(' M2: %s' % g2.cmd('G90\nM2', timeout=10))
finally:
try:
g2.cmd('M5', timeout=1)
@@ -2499,13 +2506,14 @@ def drill_senderchg(g):
print(' %+7.1f s armed=%s' % (p['t'] - t_m3, p['armed']))
last = p['armed']
print('--- emission (lit samples at 25 Hz) ---')
print(' first line, press to drop: %d samples (%.1f s)' % (len(lit1), len(lit1) / 25.0))
print(' line, press to drop: %d samples (%.1f s)' % (len(lit1), len(lit1) / 25.0))
print(' drop to second press: %d samples <- must be 0' % len(lit_gap))
print(' second line: %d samples (%.1f s)' % (len(lit2), len(lit2) / 25.0))
print(' the resumed rest of it: %d samples (%.1f s)' % (len(lit2), len(lit2) / 25.0))
ok = outcome == 'done' and prompt1 and len(lit1) > 25 and not lit_gap and len(lit2) > 5
print('\n%s: %s' % ('PASS' if ok else 'FAIL',
'the drop closed the window, nothing fired until the second press, '
'and the second laser-on prompted and armed' if ok else outcome))
'the drop held the job and closed the window, nothing fired until '
'the second press, and ~ from the new session prompted, re-armed '
'and finished the line' if ok else outcome))
rec = {'drill': 'senderchg', 'date': time.strftime('%Y-%m-%dT%H:%M:%S'),
't_m3': t_m3, 't_run1': t_run1, 't_drop': t_drop, 't_reconnect': t_reconnect,
't_run2': t_run2, 'outcome': outcome, 'pass': bool(ok),
@@ -3092,6 +3100,129 @@ def drill_expstop(g):
return trail
# --- feed hold and resume: the pause is a corner in time ----------------------
# One armed run on scrap, at F300 and S400. Under M4 a 30 mm line is held
# about 2 s in and resumed a second later, then the path turns 90 degrees:
# the pause mark near 10 mm and the corner at 30 mm sit on the same scrap
# for the eye, and should match. Under M3 the same hold on the return
# line: the resume must show no dark lead. Then a hold that outlives the
# disarm grace: the window closes in Hold, the sender's ~ must light the
# button and wait, the press re-arms, and the line finishes lit. Two
# presses: the arm, and the re-arm after the long hold.
HOLDRES_S = 400
HOLDRES_F = 300
def holdres_pause(g, label, wait_disarm=False):
"""Hold the running move ~2 s in, then resume; with wait_disarm the hold
outlives the grace and the resume goes through the re-arm prompt."""
st = g.wait_state('Run', 20)
if not st.startswith('Run'):
return 'no-run (%s)' % st
time.sleep(2.0)
g.rt(b'!')
st = g.wait_state('Hold', 5)
if not st.startswith('Hold'):
return 'no-hold (%s)' % st
seen = len(g.log)
if wait_disarm:
t0 = time.time()
while time.time() - t0 < 120:
s = sample_forgectrl()
if s and not s['armed']:
break
time.sleep(1)
else:
return 'still armed after 120 s in Hold'
print(' %s: the window closed in Hold after %.0f s; sending ~' % (label, time.time() - t0))
g.rt(b'~')
if not senderchg_wait_reply(g, 'press the button to resume', 10):
return 'no resume prompt on ~'
print(' >>> PRESS the button to resume the job')
st = g.wait_state('Run', FLOWLOAD_PRESS_WAIT_S)
if not st.startswith('Run'):
return 'no-run after the press (%s)' % st
if not any('laser armed' in ln for _t, ln in g.log[seen:]):
return 'resumed without re-arming'
else:
time.sleep(1.0)
g.rt(b'~')
st = g.wait_state('Run', 5)
if not st.startswith('Run'):
return 'no-run on ~ (%s)' % st
st = g.wait_state('Idle', 60)
return 'ok' if st.startswith('Idle') else 'never idle (%s)' % st
def drill_holdres(g):
print('=== feed hold and resume: the pause is a corner in time ===')
print('connect: %s' % prepare(g))
print('spindle off: %s' % g.cmd('M5'))
pre = sample_forgectrl()
if pre is None or pre.get('armed'):
print('REFUSED: the armed window is still open (or forgectrl is unreachable)')
return 2
arm_cue()
print('>>> Scrap with 35 mm of free +X and 25 mm of free +Y travel. TWO presses:')
print('>>> one to arm, one when the button lights again after the long hold.\n')
print('G91/G21: %s / %s' % (g.cmd('G91'), g.cmd('G21')))
results = []
def leg(lines):
for ln in lines:
g.s.sendall(ln.encode() + b'\n')
def plain(lines):
leg(lines)
g.wait_state('Run', 20)
g.wait_state('Idle', 60)
try:
# A: M4, a 30 mm line held near 10 mm, then the corner leg.
leg(('M4 S%d' % HOLDRES_S, 'G1 X30 F%d' % HOLDRES_F))
if not senderchg_wait_reply(g, 'press the button', 10):
print('ABORTED: no arm prompt')
return 1
r = holdres_pause(g, 'M4 pause')
results.append(('A M4 pause beside the corner', r))
print(' A: %s' % r)
if r != 'ok':
return 1
plain(('G1 Y10 F%d' % HOLDRES_F,))
# B: M3 on the return line, the same hold.
leg(('M3 S%d' % HOLDRES_S, 'G1 X-30 F%d' % HOLDRES_F))
r = holdres_pause(g, 'M3 pause')
results.append(('B M3 pause, no dark lead', r))
print(' B: %s' % r)
if r != 'ok':
return 1
# C: back under M4, a plain leg up, then the hold past the grace.
leg(('M4 S%d' % HOLDRES_S,))
plain(('G1 Y10 F%d' % HOLDRES_F,))
leg(('G1 X30 F%d' % HOLDRES_F,))
r = holdres_pause(g, 'grace', wait_disarm=True)
results.append(('C hold past the grace, ~ re-arms', r))
print(' C: %s' % r)
finally:
try:
g.cmd('M5', timeout=1)
except Exception:
pass
if 'Hold' in g.status():
g.rt(b'\x18')
ok = bool(results) and all(r == 'ok' for _n, r in results)
print('\n--- results ---')
for name, r in results:
print(' %s: %s' % (name, r))
print('\n%s: %s' % ('PASS' if ok else 'FAIL',
'held and resumed under M4 and M3, and the hold past the grace '
're-armed on ~ and finished; now judge the marks: the M4 pause '
'against the corner, the M3 pause for a dark lead' if ok
else 'see above'))
return 0 if ok else 1
def drill_ctrlstart(g):
"""Resume supervision after expstop: POST /controller/start, then
report /mode. No motion, no laser."""
@@ -3114,6 +3245,7 @@ def main():
'dpatch': drill_dpatch, 'flowload': drill_flowload,
'm4corner': drill_m4corner, 'm4feeds': drill_m4feeds,
'senderchg': drill_senderchg, 'overrun': drill_overrun,
'holdres': drill_holdres,
'expstop': drill_expstop, 'ctrlstart': drill_ctrlstart}
if drill not in drills:
print(__doc__)