LAB-303 / Block R: ScanImage ZMQ frame-stream wrapper node, latency report, off-rig harness

rig/README.md

@@ -305,6 +305,31 @@ build's own format setting decides. Found on the way: the pre-Next Studio's uplo
 `rename failed (0x03)` retries (the bug Next fixes), and on the Next build the Console panel's bridge
 *Connect* button did not open a socket while the client's `connect()` did — to check tomorrow.
 
+## ScanImage's ZeroMQ frame stream as ROS 2 topics (LAB-303 Block R, built off-rig 2026-10-18)
+
+ScanImage 2026 Premium publishes every acquired frame on a ZeroMQ PUB socket (`hSI.hScan2D.hAcq.ZMQAddress`,
+`tcp://127.0.0.1:65179` on the Bergamo PC, bound only while an acquisition runs): a 40-byte header of five
+doubles `[pixelsPerLine, linesPerFrame, nChannels, frameTimestamp_s, totalAcquiredFrames]` and the int16 planes.
+`nodes/scanimage_frames_node.py` subscribes and says exactly what it saw and lost; the pixels stay in the TIFF.
+Everything below was built and tested against the publisher stand-in (`tests/zmq_frames_mock.py`, the format and
+CLI of `scanimage-2026-bergamo/tests/zmq-frames/mock_zmq_publisher.py`), never yet against ScanImage — the lab day
+(Block Z+R, `procedures/rig-session-zmq-ros2-wrapper.md` in the scanimage repo) only measures.
+
+| piece | what |
+|---|---|
+| `pixi run scanimage-frames` / `pixi run rig -- --scanimage-frames` | the node: a receive thread owns the SUB socket (+ its monitor, so a publisher that goes away and comes back is logged and the next frame opens a new acquisition segment) and drains into a bounded queue; the ROS side publishes from it |
+| `/scanimage/frame_meta` | every frame: `frameNumber`, `frameTimestamp_s`, `dims`, `nchan`, `rx_epoch_ms` / `rx_mono_ns` (precise clocks on Windows too), `pub_mono_ns`, `queue_depth`, `gap_before`, `segment` |
+| `/scanimage/frame_stats` (1 Hz) | `frames`, `published`, `dropped_by_queue`, `bad_frames`, `gaps` / `gap_frames` / `gap_list` (by header frame number: what the publisher sent and this process never received), `segments` + per segment `first_frame` and `missed_before_first` (the slow-joiner loss), `connects` / `disconnects`, queue depth, in-node hop p50/p99; `received == published + dropped_by_queue + queue_depth` always |
+| `/scanimage/preview/ch<id>` + `/scanimage/preview/meta` | `sensor_msgs/CompressedImage` PNG, 128², ≤ 5 Hz per channel (`--preview-hz 0` = off; `--preview-range lo:hi` or auto = 1–99.5 percentile of the segment's first preview frame, held); the meta line records the mapping per image. Bounded on purpose — live imaging is not v1 |
+| the bag | the bridge node records all of the above as received (`STREAM_TOPICS`); the converter derives NO run-log line from them (streams, not behavior samples) |
+| `pixi run zmq-latency -- --bag <run>.bag [--tiff <acq>.tif]` | per segment: stream delay relative to the fastest frame (p50/p99/max; the acquisition and host clocks share no origin), clock drift, in-node hop, node → bridge hop (MCAP log_time), queue-depth histogram, gaps + slow-joiner loss, preview encode and frame → record latency; with the TIFF, the ZMQ-vs-TIFF timestamp difference per frame (the "same origin?" question) and the ABSOLUTE acquisition → topic latency from the TIFF `epoch`; pass bars from the procedure (meta p99 < one frame period, preview < 250 ms, 0 gaps, 0 queue drops) |
+| `pixi run test-frames` (first half of `pixi run test`) | router + node + in-process subscriber + the stand-in at 29.08 Hz × 512 × 512 × 2 ch: every frame counted and shaped right; `--drop-every 10` → 9 gaps in a new segment (a dropped LAST frame is invisible to any subscriber); publisher restart with no start delay → the slow-joiner loss recorded; the node SIGSTOPped 1.5 s with the publisher's HWM at libzmq's default 1000 (nothing lost, the publisher buffers — ScanImage's memory rise) and at 20 (the publisher drops, the node names every missing frame), publisher timing unaffected both times; clean SIGINT exit; a tiny queue + a stalled publisher thread → `dropped_by_queue` exact; the report on the dump and on a bag written with `nodes/recorder.py`, with a synthetic ScanImage-style TIFF for the clock alignment; the bridge's topic map |
+
+What the stand-in cannot tell us, so the lab day must: the absolute latency and its p99 against ScanImage's own
+`frameTimestamps_sec` (R2/R3), whether the ZMQ header timestamp and the TIFF's share an origin, how much MATLAB
+memory a paused subscriber costs (the PUB high-water mark, ~1 GB at 512 × 512 × 2 ch if it is libzmq's 1000), and
+that a subscriber being killed / paused / late never changes the acquisition's timing (`hMon`, `lostFrames`).
+
 ### Still not verified
 
 (2026-10-16: rosbag2's own reader on a torn bag is now verified — its `Reindexer` reads the crashsafe file in file order and writes `metadata.yaml`; the recovery path uses it.) Windows (now also: whether `pixi install` builds the interfaces package there — `WINDOWS-TEST.md` has the steps); the closed-loop math in a **separate** process (it stays in the bridge node here, so the `seq` ordering is available; the `--order time` numbers are the proxy); a multi-hour recording for bag growth; `ros2 bag play` of a recorded run against the Studio; a real FicTrac (the bench used the model-fly simulator).

rig/nodes/_common.py

@@ -100,6 +100,11 @@ def wrap_line(src: str, seq: int, raw: str, rx_epoch_ms: int, rx_mono_ns: int) -
 
 def spin(node) -> None:
     """rclpy.spin(node) until Ctrl-C / external shutdown, then destroy + shut down quietly."""
+    if hasattr(signal, "SIGBREAK"):
+        # Windows: rig.py and the harnesses stop children with CTRL_BREAK_EVENT (signal_group),
+        # whose default action ends the process at once (rc 0xC000013A, no clean shutdown;
+        # Bergamo 2026-10-20). Treat it as Ctrl-C, as bridge_node.py does.
+        signal.signal(signal.SIGBREAK, signal.default_int_handler)
     try:
         rclpy.spin(node)
     except (KeyboardInterrupt, ExternalShutdownException):

rig/nodes/bridge_node.py

@@ -38,7 +38,8 @@
 PLAN B SPIKE (LAB-303, 2026-10-14) — MCAP as the record of truth:
   * While a run's log is open, EVERYTHING is recorded into `<log-dir>/<run>.bag/` with
     rosbag2_py + the MCAP storage plugin (nodes/recorder.py): the device topics as received
-    (`/fictrac/raw` or the typed `/fictrac/frame`, `/kinefly/raw`, `/bias/cmd|reply|status`),
+    (`/fictrac/raw` or the typed `/fictrac/frame`, `/kinefly/raw`, `/bias/cmd|reply|status`, the
+    ScanImage frame-stream topics `/scanimage/frame_meta|frame_stats|preview/*` — streams, bag only),
     `/arena/frame` (the closed-loop evaluation with ITS OWN sample time `t`/`ms`), and two
     bridge-side streams — `/arena/browser` (every browser message, verbatim, with the receive
     stamp) and `/arena/event` (bias_config, heading_tare, connector_status, source_lost, warn)
@@ -82,6 +83,7 @@
 from rclpy.node import Node
 from rclpy.qos import QoSDurabilityPolicy, QoSHistoryPolicy, QoSProfile, QoSReliabilityPolicy
 from rclpy.serialization import deserialize_message, serialize_message
+from sensor_msgs.msg import CompressedImage
 from std_msgs.msg import String
 
 HERE = Path(__file__).resolve().parent
@@ -134,6 +136,19 @@ def load_bridge_module(path: Path = BRIDGE_PY):
 SCANIMAGE_CMD_TOPIC = "/scanimage/cmd"
 SCANIMAGE_REPLY_TOPIC = "/scanimage/reply"
 SCANIMAGE_STATUS_TOPIC = "/scanimage/status"
+# ScanImage's ZeroMQ frame stream as topics (nodes/scanimage_frames_node.py, LAB-303 Block R). STREAMS:
+# recorded into the bag as received, like /bias/frames, and never derived into a run-log line (the
+# telemetry rule: a string-tagged stream is not a behavior sample). The preview images are bounded at
+# the source (--preview-hz, 0 = off); ch1..ch4 cover ScanImage's four channels.
+SCANIMAGE_FRAME_META_TOPIC = "/scanimage/frame_meta"
+SCANIMAGE_FRAME_STATS_TOPIC = "/scanimage/frame_stats"
+SCANIMAGE_PREVIEW_META_TOPIC = "/scanimage/preview/meta"
+SCANIMAGE_PREVIEW_TOPICS = tuple(f"/scanimage/preview/ch{i}" for i in (1, 2, 3, 4))
+COMPRESSED_IMAGE_TYPE = "sensor_msgs/msg/CompressedImage"
+STREAM_TOPICS = {
+    SCANIMAGE_FRAME_META_TOPIC: STRING_TYPE, SCANIMAGE_FRAME_STATS_TOPIC: STRING_TYPE, SCANIMAGE_PREVIEW_META_TOPIC: STRING_TYPE,
+    **{t: COMPRESSED_IMAGE_TYPE for t in SCANIMAGE_PREVIEW_TOPICS},
+}
 # Device connectors the Studio may drive (LAB-303): connector_command → the node's cmd topic; its
 # reply/status topics become {"type":"connector","src":<id>,…} run-log events (derived by the
 # converter). Adding a connector = one entry here, a flag in rig.py, a topic pair in mcap_convert.py.
@@ -296,6 +311,7 @@ def topics(self) -> dict[str, str]:
             SESSION_TOPIC: STRING_TYPE, BROWSER_TOPIC: STRING_TYPE, EVENT_TOPIC: STRING_TYPE,
             KINEFLY_TOPIC: STRING_TYPE, BIAS_CMD_TOPIC: STRING_TYPE, BIAS_REPLY_TOPIC: STRING_TYPE, BIAS_STATUS_TOPIC: STRING_TYPE, BIAS_FRAMES_TOPIC: STRING_TYPE,
             SCANIMAGE_CMD_TOPIC: STRING_TYPE, SCANIMAGE_REPLY_TOPIC: STRING_TYPE, SCANIMAGE_STATUS_TOPIC: STRING_TYPE,
+            **STREAM_TOPICS,
         }
         t[FRAME_TOPIC] = ARENA_FRAME_TYPE if self._typed else STRING_TYPE
         if self.record_fictrac_raw:
@@ -639,6 +655,11 @@ def __init__(self, loop: asyncio.AbstractEventLoop, queue: asyncio.Queue, on_kin
         # Recorded on arrival, like /fictrac/raw: it never becomes a run-log line (the stopRecording reply
         # carries its summary), so there is no shadow line to stay in step with.
         self.create_subscription(String, BIAS_FRAMES_TOPIC, lambda b: self._record(BIAS_FRAMES_TOPIC, b), RAW_QOS, raw=True)
+        # The ScanImage frame stream (meta, stats, preview meta, preview images): the same arrival-time
+        # record, raw bytes, no run-log line — the bag is the only place they go.
+        for topic, typ in STREAM_TOPICS.items():
+            self.create_subscription(CompressedImage if typ == COMPRESSED_IMAGE_TYPE else String, topic,
+                                     lambda b, topic=topic: self._record(topic, b), RAW_QOS, raw=True)
 
     def _record(self, topic: str, data: bytes) -> None:
         rec = self._log.rec

rig/nodes/rig.py

@@ -7,6 +7,7 @@
     pixi run rig -- --bias --bias-sim  # + a simulated BIAS program (no camera software)
     pixi run rig -- --scanimage        # + the ScanImage node (JSON server on this PC); --si-mock for a stand-in
     pixi run rig -- --foxglove         # + foxglove_bridge (:8766) from the `viz` environment
+    pixi run rig -- --scanimage-frames # + the ScanImage ZMQ frame-stream node (SUB tcp://127.0.0.1:65179; Bergamo, Block R)
 
 Order: the router first (nodes retry until it answers on 127.0.0.1:7447), then every
 adapter/node. Each child's stderr is prefixed with its name. Ctrl-C stops everything in
@@ -138,6 +139,8 @@ def main(argv: list[str] | None = None) -> int:
     p.add_argument("--bias", action="store_true", help="also run the BIAS camera node")
     p.add_argument("--bias-sim", action="store_true", help="also run bias_sim.py (a fake BIAS program)")
     p.add_argument("--scanimage", action="store_true", help="also run the ScanImage node (talks to ScanImage's JSON server, default 127.0.0.1:5555)")
+    p.add_argument("--scanimage-frames", action="store_true", help="also run the ScanImage frame-stream node (ZMQ SUB on ScanImage's ZMQAddress → /scanimage/frame_meta, frame_stats, preview/ch<id>)")
+    p.add_argument("--scanimage-frames-args", default="", help="extra args for scanimage_frames_node.py (e.g. --address tcp://127.0.0.1:65179 --preview-hz 5 --dump frames.jsonl)")
     p.add_argument("--si-mock", action="store_true", help="also run tests/si_json_mock.py (a stand-in for ScanImage's JSON server, port 5555)")
     p.add_argument("--sim", action="store_true", help="also run the FicTrac simulator (random walk, 50 Hz)")
     p.add_argument("--viz-args", default="", help="extra args for viz_node.py (--deg-per-frame, --frame-dir, --feature-az0, --frames)")
@@ -206,6 +209,9 @@ def main(argv: list[str] | None = None) -> int:
         nodes.append(Child("si-mock", [py, str(HERE.parent / "tests" / "si_json_mock.py")]))
     if args.scanimage:
         nodes.append(Child("scanimage", [py, str(NODES / "scanimage_node.py"), *split_args(args.scanimage_args)]))
+    if args.scanimage_frames:
+        # Connects (never binds), so no port pre-flight: ScanImage binds at each acquisition start.
+        nodes.append(Child("si-frame", [py, str(NODES / "scanimage_frames_node.py"), *split_args(args.scanimage_frames_args)]))
     bridge = Child("bridge", [py, str(NODES / "bridge_node.py"), *split_args(args.bridge_args)])
     nodes.append(bridge)
     if foxglove: