Add health fault lifecycle tracking and InfluxDB export

docker/docker-compose.diagnostics.yml

@@ -19,7 +19,7 @@ services:
       - influxdb-config:/etc/influxdb2
 
   grafana:
-    image: grafana/grafana:10.4.12
+    image: grafana/grafana:13.1
     container_name: xbot2-diagnostics-grafana
     restart: unless-stopped
     depends_on:

docs/fault_health_contract.md

@@ -0,0 +1,129 @@
+# Fault diagnostic contract
+
+The aggregator interprets a diagnostic message as a fault-state source only when its normalized `node` path ends in `/fault`.
+
+Examples:
+
+- `/xbot/joint/knee_pitch_1/fault`
+- `/xbot/power/battery/fault`
+- `host/robot-pc/network/eth0/fault`
+
+Messages with other suffixes remain ordinary diagnostics even if they contain similarly named values.
+
+## ROS string-native values
+
+`diagnostic_msgs/KeyValue.value` is a string. A valid `/fault` message therefore contains:
+
+- exactly one `fault_count` entry containing a non-negative whole number. The
+  ROS bridge form `"1.000000"` is also accepted;
+- zero or more repeated `fault_report` entries;
+- `fault_count` equal to the number of unique `fault_report` values.
+
+An empty `fault_report` value is treated as an omitted report for compatibility
+with fixed-size publisher slots. It is therefore valid only together with a
+zero `fault_count`.
+
+Each `fault_report` is a standardized, stable, human-friendly identifier. It must not contain changing measurements, timestamps, counters, or other occurrence-specific text.
+
+Active example:
+
+```yaml
+name: /xbot/joint/knee_pitch_1/fault
+hardware_id: knee_pitch_1
+level: 2
+message: Drive faults active
+values:
+  - key: fault_count
+    value: "2"
+  - key: fault_report
+    value: motor over temperature
+  - key: fault_report
+    value: encoder signal lost
+```
+
+Healthy example:
+
+```yaml
+name: /xbot/joint/knee_pitch_1/fault
+hardware_id: knee_pitch_1
+level: 0
+message: OK
+values:
+  - key: fault_count
+    value: "0"
+```
+
+The reports are the complete currently active set, not deltas. XBot2 should publish immediately when the set, severity, or summary changes and should also publish a periodic heartbeat, with 1 Hz as the default recommendation.
+
+## Level semantics
+
+| Reports | `level` | Meaning |
+|---|---:|---|
+| none | `0` | Healthy; clear all faults previously reported by this source |
+| one or more | `1` | Warning-level reports active |
+| one or more | `2` | Error-level reports active |
+| unchanged/any | `3` | Source stale; do not raise or clear hardware faults |
+
+Messages missing `fault_count`, containing duplicate count entries, having an inconsistent count, or having an inconsistent level/report combination are ignored for lifecycle tracking. Ignoring malformed data is safer than clearing an existing fault.
+
+`msg` is dashboard summary text and is not part of fault identity. Dynamic measurements and vendor codes may be supplied as additional key-value entries.
+
+The lifecycle identity is:
+
+```text
+(hw_id, node, fault_report)
+```
+
+## InfluxDB schema
+
+Every valid fault publication, including the heartbeat, produces one `fault` point.
+
+Tags:
+
+- `hw_id`
+- `path`
+- `name`
+- `component`
+
+Fields:
+
+- `level`
+- `message`
+- `fault_count`
+- `active_fault_reports` (sorted reports joined for display)
+- `last_fault_report`, when known
+- `last_fault_active`, when known
+- `last_fault_level`, when known
+- `last_raised_ns`, when known
+- `last_cleared_ns`, when known
+
+This gives Grafana one current row per fault source using a latest-point query.
+
+Each raise or clear produces one `fault_event` point.
+
+Tags:
+
+- `hw_id`
+- `path`
+- `name`
+- `component`
+- `fault_report`
+- `transition` (`raised` or `cleared`)
+
+Fields:
+
+- `active`
+- `level`
+- `message`
+- `occurrence_count`
+- `first_raised_ns`
+- `last_raised_ns`
+- `last_cleared_ns`, when available
+
+`fault_report` is deliberately a tag because reports are standardized and bounded, giving the same cardinality characteristics as standardized numeric fault codes while making Grafana filtering and grouping directly human-readable.
+
+Both fault and event points use the diagnostic source timestamp when valid, falling back to aggregator wall-clock time only when necessary.
+
+## Design rationale
+
+The `/fault` suffix creates an explicit namespace boundary. Repeated `fault_report` entries are native to ROS string key-values and avoid JSON embedded inside strings. An explicit `fault_count` distinguishes a healthy authoritative snapshot from a publisher that omitted the contract. Complete-set publication is idempotent and lets the aggregator derive raises and clears through set differences. Periodic `fault` snapshots make the primary Grafana table robust to packet loss, subscriber startup order, and aggregator restarts, while `fault_event` points retain transition history without writing duplicate events on every heartbeat.

