From 6f8d633b67176005e6e98f8b990811ba9457d794 Mon Sep 17 00:00:00 2001 From: Michael B Reiser Date: Sun, 18 Oct 2026 13:30:53 -0400 Subject: [PATCH 1/5] ros2-poc: ScanImage ZMQ frame-stream wrapper node + latency report + harness (LAB-303 Block R) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Built off-rig against a stand-in of ScanImage 2026's ZeroMQ frame PUB (40-byte header of five doubles + int16 planes), so the Bergamo lab day (Block Z+R) only measures. - nodes/scanimage_frames_node.py: SUB + monitor socket on a receive thread into a bounded queue; /scanimage/frame_meta (every frame), /scanimage/frame_stats (1 Hz: gaps by header frame number, dropped-by-queue, acquisition segments with the slow-joiner loss, connects, hop p50/p99), optional bounded preview /scanimage/preview/ch (CompressedImage PNG 128², <= 5 Hz) with the display mapping on /scanimage/preview/meta. received == published + dropped_by_queue + queue_depth. - bridge_node.py: STREAM_TOPICS recorded on arrival into every bag; the converter derives nothing. - rig.py --scanimage-frames [--scanimage-frames-args]; pixi tasks scanimage-frames, test-frames, zmq-latency; `pixi run test` runs the frames harness, then the acceptance run. - tools/zmq_latency_report.py: per segment stream delay (relative), clock drift, in-node and node->bridge hops, queue-depth histogram, gaps, preview latency; with the TIFF (stdlib reader) the ZMQ-vs-TIFF timestamp difference and the absolute latency from `epoch`; the procedure's bars. - tests/test_scanimage_frames.py (39 checks) + tests/zmq_frames_mock.py (the scanimage repo's mock_zmq_publisher.py format/CLI + --sndhwm/--seed; --mock runs the original: 37/37). Co-Authored-By: Claude Fable 5.1 --- ros2-poc/README.md | 25 ++ ros2-poc/nodes/bridge_node.py | 23 +- ros2-poc/nodes/rig.py | 6 + ros2-poc/nodes/scanimage_frames_node.py | 522 +++++++++++++++++++++++ ros2-poc/pixi.toml | 8 +- ros2-poc/tests/test_scanimage_frames.py | 531 ++++++++++++++++++++++++ ros2-poc/tests/zmq_frames_mock.py | 72 ++++ ros2-poc/tools/zmq_latency_report.py | 423 +++++++++++++++++++ 8 files changed, 1608 insertions(+), 2 deletions(-) create mode 100644 ros2-poc/nodes/scanimage_frames_node.py create mode 100644 ros2-poc/tests/test_scanimage_frames.py create mode 100644 ros2-poc/tests/zmq_frames_mock.py create mode 100644 ros2-poc/tools/zmq_latency_report.py diff --git a/ros2-poc/README.md b/ros2-poc/README.md index aa147a01..4cf5e62f 100644 --- a/ros2-poc/README.md +++ b/ros2-poc/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` + `/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 .bag [--tiff .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). diff --git a/ros2-poc/nodes/bridge_node.py b/ros2-poc/nodes/bridge_node.py index a48b4800..60df8426 100644 --- a/ros2-poc/nodes/bridge_node.py +++ b/ros2-poc/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 `/.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":,…} 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. @@ -290,6 +305,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, } if self._typed: t[FRAME_TOPIC] = ARENA_FRAME_TYPE @@ -599,6 +615,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 diff --git a/ros2-poc/nodes/rig.py b/ros2-poc/nodes/rig.py index 4e617ea5..e7549254 100644 --- a/ros2-poc/nodes/rig.py +++ b/ros2-poc/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)") + 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: diff --git a/ros2-poc/nodes/scanimage_frames_node.py b/ros2-poc/nodes/scanimage_frames_node.py new file mode 100644 index 00000000..11a0926f --- /dev/null +++ b/ros2-poc/nodes/scanimage_frames_node.py @@ -0,0 +1,522 @@ +#!/usr/bin/env python3 +"""scanimage_frames_node.py — ScanImage's ZeroMQ frame stream as ROS 2 topics (LAB-303, Block R). + +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 rig; bound only while an +acquisition runs). Each frame is a 2-part message: a 40-byte header of five little-endian doubles +[pixelsPerLine, linesPerFrame, nChannels, frameTimestamp_s (since acquisition start), +totalAcquiredFrames] and the int16 planes (x fastest, then y, then channel). PUB drops silently +when a subscriber falls behind, and a subscriber that joins after the bind misses the first +frames (the "slow joiner"). This node is that subscriber, and its job is to say exactly what it +saw and what it lost — the raw pixels stay in ScanImage's TIFF. + + /scanimage/frame_meta OUT every frame: {"seq", "segment", "frameNumber", "frameTimestamp_s", + "dims": [ppl, lpf], "nchan", "bytes", "gap_before", "rx_epoch_ms", + "rx_mono_ns", "pub_mono_ns", "queue_depth", "preview"} + /scanimage/frame_stats OUT every --stats-every s: {"connected", "connects", "frames", "published", + "dropped_by_queue", "bad_frames", "gaps", "gap_frames", "gap_list", + "segments", "segment_list": [{"id", "first_frame", "missed_before_first", + "frames", "last_frame", "dims", "nchan", …}], "queue_depth", "queue_max", + "rate_hz", "preview": {…}} + /scanimage/preview/ch OUT sensor_msgs/CompressedImage (PNG, --preview-size², ≤ --preview-hz per + channel; 0 = off). Bounded on purpose: live imaging is not v1. + /scanimage/preview/meta OUT one JSON line per preview image: which frame, which channel, and the + int16 → 8-bit display mapping used ({"mode", "lo", "hi"}). + +Design: a dedicated receive thread owns the SUB socket (plus its monitor socket, which reports +connect/disconnect, so a publisher that goes away and comes back is logged and the next frames open a +new segment). It parses the header, stamps the frame (`rx_epoch_ms` wall / `rx_mono_ns` monotonic, +_common's precise clocks on Windows too), does the loss accounting — all of it on the receive thread, +so every frame that reaches this process is counted — and hands the frame to a BOUNDED queue. The +ROS side publishes from that queue; when it falls behind, the newest frame is dropped and counted +as `dropped_by_queue` (never a gap). Loss accounting, by header frame number: + + gap frameNumber jumped by more than 1 inside a segment — frames the publisher sent and + this process never received (PUB dropped them: slow subscriber, SIGSTOP, HWM); + segment a new acquisition: the frame counter restarted (ScanImage numbers each acquisition + from 1) or the socket reconnected; `missed_before_first` = frameNumber − 1 of its + first frame = the slow-joiner loss; + dropped_by_queue received, not published (this node's own back-pressure); + bad_frames a message whose header or pixel length is not the documented shape. + received == published + dropped_by_queue + queue_depth, always. + + pixi run scanimage-frames # tcp://127.0.0.1:65179, preview 5 Hz + pixi run scanimage-frames -- --preview-hz 0 --dump meta.jsonl + pixi run rig -- --scanimage-frames # supervised with the bridge node (bag records the topics) +""" + +from __future__ import annotations + +import argparse +import json +import queue +import struct +import sys +import threading +import time +import zlib + +import numpy as np +import zmq +from zmq.utils.monitor import recv_monitor_message + +import rclpy +from rclpy.node import Node +from sensor_msgs.msg import CompressedImage +from std_msgs.msg import String + +from _common import RELIABLE_QOS, mono_ns, raise_timer_resolution, spin, wall_ns + +SRC = "scanimage_frames" +META_TOPIC = "/scanimage/frame_meta" +STATS_TOPIC = "/scanimage/frame_stats" +PREVIEW_TOPIC_PREFIX = "/scanimage/preview/ch" +PREVIEW_META_TOPIC = "/scanimage/preview/meta" +DEFAULT_ADDRESS = "tcp://127.0.0.1:65179" +HEADER = struct.Struct("<5d") # pixelsPerLine, linesPerFrame, nChannels, frameTimestamp_s, totalAcquiredFrames +QOS = RELIABLE_QOS +GAP_LIST_MAX = 200 # gap events kept in memory / sent in frame_stats (the count is always exact) +STATS_GAP_LIST = 50 # the most recent gap events per frame_stats message + + +# ── pure helpers (Node-free; tests import them) ──────────────────────────────────────────── +def parse_header(hdr: bytes) -> dict: + """The 40-byte header → {"ppl", "lpf", "nchan", "frameTimestamp_s", "frameNumber"}. Raises + ValueError on the wrong length or non-integer geometry.""" + if len(hdr) != HEADER.size: + raise ValueError(f"header is {len(hdr)} B, expected {HEADER.size}") + ppl, lpf, nch, ts, n = HEADER.unpack(hdr) + for name, v in (("pixelsPerLine", ppl), ("linesPerFrame", lpf), ("nChannels", nch), ("totalAcquiredFrames", n)): + if not (v == v) or v != int(v) or v < 0 or v > 1e9: # NaN, fractional, negative, absurd + raise ValueError(f"header {name}={v!r} is not a non-negative integer") + if not (ts == ts): + raise ValueError("header frameTimestamp is NaN") + return {"ppl": int(ppl), "lpf": int(lpf), "nchan": int(nch), "frameTimestamp_s": float(ts), "frameNumber": int(n)} + + +def expected_bytes(h: dict) -> int: + return h["ppl"] * h["lpf"] * h["nchan"] * 2 + + +def planes(buf, h: dict) -> np.ndarray: + """The pixel part as an int16 array [channel, y, x] (a view; no copy).""" + return np.frombuffer(buf, dtype=" np.ndarray: + """(H, W) → (size, size) float32: block mean when both axes divide evenly (512 → 128), else + nearest-neighbour sampling (also upsamples a 96-line mock frame).""" + h, w = plane.shape + if h % size == 0 and w % size == 0 and h >= size and w >= size: + return plane.reshape(size, h // size, size, w // size).astype(np.float32).mean(axis=(1, 3)) + ys = (np.arange(size) * h) // size + xs = (np.arange(size) * w) // size + return plane[np.ix_(ys, xs)].astype(np.float32) + + +def auto_range(img: np.ndarray) -> tuple[float, float]: + """A display mapping from the data: 1st–99.5th percentile, widened to at least 1 count.""" + lo, hi = (float(x) for x in np.percentile(img, (1.0, 99.5))) + if hi <= lo: + hi = lo + 1.0 + return lo, hi + + +def to_uint8(img: np.ndarray, lo: float, hi: float) -> np.ndarray: + return np.clip((img - lo) * (255.0 / (hi - lo)), 0.0, 255.0).astype(np.uint8) + + +def png_gray(img: np.ndarray) -> bytes: + """8-bit grayscale (H, W) → PNG bytes (stdlib zlib; no imaging library in the env).""" + img = np.ascontiguousarray(img, dtype=np.uint8) + h, w = img.shape + raw = b"".join(b"\x00" + img[i].tobytes() for i in range(h)) + + def chunk(tag: bytes, body: bytes) -> bytes: + return struct.pack(">I", len(body)) + tag + body + struct.pack(">I", zlib.crc32(tag + body) & 0xFFFFFFFF) + + return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 0, 0, 0, 0)) + + chunk(b"IDAT", zlib.compress(raw, 6)) + chunk(b"IEND", b"")) + + +def parse_range(text: str) -> tuple[float, float] | None: + """--preview-range: 'auto' → None, 'lo:hi' → (lo, hi).""" + if text.strip().lower() == "auto": + return None + lo, hi = (float(x) for x in text.split(":", 1)) + if hi <= lo: + raise argparse.ArgumentTypeError("--preview-range lo:hi needs hi > lo") + return lo, hi + + +# ── the receive side ─────────────────────────────────────────────────────────────────────── +class Segment: + """One acquisition as seen on the stream.""" + + def __init__(self, sid: int, h: dict, rx_epoch_ms: int, rx_mono_ns: int) -> None: + self.id = sid + self.first_frame = h["frameNumber"] + self.missed_before_first = max(h["frameNumber"] - 1, 0) + self.last_frame = h["frameNumber"] + self.first_ts = self.last_ts = h["frameTimestamp_s"] + self.dims = [h["ppl"], h["lpf"]] + self.nchan = h["nchan"] + self.dims_changes = 0 + self.frames = 0 + self.gaps = 0 + self.gap_frames = 0 + self.t_first_ms = self.t_last_ms = rx_epoch_ms + self.mono_first_ns = self.mono_last_ns = rx_mono_ns + + def as_dict(self) -> dict: + span_s = (self.mono_last_ns - self.mono_first_ns) / 1e9 + return { + "id": self.id, "first_frame": self.first_frame, "missed_before_first": self.missed_before_first, + "last_frame": self.last_frame, "frames": self.frames, "gaps": self.gaps, "gap_frames": self.gap_frames, + "dims": self.dims, "nchan": self.nchan, "dims_changes": self.dims_changes, + "first_ts_s": self.first_ts, "last_ts_s": self.last_ts, "t_first_ms": self.t_first_ms, "t_last_ms": self.t_last_ms, + "rate_hz": round((self.frames - 1) / span_s, 3) if self.frames > 1 and span_s > 0 else None, + } + + +class FrameReceiver: + """SUB socket + monitor on its own thread → bounded queue of (meta, pixels-or-None).""" + + def __init__(self, address: str, depth: int, rcv_hwm: int, preview_hz: float, log) -> None: + self.address = address + self.q: queue.Queue = queue.Queue(maxsize=max(1, depth)) + self.rcv_hwm = rcv_hwm + self.preview_period_ns = int(1e9 / preview_hz) if preview_hz > 0 else 0 + self.log = log + self.lock = threading.Lock() + self.stop = threading.Event() + self.thread = threading.Thread(target=self._run, name="zmq-rx", daemon=True) + # counters (under lock) + self.frames = 0 + self.bytes = 0 + self.bad_frames = 0 + self.dropped_by_queue = 0 + self.gaps = 0 + self.gap_frames = 0 + self.gap_list: list[dict] = [] + self.segments: list[Segment] = [] + self.connected = False + self.connects = 0 + self.disconnects = 0 + self.reconnected = False # a (re)connect since the last frame → next frame opens a segment + self.queue_max = 0 + self._last_preview_ns = 0 + self.started_mono = mono_ns() + + def start(self) -> None: + self.thread.start() + + def close(self) -> None: + self.stop.set() + self.thread.join(timeout=3.0) + + # ── thread body ── + def _run(self) -> None: + ctx = zmq.Context.instance() + sock = ctx.socket(zmq.SUB) + sock.setsockopt(zmq.SUBSCRIBE, b"") + sock.setsockopt(zmq.LINGER, 0) + sock.setsockopt(zmq.RCVHWM, self.rcv_hwm) + sock.setsockopt(zmq.RECONNECT_IVL, 100) + sock.setsockopt(zmq.RECONNECT_IVL_MAX, 500) # ScanImage rebinds per acquisition; a localhost SYN every ≤ 0.5 s is cheap + mon = sock.get_monitor_socket(zmq.EVENT_CONNECTED | zmq.EVENT_DISCONNECTED | zmq.EVENT_CLOSED) + sock.connect(self.address) + poller = zmq.Poller() + poller.register(sock, zmq.POLLIN) + poller.register(mon, zmq.POLLIN) + try: + while not self.stop.is_set(): + for s, _ in poller.poll(100): + if s is mon: + self._monitor_event(recv_monitor_message(mon)) + else: + self._frame(sock.recv_multipart(copy=False)) + finally: + try: + sock.disable_monitor() + except Exception: + pass + mon.close(0) + sock.close(0) + + def _monitor_event(self, ev: dict) -> None: + kind = ev.get("event") + with self.lock: + if kind == zmq.EVENT_CONNECTED: + was = self.connects + self.connects += 1 + self.connected = True + if was > 0: + self.reconnected = True + self.log.info(f"{'re' if was else ''}connected to {self.address} (#{self.connects})") + elif kind in (zmq.EVENT_DISCONNECTED, zmq.EVENT_CLOSED): + if self.connected: + self.disconnects += 1 + self.log.info(f"publisher at {self.address} went away (waiting for the next bind)") + self.connected = False + + def _frame(self, parts: list) -> None: + rx_mono = mono_ns() + rx_wall = wall_ns() + try: + if len(parts) != 2: + raise ValueError(f"{len(parts)}-part message, expected 2") + h = parse_header(bytes(parts[0].buffer)) + n = expected_bytes(h) + if len(parts[1].buffer) != n: + raise ValueError(f"{len(parts[1].buffer)} pixel bytes, header says {n}") + except ValueError as exc: + with self.lock: + self.bad_frames += 1 + bad = self.bad_frames + if bad <= 5 or bad % 100 == 0: + self.log.warning(f"bad frame #{bad}: {exc}") + return + want_preview = False + if self.preview_period_ns and rx_mono - self._last_preview_ns >= self.preview_period_ns: + want_preview = True + self._last_preview_ns = rx_mono + with self.lock: + self.frames += 1 + self.bytes += n + gap_before = 0 + seg = self.segments[-1] if self.segments else None + new = seg is None or self.reconnected or h["frameNumber"] <= seg.last_frame or h["frameTimestamp_s"] < seg.last_ts - 1e-6 + if new: + seg = Segment(len(self.segments) + 1, h, rx_wall // 1_000_000, rx_mono) + self.segments.append(seg) + self.reconnected = False + self.log.info(f"segment {seg.id}: first frame {seg.first_frame} ({seg.missed_before_first} missed before it), " + f"{h['ppl']}x{h['lpf']} x {h['nchan']} ch") + else: + if h["frameNumber"] > seg.last_frame + 1: + gap_before = h["frameNumber"] - seg.last_frame - 1 + seg.gaps += 1 + seg.gap_frames += gap_before + self.gaps += 1 + self.gap_frames += gap_before + if len(self.gap_list) < GAP_LIST_MAX: + self.gap_list.append({"segment": seg.id, "after": seg.last_frame, "next": h["frameNumber"], "missing": gap_before, + "rx_epoch_ms": rx_wall // 1_000_000}) + if seg.gaps <= 5 or seg.gaps % 100 == 0: + self.log.warning(f"gap: {gap_before} frame(s) missing between {seg.last_frame} and {h['frameNumber']} (segment {seg.id})") + if [h["ppl"], h["lpf"]] != seg.dims or h["nchan"] != seg.nchan: + seg.dims_changes += 1 + seg.dims, seg.nchan = [h["ppl"], h["lpf"]], h["nchan"] + seg.last_frame = h["frameNumber"] + seg.last_ts = h["frameTimestamp_s"] + seg.frames += 1 + seg.t_last_ms = rx_wall // 1_000_000 + seg.mono_last_ns = rx_mono + meta = {"seq": self.frames, "segment": seg.id, "frameNumber": h["frameNumber"], "frameTimestamp_s": h["frameTimestamp_s"], + "dims": [h["ppl"], h["lpf"]], "nchan": h["nchan"], "bytes": n, "gap_before": gap_before, + "rx_epoch_ms": rx_wall // 1_000_000, "rx_wall_ns": rx_wall, "rx_mono_ns": rx_mono, "preview": want_preview} + try: + # copy the pixels only for a preview frame; the queue entry is otherwise tiny + self.q.put_nowait((meta, bytes(parts[1].buffer) if want_preview else None)) + depth = self.q.qsize() + if depth > self.queue_max: + self.queue_max = depth + except queue.Full: + self.dropped_by_queue += 1 + if self.dropped_by_queue <= 5 or self.dropped_by_queue % 100 == 0: + self.log.warning(f"queue full ({self.q.maxsize}): dropped frame {h['frameNumber']} ({self.dropped_by_queue} dropped so far)") + + def snapshot(self) -> dict: + with self.lock: + qmax, self.queue_max = self.queue_max, self.q.qsize() + return { + "address": self.address, "connected": self.connected, "connects": self.connects, "disconnects": self.disconnects, + "frames": self.frames, "rx_bytes": self.bytes, "bad_frames": self.bad_frames, "dropped_by_queue": self.dropped_by_queue, + "gaps": self.gaps, "gap_frames": self.gap_frames, "gap_list": list(self.gap_list[-STATS_GAP_LIST:]), + "gap_list_truncated": len(self.gap_list) >= GAP_LIST_MAX, + "segments": len(self.segments), "current_segment": self.segments[-1].id if self.segments else None, + "segment_list": [s.as_dict() for s in self.segments], + "queue_depth": self.q.qsize(), "queue_max": qmax, "queue_limit": self.q.maxsize, + } + + +# ── the ROS side ─────────────────────────────────────────────────────────────────────────── +class ScanImageFramesNode(Node): + def __init__(self, a: argparse.Namespace) -> None: + super().__init__("scanimage_frames") + self.a = a + self.meta_pub = self.create_publisher(String, META_TOPIC, QOS) + self.stats_pub = self.create_publisher(String, STATS_TOPIC, QOS) + self.preview_meta_pub = self.create_publisher(String, PREVIEW_META_TOPIC, QOS) if a.preview_hz > 0 else None + self.preview_pubs: dict[int, object] = {} + self.channel_ids = [int(x) for x in a.channel_ids.split(",") if x.strip()] if a.channel_ids else [] + self.preview_channels: set[int] = set() + if a.preview_hz > 0: + # Declared up front: creating a publisher on the first frame of the first acquisition held + # the receive side ~0.85 s (zenoh declaration under the GIL) in the off-rig smoke run. + for ch in (self.channel_ids or list(range(1, a.preview_channels + 1))): + self._preview_pub(ch) + self.fixed_range = parse_range(a.preview_range) + self._auto_ranges: dict[tuple[int, int], tuple[float, float]] = {} # (segment, channel) → (lo, hi) + self.published = 0 + self.previews = 0 + self._hops_ms: list[float] = [] # in-node hop (publish − receive) since the last stats message + self._dump = open(a.dump, "a", encoding="utf-8") if a.dump else None + self._stats_prev = (mono_ns(), 0) + self.rx = FrameReceiver(a.address, a.queue_depth, a.rcv_hwm, a.preview_hz, self.get_logger()) + self.rx.start() + self._stop = threading.Event() + self._pub_thread = threading.Thread(target=self._publish_loop, name="ros-pub", daemon=True) + self._pub_thread.start() + self.create_timer(max(0.1, a.stats_every), self._stats) + prev = f"preview ≤ {a.preview_hz:g} Hz/channel {a.preview_size}² PNG → {PREVIEW_TOPIC_PREFIX}" if a.preview_hz > 0 else "preview off" + self.get_logger().info(f"SUB {a.address} → {META_TOPIC}, {STATS_TOPIC} every {a.stats_every:g}s, {prev}, queue {a.queue_depth}") + + # ── plumbing ── + def _send(self, pub, obj: dict, dump_topic: str | None = None) -> None: + m = String() + m.data = json.dumps(obj, separators=(",", ":")) + pub.publish(m) + if self._dump is not None and dump_topic: + self._dump.write(json.dumps({"topic": dump_topic, **obj}, separators=(",", ":")) + "\n") + + def _publish_loop(self) -> None: + a = self.a + while not self._stop.is_set(): + try: + meta, pixels = self.rx.q.get(timeout=0.1) + except queue.Empty: + continue + if a.debug_stall_s > 0 and meta["seq"] == a.debug_stall_at: + self.get_logger().warning(f"debug: stalling the publisher for {a.debug_stall_s:g}s at frame seq {meta['seq']}") + time.sleep(a.debug_stall_s) + meta["queue_depth"] = self.rx.q.qsize() + meta["pub_mono_ns"] = mono_ns() + if len(self._hops_ms) < 100_000: + self._hops_ms.append((meta["pub_mono_ns"] - meta["rx_mono_ns"]) / 1e6) + out = {"src": SRC, **{k: v for k, v in meta.items() if k != "rx_wall_ns"}} + try: + self._send(self.meta_pub, out, META_TOPIC) + self.published += 1 + if pixels is not None: + self._preview(meta, pixels) + except Exception as exc: # never let one frame kill the publisher thread + self.get_logger().error(f"publish failed: {exc!r}") + + def _preview_pub(self, ch: int): + pub = self.preview_pubs.get(ch) + if pub is None: + pub = self.preview_pubs[ch] = self.create_publisher(CompressedImage, f"{PREVIEW_TOPIC_PREFIX}{ch}", QOS) + self.preview_channels.add(ch) + return pub + + def _preview(self, meta: dict, pixels: bytes) -> None: + a = self.a + h = {"ppl": meta["dims"][0], "lpf": meta["dims"][1], "nchan": meta["nchan"]} + arr = planes(pixels, h) + stamp_s, stamp_ns = divmod(int(meta["rx_wall_ns"]), 1_000_000_000) + for plane_i in range(h["nchan"]): + ch = self.channel_ids[plane_i] if plane_i < len(self.channel_ids) else plane_i + 1 + small = downsample(arr[plane_i], a.preview_size) + if self.fixed_range is not None: + lo, hi = self.fixed_range + mode = "fixed" + else: + key = (meta["segment"], ch) + if key not in self._auto_ranges: + self._auto_ranges[key] = auto_range(small) + for old in [k for k in self._auto_ranges if k[0] < meta["segment"] - 2]: + del self._auto_ranges[old] + lo, hi = self._auto_ranges[key] + mode = "auto" + png = png_gray(to_uint8(small, lo, hi)) + pub = self._preview_pub(ch) + img = CompressedImage() + img.header.stamp.sec = int(stamp_s) + img.header.stamp.nanosec = int(stamp_ns) + img.header.frame_id = f"scanimage/ch{ch}/s{meta['segment']}/f{meta['frameNumber']}" + img.format = "png" + img.data = png + pub.publish(img) + self.previews += 1 + self._send(self.preview_meta_pub, { + "src": SRC, "image_topic": f"{PREVIEW_TOPIC_PREFIX}{ch}", "channel": ch, "plane": plane_i, "segment": meta["segment"], + "frameNumber": meta["frameNumber"], "frameTimestamp_s": meta["frameTimestamp_s"], "dims": meta["dims"], + "size": [a.preview_size, a.preview_size], + "mapping": {"mode": mode, "lo": lo, "hi": hi, "to": "uint8 linear, clipped", "downsample": "block mean" if (h["lpf"] % a.preview_size == 0 and h["ppl"] % a.preview_size == 0 and h["lpf"] >= a.preview_size and h["ppl"] >= a.preview_size) else "nearest"}, + "png_bytes": len(png), "rx_epoch_ms": meta["rx_epoch_ms"], "rx_mono_ns": meta["rx_mono_ns"], "pub_mono_ns": mono_ns(), + }, PREVIEW_META_TOPIC) + + def _stats(self) -> None: + now = mono_ns() + snap = self.rx.snapshot() + t_prev, f_prev = self._stats_prev + dt = (now - t_prev) / 1e9 + self._stats_prev = (now, snap["frames"]) + hops, self._hops_ms = self._hops_ms, [] + hops.sort() + hop = ({"n": len(hops), "p50_ms": round(hops[len(hops) // 2], 3), "p99_ms": round(hops[min(len(hops) - 1, int(len(hops) * 0.99))], 3), + "max_ms": round(hops[-1], 3)} if hops else {"n": 0}) + obj = { + "src": SRC, "t_ms": wall_ns() // 1_000_000, "uptime_s": round((now - self.rx.started_mono) / 1e9, 1), + **snap, "published": self.published, + "rate_hz": round((snap["frames"] - f_prev) / dt, 2) if dt > 0 else None, + "hop": hop, # receive → publish inside this node, over the last interval + "preview": {"hz": self.a.preview_hz, "size": self.a.preview_size, "published": self.previews, + "channels": sorted(self.preview_channels), "range": self.a.preview_range}, + } + self._send(self.stats_pub, obj, STATS_TOPIC) + if self._dump is not None: + self._dump.flush() + + def destroy_node(self): + self.close() # stop the receive + publish threads before the publishers go away + return super().destroy_node() + + def close(self) -> None: + if getattr(self, "_closed", False): # destroy_node() and main()'s finally both call it + return + self._closed = True + self._stop.set() + for _ in range(3): # rclpy forwards SIGINT to Python's handler too: its KeyboardInterrupt may land in a join + try: + self.rx.close() + self._pub_thread.join(timeout=2.0) + break + except KeyboardInterrupt: + continue + if self._dump is not None: + self._dump.flush() + self._dump.close() + + +def main(argv: list[str] | None = None) -> None: + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--address", default=DEFAULT_ADDRESS, help=f"ScanImage's ZMQAddress (default {DEFAULT_ADDRESS})") + p.add_argument("--queue-depth", type=int, default=64, help="frames buffered between the receive thread and the ROS publisher; a full queue drops the newest frame (counted; default 64)") + p.add_argument("--rcv-hwm", type=int, default=32, help="ZeroMQ receive high-water mark in messages (default 32; beyond it libzmq stops reading and the PUBLISHER drops)") + p.add_argument("--stats-every", type=float, default=1.0, help=f"seconds between {STATS_TOPIC} messages (default 1)") + p.add_argument("--preview-hz", type=float, default=5.0, help="preview images per channel per second at most (default 5; 0 = no preview)") + p.add_argument("--preview-size", type=int, default=128, help="preview edge in pixels (default 128)") + p.add_argument("--preview-range", default="auto", help="int16 → 8-bit display mapping: 'auto' (1–99.5 percentile of the first preview frame of each segment, then held) or 'lo:hi' fixed (default auto)") + p.add_argument("--preview-channels", type=int, default=4, help="preview publishers declared at start, ch1..chN (default 4 = ScanImage's channels); more are added on demand") + p.add_argument("--channel-ids", default="", help="comma list naming the planes, e.g. '1,3' when ScanImage's channelSave is [1 3] (default: 1..nChannels)") + p.add_argument("--dump", default=None, help="also append every frame_meta / frame_stats / preview meta line to this JSONL file (tools/zmq_latency_report.py reads it when there is no bag)") + p.add_argument("--debug-stall-at", type=int, default=0, help="(harness) stall the ROS publisher at this received-frame seq …") + p.add_argument("--debug-stall-s", type=float, default=0.0, help="(harness) … for this many seconds, to exercise dropped_by_queue") + a = p.parse_args(argv) + if a.preview_size < 8 or a.preview_size > 1024: + raise SystemExit("--preview-size must be 8..1024") + parse_range(a.preview_range) + raise_timer_resolution() + rclpy.init() + node = ScanImageFramesNode(a) + try: + spin(node) + finally: + node.close() + + +if __name__ == "__main__": + main() diff --git a/ros2-poc/pixi.toml b/ros2-poc/pixi.toml index fbec7aeb..3d090d56 100644 --- a/ros2-poc/pixi.toml +++ b/ros2-poc/pixi.toml @@ -129,6 +129,8 @@ bias-cmd = "python nodes/bias_cmd.py" # ScanImage (built-in JSON server on the ScanImage PC, LAB-303/LAB-306): node + one-shot command. scanimage-node = "python nodes/scanimage_node.py" scanimage-cmd = "python nodes/scanimage_cmd.py" +# ScanImage's ZeroMQ frame stream (LAB-303 Block R): SUB → /scanimage/frame_meta, frame_stats, preview/ch. +scanimage-frames = "python nodes/scanimage_frames_node.py" # ── the bridge node: bridge.py's hub + log writer + dispatcher on ROS input ── bridge-node = "python nodes/bridge_node.py" # ── simulators (the root workspace's fictrac_sim.py, unmodified) ──────────── @@ -140,7 +142,11 @@ si-mock = "python tests/si_json_mock.py" # ── the one command a rig runs ────────────────────────────────────────────── rig = "python nodes/rig.py" # ── acceptance test (LAB-303 PoC criteria): whole graph + Studio emulator + model fly ── -test = "python tests/test_ros2_poc.py" +# `test` = the frame-stream harness (mock ZMQ publisher, ≈ 40 s) THEN the acceptance run; `-- args` reach the latter. +test = { cmd = "python tests/test_ros2_poc.py", depends-on = ["test-frames"] } +test-frames = "python tests/test_scanimage_frames.py" +# latency / gaps / queue-depth report from a bag or the node's --dump JSONL (+ the matching TIFF) +zmq-latency = "python tools/zmq_latency_report.py" studio-emulator = "python tests/studio_emulator.py" # ── Plan B spike: MCAP record + converter + typed interfaces ────────────────── # bag → behavior_v2 NDJSON (also `-- --level behavior_v1|full`, `-- --order time`); diff --git a/ros2-poc/tests/test_scanimage_frames.py b/ros2-poc/tests/test_scanimage_frames.py new file mode 100644 index 00000000..1eb84cb3 --- /dev/null +++ b/ros2-poc/tests/test_scanimage_frames.py @@ -0,0 +1,531 @@ +#!/usr/bin/env python3 +"""test_scanimage_frames.py — the ScanImage ZMQ frame-stream node, off-rig (LAB-303 Block R). + + pixi run test-frames # from ros2-poc/ (≈ 45 s); also the first half of `pixi run test` + pixi run test-frames -- --keep --mock ../scanimage-2026-bergamo/tests/zmq-frames/mock_zmq_publisher.py + +Brings up a Zenoh router, nodes/scanimage_frames_node.py and an in-process ROS subscriber, then +drives the node with the frame publisher stand-in (tests/zmq_frames_mock.py: Hannah's +mock_zmq_publisher.py format and CLI; `--mock` runs the original) at 29.08 Hz × 512 × 512 × 2 ch: + + A one acquisition (87 frames) every frame counted, dims/nchan right, frame numbers contiguous, + segment 1 from frame 1, previews ≤ 5 Hz per channel, 128² PNG, + display mapping recorded, in-node hop small + B gap injection (--drop-every 10) a NEW segment (frame counter restarted), the 9 visible gaps + counted (a dropped LAST frame is invisible to any subscriber) + C publisher restart, no start delay the slow-joiner loss recorded as the segment's + missed_before_first, reconnects counted + D1 slow consumer, publisher HWM 1000 the node SIGSTOPped 1.5 s mid-stream: the publisher's timing is + unaffected, nothing lost (it buffered — what ScanImage's memory + does), the backlog visible as queue depth / hop (POSIX only) + D2 slow consumer, publisher HWM 20 same pause: the publisher DROPS, the node's accounting names + every missing frame (gaps / tail), publisher timing unaffected + E node restart + queue back-pressure clean SIGINT exit; a second node with a tiny queue and a stalled + publisher thread counts dropped_by_queue exactly + F tools/zmq_latency_report.py on the node's --dump JSONL and on a bag written with + nodes/recorder.py from the received messages (the bridge's + record path), with a synthetic ScanImage-style TIFF for the + TIFF-clock alignment + G bridge_node registration the stream topics are in the bag's topic map; nothing derives them + +Exit 0 when every check passes; artifacts kept with --keep. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import os +import shutil +import signal +import socket +import struct +import subprocess +import sys +import tempfile +import threading +import time +from pathlib import Path + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parent +NODES = ROOT / "nodes" +PY = sys.executable +sys.path.insert(0, str(NODES)) +from _common import RELIABLE_QOS, SIGKILL, popen_group, signal_group, wait_port, wall_ns # noqa: E402 + +import rclpy # noqa: E402 +from rclpy.executors import SingleThreadedExecutor # noqa: E402 +from rclpy.node import Node # noqa: E402 +from rclpy.serialization import serialize_message # noqa: E402 +from sensor_msgs.msg import CompressedImage # noqa: E402 +from std_msgs.msg import String # noqa: E402 + +META_TOPIC = "/scanimage/frame_meta" +STATS_TOPIC = "/scanimage/frame_stats" +PREVIEW_META_TOPIC = "/scanimage/preview/meta" +PREVIEW_PREFIX = "/scanimage/preview/ch" +RATE = 29.08 +DIMS = (512, 512) +NCH = 2 + +results: list[tuple[str, bool, str]] = [] + + +def check(name: str, ok: bool, info: str = "") -> bool: + results.append((name, bool(ok), info)) + print(f" {'PASS' if ok else 'FAIL'} {name}{'' if ok else ' — ' + info}", file=sys.stderr) + return bool(ok) + + +def note(text: str) -> None: + print(f" info {text}", file=sys.stderr) + + +def free_tcp_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +class Proc: + def __init__(self, name: str, argv: list[str], logdir: Path) -> None: + self.name = name + self.argv = argv + self.log = open(logdir / f"{name}.log", "a", encoding="utf-8") + self.p: subprocess.Popen | None = None + self.t_start = 0.0 + + def start(self) -> "Proc": + self.t_start = time.perf_counter() + self.p = popen_group(self.argv, stdout=self.log, stderr=subprocess.STDOUT, text=True) + return self + + def alive(self) -> bool: + return self.p is not None and self.p.poll() is None + + def wait(self, timeout: float) -> float: + """Wait for exit → elapsed seconds since start.""" + assert self.p + self.p.wait(timeout) + return time.perf_counter() - self.t_start + + def stop(self, grace: float = 5.0) -> int | None: + if not self.alive(): + return self.p.returncode if self.p else None + try: + signal_group(self.p, signal.SIGINT) + self.p.wait(grace) + except subprocess.TimeoutExpired: + signal_group(self.p, signal.SIGTERM) + try: + self.p.wait(3) + except subprocess.TimeoutExpired: + signal_group(self.p, SIGKILL) + except (ProcessLookupError, OSError, ValueError): + pass + self.log.flush() + return self.p.returncode + + +class Collector(Node): + """The harness's own subscriber: everything the frames node publishes, with receive stamps.""" + + def __init__(self) -> None: + super().__init__("frames_harness") + self.lock = threading.Lock() + self.meta: list[tuple[int, dict, String]] = [] # (wall_ns, obj, msg) + self.stats: list[tuple[int, dict, String]] = [] + self.pmeta: list[tuple[int, dict, String]] = [] + self.images: dict[int, list[tuple[int, CompressedImage]]] = {} + self.create_subscription(String, META_TOPIC, lambda m: self._json(self.meta, m), RELIABLE_QOS) + self.create_subscription(String, STATS_TOPIC, lambda m: self._json(self.stats, m), RELIABLE_QOS) + self.create_subscription(String, PREVIEW_META_TOPIC, lambda m: self._json(self.pmeta, m), RELIABLE_QOS) + for ch in (1, 2, 3, 4): + self.create_subscription(CompressedImage, f"{PREVIEW_PREFIX}{ch}", lambda m, ch=ch: self._img(ch, m), RELIABLE_QOS) + + def _json(self, dest: list, m: String) -> None: + t = wall_ns() + try: + obj = json.loads(m.data) + except json.JSONDecodeError: + return + with self.lock: + dest.append((t, obj, m)) + + def _img(self, ch: int, m: CompressedImage) -> None: + t = wall_ns() + with self.lock: + self.images.setdefault(ch, []).append((t, m)) + + # snapshots + def metas(self) -> list[dict]: + with self.lock: + return [o for _, o, _ in self.meta] + + def last_stats(self) -> dict | None: + with self.lock: + return self.stats[-1][1] if self.stats else None + + def n_images(self, ch: int) -> int: + with self.lock: + return len(self.images.get(ch, [])) + + def clear_images(self) -> None: + with self.lock: + self.images.clear() + self.pmeta.clear() + + +def wait_for(pred, timeout: float, period: float = 0.05) -> bool: + end = time.monotonic() + timeout + while time.monotonic() < end: + if pred(): + return True + time.sleep(period) + return bool(pred()) + + +def wait_settled(col: Collector, timeout: float = 12.0) -> dict | None: + """Wait until the node's stats say the queue is drained and the frame count stopped moving.""" + last = None + same_since = None + end = time.monotonic() + timeout + while time.monotonic() < end: + st = col.last_stats() + if st is not None: + key = (st["frames"], st["published"], st["dropped_by_queue"]) + if st["published"] + st["dropped_by_queue"] == st["frames"] and st["queue_depth"] == 0: + if key == last: + if same_since is not None and time.monotonic() - same_since >= 1.2: + return st + else: + same_since = time.monotonic() + last = key + time.sleep(0.1) + return col.last_stats() + + +def segment(st: dict | None, sid: int) -> dict | None: + if not st: + return None + for s in st.get("segment_list", []): + if s["id"] == sid: + return s + return None + + +def png_size(data: bytes) -> tuple[int, int] | None: + if data[:8] != b"\x89PNG\r\n\x1a\n" or data[12:16] != b"IHDR": + return None + return struct.unpack(">II", data[16:24]) + + +def write_si_tiff(path: Path, frames: list[tuple[int, float]], nch: int, epoch: list[float]) -> None: + """A classic little-endian TIFF the way ScanImage lays it out: one 8×8 int16 page per channel per + frame, each page's ImageDescription carrying frameNumbers / frameTimestamps_sec / epoch. Enough for + tools/zmq_latency_report.py's stdlib reader (tifffile is not in the env).""" + w = h = 8 + pixels = bytes(w * h * 2) + out = bytearray(b"II*\x00" + struct.pack(" int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--mock", default=str(HERE / "zmq_frames_mock.py"), help="the frame publisher to drive the node with (default: tests/zmq_frames_mock.py)") + ap.add_argument("--keep", action="store_true", help="keep the work dir") + ap.add_argument("--workdir", default=None) + ap.add_argument("--no-router", action="store_true", help="a Zenoh router is already running on 127.0.0.1:7447") + ap.add_argument("--pause-s", type=float, default=1.5, help="how long the node is SIGSTOPped in D1/D2") + a = ap.parse_args(argv) + + work = Path(a.workdir) if a.workdir else Path(tempfile.mkdtemp(prefix="si-frames-test-")) + work.mkdir(parents=True, exist_ok=True) + logdir = work / "proc-logs" + logdir.mkdir(exist_ok=True) + mock = Path(a.mock) + if not mock.exists(): + print(f"[test] mock publisher not found: {mock}", file=sys.stderr) + return 2 + bundled = mock.resolve() == (HERE / "zmq_frames_mock.py").resolve() + print(f"[test] work dir {work}; publisher {mock}{'' if bundled else ' (external — D2 needs the bundled --sndhwm, skipped)'}", file=sys.stderr) + procs: list[Proc] = [] + port = free_tcp_port() + address = f"tcp://127.0.0.1:{port}" + dump_a = work / "node-a.jsonl" + dump_b = work / "node-b.jsonl" + posix = os.name != "nt" + + def start(name: str, argv: list[str]) -> Proc: + pr = Proc(name, argv, logdir).start() + procs.append(pr) + return pr + + def run_mock(name: str, frames: int, **kw) -> Proc: + argv = [PY, str(mock), "--address", address, "--frames", str(frames), "--rate", str(RATE), + "--ppl", str(DIMS[0]), "--lpf", str(DIMS[1]), "--nch", str(NCH)] + for k, v in kw.items(): + argv += [f"--{k.replace('_', '-')}", str(v)] + return start(name, argv) + + node_a = node_b = None + col = ex = None + try: + # the router FIRST: creating any rclpy node (our subscriber included) blocks until it answers + if not a.no_router: + start("router", ["ros2", "run", "rmw_zenoh_cpp", "rmw_zenohd"]) + check("zenoh router listening on 127.0.0.1:7447", wait_port("127.0.0.1", 7447, 20.0)) + rclpy.init() + col = Collector() + ex = SingleThreadedExecutor() + ex.add_node(col) + threading.Thread(target=ex.spin, name="harness-spin", daemon=True).start() + node_a = start("node-a", [PY, str(NODES / "scanimage_frames_node.py"), "--address", address, "--stats-every", "0.5", + "--preview-hz", "5", "--preview-size", "128", "--dump", str(dump_a)]) + check("node up: frame_stats arrives (0 frames, not connected)", + wait_for(lambda: col.last_stats() is not None, 20.0) and col.last_stats()["frames"] == 0, json.dumps(col.last_stats())) + time.sleep(0.5) + + # ── A: one acquisition ────────────────────────────────────────────────────────── + n_a = 87 + m = run_mock("mock-a", n_a, start_delay=1.0) + el = m.wait(30) + st = wait_settled(col) + metas = col.metas() + seg1 = segment(st, 1) + check(f"A: all {n_a} frames received and published", st and st["frames"] == n_a and st["published"] == n_a and len(metas) == n_a, + f"stats frames={st and st['frames']} published={st and st['published']} meta msgs={len(metas)}") + check("A: dims 512×512 and nchan 2 on every frame_meta", all(tuple(o["dims"]) == DIMS and o["nchan"] == NCH and o["bytes"] == DIMS[0] * DIMS[1] * NCH * 2 for o in metas), + str({(tuple(o["dims"]), o["nchan"]) for o in metas})) + check("A: frame numbers 1..87 contiguous, one segment from frame 1, 0 gaps", + [o["frameNumber"] for o in metas] == list(range(1, n_a + 1)) and seg1 and seg1["first_frame"] == 1 and seg1["missed_before_first"] == 0 and st["gaps"] == 0 and st["segments"] == 1, + json.dumps(seg1)) + check("A: node connected to the publisher (monitor) and saw it go away", st["connects"] == 1 and st["disconnects"] >= 1, f"connects={st['connects']} disconnects={st['disconnects']}") + check("A: header timestamps advance at the publisher's rate", seg1 and seg1["rate_hz"] and abs(seg1["rate_hz"] - RATE) < 1.0, f"rate_hz={seg1 and seg1['rate_hz']}") + hops = sorted((o["pub_mono_ns"] - o["rx_mono_ns"]) / 1e6 for o in metas[3:]) + hop_p99 = hops[int(len(hops) * 0.99) - 1] if hops else None + check("A: in-node hop p99 (receive → publish) < 100 ms after the first 3 frames", hop_p99 is not None and hop_p99 < 100.0, f"p99={hop_p99} ms, max={hops[-1] if hops else None}") + note(f"A: in-node hop p50 {hops[len(hops) // 2]:.2f} ms, p99 {hop_p99:.2f} ms, max {hops[-1]:.2f} ms; queue max {max(o['queue_depth'] for o in metas)}") + span = (metas[-1]["rx_mono_ns"] - metas[0]["rx_mono_ns"]) / 1e9 + for ch in (1, 2): + n_img = col.n_images(ch) + check(f"A: preview ch{ch} published and bounded: 1 ≤ {n_img} ≤ 5 Hz × {span:.1f} s + 1", 1 <= n_img <= int(5 * span) + 1, f"{n_img} images in {span:.2f} s") + with col.lock: + imgs = list(col.images.get(1, [])) + pm = [o for _, o, _ in col.pmeta] + sizes = {png_size(bytes(m.data)) for _, m in imgs} + check("A: preview images are 128×128 PNG (format 'png', frame_id names channel/segment/frame)", + sizes == {(128, 128)} and all(m.format == "png" and m.header.frame_id.startswith("scanimage/ch1/s1/f") for _, m in imgs), f"sizes={sizes}") + check("A: preview meta records the display mapping (lo < hi) for every image and names the image topic", + len(pm) == sum(col.n_images(c) for c in (1, 2)) and all(o["mapping"]["lo"] < o["mapping"]["hi"] and o["image_topic"].startswith(PREVIEW_PREFIX) and o["size"] == [128, 128] for o in pm), + f"{len(pm)} preview meta vs {sum(col.n_images(c) for c in (1, 2))} images") + check("A: publisher ran at its own pace (no back-pressure from the subscriber)", el < n_a / RATE + 1.0 + 1.5, f"{el:.2f} s for {n_a} frames + 1 s delay") + + # ── B: gap injection + a restarted frame counter ──────────────────────────────── + n_b = 100 + m = run_mock("mock-b", n_b, start_delay=1.0, drop_every=10) + m.wait(30) + st = wait_settled(col) + seg2 = segment(st, 2) + check("B: the restarted frame counter opened segment 2 from frame 1", st and st["segments"] == 2 and seg2 and seg2["first_frame"] == 1 and seg2["missed_before_first"] == 0, json.dumps(seg2)) + check("B: 9 gaps of 1 frame (10, 20, … 90) counted; frame 100 — the dropped LAST frame — is invisible to a subscriber", + seg2 and seg2["gaps"] == 9 and seg2["gap_frames"] == 9 and seg2["frames"] == 90 and seg2["last_frame"] == 99, json.dumps(seg2)) + gl = [g for g in st["gap_list"] if g["segment"] == 2] + check("B: gap_list names each gap (after/next/missing)", [(g["after"], g["next"], g["missing"]) for g in gl] == [(k - 1, k + 1, 1) for k in range(10, 100, 10)], json.dumps(gl)) + metas_b = [o for o in col.metas() if o["segment"] == 2] + check("B: frame_meta carries gap_before on the frame after each gap", [o["frameNumber"] for o in metas_b if o["gap_before"] == 1] == list(range(11, 100, 10)), "") + + # ── C: slow joiner (publisher binds and sends at once) ────────────────────────── + n_c = 60 + m = run_mock("mock-c", n_c, start_delay=0.0) + m.wait(30) + st = wait_settled(col) + seg3 = segment(st, 3) + check("C: a new segment after the publisher restart; the socket reconnected (connects = 3)", st and st["segments"] == 3 and st["connects"] == 3, f"segments={st and st['segments']} connects={st and st['connects']}") + check("C: the slow-joiner loss is recorded: missed_before_first = first_frame − 1 ≥ 1, no gaps after it", + seg3 and seg3["missed_before_first"] == seg3["first_frame"] - 1 >= 1 and seg3["gaps"] == 0 and seg3["frames"] == n_c - seg3["missed_before_first"], json.dumps(seg3)) + note(f"C: first frame seen {seg3 and seg3['first_frame']} ({seg3 and seg3['missed_before_first']} missed before it)") + + # ── D: slow consumer (the node paused) ────────────────────────────────────────── + def paused_run(name: str, frames: int, **kw) -> tuple[float, dict | None, dict | None]: + m = run_mock(name, frames, start_delay=1.0, **kw) + sid = (col.last_stats() or {}).get("segments", 0) + 1 + wait_for(lambda: len([o for o in col.metas() if o["segment"] == sid]) >= 20, 10.0) + os.kill(node_a.p.pid, signal.SIGSTOP) + time.sleep(a.pause_s) + os.kill(node_a.p.pid, signal.SIGCONT) + el = m.wait(60) + st = wait_settled(col, 20.0) + return el, st, segment(st, sid) + + if posix: + n_d = 150 + budget = n_d / RATE + 1.0 + 1.5 + el, st, seg4 = paused_run("mock-d1", n_d) + check(f"D1: publisher (HWM 1000) finished on time while the node was paused {a.pause_s:g} s", el < budget, f"{el:.2f} s vs budget {budget:.2f} s") + check("D1: nothing lost — the publisher buffered the pause (what ScanImage's memory does); accounting exact", + seg4 and seg4["frames"] == n_d and seg4["gap_frames"] == 0 and seg4["missed_before_first"] == 0 and st["dropped_by_queue"] == 0, json.dumps(seg4)) + metas_d = [o for o in col.metas() if o["segment"] == 4] + worst = max(((o["pub_mono_ns"] - o["rx_mono_ns"]) / 1e6 for o in metas_d), default=0) + check("D1: the backlog is visible: queue depth rose after the pause", metas_d and max(o["queue_depth"] for o in metas_d) >= 2, f"queue max {metas_d and max(o['queue_depth'] for o in metas_d)}, hop max {worst:.0f} ms") + if bundled: + el, st, seg5 = paused_run("mock-d2", n_d, sndhwm=20) + check(f"D2: publisher (HWM 20) finished on time while the node was paused {a.pause_s:g} s", el < budget, f"{el:.2f} s vs budget {budget:.2f} s") + lost = seg5 and (seg5["gap_frames"] + (n_d - seg5["last_frame"]) + seg5["missed_before_first"]) + check("D2: the publisher dropped frames during the pause and the node counted them (gaps and/or a missing tail)", + seg5 and lost and lost > 0 and seg5["frames"] + seg5["gap_frames"] == seg5["last_frame"] - seg5["first_frame"] + 1 and st["dropped_by_queue"] == 0, + json.dumps(seg5)) + note(f"D2: {seg5 and seg5['frames']} received, {seg5 and seg5['gap_frames']} in {seg5 and seg5['gaps']} gaps, tail {seg5 and n_d - seg5['last_frame']} missing → {lost} lost of {n_d}") + else: + note("D1/D2 skipped: no SIGSTOP on Windows (the lab-day procedure pauses the node by hand)") + + # ── E: node restart (clean exit) + queue back-pressure ────────────────────────── + rc = node_a.stop() + check("E: node exits cleanly on SIGINT (code 0)", rc == 0, f"rc={rc}") + check("E: the --dump file holds the frame_meta and frame_stats lines", dump_a.exists() and sum(1 for l in dump_a.read_text().splitlines() if '"/scanimage/frame_meta"' in l) == st["frames"], + f"{dump_a}") + prev_frames = st["frames"] + node_b = start("node-b", [PY, str(NODES / "scanimage_frames_node.py"), "--address", address, "--stats-every", "0.5", "--preview-hz", "0", + "--queue-depth", "8", "--debug-stall-at", "20", "--debug-stall-s", "1.0", "--dump", str(dump_b)]) + col.clear_images() + check("E: restarted node up (fresh counters)", wait_for(lambda: (col.last_stats() or {}).get("frames") == 0 and (col.last_stats() or {}).get("uptime_s", 99) < 5, 20.0), json.dumps(col.last_stats())) + time.sleep(0.5) + n_e = 100 + m = run_mock("mock-e", n_e, start_delay=1.0) + m.wait(30) + st = wait_settled(col) + check("E: a stalled publisher thread + queue 8 → dropped_by_queue counted, received == published + dropped exactly", + st and st["frames"] == n_e and st["dropped_by_queue"] >= 1 and st["published"] + st["dropped_by_queue"] == st["frames"] and st["gaps"] == 0, + json.dumps({k: st and st[k] for k in ("frames", "published", "dropped_by_queue", "gaps", "queue_limit")})) + check("E: --preview-hz 0 publishes no preview", st["preview"]["published"] == 0 and col.n_images(1) == 0 and col.n_images(2) == 0, json.dumps(st["preview"])) + note(f"E: {st['dropped_by_queue']} dropped by the queue (limit {st['queue_limit']}) during a 1 s stall at 29 Hz") + + # ── F: the latency report (dump + bag + TIFF) ─────────────────────────────────── + tool = ROOT / "tools" / "zmq_latency_report.py" + metas1 = [o for o in col.metas() if o["segment"] == 1][:n_a] + tiff = work / "mock_a.tif" + epoch_ms = metas1[0]["rx_epoch_ms"] - 20 # "acquisition started 20 ms before the first frame reached the node" + e = dt.datetime.fromtimestamp(epoch_ms / 1000.0, dt.timezone.utc) + write_si_tiff(tiff, [(o["frameNumber"], o["frameTimestamp_s"] - 0.005) for o in metas1], NCH, [e.year, e.month, e.day, e.hour, e.minute, e.second + e.microsecond / 1e6]) + rep_json = work / "report-a.json" + r = subprocess.run([PY, str(tool), "--jsonl", str(dump_a), "--tiff", str(tiff), "--utc-offset-h", "0", "--json", str(rep_json)], capture_output=True, text=True, timeout=120) + ok = r.returncode == 0 and rep_json.exists() + check("F: zmq_latency_report runs on the node's --dump JSONL (+ TIFF)", ok, (r.stdout + r.stderr)[-600:]) + if ok: + rep = json.loads(rep_json.read_text()) + s1 = next((s for s in rep["segments"] if s["id"] == 1), None) + s2 = next((s for s in rep["segments"] if s["id"] == 2), None) + check("F: report segment 1 = 87 frames, stream-delay/hop percentiles, queue histogram", + s1 and s1["frames"] == n_a and s1["stream_delay_rel_ms"]["n"] == n_a and s1["node_hop_ms"]["p99"] is not None and sum(s1["queue_depth"]["hist"].values()) == n_a, json.dumps(s1 and {k: s1[k] for k in ("frames", "stream_delay_rel_ms", "queue_depth")})) + check("F: report segment 2 lists the 9 injected gaps", s2 and s2["gaps"] == 9 and len(s2["gap_list"]) == 9, json.dumps(s2 and s2["gap_list"])) + t = s1 and s1.get("tiff") + check("F: TIFF alignment: 87/87 frames matched, 2 pages per frame, ZMQ − TIFF timestamp = +5 ms", + t and t["matched_frames"] == n_a and t["pages_per_frame"] == NCH and t["frame_meta_count_equals_tiff_frames"] and abs(t["zmq_minus_tiff_ts_ms"]["p50"] - 5.0) < 0.01, json.dumps(t)) + check("F: absolute latency from the TIFF epoch is finite and plausible (20 ms + the stream delay)", + t and t.get("absolute_latency_ms", {}).get("n") == n_a and 15.0 <= t["absolute_latency_ms"]["min"] <= 2000.0, json.dumps(t and t.get("absolute_latency_ms"))) + check("F: the report's pass bars include the meta-latency and gap bars per segment", any("meta latency" in b["bar"] for b in rep["bars"]) and any(b["bar"].startswith("0 gaps") and b["segment"] == 2 and not b["pass"] for b in rep["bars"]), json.dumps(rep["bars"])) + print(r.stdout[-1500:], file=sys.stderr) + # a bag the way the bridge records it (raw CDR, arrival stamps), then the report's bag path + from recorder import Recorder # noqa: E402 (nodes/) + + bag = work / "frames.bag" + topics = {META_TOPIC: "std_msgs/msg/String", STATS_TOPIC: "std_msgs/msg/String", PREVIEW_META_TOPIC: "std_msgs/msg/String", + **{f"{PREVIEW_PREFIX}{c}": "sensor_msgs/msg/CompressedImage" for c in (1, 2, 3, 4)}} + rec = Recorder(bag, topics, profile="default") + rec.open() + with col.lock: + # node A's run only: node B's frame_meta seqs restart at 1 — stop at the first seq that goes back + prev_seq = 0 + for t_ns, o, msg in col.meta: + if o["seq"] <= prev_seq: + break + prev_seq = o["seq"] + rec.write_raw(META_TOPIC, serialize_message(msg), t_ns) + for t_ns, o, msg in col.stats[:1] + [x for x in col.stats if x[1]["frames"] == prev_frames][-1:]: + rec.write_raw(STATS_TOPIC, serialize_message(msg), t_ns) + for t_ns, o, msg in col.pmeta: + rec.write_raw(PREVIEW_META_TOPIC, serialize_message(msg), t_ns) + for ch, lst in list(col.images.items()): + for t_ns, msg in lst: + rec.write_raw(f"{PREVIEW_PREFIX}{ch}", serialize_message(msg), t_ns) + info = rec.close(reindex=False) + rep_json2 = work / "report-bag.json" + r = subprocess.run([PY, str(tool), "--bag", str(bag), "--json", str(rep_json2)], capture_output=True, text=True, timeout=120) + ok = r.returncode == 0 and rep_json2.exists() + check(f"F: zmq_latency_report runs on a bag ({info.get('records')} records)", ok, (r.stdout + r.stderr)[-600:]) + if ok: + rep2 = json.loads(rep_json2.read_text()) + s1 = next((s for s in rep2["segments"] if s["id"] == 1), None) + check("F: bag path decodes the String topics (segment 1 = 87 frames) and measures the node→record hop", + s1 and s1["frames"] == n_a and s1["node_to_bridge_ms"]["n"] == n_a and 0 <= s1["node_to_bridge_ms"]["p50"] < 5000, json.dumps(s1 and {k: s1[k] for k in ("frames", "node_to_bridge_ms")})) + note(f"F: node→record hop p50 {s1 and s1['node_to_bridge_ms']['p50']} ms, p99 {s1 and s1['node_to_bridge_ms']['p99']} ms (the harness subscriber stands in for the bridge)") + + # ── G: bridge registration (static) ──────────────────────────────────────────── + import bridge_node # noqa: E402 (nodes/) + import scanimage_frames_node as sfn # noqa: E402 + + lw = bridge_node.RecordingLogWriter(None, "behavior_v2", None, profile="crashsafe", typed=False, export="converted", connectors=[]) + tmap = lw.topics() + check("G: bridge_node records the stream topics (meta, stats, preview meta, preview ch1–4) in every bag", + all(t in tmap for t in (sfn.META_TOPIC, sfn.STATS_TOPIC, sfn.PREVIEW_META_TOPIC)) and all(f"{sfn.PREVIEW_TOPIC_PREFIX}{c}" in tmap for c in (1, 2, 3, 4)) + and tmap[f"{sfn.PREVIEW_TOPIC_PREFIX}1"] == "sensor_msgs/msg/CompressedImage", json.dumps(sorted(tmap))) + import mcap_convert # noqa: E402 + + check("G: the converter derives nothing from the stream topics (streams are not behavior samples)", + not any(t in mcap_convert.CONNECTOR_TOPIC_SRC for t in bridge_node.STREAM_TOPICS) and "frame_meta" not in Path(mcap_convert.__file__).read_text(), "") + finally: + for pr in reversed(procs): + pr.stop() + for pr in procs: + pr.log.close() + try: + if ex is not None: + ex.shutdown(timeout_sec=1.0) + if col is not None: + col.destroy_node() + rclpy.try_shutdown() + except Exception: + pass + + n_ok = sum(1 for _, ok, _ in results if ok) + print(f"\n{n_ok} / {len(results)} checks passed", file=sys.stderr) + summary = ["| check | result |", "|---|---|"] + [f"| {n} | {'PASS' if ok else 'FAIL — ' + i.replace('|', '/')[:160]} |" for n, ok, i in results] + (work / "summary.md").write_text("\n".join(summary) + "\n", encoding="utf-8") + print("\n".join(summary)) + if not a.keep and n_ok == len(results): + shutil.rmtree(work, ignore_errors=True) + else: + print(f"[test] artifacts kept in {work}", file=sys.stderr) + return 0 if n_ok == len(results) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ros2-poc/tests/zmq_frames_mock.py b/ros2-poc/tests/zmq_frames_mock.py new file mode 100644 index 00000000..9ffa9a8e --- /dev/null +++ b/ros2-poc/tests/zmq_frames_mock.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 +"""zmq_frames_mock.py — a stand-in for ScanImage's ZeroMQ frame publisher, for the off-rig harness. + +The frame FORMAT is reiserlab/scanimage-2026-bergamo `tests/zmq-frames/mock_zmq_publisher.py` +(Hannah Marie Santos): a 2-part message per frame — a 40-byte header of five little-endian doubles +[pixelsPerLine, linesPerFrame, nChannels, frameTimestamp_s since acquisition start, +totalAcquiredFrames] and int16 pixels in C order of [channel, y, x] (x fastest, then y, then +channel) — and the same CLI, so `tests/test_scanimage_frames.py -- --mock ` runs the +harness against the original. Two additions for the harness only: `--sndhwm` (the publisher's +high-water mark in frames; libzmq's default 1000 ≈ 1 GB at 512×512×2 ch is what a paused subscriber +makes ScanImage buffer — Block Z's estimate) and `--seed`. Like the original, this models the +format only, nothing about ScanImage's timing or memory behaviour. + + python zmq_frames_mock.py --frames 120 --rate 29.08 --ppl 512 --lpf 512 --nch 2 + python zmq_frames_mock.py --frames 300 --drop-every 50 # simulate dropped frames + python zmq_frames_mock.py --frames 150 --sndhwm 20 # a publisher that drops when the subscriber stalls +""" +import argparse +import struct +import time + +import numpy as np +import zmq + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--address", default="tcp://127.0.0.1:65179") + ap.add_argument("--frames", type=int, default=120) + ap.add_argument("--rate", type=float, default=29.08) + ap.add_argument("--ppl", type=int, default=128) + ap.add_argument("--lpf", type=int, default=96) + ap.add_argument("--nch", type=int, default=2) + ap.add_argument("--drop-every", type=int, default=0, help="skip publishing every Nth frame") + ap.add_argument("--tiff", help="also write the frames to this TIFF (needs tifffile)") + ap.add_argument("--start-delay", type=float, default=1.0, help="seconds between bind and the first frame (0 = provoke the slow-joiner loss)") + ap.add_argument("--sndhwm", type=int, default=1000, help="publisher high-water mark in messages (libzmq default 1000)") + ap.add_argument("--seed", type=int, default=0) + a = ap.parse_args() + + ctx = zmq.Context() + s = ctx.socket(zmq.PUB) + s.setsockopt(zmq.SNDHWM, a.sndhwm) + s.bind(a.address) + time.sleep(a.start_delay) # let subscribers connect (ZeroMQ "slow joiner") + rng = np.random.default_rng(a.seed) + writer = None + if a.tiff: + import tifffile + + writer = tifffile.TiffWriter(a.tiff) + t0 = time.perf_counter() + for k in range(1, a.frames + 1): + img = rng.integers(-200, 2000, size=(a.nch, a.lpf, a.ppl), dtype=np.int16) # [c, y, x] + if writer: + for c in range(a.nch): + writer.write(img[c], description=f"frameNumbers = {k}\nframeTimestamps_sec = {(k - 1) / a.rate:.6f}") + if not (a.drop_every and k % a.drop_every == 0): + hdr = struct.pack("<5d", a.ppl, a.lpf, a.nch, (k - 1) / a.rate, k) + s.send_multipart([hdr, img.tobytes()]) # C order of [c,y,x] == x fastest, then y, then c + nxt = t0 + k / a.rate + time.sleep(max(0, nxt - time.perf_counter())) + if writer: + writer.close() + time.sleep(0.5) + s.close(linger=1000) + ctx.term() + print(f"published {a.frames} frames") + + +if __name__ == "__main__": + main() diff --git a/ros2-poc/tools/zmq_latency_report.py b/ros2-poc/tools/zmq_latency_report.py new file mode 100644 index 00000000..00ec7ca3 --- /dev/null +++ b/ros2-poc/tools/zmq_latency_report.py @@ -0,0 +1,423 @@ +#!/usr/bin/env python3 +"""zmq_latency_report.py — latency, gaps and queue depth of the ScanImage ZMQ frame stream (Block R). + +Reads what nodes/scanimage_frames_node.py published — from a run's MCAP bag (the bridge node records +/scanimage/frame_meta, /scanimage/frame_stats, /scanimage/preview/meta and the preview images) or from +the node's own `--dump` JSONL — and, per acquisition segment, reports: + + stream delay rx_mono − header frameTimestamp, RELATIVE to the fastest frame of the segment: the + acquisition clock (vDAQ, since acquisition start) and the host clock share no origin, + so without the TIFF this is the jitter/backlog part of the latency, not its constant. + p50 / p99 / max, plus the drift of the two clocks (ms per s, a linear fit). + absolute latency with --tiff: the TIFF's `epoch` (acquisition start, PC wall clock) + its + `frameTimestamps_sec` give each frame's acquisition wall time; rx_epoch_ms − that is + the frame-to-topic latency on the same PC. Also the per-frame difference between the + ZMQ header timestamp and the TIFF's (the "same origin?" question of the procedure). + hops receive → publish inside the node (`pub_mono_ns − rx_mono_ns`) and, from a bag, node → + bridge record (MCAP log_time − rx_epoch_ms), like FicTrac's connector_status. + preview encode+publish per image (preview meta) and, from a bag, frame receive → preview + record (the image's header.stamp is the frame's receive time); achieved rate per channel. + gaps frameNumber discontinuities in the meta stream (what the publisher sent and the node + never received), the slow-joiner loss (first frame − 1), and the node's own + dropped_by_queue / bad_frames from the last frame_stats. + queue depth histogram and p99 of the node's queue at publish time. + +Pass bars (the procedure's): meta p99 < one frame period (--meta-p99-ms, default = the measured period), +preview p99 < --preview-p99-ms (250), 0 gaps after the first frame, 0 dropped_by_queue. Printed as a +markdown table; --json writes everything; --strict exits 1 when a bar fails. + + pixi run zmq-latency -- --bag runlogs/.bag --json latency_focus60.json + pixi run zmq-latency -- --bag runlogs/.bag --tiff "F:\\data\\run_00003.tif" --segment 2 + pixi run zmq-latency -- --jsonl frames.jsonl +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import math +import re +import struct +import sys +from collections import Counter +from pathlib import Path + +META_TOPIC = "/scanimage/frame_meta" +STATS_TOPIC = "/scanimage/frame_stats" +PREVIEW_META_TOPIC = "/scanimage/preview/meta" +PREVIEW_PREFIX = "/scanimage/preview/ch" +FRAME_ID_RE = re.compile(r"^scanimage/ch(\d+)/s(\d+)/f(\d+)$") + + +# ── small statistics ────────────────────────────────────────────────────────────────────── +def pct(sorted_vals: list[float], q: float) -> float: + """Nearest-rank percentile of an ascending list.""" + if not sorted_vals: + return math.nan + k = max(0, min(len(sorted_vals) - 1, int(math.ceil(q / 100.0 * len(sorted_vals))) - 1)) + return sorted_vals[k] + + +def summarize(vals: list[float]) -> dict: + if not vals: + return {"n": 0} + s = sorted(vals) + return {"n": len(s), "min": round(s[0], 3), "p50": round(pct(s, 50), 3), "p99": round(pct(s, 99), 3), + "max": round(s[-1], 3), "mean": round(sum(s) / len(s), 3)} + + +def slope(xs: list[float], ys: list[float]) -> float | None: + """Least-squares slope of ys on xs (None below 10 points or without spread).""" + n = len(xs) + if n < 10: + return None + mx, my = sum(xs) / n, sum(ys) / n + sxx = sum((x - mx) ** 2 for x in xs) + if sxx <= 0: + return None + return sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / sxx + + +# ── sources: the node's JSONL dump, or a bag ───────────────────────────────────────────── +def load_jsonl(path: Path) -> list[dict]: + recs = [] + with open(path, encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except json.JSONDecodeError: + continue + topic = obj.pop("topic", None) + if topic: + recs.append({"topic": topic, "obj": obj, "log_ns": None}) + return recs + + +def cdr_string(data: bytes) -> str: + """std_msgs/String in CDR (XCDR1): 4-byte encapsulation, uint32 length, bytes (NUL-terminated).""" + le = data[1] == 1 + n = struct.unpack_from("I", data, 4)[0] + return data[8:8 + n].rstrip(b"\x00").decode("utf-8", errors="replace") + + +def cdr_compressed_image(data: bytes) -> dict: + """sensor_msgs/CompressedImage header + format + payload length (the PNG itself is not needed).""" + le = data[1] == 1 + e = "<" if le else ">" + p = 4 + + def align(n: int) -> None: + nonlocal p + p += (-(p - 4)) % n + + sec = struct.unpack_from(e + "i", data, p)[0] + p += 4 + nsec = struct.unpack_from(e + "I", data, p)[0] + p += 4 + n = struct.unpack_from(e + "I", data, p)[0] + p += 4 + frame_id = data[p:p + n].rstrip(b"\x00").decode("utf-8", errors="replace") + p += n + align(4) + n = struct.unpack_from(e + "I", data, p)[0] + p += 4 + fmt = data[p:p + n].rstrip(b"\x00").decode("utf-8", errors="replace") + p += n + align(4) + n = struct.unpack_from(e + "I", data, p)[0] + return {"stamp_ns": sec * 1_000_000_000 + nsec, "frame_id": frame_id, "format": fmt, "bytes": n} + + +def load_bag(path: Path) -> list[dict]: + from mcap.reader import make_reader + + files = sorted(f for f in path.glob("*.mcap") if not f.name.endswith((".unindexed.mcap", ".indexed.tmp"))) if path.is_dir() else [path] + if not files: + raise SystemExit(f"no .mcap files in {path}") + recs = [] + for f in files: + with open(f, "rb") as fh: + reader = make_reader(fh, validate_crcs=False) + for schema, channel, message in reader.iter_messages(log_time_order=False): + t = channel.topic + if t in (META_TOPIC, STATS_TOPIC, PREVIEW_META_TOPIC): + try: + obj = json.loads(cdr_string(message.data)) + except (json.JSONDecodeError, struct.error, UnicodeDecodeError): + continue + recs.append({"topic": t, "obj": obj, "log_ns": message.log_time}) + elif t.startswith(PREVIEW_PREFIX): + try: + img = cdr_compressed_image(message.data) + except struct.error: + continue + recs.append({"topic": t, "obj": img, "log_ns": message.log_time}) + return recs + + +# ── the TIFF side (stdlib; ScanImage writes one page per channel per frame) ────────────── +def tiff_descriptions(path: Path): + """Yield (page index, ImageDescription text) for every IFD; classic and BigTIFF, both byte orders.""" + with open(path, "rb") as f: + data = f.read() + bo = "<" if data[:2] == b"II" else ">" + big = struct.unpack_from(bo + "H", data, 2)[0] == 43 + if big: + off = struct.unpack_from(bo + "Q", data, 8)[0] + cnt_fmt, cnt_size, ent, ent_fmt, off_fmt = "Q", 8, 20, "HHQQ", "Q" + else: + off = struct.unpack_from(bo + "I", data, 4)[0] + cnt_fmt, cnt_size, ent, ent_fmt, off_fmt = "H", 2, 12, "HHII", "I" + page = 0 + seen = set() + while off and off not in seen and off + cnt_size <= len(data): + seen.add(off) + n = struct.unpack_from(bo + cnt_fmt, data, off)[0] + pos = off + cnt_size + desc = "" + for i in range(n): + tag, typ, count, val = struct.unpack_from(bo + ent_fmt, data, pos + i * ent) + if tag == 270: + inline = 8 if big else 4 + if count <= inline: # value stored in the entry itself + raw = struct.pack(bo + off_fmt, val)[:count] + else: + raw = data[val:val + count] + desc = raw.decode("latin1", "replace").rstrip("\x00") + yield page, desc + page += 1 + off = struct.unpack_from(bo + off_fmt, data, pos + n * ent)[0] + + +def parse_fields(desc: str) -> dict: + out: dict = {} + for key in ("frameNumbers", "frameTimestamps_sec", "acqTriggerTimestamps_sec", "endOfAcquisition"): + m = re.search(re.escape(key) + r"\s*=\s*([-\d.eE+]+)", desc) + out[key] = float(m.group(1)) if m else None + m = re.search(r"epoch\s*=\s*\[([^\]]*)\]", desc) + out["epoch"] = [float(x) for x in re.split(r"[\s,;]+", m.group(1).strip()) if x] if m else None + return out + + +def tiff_frames(path: Path) -> dict: + """{frameNumber: {"ts": frameTimestamps_sec, "pages": n}} + the acquisition epoch of the first page.""" + frames: dict[int, dict] = {} + epoch = None + pages = 0 + for _, desc in tiff_descriptions(path): + pages += 1 + f = parse_fields(desc) + if f["frameNumbers"] is None: + continue + k = int(f["frameNumbers"]) + if epoch is None and f["epoch"]: + epoch = f["epoch"] + if k in frames: + frames[k]["pages"] += 1 + else: + frames[k] = {"ts": f["frameTimestamps_sec"], "pages": 1} + return {"frames": frames, "epoch": epoch, "pages": pages, "path": str(path)} + + +def epoch_wall_ms(epoch: list, utc_offset_h: float | None) -> int: + """ScanImage's `epoch = [y m d H M S.fff]` (PC local time) → wall ms. The offset defaults to this + machine's local zone (the tool normally runs on the ScanImage PC).""" + y, mo, d, h, mi, s = epoch[:6] + sec = int(s) + frac = s - sec + naive = dt.datetime(int(y), int(mo), int(d), int(h), int(mi), sec) + if utc_offset_h is None: + t = naive.astimezone() # interpret as local time + else: + t = naive.replace(tzinfo=dt.timezone(dt.timedelta(hours=utc_offset_h))) + return int(t.timestamp() * 1000) + int(round(frac * 1000)) + + +# ── analysis ────────────────────────────────────────────────────────────────────────────── +def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | None = None, + frame_period_ms: float | None = None, segment_filter: int | None = None, + meta_p99_ms: float | None = None, preview_p99_ms: float = 250.0) -> dict: + metas = [r for r in records if r["topic"] == META_TOPIC and isinstance(r["obj"].get("seq"), int)] + stats = [r["obj"] for r in records if r["topic"] == STATS_TOPIC] + pmetas = [r for r in records if r["topic"] == PREVIEW_META_TOPIC] + pimgs = [r for r in records if r["topic"].startswith(PREVIEW_PREFIX)] + by_seg: dict[int, list[dict]] = {} + for r in metas: + by_seg.setdefault(int(r["obj"]["segment"]), []).append(r) + last_stats = stats[-1] if stats else {} + stats_segments = {s["id"]: s for s in last_stats.get("segment_list", [])} if last_stats else {} + + # which segment the TIFF belongs to: the one sharing the most frame numbers (or --segment) + tiff_seg = None + if tiff and tiff["frames"]: + cands = [segment_filter] if segment_filter else list(by_seg) + best = 0 + for sid in cands: + n = sum(1 for r in by_seg.get(sid, []) if int(r["obj"]["frameNumber"]) in tiff["frames"]) + if n > best: + best, tiff_seg = n, sid + + segments = [] + for sid in sorted(by_seg): + if segment_filter and sid != segment_filter: + continue + rows = sorted(by_seg[sid], key=lambda r: r["obj"]["seq"]) + objs = [r["obj"] for r in rows] + nums = [int(o["frameNumber"]) for o in objs] + ts = [float(o["frameTimestamp_s"]) for o in objs] + rx_mono_s = [o["rx_mono_ns"] / 1e9 for o in objs] + gaps = [{"after": a, "next": b, "missing": b - a - 1} for a, b in zip(nums, nums[1:]) if b > a + 1] + d = [rm - t for rm, t in zip(rx_mono_s, ts)] + off = min(d) if d else 0.0 + rel_ms = [(x - off) * 1000.0 for x in d] + dts = sorted(b - a for a, b in zip(ts, ts[1:]) if b > a) + period_ms = frame_period_ms or (pct(dts, 50) * 1000.0 if dts else None) + hop_ms = [(o["pub_mono_ns"] - o["rx_mono_ns"]) / 1e6 for o in objs if "pub_mono_ns" in o] + bridge_ms = [(r["log_ns"] - o["rx_epoch_ms"] * 1_000_000) / 1e6 for r, o in zip(rows, objs) if r["log_ns"] is not None] + depths = [int(o.get("queue_depth", 0)) for o in objs] + hist = dict(sorted(Counter(depths).items())) + pm = [r["obj"] for r in pmetas if int(r["obj"].get("segment", -1)) == sid] + enc_ms = [(o["pub_mono_ns"] - o["rx_mono_ns"]) / 1e6 for o in pm if "pub_mono_ns" in o] + per_ch: dict = {} + for o in pm: + c = per_ch.setdefault(str(o.get("channel")), {"images": 0, "first_rx_ms": o["rx_epoch_ms"], "last_rx_ms": o["rx_epoch_ms"], "mapping": o.get("mapping")}) + c["images"] += 1 + c["last_rx_ms"] = o["rx_epoch_ms"] + for c in per_ch.values(): + span = (c["last_rx_ms"] - c["first_rx_ms"]) / 1000.0 + c["achieved_hz"] = round((c["images"] - 1) / span, 2) if c["images"] > 1 and span > 0 else None + e2e_ms = [] + for r in pimgs: + m = FRAME_ID_RE.match(r["obj"].get("frame_id", "")) + if m and int(m.group(2)) == sid and r["log_ns"] is not None: + e2e_ms.append((r["log_ns"] - r["obj"]["stamp_ns"]) / 1e6) + seg = { + "id": sid, "frames": len(objs), "first_frame": nums[0], "last_frame": nums[-1], + "missed_before_first": max(nums[0] - 1, 0), "expected_from_first": nums[-1] - nums[0] + 1, + "gaps": len(gaps), "gap_frames": sum(g["missing"] for g in gaps), "gap_list": gaps[:200], + "dims": objs[0].get("dims"), "nchan": objs[0].get("nchan"), + "frame_period_ms": round(period_ms, 3) if period_ms else None, + "rate_hz": round(1000.0 / period_ms, 3) if period_ms else None, + "span_s": round(rx_mono_s[-1] - rx_mono_s[0], 3), + "stream_delay_rel_ms": summarize(rel_ms), + "clock_drift_ms_per_s": (lambda m: round(m * 1000.0, 4) if m is not None else None)(slope(ts, d)), + "node_hop_ms": summarize(hop_ms), + "node_to_bridge_ms": summarize(bridge_ms), + "queue_depth": {"hist": hist, "p99": pct(sorted(depths), 99) if depths else None, "max": max(depths) if depths else None}, + "preview": {"images": len(pm), "encode_publish_ms": summarize(enc_ms), "per_channel": per_ch, + "frame_to_record_ms": summarize(e2e_ms)}, + "from_stats": stats_segments.get(sid), + } + if tiff and sid == tiff_seg: + pairs = [(o, tiff["frames"][int(o["frameNumber"])]) for o in objs if int(o["frameNumber"]) in tiff["frames"]] + delta_ms = [(float(o["frameTimestamp_s"]) - t["ts"]) * 1000.0 for o, t in pairs if t["ts"] is not None] + t_align = {"tiff": tiff["path"], "tiff_frames": len(tiff["frames"]), "tiff_pages": tiff["pages"], + "matched_frames": len(pairs), "zmq_minus_tiff_ts_ms": summarize(delta_ms), + "pages_per_frame": Counter(t["pages"] for t in tiff["frames"].values()).most_common(1)[0][0] if tiff["frames"] else None, + "frame_meta_count_equals_tiff_frames": len(objs) == len(tiff["frames"])} + if tiff["epoch"]: + e_ms = epoch_wall_ms(tiff["epoch"], utc_offset_h) + abs_ms = [o["rx_epoch_ms"] - (e_ms + t["ts"] * 1000.0) for o, t in pairs if t["ts"] is not None] + t_align.update({"epoch": tiff["epoch"], "epoch_wall_ms": e_ms, "utc_offset_h": utc_offset_h, + "absolute_latency_ms": summarize(abs_ms)}) + seg["tiff"] = t_align + segments.append(seg) + + # pass bars + bars = [] + for seg in segments: + period = meta_p99_ms or seg["frame_period_ms"] or 34.0 + lat = seg.get("tiff", {}).get("absolute_latency_ms") or seg["stream_delay_rel_ms"] + kind = "absolute (TIFF clock)" if seg.get("tiff", {}).get("absolute_latency_ms") else "relative (fastest frame = 0)" + if lat.get("n"): + bars.append({"segment": seg["id"], "bar": f"meta latency p99 < {period:.1f} ms ({kind})", "value": lat["p99"], "pass": lat["p99"] < period}) + pv = seg["preview"]["frame_to_record_ms"] if seg["preview"]["frame_to_record_ms"].get("n") else seg["preview"]["encode_publish_ms"] + if pv.get("n"): + bars.append({"segment": seg["id"], "bar": f"preview p99 < {preview_p99_ms:g} ms", "value": pv["p99"], "pass": pv["p99"] < preview_p99_ms}) + bars.append({"segment": seg["id"], "bar": "0 gaps after the first frame", "value": seg["gap_frames"], "pass": seg["gap_frames"] == 0}) + if last_stats: + bars.append({"segment": None, "bar": "0 dropped_by_queue (node back-pressure)", "value": last_stats.get("dropped_by_queue"), "pass": not last_stats.get("dropped_by_queue")}) + bars.append({"segment": None, "bar": "0 bad frames", "value": last_stats.get("bad_frames"), "pass": not last_stats.get("bad_frames")}) + return { + "frame_meta": len(metas), "frame_stats": len(stats), "preview_meta": len(pmetas), "preview_images": len(pimgs), + "last_stats": {k: last_stats.get(k) for k in ("frames", "published", "dropped_by_queue", "bad_frames", "gaps", "gap_frames", + "segments", "connects", "disconnects", "queue_limit", "uptime_s")} if last_stats else None, + "segments": segments, "bars": bars, "all_pass": all(b["pass"] for b in bars) if bars else None, + } + + +# ── report ──────────────────────────────────────────────────────────────────────────────── +def fmt(s: dict) -> str: + return "—" if not s or not s.get("n") else f"{s['p50']:.2f} / {s['p99']:.2f} / {s['max']:.2f} (n={s['n']})" + + +def markdown(rep: dict, source: str) -> str: + out = [f"# ZMQ frame-stream report — {source}", "", + f"frame_meta {rep['frame_meta']}, frame_stats {rep['frame_stats']}, preview meta {rep['preview_meta']}, preview images {rep['preview_images']}"] + if rep["last_stats"]: + out.append("node totals (last frame_stats): " + ", ".join(f"{k}={v}" for k, v in rep["last_stats"].items())) + out += ["", "| segment | frames (first–last) | missed before first | gaps (frames) | rate Hz | stream delay rel p50/p99/max ms | node hop p50/p99/max ms | node→bridge p50/p99/max ms | queue p99/max | preview enc p50/p99/max ms | preview frame→record p50/p99/max ms |", + "|---|---|---|---|---|---|---|---|---|---|---|"] + for s in rep["segments"]: + q = s["queue_depth"] + out.append(f"| {s['id']} | {s['frames']} ({s['first_frame']}–{s['last_frame']}) | {s['missed_before_first']} | {s['gaps']} ({s['gap_frames']}) | {s['rate_hz']} | " + f"{fmt(s['stream_delay_rel_ms'])} | {fmt(s['node_hop_ms'])} | {fmt(s['node_to_bridge_ms'])} | {q['p99']}/{q['max']} | " + f"{fmt(s['preview']['encode_publish_ms'])} | {fmt(s['preview']['frame_to_record_ms'])} |") + for s in rep["segments"]: + if s.get("clock_drift_ms_per_s") is not None: + out.append(f"- segment {s['id']}: acquisition clock vs host monotonic drift {s['clock_drift_ms_per_s']:+.4f} ms/s; queue-depth histogram {s['queue_depth']['hist']}") + if s["gap_list"]: + out.append(f"- segment {s['id']} gaps: " + ", ".join(f"{g['after']}→{g['next']} (−{g['missing']})" for g in s["gap_list"][:20]) + (" …" if len(s["gap_list"]) > 20 else "")) + for ch, c in s["preview"]["per_channel"].items(): + out.append(f"- segment {s['id']} preview ch{ch}: {c['images']} images, {c['achieved_hz']} Hz achieved, mapping {c['mapping']}") + t = s.get("tiff") + if t: + out.append(f"- segment {s['id']} vs TIFF {t['tiff']}: {t['matched_frames']} of {t['tiff_frames']} TIFF frames matched ({t['tiff_pages']} pages, {t['pages_per_frame']} per frame); " + f"frame_meta count == TIFF frames: {t['frame_meta_count_equals_tiff_frames']}; ZMQ − TIFF timestamp p50/p99/max ms {fmt(t['zmq_minus_tiff_ts_ms'])}" + + (f"; ABSOLUTE latency (acquisition → topic) p50/p99/max ms {fmt(t['absolute_latency_ms'])}" if t.get("absolute_latency_ms") else "; no `epoch` in the TIFF → no absolute latency")) + out += ["", "| pass bar | segment | value | result |", "|---|---|---|---|"] + for b in rep["bars"]: + out.append(f"| {b['bar']} | {b['segment'] if b['segment'] is not None else 'all'} | {b['value']} | {'PASS' if b['pass'] else 'FAIL'} |") + return "\n".join(out) + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + src = ap.add_mutually_exclusive_group(required=True) + src.add_argument("--bag", help="a run's bag directory (or one .mcap) recorded by the bridge node") + src.add_argument("--jsonl", help="the node's --dump file") + ap.add_argument("--tiff", help="the acquisition's ScanImage TIFF (frameNumbers / frameTimestamps_sec / epoch in its page descriptions)") + ap.add_argument("--utc-offset-h", type=float, default=None, help="the ScanImage PC's UTC offset for the TIFF epoch (default: this machine's local zone)") + ap.add_argument("--segment", type=int, default=None, help="only this acquisition segment") + ap.add_argument("--frame-period-ms", type=float, default=None, help="override the measured frame period for the pass bar") + ap.add_argument("--meta-p99-ms", type=float, default=None, help="pass bar for meta latency p99 (default: one frame period)") + ap.add_argument("--preview-p99-ms", type=float, default=250.0) + ap.add_argument("--json", help="write the full report here") + ap.add_argument("--strict", action="store_true", help="exit 1 when a pass bar fails") + a = ap.parse_args(argv) + + if a.bag: + records = load_bag(Path(a.bag)) + source = a.bag + else: + records = load_jsonl(Path(a.jsonl)) + source = a.jsonl + tiff = tiff_frames(Path(a.tiff)) if a.tiff else None + rep = analyze(records, tiff, a.utc_offset_h, a.frame_period_ms, a.segment, a.meta_p99_ms, a.preview_p99_ms) + rep["source"] = source + if not rep["segments"]: + print(f"no /scanimage/frame_meta records in {source}", file=sys.stderr) + return 2 + print(markdown(rep, source)) + if a.json: + Path(a.json).write_text(json.dumps(rep, indent=1), encoding="utf-8") + print(f"\nwritten {a.json}") + return 1 if (a.strict and rep["all_pass"] is False) else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 485953c0ef03716756c62c6cabd36279be7696b8 Mon Sep 17 00:00:00 2001 From: hanmarsan Date: Wed, 21 Oct 2026 11:46:38 -0400 Subject: [PATCH 2/5] ros2-poc: treat CTRL_BREAK as Ctrl-C in the shared spin() (Windows clean exit) rig.py and the harnesses stop children with CTRL_BREAK_EVENT; its default action ended the frames node (and every spin() node) at once, rc 0xC000013A (Bergamo 2026-10-20, test-frames E 33/34). SIGBREAK now raises KeyboardInterrupt like Ctrl-C, as bridge_node.py already does. test-frames E 'node exits cleanly on SIGINT' passes on the rig PC. Co-Authored-By: Claude Opus 5.5 --- ros2-poc/nodes/_common.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/ros2-poc/nodes/_common.py b/ros2-poc/nodes/_common.py index 5e5fc2ef..d9b5c0b2 100644 --- a/ros2-poc/nodes/_common.py +++ b/ros2-poc/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): From 743bf7730ad30dadd582d1b4aeb929b16a158f8d Mon Sep 17 00:00:00 2001 From: hanmarsan Date: Wed, 21 Oct 2026 11:46:38 -0400 Subject: [PATCH 3/5] ros2-poc: ScanImage node reads the frame number from getAcqStatus (--frame-counter) hSI.hDisplay.lastFrameNumber throws 'Index exceeds array bounds' on ScanImage 2026.0.0 while idle and even during FOCUS, so every 1 Hz status poll carried an error; on a simulated vDAQ (2026-10-21) the failing get took ~59 ms against ~2 ms for AcquisitionMex(hSI.hScan2D.hAcq, 'getAcqStatus'). That error path may be the 1 Hz MATLAB stall of 2026-10-20 (Q3). New --frame-counter acqstatus (default) | display (the old get, kept to compare on the rig) | none; status and waitIdle gain acq{frameNumberAcq, totalAcquiredFrames, acqNumber, numDroppedFrames*, fifoOverflows, dcOvervoltage}. The mock answers the new fixed eval; acceptance 55/55 (one new check). Co-Authored-By: Claude Opus 5.5 --- ros2-poc/nodes/scanimage_node.py | 94 +++++++++++++++++++++++++------- ros2-poc/tests/si_json_mock.py | 13 ++++- ros2-poc/tests/test_ros2_poc.py | 4 ++ 3 files changed, 90 insertions(+), 21 deletions(-) diff --git a/ros2-poc/nodes/scanimage_node.py b/ros2-poc/nodes/scanimage_node.py index b5efc369..097ad7e4 100644 --- a/ros2-poc/nodes/scanimage_node.py +++ b/ros2-poc/nodes/scanimage_node.py @@ -11,7 +11,9 @@ "error", "steps": [...per JSON request...], "rtt_ms", "t_ms", "rx_epoch_ms", "artifact": {"kind":"tiff", "path", "stem", "counter"} (grab)} /scanimage/status OUT every --status-every s: {"reachable", "acqState", "lastFrameNumber", - "logFileCounter", "logFileStem", "logFilePath", "loggingEnable", "rtt_ms", "t_ms"} + "logFileCounter", "logFileStem", "logFilePath", "loggingEnable", "rtt_ms", "t_ms", + "acq": {frameNumberAcq, totalAcquiredFrames, acqNumber, numDroppedFramesDisplay, + numDroppedFramesLogger, fifoOverflows, dcOvervoltage} (--frame-counter acqstatus)} The `command` vocabulary is the protocol's `scanimage` plugin (js/plugin-registry.js, class "ScanImagePlugin" — MATLAB has no such class yet): @@ -29,8 +31,8 @@ class "ScanImagePlugin" — MATLAB has no such class yet): OBJECT-valued property gets no reply and the client hangs; B4 — a partial message blocks MATLAB for ~10 s, so every request is one length-prefixed write; the server is single-client, so this node is the ONLY client and status polling is coalesced; `eval` exists on the server -but this node sends exactly two fixed expressions (startGrab, abort) and never a string from -a protocol. The JSON client is ported from tests/remote-control/si_remote.py (JsonClient) in +but this node sends only fixed expressions (startGrab, abort, and the getAcqStatus frame +counter) and never a string from a protocol. The JSON client is ported from tests/remote-control/si_remote.py (JsonClient) in reiserlab/scanimage-2026-bergamo (Hannah Marie Santos), with attribution. LAB-303 semantics (as bias_node.py): `timeout_s` per request (`slow_timeout_s` for grab/abort/ @@ -97,12 +99,19 @@ class "ScanImagePlugin" — MATLAB has no such class yet): "hSI.hStackManager.framesPerSlice", "hSI.hChannels.loggingEnable", }) -# The only two expressions ever sent as `eval` (plus startFocus for the CLI's bench check). +# The only expressions ever sent as `eval` (plus startFocus for the CLI's bench check). EVAL_GRAB = "hSI.startGrab()" EVAL_ABORT = "hSI.abort()" EVAL_FOCUS = "hSI.startFocus()" +# The frame counter (--frame-counter acqstatus): the vDAQ acquisition engine's own status struct. +# hDisplay.lastFrameNumber throws "Index exceeds array bounds" on 2026.0.0 while idle and even during +# FOCUS, and the error took ~59 ms against ~2 ms for this (simulated vDAQ, 2026-10-21). +EVAL_ACQ_STATUS = "AcquisitionMex(hSI.hScan2D.hAcq,'getAcqStatus')" +ACQ_STATUS_FIELDS = ("frameNumberAcq", "totalAcquiredFrames", "acqNumber", "numDroppedFramesDisplay", + "numDroppedFramesLogger", "fifoOverflows", "dcOvervoltage") +FRAME_COUNTERS = ("acqstatus", "display", "none") SLOW_COMMANDS = frozenset({"grab", "abort", "focus"}) -STATUS_PROPS = ("hSI.acqState", "hSI.hDisplay.lastFrameNumber", "hSI.hScan2D.logFileCounter", +STATUS_PROPS = ("hSI.acqState", "hSI.hScan2D.logFileCounter", "hSI.hScan2D.logFileStem", "hSI.hScan2D.logFilePath", "hSI.hChannels.loggingEnable") @@ -226,11 +235,18 @@ def eval(self, expr: str, timeout_s: float | None = None) -> float: _, rtt = self.request({"command": "eval", "function": expr, "num_outputs": 0}, timeout_s) return rtt + def eval_value(self, expr: str, timeout_s: float | None = None) -> tuple[object, float]: + """eval with one output → (output1, rtt_ms).""" + rep, rtt = self.request({"command": "eval", "function": expr, "num_outputs": 1}, timeout_s) + return (rep.get("outputs") or {}).get("output1"), rtt + class ScanImageNode(Node): - def __init__(self, cli: JsonClient, status_every_s: float, poll_s: float, default_wait_s: float) -> None: + def __init__(self, cli: JsonClient, status_every_s: float, poll_s: float, default_wait_s: float, + frame_counter: str = "acqstatus") -> None: super().__init__("scanimage") self.cli = cli + self.frame_counter = frame_counter self.poll_s = poll_s self.default_wait_s = default_wait_s self.reply_pub = self.create_publisher(String, REPLY_TOPIC, QOS) @@ -243,7 +259,7 @@ def __init__(self, cli: JsonClient, status_every_s: float, poll_s: float, defaul if status_every_s > 0: self.create_timer(status_every_s, self._enqueue_status) self.commands = 0 - self.get_logger().info(f"ScanImage JSON server at {cli.host}:{cli.port} ← {CMD_TOPIC} → {REPLY_TOPIC}, {STATUS_TOPIC} every {status_every_s:g}s") + self.get_logger().info(f"ScanImage JSON server at {cli.host}:{cli.port} ← {CMD_TOPIC} → {REPLY_TOPIC}, {STATUS_TOPIC} every {status_every_s:g}s, frame counter {frame_counter}") # ── plumbing ──────────────────────────────────────────────────────────────────────── def _publish(self, pub, obj: dict) -> None: @@ -305,11 +321,46 @@ def _snapshot(self, steps: list | None = None) -> dict: out[prop.split(".")[-1]] = v if steps is not None: steps.append({"command": "get", "property": prop, "ok": True, "value": v, "rtt_ms": rtt}) + if self.frame_counter != "none": + last, acq, rtt = self._frame_number(errors, steps) + out["lastFrameNumber"] = last + if acq is not None: + out["acq"] = acq + rtt_total += rtt if errors: out["errors"] = errors out["rtt_ms"] = round(rtt_total, 2) return out + def _frame_number(self, errors: dict, steps: list | None = None) -> tuple[object, dict | None, float]: + """→ (lastFrameNumber, acq status subset or None, rtt_ms). Informational: a ScanImage error + lands in `errors` and gives None; timeouts and socket errors still raise. + acqstatus: frameNumberAcq from EVAL_ACQ_STATUS. display: hDisplay.lastFrameNumber (broken + on 2026.0.0; kept to compare the poll's cost on the rig, A2 in the ZMQ follow-up).""" + if self.frame_counter == "display": + prop = "hSI.hDisplay.lastFrameNumber" + try: + v, rtt = self.cli.get(prop) + except SiError as exc: + errors[prop] = str(exc).splitlines()[0] + if steps is not None: + steps.append({"command": "get", "property": prop, "ok": False, "error": errors[prop]}) + return None, None, 0.0 + if steps is not None: + steps.append({"command": "get", "property": prop, "ok": True, "value": v, "rtt_ms": rtt}) + return v, None, rtt + try: + v, rtt = self.cli.eval_value(EVAL_ACQ_STATUS) + except SiError as exc: + errors["getAcqStatus"] = str(exc).splitlines()[0] + if steps is not None: + steps.append({"command": "eval", "function": EVAL_ACQ_STATUS, "ok": False, "error": errors["getAcqStatus"]}) + return None, None, 0.0 + acq = {k: v.get(k) for k in ACQ_STATUS_FIELDS} if isinstance(v, dict) else None + if steps is not None: + steps.append({"command": "eval", "function": EVAL_ACQ_STATUS, "ok": True, "value": acq, "rtt_ms": rtt}) + return (acq or {}).get("frameNumberAcq"), acq, rtt + def _do_status(self) -> dict | None: t_ms = wall_ms() try: @@ -444,16 +495,16 @@ def _run(self, command: str, params: dict, steps: list): waited = round(time.monotonic() - t0, 3) if st != "idle": raise SiError(f"still '{st}' after {waited:g} s (timeout_s {to:g}); use abort") - # lastFrameNumber is informational: on the Bergamo (2026.0.0) it throws "Index exceeds - # array bounds" even right after a grab, so it must not turn a reached idle into an error. - try: - last = self._get("hSI.hDisplay.lastFrameNumber", steps) - except SiError as exc: - last = None - steps.append({"command": "get", "property": "hSI.hDisplay.lastFrameNumber", "ok": False, - "error": str(exc).splitlines()[0]}) - return {"acqState": st, "waited_s": waited, "lastFrameNumber": last, - "logFileCounter": self._get("hSI.hScan2D.logFileCounter", steps)} + # The frame number is informational: a failure must not turn a reached idle into an + # error (hDisplay.lastFrameNumber throws even right after a grab on 2026.0.0). + out = {"acqState": st, "waited_s": waited} + if self.frame_counter != "none": + errors: dict = {} + out["lastFrameNumber"], acq, _ = self._frame_number(errors, steps) + if acq is not None: + out["acq"] = acq + out["logFileCounter"] = self._get("hSI.hScan2D.logFileCounter", steps) + return out if command == "abort": rtt = self.cli.eval(EVAL_ABORT, self.cli.slow_timeout_s) steps.append({"command": "eval", "function": EVAL_ABORT, "ok": True, "rtt_ms": rtt}) @@ -470,13 +521,18 @@ def main(argv: list[str] | None = None) -> int: p.add_argument("--port", type=int, default=5555, help="ScanImage JSON server port (default: 5555)") p.add_argument("--timeout-s", type=float, default=2.0, help="per-request timeout; a timeout is an UNKNOWN outcome and drops the connection (default: 2)") p.add_argument("--slow-timeout-s", type=float, default=15.0, help="timeout for grab/abort (ScanImage's startGrab is synchronous, 1.2–2.4 s on the Bergamo PC; default: 15)") - p.add_argument("--status-every", type=float, default=1.0, help="status poll period in s; 0 = off (default: 1 — six ~2 ms gets)") + p.add_argument("--status-every", type=float, default=1.0, help="status poll period in s; 0 = off (default: 1 — five ~2 ms gets + the frame counter)") + p.add_argument("--frame-counter", choices=FRAME_COUNTERS, default="acqstatus", + help="how status and waitIdle read the frame number: acqstatus = one eval of getAcqStatus (~2 ms, also " + "dropped frames, FIFO overflow, DC overvoltage); display = hDisplay.lastFrameNumber (throws on 2026.0.0; " + "kept to compare on the rig); none (default: acqstatus)") p.add_argument("--poll-s", type=float, default=0.25, help="acqState poll period inside grab/waitIdle/abort (default: 0.25)") p.add_argument("--wait-s", type=float, default=30.0, help="waitIdle default timeout_s when the command carries none (default: 30)") args, ros_args = p.parse_known_args(argv) raise_timer_resolution() rclpy.init(args=ros_args) - node = ScanImageNode(JsonClient(args.host, args.port, args.timeout_s, args.slow_timeout_s), args.status_every, args.poll_s, args.wait_s) + node = ScanImageNode(JsonClient(args.host, args.port, args.timeout_s, args.slow_timeout_s), args.status_every, args.poll_s, args.wait_s, + args.frame_counter) spin(node) print(f"[scanimage-node] {node.commands} commands handled, {node.cli.reconnects} reconnects", file=sys.stderr) return 0 diff --git a/ros2-poc/tests/si_json_mock.py b/ros2-poc/tests/si_json_mock.py index a98606fb..7a3981cf 100644 --- a/ros2-poc/tests/si_json_mock.py +++ b/ros2-poc/tests/si_json_mock.py @@ -10,7 +10,8 @@ * {"command":"get","property":P} → the same object + "value" {"command":"set","property":P,"value":V} → + "actual" {"command":"eval","function":F,"num_outputs":N} → + "outputs" (only hSI.startGrab(), - hSI.startFocus(), hSI.abort() are understood; anything else is an error reply) + hSI.startFocus(), hSI.abort() and AcquisitionMex(hSI.hScan2D.hAcq,'getAcqStatus') are + understood; anything else is an error reply) * Stock bug B2 (default ON, `--no-b2` to patch it): a get of an OBJECT-valued property gets NO reply at all — the client must time out. The connector's property allowlist is what keeps a rig away from this; the harness uses it to exercise the UNKNOWN path. @@ -131,7 +132,15 @@ def _eval(expr: str, args: argparse.Namespace): _acq["t_end"] = None _acq["aborts"] += 1 return [] - raise ValueError(f"mock cannot eval {expr!r} (only hSI.startGrab(), hSI.startFocus(), hSI.abort())") + if expr == "AcquisitionMex(hSI.hScan2D.hAcq,'getAcqStatus')": + _tick() # the vDAQ engine's status struct (fields as on 2026.0.0, simulated vDAQ 2026-10-21) + with _lock: + n = STATE["hDisplay"]["lastFrameNumber"] + return [{"fifoOverflows": False, "dcOvervoltage": False, "acquisitionFinished": STATE["acqState"] == "idle", + "totalAcquiredFrames": n, "acqNumber": _acq["grabs"], "frameNumberAcq": n, + "acqTriggerTimestamp": -1, "nextFileMarkerTimestamp": -1, "numDroppedFramesDisplay": 0, + "numDroppedFramesLogger": 0, "ZMQIsThreadRunning": STATE["acqState"] != "idle", "ZMQErrorMessage": ""}] + raise ValueError(f"mock cannot eval {expr!r} (only hSI.startGrab(), hSI.startFocus(), hSI.abort(), getAcqStatus)") def _recv_exact(conn: socket.socket, n: int) -> bytes: diff --git a/ros2-poc/tests/test_ros2_poc.py b/ros2-poc/tests/test_ros2_poc.py index 263105bf..98c9f7c8 100644 --- a/ros2-poc/tests/test_ros2_poc.py +++ b/ros2-poc/tests/test_ros2_poc.py @@ -479,6 +479,10 @@ def side_actor() -> None: si_cmd = [e.get("command") for e in si_ev if e.get("kind") == "command"] check("ScanImage command/reply pairs in the run log (via=ws, plugin scanimage)", len(si_rep) >= 8 and all(e.get("via") == "ws" and e.get("plugin") == "scanimage" for e in si_rep), f"{len(si_rep)} replies: {[r.get('command') for r in si_rep]}") check("ScanImage status lines in the run log carry acqState", len(si_st) >= 2 and all("acqState" in e for e in si_st if e.get("reachable")), str(len(si_st))) + si_ok = [e for e in si_st if e.get("reachable")] + check("ScanImage status lines carry the getAcqStatus frame counter (acq.frameNumberAcq, no errors)", + bool(si_ok) and all(isinstance(e.get("acq"), dict) and "frameNumberAcq" in e["acq"] and not e.get("errors") for e in si_ok), + json.dumps(si_ok[-1] if si_ok else None)[:300]) check("Studio-path ScanImage commands logged in order", si_cmd == ["ping", "setFileStem", "set", "set", "grab", "waitIdle", "abort", "get", "ping", "bogusCommand"], str(si_cmd)) grab = [e for e in si_rep if e.get("command") == "grab"] check("grab reply in the run log carries the TIFF artifact (stem, counter, path)", bool(grab) and grab[0].get("ok") and isinstance(grab[0].get("artifact"), dict) From 7cc7710926c83470974203b7e5c78785d39d983c Mon Sep 17 00:00:00 2001 From: hanmarsan Date: Wed, 21 Oct 2026 11:46:38 -0400 Subject: [PATCH 4/5] ros2-poc: zmq_latency_report picks the TIFF's segment by frame count + epoch (P2) Every acquisition numbers frames from 1, so matching by frameNumber alone took the first segment (Bergamo 2026-10-20: a FOCUS instead of the GRAB). A candidate now holds every TIFF frame, has exactly as many frames, and starts 0..--epoch-window-s (5) after the TIFF epoch; none or several stop with exit 2 and a listing, --segment overrides. A bar checks the ZMQ and TIFF timestamps agree. The epoch difference is reported as epoch_offset_ms (it holds ScanImage's arming -> first-line gap), with no latency bar. test-frames 38/38 (four new F checks incl. three segments, TIFF = last). Co-Authored-By: Claude Opus 5.5 --- ros2-poc/tests/test_scanimage_frames.py | 40 +++++++++- ros2-poc/tools/zmq_latency_report.py | 99 +++++++++++++++++++------ 2 files changed, 115 insertions(+), 24 deletions(-) diff --git a/ros2-poc/tests/test_scanimage_frames.py b/ros2-poc/tests/test_scanimage_frames.py index 1eb84cb3..af80812b 100644 --- a/ros2-poc/tests/test_scanimage_frames.py +++ b/ros2-poc/tests/test_scanimage_frames.py @@ -449,10 +449,46 @@ def paused_run(name: str, frames: int, **kw) -> tuple[float, dict | None, dict | t = s1 and s1.get("tiff") check("F: TIFF alignment: 87/87 frames matched, 2 pages per frame, ZMQ − TIFF timestamp = +5 ms", t and t["matched_frames"] == n_a and t["pages_per_frame"] == NCH and t["frame_meta_count_equals_tiff_frames"] and abs(t["zmq_minus_tiff_ts_ms"]["p50"] - 5.0) < 0.01, json.dumps(t)) - check("F: absolute latency from the TIFF epoch is finite and plausible (20 ms + the stream delay)", - t and t.get("absolute_latency_ms", {}).get("n") == n_a and 15.0 <= t["absolute_latency_ms"]["min"] <= 2000.0, json.dumps(t and t.get("absolute_latency_ms"))) + check("F: epoch offset from the TIFF epoch is finite and plausible (20 ms + the stream delay)", + t and t.get("epoch_offset_ms", {}).get("n") == n_a and 15.0 <= t["epoch_offset_ms"]["min"] <= 2000.0, json.dumps(t and t.get("epoch_offset_ms"))) check("F: the report's pass bars include the meta-latency and gap bars per segment", any("meta latency" in b["bar"] for b in rep["bars"]) and any(b["bar"].startswith("0 gaps") and b["segment"] == 2 and not b["pass"] for b in rep["bars"]), json.dumps(rep["bars"])) + check("F: the TIFF's segment is chosen by frame count + epoch (segment 1 the only candidate; timestamps agree)", + t and [c["segment"] for c in t["segment_choice"] if c["candidate"]] == [1] and t["timestamps_agree"], json.dumps(t and t["segment_choice"])) print(r.stdout[-1500:], file=sys.stderr) + # three acquisitions that each number frames from 1, with the TIFF = the LAST one (2026-10-20's wrong match) + syn = work / "three_segments.jsonl" + base = metas1[0]["rx_epoch_ms"] + with open(syn, "w", encoding="utf-8") as fh: + for sid in (1, 2, 3): + for k in range(1, 21): + t_ms = sid * 10_000 + (k - 1) * 1000 // 29 + fh.write(json.dumps({"topic": "/scanimage/frame_meta", "seq": sid * 100 + k, "segment": sid, "frameNumber": k, + "frameTimestamp_s": (k - 1) / 29.0, "rx_epoch_ms": base + t_ms, "rx_mono_ns": t_ms * 1_000_000, + "pub_mono_ns": t_ms * 1_000_000 + 100_000, "queue_depth": 0, "dims": [8, 8, 1], "nchan": 1}) + "\n") + + def syn_tiff(name: str, epoch_ms: int) -> Path: + p = work / name + e = dt.datetime.fromtimestamp(epoch_ms / 1000.0, dt.timezone.utc) + write_si_tiff(p, [(k, (k - 1) / 29.0) for k in range(1, 21)], 1, [e.year, e.month, e.day, e.hour, e.minute, e.second + e.microsecond / 1e6]) + return p + + def syn_report(tif: Path, *extra: str): + out = work / f"{tif.stem}{'-'.join(extra)}.json" + rr = subprocess.run([PY, str(tool), "--jsonl", str(syn), "--tiff", str(tif), "--utc-offset-h", "0", "--json", str(out), *extra], + capture_output=True, text=True, timeout=60) + seg_t = {s["id"]: s.get("tiff") for s in json.loads(out.read_text())["segments"]} if rr.returncode == 0 and out.exists() else {} + return rr, seg_t + + rr, seg_t = syn_report(syn_tiff("syn_last.tif", base + 30_000 - 560)) # armed 0.56 s before segment 3's first frame + check("F: three segments numbered from 1, TIFF = the last: the report picks segment 3, all 20 frames, timestamps agree", + rr.returncode == 0 and seg_t.get(3) and not seg_t.get(1) and seg_t[3]["matched_frames"] == 20 and seg_t[3]["timestamps_agree"], + (rr.stdout + rr.stderr)[-400:]) + rr, _ = syn_report(syn_tiff("syn_none.tif", base - 60_000)) + check("F: a TIFF no segment fits (epoch 60 s early) → exit 2, listing the segments, asking for --segment", + rr.returncode == 2 and "--segment" in rr.stderr and "segment 3:" in rr.stderr, (rr.stdout + rr.stderr)[-400:]) + rr, seg_t = syn_report(work / "syn_none.tif", "--segment", "2") + check("F: --segment overrides the choice (segment 2)", rr.returncode == 0 and seg_t.get(2) and seg_t[2]["matched_frames"] == 20, + (rr.stdout + rr.stderr)[-400:]) # a bag the way the bridge records it (raw CDR, arrival stamps), then the report's bag path from recorder import Recorder # noqa: E402 (nodes/) diff --git a/ros2-poc/tools/zmq_latency_report.py b/ros2-poc/tools/zmq_latency_report.py index 00ec7ca3..a3ba5995 100644 --- a/ros2-poc/tools/zmq_latency_report.py +++ b/ros2-poc/tools/zmq_latency_report.py @@ -9,10 +9,13 @@ acquisition clock (vDAQ, since acquisition start) and the host clock share no origin, so without the TIFF this is the jitter/backlog part of the latency, not its constant. p50 / p99 / max, plus the drift of the two clocks (ms per s, a linear fit). - absolute latency with --tiff: the TIFF's `epoch` (acquisition start, PC wall clock) + its - `frameTimestamps_sec` give each frame's acquisition wall time; rx_epoch_ms − that is - the frame-to-topic latency on the same PC. Also the per-frame difference between the - ZMQ header timestamp and the TIFF's (the "same origin?" question of the procedure). + TIFF match with --tiff: which segment the TIFF is (choose_tiff_segment: frame count + epoch, an + error when none or several fit; --segment overrides), every TIFF frame found in it, + and the per-frame difference between the ZMQ header timestamp and the TIFF's (one + constant: the same acquisition clock; 0 on the Bergamo, 2026-10-20). + epoch offset with --tiff: rx_epoch_ms − (the TIFF's `epoch` + `frameTimestamps_sec`). NOT a latency: + ScanImage's epoch is when it armed the acquisition, before the shutter's openTime_s and + the first line (~0.56 s on 2026-10-20), so it has no pass bar. hops receive → publish inside the node (`pub_mono_ns − rx_mono_ns`) and, from a bag, node → bridge record (MCAP log_time − rx_epoch_ms), like FicTrac's connector_status. preview encode+publish per image (preview meta) and, from a bag, frame receive → preview @@ -238,9 +241,50 @@ def epoch_wall_ms(epoch: list, utc_offset_h: float | None) -> int: # ── analysis ────────────────────────────────────────────────────────────────────────────── +EPOCH_WINDOW_S = 5.0 # the TIFF's segment starts this soon after its epoch (Bergamo 2026-10-20: 0.56 s) + + +class SegmentChoiceError(ValueError): + """The TIFF matches no segment, or more than one; the message lists them all.""" + + +def choose_tiff_segment(by_seg: dict, tiff: dict, utc_offset_h: float | None = None, + segment: int | None = None, window_s: float = EPOCH_WINDOW_S) -> tuple[int, list[dict]]: + """Which acquisition segment a TIFF belongs to → (segment id, one row per segment). + Every segment numbers its frames from 1, so frame numbers alone match the first segment, not the + right one (Bergamo 2026-10-20: a FOCUS instead of the GRAB). A candidate holds every TIFF frame + number, has exactly as many frames as the TIFF and, when the TIFF has an epoch, its first frame + reached the node 0..window_s after that epoch. Exactly one candidate is the answer; none or + several raise SegmentChoiceError. `segment` overrides the choice.""" + n_tiff = len(tiff["frames"]) + e_ms = epoch_wall_ms(tiff["epoch"], utc_offset_h) if tiff.get("epoch") else None + rows = [] + for sid in sorted(by_seg): + recs = by_seg[sid] + nums = {int(r["obj"]["frameNumber"]) for r in recs} + found = sum(1 for k in tiff["frames"] if k in nums) + off_s = (min(r["obj"]["rx_epoch_ms"] for r in recs) - e_ms) / 1000.0 if e_ms is not None else None + rows.append({"segment": sid, "frames": len(recs), "tiff_frames_found": found, + "first_rx_after_epoch_s": round(off_s, 3) if off_s is not None else None, + "candidate": found == n_tiff and len(recs) == n_tiff and (off_s is None or 0.0 <= off_s <= window_s)}) + listing = "; ".join(f"segment {r['segment']}: {r['frames']} frames, {r['tiff_frames_found']}/{n_tiff} TIFF frames found, " + f"first frame {r['first_rx_after_epoch_s']} s after the epoch" for r in rows) + if segment is not None: + if segment not in by_seg: + raise SegmentChoiceError(f"--segment {segment} is not in the stream ({listing})") + return segment, rows + cands = [r["segment"] for r in rows if r["candidate"]] + if len(cands) == 1: + return cands[0], rows + raise SegmentChoiceError(f"{'no segment matches' if not cands else f'{len(cands)} segments match'} the TIFF " + f"({n_tiff} frames{', epoch window 0..%g s' % window_s if e_ms is not None else ', no epoch'}); " + f"pass --segment N. {listing}") + + def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | None = None, frame_period_ms: float | None = None, segment_filter: int | None = None, - meta_p99_ms: float | None = None, preview_p99_ms: float = 250.0) -> dict: + meta_p99_ms: float | None = None, preview_p99_ms: float = 250.0, + epoch_window_s: float = EPOCH_WINDOW_S) -> dict: metas = [r for r in records if r["topic"] == META_TOPIC and isinstance(r["obj"].get("seq"), int)] stats = [r["obj"] for r in records if r["topic"] == STATS_TOPIC] pmetas = [r for r in records if r["topic"] == PREVIEW_META_TOPIC] @@ -251,15 +295,10 @@ def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | last_stats = stats[-1] if stats else {} stats_segments = {s["id"]: s for s in last_stats.get("segment_list", [])} if last_stats else {} - # which segment the TIFF belongs to: the one sharing the most frame numbers (or --segment) - tiff_seg = None + # which segment the TIFF belongs to: by frame count and epoch, never by frame numbers alone + tiff_seg, choice = None, None if tiff and tiff["frames"]: - cands = [segment_filter] if segment_filter else list(by_seg) - best = 0 - for sid in cands: - n = sum(1 for r in by_seg.get(sid, []) if int(r["obj"]["frameNumber"]) in tiff["frames"]) - if n > best: - best, tiff_seg = n, sid + tiff_seg, choice = choose_tiff_segment(by_seg, tiff, utc_offset_h, segment_filter, epoch_window_s) segments = [] for sid in sorted(by_seg): @@ -317,13 +356,18 @@ def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | delta_ms = [(float(o["frameTimestamp_s"]) - t["ts"]) * 1000.0 for o, t in pairs if t["ts"] is not None] t_align = {"tiff": tiff["path"], "tiff_frames": len(tiff["frames"]), "tiff_pages": tiff["pages"], "matched_frames": len(pairs), "zmq_minus_tiff_ts_ms": summarize(delta_ms), + # the same frames carry the same acquisition clock: one constant offset (0 on the rig) + "timestamps_agree": bool(delta_ms) and len(delta_ms) == len(pairs) and max(delta_ms) - min(delta_ms) < 1.0, "pages_per_frame": Counter(t["pages"] for t in tiff["frames"].values()).most_common(1)[0][0] if tiff["frames"] else None, - "frame_meta_count_equals_tiff_frames": len(objs) == len(tiff["frames"])} + "frame_meta_count_equals_tiff_frames": len(objs) == len(tiff["frames"]), + "segment_choice": choice} if tiff["epoch"]: e_ms = epoch_wall_ms(tiff["epoch"], utc_offset_h) - abs_ms = [o["rx_epoch_ms"] - (e_ms + t["ts"] * 1000.0) for o, t in pairs if t["ts"] is not None] + # NOT a latency: ScanImage's epoch is when it armed the acquisition, before the shutter's + # openTime_s and the first line, so this holds that start-up gap (~0.56 s on 2026-10-20) + off_ms = [o["rx_epoch_ms"] - (e_ms + t["ts"] * 1000.0) for o, t in pairs if t["ts"] is not None] t_align.update({"epoch": tiff["epoch"], "epoch_wall_ms": e_ms, "utc_offset_h": utc_offset_h, - "absolute_latency_ms": summarize(abs_ms)}) + "epoch_offset_ms": summarize(off_ms)}) seg["tiff"] = t_align segments.append(seg) @@ -331,10 +375,14 @@ def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | bars = [] for seg in segments: period = meta_p99_ms or seg["frame_period_ms"] or 34.0 - lat = seg.get("tiff", {}).get("absolute_latency_ms") or seg["stream_delay_rel_ms"] - kind = "absolute (TIFF clock)" if seg.get("tiff", {}).get("absolute_latency_ms") else "relative (fastest frame = 0)" + lat = seg["stream_delay_rel_ms"] # the TIFF's epoch offset holds the start-up gap: no bar on it if lat.get("n"): - bars.append({"segment": seg["id"], "bar": f"meta latency p99 < {period:.1f} ms ({kind})", "value": lat["p99"], "pass": lat["p99"] < period}) + bars.append({"segment": seg["id"], "bar": f"meta latency p99 < {period:.1f} ms (relative (fastest frame = 0))", "value": lat["p99"], "pass": lat["p99"] < period}) + if seg.get("tiff"): + t = seg["tiff"] + bars.append({"segment": seg["id"], "bar": "TIFF frames all in the stream, timestamps agree (constant offset)", + "value": f"{t['matched_frames']}/{t['tiff_frames']}, ZMQ − TIFF p50 {t['zmq_minus_tiff_ts_ms'].get('p50')} ms", + "pass": t["matched_frames"] == t["tiff_frames"] and t["timestamps_agree"]}) pv = seg["preview"]["frame_to_record_ms"] if seg["preview"]["frame_to_record_ms"].get("n") else seg["preview"]["encode_publish_ms"] if pv.get("n"): bars.append({"segment": seg["id"], "bar": f"preview p99 < {preview_p99_ms:g} ms", "value": pv["p99"], "pass": pv["p99"] < preview_p99_ms}) @@ -378,7 +426,8 @@ def markdown(rep: dict, source: str) -> str: if t: out.append(f"- segment {s['id']} vs TIFF {t['tiff']}: {t['matched_frames']} of {t['tiff_frames']} TIFF frames matched ({t['tiff_pages']} pages, {t['pages_per_frame']} per frame); " f"frame_meta count == TIFF frames: {t['frame_meta_count_equals_tiff_frames']}; ZMQ − TIFF timestamp p50/p99/max ms {fmt(t['zmq_minus_tiff_ts_ms'])}" - + (f"; ABSOLUTE latency (acquisition → topic) p50/p99/max ms {fmt(t['absolute_latency_ms'])}" if t.get("absolute_latency_ms") else "; no `epoch` in the TIFF → no absolute latency")) + + (f"; epoch offset (topic − (epoch + frame time); includes ScanImage's arming → first-line gap, NOT a latency) p50/p99/max ms {fmt(t['epoch_offset_ms'])}" + if t.get("epoch_offset_ms") else "; no `epoch` in the TIFF → no epoch offset")) out += ["", "| pass bar | segment | value | result |", "|---|---|---|---|"] for b in rep["bars"]: out.append(f"| {b['bar']} | {b['segment'] if b['segment'] is not None else 'all'} | {b['value']} | {'PASS' if b['pass'] else 'FAIL'} |") @@ -392,7 +441,9 @@ def main(argv=None) -> int: src.add_argument("--jsonl", help="the node's --dump file") ap.add_argument("--tiff", help="the acquisition's ScanImage TIFF (frameNumbers / frameTimestamps_sec / epoch in its page descriptions)") ap.add_argument("--utc-offset-h", type=float, default=None, help="the ScanImage PC's UTC offset for the TIFF epoch (default: this machine's local zone)") - ap.add_argument("--segment", type=int, default=None, help="only this acquisition segment") + ap.add_argument("--segment", type=int, default=None, help="only this acquisition segment (with --tiff: the TIFF's segment, overriding the choice)") + ap.add_argument("--epoch-window-s", type=float, default=EPOCH_WINDOW_S, + help=f"with --tiff: the TIFF's segment starts 0..this many s after its epoch (default {EPOCH_WINDOW_S:g})") ap.add_argument("--frame-period-ms", type=float, default=None, help="override the measured frame period for the pass bar") ap.add_argument("--meta-p99-ms", type=float, default=None, help="pass bar for meta latency p99 (default: one frame period)") ap.add_argument("--preview-p99-ms", type=float, default=250.0) @@ -407,7 +458,11 @@ def main(argv=None) -> int: records = load_jsonl(Path(a.jsonl)) source = a.jsonl tiff = tiff_frames(Path(a.tiff)) if a.tiff else None - rep = analyze(records, tiff, a.utc_offset_h, a.frame_period_ms, a.segment, a.meta_p99_ms, a.preview_p99_ms) + try: + rep = analyze(records, tiff, a.utc_offset_h, a.frame_period_ms, a.segment, a.meta_p99_ms, a.preview_p99_ms, a.epoch_window_s) + except SegmentChoiceError as exc: + print(f"which segment is the TIFF? {exc}", file=sys.stderr) + return 2 rep["source"] = source if not rep["segments"]: print(f"no /scanimage/frame_meta records in {source}", file=sys.stderr) From 15db22c11ab55c7fdc153595b0fc7c69c9cf7e20 Mon Sep 17 00:00:00 2001 From: Michael B Reiser Date: Thu, 22 Oct 2026 08:34:03 -0400 Subject: [PATCH 5/5] ros2-poc: zmq_latency_report survives the simulated vDAQ's 1601 TIFF epoch + warns on missed_before_first (LAB-303) ScanImage 2026's simulated vDAQ writes epoch = [1601 1 1 0 0 0]. On Windows the local-zone path (no --utc-offset-h) crashed in naive.astimezone() (OSError before 1970; Bergamo PC 2026-10-22), and with an offset the 1601 epoch made every segment miss the epoch window, so --segment was needed. - epoch_wall_ms returns None for a year before 2000 or a date the platform can't convert; the epoch is then ignored with a warning: no epoch offset, the TIFF's segment chosen by frame count alone. - A segment whose first frame is > 1 (a late join, or a reconnect mid-acquisition) now gets a warning in the markdown, the JSON (`warnings`) and on stderr. Still not a pass bar. - Harness (tests/test_scanimage_frames.py): 3 new F checks; 46/46 pass on the Mac. Co-Authored-By: Claude Opus 5.5 --- ros2-poc/tests/test_scanimage_frames.py | 31 +++++++++++++ ros2-poc/tools/zmq_latency_report.py | 61 +++++++++++++++++++------ 2 files changed, 78 insertions(+), 14 deletions(-) diff --git a/ros2-poc/tests/test_scanimage_frames.py b/ros2-poc/tests/test_scanimage_frames.py index af80812b..61e60485 100644 --- a/ros2-poc/tests/test_scanimage_frames.py +++ b/ros2-poc/tests/test_scanimage_frames.py @@ -489,6 +489,37 @@ def syn_report(tif: Path, *extra: str): rr, seg_t = syn_report(work / "syn_none.tif", "--segment", "2") check("F: --segment overrides the choice (segment 2)", rr.returncode == 0 and seg_t.get(2) and seg_t[2]["matched_frames"] == 20, (rr.stdout + rr.stderr)[-400:]) + # the simulated vDAQ's TIFF epoch [1601 1 1 0 0 0] in the local-zone path (no --utc-offset-h): it crashed + # the report on Windows (astimezone() before 1970, Bergamo PC 2026-10-22); now ignored with a warning + sim_tif = work / "syn_sim1601.tif" + write_si_tiff(sim_tif, [(k, (k - 1) / 29.0) for k in range(1, 21)], 1, [1601, 1, 1, 0, 0, 0]) + sim_out = work / "syn_sim1601.json" + rr = subprocess.run([PY, str(tool), "--jsonl", str(syn), "--tiff", str(sim_tif), "--segment", "2", "--json", str(sim_out)], + capture_output=True, text=True, timeout=60) + rep_s = json.loads(sim_out.read_text()) if rr.returncode == 0 and sim_out.exists() else {} + t2 = next((s.get("tiff") for s in rep_s.get("segments", []) if s["id"] == 2), None) + check("F: a 1601 TIFF epoch (simulated vDAQ) is ignored with a warning: no crash, no epoch offset, 20/20 frames matched", + rr.returncode == 0 and t2 and t2["matched_frames"] == 20 and t2.get("epoch_usable") is False and "epoch_offset_ms" not in t2 + and any("not a usable wall time" in w for w in rep_s.get("warnings", [])) and "warning: TIFF epoch" in rr.stderr, + (rr.stdout + rr.stderr)[-400:]) + rr = subprocess.run([PY, str(tool), "--jsonl", str(syn), "--tiff", str(sim_tif)], capture_output=True, text=True, timeout=60) + check("F: with the epoch ignored, three equal segments are ambiguous → exit 2 asking for --segment (no traceback)", + rr.returncode == 2 and "3 segments match" in rr.stderr and "Traceback" not in rr.stderr, (rr.stdout + rr.stderr)[-400:]) + # a late join: the node's first frame is 6 → missed_before_first 5, reported as a warning (never a bar) + late = work / "late_join.jsonl" + with open(late, "w", encoding="utf-8") as fh: + for k in range(6, 21): + t_ms = (k - 1) * 1000 // 29 + fh.write(json.dumps({"topic": "/scanimage/frame_meta", "seq": k - 5, "segment": 1, "frameNumber": k, + "frameTimestamp_s": (k - 1) / 29.0, "rx_epoch_ms": base + t_ms, "rx_mono_ns": t_ms * 1_000_000, + "pub_mono_ns": t_ms * 1_000_000 + 100_000, "queue_depth": 0, "dims": [8, 8, 1], "nchan": 1}) + "\n") + late_out = work / "late_join.json" + rr = subprocess.run([PY, str(tool), "--jsonl", str(late), "--json", str(late_out)], capture_output=True, text=True, timeout=60) + rep_l = json.loads(late_out.read_text()) if rr.returncode == 0 and late_out.exists() else {} + check("F: a late join (first frame 6) → missed_before_first 5, a warning in the JSON, the markdown and on stderr", + bool(rep_l) and rep_l["segments"][0]["missed_before_first"] == 5 + and any("first 5 frame(s) never reached the node" in w for w in rep_l["warnings"]) + and "WARNING: segment 1" in rr.stdout and "warning: segment 1" in rr.stderr, (rr.stdout + rr.stderr)[-400:]) # a bag the way the bridge records it (raw CDR, arrival stamps), then the report's bag path from recorder import Recorder # noqa: E402 (nodes/) diff --git a/ros2-poc/tools/zmq_latency_report.py b/ros2-poc/tools/zmq_latency_report.py index a3ba5995..35369e1d 100644 --- a/ros2-poc/tools/zmq_latency_report.py +++ b/ros2-poc/tools/zmq_latency_report.py @@ -29,6 +29,11 @@ preview p99 < --preview-p99-ms (250), 0 gaps after the first frame, 0 dropped_by_queue. Printed as a markdown table; --json writes everything; --strict exits 1 when a bar fails. +Warnings (printed and in the JSON, never a bar): frames that never reached the node before a segment's +first frame (missed_before_first > 0: a late join, or a reconnect mid-acquisition), and a TIFF `epoch` +that is not a wall time (ScanImage's simulated vDAQ writes [1601 1 1 0 0 0]), which is then ignored: +no epoch offset, and the TIFF's segment is chosen by frame count alone. + pixi run zmq-latency -- --bag runlogs/.bag --json latency_focus60.json pixi run zmq-latency -- --bag runlogs/.bag --tiff "F:\\data\\run_00003.tif" --segment 2 pixi run zmq-latency -- --jsonl frames.jsonl @@ -226,18 +231,29 @@ def tiff_frames(path: Path) -> dict: return {"frames": frames, "epoch": epoch, "pages": pages, "path": str(path)} -def epoch_wall_ms(epoch: list, utc_offset_h: float | None) -> int: +EPOCH_MIN_YEAR = 2000 # an earlier TIFF `epoch` is not a wall time (the simulated vDAQ writes 1601-01-01) + + +def epoch_wall_ms(epoch: list, utc_offset_h: float | None) -> int | None: """ScanImage's `epoch = [y m d H M S.fff]` (PC local time) → wall ms. The offset defaults to this - machine's local zone (the tool normally runs on the ScanImage PC).""" + machine's local zone (the tool normally runs on the ScanImage PC). None when the epoch is not a + usable wall time: a year before EPOCH_MIN_YEAR (ScanImage 2026's simulated vDAQ writes + [1601 1 1 0 0 0], the Windows FILETIME zero; Bergamo PC 2026-10-22) or a date this platform cannot + convert (on Windows, `astimezone()` raises OSError for anything before 1970).""" y, mo, d, h, mi, s = epoch[:6] + if y < EPOCH_MIN_YEAR: + return None sec = int(s) frac = s - sec - naive = dt.datetime(int(y), int(mo), int(d), int(h), int(mi), sec) - if utc_offset_h is None: - t = naive.astimezone() # interpret as local time - else: - t = naive.replace(tzinfo=dt.timezone(dt.timedelta(hours=utc_offset_h))) - return int(t.timestamp() * 1000) + int(round(frac * 1000)) + try: + naive = dt.datetime(int(y), int(mo), int(d), int(h), int(mi), sec) + if utc_offset_h is None: + t = naive.astimezone() # interpret as local time + else: + t = naive.replace(tzinfo=dt.timezone(dt.timedelta(hours=utc_offset_h))) + return int(t.timestamp() * 1000) + int(round(frac * 1000)) + except (OverflowError, OSError, ValueError): + return None # ── analysis ────────────────────────────────────────────────────────────────────────────── @@ -295,6 +311,12 @@ def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | last_stats = stats[-1] if stats else {} stats_segments = {s["id"]: s for s in last_stats.get("segment_list", [])} if last_stats else {} + warnings: list[str] = [] + if tiff and tiff.get("epoch") and epoch_wall_ms(tiff["epoch"], utc_offset_h) is None: + warnings.append(f"TIFF epoch {[round(v, 3) for v in tiff['epoch'][:6]]} is not a usable wall time " + "(ScanImage's simulated vDAQ writes 1601-01-01): ignored, so there is no epoch offset " + "and the TIFF's segment is chosen by frame count alone") + # which segment the TIFF belongs to: by frame count and epoch, never by frame numbers alone tiff_seg, choice = None, None if tiff and tiff["frames"]: @@ -363,12 +385,17 @@ def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | "segment_choice": choice} if tiff["epoch"]: e_ms = epoch_wall_ms(tiff["epoch"], utc_offset_h) - # NOT a latency: ScanImage's epoch is when it armed the acquisition, before the shutter's - # openTime_s and the first line, so this holds that start-up gap (~0.56 s on 2026-10-20) - off_ms = [o["rx_epoch_ms"] - (e_ms + t["ts"] * 1000.0) for o, t in pairs if t["ts"] is not None] - t_align.update({"epoch": tiff["epoch"], "epoch_wall_ms": e_ms, "utc_offset_h": utc_offset_h, - "epoch_offset_ms": summarize(off_ms)}) + t_align.update({"epoch": tiff["epoch"], "epoch_usable": e_ms is not None}) + if e_ms is not None: + # NOT a latency: ScanImage's epoch is when it armed the acquisition, before the shutter's + # openTime_s and the first line, so this holds that start-up gap (~0.56 s on 2026-10-20) + off_ms = [o["rx_epoch_ms"] - (e_ms + t["ts"] * 1000.0) for o, t in pairs if t["ts"] is not None] + t_align.update({"epoch_wall_ms": e_ms, "utc_offset_h": utc_offset_h, "epoch_offset_ms": summarize(off_ms)}) seg["tiff"] = t_align + if seg["missed_before_first"]: + warnings.append(f"segment {sid}: the first {seg['missed_before_first']} frame(s) never reached the node " + f"(its first frame is {seg['first_frame']}): a late join (start the node before Grab) " + "or a reconnect mid-acquisition") segments.append(seg) # pass bars @@ -395,6 +422,7 @@ def analyze(records: list[dict], tiff: dict | None = None, utc_offset_h: float | "last_stats": {k: last_stats.get(k) for k in ("frames", "published", "dropped_by_queue", "bad_frames", "gaps", "gap_frames", "segments", "connects", "disconnects", "queue_limit", "uptime_s")} if last_stats else None, "segments": segments, "bars": bars, "all_pass": all(b["pass"] for b in bars) if bars else None, + "warnings": warnings, } @@ -408,6 +436,8 @@ def markdown(rep: dict, source: str) -> str: f"frame_meta {rep['frame_meta']}, frame_stats {rep['frame_stats']}, preview meta {rep['preview_meta']}, preview images {rep['preview_images']}"] if rep["last_stats"]: out.append("node totals (last frame_stats): " + ", ".join(f"{k}={v}" for k, v in rep["last_stats"].items())) + for w in rep.get("warnings", []): + out.append(f"- WARNING: {w}") out += ["", "| segment | frames (first–last) | missed before first | gaps (frames) | rate Hz | stream delay rel p50/p99/max ms | node hop p50/p99/max ms | node→bridge p50/p99/max ms | queue p99/max | preview enc p50/p99/max ms | preview frame→record p50/p99/max ms |", "|---|---|---|---|---|---|---|---|---|---|---|"] for s in rep["segments"]: @@ -427,7 +457,8 @@ def markdown(rep: dict, source: str) -> str: out.append(f"- segment {s['id']} vs TIFF {t['tiff']}: {t['matched_frames']} of {t['tiff_frames']} TIFF frames matched ({t['tiff_pages']} pages, {t['pages_per_frame']} per frame); " f"frame_meta count == TIFF frames: {t['frame_meta_count_equals_tiff_frames']}; ZMQ − TIFF timestamp p50/p99/max ms {fmt(t['zmq_minus_tiff_ts_ms'])}" + (f"; epoch offset (topic − (epoch + frame time); includes ScanImage's arming → first-line gap, NOT a latency) p50/p99/max ms {fmt(t['epoch_offset_ms'])}" - if t.get("epoch_offset_ms") else "; no `epoch` in the TIFF → no epoch offset")) + if t.get("epoch_offset_ms") else "; the TIFF's `epoch` is not a usable wall time → no epoch offset" + if t.get("epoch") else "; no `epoch` in the TIFF → no epoch offset")) out += ["", "| pass bar | segment | value | result |", "|---|---|---|---|"] for b in rep["bars"]: out.append(f"| {b['bar']} | {b['segment'] if b['segment'] is not None else 'all'} | {b['value']} | {'PASS' if b['pass'] else 'FAIL'} |") @@ -468,6 +499,8 @@ def main(argv=None) -> int: print(f"no /scanimage/frame_meta records in {source}", file=sys.stderr) return 2 print(markdown(rep, source)) + for w in rep["warnings"]: + print(f"warning: {w}", file=sys.stderr) if a.json: Path(a.json).write_text(json.dumps(rep, indent=1), encoding="utf-8") print(f"\nwritten {a.json}")