updates

firmware/pi_vision/README.md

@@ -0,0 +1,13 @@
+# Pi-to-eye gaze link
+
+`gaze_link.py` prepares the Raspberry Pi Zero 2 W to replace the DFRobot camera **as the one gaze-message sender**. It reproduces the ESP32 eye boards' [existing v1 packet](../../protocol/gaze.md), including CRC and sequence wrap; the eye firmware does not need reflashing. The Pi now boots Raspberry Pi OS Lite and detects an IMX708 camera, but the person detector and camera adapter remain to be implemented and benchmarked. This module has been tested on the laptop only; it has not sent a command from the Pi to the physical eyes.
+
+On the Pi, enable the hardware UART and disable its serial login console. After the 2026-10-01 reboot, `/dev/serial0` exists, points to `/dev/ttyS0`, and the kernel command line no longer assigns a serial console. `open_pi_uart()` uses Python's standard library, so no `python3-serial` install is needed for the GPIO link. It sends normalized `x`, `y`, and `valid` values. Use `GazeLink` with an in-memory stream for development without driving the real eyes. The Pi GPIO14 TX/pin 8 goes to **both** eye GPIO44 RX pins and all three boards share ground. Disconnect the DFRobot camera TX first, and keep the Pi and ESP32 power rails separate. The eye UART USB-C connectors are a viable alternate path through a Pi USB hub, but they disable the eye UART headers while connected. See [Pi trial plan](../../planning/pi-vision.md).
+
+Run the portable protocol check from this directory with `python3 -m unittest test_gaze_link.py`. It compares one encoded packet against a fixture generated by the existing C++ ESP32 protocol implementation. No camera or serial device is required for this check.
+
+With the Pi UART enabled, the DFRobot TX wire disconnected, and both eye RX pins connected to Pi GPIO14, run `python3 smoke_test_link.py`. It sends left, center, and right gaze positions with the red tracking state for two seconds each, then returns both eyes to blue idle. On 2026-10-01 the script ran twice on the Pi, and Joshua visually confirmed both eyes matched throughout the second run.
+
+For an initial **full-body detector benchmark**, install the Pi OS packages `python3-picamera2` and `python3-opencv`, then run `python3 benchmark_hog.py --duration 30 --rotation 180` with the current upside-down IMX708. It logs every frame's detection result and inference time under the project's `diagnostics/` directory. This OpenCV HOG detector is a baseline to measure; it requires most of a person's body in view and has not been accepted as the final person tracker. It does **not** send commands to the eyes. Judge its speed and false positives before connecting a detector to the gaze link. See the [Pi trial plan](../../planning/pi-vision.md).
+
+The first integrated trial is `python3 run_hog_eye.py --duration 30 --rotation 180`. It logs what it **would** send while leaving the eyes untouched. Add `--drive-eyes` for a timed hardware trial. It confirms a person after two detections and holds the last gaze through short misses. The first moving-person run on 2026-10-01 had a longest raw-detection gap of 3.34 seconds and Joshua saw the eyes briefly return to blue. Replaying that log showed a 3.5-second hold bridges all three observed red-to-blue drops, so the next trial uses 3.5 seconds. This also delays the return to blue after a person leaves and does not move the gaze during a detection gap. At the end it sends blue idle. This is a one-person proof of concept; continuous target tracking, multiple-target continuity, day/night testing, and false-positive evaluation remain.

firmware/pi_vision/benchmark_hog.py

