rig: record the rig-nodes host monitor (/rig/host) in every run's bag

rig/README.md

@@ -307,6 +307,7 @@ sends it at run end only on the commit path today) or on a run-end event.
 | live graph, interactive | `pixi install -e viz`, then `pixi run -e viz rqt-graph` | the standard ROS node/topic graph window (Qt); `pixi run -e viz rqt-topic` shows rates and payloads |
 | the recorded run | Foxglove (free desktop app) → *Open local file* → `runlogs/<run>.bag/<run>_0.mcap`; Layout → *Import from file* → `docs/foxglove-arena-layout.json` | every topic of the run on a timeline: plots of `/arena/frame` `idx`, `hd`, `turning_dps` + `forward_mms`, `bias_deg`, `epoch` (typed runs; String runs show the JSON text), the browser messages, events and BIAS replies as raw views, the schemas rosbag2 embedded — no ROS needed on that machine |
 | the live run in Foxglove | `pixi run rig -- --foxglove` (adds foxglove_bridge from the `viz` environment to the supervised rig; a note and no bridge when `pixi install -e viz` has not been run) → Foxglove *Open connection* `ws://localhost:8766`; or by hand beside a running rig: `pixi run -e viz foxglove-bridge` | the same panels on the running graph, including the derived `turning_dps` / `forward_mms` plot |
+| how loaded the PC is | nothing to start: `pixi run rig` runs `python -m rig_nodes.host_monitor` from the installed rig-nodes package when it has one (off: `--no-host-monitor`; a note and no monitor with an older package) | `/rig/host` once a second, recorded in every run's bag: CPU per core (mean + busiest 100 ms), P/E cores, sampler lateness, memory, page faults, disks, network, NVIDIA GPU (incl. NVENC), top processes + this workspace's nodes as `py:<script>`; ≈ 0.9 % of one core on ww18. Node, message and design: reiserlab/rig-nodes (`rig_nodes.host_monitor`, `rig_interfaces/HostStats`) |
 | the plain CLI | `pixi run ros2 node list`, `ros2 topic list -t`, `ros2 topic hz /arena/frame`, `ros2 topic echo /bias/status`, `ros2 bag info runlogs/<run>.bag` | what every ROS user reaches for first; every topic here is readable JSON or a flat typed message |
 
 **Color optomotor smoke run (22:00 ET, same bench, after merging Next into the branch).** The

rig/nodes/bridge_node.py

@@ -31,6 +31,8 @@
     node killed mid-run therefore leaves the log intact AND documented.
   * `connector_status` lines (frames seen, adapter→bridge latency p50/p99 ms per source)
     at log close / export, so a run carries its own transport health.
+  * `/rig/host` (rig_nodes.host_monitor: CPU per core, memory, disk, network, GPU, top processes) is
+    recorded into the bag on arrival, so every run carries the PC's pressure alongside it.
 
 Not supported here (by design of the PoC): the `fictrac_port` config key cannot re-bind
 the socket — the adapter owns it. The request is acknowledged with a warning.
@@ -168,6 +170,13 @@ def package_versions() -> dict:
     "scanimage": {"cmd": SCANIMAGE_CMD_TOPIC, "reply": SCANIMAGE_REPLY_TOPIC, "status": SCANIMAGE_STATUS_TOPIC},
 }
 FRAME_TOPIC = "/arena/frame"
+# Host pressure from rig_nodes.host_monitor (reiserlab/rig-nodes): recorded on arrival, no run-log line.
+HOST_TOPIC = "/rig/host"
+HOST_STATS_TYPE = "rig_interfaces/msg/HostStats"
+try:  # rig-nodes ≥ 0.3.0; the monitor publishes the same type when it imports (its --typed auto)
+    from rig_interfaces.msg import HostStats
+except (ImportError, AttributeError):
+    HostStats = None
 RAW_QOS = QoSProfile(
     reliability=QoSReliabilityPolicy.RELIABLE,
     history=QoSHistoryPolicy.KEEP_LAST,
@@ -324,6 +333,7 @@ def topics(self) -> dict[str, str]:
             SCANIMAGE_CMD_TOPIC: STRING_TYPE, SCANIMAGE_REPLY_TOPIC: STRING_TYPE, SCANIMAGE_STATUS_TOPIC: STRING_TYPE,
         }
         t[FRAME_TOPIC] = ARENA_FRAME_TYPE if self._typed else STRING_TYPE
