vh/forgefirm
1
0
Fork 0
mirror of https://github.com/openglow-org/forgefirm.git synced 2026-09-27 08:41:13 -07:00
Code Issues Packages Projects Releases Wiki Activity

bench: the controller port and manual home harnesses

Browse Source 
Two host harnesses for the grblHAL driver's null-sink build, in the bench
registry and the README with the others. The driver's CI runs them.

ctlport_test.py drives the controller port beside a scripted Grbl sender
that counts every ok and error it is sent, which is the only way to see a
status routed to the wrong source. 13 cases: the socket's mode; a port
jog's status going to the port with the sender's count exact and the jog
run whole under the sender's '?' polls; a port error the sender's next line
does not inherit; the sender's line canceling a fast port jog and drawing
its own ok (fast on purpose: a slow jog stops at once and would pass with no
hold at all); a sender line queued right behind the port's; the refusals;
one client, and five reconnects right after a close; the status hook across
a soft reset; the dead-man; a CR LF sender; a sender that polls the way
LightBurn does, '?' with an end of line behind it, LF and CR LF (every port
jog accepted, a 60 mm port jog run whole with the polls landing inside it,
one ok per poll, a real line still canceling a port jog); and every
operation of both sets under an open armed window with M3 modal and S500,
where the dump must hold no FIRE tick.

manual_home_test.py reads the stream dump (GFSINK_DUMP) and the attribute
log (GFSINK_ATTR_LOG), so it can say that nothing was shipped and which
current was written, and how many times. 9 cases: a manual $H ships no step
and no FIRE tick and declares the offsets with the soft limits on and Z
kept; $H refused in a cycle; $MD refused under an open armed window with
nothing written; every motion source and $X refused while released; each
energize written exactly once; the port's panel operations; both pairs of
home offsets, alone and at once; and a controller killed under a release,
whose replacement writes only 0 and 0.