@@ -0,0 +1,104 @@
+"""Measure a built-in OpenCV full-body detector on the Pi CSI camera.
+
+This is an offline baseline: it logs detections but does not drive the eyes.
+It will miss people whose full bodies are outside the frame.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+import time
+from datetime import datetime
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--duration", type=float, default=30.0, help="seconds to sample")
+    parser.add_argument("--rotation", type=int, choices=(0, 180), default=0)
+    parser.add_argument("--width", type=int, default=320)
+    parser.add_argument("--height", type=int, default=240)
+    parser.add_argument("--log", type=Path, help="JSONL path (default: project diagnostics)")
+    args = parser.parse_args()
+    if args.duration <= 0 or args.width < 128 or args.height < 128:
+        parser.error("duration must be positive and frame dimensions at least 128 pixels")
+
+    import cv2
+    from libcamera import Transform
+    from picamera2 import Picamera2
+
+    cv2.setNumThreads(2)
+    hog = cv2.HOGDescriptor()
+    hog.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector())
+    transform = Transform(hflip=1, vflip=1) if args.rotation == 180 else Transform()
+    camera = Picamera2()
+    config = camera.create_video_configuration(
+        main={"size": (args.width, args.height), "format": "RGB888"},
+        transform=transform,
+        buffer_count=3,
+    )
+    camera.configure(config)
+
+    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    log_path = args.log or PROJECT_ROOT / "diagnostics" / f"pi_hog_{stamp}.jsonl"
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    detection_ms: list[float] = []
+    detected_frames = 0
+    frames = 0
+    start = 0.0
+    try:
+        camera.start()
+        time.sleep(1.0)  # Let exposure settle before measuring.
+        start = time.monotonic()
+        with log_path.open("w", encoding="utf-8") as log:
+            while time.monotonic() - start < args.duration:
+                frame = camera.capture_array("main")
+                detect_start = time.monotonic()
+                boxes, weights = hog.detectMultiScale(
+                    frame,
+                    hitThreshold=0.0,
+                    winStride=(8, 8),
+                    padding=(8, 8),
+                    scale=1.05,
+                    groupThreshold=1,
+                )
+                elapsed_ms = (time.monotonic() - detect_start) * 1000
+                detections = [
+                    {"x": int(x), "y": int(y), "width": int(w), "height": int(h),
+                     "score": float(score)}
+                    for (x, y, w, h), score in zip(boxes, weights)
+                ]
+                frames += 1
+                detected_frames += bool(detections)
+                detection_ms.append(elapsed_ms)
+                log.write(json.dumps({
+                    "time": datetime.now().astimezone().isoformat(),
+                    "frame": frames,
+                    "detect_ms": round(elapsed_ms, 1),
+                    "detections": detections,
+                }) + "\n")
+                if frames % 10 == 0:
+                    print(f"frames={frames} detected={detected_frames} "
+                          f"last_detect_ms={elapsed_ms:.0f}", flush=True)
+    finally:
+        camera.stop()
+        camera.close()
+
+    total = time.monotonic() - start
+    print(json.dumps({
+        "frames": frames,
+        "seconds": round(total, 2),
+        "fps": round(frames / total, 2) if total else 0,
+        "frames_with_person": detected_frames,
+        "median_detect_ms": round(statistics.median(detection_ms), 1) if detection_ms else None,
+        "log": str(log_path),
+    }, indent=2))
+
+
+if __name__ == "__main__":
+    main()

firmware/pi_vision/gaze_link.py

@@ -0,0 +1,72 @@
+"""Raspberry Pi sender for the two existing ESP32 eye boards' gaze protocol."""
+
+from __future__ import annotations
+
+import struct
+from typing import BinaryIO
+
+
+def crc16_ccitt(data: bytes) -> int:
+    crc = 0xFFFF
+    for byte in data:
+        crc ^= byte << 8
+        for _ in range(8):
+            crc = ((crc << 1) ^ 0x1021) & 0xFFFF if crc & 0x8000 else (crc << 1) & 0xFFFF
+    return crc
+
+
+def encode_frame(sequence: int, x: int, y: int, valid: bool) -> bytes:
+    """Encode protocol v1; x and y are normalized camera coordinates (0..1000)."""
+    if not 0 <= sequence <= 0xFFFF or not 0 <= x <= 1000 or not 0 <= y <= 1000:
+        raise ValueError("sequence or coordinates out of range")
+    body = struct.pack("<2sBBHHH", b"\xA5\x5A", 1, int(bool(valid)), sequence, x, y)
+    return body + struct.pack("<H", crc16_ccitt(body))
+
+
+def center_to_gaze(box: tuple[int, int, int, int], width: int, height: int) -> tuple[int, int]:
+    """Convert a person box (left, top, right, bottom) to the eye protocol range."""
+    if width < 2 or height < 2:
+        raise ValueError("frame must be at least 2 by 2 pixels")
+    left, top, right, bottom = box
+    if not (0 <= left <= right < width and 0 <= top <= bottom < height):
+        raise ValueError("box must lie within frame")
+    return ((left + right) // 2 * 1000 // (width - 1),
+            (top + bottom) // 2 * 1000 // (height - 1))
+
+
+class GazeLink:
+    """Write one shared gaze stream; sequence advances even for invalid targets."""
+
+    def __init__(self, stream: BinaryIO) -> None:
+        self.stream = stream
+        self.sequence = 0
+
+    def send(self, x: int, y: int, valid: bool) -> int:
+        sequence = self.sequence
+        frame = encode_frame(sequence, x, y, valid)
+        written = self.stream.write(frame)
+        if written != len(frame):
+            raise OSError(f"short UART write: {written} of {len(frame)} bytes")
+        self.sequence = (sequence + 1) & 0xFFFF
+        return sequence
+
+
+def open_pi_uart(device: str = "/dev/serial0") -> GazeLink:
+    """Open the Pi's enabled hardware UART after disabling its login console."""
+    import os
+    import termios
+
+    fd = os.open(device, os.O_WRONLY | os.O_NOCTTY)
+    try:
+        settings = termios.tcgetattr(fd)
+        settings[0] = 0  # No input translation or software flow control.
+        settings[1] = 0  # No output translation.
+        settings[2] = termios.CLOCAL | termios.CREAD | termios.CS8  # 8N1, no RTS/CTS.
+        settings[3] = 0  # Raw mode; no terminal processing.
+        settings[4] = termios.B115200
+        settings[5] = termios.B115200
+        termios.tcsetattr(fd, termios.TCSANOW, settings)
+        return GazeLink(os.fdopen(fd, "wb", buffering=0))
+    except BaseException:
+        os.close(fd)
+        raise

firmware/pi_vision/run_hog_eye.py

@@ -0,0 +1,117 @@
+"""Trial IMX708 full-body detection driving the two eyes over one Pi UART.
+
+By default this only logs. Pass --drive-eyes for a timed hardware trial.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import time
+from datetime import datetime
+from pathlib import Path
+
+from gaze_link import GazeLink, center_to_gaze, open_pi_uart
+from tracking_state import TrackingState
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--duration", type=float, default=30.0)
+    parser.add_argument("--rotation", type=int, choices=(0, 180), default=0)
+    parser.add_argument("--drive-eyes", action="store_true")
+    parser.add_argument("--device", default="/dev/serial0")
+    parser.add_argument("--log", type=Path)
+    args = parser.parse_args()
+    if args.duration <= 0:
+        parser.error("duration must be positive")
+
+    import cv2
+    from libcamera import Transform
+    from picamera2 import Picamera2
+
+    cv2.setNumThreads(2)
+    hog = cv2.HOGDescriptor()
+    hog.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector())
+    camera = Picamera2()
+    transform = Transform(hflip=1, vflip=1) if args.rotation == 180 else Transform()
+    camera.configure(camera.create_video_configuration(
+        main={"size": (320, 240), "format": "RGB888"},
+        transform=transform,
+        buffer_count=3,
+    ))
+    # The first moving-person log had a 3.34 s detector gap despite a visible person.
+    state = TrackingState(confirm_hits=2, hold_seconds=3.5)
+    link = open_pi_uart(args.device) if args.drive_eyes else GazeLink(open(os.devnull, "wb"))
+    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    log_path = args.log or PROJECT_ROOT / "diagnostics" / f"pi_eye_trial_{stamp}.jsonl"
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    started = False
+    frames = 0
+    tracked = 0
+    try:
+        camera.start()
+        started = True
+        time.sleep(1.0)
+        deadline = time.monotonic() + args.duration
+        with log_path.open("w", encoding="utf-8") as log:
+            while time.monotonic() < deadline:
+                frame = camera.capture_array("main")
+                detect_start = time.monotonic()
+                boxes, weights = hog.detectMultiScale(
+                    frame, hitThreshold=0.0, winStride=(8, 8),
+                    padding=(8, 8), scale=1.05, groupThreshold=1,
+                )
+                detect_ms = round((time.monotonic() - detect_start) * 1000, 1)
+                target = None
+                candidate = None
+                if len(boxes):
+                    index = max(range(len(boxes)), key=lambda i: float(weights[i]))
+                    box_x, box_y, width, height = (int(v) for v in boxes[index])
+                    left = max(0, box_x)
+                    top = max(0, box_y)
+                    right = min(box_x + width - 1, frame.shape[1] - 1)
+                    bottom = min(box_y + height - 1, frame.shape[0] - 1)
+                    if left <= right and top <= bottom:
+                        candidate = center_to_gaze(
+                            (left, top, right, bottom), frame.shape[1], frame.shape[0]
+                        )
+                        target = {"x": box_x, "y": box_y, "width": width,
+                                  "height": height, "score": float(weights[index])}
+                x, y, valid = state.update(candidate, time.monotonic())
+                seq = link.send(x, y, valid)
+                frames += 1
+                tracked += valid
+                log.write(json.dumps({
+                    "time": datetime.now().astimezone().isoformat(),
+                    "frame": frames,
+                    "detect_ms": detect_ms,
+                    "target": target,
+                    "gaze": {"x": x, "y": y, "valid": valid, "sequence": seq},
+                }) + "\n")
+                if frames % 10 == 0:
+                    print(f"frames={frames} tracked={tracked} "
+                          f"last_gaze={x},{y},{int(valid)}", flush=True)
+    finally:
+        # The eyes should return to blue whether the trial ends normally or fails.
+        try:
+            if args.drive_eyes:
+                for _ in range(3):
+                    link.send(500, 500, False)
+                    time.sleep(0.1)
+        finally:
+            link.stream.close()
+            if started:
+                camera.stop()
+            camera.close()
+
+    print(json.dumps({"frames": frames, "tracked_frames": tracked,
+                      "log": str(log_path), "eyes_driven": args.drive_eyes}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

firmware/pi_vision/smoke_test_link.py

@@ -0,0 +1,42 @@
+"""Briefly drive both eyes through the existing shared UART gaze link."""
+
+from __future__ import annotations
+
+import argparse
+import time
+
+from gaze_link import GazeLink, open_pi_uart
+
+
+def send_state(link: GazeLink, x: int, y: int, valid: bool, duration: float) -> None:
+    deadline = time.monotonic() + duration
+    while time.monotonic() < deadline:
+        link.send(x, y, valid)
+        time.sleep(0.1)  # The eye receiver returns to idle after 750 ms without a frame.
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--device", default="/dev/serial0", help="Pi GPIO UART device")
+    args = parser.parse_args()
+
+    link = open_pi_uart(args.device)
+    try:
+        for label, x, valid in (
+            ("left/red", 200, True),
+            ("center/red", 500, True),
+            ("right/red", 800, True),
+            ("idle/blue", 500, False),
+        ):
+            print(label, flush=True)
+            send_state(link, x, 500, valid, 2.0)
+    finally:
+        # Leave the eyes blue even if the test is interrupted.
+        for _ in range(3):
+            link.send(500, 500, False)
+            time.sleep(0.1)
+        link.stream.close()
+
+
+if __name__ == "__main__":
+    main()

firmware/pi_vision/test_gaze_link.py

@@ -0,0 +1,54 @@
+import io
+import os
+import pty
+import select
+import termios
+import unittest
+
+from gaze_link import GazeLink, center_to_gaze, crc16_ccitt, encode_frame, open_pi_uart
+
+
+class GazeLinkTests(unittest.TestCase):
+    def test_crc_standard_vector_and_esp32_fixture(self):
+        self.assertEqual(crc16_ccitt(b"123456789"), 0x29B1)
+        # Generated by the current C++ gaze::encode used on the ESP32 boards.
+        self.assertEqual(
+            encode_frame(0x1234, 655, 774, True).hex(),
+            "a55a010134128f020603fa76",
+        )
+
+    def test_stream_and_sequence_wrap(self):
+        stream = io.BytesIO()
+        link = GazeLink(stream)
+        link.sequence = 0xFFFF
+        self.assertEqual(link.send(500, 500, False), 0xFFFF)
+        self.assertEqual(link.send(0, 1000, True), 0)
+        self.assertEqual(len(stream.getvalue()), 24)
+        self.assertEqual(stream.getvalue()[12:14], b"\xA5\x5A")
+
+    def test_box_center_and_invalid_inputs(self):
+        self.assertEqual(center_to_gaze((0, 0, 319, 239), 320, 240), (498, 497))
+        with self.assertRaises(ValueError):
+            center_to_gaze((0, 0, 320, 239), 320, 240)
+        with self.assertRaises(ValueError):
+            encode_frame(0, 1001, 0, True)
+
+    def test_serial_port_sends_115200_baud_frame(self):
+        master, slave = pty.openpty()
+        try:
+            link = open_pi_uart(os.ttyname(slave))
+            try:
+                settings = termios.tcgetattr(slave)
+                self.assertEqual(settings[4:6], [termios.B115200, termios.B115200])
+                link.send(200, 500, True)
+                self.assertTrue(select.select([master], [], [], 1)[0])
+                self.assertEqual(os.read(master, 12), encode_frame(0, 200, 500, True))
+            finally:
+                link.stream.close()
+        finally:
+            os.close(slave)
+            os.close(master)
+
+
+if __name__ == "__main__":
+    unittest.main()

firmware/pi_vision/test_tracking_state.py

@@ -0,0 +1,39 @@
+import unittest
+
+from tracking_state import TrackingState
+
+
+class TrackingStateTests(unittest.TestCase):
+    def test_two_hits_confirm_and_short_misses_hold(self):
+        state = TrackingState()
+        self.assertEqual(state.update((200, 400), 0.0), (500, 500, False))
+        self.assertEqual(state.update((250, 450), 0.2), (250, 450, True))
+        self.assertEqual(state.update(None, 1.8), (250, 450, True))
+        self.assertEqual(state.update((300, 450), 1.9), (300, 450, True))
+        self.assertEqual(state.update(None, 3.9), (500, 500, False))
+
+    def test_isolated_hit_does_not_trigger(self):
+        state = TrackingState()
+        self.assertFalse(state.update((100, 100), 0.0)[2])
+        self.assertFalse(state.update(None, 0.8)[2])
+        self.assertFalse(state.update((900, 900), 1.0)[2])
+
+    def test_recorded_moving_person_gap_is_held(self):
+        state = TrackingState(hold_seconds=3.5)
+        state.update((200, 600), 0.0)
+        self.assertTrue(state.update((210, 600), 0.2)[2])
+        self.assertTrue(state.update(None, 3.53)[2])
+        self.assertTrue(state.update((350, 600), 3.54)[2])
+        self.assertFalse(state.update(None, 7.04)[2])
+
+    def test_bad_configuration_and_coordinates(self):
+        with self.assertRaises(ValueError):
+            TrackingState(confirm_hits=0)
+        with self.assertRaises(ValueError):
+            TrackingState(hold_seconds=0)
+        with self.assertRaises(ValueError):
+            TrackingState().update((1001, 0), 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()

firmware/pi_vision/tracking_state.py

@@ -0,0 +1,36 @@
+"""Small Pi-side gate to keep both eyes on one person through brief misses."""
+
+from __future__ import annotations
+
+
+class TrackingState:
+    def __init__(self, confirm_hits: int = 2, hold_seconds: float = 2.0) -> None:
+        if confirm_hits < 1 or hold_seconds <= 0:
+            raise ValueError("confirmation and hold must be positive")
+        self.confirm_hits = confirm_hits
+        self.hold_seconds = hold_seconds
+        self.hits = 0
+        self.last_seen: float | None = None
+        self.valid = False
+        self.x = 500
+        self.y = 500
+
+    def update(self, candidate: tuple[int, int] | None, now: float) -> tuple[int, int, bool]:
+        if candidate is not None:
+            x, y = candidate
+            if not 0 <= x <= 1000 or not 0 <= y <= 1000:
+                raise ValueError("candidate gaze coordinates out of range")
+            if self.last_seen is None or now - self.last_seen > 0.75:
+                self.hits = 0
+            self.hits += 1
+            self.last_seen = now
+            if self.valid or self.hits >= self.confirm_hits:
+                self.valid = True
+                self.x, self.y = x, y
+        elif self.last_seen is not None:
+            if self.valid and now - self.last_seen >= self.hold_seconds:
+                self.valid = False
+                self.x = self.y = 500
+            if not self.valid and now - self.last_seen > 0.75:
+                self.hits = 0
+        return self.x, self.y, self.valid