+        t[HOST_TOPIC] = HOST_STATS_TYPE if HostStats is not None else STRING_TYPE
         if self.record_fictrac_raw:
             if self._typed:
                 t[FICTRAC_FRAME_TOPIC] = FICTRAC_FRAME_TYPE
@@ -665,6 +675,9 @@ 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)
+        # rig_nodes.host_monitor samples (1 Hz; the monitor publishes RELIABLE, KEEP_LAST 10).
+        self.create_subscription(HostStats if HostStats is not None else String, HOST_TOPIC,
+                                 lambda b: self._record(HOST_TOPIC, b), RAW_QOS, raw=True)
 
     def _record(self, topic: str, data: bytes) -> None:
         rec = self._log.rec

rig/nodes/rig.py

@@ -4,6 +4,7 @@
     pixi run rig                       # router, fictrac adapter, bridge node
     pixi run rig -- --kinefly --bias   # + Kinefly adapter + BIAS node
     pixi run rig -- --sim              # + the FicTrac simulator (demo without a camera)
+    pixi run rig -- --no-host-monitor  # without the host monitor (CPU/RAM/disk/net/GPU on /rig/host; on by default)
     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
@@ -26,6 +27,7 @@
 from __future__ import annotations
 
 import argparse
+import importlib.util
 import os
 import shutil
 import signal
@@ -46,6 +48,15 @@
 _ENVS = Path(os.environ["CONDA_PREFIX"]).parent if os.environ.get("CONDA_PREFIX") else ROOT / ".pixi" / "envs"
 VIZ_ENV = _ENVS / "viz"
 FOXGLOVE_PORT = 8766
+HOST_MONITOR_MODULE = "rig_nodes.host_monitor"
+
+
+def host_monitor_available() -> bool:
+    """True when the installed rig_nodes package ships the host monitor."""
+    try:
+        return importlib.util.find_spec(HOST_MONITOR_MODULE) is not None
+    except ImportError:  # no rig_nodes package at all
+        return False
 
 
 def node_argv(module: str, *args: str) -> list[str]:
@@ -71,6 +82,8 @@ def main(argv: list[str] | None = None) -> int:
     p.add_argument("--bias-args", default="", help="extra args for the BIAS node")
     p.add_argument("--scanimage-args", default="", help="extra args for the ScanImage node (e.g. --port 5555 --status-every 1)")
     p.add_argument("--sim-args", default="--proto udp --rate 50", help="args for the FicTrac simulator")
+    p.add_argument("--no-host-monitor", action="store_true", help=f"do not run {HOST_MONITOR_MODULE} (the PC's CPU per core, memory, disk, network, GPU, top processes on /rig/host, recorded in every run's bag)")
+    p.add_argument("--host-monitor-args", default="", help=f"extra args for {HOST_MONITOR_MODULE} (e.g. --period 0.5 --print-every 30)")
     args = p.parse_args(argv)
 
     py = sys.executable
@@ -124,6 +137,15 @@ def main(argv: list[str] | None = None) -> int:
         nodes.append(Child("bias-sim", node_argv("rig_nodes.sims.bias_sim")))
     if args.bias:
         nodes.append(Child("bias", node_argv("rig_nodes.bias_node", *split_args(args.bias_args))))
+    if not args.no_host_monitor:
+        # The generic host monitor ships in the rig_nodes package (reiserlab/rig-nodes). It starts
+        # before the gateway (bridge_node.py, below), so a run opened right away already has host
+        # samples in its bag; `--py-root` lets it name this workspace's Python nodes `py:<script>`.
+        if host_monitor_available():
+            nodes.append(Child("host", node_argv(HOST_MONITOR_MODULE, "--log-dir", str(ROOT / "runlogs"),
+                                                  "--py-root", str(ROOT), *split_args(args.host_monitor_args))))
+        else:
+            print(f"[rig] host monitor: the installed rig_nodes has no {HOST_MONITOR_MODULE} (needs a rig-nodes release with it); starting without it", file=sys.stderr)
     if args.si_mock:
         nodes.append(Child("si-mock", node_argv("rig_nodes.sims.si_json_mock")))
     if args.scanimage:

rig/tests/test_ros2_poc.py

