forgetest: the jog witness judged over the sequence, as the sampler can see it

On the bench the head accelerometer's sysfs read lands two or three
samples in a one-second jog leg, and two samples on the constant-velocity
stretch read near idle with the head in full flight: the accelerometer
sees the ramps, not the travel. The first unattended queue on dev image
20260822204234 failed motion.jog-roundtrip on that (2 of 8 legs judged)
with the head plainly moving where a ramp was caught (p2p 3019, 1330,
1698, 1663). The verdict is now over the sequence: p2p across all the
legs at or above the liveness threshold, and motion on at least two
distinct legs (one jolt is not a gantry moving on every jog); the
sampler must have landed samples at all. Rerun: p2p 2897 over 17
samples, motion on 3 legs, PASS; the whole unattended queue passed (15,
then 9 re-required by the module hash) on that image.

camera.snapshot's PASS line names the half-res frame it compared (the
full-res size had been printed in its place). AccelSampler gets host
tests over a fake iio tree.

forgetest/forgetest/suite/camera.py

@@ -111,7 +111,8 @@ def snapshot(ctx):
     st, body = fc.get("/cam/status")
     ev["cam_status_after"] = body
     ctx.log("cam status after: %s", body)
-    ctx.log("PASS: snapshot %d bytes, %d with the lamp off, stream %s", len(data), len(dark), ctype)
+    ctx.log("PASS: half-res snapshot %d bytes, %d with the lamp off; full-res %d; stream %s",
+            ev["snapshot_bytes"], len(dark), len(data), ctype)
 
 
 @test("camera.sensor-profile", title="Camera geometry follows the fitted sensor", subsystem="camera",

forgetest/forgetest/suite/motion.py

@@ -229,24 +229,35 @@ def jog_roundtrip(ctx):
                 r = g.command(jog)
                 ctx.check(not any(x.startswith("error") for x in r), "%s: jog refused: %s", name, r)
                 peak, states, _ = wait_idle(ctx, g)
-                p2px, p2py, n = accel.p2p(t0)
+                t1 = time.time()
+                p2px, p2py, n = accel.p2p(t0, t1)
                 leg = "out" if jog == out else "back"
                 ctx.log("%s %s: peak %.0f mm/min, states %s; accel p2p x=%d y=%d over %d samples",
                         name, leg, peak, states, p2px, p2py, n)
                 moves.append({"name": name, "leg": leg, "peak": peak, "states": states,
-                              "accel_p2p": [p2px, p2py], "accel_samples": n})
+                              "accel_p2p": [p2px, p2py], "accel_samples": n, "t0": t0, "t1": t1})
                 ctx.check("TIMEOUT" not in states, "%s %s did not return to Idle", name, leg)
         ev["moves"] = moves
-        # The witness: every leg the sampler caught with enough samples
-        # must have moved the head; the short max-rate legs may land too
-        # few samples to judge on their own and are judged with the rest.
-        judged = [m for m in moves if m["accel_samples"] >= 3]
-        ctx.check(len(judged) >= 4, "the accelerometer sampled too few legs to judge motion (%d of %d; "
-                  "%d read errors)", len(judged), len(moves), accel.errors)
-        still = [m for m in judged if max(m["accel_p2p"]) < hw.ACCEL_P2P_MOVING]
-        ev["accel_still_legs"] = [m["name"] + " " + m["leg"] for m in still]
-        ctx.check(not still, "the head did not move on %s (accel p2p below %d): the counters ran "
-                  "without the gantry", ", ".join(ev["accel_still_legs"]), hw.ACCEL_P2P_MOVING)
+        # The witness sees the ramps, not the travel: a sysfs read lands
+        # two or three samples in a one-second leg, and two samples on
+        # the constant-velocity stretch read near the idle level with the
+        # head in full flight (bench: 17 samples over the eight legs, a
+        # ramp caught on three of them). So the verdict is over the
+        # sequence: the head moved during the jogs (p2p over all the
+        # legs), and on more than one of them, never one leg alone.
+        p2px, p2py, n = accel.p2p(moves[0]["t0"], moves[-1]["t1"])
+        moving = [m for m in moves if max(m["accel_p2p"]) >= hw.ACCEL_P2P_MOVING]
+        ev["accel_overall"] = {"p2p": [p2px, p2py], "samples": n, "errors": accel.errors}
+        ev["accel_moving_legs"] = [m["name"] + " " + m["leg"] for m in moving]
+        ctx.log("accel over all %d legs: p2p x=%d y=%d (%d samples, %d errors); motion seen on %d legs",
+                len(moves), p2px, p2py, n, accel.errors, len(moving))
+        ctx.check(n >= 8, "the accelerometer landed only %d samples over the jogs (%d read errors): "
+                  "the motion witness is not reading", n, accel.errors)
+        ctx.check(max(p2px, p2py) >= hw.ACCEL_P2P_MOVING,
+                  "the head did not move during the jogs (accel p2p x=%d y=%d, below %d): the counters "
+                  "ran without the gantry", p2px, p2py, hw.ACCEL_P2P_MOVING)
+        ctx.check(len(moving) >= 2, "the accelerometer saw motion on only %d leg(s) of %d (one jolt is "
+                  "not a gantry moving on every jog)", len(moving), len(moves))
         maxrate = max(m["peak"] for m in moves if m["name"].startswith("X max-rate"))
         ev["max_rate_peak"] = maxrate
         ctx.check(maxrate >= 6000, "max-rate jog peaked at only %.0f mm/min", maxrate)
