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()