From cbf604f14fcc41828e652fef8c4349072781eb28 Mon Sep 17 00:00:00 2001 From: Michael B Reiser Date: Sun, 18 Oct 2026 23:00:03 -0400 Subject: [PATCH] ros2-poc: the FicTrac record in the bag = ONE /arena/frame message per frame with 10 columns (LAB-303) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Decided 2026-10-18 (docs/development/fictrac-columns-in-the-bag.md): the bag keeps 10 of FicTrac's 25 columns — the behavior row's 5 plus the error score (col 5), the lab-frame delta rotation (cols 6-8) and the sequence counter (col 23) — on the per-frame /arena/frame record, float64, one MCAP message per frame; the adapter's raw topics (/fictrac/frame, /fictrac/raw: 25 columns + the verbatim line, 68-77 % of a bag) are recorded only with --record-fictrac-raw. - ArenaFrame.msg: err, dr_lab_x/y/z, ft_seq, ft_ns (col 22 as received), flags (bit 0 = untracked FT_BADFR frame). The recorded copy carries NaN in turning_dps/forward_mms: they are live-only, exactly recomputable, and measured at 16 B/frame of incompressible doubles. - bridge_node: the consumer parses each line once (hub.fields) so the record carries the columns of the SAME line as x/y/hd; an FT_BADFR line becomes a flagged record (no run-log row) and connector_status counts bad_frames; raw FicTrac topics gated by --record-fictrac-raw. - _rows.py: parse_fictrac_line (one parser for adapter, node, converter), fictrac_extras, arena_frame_to_row (the ONE record→row mapping; None for a flagged frame). - mcap_convert: rows via _rows; flagged frames skipped; --level full refused without the raw topics (old bags that have them still convert, full included). - mcap_index.reindex: chunk-only index + zstd 19 (the final layout); recorder `default` profile: Zstd / Slowest / noMessageIndex through the plugin's storage options. - sizing/columns.py + `pixi run size-columns`: the bag-vs-jsonl.gz study, reproducible. - tests: merged-record alignment (wrap(dhd) == dr_lab_z), bad-frame injection, full-level rule, pre-2026-10-18 bag still converts, FrameDeriver math, 72 B/frame size guard. pixi run test 60/60 (default env), pixi run -e typed test 63/63. Co-Authored-By: Claude Fable 5.1 --- .../telemetry-logging-reference.md | 11 + ros2-poc/README.md | 6 +- .../arena_interfaces/msg/ArenaFrame.msg | 19 +- ros2-poc/nodes/_rows.py | 82 +++++- ros2-poc/nodes/bridge_node.py | 127 +++++--- ros2-poc/nodes/fictrac_udp_adapter.py | 17 +- ros2-poc/nodes/mcap_convert.py | 46 ++- ros2-poc/nodes/mcap_index.py | 41 ++- ros2-poc/nodes/recorder.py | 10 +- ros2-poc/pixi.toml | 3 + ros2-poc/sizing/columns.py | 275 ++++++++++++++++++ ros2-poc/tests/test_ros2_poc.py | 185 ++++++++++-- 12 files changed, 719 insertions(+), 103 deletions(-) create mode 100644 ros2-poc/sizing/columns.py diff --git a/docs/development/telemetry-logging-reference.md b/docs/development/telemetry-logging-reference.md index 8fd55c05..b3acd26a 100644 --- a/docs/development/telemetry-logging-reference.md +++ b/docs/development/telemetry-logging-reference.md @@ -57,6 +57,17 @@ must skip it (rule 2 in the hand-off §7.2). After #198 merges, all readers belo `readRunlogText`). **v0.78 guarantee:** the final ring drain and the `trial_quality` event are written **before** the export, on normal and on fault runs (v0.77 files can have controller rows after `trial_quality` — see §7). + +### 1.x The MCAP bag (ROS 2 bridge, LAB-303) and the FicTrac columns it keeps + +On the ROS 2 graph (`ros2-poc/`, PR #254) the record of truth is an MCAP bag per run and this +NDJSON is DERIVED from it by `ros2-poc/nodes/mcap_convert.py`; every row and event above is +unchanged. The bag's FicTrac record is the per-frame `/arena/frame` message: the behavior row's +five columns plus FicTrac's error score (col 5), lab-frame delta rotation (cols 6–8) and sequence +counter (col 23), `float64`, one message per frame — the other 15 columns and the verbatim line are +not recorded by default (`--record-fictrac-raw` opts in; the `full` level needs it). Why those ten +and what it costs (≈ 2.2–2.5× this file's `.gz`): `fictrac-columns-in-the-bag.md`. + ## 2. Row catalogue (arrays) All examples are verbatim lines from `arena-log-20260913-120611-713.jsonl` unless noted. "Clock" names the owner of the diff --git a/ros2-poc/README.md b/ros2-poc/README.md index aa147a01..42dfe39b 100644 --- a/ros2-poc/README.md +++ b/ros2-poc/README.md @@ -187,10 +187,10 @@ without further changes; **"ROS Studio" = that Studio served from the `/ros/` Pa | item | where | what | |---|---|---| -| MCAP record | `nodes/recorder.py`, `bridge_node.py` | one bag per run (opened with the run's log, closed at `log_control off` / `log_export` / shutdown) holding every device topic as received (raw CDR bytes), `/arena/frame` (the closed-loop evaluation with the node's **own sample time** `t_ms`/`ms` as fields, and as MCAP `publish_time`), `/arena/browser` (**every** browser message, verbatim, with the receive stamp), `/arena/event` (bias_config, heading_tare, connector_status, source_lost/resumed, warn), `/arena/session`. Bridge-side records carry the writer's emission sequence number `seq`. Two writer profiles: `crashsafe` (no chunking, no cache; the default) and rosbag2's `default`. | +| MCAP record | `nodes/recorder.py`, `bridge_node.py` | one bag per run (opened with the run's log, closed at `log_control off` / `log_export` / shutdown) holding every device topic as received (raw CDR bytes) — except the FicTrac topics, which since 2026-10-18 are recorded only with `--record-fictrac-raw` ([docs/development/fictrac-columns-in-the-bag.md](../docs/development/fictrac-columns-in-the-bag.md)) — `/arena/frame` (the closed-loop evaluation with the node's **own sample time** `t_ms`/`ms` as fields, and as MCAP `publish_time`, plus the 10 FicTrac columns the lab keeps: the behavior row's 5 and `err`, `dr_lab_x/y/z`, `ft_seq`, `ft_ns`; an untracked `FT_BADFR` frame is a flagged record), `/arena/browser` (**every** browser message, verbatim, with the receive stamp), `/arena/event` (bias_config, heading_tare, connector_status, source_lost/resumed, warn), `/arena/session`. Bridge-side records carry the writer's emission sequence number `seq`. Two writer profiles: `crashsafe` (no chunking, no cache; the default) and rosbag2's `default`. | | converter | `nodes/mcap_convert.py` (`pixi run convert`) | bag → NDJSON, pure Python (`mcap` + a 60-line CDR decoder driven by the schema text rosbag2 embeds): no ROS process needed, reads a torn bag with `--recover`. Reuses bridge.py's own pure encoders (`compact_arena_command`, schema line). Levels are a converter option (`--level behavior_v2|behavior_v1|full`); the run's acked level is the default. `--order seq` reproduces the live writer's order; `--order time` is what a separate closed-loop process would have to do. ONE ms base per run (the pipeline's t0 is set from the log's open stamp — bridge 3.4 takes two samples there, comparison.md appendix 4). `log_export` returns the converted log; the live NDJSON stays as a shadow file during the spike and `converted == live` is checked and printed at every export. | | typed messages | `interfaces/arena_interfaces/` (`FicTracFrame`, `ArenaFrame`, `ConnectorReply`) | built **by `pixi install -e typed`** (pixi-build preview + the RoboStack `pixi-build-ros` backend; the opt-in `typed` environment depends on the package by path and carries the ament/rosidl toolchain). The **default environment stays String-only** so a rig PC without a compiler can always install and run the graph; the nodes choose typed topics only when the package imports AND serializes (`_common.probe_interfaces`). `pixi run build-interfaces` is the colcon alternative. `ArenaFrame` and `FicTracFrame` are in use; `ArenaFrame` also carries two DERIVED channels for live viewing — `turning_dps` (heading rate, deg/s) and `forward_mms` (forward velocity, mm/s; ball diameter `--ball-mm`, 9 mm default) — computed in the bridge node from consecutive frames the way `js/kinematics.js` does (two-point difference on the FicTrac camera clock, NaN on a run's first frame); the hub message to the Studio and the behavior_v1 columns are unchanged. `ConnectorReply` is defined and built but the BIAS reply stays JSON-in-String (its shape is BIAS's, not ours — typing it would make the derived connector event a rebuild instead of a copy). | -| FicTrac as a typed topic | `fictrac_udp_adapter.py --typed`, `bridge_node.py --typed` | the adapter publishes `/fictrac/frame` (parsed columns + FT tag + receive stamps + the verbatim line) beside `/fictrac/raw`; the bridge subscribes to the typed topic when the package imports (`auto`). The mapping math still parses the verbatim line, so it stays byte-for-byte bridge 3.4's. | +| FicTrac as a typed topic | `fictrac_udp_adapter.py --typed`, `bridge_node.py --typed` | the adapter publishes `/fictrac/frame` (parsed columns + FT tag + receive stamps + the verbatim line) beside `/fictrac/raw`; the bridge subscribes to the typed topic when the package imports (`auto`). The mapping math still parses the verbatim line, so it stays byte-for-byte bridge 3.4's. The typed FicTrac topic is a LIVE topic; the bag records it only with `--record-fictrac-raw` (2026-10-18). | | harness | `tests/test_ros2_poc.py` | the 32 PoC checks now judged against the **converted** log, plus bag presence per topic, fidelity vs the shadow, levels, ordering, typed schemas, and a crash phase (bridge node SIGKILLed mid-run, both profiles): **46 / 46 in 46 s** on the Mac. | | Pages path | `scripts/tiers/stamp.py --ros REF`, `deploy-pages.yml` input `ros`, `pixi run ros-preview`, `js/build-channel.js` channel `ros` | a third, optional path `/ros/` beside Production and Next: this branch's tree, stamped `channel: "ros"`, magenta badge/banner "ROS STUDIO", tier-suffixed window names, `run_metadata.channel = "ros"`. Off by default; a problem there never blocks the other tiers; the production tree check skips it. Production and Next are untouched (tiers test 109 / 109). **Enabling it on the live site needs this tooling commit on `main`** (the deploy workflow runs from `main`): a tooling-only PR, then `gh workflow run deploy-pages.yml -f ros=`. Until then: `pixi run ros-preview` + `python -m http.server -d _site`. | @@ -206,7 +206,7 @@ without further changes; **"ROS Studio" = that Studio served from the `/ros/` Pa | recording overhead on the closed loop (adapter → bridge, `connector_status`) | FicTrac p50 0.87 / p99 1.65 ms, Kinefly p50 0.82 / p99 1.51 ms — the same as the PoC without recording (0.89 / 1.43) | | recorder receive time vs the closed-loop node's own sample time (`/arena/frame` `log_time − publish_time`) | p50 0.53 ms, p99 1.07 ms, max 1.2 ms (the sample time is a whole ms). The record carries both, so `idx` reconstruction uses the node's clock, never the recorder's. | | conversion | 962 lines in 50 ms (smoke run); 3949 lines (harness run, 197 kB NDJSON) from a 4.1 MB bag of 4889 records while open. Open, the bag is ~20× the NDJSON: it holds the raw inputs (every FicTrac line twice — typed fields + verbatim — every Kinefly envelope) and the outputs, uncompressed so a crash loses nothing; after the close-time re-index (zstd) it is ~3× the NDJSON (bench: 6.76 MB → 2.33 MB vs a 954 kB log). | -| **fidelity, sequenced lines** (frames, browser lines, bridge events, session/schema): converted vs the live writer | **byte-identical and in the same order** in every run (smoke, harness, both levels); `bag → behavior_v1` equals `bridge.py --convert` of the live v2 byte for byte; `full` carries the 25 FicTrac columns for every frame | +| **fidelity, sequenced lines** (frames, browser lines, bridge events, session/schema): converted vs the live writer | **byte-identical and in the same order** in every run (smoke, harness, both levels); `bag → behavior_v1` equals `bridge.py --convert` of the live v2 byte for byte; `full` carries the 25 FicTrac columns for every frame (bags recorded with the raw FicTrac topics; since 2026-10-18 that needs `--record-fictrac-raw`, otherwise `full` is refused) | | fidelity, DERIVED lines (2204 Kinefly rows + 31 BIAS reply/status events): content | identical apart from `ms`; `\|Δms\|` **p50 0 / p99 1 ms**, worst 2 ms in the typed run and 12 ms once (one row of 2236) in a String-topic run while the harness was killing and restarting the adapter (recorder receive vs the live writer's processing time) | | fidelity, derived lines: placement | 113 of 2235 (5 %) land after a different neighbouring line than in the live file (within the same 1–2 ms). The file as a whole is therefore **not** byte-identical when Kinefly/BIAS are present; it is when they are absent. | | `--order time` vs `--order seq` | same multiset of lines; **17 of 3949** positions differ (12 of 962 in the smoke run), all same-millisecond ties (`log_control` vs the first frame, runner `fictracApply` vs `bias_config`, `config` vs `heading_tare`); `closed-loop-report.py` output identical for both orders | diff --git a/ros2-poc/interfaces/arena_interfaces/msg/ArenaFrame.msg b/ros2-poc/interfaces/arena_interfaces/msg/ArenaFrame.msg index 4a8a2403..344ad927 100644 --- a/ros2-poc/interfaces/arena_interfaces/msg/ArenaFrame.msg +++ b/ros2-poc/interfaces/arena_interfaces/msg/ArenaFrame.msg @@ -2,9 +2,14 @@ # frame (LAB-303 Plan B, /arena/frame). THE field the record of truth needs is t_ms — the # closed-loop node's OWN sample time, the instant idx, the bias phase and a pending tare were # evaluated. A recorder's receive time lags it; offline reconstruction must use t_ms / ms. +# +# Since 2026-10-18 this is also THE FicTrac record in the bag (docs/development/ +# fictrac-columns-in-the-bag.md): the raw FicTrac topics are no longer recorded by default, and +# the 10 FicTrac columns the lab keeps ride here — cols 1, 15, 16, 17, 22 as before, plus +# cols 5, 6-8 and 23 below. One message per frame; one type for the live topic and the bag. int64 t_ms # the node's wall clock at evaluation (ms since the Unix epoch) int32 ms # the same instant relative to the run's t0 (behavior_v1 column `ms`) -int64 fc # FicTrac frame counter (col 1) +int64 fc # FicTrac frame counter (col 1); 0 when an untracked frame carried none int32 idx # displayed frame index (0-based) bool has_ft # false when FicTrac's col-22 timestamp was absent float64 ft # FicTrac camera timestamp relative to the first frame, ms (behavior_v1 `ft`) @@ -18,6 +23,16 @@ uint64 seq # the bridge writer's emission order within the run (0 whe # to the Studio and the behavior_v1 columns do not carry them. Two-point differences over # consecutive frames on the FicTrac camera clock `ft` (falling back to the bridge clock when # ft is absent), the way js/kinematics.js derives the same channels; NaN on the first frame -# of a run, when the clock does not advance, or when FicTrac restarted. +# of a run, when the clock does not advance, or when FicTrac restarted. In the BAG these two +# are always NaN (live-only since 2026-10-18): exactly recomputable offline, and 16 B per frame +# of incompressible doubles otherwise — a fifth of the record. float64 turning_dps # heading rate, deg/s (positive = increasing FicTrac heading) float64 forward_mms # forward velocity, mm/s (x/y step projected onto the heading × ball radius; --ball-mm, 9 mm default) +# The FicTrac columns kept beyond the behavior row (NaN / 0 when the line did not carry them). +float64 err # col 5: delta-rotation error score (FicTrac's optimizer residual; its only per-frame quality signal) +float64 dr_lab_x # cols 6-8: delta rotation vector in laboratory coordinates, radians/frame — +float64 dr_lab_y # the primary measurement everything else is integrated from +float64 dr_lab_z # (FicTrac: heading -= dr_lab_z; velx = dr_lab_y; vely = -dr_lab_x) +int64 ft_seq # col 23: sequence counter — equals fc until FicTrac's tracking is reset, then restarts at 1 +int64 ft_ns # col 22 as received, truncated to an integer (a ns camera clock on our rigs; FicTrac's doc says ms) +uint8 flags # bit 0: untracked frame (an FT_BADFR line — no columns; recorded so the bag says "untracked", not "missing"; no run-log row) diff --git a/ros2-poc/nodes/_rows.py b/ros2-poc/nodes/_rows.py index 97f18269..ca975912 100644 --- a/ros2-poc/nodes/_rows.py +++ b/ros2-poc/nodes/_rows.py @@ -1,14 +1,90 @@ -"""_rows.py — pure row encoders shared by the bridge node (live shadow file) and the converter. +"""_rows.py — pure row/record helpers shared by the bridge node (live shadow file) and the converter. ONE definition on purpose: the converter's fidelity promise (`converted == live`) depends on both -writers encoding a Kinefly record identically. No ROS imports here — mcap_convert.py must stay -runnable without a ROS process. +writers encoding a record identically. No ROS imports here — mcap_convert.py must stay runnable +without a ROS process. + + * Kinefly KF/1 → the compact ["kf", …] row (kf_row) + * a FicTrac line → (tag, fields) (parse_fictrac_line) + * the FicTrac columns kept in the ArenaFrame record (docs/development/fictrac-columns-in-the-bag.md, + decided 2026-10-18): cols 5, 6-8, 22 native, 23 (fictrac_extras) + * an ArenaFrame record (typed fields or the hub message) → the behavior_v1 row + [ms, fc, idx, ft, x, y, hd], or None for a flagged (untracked) frame (arena_frame_to_row) """ from __future__ import annotations +import math + KF_TAG = "kf" +NAN = float("nan") + +# ArenaFrame.flags +FLAG_BAD_FRAME = 1 # bit 0: an FT_BADFR line — FicTrac could not track; no columns, no run-log row + +# The FicTrac columns (1-based) kept beyond the behavior row, and the ArenaFrame field each lands in. +FICTRAC_EXTRA_COLS = {"err": 5, "dr_lab_x": 6, "dr_lab_y": 7, "dr_lab_z": 8, "ft_ns": 22, "ft_seq": 23} +BEHAVIOR_ROW_COLS = ("ms", "fc", "idx", "ft", "x", "y", "hd") # == bridge.py BEHAVIOR_V1_COLS (never widened) + + +def parse_fictrac_line(raw: str) -> tuple[str, list[float]]: + """(tag, fields) the way bridge.py's Pipeline parses: a leading FT* token is the tag + ('' when absent, as in offline .dat lines and the simulator); fields = the floats after + it, or [] when the line is not numeric (a bad frame has no usable data).""" + parts = raw.strip().split(",") + tag = "" + if parts and parts[0].strip()[:2].upper() == "FT": + tag = parts[0].strip() + parts = parts[1:] + try: + return tag, [float(p) for p in parts] + except ValueError: + return tag, [] + + +def is_bad_frame_tag(tag: str) -> bool: + """FicTrac's socket output tags every record: "FT" = tracked, anything else starting with + FT (seen: "FT_BADFR") = a frame it could not track.""" + return tag[:2].upper() == "FT" and tag != "FT" + + +def fictrac_extras(fields: list[float] | None) -> dict: + """The kept FicTrac columns beyond the behavior row, as ArenaFrame fields. PURE. A short or + absent line gives NaN for the floats and 0 for the integers (the typed message cannot carry + null); `ft_ns` is col 22 truncated to an integer — on our rigs it already is one (ns).""" + f = fields or [] + + def fl(col: int) -> float: + return float(f[col - 1]) if len(f) >= col else NAN + + def it(col: int) -> int: + if len(f) < col: + return 0 + v = f[col - 1] + return int(v) if math.isfinite(v) else 0 + + return {"err": fl(5), "dr_lab_x": fl(6), "dr_lab_y": fl(7), "dr_lab_z": fl(8), "ft_seq": it(23), "ft_ns": it(22)} + + +def frame_flags(fr: dict) -> int: + v = fr.get("flags", 0) + return int(v) if isinstance(v, (int, float)) else 0 + + +def arena_frame_to_row(fr: dict) -> list | None: + """An /arena/frame record → the behavior_v1 row `[ms, fc, idx, ft, x, y, hd]` the live writer + emits (bridge.py write_frame), or None when the record is a flagged untracked frame (the + live writer wrote no row). Accepts the typed ArenaFrame fields (`has_ft`) or the hub message + the String path records (`ft` None when absent).""" + if frame_flags(fr) & FLAG_BAD_FRAME: + return None + if "has_ft" in fr: # typed ArenaFrame + ft = float(fr["ft"]) if fr["has_ft"] else None + return [int(fr["ms"]), int(fr["fc"]), int(fr["idx"]), ft, float(fr["x"]), float(fr["y"]), float(fr["hd"])] + return [fr.get(c) for c in BEHAVIOR_ROW_COLS] + + def kf_row(rel_ms: int, pkt: dict, seq: int) -> list: """KF/1 record → compact tagged row. PURE. Angles are the FIRST (strongest-edge) angle of each part when the part is valid, else null — Kinefly's bridge plan: the diff --git a/ros2-poc/nodes/bridge_node.py b/ros2-poc/nodes/bridge_node.py index a48b4800..a1d30bbc 100644 --- a/ros2-poc/nodes/bridge_node.py +++ b/ros2-poc/nodes/bridge_node.py @@ -92,7 +92,7 @@ from mcap_convert import convert_bag # noqa: E402 from mcap_index import needs_recovery, recover_bag # noqa: E402 from recorder import PROFILES, STRING_TYPE, Recorder # noqa: E402 -from _rows import KF_TAG, kf_row # noqa: E402 (one encoder for the node and the converter) +from _rows import FLAG_BAD_FRAME, KF_TAG, fictrac_extras, is_bad_frame_tag, kf_row, parse_fictrac_line # noqa: E402 (one encoder for the node and the converter) # Typed messages only when the package imports AND serializes (pixi's `typed` env, or a # colcon install whose libraries load); otherwise String JSON everywhere. @@ -118,7 +118,7 @@ def load_bridge_module(path: Path = BRIDGE_PY): # it in THIS process to the precise wall clock; bridge.py itself (bridge 3.4) is unchanged. B.now_ms = wall_ms -NODE_VERSION = f"{B.BRIDGE_VERSION} · ros2-poc plan-b (rmw_zenoh, mcap record, connectors)" +NODE_VERSION = f"{B.BRIDGE_VERSION} · ros2-poc plan-b (rmw_zenoh, mcap record, connectors, merged frame record)" FICTRAC_TOPIC = "/fictrac/raw" FICTRAC_FRAME_TOPIC = "/fictrac/frame" # typed arena_interfaces/FicTracFrame BROWSER_TOPIC = "/arena/browser" @@ -203,6 +203,7 @@ def __init__(self, name: str) -> None: self.last_rx_mono = None # bridge monotonic at last message self.lat_ms: list[float] = [] self.lost = False + self.bad = 0 # untracked (FT_BADFR) frames in the current run; FicTrac only def seen(self, rx_epoch_ms: int | None, rx_mono_ns: int | None = None) -> None: self.count += 1 @@ -225,6 +226,7 @@ def summary(self) -> dict: return { "src": self.name, "frames": self.count, + "bad_frames": self.bad, "lat_ms": { "p50": None if not self.lat_ms else round(percentile(self.lat_ms, 0.5), 2), "p99": None if not self.lat_ms else round(percentile(self.lat_ms, 0.99), 2), @@ -261,8 +263,12 @@ class RecordingLogWriter(B.LogWriter): on_close = None def __init__(self, path, level, log_dir, *, profile: str, typed: bool, export: str, connectors: list[str], - max_bag_bytes: int = 0) -> None: + max_bag_bytes: int = 0, record_fictrac_raw: bool = False) -> None: self.seq = 0 + # 2026-10-18 (docs/development/fictrac-columns-in-the-bag.md): the FicTrac record in the bag + # is the merged /arena/frame message; the adapter's topics (25 columns + the verbatim line, + # 68-77 % of a bag) are recorded only on request (--record-fictrac-raw, debugging a tracker). + self.record_fictrac_raw = bool(record_fictrac_raw) self._max_bag_bytes = int(max_bag_bytes or 0) self.rec: Recorder | None = None self.bag_name: str | None = None @@ -291,12 +297,12 @@ def topics(self) -> dict[str, str]: 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, } - if self._typed: - t[FRAME_TOPIC] = ARENA_FRAME_TYPE - t[FICTRAC_FRAME_TOPIC] = FICTRAC_FRAME_TYPE - else: - t[FRAME_TOPIC] = STRING_TYPE - t[FICTRAC_TOPIC] = STRING_TYPE + t[FRAME_TOPIC] = ARENA_FRAME_TYPE if self._typed else STRING_TYPE + if self.record_fictrac_raw: + if self._typed: + t[FICTRAC_FRAME_TOPIC] = FICTRAC_FRAME_TYPE + else: + t[FICTRAC_TOPIC] = STRING_TYPE return t # ── the LogWriter surface ──────────────────────────────────────────────────────── @@ -387,17 +393,38 @@ def record_device(self, topic: str, data: bytes, t_ns: int) -> None: if self._fh and self.rec is not None: self.rec.write_raw(topic, data, t_ns) - def record_frame(self, msg: dict, derived: tuple[float, float] = NO_DERIVED) -> None: + def record_frame(self, msg: dict, extras: dict | None = None, flags: int = 0) -> None: """Called by RosHub.publish — the frame's emission point (write_frame follows at once). - `derived` = (turning_dps, forward_mms) from the hub's FrameDeriver; typed record only.""" + `extras` = the kept FicTrac columns beyond the behavior row (_rows.fictrac_extras), in + both record forms — the bag's ONLY copy of them since the raw topics stopped being recorded. + The derived turning_dps / forward_mms are NOT recorded (NaN in the typed record): they are + live-only, exactly recomputable from x/y/hd/ft, and measured at 16 B per frame of + incompressible doubles — a fifth of the record (harness bag, 2026-10-18).""" if not self._fh or self.rec is None: return seq = self.next_seq() pub_ns = int(msg["t"]) * 1_000_000 if self._typed: - self.rec.write_msg(FRAME_TOPIC, arena_frame_msg(msg, seq, derived), pub_ns=pub_ns) + self.rec.write_msg(FRAME_TOPIC, arena_frame_msg(msg, seq, NO_DERIVED, extras, flags), pub_ns=pub_ns) else: - self.rec.write_json(FRAME_TOPIC, {"wseq": seq, **msg}, pub_ns=pub_ns) + rec = {"wseq": seq, **msg, **(extras or {})} + if flags: + rec["flags"] = int(flags) + self.rec.write_json(FRAME_TOPIC, rec, pub_ns=pub_ns) + + def record_bad_frame(self, tag: str, fields: list[float], now: int, t0: int, epoch: int) -> None: + """An untracked FicTrac frame (tag FT_BADFR): the Pipeline emits no row and no hub frame, + so without this the bag would show a gap it cannot tell from a dropped datagram. One + flagged /arena/frame record (flags bit 0; fc/ft_seq when the line carried them, idx -1, + the floats NaN / null) — the converter writes no line for it (_rows.arena_frame_to_row).""" + if not self._fh or self.rec is None: + return + fc = int(fields[0]) if fields and math.isfinite(fields[0]) else 0 + msg = {"t": now, "ms": now - t0, "fc": fc, "idx": -1, "ft": None, "x": None, "y": None, "hd": None, "epoch": int(epoch)} + extras = fictrac_extras(fields) + if not self._typed: # JSON must stay strict (no NaN tokens): null for the absent floats + extras = {k: (None if isinstance(v, float) and math.isnan(v) else v) for k, v in extras.items()} + self.record_frame(msg, extras, FLAG_BAD_FRAME) def close(self) -> None: if not self._fh: @@ -536,14 +563,22 @@ def step(self, msg: dict) -> tuple[float, float]: return (turning_dps, forward_mms) -def arena_frame_msg(msg: dict, seq: int, derived: tuple[float, float] = NO_DERIVED): +def arena_frame_msg(msg: dict, seq: int, derived: tuple[float, float] = NO_DERIVED, extras: dict | None = None, flags: int = 0): """The hub frame → arena_interfaces/ArenaFrame. ONE builder for the published message and - the recorded one, so the bag and the live topic can never disagree on a field.""" + the recorded one, so the bag and the live topic can never disagree on a field. `extras` = + _rows.fictrac_extras(fields) of the line the frame came from (cols 5, 6-8, 22 native, 23).""" + e = extras or fictrac_extras(None) + + def fl(v) -> float: + return float(v) if v is not None else NAN + return ArenaFrame(t_ms=int(msg["t"]), ms=int(msg["ms"]), fc=int(msg["fc"]), idx=int(msg["idx"]), - has_ft=msg.get("ft") is not None, ft=float(msg["ft"]) if msg.get("ft") is not None else NAN, - x=float(msg["x"]), y=float(msg["y"]), hd=float(msg["hd"]), epoch=int(msg.get("epoch", 0)), + has_ft=msg.get("ft") is not None, ft=fl(msg.get("ft")), + x=fl(msg.get("x")), y=fl(msg.get("y")), hd=fl(msg.get("hd")), epoch=int(msg.get("epoch", 0)), bias_deg=float(msg.get("bias", 0.0)), seq=seq, - turning_dps=float(derived[0]), forward_mms=float(derived[1])) + turning_dps=float(derived[0]), forward_mms=float(derived[1]), + err=fl(e["err"]), dr_lab_x=fl(e["dr_lab_x"]), dr_lab_y=fl(e["dr_lab_y"]), dr_lab_z=fl(e["dr_lab_z"]), + ft_seq=int(e["ft_seq"]), ft_ns=int(e["ft_ns"]), flags=int(flags) & 0xFF) class RosHub(B.Hub): @@ -558,17 +593,22 @@ def __init__(self, on_message, ros_pub, log: RecordingLogWriter, typed: bool, ba self._log = log self._typed = typed self.deriver = FrameDeriver(ball_mm) + # The parsed columns of the FicTrac line the consumer is feeding to Pipeline.handle_line + # right now (set just before the call; handle_line publishes synchronously inside it, and + # the consumer awaits one line at a time, so the slot is always this frame's). + self.fields: list[float] | None = None async def publish(self, msg: dict) -> None: - await super().publish(msg) - derived = self.deriver.step(msg) if self._typed else NO_DERIVED - self._log.record_frame(msg, derived) + await super().publish(msg) # the WebSocket hub message to the Studio: unchanged + derived = self.deriver.step(msg) if self._typed else NO_DERIVED # live topic only; the record carries NaN + extras = fictrac_extras(self.fields) + self._log.record_frame(msg, extras) if self._ros_pub is not None: if self._typed: - m = arena_frame_msg(msg, self._log.seq, derived) + m = arena_frame_msg(msg, self._log.seq, derived, extras) else: - m = String() - m.data = json.dumps(msg, separators=(",", ":")) + m = String() # the hub message + the kept FicTrac columns (what the bag records) + m.data = json.dumps({**msg, **extras}, separators=(",", ":")) self._ros_pub.publish(m) @@ -621,7 +661,8 @@ def _fictrac(self, data: bytes) -> None: t_ns = wall_ns() env = self._envelope(data) if env is None or not isinstance(env.get("raw"), str): - self._record(FICTRAC_TOPIC, data) + if self._log.record_fictrac_raw: + self._record(FICTRAC_TOPIC, data) return self.stats["fictrac"].seen(env.get("rx_epoch_ms"), env.get("rx_mono_ns")) # Same hand-off bridge.py's UDP protocol does (queue.put_nowait), from the ROS thread. @@ -632,7 +673,8 @@ def _fictrac_typed(self, data: bytes) -> None: try: m = deserialize_message(data, FicTracFrame) except Exception: - self._record(FICTRAC_FRAME_TOPIC, data) + if self._log.record_fictrac_raw: + self._record(FICTRAC_FRAME_TOPIC, data) return self.stats["fictrac"].seen(int(m.rx_epoch_ms) if m.rx_epoch_ms else None, int(m.rx_mono_ns) if m.rx_mono_ns else None) # The closed-loop math stays bridge 3.4's byte for byte: it parses the verbatim line @@ -671,7 +713,8 @@ async def run(args: argparse.Namespace) -> None: CONNECTORS = {c: t["cmd"] for c, t in CONNECTOR_TOPICS.items()} log = RecordingLogWriter(args.log, args.log_level, args.log_dir, profile=args.bag_profile, typed=typed, export=args.export, connectors=list(CONNECTORS), - max_bag_bytes=int(round(args.max_bag_mb * 1024 * 1024)) if args.max_bag_mb > 0 else 0) + max_bag_bytes=int(round(args.max_bag_mb * 1024 * 1024)) if args.max_bag_mb > 0 else 0, + record_fictrac_raw=args.record_fictrac_raw) queue: asyncio.Queue[tuple[str, str, bytes, int]] = asyncio.Queue() # (line, topic, raw bytes, receive ns) pipeline = B.Pipeline( None, @@ -827,6 +870,7 @@ def on_log_open() -> None: for s in node.stats.values(): s.lost = False s.last_rx_mono = None + s.bad = 0 log.write_event( { "type": "connector_status", @@ -835,7 +879,7 @@ def on_log_open() -> None: "src": "bridge", "node": NODE_VERSION, "rmw": os.environ.get("RMW_IMPLEMENTATION"), - "record": {"bag": log.bag_name, "profile": log._profile, "typed": typed}, + "record": {"bag": log.bag_name, "profile": log._profile, "typed": typed, "fictrac_raw": log.record_fictrac_raw}, "topics": {"fictrac": FICTRAC_FRAME_TOPIC if typed else FICTRAC_TOPIC, "kinefly": KINEFLY_TOPIC, "bias": [BIAS_CMD_TOPIC, BIAS_REPLY_TOPIC, BIAS_STATUS_TOPIC, BIAS_FRAMES_TOPIC], "scanimage": [SCANIMAGE_CMD_TOPIC, SCANIMAGE_REPLY_TOPIC, SCANIMAGE_STATUS_TOPIC], "frame": FRAME_TOPIC, @@ -901,14 +945,27 @@ async def watchdog() -> None: ros_thread.start() async def consume_recording() -> None: - """bridge.py's consume() (line → pipeline.handle_line), then the FicTrac message is - recorded under the run's open gate (RecordingLogWriter.record_device) — after the - frame went out, so the bag write is not on the closed-loop path; handle_line never + """bridge.py's consume() (line → pipeline.handle_line). The line is parsed ONCE more here + (the same split bridge.py does) so the frame's record can carry the kept FicTrac columns + beyond the behavior row (hub.fields → RosHub.publish → ArenaFrame err/dr_lab/ft_seq/ft_ns); + an untracked frame (FT_BADFR) gets a flagged record of its own. With --record-fictrac-raw + the adapter's message is also recorded under the run's open gate (record_device) — after + the frame went out, so the bag write is not on the closed-loop path; handle_line never opens or closes the log, so the gate is the frame's.""" + fictrac_stats = node.stats["fictrac"] while True: line, topic, data, t_ns = await queue.get() - await pipeline.handle_line(line) - log.record_device(topic, data, t_ns) + tag, fields = parse_fictrac_line(line) + hub.fields = fields + try: + await pipeline.handle_line(line) + finally: + hub.fields = None + if is_bad_frame_tag(tag): + fictrac_stats.bad += 1 + log.record_bad_frame(tag, fields, B.now_ms(), pipeline.t0_ms, pipeline.epoch_id) + if log.record_fictrac_raw: + log.record_device(topic, data, t_ns) consumer = asyncio.create_task(consume_recording()) wd = asyncio.create_task(watchdog()) @@ -966,6 +1023,10 @@ def main(argv: list[str] | None = None) -> int: help="MCAP writer profile: crashsafe (no chunking, no cache; default — rosbag2's defaults hold a whole short run in an unflushed 768 KiB chunk) or default (rosbag2 defaults)") p.add_argument("--max-bag-mb", type=float, default=0.0, help="OPTIONAL: split the bag into _0.mcap, _1.mcap, … at about this many MiB (rosbag2 max_bagfile_size; a threshold, not a hard cap). Default 0 = one file; bags are archived on a share, not pushed to GitHub, so only a size-limited store needs this") + p.add_argument("--record-fictrac-raw", action="store_true", + help="ALSO record the adapter's FicTrac topic (/fictrac/frame or /fictrac/raw: all 25 columns + the verbatim line) in the bag. " + "Off by default since 2026-10-18 — the merged /arena/frame record carries the 10 kept columns (docs/development/fictrac-columns-in-the-bag.md); " + "`--level full` needs this") p.add_argument("--export", choices=("converted", "live"), default="converted", help="what log_export returns: the NDJSON converted from the bag (default) or the live shadow file") args, ros_args = p.parse_known_args(argv) diff --git a/ros2-poc/nodes/fictrac_udp_adapter.py b/ros2-poc/nodes/fictrac_udp_adapter.py index 2b4f0065..37f7f4dc 100644 --- a/ros2-poc/nodes/fictrac_udp_adapter.py +++ b/ros2-poc/nodes/fictrac_udp_adapter.py @@ -43,6 +43,7 @@ from std_msgs.msg import String from _common import RELIABLE_QOS, mono_ns, probe_interfaces, raise_timer_resolution, spin, wall_ms, wrap_line +from _rows import parse_fictrac_line # ONE parser for the adapter, the bridge node and the converter _IFACES = probe_interfaces() # import + serialize; None → String only HAVE_IFACES = _IFACES is not None @@ -53,22 +54,6 @@ RAW_QOS = RELIABLE_QOS -def parse_fictrac_line(raw: str) -> tuple[str, list[float]]: - """(tag, fields) the way bridge.py's Pipeline parses: a leading FT* token is the tag - ('' when absent, as in offline .dat lines and the simulator); fields = the floats after - it, or [] when the line is not numeric (a bad frame has no usable data).""" - parts = raw.strip().split(",") - tag = "" - if parts and parts[0].strip()[:2].upper() == "FT": - tag = parts[0].strip() - parts = parts[1:] - try: - return tag, [float(p) for p in parts] - except ValueError: - return tag, [] - - - class FicTracUdpAdapter(Node): def __init__(self, host: str, port: int, status_every_s: float, typed: str = "auto") -> None: super().__init__("fictrac_udp_adapter") diff --git a/ros2-poc/nodes/mcap_convert.py b/ros2-poc/nodes/mcap_convert.py index fc3e5899..1a51ba68 100644 --- a/ros2-poc/nodes/mcap_convert.py +++ b/ros2-poc/nodes/mcap_convert.py @@ -20,13 +20,17 @@ `rows` messages expand to their rows; `connector_command` → its connector event; `log_export` → nothing (as today). /arena/frame the closed-loop evaluation (String JSON or arena_interfaces/ArenaFrame), - with the node's OWN sample time `t`/`ms` → [ms, fc, idx, ft, x, y, hd] + with the node's OWN sample time `t`/`ms` AND, since 2026-10-18, the kept FicTrac + columns beyond the behavior row (err, dr_lab_x/y/z, ft_seq, ft_ns — docs/development/ + fictrac-columns-in-the-bag.md); a flagged untracked frame (flags bit 0) → no line + → [ms, fc, idx, ft, x, y, hd] /arena/event {"seq", …a bridge-authored event…}: bias_config, heading_tare, connector_status, source_lost/resumed, warn → verbatim /kinefly/raw the adapter envelope (DERIVED stream) → ["kf", ms, seq, t_src, L, R, head, abd] /bias/reply, /bias/status the BIAS node's JSON (DERIVED) → {"type":"connector","src":"bias",…} /scanimage/reply, /scanimage/status the ScanImage node's JSON (DERIVED) → {"type":"connector","src":"scanimage",…} - /fictrac/raw | /fictrac/frame the adapter's FicTrac records → used only for level `full` + /fictrac/raw | /fictrac/frame the adapter's FicTrac records (only when the bridge node ran with + --record-fictrac-raw; bags before 2026-10-18 always have them) → used only for level `full` Clock and order rules (comparison.md §4, B1/B2): * ONE ms base: t0 = the session-open record's `ms`, set when the run's log opened. Frame @@ -42,7 +46,8 @@ to do; the harness measures how far it lands from the live writer's order. Levels are a converter option (the bag holds everything): behavior_v2 compacts arena echoes to ["a",…] and carries t0 in the schema line; behavior_v1 keeps the echo objects; full keeps -the 25-column FicTrac records (joined from /fictrac/* by frame counter) and writes no schema. +the 25-column FicTrac records (joined from /fictrac/* by frame counter) and writes no schema — it +is refused for a bag recorded without the raw FicTrac topics (the default since 2026-10-18). """ from __future__ import annotations @@ -58,7 +63,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) -from _rows import KF_TAG, kf_row # noqa: E402 (shared with bridge_node.py) +from _rows import KF_TAG, arena_frame_to_row, kf_row # noqa: E402 (shared with bridge_node.py) HERE = Path(__file__).resolve().parent REPO = HERE.parents[1] @@ -256,19 +261,19 @@ def _json_of(rec: dict): def frame_line(fr: dict, level: str, fields_by_fc: dict | None): - """A /arena/frame record (hub message JSON or ArenaFrame fields) → the logged frame.""" - if "idx" in fr and "has_ft" in fr: # typed ArenaFrame - beh = {"ms": int(fr["ms"]), "fc": int(fr["fc"]), "idx": int(fr["idx"]), - "ft": float(fr["ft"]) if fr["has_ft"] else None, - "x": float(fr["x"]), "y": float(fr["y"]), "hd": float(fr["hd"])} - else: # the hub message, as published - beh = {c: fr.get(c) for c in B.BEHAVIOR_V1_COLS} + """A /arena/frame record (hub message JSON or ArenaFrame fields) → the logged frame, or None + for a flagged untracked frame (the live writer wrote no row). The column mapping is + _rows.arena_frame_to_row, shared with the node.""" + row = arena_frame_to_row(fr) + if row is None: + return None if level == "full": - fields = (fields_by_fc or {}).get(int(beh["fc"])) - rec = {"type": "fictrac_frame", "seq": beh["fc"], "index": beh["idx"], "t": beh["ms"]} + ms, fc, idx = row[0], row[1], row[2] + fields = (fields_by_fc or {}).get(int(fc)) + rec = {"type": "fictrac_frame", "seq": fc, "index": idx, "t": ms} rec["fictrac"] = fields if fields is not None else [] return rec - return [beh[c] for c in B.BEHAVIOR_V1_COLS] + return row def fictrac_fields(rec: dict) -> tuple[int, list[float]] | None: @@ -329,6 +334,13 @@ def add(seq: int, own_ms: int, rank: str, obj) -> None: ff = fictrac_fields(r) if ff: fields_by_fc[ff[0]] = ff[1] + if not fields_by_fc: + # The 25 columns exist only in the adapter's topics, which the bridge node records only + # with --record-fictrac-raw (off since 2026-10-18); the merged /arena/frame record keeps + # 10 of them. The readers of `full` (runlog-replay.js, closed-loop-report.py) use cols + # 1/15-17/22, which behavior_v2 carries — derive that instead. + raise ValueError("level 'full' needs the raw FicTrac topics (/fictrac/frame or /fictrac/raw), which this bag " + "does not hold: the bridge node records them only with --record-fictrac-raw. Use --level behavior_v2.") for r in records: t = r["topic"] @@ -413,7 +425,11 @@ def add(seq: int, own_ms: int, rank: str, obj) -> None: if seq is None: stats["skipped"] += 1 continue - add(seq, int(fr["ms"]), "frame", frame_line(fr, level, fields_by_fc)) + line = frame_line(fr, level, fields_by_fc) + if line is None: # a flagged untracked frame: recorded, never a line + stats["bad_frames"] = stats.get("bad_frames", 0) + 1 + continue + add(seq, int(fr["ms"]), "frame", line) stats["frames"] += 1 elif t == TOPIC_EVENT: o = _json_of(r) diff --git a/ros2-poc/nodes/mcap_index.py b/ros2-poc/nodes/mcap_index.py index ddb2d062..dd0f8771 100644 --- a/ros2-poc/nodes/mcap_index.py +++ b/ros2-poc/nodes/mcap_index.py @@ -66,13 +66,41 @@ def _scan(src: Path, recover: bool): return torn, error +ZSTD_LEVEL = 19 # the final layout's compression level (docs/development/fictrac-columns-in-the-bag.md §3) + + +class _zstd_level: + """The `mcap` writer calls `zstandard.compress(data)` — level 3, no knob. Swap the module + function for the duration of one rewrite (the same trick as sizing/compare.py); restored in + __exit__ even on an exception. Process-wide, so never nest two rewrites concurrently.""" + + def __init__(self, level: int) -> None: + self.level = level + + def __enter__(self): + import zstandard + + self._mod = zstandard + self._saved = zstandard.compress + lvl = self.level + zstandard.compress = lambda data, _l=lvl: zstandard.ZstdCompressor(level=_l).compress(data) + return self + + def __exit__(self, *exc) -> None: + self._mod.compress = self._saved + + def reindex(mcap_path: str | os.PathLike, keep_original: bool = False, compression: str = "zstd", - recover: bool = False, marker: dict | None = None) -> dict: + recover: bool = False, marker: dict | None = None, level: int = ZSTD_LEVEL) -> dict: """Rewrite one MCAP indexed + compressed. `recover=True` tolerates a torn tail (the records read up to it are kept), appends the `recovered` session marker (`marker` adds fields) and - keeps the original as `.torn.bak`. Returns a report; `verified` False = nothing replaced.""" + keeps the original as `.torn.bak`. Returns a report; `verified` False = nothing replaced. + + The final layout (2026-10-18): a CHUNK index only — no per-message index, which costs more + than the messages of a 50 Hz stream compress to — and zstd at `level` (19; the mcap writer's + own 3 is a streaming default). Readers seek by chunk; Foxglove and `ros2 bag` are unaffected.""" from mcap.records import Attachment, Channel, Header, Message, Metadata, Schema - from mcap.writer import CompressionType, Writer + from mcap.writer import CompressionType, IndexType, Writer src = Path(mcap_path) tmp = src.with_suffix(".indexed.tmp") @@ -85,8 +113,8 @@ def reindex(mcap_path: str | os.PathLike, keep_original: bool = False, compressi n = 0 torn, error = False, None marker_written = False - with open(tmp, "wb") as fout: - w = Writer(fout, compression=comp) + with open(tmp, "wb") as fout, _zstd_level(level if compression == "zstd" else 3): + w = Writer(fout, compression=comp, index_types=IndexType.ALL & ~IndexType.MESSAGE) started = False gen = _scan(src, recover) while True: @@ -152,7 +180,8 @@ def reindex(mcap_path: str | os.PathLike, keep_original: bool = False, compressi got[chn.topic] = got.get(chn.topic, 0) + 1 ok = got == want and summ is not None and bool(summ.chunk_indexes) res = {"file": str(src), "messages": n, "bytes_before": src.stat().st_size, "bytes_after": tmp.stat().st_size, - "chunks": len(summ.chunk_indexes) if summ else 0, "compression": compression, "verified": ok, + "chunks": len(summ.chunk_indexes) if summ else 0, "compression": compression, + "level": level if compression == "zstd" else None, "message_index": False, "verified": ok, "seconds": round(time.perf_counter() - t0, 2)} if recover: res.update({"recovered": True, "torn": torn, "torn_error": error, "marker": marker_written}) diff --git a/ros2-poc/nodes/recorder.py b/ros2-poc/nodes/recorder.py index bfbcbd92..07e2698e 100644 --- a/ros2-poc/nodes/recorder.py +++ b/ros2-poc/nodes/recorder.py @@ -42,7 +42,12 @@ PROFILES = { # name: (storage_preset_profile, storage_config (dict → YAML), max_cache_size bytes) - "default": ("", None, 100 * 1024 * 1024), + # `default` = rosbag2's chunked writer (768 KiB chunks + a 100 MiB message cache, flushed when + # full or at close) with the final layout's options (2026-10-18, docs/development/ + # fictrac-columns-in-the-bag.md §4): zstd at the plugin's slowest level, no per-message index. + # `crashsafe` writes every record at once, uncompressed; Recorder.close() rewrites it into + # the same final layout (mcap_index.reindex) once the run has closed cleanly. + "default": ("", {"compression": "Zstd", "compressionLevel": "Slowest", "noMessageIndex": True}, 100 * 1024 * 1024), "crashsafe": ("", {"noChunking": True, "noChunkCRC": True, "compression": "None"}, 0), } @@ -57,8 +62,7 @@ def __init__(self, bag_dir: str | os.PathLike, topics: dict[str, str], profile: max_bag_bytes: int = 0) -> None: """`topics`: topic name → ROS type name (e.g. 'std_msgs/msg/String'). `max_bag_bytes` > 0 asks rosbag2 to split the bag into `_0.mcap`, `_1.mcap`, … at about that size (a - threshold rosbag2 checks after each write, not a hard cap; off by default — bags are - archived on a share, not pushed to GitHub, so size limits are the exception).""" + threshold rosbag2 checks after each write, not a hard cap; off by default, plan §3).""" self.bag_dir = Path(bag_dir) self.topics = dict(topics) self.profile = profile diff --git a/ros2-poc/pixi.toml b/ros2-poc/pixi.toml index fbec7aeb..bef58f6c 100644 --- a/ros2-poc/pixi.toml +++ b/ros2-poc/pixi.toml @@ -153,6 +153,9 @@ export = "python nodes/export_log.py" # Rewrite a finished bag indexed + zstd-compressed (the bridge does this itself when a run # closes cleanly; use it on a bag recorded before that, or after `--recover`). reindex = "python nodes/mcap_index.py" +# Bag-size study of the FicTrac record (docs/development/fictrac-columns-in-the-bag.md): a bag vs its +# jsonl.gz under the final layout and the alternatives that were measured and declined. +size-columns = "python sizing/columns.py" # Build interfaces/arena_interfaces with colcon inside the `typed` env (the alternative to # the pixi-build route); timed and recorded in README.md → "Plan B spike". build-interfaces = "python nodes/build_interfaces.py" diff --git a/ros2-poc/sizing/columns.py b/ros2-poc/sizing/columns.py new file mode 100644 index 00000000..6dafb65c --- /dev/null +++ b/ros2-poc/sizing/columns.py @@ -0,0 +1,275 @@ +"""columns.py — size a bridge-node bag vs its jsonl.gz under the FicTrac-record layouts. + + pixi run size-columns -- runlogs/.bag [runlogs/.jsonl] [--all] + +The study behind docs/development/fictrac-columns-in-the-bag.md (decided 2026-10-18): ONE record per +FicTrac frame on /arena/frame carrying 10 of the 25 FicTrac columns, float64, one MCAP message per +frame, chunk-only index, zstd 19 — measured on the two 2026-10-14 bench bags at 2.24× / 2.48× the +jsonl.gz. Every variant is written by the pure-Python `mcap` writer (the same one mcap_index.reindex +uses for the final layout), so the numbers are what a cleanly closed run ends up with. + +Without --all: the reference jsonl.gz, the bag as it is on disk, and the ADOPTED layout rebuilt from +the bag's records. With --all: the alternatives that were measured and declined (two-topic layouts, +column subsets, 1 s batching, int32 fixed-point), as in the doc's §3 table. A bag recorded before +2026-10-18 holds the raw FicTrac topics (/fictrac/frame or /fictrac/raw), so every variant can be +rebuilt from it; a bag recorded after holds the kept columns in /arena/frame only, and the +25-column rows are reported as "n/a". + +Pure Python (mcap, zstandard); no ROS process needed. +""" + +from __future__ import annotations + +import argparse +import collections +import gzip +import io +import json +import math +import os +import struct +import sys +from pathlib import Path + +import zstandard +from mcap.reader import make_reader +from mcap.writer import CompressionType, IndexType, Writer + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "nodes")) +from mcap_convert import decode_cdr # noqa: E402 (the converter's CDR decoder; reads the schema text in the bag) +from mcap_index import ZSTD_LEVEL, _zstd_level # noqa: E402 + +TOPIC_FRAME = "/arena/frame" +FICTRAC_TOPICS = ("/fictrac/frame", "/fictrac/raw") + +# 1-based FicTrac columns (doc §1) +BEHAVIOR_5 = [1, 15, 16, 17, 22] +KEPT_10 = [1, 5, 6, 7, 8, 15, 16, 17, 22, 23] +ALL_25 = list(range(1, 26)) +EXTRA_COLS = [5, 6, 7, 8, 23] # kept beyond the behavior row (col 22 native rides as ft_ns) +INT_COLS = {1, 22, 23} # packed as int64 (col 22 = ns on our rigs) + +ARENA_FLOATS = ("ft", "x", "y", "hd", "bias_deg", "turning_dps", "forward_mms", "err", "dr_lab_x", "dr_lab_y", "dr_lab_z") + + +def load(path: Path) -> list[tuple]: + """(topic, schema name, log_ns, pub_ns, data, decoded) per record, file order.""" + files = sorted(path.glob("*.mcap")) if path.is_dir() else [path] + recs = [] + for f in files: + with open(f, "rb") as fh: + for schema, ch, m in make_reader(fh).iter_messages(log_time_order=False): + sn = schema.name if schema else None + st = schema.data.decode() if schema and schema.data else None + recs.append((ch.topic, sn, m.log_time, m.publish_time, m.data, decode_cdr(sn, st, m.data))) + return recs + + +def fictrac_fields(sn: str | None, msg: dict) -> list[float] | None: + if sn == "arena_interfaces/msg/FicTracFrame": + return list(msg["fields"]) + try: + env = json.loads(msg["data"]) + parts = env["raw"].strip().split(",") + except (KeyError, TypeError, json.JSONDecodeError): + return None + if parts and parts[0].strip()[:2].upper() == "FT": + parts = parts[1:] + try: + return [float(p) for p in parts] + except ValueError: + return None + + +def arena_fields(sn: str | None, msg: dict) -> dict: + """The /arena/frame record as a flat dict, from the typed message or the String-path JSON.""" + if sn == "arena_interfaces/msg/ArenaFrame": + return dict(msg) + o = json.loads(msg["data"]) + return {"t_ms": o["t"], "ms": o["ms"], "fc": o["fc"], "idx": o["idx"], "has_ft": o.get("ft") is not None, + "ft": o.get("ft") if o.get("ft") is not None else math.nan, "x": o.get("x"), "y": o.get("y"), "hd": o.get("hd"), + "epoch": o.get("epoch", 0), "bias_deg": o.get("bias", 0.0) or 0.0, "seq": o.get("wseq", 0), + "turning_dps": math.nan, "forward_mms": math.nan, + "err": o.get("err", math.nan), "dr_lab_x": o.get("dr_lab_x", math.nan), "dr_lab_y": o.get("dr_lab_y", math.nan), + "dr_lab_z": o.get("dr_lab_z", math.nan), "ft_seq": o.get("ft_seq", 0), "ft_ns": o.get("ft_ns", 0), "flags": o.get("flags", 0)} + + +def _f(v) -> float: + return float(v) if v is not None else math.nan + + +def pack_num(v, enc: str) -> bytes: + if enc == "f64": + return struct.pack(" bytes: + out = b"" + for c in cols: + v = fields[c - 1] if len(fields) >= c else math.nan + out += struct.pack(" bytes: + """≈ the typed ArenaFrame's CDR cost: ints as declared, floats per `enc`; `extras` adds the + kept FicTrac columns (the adopted record); `dedup` drops x/y/hd/ft (the join-by-fc variant); + `derived` keeps the live turning_dps/forward_mms values — the adopted record stores NaN there + (a constant, free), which is what the node records since 2026-10-18.""" + out = struct.pack(" int | None: + """Rewrite the bag's records in one layout; returns the MCAP's size in bytes. + layout: 'merged' (the adopted record: /arena/frame + extras, no FicTrac topic) · 'split' (a FicTrac + topic with `cols` beside today's ArenaFrame) · 'dedup' (split, ArenaFrame without x/y/hd/ft) · + 'floor' (ArenaFrame alone, no extras). None when the bag lacks what the variant needs.""" + ft_by_fc: dict[int, list[float]] = {} + have_ft_topic = any(r[0] in FICTRAC_TOPICS for r in recs) + if layout in ("split", "dedup") and not have_ft_topic: + return None + if have_ft_topic: + for topic, sn, lt, pt, data, msg in recs: + if topic in FICTRAC_TOPICS: + f = fictrac_fields(sn, msg) + if f and len(f) >= 1: + ft_by_fc[int(f[0])] = f + buf = io.BytesIO() + with _zstd_level(level): + w = Writer(buf, compression=CompressionType.ZSTD, index_types=IndexType.ALL & ~IndexType.MESSAGE) + w.start(profile="ros2", library="columns.py") + chans: dict[str, int] = {} + + def chan(name: str) -> int: + if name not in chans: + chans[name] = w.register_channel(name, "cdr", w.register_schema(name, "ros2msg", b"")) + return chans[name] + + seq = 0 + buckets: dict[tuple, list] = collections.OrderedDict() + + def emit(topic: str, lt: int, pt: int, payload: bytes, batch: bool) -> None: + nonlocal seq + if batch: + buckets.setdefault((topic, lt // 1_000_000_000), []).append((lt, payload)) + else: + seq += 1 + w.add_message(chan(topic), log_time=lt, publish_time=pt, data=payload, sequence=seq) + + for topic, sn, lt, pt, data, msg in recs: + if topic in FICTRAC_TOPICS: + if layout in ("split", "dedup") and cols: + f = fictrac_fields(sn, msg) + if f is not None: + emit("/fictrac/frame", lt, pt, pack_fictrac(f, cols, enc), batch_ft) + continue # merged / floor: the raw topic is not recorded + if topic == TOPIC_FRAME: + a = arena_fields(sn, msg) + if layout == "merged": + # The extras come from the record itself (a bag after 2026-10-18) or, for an old + # bag, from the raw topic joined by fc — what the node now writes directly. + if math.isnan(_f(a.get("err"))) and int(a["fc"]) in ft_by_fc: + f = ft_by_fc[int(a["fc"])] + a.update({"err": f[4] if len(f) > 4 else math.nan, "dr_lab_x": f[5] if len(f) > 5 else math.nan, + "dr_lab_y": f[6] if len(f) > 6 else math.nan, "dr_lab_z": f[7] if len(f) > 7 else math.nan, + "ft_seq": int(f[22]) if len(f) > 22 else 0, "ft_ns": int(f[21]) if len(f) > 21 else 0}) + payload = pack_arena(a, enc, extras=True) + elif layout == "dedup": + payload = pack_arena(a, enc, extras=False, dedup=True) + else: + payload = pack_arena(a, enc, extras=False, derived=True) # the pre-2026-10-18 record as it was + emit(topic, lt, pt, payload, batch_arena) + else: + emit(topic, lt, pt, data, False) + for (topic, _b), items in sorted(buckets.items(), key=lambda kv: kv[0][1]): + body = struct.pack(" dict: + recs = load(bag) + n_frames = sum(1 for r in recs if r[0] == TOPIC_FRAME) + on_disk = sum(f.stat().st_size for f in (bag.glob("*.mcap") if bag.is_dir() else [bag])) + rows: list[tuple[str, int | None]] = [] + gz = None + if jsonl is None: + cand = bag.parent / (bag.name[:-4] + ".jsonl") if bag.name.endswith(".bag") else None + if cand and cand.exists(): + jsonl = cand + if jsonl and jsonl.exists(): + raw = jsonl.read_bytes() + gz = len(raw) if jsonl.suffix == ".gz" else len(gzip.compress(raw, 9)) + rows.append(("jsonl.gz (reference)", gz)) + rows.append(("bag as on disk", on_disk)) + rows.append(("ADOPTED: one record per frame, kept 10 cols, float64, one message per frame, chunk-only index, zstd 19", + write_variant(recs, **ADOPTED))) + if all_variants: + for label, kw in ALTERNATIVES: + rows.append((label, write_variant(recs, **kw))) + return {"bag": str(bag), "frames": n_frames, "jsonl_gz": gz, "rows": rows} + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("bag", help="a rosbag2 bag directory or a single .mcap file") + ap.add_argument("jsonl", nargs="?", help="the run's exported .jsonl or .jsonl.gz (default: .jsonl beside the bag)") + ap.add_argument("--all", action="store_true", help="also the declined alternatives (doc §3 table)") + ap.add_argument("--json", action="store_true", help="machine-readable output") + a = ap.parse_args(argv) + res = study(Path(a.bag), Path(a.jsonl) if a.jsonl else None, a.all) + if a.json: + print(json.dumps(res)) + return 0 + gz = res["jsonl_gz"] + print(f"{Path(res['bag']).name}: {res['frames']} frames" + (f"; jsonl.gz {gz} B" if gz else "; no jsonl beside the bag (ratios omitted)")) + print(f"{'variant':100s} {'bytes':>9s} {'x jsonl.gz':>10s} {'B/frame':>8s}") + for label, n in res["rows"]: + if n is None: + print(f"{label:100s} {'n/a':>9s} {'':>10s} {'':>8s} (needs the raw FicTrac topics)") + continue + ratio = f"{n / gz:9.2f}x" if gz else "" + per = f"{n / res['frames']:8.1f}" if res["frames"] else "" + print(f"{label:100s} {n:9d} {ratio:>10s} {per:>8s}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ros2-poc/tests/test_ros2_poc.py b/ros2-poc/tests/test_ros2_poc.py index 263105bf..c646e9c0 100644 --- a/ros2-poc/tests/test_ros2_poc.py +++ b/ros2-poc/tests/test_ros2_poc.py @@ -36,7 +36,12 @@ order; DERIVED lines (Kinefly rows, BIAS events) same content, |Δms| within tolerance log levels as a converter option bag → behavior_v1 == bridge.py --convert of the - v2; full level carries the 25 FicTrac columns + v2; full level refused without the raw FicTrac + topics (2026-10-18), still served by an old bag + the merged FicTrac record (2026-10-18) /arena/frame carries err, dr_lab_x/y/z, ft_seq, ft_ns + from the SAME line as x/y/hd; no /fictrac/* topic + recorded by default; an FT_BADFR frame = a flagged + record and no row; frame records ≤ 72 B/frame compressed ordering: `seq` (live order) vs `time` same lines; count of same-ms swaps; identical closed-loop report either way typed topics when arena_interfaces is built ArenaFrame / FicTracFrame schemas in the bag @@ -247,7 +252,8 @@ def typed_frame_values(bag: Path) -> dict: from mcap_convert import read_all # noqa: E402 recs, _ = read_all(bag) - frames = [r["msg"] for r in recs if r["topic"] == "/arena/frame" and r.get("msg") and "has_ft" in r["msg"]] + frames = [r["msg"] for r in recs if r["topic"] == "/arena/frame" and r.get("msg") and "has_ft" in r["msg"] + and not (int(r["msg"].get("flags") or 0) & 1)] # flagged untracked frames carry NaN by design frames.sort(key=lambda m: m["seq"]) out = {"n": len(frames), "decode_errors": sum(1 for r in recs if r["topic"] == "/arena/frame" and r.get("decode_error")), "has_derived": bool(frames) and all(k in frames[0] for k in ("turning_dps", "forward_mms"))} @@ -256,17 +262,102 @@ def typed_frame_values(bag: Path) -> dict: fin = lambda v: isinstance(v, float) and math.isfinite(v) # noqa: E731 out["core_finite"] = all(fin(m["x"]) and fin(m["y"]) and fin(m["hd"]) and (fin(m["ft"]) or not m["has_ft"]) for m in frames) out["fc_monotonic"] = all(b["fc"] > a["fc"] for a, b in zip(frames, frames[1:])) - out["first_nan"] = math.isnan(frames[0]["turning_dps"]) and math.isnan(frames[0]["forward_mms"]) - rest = frames[1:] - good = [m for m in rest if fin(m["turning_dps"]) and fin(m["forward_mms"])] - out["finite_frac"] = round(len(good) / len(rest), 3) if rest else None - out["turning_max"] = round(max((abs(m["turning_dps"]) for m in good), default=0.0), 1) - out["forward_max"] = round(max((abs(m["forward_mms"]) for m in good), default=0.0), 2) - out["in_range"] = out["turning_max"] < 5000 and out["forward_max"] < 200 # the model fly, not a real one - out["ok"] = out["core_finite"] and out["fc_monotonic"] and out["first_nan"] and (out["finite_frac"] or 0) >= 0.95 and out["in_range"] + # Live-only since 2026-10-18: the recorded copy carries NaN (16 B/frame of incompressible doubles, + # exactly recomputable); the derivation itself is checked on FrameDeriver directly (below). + out["derived_nan_in_bag"] = all(math.isnan(m["turning_dps"]) and math.isnan(m["forward_mms"]) for m in frames) + out["ok"] = out["core_finite"] and out["fc_monotonic"] and out["derived_nan_in_bag"] return out +def frame_deriver_values() -> dict: + """The live derived channels' math, on the class itself (bridge_node.FrameDeriver): two frames + 20 ms apart, 0.02 rad along the heading and 0.01 rad of turn → js/kinematics.js's numbers.""" + sys.path.insert(0, str(NODES)) + from bridge_node import FrameDeriver # noqa: E402 (imports rclpy; no node is created) + + d = FrameDeriver(9.0) + first = d.step({"fc": 1, "ft": 0.0, "t": 0, "x": 0.0, "y": 0.0, "hd": 0.0}) + turning, forward = d.step({"fc": 2, "ft": 20.0, "t": 20, "x": 0.02, "y": 0.0, "hd": 0.01}) + want_t = math.degrees(0.01) / 0.02 + want_f = (0.02 * math.cos(0.005)) / 0.02 * 4.5 + return {"first_nan": all(math.isnan(v) for v in first), "turning_dps": round(turning, 4), "forward_mms": round(forward, 4), + "ok": all(math.isnan(v) for v in first) and abs(turning - want_t) < 1e-6 and abs(forward - want_f) < 1e-6} + + +def frame_extras_values(bag: Path) -> dict: + """The kept FicTrac columns on the /arena/frame records (typed or String path), judged by VALUE + against the behavior columns of the SAME record: fictrac_sim writes col 8 = its heading step + and col 17 = the accumulated heading, so wrap(Δhd) == dr_lab_z frame to frame proves the extras + come from the line that produced x/y/hd (an off-by-one join would fail it); col 23 = the frame + number (ft_seq == fc); col 5 finite (the sim writes 0). Flagged records are the untracked frames.""" + sys.path.insert(0, str(NODES)) + from mcap_convert import _json_of, read_all # noqa: E402 + + recs, _ = read_all(bag) + frames, bad = [], [] + for r in recs: + if r["topic"] != "/arena/frame" or not r.get("msg"): + continue + fr = r["msg"] if "has_ft" in r["msg"] else _json_of(r) + if not isinstance(fr, dict): + continue + (bad if int(fr.get("flags") or 0) & 1 else frames).append(fr) + frames.sort(key=lambda f: int(f.get("seq", f.get("wseq", 0)))) + out = {"n": len(frames), "bad": len(bad), "bad_fc": [int(b.get("fc") or 0) for b in bad][:5]} + if not frames: + return out + fin = lambda v: isinstance(v, (int, float)) and math.isfinite(v) # noqa: E731 + out["extras_present"] = all(k in frames[0] for k in ("err", "dr_lab_x", "dr_lab_y", "dr_lab_z", "ft_seq", "ft_ns")) + if not out["extras_present"]: + return out + out["err_finite"] = all(fin(f["err"]) for f in frames) + out["dr_finite"] = all(fin(f["dr_lab_x"]) and fin(f["dr_lab_y"]) and fin(f["dr_lab_z"]) for f in frames) + out["ft_seq_eq_fc"] = all(int(f["ft_seq"]) == int(f["fc"]) for f in frames) + out["ft_ns_monotonic"] = all(int(b["ft_ns"]) > int(a["ft_ns"]) for a, b in zip(frames, frames[1:]) if int(b["fc"]) > int(a["fc"])) + + def wrap(r: float) -> float: + return (r + math.pi) % (2 * math.pi) - math.pi + + pairs = [(a, b) for a, b in zip(frames, frames[1:]) if int(b["fc"]) == int(a["fc"]) + 1] + aligned = sum(1 for a, b in pairs if abs(wrap(float(b["hd"]) - float(a["hd"])) - float(b["dr_lab_z"])) < 5e-5) # hd is 5-dp rounded + out["pairs"] = len(pairs) + out["aligned_frac"] = round(aligned / len(pairs), 3) if pairs else None + out["ok"] = out["err_finite"] and out["dr_finite"] and out["ft_seq_eq_fc"] and out["ft_ns_monotonic"] and (out["aligned_frac"] or 0) >= 0.95 + return out + + +def frame_record_size(bag: Path, export: Path) -> dict: + """The /arena/frame records AS RECORDED, rewritten alone in the final layout (chunk-only index, + zstd 19 — mcap_index.reindex's settings), against the gzip of the export's frame rows: the ratio + the 2026-10-18 decision accepted (docs/development/fictrac-columns-in-the-bag.md §3).""" + import gzip + import io + + from mcap.reader import make_reader + from mcap.writer import CompressionType, IndexType, Writer + + sys.path.insert(0, str(NODES)) + from mcap_index import ZSTD_LEVEL, _zstd_level # noqa: E402 + + buf = io.BytesIO() + n = 0 + with _zstd_level(ZSTD_LEVEL): + w = Writer(buf, compression=CompressionType.ZSTD, index_types=IndexType.ALL & ~IndexType.MESSAGE) + w.start(profile="ros2", library="test_ros2_poc") + ch = None + for f in sorted(bag.glob("*.mcap")): + with open(f, "rb") as fh: + for schema, _chan, m in make_reader(fh).iter_messages(topics=["/arena/frame"], log_time_order=False): + if ch is None: + ch = w.register_channel("/arena/frame", "cdr", w.register_schema(schema.name if schema else "frame", "ros2msg", schema.data if schema else b"")) + n += 1 + w.add_message(ch, log_time=m.log_time, publish_time=m.publish_time, data=m.data, sequence=n) + w.finish() + rows = b"\n".join(l for l in export.read_bytes().splitlines() if l.startswith(b"[") and not l.startswith(b'["')) + b"\n" + gz = len(gzip.compress(rows, 9)) + return {"frames": n, "bag_frame_bytes": len(buf.getvalue()), "rows_gz": gz, "ratio": round(len(buf.getvalue()) / gz, 2) if gz else None} + + def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--epochs", type=int, default=3) @@ -364,6 +455,15 @@ def side_actor() -> None: fictrac.start() kill_report["restarted_at_ms"] = int((time.monotonic() - t_run0) * 1000) time.sleep(1.0) + # One untracked FicTrac frame, tagged as the real tracker tags it (bridge.py: "FT_BADFR"). + # The adapter forwards any datagram, the pipeline skips it; the bag must still show it as + # a flagged /arena/frame record (2026-10-18), the run log must not. + try: + sk = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sk.sendto(b"FT_BADFR, 424242, 0\n", ("127.0.0.1", 60000)) + sk.close() + except OSError as exc: + kill_report["bad_frame_send_error"] = repr(exc) bias_replies["getTimestamp"] = bias_cmd("getTimestamp") # stopRecording once the LAST epoch has opened (same anchor, so it lands inside the run). last = f"cl_constant_{a.epochs - 1}" @@ -400,8 +500,11 @@ def side_actor() -> None: want_topics = ["/arena/session", "/arena/browser", "/arena/event", "/arena/frame", "/kinefly/raw", "/bias/cmd", "/bias/reply", "/bias/status", "/scanimage/cmd", "/scanimage/reply", "/scanimage/status"] missing = [t for t in want_topics if not bagsum.get("topics", {}).get(t)] - check("every stream recorded in the bag (session, browser, event, frame, kinefly, bias + scanimage cmd/reply/status, fictrac)", - not missing and (bagsum.get("topics", {}).get("/fictrac/frame") or bagsum.get("topics", {}).get("/fictrac/raw")), f"missing {missing}") + check("every stream recorded in the bag (session, browser, event, frame, kinefly, bias + scanimage cmd/reply/status)", + not missing, f"missing {missing}") + raw_ft = {t: n for t, n in bagsum.get("topics", {}).items() if t.startswith("/fictrac/")} + check("raw FicTrac topics NOT recorded by default (2026-10-18: the merged /arena/frame record is the FicTrac record; --record-fictrac-raw opts in)", + not any(raw_ft.values()), str(raw_ft)) typed = any(s.startswith("arena_interfaces/") for s in bagsum.get("schemas", [])) check("typed messages in the bag when arena_interfaces is built (else String fallback, reported)", True, "") @@ -434,6 +537,17 @@ def side_actor() -> None: by_type[e.get("type", "?")] = by_type.get(e.get("type", "?"), 0) + 1 print(f"[test] log: {len(objs)} lines — fictrac rows {len(rows_ft)}, arena echoes {len(rows_a)}, kf rows {len(rows_kf)}, events {by_type}", file=sys.stderr) check("FicTrac rows logged (every frame, not only applied)", len(rows_ft) > 40 * a.epochs, str(len(rows_ft))) + # ── the merged FicTrac record (2026-10-18) ─────────────────────────────────── + ev = frame_extras_values(main_bag) if main_bag else {} + check("merged record: err, dr_lab_x/y/z, ft_seq, ft_ns ride on every /arena/frame record and come from the SAME line as x/y/hd " + "(wrap(Δhd) == dr_lab_z on ≥ 95 % of consecutive frames; ft_seq == fc; ft_ns monotonic)", + ev.get("ok", False), json.dumps(ev)) + bad_row = any(isinstance(o, list) and isinstance(o[0], (int, float)) and o[1] == 424242 for o in objs) + check("untracked frame (FT_BADFR) → one flagged /arena/frame record carrying its fc, and NO run-log row", + ev.get("bad", 0) >= 1 and 424242 in ev.get("bad_fc", []) and not bad_row, + json.dumps({"bad": ev.get("bad"), "bad_fc": ev.get("bad_fc"), "row_in_log": bad_row, "send_error": kill_report.get("bad_frame_send_error")})) + cs_ft = [e for e in objs if isinstance(e, dict) and e.get("type") == "connector_status" and e.get("src") == "fictrac"] + check("connector_status (fictrac) reports bad_frames ≥ 1", bool(cs_ft) and (cs_ft[-1].get("bad_frames") or 0) >= 1, str(cs_ft[-1] if cs_ft else None)) check('["a",…] compact arena echoes logged', len(rows_a) > 40 * (a.epochs - 1), str(len(rows_a))) check('["kf",…] Kinefly rows logged (KF/1 → adapter → bridge)', len(rows_kf) > 100, str(len(rows_kf))) kf_ok = all(len(r) == 8 and isinstance(r[1], int) and isinstance(r[2], int) and r[4] is not None and r[5] is not None for r in rows_kf[:50]) @@ -533,11 +647,25 @@ def side_actor() -> None: check("bag → behavior_v1 == bridge.py --convert of the behavior_v2 (byte-identical)", r1.returncode == 0 and c1.returncode == 0 and bv1.read_bytes() == v1.read_bytes(), r1.stderr[-200:]) rf = run_capture(conv + ["--level", "full", "--out", str(bfull)]) - full_objs = read_jsonl(bfull) if rf.returncode == 0 else [] - ff = [o for o in full_objs if isinstance(o, dict) and o.get("type") == "fictrac_frame"] - check("bag → full level: one fictrac_frame per frame with the 25 FicTrac columns, no schema line", - rf.returncode == 0 and len(ff) == len(rows_ft) and all(len(o.get("fictrac") or []) == 25 for o in ff) - and not any(isinstance(o, dict) and o.get("type") == "frame_schema" for o in full_objs), f"{len(ff)} frames vs {len(rows_ft)}; cols {sorted({len(o.get('fictrac') or []) for o in ff})}") + check("bag → full level REFUSED without the raw FicTrac topics (the error names --record-fictrac-raw)", + rf.returncode != 0 and "--record-fictrac-raw" in rf.stderr, rf.stderr[-200:]) + # A bag from before 2026-10-18 (two records per frame, raw FicTrac topic present) still + # converts, full level included. The bench bags are untracked: checked when present. + old_bags = [b for b in sorted((POC / "runlogs").glob("arena-log-20261014-*.bag")) if (b / "metadata.yaml").exists()] + if old_bags: + ob = old_bags[0] + ob_out, ob_full = work / "oldbag.v2.jsonl", work / "oldbag.full.jsonl" + r_ob = run_capture([PY, str(NODES / "mcap_convert.py"), str(ob), "--out", str(ob_out)], timeout=180) + r_of = run_capture([PY, str(NODES / "mcap_convert.py"), str(ob), "--level", "full", "--out", str(ob_full)], timeout=180) + ob_rows = sum(1 for o in read_jsonl(ob_out) if isinstance(o, list) and isinstance(o[0], (int, float))) if r_ob.returncode == 0 else 0 + of_ff = [o for o in (read_jsonl(ob_full) if r_of.returncode == 0 else []) if isinstance(o, dict) and o.get("type") == "fictrac_frame"] + n25 = sum(1 for o in of_ff if len(o.get("fictrac") or []) == 25) + # ww18 2026-10-15: one frame per bag may lack its raw record (a line that arrived just before the open). + check(f"pre-2026-10-18 bag ({ob.name}) still converts; its full level carries the 25 FicTrac columns", + r_ob.returncode == 0 and ob_rows > 0 and r_of.returncode == 0 and len(of_ff) == ob_rows and n25 >= len(of_ff) - 1, + f"rc {r_ob.returncode}/{r_of.returncode}, rows {ob_rows}, full frames {len(of_ff)}, with 25 cols {n25}") + else: + print(" skip pre-2026-10-18 bag check: no runlogs/arena-log-20261014-*.bag in this checkout (the bench bags are untracked)", file=sys.stderr) rt = run_capture(conv + ["--order", "time", "--out", str(btime)]) time_lines = read_jsonl_lines(btime) if rt.returncode == 0 else [] swaps = sum(1 for x, y in zip(conv_lines, time_lines) if x != y) @@ -565,16 +693,29 @@ def side_actor() -> None: ifc = run_capture([PY, str(NODES / "build_interfaces.py"), "--check"], timeout=60) have_ifaces = ifc.returncode == 0 if have_ifaces: - check("typed: arena_interfaces importable → ArenaFrame + FicTracFrame recorded in the bag", - typed and bagsum.get("topics", {}).get("/fictrac/frame", 0) > 0, str(bagsum.get("schemas"))) + check("typed: arena_interfaces importable → ArenaFrame recorded in the bag (and no FicTracFrame by default)", + typed and bagsum.get("topics", {}).get("/arena/frame", 0) > 0 and not bagsum.get("topics", {}).get("/fictrac/frame"), str(bagsum.get("schemas"))) tv = typed_frame_values(main_bag) if main_bag else {} check("typed: ArenaFrame VALUES decode from the bag's own schema — core fields finite, fc monotonic, " - "turning_dps/forward_mms NaN on the first frame and finite (model-fly range) on ≥ 95 % of the rest", + "turning_dps/forward_mms NaN in the bag (live-only)", tv.get("ok", False), json.dumps(tv)) - print(f" info typed frames: {tv.get('n')}, derived finite on {tv.get('finite_frac')} of frames 2+, " - f"|turning| ≤ {tv.get('turning_max')} deg/s, |forward| ≤ {tv.get('forward_max')} mm/s", file=sys.stderr) else: print(" info arena_interfaces not importable here: String topics used (pixi install builds it on osx/linux; Windows needs MSVC)", file=sys.stderr) + fd = frame_deriver_values() + check("FrameDeriver (live turning_dps / forward_mms): NaN on the first frame, kinematics.js's numbers on the second", fd.get("ok", False), json.dumps(fd)) + # Size: the FicTrac record alone in the final layout. The guard is an absolute per-frame budget + # (adding one incompressible double costs ≈ 8 B/frame): 72 B/frame compressed, against the + # 63.7 B/frame this 25 s run measured on 2026-10-18 (≈ 3 B of that is the file's fixed overhead + # spread over 1 084 frames) and the bench bags' ≈ 45-55 (docs/development/fictrac-columns-in-the- + # bag.md §3). The ratio to the gzip of the frame rows is reported only — it swings with the + # fly's dynamics and a short run's fixed overhead. The String path's JSON record is reported only. + sz = frame_record_size(main_bag, export) if main_bag and export.exists() else {} + if sz.get("frames"): + sz["bytes_per_frame"] = round(sz["bag_frame_bytes"] / sz["frames"], 1) + whole = {"bag_bytes": bagsum.get("bytes"), "export_gz": len(__import__("gzip").compress(export.read_bytes(), 9)) if export.exists() else None} + print(f" info size: frame records {json.dumps(sz)}; whole bag (incl. kinefly/bias/browser) {json.dumps(whole)}", file=sys.stderr) + if typed: + check("size: /arena/frame records in the final layout (chunk-only index, zstd 19) ≤ 72 B/frame compressed", bool(sz) and (sz.get("bytes_per_frame") or 999) <= 72.0, json.dumps(sz)) check("bridge node still alive at the end of the main run", bridge.alive())