W5: scope captures of several nodes on one trigger

README.md

@@ -138,7 +138,14 @@ bottom bar):
   enabled without a click.
 - **Plots:** the published subjects of any nodes on one time axis, a window,
   pause, CSV export; commands, stops and register writes are markers.
-- **Scope:** cymon captures with trigger and pre-trigger, CSV export.
+- **Scope:** cymon captures with trigger and pre-trigger on one node or on
+  several at once: one node triggers on its condition, and the others trigger
+  on its `cymon.TriggerEvent` (about 1 to 2 ms later on the bus). For that the
+  nodes need the shared trigger subject (`uavcan.pub.cymon_trigger.id` and
+  `uavcan.sub.cymon_trigger.id`, disabled by default on UNIMOC); the view sets
+  it (subject 220), stores it and restarts the nodes. All channels in one plot,
+  each node aligned on its own trigger sample; CSV export; captures taken
+  during a run are stored in the run.
 - **Simulation:** the virtual clock of a simulated drive (pause, run, advance,
   reset) and its plant inputs (`unimoc.sim.*`).
 - **Runs:** see below.

docs/web_plan.md

@@ -1,6 +1,6 @@
 # One app: cypilot with the cymon features and a web front end
 
-Status (2026-10-25): **decided; W1 to W4 done** (section 8). It changes decisions
+Status (2026-10-25): **decided; W1 to W5 done** (section 8). It changes decisions
 of [plan.md](plan.md) (section 2: the Qt GUI, and cymon as a separate tool
 the GUI depends on) and replaces its phases G5 to G7 by the phases W0 to W7
 below. The firmware contract (plan sections 5.1, 6, 7.1, 9) stays as it is.
@@ -221,7 +221,7 @@ connected, and a browser that connects mid-run sees it.
 | W2 | **Done 2026-10-25 (the page is a placeholder until W3):** **Web server:** `cypilot.web` (FastAPI, event hub, static files), cymon's API plus control, profiles, simulation, plots; auth (PIN or token) and the control lease | API tests (httpx) against the simulation, lease loss stops a drive |
 | W3 | **Done 2026-10-25 (replay of recordings moves to W4):** **Web UI** with the G2 to G4 functions: bus, nodes with chips, registers with the description file, control, plots, simulation, log; responsive for the phone; then the **Qt GUI is removed** | Browser tests with Playwright (headless Chromium) against the simulation, also at phone size |
 | W4 | **Done 2026-10-25 (the telemetry rate stays as configured):** **Run log:** runs started and stopped by the user, raw frames plus decoded channels plus `run.json`, list, download and delete in the web app, replay, free-space limit | A run of an engage, a setpoint step and a trip in the simulation, downloaded, replayed and deleted from the UI |
-| W5 | **Scope** (old G5): cymon captures, several nodes on one shared trigger, overlay, stored in the run | Capture of two simulated drives on one trigger |
+| W5 | **Done 2026-10-25:** **Scope** (old G5): cymon captures, several nodes on one shared trigger, overlay, stored in the run | Capture of two simulated drives on one trigger |
 | W6 | **Raspberry Pi:** setup script, systemd service, access point, CAN FD HAT; load test (four drives, all subjects recorded, two browsers) | Runs a day on the cart bench with the drives |
 | W7 | **Packaging and the end of cymon:** wheel, Windows and Linux executables without Qt, README workflows; `Tecnologic/cymon` archived | Install on a clean Windows machine and a fresh Pi |
 

src/cypilot/core/scope.py

@@ -6,12 +6,21 @@
 service whose ``.type`` is ``cymon.<Service>.0.1`` (``docs/plan.md``,
 section 8.2). A capture is set up, armed, polled until complete and read out
 page by page; the time axis is relative to the trigger sample.
