#!/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 XY microstep mode wiring. The XY microstep mode is one number in the shared config, xy_microsteps (8, 16 or 32), and the driver derives three things from it at start and never takes them typed: $100/$101, the machine tick, and the kernel stop ramp. This harness drives the native grblHAL_glowforge binary in null-sink mode (no hardware, no root) over TCP, one process per config: 1. no key: x32, $100/$101 = 213.333, and $110 stands 2. xy_microsteps = 16: 106.667; a typed $100 is overwritten on the spot; $110 is left alone under a tick that carries it 3. GFSINK_RATE lowered under the mode's tick: $110/$111 are held at the feed the tick carries, one step per tick per axis Each mode's derivations, and a value that is not a mode, are the host unit test's (xy_scale_test.c in grblHAL-glowforge). 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: xy_mode_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 = 2401 def fail(msg): print("FAIL: %s" % msg) sys.exit(1) class Session: """One null-sink controller process with its own settings store and its own shared config file.""" def __init__(self, conf=None, env_extra=None): self.workdir = tempfile.mkdtemp(prefix="xy-mode-") env = dict(os.environ, GF_STATE_DIR=self.workdir, FFLOG_STDERR="1", FFLOG_LEVEL="info") for key in ("GFSINK", "GFSINK_RATE", "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: f.write(conf or "") env["GFHOME_CONF"] = conf_path env.update(env_extra or {}) 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((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 float(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 close(self): """Stop the process and return everything it logged.""" try: self.sock.close() except OSError: pass self.proc.terminate() try: _out, err = self.proc.communicate(timeout=5) except subprocess.TimeoutExpired: self.proc.kill() _out, err = self.proc.communicate() shutil.rmtree(self.workdir, ignore_errors=True) return (err or b"").decode(errors="replace") def check(ok, ok_msg, fail_msg): if not ok: fail(fail_msg) print("ok %s" % ok_msg) def near(got, want, tol=0.0005): return abs(got - want) <= tol def expect_scale(s, mode, spm): for key in ("$100", "$101"): got = s.setting(key) check(near(got, spm), "%s=%.3f at x%d" % (key, got, mode), "%s is %.3f at x%d, want %.3f" % (key, got, mode, spm)) def case_default(): s = Session() try: expect_scale(s, 32, 213.333) rate = s.setting("$110") check(near(rate, 12000.0), "$110=%.0f stands at x32" % rate, "$110 is %.3f at x32, want 12000" % rate) finally: s.close() def case_x16(): s = Session("xy_microsteps = 16\n") try: expect_scale(s, 16, 106.667) # A typed $100 is accepted and overwritten on the spot: the scale # is the mode's, never the sender's. s.write_setting("$100=53.333") got = s.setting("$100") check(near(got, 106.667), "a typed $100 is back at %.3f" % got, "a typed $100 stuck at %.3f under x16" % got) rate = s.setting("$110") check(near(rate, 12000.0), "$110=%.0f stands under a tick that carries it" % rate, "$110 is %.3f under x16, want 12000" % rate) finally: s.close() def case_low_tick(): # 1000 Hz carries 1000 steps/s per axis: 562.5 mm/min at 106.667 # steps/mm. $110/$111 are held there, and a typed value above it # comes back held. s = Session("xy_microsteps = 16\n", {"GFSINK_RATE": "1000"}) try: expect_scale(s, 16, 106.667) for key in ("$110", "$111"): got = s.setting(key) check(near(got, 562.5, 0.05), "%s held at %.3f under a 1000 Hz tick" % (key, got), "%s is %.3f under a 1000 Hz tick, want 562.5" % (key, got)) s.write_setting("$110=6000") got = s.setting("$110") check(near(got, 562.5, 0.05), "a typed $110=6000 comes back held at %.3f" % got, "a typed $110=6000 stuck at %.3f under a 1000 Hz tick" % got) finally: s.close() def main(): if not os.path.isfile(BIN): fail("no controller binary at %s" % BIN) case_default() case_x16() case_low_tick() print("PASS xy_mode_test") if __name__ == "__main__": main()