@@ -50,6 +50,11 @@
                                                         NDJSON — reported, crashsafe profile gated;
                                                         then a RESTARTED node's log_export recovers the
                                                         bag on its own (recovered:true, whole bag left)
+  host monitor (2026-10-28; rig_nodes.host_monitor)   /rig/host in the bag ≥ 0.8 Hz, fields = the installed
+                                                        HostStats (typed) / the read fields (JSON), the graph's
+                                                        nodes pinned; its own CPU ≤ 2 % of a core, ≤ 5 % of the
+                                                        compressed bag — skipped (info) when the installed
+                                                        rig_nodes has no host monitor
 
 Exit 0 when all checks pass. Prints a markdown summary. Artifacts are kept with --keep.
 """
@@ -241,6 +246,68 @@ def bag_summary(bag: Path) -> dict:
             "frame_lag_ms": {"p50": round(lag[len(lag) // 2], 3), "p99": round(lag[int(len(lag) * 0.99)], 3), "max": round(lag[-1], 3)} if lag else None}
 
 
+def host_monitor_values(bag: Path) -> dict:
+    """/rig/host in the bag: sample count over the run, field names vs the request's HostStats.msg,
+    per-core sanity, pinned groups, the monitor's own CPU, and its share of the compressed payload."""
+    import statistics
+
+    import zstandard
+
+    sys.path.insert(0, str(NODES))
+    from mcap_convert import _json_of, decode_cdr, read_records  # noqa: E402
+
+    try:  # the field contract is rig-nodes' (rig_interfaces/HostStats); without the typed build the
+        # JSON samples are checked for the fields this harness reads
+        from rig_interfaces.msg import HostStats
+
+        want = sorted(HostStats.get_fields_and_field_types())
+    except (ImportError, AttributeError):
+        want = None
+    host, other, t_ns, objs, schemas = [], [], [], [], set()
+    gen = read_records(bag)
+    try:
+        while True:
+            r = next(gen)
+            t_ns.append(r["log_ns"])
+            if r["topic"] != "/rig/host":
+                other.append(r["data"])
+                continue
+            host.append(r["data"])
+            schemas.add(r["schema"])
+            try:
+                if want is not None and r["schema"] == "rig_interfaces/msg/HostStats":
+                    # typed: ROS's own deserializer (the converter's pure decoder reads no nested
+                    # message arrays — DiskStats[] disks … — and the converter never needs /rig/host)
+                    from rclpy.serialization import deserialize_message
+                    from rosidl_runtime_py.convert import message_to_ordereddict
+
+                    objs.append(json.loads(json.dumps(message_to_ordereddict(deserialize_message(r["data"], HostStats)))))
+                else:  # String JSON → its object
+                    objs.append(_json_of({"msg": decode_cdr(r["schema"], r.get("schema_text"), r["data"])}))
+            except Exception:
+                objs.append(None)
+    except StopIteration:
+        pass
+    good = [o for o in objs if isinstance(o, dict)]
+    need = {"core_busy_mean", "core_busy_peak", "monitor_cpu_cores", "procs"}
+    bad = sorted({k for o in good for k in (set(o) ^ set(want) if want else need - set(o))})
+    cores_ok = bool(good) and all(len(o["core_busy_mean"]) == len(o["core_busy_peak"]) == (os.cpu_count() or 0)
+                                  and all(p + 1e-6 >= m for m, p in zip(o["core_busy_mean"], o["core_busy_peak"])) for o in good)
+    self_cpu = [o["monitor_cpu_cores"] for o in good[2:] if isinstance(o.get("monitor_cpu_cores"), (int, float))]
+    level = 19  # the bag's zstd "Slowest" (recorder.py `default` profile; crashsafe is reindexed to it)
+    zh = len(zstandard.ZstdCompressor(level=level).compress(b"".join(host))) if host else 0
+    za = len(zstandard.ZstdCompressor(level=level).compress(b"".join(host + other))) if host or other else 0
+    run_s = (max(t_ns) - min(t_ns)) / 1e9 if t_ns else None
+    return {"samples": len(host), "schema": sorted(s or "std_msgs/msg/String" for s in schemas), "run_s": round(run_s, 1) if run_s else None, "fields_ok": bool(good) and len(good) == len(objs) and not bad,
+            "bad_fields": bad, "cores_ok": cores_ok, "ncpu": len(good[0]["core_busy_mean"]) if good else None,
+            "groups": sorted({p["group"] for o in good for p in o.get("procs", []) if p.get("group")}),
+            "self_cpu_median": round(statistics.median(self_cpu), 4) if self_cpu else None,
+            "self_cpu_max": round(max(self_cpu), 4) if self_cpu else None,
+            "bytes_per_sample": round(sum(map(len, host)) / len(host)) if host else None,
+            "share_compressed": round(zh / za, 4) if za else None,
+            "host_kb_per_h_zstd": round(zh / run_s * 3600 / 1024) if run_s else None}
+
+
 def typed_frame_values(bag: Path) -> dict:
     """The recorded /arena/frame ArenaFrame messages, decoded with the converter's pure-Python
     CDR decoder from the schema text IN THE BAG, judged by VALUE — a schema name alone said
@@ -403,6 +470,13 @@ def start(name: str, argv: list[str]) -> Proc:
         si_port = free_tcp_port()
         start("si-mock", [PY, "-m", "rig_nodes.sims.si_json_mock", "--port", str(si_port), "--grab-s", "1.0", "--start-delay-s", "0.2", "--stall-on", "hSI.hRoiManager.scanZoomFactor"])
         start("scanimage-node", [PY, "-m", "rig_nodes.scanimage_node", "--port", str(si_port), "--status-every", "1", "--timeout-s", "1.5"])
+        from rig import HOST_MONITOR_MODULE, host_monitor_available  # noqa: E402
+
+        host_mon = host_monitor_available()
+        if host_mon:  # the command line rig.py uses
+            start("host-monitor", [PY, "-m", HOST_MONITOR_MODULE, "--log-dir", str(runlogs), "--py-root", str(POC)])
+        else:
+            print(f"  info  host monitor: the installed rig_nodes has no {HOST_MONITOR_MODULE} — its checks are skipped", file=sys.stderr)
         bridge_argv = [PY, str(NODES / "bridge_node.py"), "--log-dir", str(runlogs), "--source-timeout", "1.5", "--bag-profile", a.bag_profile]
         bridge = start("bridge-node", bridge_argv)
         check("bridge node serving ws://127.0.0.1:8765", wait_port("127.0.0.1", 8765, 20.0))
@@ -716,6 +790,22 @@ def side_actor() -> None:
         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))
 
+        # ── host monitor (/rig/host, rig_nodes.host_monitor) ─────────────────────────────
+        hm = host_monitor_values(main_bag) if main_bag and host_mon else {}
+        if host_mon:
+            print(f"  info  host monitor: {json.dumps({k: v for k, v in hm.items() if k not in ('fields_ok', 'groups')})}", file=sys.stderr)
+            check("host monitor: /rig/host recorded in the bag, ≥ 0.8 samples per second of the run",
+                  (hm.get("samples") or 0) >= 0.8 * (hm.get("run_s") or 99), f"{hm.get('samples')} samples over {hm.get('run_s')} s")
+            check("host monitor: every sample has the HostStats fields (the installed rig_interfaces' when typed)",
+                  hm.get("fields_ok", False), str(hm.get("bad_fields")))
+            check("host monitor: one entry per logical CPU, busiest sub-window ≥ the window mean on every core",
+                  hm.get("cores_ok", False), f"{hm.get('ncpu')} CPUs")
+            check("host monitor: the graph's processes pinned by script (py:bridge_node, py:fictrac_udp_adapter, py:host_monitor)",
+                  {"py:bridge_node", "py:fictrac_udp_adapter", "py:host_monitor"} <= set(hm.get("groups") or []), str(hm.get("groups")))
+            check("host monitor cost: its own CPU ≤ 2 % of one core (median after the first 2 samples)",
+                  hm.get("self_cpu_median") is not None and hm["self_cpu_median"] <= 0.02, f"median {hm.get('self_cpu_median')} max {hm.get('self_cpu_max')} cores")
+            check("host monitor storage: /rig/host ≤ 5 % of the run's compressed payload (zstd, as the bag)",
+                  hm.get("share_compressed") is not None and hm["share_compressed"] <= 0.05, f"{hm.get('share_compressed')} ({hm.get('host_kb_per_h_zstd')} KB/h)")
         check("bridge node still alive at the end of the main run", bridge.alive())
 
         # ── crash phase: SIGKILL the bridge node mid-run, both writer profiles ────────