@@ -274,8 +285,8 @@ def jog_roundtrip(ctx):
     machine_idle(ctx)
     ctx.check("Hold" in held["state"], "feed hold did not park (state %s)", held["state"])
     ctx.check(drift <= 0.05, "position drift %.3f mm", drift)
-    ctx.log("PASS: %d jogs, peak %.0f mm/min, hold parked, drift %.3f mm, the head seen moving on "
-            "%d of %d legs", len(moves), maxrate, drift, len(judged), len(moves))
+    ctx.log("PASS: %d jogs, peak %.0f mm/min, hold parked, drift %.3f mm, accel p2p %d over the jogs, "
+            "motion on %d of %d legs", len(moves), maxrate, drift, max(p2px, p2py), len(moving), len(moves))
 
 
 # ---------------------------------------------------------------- liveness

forgetest/tests/test_operator.py

@@ -297,3 +297,54 @@ class JournalTests(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class AccelSamplerTests(unittest.TestCase):
+    """The head accelerometer sampler over a fake iio tree."""
+
+    def setUp(self):
+        from forgetest import hw
+        self.hw = hw
+        self.tmp = tempfile.mkdtemp(prefix="forgetest-iio-")
+        d = os.path.join(self.tmp, "iio_device1")      # no colon: the host may be Windows
+        os.makedirs(d)
+        with open(os.path.join(d, "name"), "w") as f:
+            f.write("lis2hh12 3-001e\n")
+        self.dir = d
+        self.write(100, 200)
+        os.environ["GF_IIO_ROOT"] = self.tmp
+
+    def tearDown(self):
+        os.environ.pop("GF_IIO_ROOT", None)
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def write(self, x, y):
+        for axis, v in (("x", x), ("y", y)):
+            with open(os.path.join(self.dir, "in_accel_%s_raw" % axis), "w") as f:
+                f.write("%d\n" % v)
+
+    def test_found_by_bus_address_and_sampled(self):
+        self.assertEqual(self.hw.head_accel_dir(), self.dir)
+        s = self.hw.AccelSampler(period=0.02)
+        self.assertTrue(s.available)
+        with s:
+            t0 = time.time()
+            time.sleep(0.1)
+            self.write(1100, 200)
+            time.sleep(0.1)
+            self.write(100, 200)
+            time.sleep(0.1)
+            t1 = time.time()
+        p2px, p2py, n = s.p2p(t0, t1)
+        self.assertGreaterEqual(n, 5)
+        self.assertEqual((p2px, p2py), (1000, 0))
+        # a window with no samples reads as nothing, not as an error
+        self.assertEqual(s.p2p(t1 + 10, t1 + 20), (0, 0, 0))
+
+    def test_absent_device_is_unavailable(self):
+        os.environ["GF_IIO_ROOT"] = os.path.join(self.tmp, "nowhere")
+        s = self.hw.AccelSampler()
+        self.assertFalse(s.available)
+        with s:
+            pass
+        self.assertEqual(s.p2p(0), (0, 0, 0))