From 1191389a2bddd7113baa7c1d52a063173751262b Mon Sep 17 00:00:00 2001 From: ScottW514 Date: Mon, 12 Oct 2026 21:51:45 -0400 Subject: [PATCH] forgetest: no press is ever held; the arm notice goes down at the press - The presence press at the Ready gate polled forgectrl's /status over HTTPS every 0.25 s plus the request's own time, so a short press fell between two reads and the operator held the button for seconds. It is read as an edge from the switches' input events (hw.ButtonEdges, no grab), opened before the notice goes up: a press of any length counts. Events read and not yet looked at are kept, so a short press whose release arrives in the same read gives the release to the next wait. With no switches device (the host harnesses) the status is polled at 50 ms. - An operator's arm notice (arm_press, press_now) stood until the test's next step, over a press already taken. A watcher takes it down the moment the controller puts the button out at the press; a notice put up meanwhile is left alone. Proof: host tests: a press and release written back to back on a FIFO standing in for the switches' device pass the Ready gate in well under a second with the status never showing the button down (the first version dropped the release with the press and waited 10 s: the test caught it); the arm notice stays while the button is dark or lit and goes down when it goes out, and a notice put up meanwhile stays. The fixture (35), setup (87), manifest (26) and operator (22) host tests pass. Acceptance: these are the harness's own paths; every live and operator test with a press runs through them. --- forgetest/forgetest/hw.py | 86 +++++++++++++++++++++++++++++++++ forgetest/forgetest/runner.py | 58 ++++++++++++++++++++-- forgetest/tests/test_fixture.py | 67 +++++++++++++++++++++++++ 3 files changed, 206 insertions(+), 5 deletions(-) diff --git a/forgetest/forgetest/hw.py b/forgetest/forgetest/hw.py index b8e3633..56f0bff 100644 --- a/forgetest/forgetest/hw.py +++ b/forgetest/forgetest/hw.py @@ -14,9 +14,12 @@ exercised against a mock on a host: GRBL_HOST / GRBL_PORT default 127.0.0.1 / 23 FORGETEST_INITD default /etc/init.d """ +import glob import json import os +import select import socket +import struct import subprocess import threading import time @@ -310,6 +313,89 @@ def button_lit(): return any(v > 0 for v in vals) +# ----------------------------------------------------------- the button + +EV_SW = 0x05 +SW_BUTTON = 2 # the button's EV_SW code (bit 2 of the switch word) + + +def switches_device(): + """The machine's switches as forgectrl and the controller find them: + GF_SWITCH_DEV, else the input device the device tree names + "switches". None when there is none.""" + dev = os.environ.get("GF_SWITCH_DEV") + if dev: + return dev + for d in sorted(glob.glob("/sys/class/input/event*")): + try: + with open(os.path.join(d, "device", "name")) as f: + if f.read().strip() == "switches": + return "/dev/input/" + os.path.basename(d) + except OSError: + continue + return None + + +class ButtonEdges: + """The button as edges, from the switches' input events. A press is an + event however short it is, so nothing has to be held long enough for a + poll to land on it: an operator's press is never longer than a quarter + second. Open it before the press is asked for. It reads without a + grab, so forgectrl and the controller see the switches as always.""" + + EVENT = struct.Struct("llHHi") # struct input_event: time, type, code, value + + def __init__(self): + self.fd = None + self.pending = b"" # events read and not yet looked at + dev = switches_device() + if dev: + try: + self.fd = os.open(dev, os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0)) + except OSError: + self.fd = None + + @property + def available(self): + return self.fd is not None + + def wait(self, value, timeout, checkpoint=None): + """The seconds until the button's next edge to `value` (1 pressed, + 0 released), or None at the timeout. checkpoint() is called about + every 0.1 s (an abort raises out of it).""" + size = self.EVENT.size + t0 = time.time() + while True: + # The events already read come first: a short press arrives + # with its release in one read, and the release is the next + # wait's. + while len(self.pending) >= size: + _sec, _usec, typ, code, val = self.EVENT.unpack_from(self.pending, 0) + self.pending = self.pending[size:] + if typ == EV_SW and code == SW_BUTTON and val == value: + return time.time() - t0 + if time.time() - t0 >= timeout: + return None + if checkpoint: + checkpoint() + ready, _, _ = select.select([self.fd], [], [], 0.1) + if not ready: + continue + try: + data = os.read(self.fd, size * 64) + except BlockingIOError: + continue + if not data: + time.sleep(0.02) # a FIFO with no writer (the host tests) reads empty + continue + self.pending += data + + def close(self): + if self.fd is not None: + os.close(self.fd) + self.fd = None + + # --------------------------------------------------------------- init.d def initd(service, action, timeout=60): diff --git a/forgetest/forgetest/runner.py b/forgetest/forgetest/runner.py index 86d10ef..906996c 100644 --- a/forgetest/forgetest/runner.py +++ b/forgetest/forgetest/runner.py @@ -125,6 +125,8 @@ ACT_TIMEOUT_S = 180 # machine by pressing its button. Generous: they may be setting up the # scrap, and nothing fires until it lands. PRESENCE_TIMEOUT_S = 600 +ARM_LIT_WAIT_S = 1800 # an arm can wait minutes (a coolant settle) before it lights +ARM_PRESS_WAIT_S = 3600 # the controller's own button timeout, at most class Run: @@ -316,17 +318,31 @@ class Context: fixture = getattr(self.runner, "fixture", None) if self.runner is not None else None if fixture is not None and fixture.covers("button"): + # The press is read as an edge from the switches' events, opened + # before the notice goes up: a press of any length counts, and + # nobody holds the button for a poll to land on it. + edges = hw.ButtonEdges() self.notice(text + " Then press the button on the machine to start.") self.log("READY: waiting for the operator's press on the machine " "(the bench then does every press in this test)") - pressed = self.wait_for(lambda: self.switch("button") is True, PRESENCE_TIMEOUT_S) - self.clear_notice() + try: + if edges.available: + pressed = edges.wait(1, PRESENCE_TIMEOUT_S, self.checkpoint) + # The release, so this press is never read as the arm + # press or a pause toggle; a short press has queued it. + if pressed is not None: + edges.wait(0, 10, self.checkpoint) + else: + pressed = self.wait_for(lambda: self.switch("button") is True, PRESENCE_TIMEOUT_S, + poll=0.05) + if pressed is not None: + self.wait_for(lambda: self.switch("button") is False, 10, poll=0.05) + finally: + edges.close() + self.clear_notice() if pressed is None: raise Failed("no press on the machine within %d s: %s" % (PRESENCE_TIMEOUT_S, text)) - # Let the button come back up, so this press is never read as - # the arm press or a pause toggle. - self.wait_for(lambda: self.switch("button") is False, 10) self.run.fixture_takeover = True self.evidence.setdefault("actions", []).append( {"channel": "button", "state": "presence", "by": "operator", @@ -461,6 +477,7 @@ class Context: self.log("ARM: WARNING the bench actuator took this test's presses and is now " "gone - asking the operator for the arm press") self.notice(text) + self._clear_when_taken() return False try: fixture.act("button", "press") @@ -473,6 +490,36 @@ class Context: self.log("ARM press by the fixture, at the machine's own press prompt") return True + def _clear_when_taken(self, lit_timeout=ARM_LIT_WAIT_S): + """Take the arm notice down the moment the press is taken: the + controller lights the button white for its arm wait and puts it out + at the press. Without this the notice stood until the test's next + step, and an operator who still saw it pressed again, which pauses + the job. A notice replaced meanwhile is left alone.""" + with self.run._lock: + nid = (self.run.notice or {}).get("id") + if nid is None: + return + + def watch(): + t0 = time.time() + while hw.button_lit() is not True: + if time.time() - t0 > lit_timeout: + return + time.sleep(0.05) + while hw.button_lit() is not False: + if time.time() - t0 > lit_timeout + ARM_PRESS_WAIT_S: + return + time.sleep(0.05) + with self.run._lock: + same = (self.run.notice or {}).get("id") == nid + if same: + self.run.notice = None + if same: + self.log("ARM: the button went out at the press; the notice is down") + + threading.Thread(target=watch, daemon=True, name="forgetest-arm-notice").start() + def arm_press(self, text="The button lights white: press it to arm. The machine fires after your press.", lit_timeout=60): """The arm cue of a live test. A person's press by default: a @@ -499,6 +546,7 @@ class Context: self.log("ARM: WARNING the bench actuator took this test's presses and is now " "gone - asking the operator for the arm press") self.notice(text) + self._clear_when_taken() return False self.log("ARM: the fixture presses when the button lights (the bench's arm_press opt-in)") diff --git a/forgetest/tests/test_fixture.py b/forgetest/tests/test_fixture.py index 50c1f4a..5ddf5fc 100644 --- a/forgetest/tests/test_fixture.py +++ b/forgetest/tests/test_fixture.py @@ -734,6 +734,73 @@ class RoutingTests(unittest.TestCase): finally: runner_mod.hw.button_lit = saved + @unittest.skipUnless(hasattr(os, "mkfifo"), "needs a FIFO for the switches' event device") + def test_ready_reads_a_short_press_as_an_edge(self): + """The presence press is the switches' input events, so a press + far shorter than a quarter second counts: the press and release + land back to back, and the status the old poll read never shows + the button down at all.""" + dev = os.path.join(self.tmp, "switches") + os.mkfifo(dev) + os.environ["GF_SWITCH_DEV"] = dev + ev = runner_mod.hw.ButtonEdges.EVENT + + def press(): + with open(dev, "wb") as w: + w.write(ev.pack(0, 0, runner_mod.hw.EV_SW, runner_mod.hw.SW_BUTTON, 1) + + ev.pack(0, 0, 0, 0, 0) + + ev.pack(0, 0, runner_mod.hw.EV_SW, runner_mod.hw.SW_BUTTON, 0)) + try: + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + self.press_button(False) # the status never shows it down + threading.Timer(0.3, press).start() + t0 = time.time() + ctx.ready("LIVE FIRE. Scrap under the head.") + self.assertLess(time.time() - t0, 5) + self.assertTrue(run.fixture_takeover) + rec = [r for r in run.evidence["actions"] if r["state"] == "presence"] + self.assertEqual(len(rec), 1) + self.assertIsNone(run.notice) # the notice is down with the press + finally: + os.environ.pop("GF_SWITCH_DEV", None) + + def test_the_arm_notice_comes_down_when_the_press_is_taken(self): + """The controller puts the button out at the press: the operator's + arm notice goes down then, not at the test's next step, so nobody + presses again (a press during the job pauses it).""" + run = Run("test", "r.live", "r.live") + ctx = Context(run, self.runner, self.reg["r.live"]) + self.runner.probe_fixture(force=True) + lit = {"v": False} + saved = runner_mod.hw.button_lit + runner_mod.hw.button_lit = lambda: lit["v"] + try: + self.assertFalse(ctx.arm_press()) # the operator's press + self.assertIsNotNone(run.notice) + time.sleep(0.2) + self.assertIsNotNone(run.notice) # dark before the arm: still asked for + lit["v"] = True + time.sleep(0.2) + self.assertIsNotNone(run.notice) # lit: the arm waits for the press + lit["v"] = False # the press taken + deadline = time.time() + 2 + while run.notice is not None and time.time() < deadline: + time.sleep(0.02) + self.assertIsNone(run.notice) + # A notice put up meanwhile is not the arm's: it stays. + lit["v"] = False + self.assertFalse(ctx.arm_press()) + ctx.notice("Something else.") + lit["v"] = True + time.sleep(0.2) + lit["v"] = False + time.sleep(0.3) + self.assertEqual((run.notice or {}).get("text"), "Something else.") + finally: + runner_mod.hw.button_lit = saved + if __name__ == "__main__": unittest.main()