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

forgetest: the fixture spaces its presses; an unattended refusal is an error; the baseline resets out of a hold

Browse Source 
The first campaign with the bench actuator wired failed
motion.button-hold-resume on the tool, not the machine: the second press
was asked while the first 200 ms pulse was still on, the fixture answered
409, the runner handed the step to an operator who was not in the room,
and the post pass could not jog a controller left in Hold.

- fixture.py: a press waits for the last pulse to end (the fixture's
  pulse_ms) plus a 300 ms release, so the controller sees the edge; a
  409 for a pulse in progress is waited out against button_pulsing and
  retried once.
- runner.py: in an unattended run a fixture refusal ends the test at
  once as ERROR naming the refusal; the operator fallback stays for
  attended runs.
- baseline.py: a controller in Hold or Door gets a soft reset before the
  return jog, position kept.
- tests: the fake fixture refuses a press while one is in progress and
  reports button_pulsing; FakeGrbl records ^X and can land a reset in a
  chosen state; five new tests.
- docs: ACCEPTANCE.md fixture rules, fixture/README.md tool's side.

No catalog consequence: tool-side change, no covers map moves.
Bench: campaign c-20260824174545-0bdc 25/25 with every action by the
fixture; the hold reset proven by a dry drill.
This commit is contained in:
ScottW514
2026-08-24 14:02:59 -04:00
parent e99888753f
commit 00ded74189
8 changed files with 147 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
forgetest/forgetest/baseline.py
+13
View File
@@ -550,6 +550,19 @@ class Baseline:
try:
with hw.Grbl() as g:
rep = g.status_report()
if rep["state"].startswith(("Hold", "Door")):
# a job left held (a pause test that failed there) refuses
# a jog: the soft reset ends it where it stopped, position
# kept, and hands the controller back
held = rep["state"]
g.realtime(0x18)
deadline = time.time() + 5
while time.time() < deadline:
rep = g.status_report()
if not rep["state"].startswith(("Hold", "Door")):
break
time.sleep(0.2)
self.log("controller reset out of %s: now %s" % (held, rep["state"]))
if rep["state"].startswith("Alarm"):
g.command("$X")
g.command("$J=G91X%.3fY%.3fF1200" % (-dx, -dy))
forgetest/forgetest/fixture.py
+28 -1
View File
@@ -33,6 +33,8 @@ CHANNELS = ("lid", "interlock", "button")
MDNS_GROUP = "224.0.0.251"
MDNS_PORT = 5353
RESOLVE_TTL_S = 300.0 # a resolved address is trusted this long
BUTTON_GAP_S = 0.3 # the release between two presses: long enough for the controller to see it
PULSE_WAIT_S = 1.5 # a pulse in progress ends within the firmware's 500 ms clamp
class FixtureError(Exception):
@@ -226,6 +228,7 @@ class Fixture:
self._resolved_at = time.time() if self.ip_override else 0.0
self._lock = threading.Lock()
self.last_state = None
self._pulse_end = 0.0 # when this client's last press ends (its pulse_ms)
# -- address -------------------------------------------------------
def address(self, refresh=False):
@@ -293,7 +296,7 @@ class Fixture:
if channel == "button":
if state != "press":
raise FixtureError("the button is only ever pressed")
st, body = self._request("POST", "/button", {})
st, body = self._press()
else:
if state not in ("open", "close"):
raise FixtureError("%s: unknown state %r" % (channel, state))
@@ -304,6 +307,30 @@ class Fixture:
self.last_state = body
return body
def _press(self):
"""One press, spaced from the last one. The controller detects
the button's rising edge, so a press must follow a release it has
seen: the next press waits for the last pulse to end (the
fixture's own pulse_ms) plus BUTTON_GAP_S. A 409 for a pulse in
progress (a press this client did not time) is waited out once
against the fixture's state, then retried."""
wait = self._pulse_end + BUTTON_GAP_S - time.time()
if wait > 0:
time.sleep(wait)
st, body = self._request("POST", "/button", {})
if st == 409 and "in progress" in str(body.get("error", "")):
deadline = time.time() + PULSE_WAIT_S
while time.time() < deadline:
time.sleep(0.1)
s, b = self._request("GET", "/")
if s == 200 and not b.get("button_pulsing"):
break
time.sleep(BUTTON_GAP_S)
st, body = self._request("POST", "/button", {})
if st == 200:
self._pulse_end = time.time() + float(body.get("pulse_ms") or 500) / 1000.0
return st, body
def release(self):
st, body = self._request("POST", "/release", {})
if st != 200:
forgetest/forgetest/runner.py
+10 -1
View File
@@ -338,11 +338,16 @@ class Context:
try:
fixture.act(channel, state)
except _fixture.FixtureError as e:
rec["fixture_error"] = str(e)
if self.run.unattended:
# nobody is in the room to do it instead: the run ends
# here, as the harness's failure, not the machine's
self.log("ACT %s %s: fixture failed (%s) - unattended, nobody to ask", channel, state, e)
raise
# the box did not do it: the operator is asked instead,
# and the record says so
self.log("ACT %s %s: fixture failed (%s) - asking the operator", channel, state, e)
rec["by"] = "operator"
rec["fixture_error"] = str(e)
self.notice(wording)
else:
self.notice(wording)
@@ -900,6 +905,10 @@ class Runner:
result, message = _campaign.ABORTED, str(e) or "aborted"
except Failed as e:
result, message = _campaign.FAIL, str(e)
except _fixture.FixtureError as e:
# the bench actuator could not perform a step: the test was
# not judged, and the machine is not the one at fault
result, message = _campaign.ERROR, "the fixture could not perform a step: %s" % e
except Exception as e: # noqa: BLE001 - an erroring test is a failed test
result, message = _campaign.ERROR, "%s: %s" % (type(e).__name__, e)
run.log(traceback.format_exc().rstrip())
forgetest/tests/helpers.py
+8
View File
@@ -184,6 +184,8 @@ class FakeGrbl:
self.mpos = [0.0, 0.0, 0.0]
self.sent = []
self.extra = b"" # text pushed to the client on the next poll
self.reset_to = None # the state a soft reset (^X) lands in, when set
self.on_command = None # called with every command line received
self._sock = _socket.socket()
self._sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1)
self._sock.bind(("127.0.0.1", 0))
@@ -228,10 +230,16 @@ class FakeGrbl:
buf = buf[1:]
out += ("<%s|MPos:%.3f,%.3f,%.3f|FS:0,0>\r\n" % (self.state, *self.mpos)).encode()
elif buf[0] in (0x18, 0x85, 0x7E, 0x21):
if buf[0] == 0x18:
self.sent.append("^X")
if self.reset_to:
self.state = self.reset_to
buf = buf[1:]
elif b"\n" in buf:
line, buf = buf.split(b"\n", 1)
self.sent.append(line.strip().decode("utf-8", "replace"))
if self.on_command:
self.on_command(self.sent[-1])
out += b"ok\r\n"
else:
break
forgetest/tests/test_baseline.py
+27
View File
@@ -119,6 +119,33 @@ class BaselineTests(unittest.TestCase):
self.assertTrue(items["position"].action.startswith("unrestorable"), items["position"].action)
self.assertEqual(items["position"].found, [1000, 0, 0])
def test_a_held_controller_is_reset_before_the_return_jog(self):
# a pause test that failed while held leaves the controller in
# Hold, which refuses a jog: the baseline resets out of it first
import helpers
fc = helpers.FakeForgectrl().start()
dev = helpers.FakeGrbl().start()
try:
dev.state = "Hold:0"
dev.reset_to = "Idle"
dev.on_command = lambda line: self._pos(0, 0, 0) if line.startswith("$J=") else None
b = self.bl()
cap = b.capture()
self._pos(221, 0, 0) # 4.144 mm into the held move
left = b.enforce("post", captured=cap)
items = {x.item: x for x in left}
self.assertTrue(items["position"].action.startswith("restored"), items["position"].action)
self.assertEqual(dev.sent[0], "^X")
jogs = [l for l in dev.sent if l.startswith("$J=")]
self.assertEqual(len(jogs), 1)
self.assertIn("X-4.144", jogs[0])
self.assertTrue(any("reset out of Hold:0" in l for l in self.lines), self.lines)
finally:
dev.stop()
fc.stop()
for k in ("GRBL_HOST", "GRBL_PORT"):
os.environ.pop(k, None)
def _pos_bytes(self, x, y, z, processed, total):
with open(self.sysfs + "cnc/position", "wb") as f:
f.write(struct.pack("<3i2I", x, y, z, processed, total))
forgetest/tests/test_fixture.py
+53 -1
View File
@@ -37,6 +37,7 @@ class FakeFixture:
self.state = {"lid": "closed", "interlock": "closed", "button": "idle"}
self.button_enabled = button_enabled
self.calls = []
self.presses = 0 # presses the device performed
srv = self
class H(BaseHTTPRequestHandler):
@@ -54,7 +55,8 @@ class FakeFixture:
def _state(self):
return {"device": "forgefixture", "hostname": "forgefixture", "version": "1.0.0",
"idf": "v5.5.5", "uptime_s": 12, "channels": dict(srv.state),
"button_enabled": srv.button_enabled, "button_pulsing": False,
"button_enabled": srv.button_enabled,
"button_pulsing": srv.state["button"] == "pressed",
"wifi": {"connected": True, "ip": "127.0.0.1", "rssi": -50}}
def _auth(self):
@@ -84,7 +86,10 @@ class FakeFixture:
if self.path == "/button":
if not srv.button_enabled:
return self._json(409, {"error": "button disabled: the enable jumper is out"})
if srv.state["button"] == "pressed":
return self._json(409, {"error": "a button pulse is in progress"})
srv.state["button"] = "pressed"
srv.presses += 1
threading.Timer(0.05, lambda: srv.state.__setitem__("button", "idle")).start()
s = self._state()
s["pulse_ms"] = 200
@@ -217,6 +222,37 @@ class ClientTests(unittest.TestCase):
f.act("button", "press")
self.assertIn("jumper", str(cm.exception))
def test_two_presses_are_spaced_so_the_controller_sees_the_release(self):
f = self.client()
f.status()
t0 = time.time()
f.act("button", "press")
f.act("button", "press")
took = time.time() - t0
posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")]
self.assertEqual(len(posts), 2) # no 409 round trip was needed
self.assertEqual(self.dev.presses, 2)
# the second waited for the first pulse (200 ms as reported) and the gap
self.assertGreaterEqual(took, 0.2 + fx.BUTTON_GAP_S - 0.05)
def test_a_pulse_in_progress_is_waited_out_then_retried(self):
f = self.client()
f.status()
self.dev.state["button"] = "pressed" # a press this client did not time
threading.Timer(0.3, lambda: self.dev.state.__setitem__("button", "idle")).start()
f.act("button", "press")
posts = [c for c in self.dev.calls if c[:2] == ("POST", "/button")]
self.assertEqual(len(posts), 2) # the 409, then the press
self.assertEqual(self.dev.presses, 1)
def test_a_pulse_that_never_ends_is_the_fixtures_error(self):
f = self.client()
f.status()
self.dev.state["button"] = "pressed"
with self.assertRaises(fx.FixtureError) as cm:
f.act("button", "press")
self.assertIn("in progress", str(cm.exception))
def test_a_wrong_key_is_refused(self):
f = self.client(key="wrong")
with self.assertRaises(fx.FixtureError) as cm:
@@ -426,6 +462,22 @@ class RoutingTests(unittest.TestCase):
self.assertIn("the box is off", rec["fixture_error"])
self.assertTrue(seen and "lid" in seen[0].lower())
def test_a_failing_box_in_an_unattended_run_ends_the_test_as_an_error(self):
# nobody is in the room: the run must not wait ACT_TIMEOUT_S for a
# hand that is not there; it ends at once, as the harness's error
self.stub.fail = True
t0 = time.time()
b, order = self.run_queue("unattended")
results = {x["test"]: x["result"] for x in b["done"]}
self.assertEqual(results["r.lid"], "ERROR")
self.assertLess(time.time() - t0, 20)
rec = self.last_result("r.lid")
self.assertIn("the fixture could not perform a step", rec["message"])
self.assertIn("the box is off", rec["message"])
act = rec["evidence"]["actions"][0]
self.assertEqual((act["by"], act["fixture_error"]), ("fixture", "the box is off"))
self.assertFalse(any("asking the operator" in l for l in rec["log"]), rec["log"])
def test_what_the_box_still_holds_is_released_after_a_run(self):
def holds(ctx):
ctx.act("lid", "open") # and never closes it