forgefirm/scripts/bench/xy_mode_test.py

#!/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()