mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-28 01:01:12 -07:00
354 lines
14 KiB
Python
354 lines
14 KiB
Python
#!/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 that the Z envelope survives a settings write.
|
|
|
|
The Z soft limit belongs to the driver, not to $20. glowforge_homing.c
|
|
owns sys.work_envelope, sys.homed and sys.soft_limits for Z: Z is always
|
|
referenced, to the lens hall edge or to where the lens stands, and the
|
|
core knows neither. The core recomputes both masks from the settings and
|
|
drops Z when it does, so driver.c re-applies the envelope from the
|
|
settings-changed chain, after the rest of the chain has run.
|
|
|
|
The core drops Z in two places this harness drives:
|
|
|
|
$20 clears sys.soft_limits for every axis in the setter
|
|
$13x clears sys.homed for the axis as well, when the value changes
|
|
|
|
Runs the native grblHAL_glowforge binary in null-sink mode (no hardware,
|
|
no root) and drives it over TCP:
|
|
|
|
1. the envelope is in force from boot, with $20 off: an unreferenced Z
|
|
is collapsed to where the lens stands, so a Z move alarms
|
|
2. it is collapsed, not merely present: a move each way alarms
|
|
3. it is Z alone: X and Y follow $20 as the core intends, and stay free
|
|
4. $20=0, soft limits off for every axis, does not free Z
|
|
5. a $132 write, which un-homes Z and clears the mask in the setter,
|
|
does not free Z either
|
|
6. X and Y are still free after both writes
|
|
|
|
A referenced lens opens the envelope to the window the shared settings
|
|
hold, the same keys the daemon's wizards build their Z from. forgectrl
|
|
leaves the lens on its hall edge and a marker in the state directory
|
|
before the controller starts; the driver takes it as it loads its
|
|
settings and places the edge on its step grid:
|
|
|
|
7. the marker with the edge alone: Z sits at the edge's grid step, the
|
|
fallback window (10 below, 12 above) runs to its ends, and two
|
|
half-steps past either end alarms
|
|
8. the marker with the focus card's stops (14 below, 20 above): the
|
|
wider window runs to its ends, two half-steps past either alarms
|
|
9. a stop count out of range (41) falls back on its side alone
|
|
|
|
The X and Y soft limits are the driver's as well: off while the
|
|
position is not trusted (the core's $20 cannot be turned on with $22
|
|
off), on after a home, when the envelope is the bed. A scripted home
|
|
(homing_mode = gfcloud with a runner that exits 0, the host hook) sets
|
|
them:
|
|
|
|
10. after the home a program move past X or Y max raises ALARM:2
|
|
before any motion, a jog past it is refused with error 15, and a
|
|
move inside the bed runs; a stream fault or a controller start
|
|
drops them again with the reference
|
|
|
|
The binary keeps its settings in EEPROM.DAT in the working directory, so
|
|
each run starts from defaults in a temporary directory and leaves
|
|
nothing behind.
|
|
|
|
Usage: z_envelope_test.py [path-to-binary]
|
|
(default ./build-native/grblHAL_glowforge)
|
|
"""
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
BIN = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else "build-native/grblHAL_glowforge")
|
|
PORT = 2400
|
|
|
|
ALARM_SOFT_LIMIT = "ALARM:2"
|
|
|
|
|
|
def fail(msg):
|
|
print("FAIL: %s" % msg)
|
|
sys.exit(1)
|
|
|
|
|
|
class Session:
|
|
"""One null-sink controller process with its own settings store.
|
|
`conf` (key: value) is the shared config the driver reads; with
|
|
`referenced` the lens marker forgectrl leaves is in the state
|
|
directory, so the driver takes the lens reference at its start."""
|
|
|
|
def __init__(self, conf=None, referenced=False):
|
|
self.workdir = tempfile.mkdtemp(prefix="z-envelope-")
|
|
env = dict(os.environ, GF_STATE_DIR=self.workdir, FFLOG_STDERR="1")
|
|
for key in ("GFSINK", "GF_SWITCH_FILE", "GF_VERDICT_FILE"):
|
|
env.pop(key, None)
|
|
conf_path = os.path.join(self.workdir, "forgefirm.conf")
|
|
with open(conf_path, "w") as f:
|
|
for k, v in (conf or {}).items():
|
|
f.write("%s = %s\n" % (k, v))
|
|
env["GFHOME_CONF"] = conf_path
|
|
if referenced:
|
|
with open(os.path.join(self.workdir, "lens.home"), "w") as f:
|
|
f.write("edge 0 3 3\n")
|
|
self.proc = subprocess.Popen([BIN, "-p", str(PORT)],
|
|
cwd=self.workdir, env=env,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.PIPE)
|
|
self.sock = self.connect()
|
|
self.read(1.0) # banner and any boot messages
|
|
|
|
def connect(self):
|
|
for _ in range(50):
|
|
try:
|
|
return socket.create_connection(("127.0.0.1", PORT), timeout=1)
|
|
except OSError:
|
|
time.sleep(0.1)
|
|
err = b""
|
|
if self.proc.poll() is not None:
|
|
err = self.proc.stderr.read() or b""
|
|
fail("cannot connect to the controller (exit=%s)\n%s"
|
|
% (self.proc.poll(), err.decode(errors="replace")))
|
|
|
|
def read(self, timeout=0.8):
|
|
out = b""
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
self.sock.settimeout(max(0.05, end - time.time()))
|
|
try:
|
|
data = self.sock.recv(4096)
|
|
except (socket.timeout, OSError):
|
|
break
|
|
if not data:
|
|
break
|
|
out += data
|
|
return out.decode(errors="replace")
|
|
|
|
def send(self, line, timeout=1.0):
|
|
self.sock.sendall(b"?" if line == "?" else (line + "\n").encode())
|
|
return self.read(timeout)
|
|
|
|
def setting(self, key):
|
|
reply = self.send(key)
|
|
m = re.search(r"^%s=(\S+)" % re.escape(key), reply, re.M)
|
|
if not m:
|
|
fail("no value reported for %s: %r" % (key, reply))
|
|
return m.group(1)
|
|
|
|
def write_setting(self, assignment):
|
|
reply = self.send(assignment)
|
|
if "ok" not in reply:
|
|
fail("settings write %r refused: %r" % (assignment, reply))
|
|
|
|
def state(self):
|
|
m = re.search(r"<(\w+)", self.send("?"))
|
|
return m.group(1) if m else "?"
|
|
|
|
def mpos_z(self):
|
|
"""The reported machine Z (no work offset is set in a session)."""
|
|
m = re.search(r"[MW]Pos:(-?[\d.]+),(-?[\d.]+),(-?[\d.]+)", self.send("?"))
|
|
if not m:
|
|
fail("no position in the status report")
|
|
return float(m.group(3))
|
|
|
|
def wait_idle(self, timeout=15.0):
|
|
"""A move that ran plays out before the next line is judged: the
|
|
soft limit on a moving machine holds first and alarms after."""
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
if self.state() == "Idle":
|
|
return
|
|
time.sleep(0.1)
|
|
fail("the controller did not return to Idle")
|
|
|
|
def move(self, gcode):
|
|
"""Run one move. Returns True when the soft limit blocked it.
|
|
An alarm needs a reset and an unlock before the next move."""
|
|
reply = self.send(gcode, 1.5)
|
|
if ALARM_SOFT_LIMIT in reply:
|
|
self.recover()
|
|
return True
|
|
if "ok" not in reply:
|
|
fail("unexpected reply to %r: %r" % (gcode, reply))
|
|
return False
|
|
|
|
def recover(self):
|
|
self.sock.sendall(b"\x18")
|
|
self.read(1.0)
|
|
self.send("$X")
|
|
if self.state() != "Idle":
|
|
fail("controller did not return to Idle after an alarm")
|
|
|
|
def close(self):
|
|
try:
|
|
self.sock.close()
|
|
except OSError:
|
|
pass
|
|
self.proc.terminate()
|
|
try:
|
|
self.proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
self.proc.kill()
|
|
shutil.rmtree(self.workdir, ignore_errors=True)
|
|
|
|
|
|
def check(ok, ok_msg, fail_msg):
|
|
if not ok:
|
|
fail(fail_msg)
|
|
print("ok %s" % ok_msg)
|
|
|
|
|
|
def blocked_both_ways(s, when):
|
|
check(s.move("G0 Z1"), "Z+ blocked %s" % when, "a Z+ move ran %s" % when)
|
|
check(s.move("G0 Z-1"), "Z- blocked %s" % when, "a Z- move ran %s" % when)
|
|
|
|
|
|
def free_in_xy(s, when):
|
|
ran = not s.move("G0 X1") and not s.move("G0 Y1")
|
|
check(ran, "X and Y free %s" % when, "an X or Y move was blocked %s" % when)
|
|
|
|
|
|
def window_holds(s, steps, below, above, what):
|
|
"""The envelope of a referenced lens whose edge sits on grid step
|
|
`steps`: the ends of the reach (`below` and `above` half-steps from
|
|
the edge) run, two half-steps past either end alarms. The controller
|
|
keeps a half-step of slack beyond the reach, so one past is not
|
|
judged here; two is outside on every head."""
|
|
spm = float(s.setting("$102"))
|
|
top, bottom = (steps + above) / spm, (steps - below) / spm
|
|
s.wait_idle()
|
|
check(not s.move("G0 Z%.3f" % top), "%s: the top of the reach, Z %.3f, runs" % (what, top),
|
|
"%s: Z %.3f, the top of the reach, was blocked" % (what, top))
|
|
s.wait_idle()
|
|
check(not s.move("G0 Z%.3f" % bottom), "%s: the bottom of the reach, Z %.3f, runs" % (what, bottom),
|
|
"%s: Z %.3f, the bottom of the reach, was blocked" % (what, bottom))
|
|
over, under = (steps + above + 2) / spm, (steps - below - 2) / spm
|
|
s.wait_idle()
|
|
check(s.move("G0 Z%.3f" % over), "%s: two half-steps over the top, Z %.3f, alarms" % (what, over),
|
|
"%s: Z %.3f, two half-steps over the top, ran" % (what, over))
|
|
s.wait_idle()
|
|
check(s.move("G0 Z%.3f" % under), "%s: two half-steps under the bottom, Z %.3f, alarms" % (what, under),
|
|
"%s: Z %.3f, two half-steps under the bottom, ran" % (what, under))
|
|
|
|
|
|
def referenced_cases():
|
|
"""The lens referenced by forgectrl's marker: the envelope is the
|
|
window the shared settings hold, as the daemon's wizards read it."""
|
|
edge = 3.35
|
|
conf = {"lens_hall_edge_z_mm": "%.2f" % edge}
|
|
s = Session(conf=conf, referenced=True)
|
|
try:
|
|
spm = float(s.setting("$102"))
|
|
steps = int(round(edge * spm))
|
|
z = s.mpos_z()
|
|
check(abs(z - steps / spm) < 0.002,
|
|
"referenced: Z sits at the edge's grid step, %.3f" % z,
|
|
"referenced: Z reads %.3f, not the edge's grid step %.3f" % (z, steps / spm))
|
|
window_holds(s, steps, 10, 12, "the fallback window")
|
|
free_in_xy(s, "with the lens referenced")
|
|
finally:
|
|
s.close()
|
|
|
|
s = Session(conf=dict(conf, lens_stop_below_steps="14", lens_stop_above_steps="20"), referenced=True)
|
|
try:
|
|
window_holds(s, steps, 14, 20, "the focus card's window")
|
|
finally:
|
|
s.close()
|
|
|
|
s = Session(conf=dict(conf, lens_stop_below_steps="41", lens_stop_above_steps="20"), referenced=True)
|
|
try:
|
|
window_holds(s, steps, 10, 20, "a below count out of range")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def homed_cases():
|
|
"""After a scripted gfcloud home the bed is the X/Y envelope."""
|
|
conf = {"lens_hall_edge_z_mm": "3.35", "homing_mode": "gfcloud", "gfcloud_home_cmd": "true"}
|
|
s = Session(conf=conf, referenced=True)
|
|
try:
|
|
free_in_xy(s, "before the home")
|
|
x_travel = float(s.setting("$130"))
|
|
y_travel = float(s.setting("$131"))
|
|
reply = s.send("$H", 5.0)
|
|
if "ok" not in reply:
|
|
fail("the scripted home was refused: %r" % reply)
|
|
s.wait_idle()
|
|
st = s.send("?")
|
|
m = re.search(r"MPos:(-?[\d.]+),(-?[\d.]+)", st)
|
|
check(m and abs(float(m.group(1))) < 0.01 and abs(float(m.group(2))) < 0.01,
|
|
"homed: X and Y sit at the origin", "homed: X/Y are not at the origin: %r" % st)
|
|
check(s.move("G90 G0 X%.1f" % (x_travel + 5)), "homed: X past the bed alarms",
|
|
"homed: a move past X max ran")
|
|
s.wait_idle()
|
|
check(s.move("G90 G0 Y%.1f" % (y_travel + 5)), "homed: Y past the bed alarms",
|
|
"homed: a move past Y max ran")
|
|
s.wait_idle()
|
|
check(s.move("G90 G0 X-1"), "homed: X past the near edge alarms",
|
|
"homed: a move past X min ran")
|
|
s.wait_idle()
|
|
reply = s.send("$J=G91X%.1fF1200" % (x_travel + 5), 1.5)
|
|
check("error:15" in reply, "homed: a jog past the bed is refused with error 15",
|
|
"homed: a jog past the bed was not refused with error 15: %r" % reply)
|
|
# The core answers the line after an error with that error again
|
|
# until an empty line (or a $ command) clears it: the sender's
|
|
# acknowledgment at compatibility level 0.
|
|
s.send("", 1.0)
|
|
check(not s.move("G90 G0 X10 Y10"), "homed: a move inside the bed runs",
|
|
"homed: a move inside the bed was blocked")
|
|
s.wait_idle()
|
|
# A settings write does not drop them: the driver re-applies the
|
|
# mask after the core clears it.
|
|
s.write_setting("$20=0")
|
|
check(s.move("G90 G0 X%.1f" % (x_travel + 5)), "homed: X past the bed still alarms after a $20 write",
|
|
"homed: a $20 write freed X")
|
|
s.wait_idle()
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def main():
|
|
if not os.path.isfile(BIN):
|
|
fail("no controller binary at %s" % BIN)
|
|
s = Session()
|
|
try:
|
|
soft = s.setting("$20")
|
|
check(soft == "0", "$20=0 at boot, so the Z envelope is the driver's alone",
|
|
"$20 is %s, not 0: this harness needs the default" % soft)
|
|
|
|
blocked_both_ways(s, "at boot")
|
|
free_in_xy(s, "at boot")
|
|
|
|
# $20 clears sys.soft_limits for every axis inside the setter.
|
|
s.write_setting("$20=0")
|
|
blocked_both_ways(s, "after a $20 write")
|
|
|
|
# $13x clears sys.homed for the axis too, but only when the value
|
|
# changes, so move it and put it back. Both writes drop Z.
|
|
z_travel = s.setting("$132")
|
|
s.write_setting("$132=%.3f" % (float(z_travel) + 1.0))
|
|
blocked_both_ways(s, "after a $132 write")
|
|
s.write_setting("$132=%s" % z_travel)
|
|
blocked_both_ways(s, "after $132 is restored")
|
|
check(s.setting("$132") == z_travel, "$132 restored to %s" % z_travel,
|
|
"$132 did not go back to %s" % z_travel)
|
|
|
|
free_in_xy(s, "after the settings writes")
|
|
finally:
|
|
s.close()
|
|
referenced_cases()
|
|
homed_cases()
|
|
print("PASS z_envelope_test")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|