mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-29 09:31:17 -07:00
ScottW514
b494bbecb5
homing.manual runs two homes. With manual_home_x and _y unset, $H plays no byte and declares X0 Y0 (the checks it had). With 5 and 3, $H declares the head at minus them and jogs it to X0 Y0 before it answers: the sender is told the declared position, the kernel plays the move, Grbl and forgectrl read X0 Y0, and a jog behind the origin draws error:15. The test restores homing_mode and both offsets as found, and returns the head by what the machine read, so a move cut short still comes home. The change stays inside the test's own function. manual_home_test.py: case 7 is the offset's new meaning (declared at minus the offset, the move dark under a modal M3, the port's home, the clamps); a new case cancels the move with a jog cancel and expects it to stop short with $H saying where; case 8 reads the declared position from the sender's message. The bench README says so. Proof: manual_home_test.py ALL PASS against the driver change (ten cases), and its cancel case fails against a build with the driver's cancel hook left out. On the bench reference, fx14 with the driver change and this suite file in place of the image's, homing.manual PASS in campaign c-20260926000059-e731: with no offset no byte and X0 Y0; with 5 and 3 the sender was told X-5.002 Y-3.000, the kernel played 1067 and 640 steps, Grbl and forgectrl read X0 Y0, error:15 behind the origin; settings restored and the baseline clean. The forgetest unit tests pass. Acceptance: homing.manual is the test; its covers are unchanged (the driver's src/**, forgectrl's status and main.c).
543 lines
25 KiB
Python
543 lines
25 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 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: how far in front of the stop blocks
|
|
the origin lies, never negative. With an offset the head at the blocks
|
|
is declared at minus it and jogs to the origin, dark, before $H
|
|
answers, through the sender and through the port; the envelope starts
|
|
at the origin, and an out-of-range offset is clamped; a jog cancel
|
|
ends the move short of the origin and $H says where it stopped
|
|
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 minus 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 re
|
|
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 fire_ticks(self):
|
|
"""FIRE ticks shipped so far."""
|
|
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 & 0x10)
|
|
|
|
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 declared(s):
|
|
"""The position the last manual home with an offset told the sender the
|
|
head stood at, on the step grid, or None."""
|
|
with s.sender.lock:
|
|
lines = list(s.sender.other)
|
|
for line in reversed(lines):
|
|
m = re.search(r"Manual home: the head is at X(-?[0-9.]+) Y(-?[0-9.]+); moving it to the origin", line)
|
|
if m:
|
|
return float(m.group(1)), float(m.group(2))
|
|
return None
|
|
|
|
|
|
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 how far in front of the stop blocks the origin
|
|
lies, never negative: the head at the blocks is declared at minus them and
|
|
jogs to the origin, dark, before $H answers, and the envelope starts at
|
|
the origin."""
|
|
s.sender.wait_state("Idle")
|
|
s.set_mode("manual", "manual_home_x = 12.5\nmanual_home_y = 8\n")
|
|
s.sender.send("M3 S100") # a modal beam the jog must not carry
|
|
ticks, fire = s.motion_ticks(), s.fire_ticks()
|
|
if s.sender.send("$H") != "ok":
|
|
fail("[offsets] $H under manual with offsets was refused")
|
|
st = s.port.state() # $H answers once the head has stopped
|
|
s.sender.send("M5")
|
|
if st["state"] != "Idle" or st["mpos"][0] != 0 or st["mpos"][1] != 0:
|
|
fail("[offsets] the head was not idle at the origin when $H answered: %r" % st)
|
|
if st["homed"] & 3 != 3:
|
|
fail("[offsets] X and Y are not marked homed: %r" % st)
|
|
d = declared(s)
|
|
if d is None or abs(d[0] + 12.5) > 0.01 or abs(d[1] + 8.0) > 0.01:
|
|
fail("[offsets] the sender was told the head stood at %r, not (-12.5, -8)" % (d,))
|
|
if s.motion_ticks() == ticks:
|
|
fail("[offsets] the move to the origin shipped nothing")
|
|
if s.fire_ticks() != fire:
|
|
fail("[offsets] the move to the origin shipped %d FIRE ticks" % (s.fire_ticks() - fire))
|
|
if s.sender.send("$J=G91 X5 Y5 F3000") != "ok":
|
|
fail("[offsets] a jog from the origin 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 origin 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 origin, 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", timeout=30.0) != "ok":
|
|
fail("[offsets] $H with out-of-range offsets was refused")
|
|
st = s.port.state()
|
|
if st["mpos"][0] != 0 or st["mpos"][1] != 0:
|
|
fail("[offsets] the head did not end at the origin: %r" % st["mpos"][:2])
|
|
d = declared(s)
|
|
if d is None or d[0] != 0 or abs(d[1] + 279.0) > 0.01:
|
|
fail("[offsets] a negative offset and an oversized one declared %r, not (0, -279)" % (d,))
|
|
# The panel's Set home here: the port answers once the head has stopped,
|
|
# and the sender's count stays exact.
|
|
s.set_mode("manual", "manual_home_x = 12.5\nmanual_home_y = 8\n")
|
|
s.quiet()
|
|
before = s.sender.count()
|
|
if s.port.request("home") != "ok":
|
|
fail("[offsets] the port's home with offsets was refused")
|
|
st = s.port.state()
|
|
if st["state"] != "Idle" or st["mpos"][0] != 0 or st["mpos"][1] != 0:
|
|
fail("[offsets] the head was not idle at the origin when the port's home answered: %r" % st)
|
|
if s.sender.count() != before:
|
|
fail("[offsets] the sender got %d responses for the port's home" % (s.sender.count() - before))
|
|
s.set_mode("manual")
|
|
print("PASS [offsets]: a manual home declared (-12.5, -8) and jogged to the origin before $H "
|
|
"answered, dark under a modal M3; the envelope started at the origin; a negative offset "
|
|
"became 0 and an oversized one the axis travel; the port's home answered at the origin")
|
|
|
|
|
|
def test_origin_jog_cancel(s):
|
|
"""A jog cancel during a manual home's move to the origin ends the move at
|
|
once, as it ends any jog, though the core's main loop is waiting for $H:
|
|
$H answers with the head short of the origin, and the sender is told
|
|
where it stopped."""
|
|
s.sender.wait_state("Idle")
|
|
s.set_mode("manual", "manual_home_y = 250\n")
|
|
n = s.sender.count()
|
|
s.sender.sock.sendall(("$H" + s.sender.eol).encode())
|
|
time.sleep(0.4) # into a move of about 1.6 s
|
|
s.sender.realtime(b"\x85")
|
|
end = time.time() + 5.0
|
|
while time.time() < end and s.sender.count() == n:
|
|
time.sleep(0.01)
|
|
with s.sender.lock:
|
|
r = s.sender.responses[n] if len(s.sender.responses) > n else None
|
|
st = s.port.state()
|
|
s.set_mode("manual")
|
|
if r != "ok":
|
|
fail("[origin-cancel] $H got %r" % r)
|
|
if st["state"] != "Idle" or st["mpos"][1] > -10.0:
|
|
fail("[origin-cancel] the jog cancel did not stop the move short of the origin: %r" % st)
|
|
if not s.sender.saw("Manual home: the move to the origin stopped at"):
|
|
fail("[origin-cancel] the sender was not told where the move stopped")
|
|
print("PASS [origin-cancel]: a jog cancel 0.4 s into the move to the origin stopped it at Y%.1f; "
|
|
"$H answered ok and said where it stopped" % st["mpos"][1])
|
|
|
|
|
|
def test_offsets_belong_to_their_provider(s):
|
|
"""Both pairs set at once: a manual home declares minus 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")
|
|
d = declared(s)
|
|
if d is None or abs(d[0] + 7.0) > 0.01 or abs(d[1] + 9.0) > 0.01:
|
|
fail("[own-offsets] a manual home declared %r, not its own (-7, -9)" % (d,))
|
|
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_origin_jog_cancel(s)
|
|
test_offsets_belong_to_their_provider(s)
|
|
finally:
|
|
s.close()
|
|
print("ALL PASS")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|