+
+**Several nodes on one trigger** (:meth:`ScopeClient.multi_capture`): a node
+publishes ``cymon.TriggerEvent`` when its capture triggers, and an armed node
+that subscribes to it triggers at its next sample. With the trigger ports
+(``uavcan.pub.cymon_trigger.id``, ``uavcan.sub.cymon_trigger.id``) on one
+subject on every node, the other nodes are armed on a condition that never
+holds, then the trigger node on the user's condition. Their trigger samples are
+late by the time the event takes over the bus (about 1 to 2 ms on UNIMOC).
 """
 
 from __future__ import annotations
 
 import asyncio
 import contextlib
+import dataclasses
 from dataclasses import dataclass
 from typing import Any
 
@@ -39,6 +48,11 @@
     9: "bad config",
 }
 _COMPLETE = 4
+TRIGGER_PORTS = ("uavcan.pub.cymon_trigger.id", "uavcan.sub.cymon_trigger.id")
+DEFAULT_TRIGGER_SUBJECT = 220
+"""The subject of the shared trigger that cypilot offers to set (free on UNIMOC)."""
+_NEVER = 3.0e38
+"""A level no variable reaches: the followers fire only on the trigger event."""
 _PAGE = 255
 """Frames asked for per ReadSamples request; the node answers with as many as fit."""
 
@@ -79,6 +93,7 @@ def __init__(self, session: Any, events: EventHub | None = None) -> None:
         self._events = events or EventHub()
         self._service_ids: dict[int, dict[str, int]] = {}
         self._clients: dict[tuple[int, str], Any] = {}
+        self._periods: dict[int, float] = {}
 
     async def discover(self, node_id: int, refresh: bool = False) -> dict[str, int]:
         """The service-IDs of the cymon services by key; empty when the node has none."""
@@ -183,10 +198,109 @@ async def disarm(self, node_id: int) -> None:
 
     async def capture(self, node_id: int, request: CaptureRequest) -> Capture:
         """Set up, arm, wait for completion, read out; emits ``capture`` events on the way."""
-        import cymon
+        if request.trigger_mode not in TRIGGER_MODES:
+            raise ScopeError(f"unknown trigger mode {request.trigger_mode!r}")
+        await self._setup(node_id, request)
+        await self._arm(node_id, request)
+        loop = asyncio.get_running_loop()
+        await self._wait(node_id, loop.time() + request.timeout_s, request.timeout_s)
+        return await self.read(node_id)
 
+    async def multi_capture(
+        self, channels: dict[int, list[int]], trigger_node: int, request: CaptureRequest
+    ) -> dict[int, Capture]:
+        """One capture on several nodes, triggered by ``trigger_node`` on the condition of ``request``.
+
+        ``channels``: the variable IDs per node (``request.channels`` is ignored); the
+        period, the length and the pre-trigger part are those of ``request`` on every node.
+        The nodes need the shared trigger subject (:meth:`trigger_ports`).
+        """
+        if trigger_node not in channels:
+            raise ScopeError(f"the trigger node {trigger_node} takes part in the capture")
         if request.trigger_mode not in TRIGGER_MODES:
             raise ScopeError(f"unknown trigger mode {request.trigger_mode!r}")
+        if len(channels) > 1:
+            state = await self.trigger_ports(list(channels))
+            if not state["shared"]:
+                raise ScopeError(f"the nodes have no shared trigger subject: {state['problem']}")
+        followers = [node for node in channels if node != trigger_node]
+        for node, node_channels in channels.items():
+            await self._setup(node, dataclasses.replace(request, channels=node_channels))
+        armed: list[int] = []
+        try:
+            for node in followers:
+                never = dataclasses.replace(
+                    request, trigger_mode="rising", trigger_source=channels[node][0], trigger_level=_NEVER
+                )
+                await self._arm(node, never)
+                armed.append(node)
+            await self._arm(trigger_node, request)
+            armed.append(trigger_node)
+            deadline = asyncio.get_running_loop().time() + request.timeout_s
+            for node in [trigger_node, *followers]:
+                await self._wait(node, deadline, request.timeout_s)
+        except BaseException:
+            for node in armed:
+                with contextlib.suppress(ScopeError):
+                    await self.disarm(node)
+            raise
+        return {node: await self.read(node) for node in [trigger_node, *followers]}
+
+    async def trigger_ports(self, node_ids: list[int]) -> dict[str, Any]:
+        """The trigger subjects of the nodes and whether they share one (published and subscribed)."""
+        nodes: dict[int, dict[str, int | None]] = {}
+        for node_id in node_ids:
+            client = self._session.registers(node_id)
+            try:
+                registers = await client.scan(list(TRIGGER_PORTS))
+            except RegisterError as error:
+                raise ScopeError(f"cannot read the trigger ports of node {node_id}: {error}") from error
+            finally:
+                client.close()
+            values: dict[str, int | None] = {"pub": None, "sub": None}
+            for register in registers:
+                content = register.value.content
+                if register.value.kind != "empty" and isinstance(content, list) and content:
+                    values["pub" if ".pub." in register.name else "sub"] = int(content[0])
+            nodes[node_id] = values
+        problem = ""
+        subjects = {v["pub"] for v in nodes.values()} | {v["sub"] for v in nodes.values()}
+        missing = [n for n, v in nodes.items() if v["pub"] is None or v["sub"] is None]
+        if missing:
+            problem = f"node {missing[0]} has no trigger ports"
+        elif len(subjects) != 1 or next(iter(subjects)) is None or int(next(iter(subjects)) or 0) > 8191:
+            problem = "the trigger ports differ or are disabled: " + ", ".join(
+                f"{n} pub {v['pub']} sub {v['sub']}" for n, v in nodes.items()
+            )
+        shared = next(iter(subjects)) if not problem else None
+        return {
+            "nodes": {str(n): v for n, v in nodes.items()},
+            "shared": not problem,
+            "subject": shared,
+            "problem": problem,
+        }
+
+    async def set_trigger_ports(self, node_ids: list[int], subject: int = DEFAULT_TRIGGER_SUBJECT) -> None:
+        """Writes the shared trigger subject into the nodes; it takes effect after a store and a restart."""
+        from cypilot.core.registers import RegisterValue
+
+        if not 0 <= subject <= 8191:
+            raise ScopeError(f"a subject-ID is 0 .. 8191, not {subject}")
+        for node_id in node_ids:
+            client = self._session.registers(node_id)
+            try:
+                for name in TRIGGER_PORTS:
+                    result = await client.write(name, RegisterValue("natural16", [subject]))
+                    if not result.accepted:
+                        raise ScopeError(f"node {node_id} refused {name} = {subject}")
+            except RegisterError as error:
+                raise ScopeError(f"cannot write the trigger ports of node {node_id}: {error}") from error
+            finally:
+                client.close()
+
+    async def _setup(self, node_id: int, request: CaptureRequest) -> None:
+        import cymon
+
         setup = await self._call(
             node_id,
             "setup",
@@ -198,6 +312,11 @@ async def capture(self, node_id: int, request: CaptureRequest) -> Capture:
             ),
         )
         _check(setup.result.value, "SetupCapture")
+        self._periods[node_id] = float(setup.actual_sample_period_us)
+
+    async def _arm(self, node_id: int, request: CaptureRequest) -> None:
+        import cymon
+
         arm = await self._call(
             node_id,
             "arm",
@@ -210,28 +329,23 @@ async def capture(self, node_id: int, request: CaptureRequest) -> Capture:
             ),
         )
         _check(arm.result.value, "ArmTrigger")
-        self._events.emit(
-            "capture", node_id=node_id, state="armed", period_us=float(setup.actual_sample_period_us)
-        )
+        self._events.emit("capture", node_id=node_id, state="armed", period_us=self._periods.get(node_id))
 
+    async def _wait(self, node_id: int, deadline: float, timeout_s: float) -> None:
         loop = asyncio.get_running_loop()
-        deadline = loop.time() + request.timeout_s
         state = "armed"
         while state != "complete":
             if loop.time() > deadline:
                 with contextlib.suppress(ScopeError):
                     await self.disarm(node_id)
-                raise ScopeError(
-                    f"capture on node {node_id} did not complete within {request.timeout_s} s ({state})"
-                )
+                raise ScopeError(f"capture on node {node_id} did not complete within {timeout_s} s ({state})")
             await asyncio.sleep(0.05)
             new_state = (await self.status(node_id))["state"]
             if new_state != state:
                 state = new_state
                 self._events.emit("capture", node_id=node_id, state=state)
             if state in ("idle", "configured"):
                 raise ScopeError(f"capture on node {node_id} was disarmed")
-        return await self.read(node_id)
 
     async def read(self, node_id: int) -> Capture:
         """Reads a complete capture again."""

src/cypilot/demo.py

@@ -35,7 +35,7 @@
 import uavcan.diagnostic
 import uavcan.node
 import uavcan.primitive.array
-from pycyphal.application.register import Bit, Natural16, Real32, Value
+from pycyphal.application.register import Bit, Natural16, Real32, String, Value
 
 from cypilot import __version__
 
@@ -86,6 +86,8 @@ def __init__(
         self.prev = 0.0
         self.awaiting_low = False
         self.trigger_requested = False
+        self.trigger_event: tuple[int, int, float] | None = None
+        """Source, mode and value of the last trigger, until published as ``cymon.TriggerEvent``."""
 
     def setup(
         self, channels: list[int], period_us: float, num_samples: int, pretrigger: int
@@ -122,6 +124,7 @@ def arm(self, arm: bool, source: int, mode: int, level: float, band: float) -> i
         self.prev = self.read(source) if source < len(self.variables) else 0.0
         self.awaiting_low = mode == HYSTERESIS and self.prev >= level
         self.pre, self.post, self.counter, self.trigger_requested = [], [], 0, False
+        self.trigger_event = None
         self.state = ARMED
         return 0
 
@@ -150,6 +153,7 @@ def tick(self) -> None:
             filled = len(self.pre) >= self.pretrigger
             if filled and (self._fires(value) or self.trigger_requested):
                 self.state = TRIGGERED
+                self.trigger_event = (self.source, self.mode, value)
             self.prev = value
         frame = [self.read(c) for c in self.channels]
         if self.state == ARMED:
@@ -172,7 +176,9 @@ def __init__(
         transport: pycyphal.transport.Transport,
         name: str = "org.tecnologic.cypilot.demo",
         tick_period_us: float = 100.0,
+        trigger_subject: int = 65535,
     ) -> None:
+        """``trigger_subject``: the ports of ``cymon.TriggerEvent`` (65535: disabled, as on UNIMOC)."""
         env = {f"UAVCAN__SRV__{k.upper()}__ID": str(v) for k, v in SERVICE_IDS.items()}
         env["UAVCAN__PUB__TELEMETRY__ID"] = str(TELEMETRY_SUBJECT)
         env["UAVCAN__SUB__READINESS__ID"] = str(READINESS_SUBJECT)
@@ -199,6 +205,11 @@ def __init__(
         registers["demo.engaged"] = lambda: Value(natural16=Natural16([int(self.engaged)]))
         registers["demo.store_count"] = lambda: Value(natural16=Natural16([self.store_count]))
         registers["unimoc.settings.pending"] = lambda: Value(bit=Bit([False]))
+        for name in ("uavcan.pub.cymon_trigger", "uavcan.sub.cymon_trigger"):
+            registers[f"{name}.id"] = Value(natural16=Natural16([trigger_subject]))
+            registers[f"{name}.type"] = Value(string=String("cymon.TriggerEvent.0.1"))
+        self._trigger_pub: Any = None
+        self._trigger_sub: Any = None
 
         variables = [("sine", "A"), ("cosine", "A"), ("ramp", ""), ("engaged", "")]
         self.capture = CaptureEngine(variables, self._read_variable, tick_period_us)
@@ -232,8 +243,38 @@ def _read_variable(self, index: int) -> float:
             float(self.engaged),
         ][index]
 
+    def _apply_trigger_ports(self) -> None:
+        """(Re)creates the trigger ports from their registers, as UNIMOC does at start."""
+        for port in (self._trigger_pub, self._trigger_sub):
+            if port is not None:
+                port.close()
+        self._trigger_pub = self._trigger_sub = None
+        registry = self.node.registry
+        pub = int(registry["uavcan.pub.cymon_trigger.id"])
+        sub = int(registry["uavcan.sub.cymon_trigger.id"])
+        if pub <= 8191:
+            self._trigger_pub = self.node.presentation.make_publisher(cymon_dsdl.TriggerEvent_0_1, pub)
+        if sub <= 8191:
+            self._trigger_sub = self.node.presentation.make_subscriber(cymon_dsdl.TriggerEvent_0_1, sub)
+            self._trigger_sub.receive_in_background(self._on_trigger_event)
+
+    async def _on_trigger_event(self, _: Any, transfer: Any) -> None:
+        if transfer.source_node_id != self.node.id and self.capture.state == ARMED:
+            self.capture.trigger_requested = True
+
+    async def _publish_trigger(self) -> None:
+        event, self.capture.trigger_event = self.capture.trigger_event, None
+        if event is not None and self._trigger_pub is not None:
+            source, mode, value = event
+            await self._trigger_pub.publish(
+                cymon_dsdl.TriggerEvent_0_1(
+                    source_variable_id=source, mode=cymon_dsdl.TriggerMode_0_1(mode), trigger_value=value
+                )
+            )
+
     def start(self) -> None:
         self.node.start()
+        self._apply_trigger_ports()
         loop = asyncio.get_running_loop()
         self._tasks = [loop.create_task(self._sampling()), loop.create_task(self._publishing())]
 
@@ -253,6 +294,8 @@ async def _sampling(self) -> None:
             for _ in range(ticks):
                 self.t += self._tick_period_s
                 self.capture.tick()
+            if self.capture.trigger_event is not None:
+                await self._publish_trigger()
 
     async def _publishing(self) -> None:
         while True:
@@ -304,6 +347,7 @@ async def _on_command(self, request: Any, _: Any) -> Any:
         if request.command == 65535:
             self.estopped = False
             self._update_status()
+            self._apply_trigger_ports()
             return Status(status=Status.STATUS_SUCCESS)
         return Status(status=Status.STATUS_BAD_COMMAND)
 

src/cypilot/simulation.py

@@ -133,6 +133,9 @@ def slcan_interfaces(self) -> list[str]:
         return [f"slcan:{path}" for name, path in sorted(self.ptys.items()) if name != "uart"]
 
     def start(self, timeout: float = 15.0) -> None:
+        """Starts the simulation (again after :meth:`stop`: new pseudo terminals, the same flash)."""
+        self.ptys.clear()
+        self._started.clear()
         command = [str(self.executable), f"-flash={self.flash}", "-no-color"]
         if self.node_id is not None:
             command.append(f"-node-id={self.node_id}")