Both pass against the driver's extensions tree. The poll cases fail against
the driver without its empty-line rule ("8 of 8 port jogs were refused under
a status poll"), which is the defect they were written from: it was found
on the bench reference with LightBurn connected, and the sender these
harnesses had until then polled a bare '?'.
This commit is contained in:
ScottW514
2026-09-20 07:13:31 -04:00
parent 11f4c42705
commit b29bb9e023
4 changed files with 1137 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/bench.py
+17
View File
@@ -257,6 +257,23 @@ TOOLS = [
"card's stops: the ends of the reach run, two half-steps past either end alarms. A CI harness "
"(the grblHAL repo): needs the host-built null-sink controller, not the machine, so it is not a "
"bench-page tool."},
{"id": "ctlport-test", "title": "Controller port and line multiplexer harness", "script": "ctlport_test.py",
"safety": "dry", "where": "host", "ported": False, "args": [],
"desc": "A scripted sender that counts every response, beside the controller port's one client, on the "
"null-sink controller: a port jog's status goes to the port and never to the sender, a port "
"error does not reach the sender's parser, a sender line cancels a port jog and gets its own ok "
"(a line queued right behind the port's included), the refusals, the single client, a CR LF "
"sender, a soft reset, and the client as the dead-man. A CI harness (the grblHAL repo): needs "
"the host-built null-sink controller, not the machine, so it is not a bench-page tool."},
{"id": "manual-home-test", "title": "Manual home and motor release harness", "script": "manual_home_test.py",
"safety": "dry", "where": "host", "ported": False, "args": [],
"desc": "The manual homing provider and the motor release on the null-sink controller: $H under manual "
"ships nothing and declares X0 Y0 with the soft limits on and Z kept; while the motors are "
"released every motion source is refused and ships nothing, $X and a soft reset do not unlock "
"it, and only $ME and a manual $H write the energize, once each; the port's panel operations "
"do the same with the sender's count exact. Imports its sender and port client from "
"ctlport_test.py. A CI harness (the grblHAL repo): needs the host-built null-sink controller, "
"not the machine, so it is not a bench-page tool."},
{"id": "raster-dry", "title": "Dry top-speed raster per XY microstep mode", "script": "raster_dry.py",
"safety": "dry", "where": "board", "ported": True,
"args": [_arg("modes", "choice", "8 16 32", "the modes to run, in order", ["8", "16", "32", "8 16 32"])],
scripts/bench/README.md
+2
View File
@@ -33,6 +33,8 @@ page's takeover does that; from a host, stop them first.
| `laser_stream_test.py` | Host-side laser pulse-stream emission harness: runs the native null-sink controller with `GFSINK_DUMP`, drives small laser jobs over TCP, and checks the dumped bytes against the kernel feeder contract (leading power byte, no back-to-back power bytes, FIRE only inside cutting moves, every stream ends FIRE-clear, no FIRE on a stepless gap, no FIRE leak across cycle churn, the rapids after an M5 executed at idle ship dark, and the next job in the same process fires at the level the previous one ended at). Runs in the grblHAL repo's CI. |
| `laser_lifecycle_test.py` | Host-side operator-armed-window lifecycle harness (null-sink controller): arm once per job with M5/M3 persistence, the M2 close, sender-change re-consent, the disarm grace counting down in Hold, and arm refusal under a blocking cooling verdict. Runs in the grblHAL repo's CI. |
| `z_envelope_test.py` | Host-side Z envelope harness (null-sink controller): the Z soft limit belongs to the driver, not to `$20`, so the driver re-applies `sys.work_envelope`, `sys.homed` and `sys.soft_limits` for Z from the settings-changed chain. Checks that an unreferenced Z is collapsed to where the lens stands and blocks a move each way, that X and Y stay free, and that neither a `$20` write (the core clears the soft-limit mask in the setter) nor a `$132` write (which un-homes the axis as well) frees Z. Runs in the grblHAL repo's CI. |
| `ctlport_test.py` | Host-side controller port and line-multiplexer harness (null-sink controller): a scripted sender that counts every `ok` and `error` runs beside the port's one client on `grbl.ctl`. A port jog's status goes to the port and never to the sender, a port error does not reach the sender's parser, a sender line cancels a port jog and gets its own `ok` (a line queued right behind the port's included), plus the refusals, the single client, a CR LF sender, a soft reset, and the client as the dead-man. `python3 ctlport_test.py <grblHAL_glowforge>`. |
| `manual_home_test.py` | Host-side manual homing and motor release harness (null-sink controller, `GFSINK_DUMP` and `GFSINK_ATTR_LOG`): `$H` under `homing_mode = manual` ships nothing and declares X0 Y0; while the motors are released (`$MD`) every motion source is refused and ships nothing, `$X` and a soft reset do not unlock it, and only `$ME` and a manual `$H` write the energize. Imports its sender and port client from `ctlport_test.py`. `python3 manual_home_test.py <grblHAL_glowforge>`. |
| `live_fire_drills.py` | **LIVE LASER** drills, on the board (the bench page) or from a LAN host (`GF_HOST`): `live_fire_drills.py <drill> [S] [F]` - `witness` (emission witness, lid-IR peaks vs the ambient baseline, HV current, job-based disarm on M2), `hold` (disarm grace in Hold), `faultpos` (armed job refuses a stale origin after an underrun), `ircut` (lid-IR characterization cut at S/F), `pthresh` (laser power-threshold ladder: 13 constant-power rungs from 2 % to 30 % of full on scrap; the lowest rung that marks is the tube's striking threshold and reads directly as the `$35` value - requires `$35` = 0 for the run), `dladder` (density ladder at a chosen base period), `pcurve` (laser performance-curve ladder: one 100 mm line per level at 10 mm/s under M3, the laser off between rungs and a mid-ladder rung repeated at the end; reads `pic/hv_current` and the head thermopile `head/beam_detect_analog` (a scatter detector in the beam path upstream of the final mirror, so it sees the beam, not the material) from sysfs at ~25 Hz on the board, brackets each rung on the controller's Run/Idle states, and reports per rung the current with a clipped-at-1023 flag, the thermopile delta over its laser-off baseline and in-line drift, then the normalized curve, monotonicity, a line fit with its threshold intercept and the repeat-rung drift; JSON record with the raw trace in the bench data directory; rungs follow `laser_power_model`, a comma list overrides; a curve measurement wants `$35` = 0), `dpatch [F] [pitch] [length]` (depth witness for the density dose curve: two rows of small serpentine-filled patches, row A CW at feeds giving relative doses 1.0 to 0.25 of the reference feed, row B at the reference feed at 100/80/60/45/30 % density; the operator matches each row-B patch to the row-A patch of equal depth, which reads the density's light fraction off the material beside the thermopile's prediction; JSON record), `m4feeds [S] [F1] [F2]` (the density time base across feeds: one out-and-back line pair per feed at the same S under M4 density, one armed run; the operator reads within-line evenness and reversal darkness at both feeds - M4's velocity scaling is what holds dose per mm through the accel), `m4corner [S] [F]` (M4 velocity-scaled power into corners: a corner-heavy vector pattern at 30 % under M4 density, one armed run; the operator confirms every commanded segment marks - the floor makes a dropout unreachable - and the drill asserts the arm report, one discharge window and dark after), `m5dark` (the rapids after an M5 ship dark: one 20 mm line at M3 S400, M5, dwell, rapid back, dwell, rapid forward; PASS when the 25 Hz current trace shows one discharge segment and reads dark after the M5 and `laser_on_sampled` never re-lights; the catalog's `laser.m5-rapid-dark` is its port), `flowload` (cooling under laser load, one armed run per invocation, the conf keys it writes put back at the end, the pump never commanded off: `t1` reproduces the flow-check trip with the check on at its defaults and two 30 x 4 mm CW fills at F1500 starting on the press with no dark dwell, and reports the engine's rise/dT verdict beside the 25 Hz trace of both coolant sensors, the current, the digital witness and the heater output in 5 s bins across the window, with the shape at fire start; `t2 <secs> [pct]` runs with the check off and one fill of about `secs` lit seconds at CW or at `pct` density, and reports the lag to each sensor, the rise per raw-second of `hv_current` and what a full 50 s window would add against the 1.6 C margin; `fit` fits rise against dose over every t2 record; JSON records), `expstop` (armed kill on the expected-stop path; needs the panel token - `GF_TOKEN`, or the board's token file) and `ctrlstart` (the separate controller restart after it). Every drill waits for the operator's physical arm press; eye protection, fire watch, extinguisher, and exhaust are mandatory. |
| `pacing_test.py` | Protocol-loop pacing check (runs on the board, dry motion): idle and parked-in-Hold states are coarse-paced, active motion is tight-paced, and a feed-hold/resume mid-move preserves position with no feeder starve. |
| `planner_blocks_test.py` | Host-side planner buffer depth harness (null-sink controller): `$398` written to 400 and to 1000, the controller restarted on the same settings store each time, and it must answer on the port, report the depth in its status report and run a move. A byte-wide ring index once made 255 blocks or more spin the start. Runs in the grblHAL repo's CI. |
scripts/bench/ctlport_test.py
+660
View File
@@ -0,0 +1,660 @@
#!/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 controller port and the stream's line
multiplexer.
Runs the native grblHAL_glowforge binary in null-sink mode and drives two
clients at once: a scripted sender on the Grbl TCP socket that counts every
response it gets, and the port's one client on the Unix socket grbl.ctl in
the state directory. The sender's count is the point: a sender synchronizes
on ok and error, so one response too many or too few desynchronizes every
job after it.
1. a port jog's status goes to the port and never to the sender, which
sees a [MSG:] and nothing else
2. a port jog that errors does not leave its error with the sender: with
COMPATIBILITY_LEVEL 0 the core holds a parser error against the next
g-code line, and the next line here is the sender's
3. the sender always wins: a sender line that arrives during a port jog
cancels the jog, waits, and gets its own ok, never error:9
4. a port jog is refused while the sender is active, while a program
runs, and when its words carry anything a jog does not need; a motion
word the core's jog grammar rejects comes back as the core's error
5. the port has one client: a second connection is refused and the first
keeps working (it never displaces, unlike the Grbl socket); a client
that closes and reconnects at once is served
6. a CR LF sender keeps its count across a port jog, an empty line
included
7. the status hook survives a soft reset (the core puts its own report
handlers back at every reset)
8. the port's client is the dead-man: closing it cancels its jog
9. every port operation ships dark. Under an open armed window, with M3
modal and S above zero, a port jog, a canceled port jog, the refused
jogs, and the panel operations add steps to the shipped stream and not
one FIRE tick. A jog block carries the modal spindle state, so the
stream's mask on jogging is the only thing that keeps a jog dark, and
this is what holds the port to it
Usage: ctlport_test.py [path/to/grblHAL_glowforge]
"""
import json
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 = 2397
def fail(msg):
print("FAIL: %s" % msg)
sys.exit(1)
class Sender:
"""The scripted sender: every line it receives is classified, and the
responses (ok, error:N) are counted apart from everything else."""
def __init__(self, port, eol="\n"):
self.eol = eol
self.responses = [] # "ok" / "error:N", in order
self.other = [] # messages, reports, the banner
self.lock = threading.Lock()
for _ in range(50):
try:
self.sock = socket.create_connection(("127.0.0.1", port), timeout=1)
break
except OSError:
time.sleep(0.1)
else:
fail("cannot connect to the controller")
self.sock.settimeout(0.2)
self.alive = True
self.thread = threading.Thread(target=self._read, daemon=True)
self.thread.start()
time.sleep(0.5) # the banner
def _read(self):
buf = b""
while self.alive:
try:
data = self.sock.recv(4096)
except socket.timeout:
continue
except OSError:
return
if not data:
return
buf += data
while b"\n" in buf:
raw, buf = buf.split(b"\n", 1)
line = raw.decode(errors="replace").strip()
if not line:
continue
with self.lock:
if line == "ok" or line.startswith("error:"):
self.responses.append(line)
else:
self.other.append(line)
def count(self):
with self.lock:
return len(self.responses)
def send(self, line, timeout=5.0):
"""Send one line and return its response: the one response that
arrives for it."""
n = self.count()
self.sock.sendall((line + self.eol).encode())
end = time.time() + timeout
while time.time() < end:
if self.count() > n:
with self.lock:
return self.responses[n]
time.sleep(0.01)
fail("no response to the sender's %r" % line)
def realtime(self, byte):
self.sock.sendall(byte)
def state(self):
with self.lock:
n = len(self.other)
self.sock.sendall(b"?")
end = time.time() + 2.0
while time.time() < end:
with self.lock:
for line in self.other[n:]:
m = re.match(r"<([A-Za-z]+)", line)
if m:
return m.group(1)
time.sleep(0.01)
fail("no status report")
def wait_state(self, want, timeout=20.0):
end = time.time() + timeout
while time.time() < end:
if self.state() == want:
return
time.sleep(0.1)
fail("the controller never reached %s (now %s)" % (want, self.state()))
def saw(self, needle):
with self.lock:
return sum(needle in line for line in self.other)
def close(self):
self.alive = False
try:
self.sock.close()
except OSError:
pass
class PortClient:
def __init__(self, path):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.settimeout(5.0)
self.sock.connect(path)
self.buf = b""
def request(self, line):
self.sock.sendall((line + "\n").encode())
return self.reply()
def reply(self):
while b"\n" not in self.buf:
data = self.sock.recv(4096)
if not data:
return None
self.buf += data
raw, self.buf = self.buf.split(b"\n", 1)
return raw.decode()
def state(self):
return json.loads(self.request("state"))
def close(self):
try:
self.sock.close()
except OSError:
pass
def publish_verdicts(path, stop):
"""Stand in for the cooling engine: a clean verdict that acknowledges the
armed window, refreshed twice a second. A laser line arms against it."""
while not stop.is_set():
tmp = path + ".tmp"
with open(tmp, "w") as f:
f.write('{"ts_mono":%.3f,"fire_ok":true,"verdict":"OK","hold":false,'
'"resume_ok":true,"armed":true,"reason":""}'
% time.clock_gettime(time.CLOCK_MONOTONIC))
os.replace(tmp, path)
stop.wait(0.5)
class Session:
def __init__(self, eol="\n", laser=False):
self.workdir = tempfile.mkdtemp(prefix="ctlport-")
conf = os.path.join(self.workdir, "forgefirm.conf")
with open(conf, "w") as f:
f.write("cool_fan_grace_s = 0\nlaser_disarm_s = 30\nhoming_mode = manual\n")
env = dict(os.environ, GFHOME_CONF=conf, GF_STATE_DIR=self.workdir, FFLOG_STDERR="1")
env.pop("GFSINK", None)
env.pop("GF_SWITCH_FILE", None)
self.dump = os.path.join(self.workdir, "stream.bin")
self.stop = threading.Event()
self.pub = None
if laser:
verdict = os.path.join(self.workdir, "cooling.state")
env.update(GFSINK_DUMP=self.dump, GF_VERDICT_FILE=verdict)
self.pub = threading.Thread(target=publish_verdicts, args=(verdict, self.stop),
daemon=True)
self.pub.start()
self.proc = subprocess.Popen([BIN, "-p", str(PORT)], cwd=self.workdir, env=env,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.sender = Sender(PORT, eol)
self.path = os.path.join(self.workdir, "grbl.ctl")
for _ in range(50):
if os.path.exists(self.path):
break
time.sleep(0.1)
else:
fail("the controller did not create %s" % self.path)
self.port = PortClient(self.path)
def quiet(self):
"""Past the port's sender-quiet interval."""
time.sleep(0.45)
def ticks(self):
"""(step ticks, FIRE ticks) shipped so far. A byte with bit 7 set is a
power byte; in a tick byte 0x25 are the X, Y and Z steps and 0x10 is FIRE."""
with open(self.dump, "rb") as f:
data = f.read()
return (sum(1 for b in data if not b & 0x80 and b & 0x25),
sum(1 for b in data if not b & 0x80 and b & 0x10))
def close(self):
self.port.close()
self.sender.close()
self.proc.send_signal(signal.SIGINT)
try:
self.proc.wait(5)
except subprocess.TimeoutExpired:
self.proc.kill()
self.stop.set()
if self.pub is not None:
self.pub.join(2)
shutil.rmtree(self.workdir, ignore_errors=True)
def test_socket_mode(s):
mode = os.stat(s.path).st_mode & 0o777
if mode != 0o600:
fail("[mode] grbl.ctl is %o, not 600" % mode)
print("PASS [mode]: grbl.ctl is a socket with mode 600")
def test_status_routing(s):
if s.sender.send("G91") != "ok" or s.sender.send("G0 X1") != "ok":
fail("[routing] the sender's own lines did not get ok")
s.sender.wait_state("Idle")
s.quiet()
before = s.sender.count()
x0 = s.port.state()["mpos"][0]
r = s.port.request("jog G91 X2 F600")
if r != "ok":
fail("[routing] the port jog got %r, not ok" % r)
s.sender.wait_state("Idle")
# The sender polled its status all the way through the jog. A realtime
# character is not a line: it cancels nothing, and the jog ran whole.
moved = s.port.state()["mpos"][0] - x0
if abs(moved - 2.0) > 0.01:
fail("[routing] the sender's status polls cut the port jog short: moved %.3f of 2" % moved)
if s.sender.count() != before:
fail("[routing] the sender got a response for the port's line: %r"
% s.sender.responses[before:])
if not s.sender.saw("[MSG:Panel jog]"):
fail("[routing] the sender got no [MSG:] for the port jog")
if s.sender.send("G0 X1") != "ok":
fail("[routing] the sender's next line did not get ok")
print("PASS [routing]: the port jog's ok went to the port and the jog ran whole under the "
"sender's status polls; the sender saw a [MSG:] and its count is intact")
def test_error_not_inherited(s):
s.sender.wait_state("Idle")
s.quiet()
before = s.sender.count()
r = s.port.request("jog G91 X1") # no feed rate: the core refuses it
if not (r or "").startswith("error:"):
fail("[error] a jog with no feed rate got %r, not an error" % r)
if s.sender.count() != before:
fail("[error] the sender got the port's error: %r" % s.sender.responses[before:])
r2 = s.sender.send("G0 X1")
if r2 != "ok":
fail("[error] the sender's next g-code line got %r: it inherited the port's %s" % (r2, r))
print("PASS [error]: the port got %s; the sender's next line ran and got ok" % r)
def test_sender_wins(s):
s.sender.wait_state("Idle")
s.quiet()
x0 = s.port.state()["mpos"][0]
# Fast, so the cancel is a real deceleration: the core stays in the jog
# state for the length of it, and a sender line read in that window is
# the error:9 the hold exists to prevent. A slow jog stops at once and
# would pass with no hold at all.
if s.port.request("jog G91 X300 F6000") != "ok": # 3 s if nothing stops it
fail("[sender-wins] the long port jog was refused")
time.sleep(0.7)
st = s.port.state()
if st["state"] != "Jog" or not st["port_jog"]:
fail("[sender-wins] not jogging for the port: %r" % st)
t0 = time.time()
r = s.sender.send("G0 X1", timeout=10.0)
dt = time.time() - t0
if r != "ok":
fail("[sender-wins] the sender's line during a port jog got %r, not ok" % r)
s.sender.wait_state("Idle")
st = s.port.state()
if st["port_jog"]:
fail("[sender-wins] the port still claims the jog: %r" % st)
moved = st["mpos"][0] - x0
if moved > 200:
fail("[sender-wins] the port jog was not canceled (X moved %.1f mm)" % moved)
print("PASS [sender-wins]: the sender's line canceled the port jog, waited %.2f s, "
"and got ok" % dt)
def test_sender_wins_queued(s):
"""The window the read gate exists for. The core reads a line's bytes
in one pass and polls the stream only at the end of line, so a sender
line that sits in the ring right behind the port's jog line is read
while the jog it follows is still running, with no poll in between to
cancel it. A dwell opens the window on purpose: the port's line is
queued during it, and the sender streams its next line behind that."""
s.sender.wait_state("Idle")
s.sender.send("G91")
n = s.sender.count()
s.sender.sock.sendall(("G4 P1.5" + s.sender.eol).encode()) # no ok until the dwell ends
s.quiet()
s.port.sock.sendall(b"jog G91 X300 F6000\n") # queued behind the dwell
time.sleep(0.2)
s.sender.sock.sendall(("G0 X1" + s.sender.eol).encode()) # streamed behind the port's line
end = time.time() + 10.0
while time.time() < end and s.sender.count() < n + 2:
time.sleep(0.02)
got = s.sender.responses[n:n + 2]
port_reply = s.port.reply()
if port_reply != "ok":
fail("[sender-wins-queued] the queued port jog got %r, not ok" % port_reply)
if got != ["ok", "ok"]:
fail("[sender-wins-queued] the sender's dwell and the line behind the port jog got %r, "
"not two ok" % got)
s.sender.wait_state("Idle")
print("PASS [sender-wins-queued]: a sender line queued right behind the port's jog line "
"waited for the cancel and got ok")
class NewlinePoller(threading.Thread):
"""A sender that polls the way LightBurn does: '?' with an end of line
behind it, on a timer. The '?' is a realtime character; the end of line
is an empty line, which the core answers ok."""
def __init__(self, sender, period=0.25):
super().__init__(daemon=True)
self.sender, self.period = sender, period
self.polls = 0
self.halt = threading.Event()
def run(self):
while not self.halt.is_set():
self.sender.sock.sendall(b"?" + self.sender.eol.encode())
self.polls += 1
self.halt.wait(self.period)
def stop(self):
self.halt.set()
self.join()
def poll_case(tag, eol):
"""An empty line is not the sender speaking. With a sender polling
'?'<eol> four times a second (LightBurn polls about twice a second): a
port jog is never refused for it, a long port jog runs whole under it,
and every poll still draws its one ok, during the jog too. A real line
still goes first."""
s = Session(eol=eol)
try:
s.sender.wait_state("Idle")
s.quiet()
before = s.sender.count()
poller = NewlinePoller(s.sender)
poller.start()
time.sleep(0.6)
refused = []
for i in range(8): # spread across the poll period
r = s.port.request("jog G91 X0.5 F3000")
if r != "ok":
refused.append(r)
time.sleep(0.11)
if refused:
fail("[%s] %d of 8 port jogs were refused under a status poll: %s" % (tag, len(refused), refused))
while s.port.state()["state"] != "Idle":
time.sleep(0.05)
x0 = s.port.state()["mpos"][0]
if s.port.request("jog G91 X60 F3000") != "ok": # 1.2 s and more: several polls long
fail("[%s] the long port jog was refused under a status poll" % tag)
t0 = time.time()
while time.time() - t0 < 10:
st = s.port.state()
if st["state"] == "Idle" and not st["port_jog"]:
break
time.sleep(0.05)
moved = s.port.state()["mpos"][0] - x0
if abs(moved - 60.0) > 0.02:
fail("[%s] a status poll cut the port jog short: moved %.3f of 60" % (tag, moved))
polls_mid = poller.polls
# A real line still goes first: it cancels a port jog and draws its ok.
x1 = s.port.state()["mpos"][0]
if s.port.request("jog G91 X-300 F6000") != "ok":
fail("[%s] the fast port jog was refused" % tag)
time.sleep(0.7)
poller.stop() # the real line's response is then the only one outstanding
time.sleep(0.3)
n = s.sender.count()
s.sender.sock.sendall(("G4 P0" + eol).encode())
end = time.time() + 10
while time.time() < end and s.sender.count() <= n:
time.sleep(0.01)
if s.sender.count() != n + 1 or s.sender.responses[n] != "ok":
fail("[%s] the sender's real line under a port jog got %r" % (tag, s.sender.responses[n:]))
s.sender.wait_state("Idle")
back = x1 - s.port.state()["mpos"][0]
if back > 250:
fail("[%s] the sender's real line did not cancel the port jog: it ran %.1f of 300" % (tag, back))
time.sleep(0.5)
got = s.sender.count() - before
want = poller.polls + 1 # one ok per poll's empty line, and the G4
if got != want:
fail("[%s] %d polls and one line drew %d responses, not %d" % (tag, poller.polls, got, want))
print("PASS [%s]: under a '?'+EOL poll 8 of 8 port jogs were accepted, a 60 mm port jog ran whole "
"(%d polls into it), every poll drew its ok (%d), and a real line still canceled a port jog at "
"%.1f of 300" % (tag, polls_mid, poller.polls, back))
finally:
s.close()
def test_poll_lf():
poll_case("poll-lf", "\n")
def test_poll_crlf():
poll_case("poll-crlf", "\r\n")
def test_refusals(s):
s.sender.wait_state("Idle")
s.sender.send("G90")
r = s.port.request("jog G91 X1 F600") # inside the quiet interval
if r != "busy:sender":
fail("[refusals] a jog right after a sender line got %r, not busy:sender" % r)
s.quiet()
for words in ("G91 X1 F600 $H", "g91 x1 f600", "G91 X1 F600;M3", ""):
r = s.port.request("jog " + words)
if r != "error:invalid":
fail("[refusals] jog %r got %r, not error:invalid" % (words, r))
x0 = s.port.state()["mpos"][0]
r = s.port.request("jog G1 X5 F600") # a motion mode is not a jog word
if not (r or "").startswith("error:"):
fail("[refusals] jog G1 got %r, not the core's error" % r)
r = s.port.request("jog G91 X1 F600 S500 M3")
if not (r or "").startswith("error:"):
fail("[refusals] a jog with S and M3 got %r, not the core's error" % r)
time.sleep(0.3)
if abs(s.port.state()["mpos"][0] - x0) > 0.001:
fail("[refusals] a refused jog moved the machine")
s.sender.send("G91")
s.sender.send("G1 X40 F300") # a program: 8 s of Run
s.quiet()
st = s.port.state()
r = s.port.request("jog G91 Y1 F600")
if r != "busy:state":
fail("[refusals] a jog during a program (%s) got %r, not busy:state" % (st["state"], r))
s.sender.realtime(b"\x18") # end the program
time.sleep(1.0)
s.sender.send("$X")
print("PASS [refusals]: busy:sender inside the quiet interval, error:invalid for stray "
"characters, the core's error for G1 and for S/M3, busy:state during a program")
def test_single_client(s):
second = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
second.settimeout(3.0)
second.connect(s.path)
try:
second.sendall(b"state\n")
data = second.recv(100)
except OSError:
data = b""
second.close()
if data:
fail("[single-client] a second port client was served: %r" % data)
if s.port.state().get("state") is None:
fail("[single-client] the first client stopped working")
# A client that closes and reconnects at once is not a second client: the
# port has to notice the hang-up before it judges the new connection.
for i in range(5):
s.port.close()
s.port = PortClient(s.path)
try:
ok = s.port.state().get("state") is not None
except (OSError, TypeError, ValueError):
ok = False
if not ok:
fail("[single-client] reconnect %d right after a close was refused" % (i + 1))
print("PASS [single-client]: the second connection was refused, the first kept working, and "
"five reconnects right after a close were each served")
def test_crlf():
s = Session(eol="\r\n")
try:
for line in ("G91", "G0 X1", ""):
if s.sender.send(line) != "ok":
fail("[crlf] %r did not get ok" % line)
s.sender.wait_state("Idle")
s.quiet()
before = s.sender.count()
if s.port.request("jog G91 X1 F600") != "ok":
fail("[crlf] the port jog was refused")
s.sender.wait_state("Idle")
for line in ("", "G0 X1", "", ""):
if s.sender.send(line) != "ok":
fail("[crlf] %r after the port jog did not get ok" % line)
time.sleep(0.5)
got = s.sender.count() - before
if got != 4:
fail("[crlf] 4 sender lines after the port jog got %d responses" % got)
print("PASS [crlf]: a CR LF sender's count is exact across a port jog, empty lines included")
finally:
s.close()
def test_port_dark():
s = Session(laser=True)
try:
s.sender.send("G91")
s.sender.send("M3 S500")
s.sender.send("G1 X10 F1500")
s.sender.wait_state("Idle")
time.sleep(0.5)
if not s.sender.saw("laser armed"):
fail("[dark] the laser line did not arm: the case would prove nothing")
steps0, fire0 = s.ticks()
if fire0 == 0:
fail("[dark] the armed G1 shipped no FIRE tick: the dump cannot see emission")
s.quiet()
replies = {}
replies["state"] = s.port.request("state")
replies["jog"] = s.port.request("jog G91 X8 F1200")
s.sender.wait_state("Idle")
replies["jog-long"] = s.port.request("jog G91 X200 F6000")
time.sleep(0.4)
replies["cancel"] = s.port.request("cancel")
s.sender.wait_state("Idle")
for words in ("G1 X5 F600", "G91 X5 F600 S900 M3", "G91 X5 F600 M4", "G91 G2 X5 I2 F600"):
replies["jog " + words] = s.port.request("jog " + words)
replies["release"] = s.port.request("release") # refused: a laser job is armed
replies["energize"] = s.port.request("energize")
replies["home"] = s.port.request("home") # manual: declares X0 Y0, moves nothing
replies["jog-after-home"] = s.port.request("jog G91 X6 F1200")
s.sender.wait_state("Idle")
time.sleep(0.5)
steps1, fire1 = s.ticks()
if replies["jog"] != "ok" or replies["jog-long"] != "ok" or replies["jog-after-home"] != "ok":
fail("[dark] a port jog under the open window was refused: %r" % replies)
if steps1 <= steps0:
fail("[dark] the port jogs shipped no step: the case would prove nothing")
if s.sender.saw("laser disarmed"):
fail("[dark] the armed window closed during the case: %r" % replies)
if fire1 != fire0:
fail("[dark] port operations under an open window with M3 S500 shipped %d FIRE ticks: %r"
% (fire1 - fire0, replies))
print("PASS [dark]: under an open armed window with M3 S500, %d port operations shipped "
"%d steps and no FIRE tick" % (len(replies), steps1 - steps0))
finally:
s.close()
def test_reset_keeps_hook(s):
s.sender.realtime(b"\x18")
time.sleep(1.0)
s.sender.send("$X")
s.sender.wait_state("Idle")
s.quiet()
before = s.sender.count()
if s.port.request("jog G91 X1 F600") != "ok":
fail("[reset] the port jog after a soft reset was refused")
s.sender.wait_state("Idle")
if s.sender.count() != before:
fail("[reset] after a soft reset the port's ok went to the sender")
print("PASS [reset]: the status hook survives a soft reset")
def test_dead_man(s):
s.sender.wait_state("Idle")
s.quiet()
if s.port.request("jog G91 X60 F200") != "ok":
fail("[dead-man] the long port jog was refused")
time.sleep(0.5)
s.port.close()
t0 = time.time()
s.sender.wait_state("Idle", timeout=6.0)
print("PASS [dead-man]: closing the port's client canceled its jog (idle after %.2f s)"
% (time.time() - t0))
s.port = PortClient(s.path)
if s.port.state()["port_jog"]:
fail("[dead-man] a new client inherited the old jog")
def main():
if not os.path.exists(BIN):
fail("no controller binary at %s" % BIN)
s = Session()
try:
test_socket_mode(s)
test_status_routing(s)
test_error_not_inherited(s)
test_sender_wins(s)
test_sender_wins_queued(s)
test_refusals(s)
test_single_client(s)
test_reset_keeps_hook(s)
test_dead_man(s)
finally:
s.close()
test_crlf()
test_poll_lf()
test_poll_crlf()
test_port_dark()
print("ALL PASS")
if __name__ == "__main__":
main()
scripts/bench/manual_home_test.py
+458
View File
@@ -0,0 +1,458 @@
#!/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 manual homing provider and the motor release.
Runs the native grblHAL_glowforge binary in null-sink mode with GFSINK_DUMP
capturing the shipped byte stream and GFSINK_ATTR_LOG capturing the sysfs
writes the hardware would get (the release is the X and Y step currents at 0). An operator's hands are on a released gantry,
so the rule these cases pin is that nothing moves it and nothing but the
operator's own $ME or a manual $H energizes it:
1. $H with homing_mode = manual ships no step, declares X0 Y0 where the
head stands, marks X and Y homed and turns their soft limits on, leaves
Z where it was, and tells the sender the home was set by hand
2. $H is refused while a program runs
3. $MD writes the release, drops the X and Y reference at once, and locks
the machine in the alarm state; it is refused under an open armed window
4. while released, every motion source is refused and ships nothing: a
sender jog, a sender G0, a program line, a port jog, and a homing runner;
$X cannot unlock it and a soft reset does not either
5. only $ME and a manual $H write the energize, each exactly once; after
$ME the position is still unknown, and after the $H it is homed
6. the port's panel operations (release, energize, home) do the same with
a sender connected, and the sender's response count stays exact; home
is refused unless homing_mode = manual
7. manual_home_x and manual_home_y: the coordinate the stop blocks stand
for, the origin by default and never negative; the envelope starts at
the blocks, and an out-of-range offset is clamped
9. a controller that starts over a released gantry (the marker in the
state directory) comes up locked and writes no current until $ME
8. each provider reads its own offsets and no other's: with both pairs set,
a manual home declares manual_home_* and a camera home declares
gfcloud_home_*, which may be negative, with the envelope reaching back
to it
Usage: manual_home_test.py [path/to/grblHAL_glowforge]
"""
import os
import shutil
import signal
import subprocess
import tempfile
import threading
import time
from ctlport_test import BIN, PortClient, Sender, fail, publish_verdicts
PORT = 2396
NOTICE = "Manual home: position set where the head was placed"
RELEASED = "Motors released"
RUNNER_MARK = "runner-ran"
class Session:
def __init__(self):
self.workdir = tempfile.mkdtemp(prefix="manual-home-")
self.conf = os.path.join(self.workdir, "forgefirm.conf")
self.dump = os.path.join(self.workdir, "stream.bin")
self.attr_log = os.path.join(self.workdir, "attr.log")
self.set_mode("manual")
verdict = os.path.join(self.workdir, "cooling.state")
self.stop = threading.Event()
self.pub = threading.Thread(target=publish_verdicts, args=(verdict, self.stop), daemon=True)
self.pub.start()
env = dict(os.environ, GFHOME_CONF=self.conf, GF_STATE_DIR=self.workdir,
GF_VERDICT_FILE=verdict, GFSINK_DUMP=self.dump,
GFSINK_ATTR_LOG=self.attr_log, FFLOG_STDERR="1")
env.pop("GFSINK", None)
env.pop("GF_SWITCH_FILE", None)
self.env = env
self.proc = subprocess.Popen([BIN, "-p", str(PORT)], cwd=self.workdir, env=env,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.sender = Sender(PORT)
path = os.path.join(self.workdir, "grbl.ctl")
for _ in range(50):
if os.path.exists(path):
break
time.sleep(0.1)
self.port = PortClient(path)
# A controller writes the hold currents at its start; what the cases
# count comes after that.
self.base = 0
start = self.attr_writes()
if start != [("33", "5")]:
fail("the controller's start wrote %r to the step currents, not the hold currents" % start)
self.base = len(start)
def set_mode(self, mode, extra=""):
"""The driver re-reads the file at every $H."""
with open(self.conf, "w") as f:
f.write("cool_fan_grace_s = 0\nlaser_disarm_s = 1\nhoming_mode = %s\n"
"gfcloud_home_cmd = touch %s\n%s"
% (mode, os.path.join(self.workdir, RUNNER_MARK), extra))
def motion_ticks(self):
"""Step and FIRE ticks shipped so far (a byte with bit 7 set is a power byte)."""
try:
with open(self.dump, "rb") as f:
data = f.read()
except OSError:
return 0
return sum(1 for b in data if not b & 0x80 and b & 0x35)
def attr_writes(self):
"""The X and Y step-current writes the hardware would have got, in
order, as (x, y) pairs: ("0", "0") is a release, ("33", "5") the hold
currents, which is what an energize writes."""
vals = {"pic/x_step_current": [], "pic/y_step_current": []}
try:
with open(self.attr_log) as f:
for line in f:
parts = line.split()
if len(parts) == 2 and parts[0] in vals:
vals[parts[0]].append(parts[1])
except OSError:
pass
return list(zip(vals["pic/x_step_current"], vals["pic/y_step_current"]))[self.base:]
def respawn(self, hard=True):
"""End the controller (SIGKILL when hard: it gets no chance to tidy
up) and start another over the same state directory."""
self.port.close()
self.sender.close()
if hard:
self.proc.kill()
else:
self.proc.send_signal(signal.SIGINT)
self.proc.wait(10)
self.proc = subprocess.Popen([BIN, "-p", str(PORT)], cwd=self.workdir, env=self.env,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.sender = Sender(PORT)
time.sleep(0.5)
self.port = PortClient(os.path.join(self.workdir, "grbl.ctl"))
def quiet(self):
time.sleep(0.45)
def close(self):
self.port.close()
self.sender.close()
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 refused(s, line):
"""Send a line that must be refused, and return its error. The core holds
a parser error against every g-code line that follows until an empty line
acknowledges it (the streaming contract), so the acknowledgment is sent
here: the next line is then judged on its own merits."""
r = s.sender.send(line)
if r.startswith("error"):
if s.sender.send("") != "ok":
fail("the empty line that acknowledges %s did not get ok" % r)
return r
return None
def test_manual_home(s):
s.sender.send("G91")
s.sender.send("G0 X20 Y10") # away from wherever X0 Y0 was
s.sender.wait_state("Idle")
before = s.port.state()
if abs(before["mpos"][0]) < 1:
fail("[manual-home] the head did not move before the home: %r" % before)
if before["homed"] & 3:
fail("[manual-home] X or Y homed before any home: %r" % before)
if s.sender.send("$J=G91 X-100 F3000") != "ok": # unhomed: no X soft limit yet
fail("[manual-home] an unhomed jog past X0 was refused")
s.sender.wait_state("Idle")
time.sleep(0.5)
ticks = s.motion_ticks()
z_before = s.port.state()["mpos"][2]
if s.sender.send("$H") != "ok":
fail("[manual-home] $H under manual did not get ok")
time.sleep(0.5)
st = s.port.state()
if s.motion_ticks() != ticks:
fail("[manual-home] $H shipped %d motion ticks" % (s.motion_ticks() - ticks))
if st["mpos"][0] != 0 or st["mpos"][1] != 0:
fail("[manual-home] X and Y are not 0 after the home: %r" % st)
if st["homed"] & 3 != 3:
fail("[manual-home] X and Y are not marked homed: %r" % st)
if st["mpos"][2] != z_before:
fail("[manual-home] Z changed across the home: %r -> %r" % (z_before, st["mpos"][2]))
if st["state"] != "Idle":
fail("[manual-home] not idle after the home: %r" % st)
if not s.sender.saw(NOTICE):
fail("[manual-home] the sender was not told the home was set by hand")
r = refused(s, "$J=G91 X-5 F600") # off the bed: the soft limit is on now
if r != "error:15":
fail("[manual-home] a jog past X0 after the home got %r, not error:15" % r)
print("PASS [manual-home]: $H shipped nothing, declared X0 Y0, homed X and Y with soft "
"limits on, kept Z, and told the sender")
def test_home_refused_in_cycle(s):
s.sender.send("G90")
s.sender.send("G1 X40 F300")
time.sleep(0.5)
if s.port.state()["state"] != "Run":
fail("[home-in-cycle] the program is not running: %r" % s.port.state())
r = refused(s, "$H")
if r is None:
fail("[home-in-cycle] $H during a program was accepted")
if s.port.state()["homed"] & 3 != 3:
fail("[home-in-cycle] the refused $H dropped the reference")
s.sender.wait_state("Idle", timeout=15.0)
print("PASS [home-in-cycle]: $H during a program got %s and changed nothing" % r)
def test_release_refused_armed(s):
s.sender.send("M3 S100")
s.sender.send("G1 X45 F1500")
s.sender.wait_state("Idle") # motion done, the armed window still open
if not s.sender.saw("laser armed"):
fail("[release-armed] the laser line did not arm, so the case proves nothing")
r = refused(s, "$MD")
if r is None:
fail("[release-armed] $MD under an open armed window was accepted")
if s.attr_writes():
fail("[release-armed] the refused $MD wrote the release: %r" % s.attr_writes())
s.sender.send("M5")
time.sleep(2.5) # past laser_disarm_s: the window closes
print("PASS [release-armed]: $MD under an open armed window got %s and wrote nothing" % r)
def test_release_locks(s):
if s.sender.send("$MD") != "ok":
fail("[release] $MD at idle was refused")
st = s.port.state()
if s.attr_writes() != [("0", "0")]:
fail("[release] the release was not written once: %r" % s.attr_writes())
if not st["released"] or st["state"] != "Alarm" or st["homed"] & 3:
fail("[release] not released, locked and unreferenced: %r" % st)
if not st["homed"] & 4:
fail("[release] Z lost its reference: %r" % st)
if not s.sender.saw(RELEASED):
fail("[release] the sender was not told")
ticks = s.motion_ticks()
x0 = st["mpos"][0]
for line in ("$J=G91 X5 F600", "G0 X5", "G1 X5 F500", "G91 G1 Y3 F500"):
if refused(s, line) is None:
fail("[release] the sender's %r was accepted while released" % line)
s.quiet()
r = s.port.request("jog G91 X5 F600")
if r != "busy:released":
fail("[release] a port jog got %r while released" % r)
if refused(s, "$X") is None:
fail("[release] $X was accepted while released")
s.set_mode("gfcloud")
r = refused(s, "$H")
s.set_mode("manual")
if r is None or os.path.exists(os.path.join(s.workdir, RUNNER_MARK)):
fail("[release] a homing runner was not refused while released (%r)" % r)
s.sender.realtime(b"\x18")
time.sleep(1.0)
r = refused(s, "$X")
st = s.port.state()
if r is None or st["state"] != "Alarm" or not st["released"]:
fail("[release] a soft reset and $X unlocked a released machine: %r %r" % (r, st))
if refused(s, "G0 X5") is None:
fail("[release] G0 after a soft reset was accepted while released")
time.sleep(0.5)
if s.motion_ticks() != ticks or s.port.state()["mpos"][0] != x0:
fail("[release] something moved while released")
if s.attr_writes() != [("0", "0")]:
fail("[release] something other than $ME or $H wrote the step currents: %r" % s.attr_writes())
print("PASS [release]: released, locked in alarm, X and Y unreferenced, Z kept; a sender "
"jog, G0, program lines, a port jog, a homing runner, $X and a soft reset all "
"refused; nothing shipped, nothing energized")
def test_energize(s):
if s.sender.send("$ME") != "ok":
fail("[energize] $ME was refused")
st = s.port.state()
if s.attr_writes() != [("0", "0"), ("33", "5")]:
fail("[energize] the energize was not written once: %r" % s.attr_writes())
if st["released"] or st["state"] != "Idle" or st["homed"] & 3:
fail("[energize] not energized, idle and still unreferenced: %r" % st)
if s.sender.send("$ME") != "ok" or s.attr_writes() != [("0", "0"), ("33", "5")]:
fail("[energize] a second $ME wrote again: %r" % s.attr_writes())
if s.sender.send("$J=G91 X1 F600") != "ok":
fail("[energize] a jog after $ME was refused")
s.sender.wait_state("Idle")
s.sender.send("$MD")
if s.sender.send("$H") != "ok":
fail("[energize] a manual $H while released was refused")
st = s.port.state()
if s.attr_writes() != [("0", "0"), ("33", "5"), ("0", "0"), ("33", "5")]:
fail("[energize] the manual $H did not energize exactly once: %r" % s.attr_writes())
if st["released"] or st["state"] != "Idle" or st["homed"] & 3 != 3:
fail("[energize] not energized, idle and homed after the manual $H: %r" % st)
print("PASS [energize]: $ME energized once and left X and Y unreferenced; a manual $H "
"energized once and homed")
def test_port_panel_ops(s):
s.sender.wait_state("Idle")
s.quiet()
before = s.sender.count()
for op, want in (("release", True), ("energize", False), ("release", True)):
if s.port.request(op) != "ok":
fail("[port] %s was refused" % op)
if s.port.state()["released"] != want:
fail("[port] %s left released=%r" % (op, not want))
s.set_mode("gfcloud")
r = s.port.request("home")
s.set_mode("manual")
if r != "error:mode" or os.path.exists(os.path.join(s.workdir, RUNNER_MARK)):
fail("[port] home under gfcloud got %r" % r)
if s.port.request("home") != "ok":
fail("[port] home under manual was refused")
st = s.port.state()
if st["released"] or st["homed"] & 3 != 3 or st["state"] != "Idle":
fail("[port] not energized, homed and idle after the port's home: %r" % st)
if s.sender.count() != before:
fail("[port] the sender got %d responses for the port's commands"
% (s.sender.count() - before))
if s.sender.send("G0 X1") != "ok":
fail("[port] the sender's next line did not get ok")
print("PASS [port]: release, energize and home work through the port with a sender "
"connected, home is refused outside manual mode, and the sender's count is exact")
def test_restart_while_released(s):
"""A controller that dies under a released gantry is replaced by one that
takes the release over before it writes its first current: it comes up
locked, the hold currents are never written, and $ME still ends it."""
s.sender.wait_state("Idle")
if s.sender.send("$MD") != "ok":
fail("[restart] $MD was refused")
before = s.attr_writes()
marker = os.path.join(s.workdir, "motors.released")
if not os.path.exists(marker):
fail("[restart] the release left no marker in the state directory")
s.respawn(hard=True)
st = s.port.state()
if not st["released"] or st["state"] != "Alarm":
fail("[restart] the new controller did not take the release over: %r" % st)
started = s.attr_writes()[len(before):]
if started != [("0", "0")]:
fail("[restart] the new controller's start wrote %r to a released gantry, not 0 and 0" % started)
if refused(s, "$X") is None or refused(s, "$J=G91 X5 F600") is None:
fail("[restart] the new controller let a released machine be unlocked or jogged")
if s.sender.send("$ME") != "ok":
fail("[restart] $ME was refused after the restart")
st = s.port.state()
if st["released"] or st["state"] != "Idle" or s.attr_writes()[-1] != ("33", "5") or os.path.exists(marker):
fail("[restart] $ME did not energize and clear the marker: %r %r" % (st, s.attr_writes()[-1:]))
s.respawn(hard=False)
st = s.port.state()
if st["released"] or st["state"] != "Idle":
fail("[restart] a controller started with no marker came up released: %r" % st)
print("PASS [restart]: a controller killed under a released gantry was replaced by one that came "
"up locked and wrote 0 to X and Y at its start; $ME energized and cleared the marker; the next start was normal")
def test_stop_block_offsets(s):
"""manual_home_x and _y are the coordinate the stop blocks stand for: the
origin by default, never negative, and the wall the envelope starts at."""
s.sender.wait_state("Idle")
s.set_mode("manual", "manual_home_x = 12.5\nmanual_home_y = 8\n")
if s.sender.send("$H") != "ok":
fail("[offsets] $H under manual with offsets was refused")
st = s.port.state()
if abs(st["mpos"][0] - 12.5) > 0.01 or abs(st["mpos"][1] - 8.0) > 0.01:
fail("[offsets] the home declared %r, not (12.5, 8)" % st["mpos"][:2])
if s.sender.send("$J=G91 X5 Y5 F3000") != "ok":
fail("[offsets] a jog from the blocks onto the bed was refused")
s.sender.wait_state("Idle")
if s.sender.send("$J=G91 X-5 Y-5 F3000") != "ok":
fail("[offsets] the jog back to the blocks was refused")
s.sender.wait_state("Idle")
for line in ("$J=G91 X-1 F600", "$J=G91 Y-1 F600"):
r = refused(s, line)
if r != "error:15":
fail("[offsets] %s, behind the blocks, got %r, not error:15" % (line, r))
s.set_mode("manual", "manual_home_x = -10.3\nmanual_home_y = 1e9\n")
if s.sender.send("$H") != "ok":
fail("[offsets] $H with out-of-range offsets was refused")
st = s.port.state()
if st["mpos"][0] != 0 or st["mpos"][1] > 1000:
fail("[offsets] a negative or oversized offset was not clamped: %r" % st["mpos"][:2])
s.set_mode("manual")
print("PASS [offsets]: a manual home declared (12.5, 8), the envelope started at the blocks, "
"a negative offset became the origin and an oversized one the axis travel")
def test_offsets_belong_to_their_provider(s):
"""Both pairs set at once: a manual home declares manual_home_* and never
looks at gfcloud_home_*, and a camera home (a stand-in runner that exits 0)
declares gfcloud_home_* and never looks at manual_home_*. A camera home may
lie behind the origin, and the envelope reaches back to it."""
both = "manual_home_x = 7\nmanual_home_y = 9\ngfcloud_home_x = -4.5\ngfcloud_home_y = 3.25\n"
s.sender.wait_state("Idle")
s.set_mode("manual", both)
if s.sender.send("$H") != "ok":
fail("[own-offsets] the manual home was refused")
st = s.port.state()
if abs(st["mpos"][0] - 7.0) > 0.01 or abs(st["mpos"][1] - 9.0) > 0.01:
fail("[own-offsets] a manual home declared %r, not its own (7, 9)" % st["mpos"][:2])
s.set_mode("gfcloud", both)
if s.sender.send("$H", timeout=30.0) != "ok":
fail("[own-offsets] the camera home (stand-in runner) was refused")
s.sender.wait_state("Idle")
st = s.port.state()
if abs(st["mpos"][0] + 4.5) > 0.01 or abs(st["mpos"][1] - 3.25) > 0.01:
fail("[own-offsets] a camera home declared %r, not its own (-4.5, 3.25)" % st["mpos"][:2])
if s.sender.send("$J=G91 X2 F3000") != "ok":
fail("[own-offsets] a jog from a camera home behind the origin was refused")
s.sender.wait_state("Idle")
if s.sender.send("$J=G91 X-2 F3000") != "ok":
fail("[own-offsets] the jog back to the camera home was refused")
s.sender.wait_state("Idle")
r = refused(s, "$J=G91 X-1 F600")
if r != "error:15":
fail("[own-offsets] a jog behind the camera home got %r, not error:15" % r)
r = refused(s, "$J=G91 Y-4 F600")
if r != "error:15":
fail("[own-offsets] a jog to Y-0.75, off the bed, got %r, not error:15" % r)
s.set_mode("manual")
print("PASS [own-offsets]: with both pairs set, a manual home declared (7, 9) and a camera home "
"(-4.5, 3.25); the envelope reached back to the camera home on X and stopped at the origin on Y")
def main():
if not os.path.exists(BIN):
fail("no controller binary at %s" % BIN)
s = Session()
try:
test_manual_home(s)
test_home_refused_in_cycle(s)
test_release_refused_armed(s)
test_release_locks(s)
test_energize(s)
test_port_panel_ops(s)
test_restart_while_released(s)
test_stop_block_offsets(s)
test_offsets_belong_to_their_provider(s)
finally:
s.close()
print("ALL PASS")
if __name__ == "__main__":
main()