python/src/pyxbot2_diagnostics/aggregator/aggregator.py

@@ -14,6 +14,11 @@
 from jsonschema import ValidationError, validate
 
 from pyxbot2_diagnostics.aggregator.config import AggregatorConfig
+from pyxbot2_diagnostics.aggregator.fault_tracker import (
+    FaultLifecycleTracker,
+    FaultState,
+    FaultTransition,
+)
 
 LOGGER = logging.getLogger(__name__)
 MESSAGE_SCHEMA_PATH = Path(__file__).resolve().parent / "schema" / "diagnostics_message.schema.json"
@@ -116,10 +121,16 @@ def __init__(
 
         self.state_cache: dict[str, DiagnosticsMessage] = {}
         self._last_seen: dict[str, float] = {}
+        self._fault_tracker = FaultLifecycleTracker()
 
         self._running = False
         self._last_stale_check = self._time_fn()
 
+    @property
+    def fault_states(self) -> dict[Any, FaultState]:
+        """Return a snapshot of all known standardized fault-report states."""
+        return dict(self._fault_tracker.states)
+
     @staticmethod
     def validate_and_normalize_message(raw: Any) -> DiagnosticsMessage:
         """Validate and normalize raw JSON-decoded payload."""
@@ -183,11 +194,25 @@ def _publish_state(self) -> None:
         for sink in self._sinks:
             sink.publish_state(snapshot)
 
+    def _publish_fault_updates(self, transitions: list[FaultTransition]) -> None:
+        if not transitions:
+            return
+        states = dict(self._fault_tracker.states)
+        for sink in self._sinks:
+            handler = getattr(sink, "handle_fault_transitions", None)
+            if callable(handler):
+                handler(transitions, states)
+
     def process_message(self, message: DiagnosticsMessage, now: float | None = None) -> bool:
         """Process one normalized diagnostics message."""
         recv_time = now if now is not None else self._time_fn()
         self.state_cache[message.node] = message
         self._last_seen[message.node] = recv_time
+        transitions = self._fault_tracker.update(message, recv_time)
+
+        # Transition-aware sinks update their per-source summary before the
+        # corresponding /fault snapshot is serialized.
+        self._publish_fault_updates(transitions)
         for sink in self._sinks:
             sink.handle_message(message)
         self._publish_state()

python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py

@@ -0,0 +1,181 @@
+"""Fault lifecycle tracking for normalized fault diagnostics messages."""
+
+from __future__ import annotations
+
+import math
+import re
+from dataclasses import dataclass
+from typing import Any, Iterable
+
+
+_FAULT_COUNT_PATTERN = re.compile(r"[0-9]+(?:\.0+)?")
+
+
+def parse_fault_count(value: Any) -> int | None:
+    """Parse a non-negative whole-number count from a ROS key-value string.
+
+    ROS bridges commonly format numeric diagnostic values as ``"1.000000"``.
+    Accept that representation while rejecting fractional, signed, and non-finite
+    values so a count remains an authoritative cardinality declaration.
+    """
+
+    text = str(value).strip()
+    if not _FAULT_COUNT_PATTERN.fullmatch(text):
+        return None
+    return int(text.partition(".")[0], 10)
+
+
+@dataclass(frozen=True)
+class FaultKey:
+    """Stable identity for one standardized fault report."""
+
+    hw_id: str
+    node: str
+    report: str
+
+
+@dataclass(frozen=True)
+class FaultState:
+    """Current lifecycle state for one standardized fault report."""
+
+    key: FaultKey
+    active: bool
+    level: int
+    message: str
+    first_raised: float
+    last_raised: float
+    last_cleared: float | None
+    occurrence_count: int
+
+
+@dataclass(frozen=True)
+class FaultTransition:
+    """A fault state transition emitted by :class:`FaultLifecycleTracker`."""
+
+    kind: str
+    state: FaultState
+    stamp: float
+
+
+class FaultLifecycleTracker:
+    """Track faults published through the string-native fault contract.
+
+    Only nodes whose final path segment is ``fault`` participate. A valid
+    message contains exactly one ``fault_count`` value and zero or more repeated
+    ``fault_report`` values. Reports are the complete active set, not deltas.
+    """
+
+    def __init__(self) -> None:
+        self.states: dict[FaultKey, FaultState] = {}
+        self._active_by_source: dict[tuple[str, str], set[str]] = {}
+
+    def update(self, message: Any, recv_time: float) -> list[FaultTransition]:
+        if not self._is_fault_node(message.node) or message.level == 3:
+            return []
+
+        reports, declared_count, valid = self._extract_reports(message.values)
+        if not valid or declared_count != len(reports):
+            return []
+        if reports and message.level not in (1, 2):
+            return []
+        if not reports and message.level != 0:
+            return []
+
+        source = (message.hw_id or "unknown", message.node)
+        previous_reports = set(self._active_by_source.get(source, set()))
+        stamp = self._event_stamp(message.stamp, recv_time)
+        transitions: list[FaultTransition] = []
+
+        for report in sorted(previous_reports - reports):
+            key = FaultKey(source[0], source[1], report)
+            previous = self.states[key]
+            state = FaultState(
+                key=key,
+                active=False,
+                level=0,
+                message=message.msg,
+                first_raised=previous.first_raised,
+                last_raised=previous.last_raised,
+                last_cleared=stamp,
+                occurrence_count=previous.occurrence_count,
+            )
+            self.states[key] = state
+            transitions.append(FaultTransition("cleared", state, stamp))
+
+        for report in sorted(reports - previous_reports):
+            key = FaultKey(source[0], source[1], report)
+            previous = self.states.get(key)
+            state = FaultState(
+                key=key,
+                active=True,
+                level=message.level,
+                message=message.msg,
+                first_raised=stamp if previous is None else previous.first_raised,
+                last_raised=stamp,
+                last_cleared=None if previous is None else previous.last_cleared,
+                occurrence_count=1 if previous is None else previous.occurrence_count + 1,
+            )
+            self.states[key] = state
+            transitions.append(FaultTransition("raised", state, stamp))
+
+        for report in sorted(reports & previous_reports):
+            key = FaultKey(source[0], source[1], report)
+            previous = self.states[key]
+            self.states[key] = FaultState(
+                key=key,
+                active=True,
+                level=message.level,
+                message=message.msg,
+                first_raised=previous.first_raised,
+                last_raised=previous.last_raised,
+                last_cleared=previous.last_cleared,
+                occurrence_count=previous.occurrence_count,
+            )
+
+        self._active_by_source[source] = reports
+        return transitions
+
+    @staticmethod
+    def _is_fault_node(node: Any) -> bool:
+        parts = [part for part in str(node).split("/") if part]
+        return bool(parts) and parts[-1].lower() == "fault"
+
+    @staticmethod
+    def _event_stamp(source_stamp: Any, recv_time: float) -> float:
+        try:
+            stamp = float(source_stamp)
+        except (TypeError, ValueError):
+            return recv_time
+        return stamp if math.isfinite(stamp) and stamp > 0.0 else recv_time
+
+    @classmethod
+    def _extract_reports(
+        cls, values: Iterable[Any]
+    ) -> tuple[set[str], int | None, bool]:
+        reports: set[str] = set()
+        counts: list[int] = []
+
+        for item in values:
+            key = str(item.key).strip().lower()
+            if key == "fault_report":
+                report = cls._normalize_report(item.value)
+                if report is None:
+                    # Fixed-size publisher slots are represented by an empty
+                    # ROS string when no fault is active.  Treat that as an
+                    # omitted report; a non-zero count still fails below.
+                    continue
+                reports.add(report)
+            elif key == "fault_count":
+                count = parse_fault_count(item.value)
+                if count is None:
+                    return set(), None, False
+                counts.append(count)
+
+        if len(counts) != 1:
+            return set(), None, False
+        return reports, counts[0], True
+
+    @staticmethod
+    def _normalize_report(value: Any) -> str | None:
+        report = str(value).strip()
+        return report if report else None