From 332c59ed5f3753008373fa1bc5c0e5f3ece3c90e Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Wed, 5 Aug 2026 23:41:22 +0200 Subject: [PATCH 01/12] add fault lifecycle tracking --- .../aggregator/aggregator.py | 22 ++ .../aggregator/fault_tracker.py | 216 ++++++++++++++++++ tests/test_fault_tracker.py | 179 +++++++++++++++ 3 files changed, 417 insertions(+) create mode 100644 python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py create mode 100644 tests/test_fault_tracker.py diff --git a/python/src/pyxbot2_diagnostics/aggregator/aggregator.py b/python/src/pyxbot2_diagnostics/aggregator/aggregator.py index 119d83d..d19d70c 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/aggregator.py +++ b/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 coded fault lifecycle states.""" + return dict(self._fault_tracker.states) + @staticmethod def validate_and_normalize_message(raw: Any) -> DiagnosticsMessage: """Validate and normalize raw JSON-decoded payload.""" @@ -183,13 +194,24 @@ 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) for sink in self._sinks: sink.handle_message(message) + self._publish_fault_updates(transitions) self._publish_state() return True diff --git a/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py new file mode 100644 index 0000000..cd1485b --- /dev/null +++ b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py @@ -0,0 +1,216 @@ +"""Fault lifecycle tracking for normalized diagnostics messages.""" + +from __future__ import annotations + +import math +from dataclasses import dataclass +from typing import Any, Iterable + + +_SINGLE_CODE_KEYS = {"fault_code", "error_code"} +_MULTI_CODE_KEYS = { + "fault_codes", + "error_codes", + "active_fault_codes", + "active_error_codes", +} +_ACTIVE_KEYS = {"fault_active", "error_active"} + + +@dataclass(frozen=True) +class FaultKey: + """Stable identity for one device fault.""" + + hw_id: str + node: str + code: str + + +@dataclass(frozen=True) +class FaultState: + """Current lifecycle state for one device fault.""" + + 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 raise/clear transitions for diagnostic error codes. + + The tracker recognizes the value keys ``fault_code`` and ``error_code`` for + a single code, and their plural/``active_*`` variants for multiple active + codes. Diagnostic levels 1 and 2 imply active faults, level 0 clears the + current source faults, and level 3 (STALE) deliberately leaves hardware + fault state unchanged. + """ + + 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]: + """Consume one normalized diagnostics message and return transitions.""" + if message.level == 3: + return [] + + source = (message.hw_id or "unknown", message.node) + codes, explicit_active = self._extract_codes(message.values) + previous_codes = set(self._active_by_source.get(source, set())) + + if explicit_active is False or message.level == 0: + current_codes: set[str] = set() + elif explicit_active is True or message.level in (1, 2): + current_codes = codes + else: + current_codes = previous_codes + + # Messages without a recognizable code cannot create a stable fault + # identity. They can still clear previously active coded faults on OK. + if not codes and message.level in (1, 2): + return [] + + stamp = self._event_stamp(message.stamp, recv_time) + transitions: list[FaultTransition] = [] + + for code in sorted(previous_codes - current_codes): + key = FaultKey(source[0], source[1], code) + 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 code in sorted(current_codes - previous_codes): + key = FaultKey(source[0], source[1], code) + previous = self.states.get(key) + count = 1 if previous is None else previous.occurrence_count + 1 + first_raised = stamp if previous is None else previous.first_raised + last_cleared = None if previous is None else previous.last_cleared + state = FaultState( + key=key, + active=True, + level=message.level, + message=message.msg, + first_raised=first_raised, + last_raised=stamp, + last_cleared=last_cleared, + occurrence_count=count, + ) + self.states[key] = state + transitions.append(FaultTransition("raised", state, stamp)) + + for code in sorted(current_codes & previous_codes): + key = FaultKey(source[0], source[1], code) + 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] = current_codes + return transitions + + @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_codes(cls, values: Iterable[Any]) -> tuple[set[str], bool | None]: + codes: set[str] = set() + explicit_active: bool | None = None + + for item in values: + key = str(item.key).strip().lower() + value = item.value + if key in _SINGLE_CODE_KEYS: + code = cls._normalize_code(value) + if code is not None: + codes.add(code) + elif key in _MULTI_CODE_KEYS: + for raw_code in cls._iter_codes(value): + code = cls._normalize_code(raw_code) + if code is not None: + codes.add(code) + elif key in _ACTIVE_KEYS: + explicit_active = cls._as_bool(value) + + return codes, explicit_active + + @staticmethod + def _iter_codes(value: Any) -> Iterable[Any]: + if isinstance(value, (list, tuple, set)): + return value + if isinstance(value, str): + return [part.strip() for part in value.split(",") if part.strip()] + return [value] + + @staticmethod + def _normalize_code(value: Any) -> str | None: + if value is None or isinstance(value, bool): + return None + if isinstance(value, int): + if value == 0: + return None + return f"0x{value:X}" + if isinstance(value, float) and value.is_integer(): + integer = int(value) + if integer == 0: + return None + return f"0x{integer:X}" + + text = str(value).strip() + if not text or text.lower() in {"0", "0x0", "0x0000", "none", "ok"}: + return None + if text.lower().startswith("0x"): + try: + return f"0x{int(text, 16):X}" + except ValueError: + pass + return text + + @staticmethod + def _as_bool(value: Any) -> bool | None: + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"true", "yes", "on", "1"}: + return True + if normalized in {"false", "no", "off", "0"}: + return False + return None diff --git a/tests/test_fault_tracker.py b/tests/test_fault_tracker.py new file mode 100644 index 0000000..f566340 --- /dev/null +++ b/tests/test_fault_tracker.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + +from pyxbot2_diagnostics.aggregator.aggregator import ( + DiagnosticKeyValue, + DiagnosticsAggregator, + DiagnosticsMessage, +) +from pyxbot2_diagnostics.aggregator.config import ( + AggregatorConfig, + AggregatorSection, + SinksSection, +) +from pyxbot2_diagnostics.aggregator.fault_tracker import FaultLifecycleTracker + + +def _message( + *, + level: int, + values: tuple[DiagnosticKeyValue, ...], + stamp: float = 10.0, + msg: str = "fault", +) -> DiagnosticsMessage: + return DiagnosticsMessage( + v=1, + node="xbot/joint/knee_pitch_1/drive_fault", + hw_id="knee_pitch_1", + stamp=stamp, + level=level, + msg=msg, + values=values, + ) + + +def test_single_fault_raise_duplicate_and_clear() -> None: + tracker = FaultLifecycleTracker() + fault = _message( + level=2, + values=(DiagnosticKeyValue("error_code", "0x4210"),), + ) + + transitions = tracker.update(fault, recv_time=11.0) + assert [item.kind for item in transitions] == ["raised"] + state = transitions[0].state + assert state.key.code == "0x4210" + assert state.active + assert state.last_raised == 10.0 + assert state.occurrence_count == 1 + + assert tracker.update(fault, recv_time=12.0) == [] + assert next(iter(tracker.states.values())).occurrence_count == 1 + + cleared = tracker.update( + _message(level=0, values=(), stamp=20.0, msg="OK"), recv_time=21.0 + ) + assert [item.kind for item in cleared] == ["cleared"] + assert not cleared[0].state.active + assert cleared[0].state.last_cleared == 20.0 + + +def test_code_change_clears_old_and_raises_new() -> None: + tracker = FaultLifecycleTracker() + tracker.update( + _message(level=2, values=(DiagnosticKeyValue("fault_code", "0x4210"),)), + recv_time=10.0, + ) + + transitions = tracker.update( + _message( + level=2, + stamp=30.0, + values=(DiagnosticKeyValue("fault_code", "0x7500"),), + ), + recv_time=31.0, + ) + + assert [(item.kind, item.state.key.code) for item in transitions] == [ + ("cleared", "0x4210"), + ("raised", "0x7500"), + ] + + +def test_multiple_simultaneous_faults_are_diffed_as_sets() -> None: + tracker = FaultLifecycleTracker() + first = tracker.update( + _message( + level=2, + values=(DiagnosticKeyValue("active_error_codes", [0x4210, "0x7500"]),), + ), + recv_time=10.0, + ) + assert {(item.kind, item.state.key.code) for item in first} == { + ("raised", "0x4210"), + ("raised", "0x7500"), + } + + second = tracker.update( + _message( + level=2, + stamp=40.0, + values=(DiagnosticKeyValue("active_error_codes", ["0x7500", "0x8611"]),), + ), + recv_time=41.0, + ) + assert {(item.kind, item.state.key.code) for item in second} == { + ("cleared", "0x4210"), + ("raised", "0x8611"), + } + + +def test_stale_does_not_clear_hardware_faults() -> None: + tracker = FaultLifecycleTracker() + tracker.update( + _message(level=2, values=(DiagnosticKeyValue("error_code", "0x4210"),)), + recv_time=10.0, + ) + + assert tracker.update( + _message( + level=3, + stamp=50.0, + msg="STALE", + values=(DiagnosticKeyValue("error_code", "0x4210"),), + ), + recv_time=50.0, + ) == [] + assert next(iter(tracker.states.values())).active + + +class NullSource: + def poll(self, timeout_ms: int = 100): + del timeout_ms + return [] + + def close(self) -> None: + return + + +@dataclass +class FaultSink: + transitions: list[object] = field(default_factory=list) + state_snapshots: list[dict[object, object]] = field(default_factory=list) + + def handle_message(self, message) -> None: + del message + + def handle_fault_transitions(self, transitions, states) -> None: + self.transitions.extend(transitions) + self.state_snapshots.append(dict(states)) + + def publish_state(self, states) -> None: + del states + + def close(self) -> None: + return + + +def test_aggregator_publishes_fault_transitions_to_opt_in_sink() -> None: + config = AggregatorConfig( + aggregator=AggregatorSection( + zmq_endpoint="inproc://unused", + stale_timeout_sec=5.0, + stale_check_interval_sec=1.0, + ), + sinks=SinksSection(), + ) + sink = FaultSink() + aggregator = DiagnosticsAggregator(config, [sink], sources=[NullSource()]) + + aggregator.process_message( + _message(level=2, values=(DiagnosticKeyValue("error_code", "0x4210"),)), + now=11.0, + ) + + assert len(sink.transitions) == 1 + assert sink.transitions[0].kind == "raised" + assert next(iter(aggregator.fault_states.values())).active + aggregator.close() From e0fac712dd793da5143379c4e8c1e33ec6154937 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Wed, 5 Aug 2026 23:45:06 +0200 Subject: [PATCH 02/12] define health fault message contract --- .../aggregator/fault_tracker.py | 113 ++++++++---------- 1 file changed, 53 insertions(+), 60 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py index cd1485b..0c9967d 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py +++ b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py @@ -1,4 +1,4 @@ -"""Fault lifecycle tracking for normalized diagnostics messages.""" +"""Fault lifecycle tracking for normalized health diagnostics messages.""" from __future__ import annotations @@ -7,14 +7,19 @@ from typing import Any, Iterable -_SINGLE_CODE_KEYS = {"fault_code", "error_code"} -_MULTI_CODE_KEYS = { - "fault_codes", +# Fault lifecycle contract: +# - only diagnostic paths ending in /health are considered; +# - fault_codes is the canonical required value key; +# - an empty fault_codes collection means that all previous faults are cleared; +# - levels 1/2 describe active faults, level 0 describes no active faults; +# - level 3 is transport staleness and must not change hardware fault state. +_CANONICAL_CODE_KEY = "fault_codes" +_LEGACY_SINGLE_CODE_KEYS = {"fault_code", "error_code"} +_LEGACY_MULTI_CODE_KEYS = { "error_codes", "active_fault_codes", "active_error_codes", } -_ACTIVE_KEYS = {"fault_active", "error_active"} @dataclass(frozen=True) @@ -50,13 +55,16 @@ class FaultTransition: class FaultLifecycleTracker: - """Track raise/clear transitions for diagnostic error codes. + """Track coded faults published through the health-message contract. - The tracker recognizes the value keys ``fault_code`` and ``error_code`` for - a single code, and their plural/``active_*`` variants for multiple active - codes. Diagnostic levels 1 and 2 imply active faults, level 0 clears the - current source faults, and level 3 (STALE) deliberately leaves hardware - fault state unchanged. + Canonical publishers use a node path ending in ``/health`` and include one + ``fault_codes`` value. Its value is the complete set of faults active at the + message timestamp. An empty collection clears faults previously reported by + that health source. + + ``fault_code``, ``error_code``, ``error_codes``, ``active_fault_codes`` and + ``active_error_codes`` are accepted as compatibility aliases, but new + publishers should only emit ``fault_codes``. """ def __init__(self) -> None: @@ -64,24 +72,25 @@ def __init__(self) -> None: self._active_by_source: dict[tuple[str, str], set[str]] = {} def update(self, message: Any, recv_time: float) -> list[FaultTransition]: - """Consume one normalized diagnostics message and return transitions.""" - if message.level == 3: + """Consume one normalized message and return fault transitions.""" + if not self._is_health_node(message.node) or message.level == 3: + return [] + + codes, has_code_key = self._extract_codes(message.values) + if not has_code_key: + # A /health message without the required code set is malformed for + # lifecycle purposes. Ignoring it is safer than clearing state. return [] source = (message.hw_id or "unknown", message.node) - codes, explicit_active = self._extract_codes(message.values) previous_codes = set(self._active_by_source.get(source, set())) + current_codes = codes - if explicit_active is False or message.level == 0: - current_codes: set[str] = set() - elif explicit_active is True or message.level in (1, 2): - current_codes = codes - else: - current_codes = previous_codes - - # Messages without a recognizable code cannot create a stable fault - # identity. They can still clear previously active coded faults on OK. - if not codes and message.level in (1, 2): + # fault_codes is the source of truth. Level conveys severity only. + # Inconsistent messages are ignored to avoid false raises or clears. + if current_codes and message.level not in (1, 2): + return [] + if not current_codes and message.level != 0: return [] stamp = self._event_stamp(message.stamp, recv_time) @@ -106,18 +115,15 @@ def update(self, message: Any, recv_time: float) -> list[FaultTransition]: for code in sorted(current_codes - previous_codes): key = FaultKey(source[0], source[1], code) previous = self.states.get(key) - count = 1 if previous is None else previous.occurrence_count + 1 - first_raised = stamp if previous is None else previous.first_raised - last_cleared = None if previous is None else previous.last_cleared state = FaultState( key=key, active=True, level=message.level, message=message.msg, - first_raised=first_raised, + first_raised=stamp if previous is None else previous.first_raised, last_raised=stamp, - last_cleared=last_cleared, - occurrence_count=count, + 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)) @@ -139,6 +145,11 @@ def update(self, message: Any, recv_time: float) -> list[FaultTransition]: self._active_by_source[source] = current_codes return transitions + @staticmethod + def _is_health_node(node: Any) -> bool: + parts = [part for part in str(node).split("/") if part] + return bool(parts) and parts[-1].lower() == "health" + @staticmethod def _event_stamp(source_stamp: Any, recv_time: float) -> float: try: @@ -148,26 +159,26 @@ def _event_stamp(source_stamp: Any, recv_time: float) -> float: return stamp if math.isfinite(stamp) and stamp > 0.0 else recv_time @classmethod - def _extract_codes(cls, values: Iterable[Any]) -> tuple[set[str], bool | None]: + def _extract_codes(cls, values: Iterable[Any]) -> tuple[set[str], bool]: codes: set[str] = set() - explicit_active: bool | None = None + has_code_key = False for item in values: key = str(item.key).strip().lower() value = item.value - if key in _SINGLE_CODE_KEYS: - code = cls._normalize_code(value) - if code is not None: - codes.add(code) - elif key in _MULTI_CODE_KEYS: + if key == _CANONICAL_CODE_KEY or key in _LEGACY_MULTI_CODE_KEYS: + has_code_key = True for raw_code in cls._iter_codes(value): code = cls._normalize_code(raw_code) if code is not None: codes.add(code) - elif key in _ACTIVE_KEYS: - explicit_active = cls._as_bool(value) + elif key in _LEGACY_SINGLE_CODE_KEYS: + has_code_key = True + code = cls._normalize_code(value) + if code is not None: + codes.add(code) - return codes, explicit_active + return codes, has_code_key @staticmethod def _iter_codes(value: Any) -> Iterable[Any]: @@ -182,14 +193,10 @@ def _normalize_code(value: Any) -> str | None: if value is None or isinstance(value, bool): return None if isinstance(value, int): - if value == 0: - return None - return f"0x{value:X}" + return None if value == 0 else f"0x{value:X}" if isinstance(value, float) and value.is_integer(): integer = int(value) - if integer == 0: - return None - return f"0x{integer:X}" + return None if integer == 0 else f"0x{integer:X}" text = str(value).strip() if not text or text.lower() in {"0", "0x0", "0x0000", "none", "ok"}: @@ -200,17 +207,3 @@ def _normalize_code(value: Any) -> str | None: except ValueError: pass return text - - @staticmethod - def _as_bool(value: Any) -> bool | None: - if isinstance(value, bool): - return value - if isinstance(value, (int, float)): - return bool(value) - if isinstance(value, str): - normalized = value.strip().lower() - if normalized in {"true", "yes", "on", "1"}: - return True - if normalized in {"false", "no", "off", "0"}: - return False - return None From 3160920eb92a51cb81b46b5a63560e96dd2d3b67 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Wed, 5 Aug 2026 23:45:24 +0200 Subject: [PATCH 03/12] test health fault message contract --- tests/test_fault_tracker.py | 64 +++++++++++++++++++++++++++++++------ 1 file changed, 54 insertions(+), 10 deletions(-) diff --git a/tests/test_fault_tracker.py b/tests/test_fault_tracker.py index f566340..60d90db 100644 --- a/tests/test_fault_tracker.py +++ b/tests/test_fault_tracker.py @@ -21,10 +21,11 @@ def _message( values: tuple[DiagnosticKeyValue, ...], stamp: float = 10.0, msg: str = "fault", + node: str = "/xbot/joint/knee_pitch_1/health", ) -> DiagnosticsMessage: return DiagnosticsMessage( v=1, - node="xbot/joint/knee_pitch_1/drive_fault", + node=node, hw_id="knee_pitch_1", stamp=stamp, level=level, @@ -37,7 +38,7 @@ def test_single_fault_raise_duplicate_and_clear() -> None: tracker = FaultLifecycleTracker() fault = _message( level=2, - values=(DiagnosticKeyValue("error_code", "0x4210"),), + values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),), ) transitions = tracker.update(fault, recv_time=11.0) @@ -52,7 +53,13 @@ def test_single_fault_raise_duplicate_and_clear() -> None: assert next(iter(tracker.states.values())).occurrence_count == 1 cleared = tracker.update( - _message(level=0, values=(), stamp=20.0, msg="OK"), recv_time=21.0 + _message( + level=0, + values=(DiagnosticKeyValue("fault_codes", []),), + stamp=20.0, + msg="OK", + ), + recv_time=21.0, ) assert [item.kind for item in cleared] == ["cleared"] assert not cleared[0].state.active @@ -62,7 +69,7 @@ def test_single_fault_raise_duplicate_and_clear() -> None: def test_code_change_clears_old_and_raises_new() -> None: tracker = FaultLifecycleTracker() tracker.update( - _message(level=2, values=(DiagnosticKeyValue("fault_code", "0x4210"),)), + _message(level=2, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), recv_time=10.0, ) @@ -70,7 +77,7 @@ def test_code_change_clears_old_and_raises_new() -> None: _message( level=2, stamp=30.0, - values=(DiagnosticKeyValue("fault_code", "0x7500"),), + values=(DiagnosticKeyValue("fault_codes", ["0x7500"]),), ), recv_time=31.0, ) @@ -86,7 +93,7 @@ def test_multiple_simultaneous_faults_are_diffed_as_sets() -> None: first = tracker.update( _message( level=2, - values=(DiagnosticKeyValue("active_error_codes", [0x4210, "0x7500"]),), + values=(DiagnosticKeyValue("fault_codes", [0x4210, "0x7500"]),), ), recv_time=10.0, ) @@ -99,7 +106,7 @@ def test_multiple_simultaneous_faults_are_diffed_as_sets() -> None: _message( level=2, stamp=40.0, - values=(DiagnosticKeyValue("active_error_codes", ["0x7500", "0x8611"]),), + values=(DiagnosticKeyValue("fault_codes", ["0x7500", "0x8611"]),), ), recv_time=41.0, ) @@ -112,7 +119,7 @@ def test_multiple_simultaneous_faults_are_diffed_as_sets() -> None: def test_stale_does_not_clear_hardware_faults() -> None: tracker = FaultLifecycleTracker() tracker.update( - _message(level=2, values=(DiagnosticKeyValue("error_code", "0x4210"),)), + _message(level=2, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), recv_time=10.0, ) @@ -121,13 +128,50 @@ def test_stale_does_not_clear_hardware_faults() -> None: level=3, stamp=50.0, msg="STALE", - values=(DiagnosticKeyValue("error_code", "0x4210"),), + values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),), ), recv_time=50.0, ) == [] assert next(iter(tracker.states.values())).active +def test_non_health_message_is_not_interpreted_as_fault_contract() -> None: + tracker = FaultLifecycleTracker() + transitions = tracker.update( + _message( + level=2, + node="/xbot/joint/knee_pitch_1/temperature", + values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),), + ), + recv_time=10.0, + ) + assert transitions == [] + assert tracker.states == {} + + +def test_health_message_requires_fault_codes_key() -> None: + tracker = FaultLifecycleTracker() + transitions = tracker.update( + _message(level=2, values=(DiagnosticKeyValue("temperature", 90.0),)), + recv_time=10.0, + ) + assert transitions == [] + assert tracker.states == {} + + +def test_inconsistent_level_and_fault_codes_is_ignored() -> None: + tracker = FaultLifecycleTracker() + assert tracker.update( + _message(level=0, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), + recv_time=10.0, + ) == [] + assert tracker.update( + _message(level=2, values=(DiagnosticKeyValue("fault_codes", []),)), + recv_time=10.0, + ) == [] + assert tracker.states == {} + + class NullSource: def poll(self, timeout_ms: int = 100): del timeout_ms @@ -169,7 +213,7 @@ def test_aggregator_publishes_fault_transitions_to_opt_in_sink() -> None: aggregator = DiagnosticsAggregator(config, [sink], sources=[NullSource()]) aggregator.process_message( - _message(level=2, values=(DiagnosticKeyValue("error_code", "0x4210"),)), + _message(level=2, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), now=11.0, ) From 48e65818d43401d7c83aa28b23cf3cb0703c007f Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Wed, 5 Aug 2026 23:45:43 +0200 Subject: [PATCH 04/12] document health fault contract --- docs/fault_health_contract.md | 99 +++++++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 docs/fault_health_contract.md diff --git a/docs/fault_health_contract.md b/docs/fault_health_contract.md new file mode 100644 index 0000000..3abe859 --- /dev/null +++ b/docs/fault_health_contract.md @@ -0,0 +1,99 @@ +# Fault health diagnostic contract + +The aggregator interprets a diagnostic message as a fault-state source only when its normalized `node` path ends in `/health`. + +Examples of health-source paths: + +- `/xbot/joint/knee_pitch_1/health` +- `/xbot/power/battery/health` +- `host/robot-pc/network/eth0/health` + +Messages with other suffixes remain ordinary diagnostics and are not inspected for fault lifecycle information, even if they contain similarly named values. + +## Required values + +Every `/health` message must contain exactly one canonical key-value entry named `fault_codes`. + +```json +{"key": "fault_codes", "value": ["0x4210", "0x7500"]} +``` + +`fault_codes` is the complete set of fault codes active for that health source at the message timestamp. It is not a delta and must not contain only newly raised faults. + +The value should be an array. Each code may be an integer or a stable string identifier. Integer codes and hexadecimal strings are normalized to uppercase hexadecimal strings by the aggregator. Zero-like values (`0`, `0x0000`, `none`, `ok`) are ignored and must not be used as real fault identifiers. + +An empty array explicitly means that the source has no active faults: + +```json +{"key": "fault_codes", "value": []} +``` + +A `/health` message without `fault_codes` is invalid for lifecycle tracking and is ignored. It does not clear previously active faults. + +## Level and message semantics + +The ROS diagnostics level must agree with `fault_codes`: + +| `fault_codes` | `level` | Meaning | +|---|---:|---| +| empty | `0` | Healthy; clear all faults previously reported by this source | +| non-empty | `1` | One or more warning-level faults are active | +| non-empty | `2` | One or more error-level faults are active | +| unchanged/any | `3` | Source is stale; do not raise or clear hardware faults | + +Inconsistent combinations, such as non-empty `fault_codes` with level `0`, are ignored for lifecycle tracking. + +`msg` is a human-readable summary for dashboards and logs. It is not part of fault identity and must not be parsed to determine active faults. + +`hw_id` identifies the physical device. A fault lifecycle is keyed by: + +```text +(hw_id, node, fault_code) +``` + +## Examples + +Active faults: + +```json +{ + "v": 1, + "node": "/xbot/joint/knee_pitch_1/health", + "hw_id": "knee_pitch_1", + "stamp": 1785967012.0, + "level": 2, + "msg": "Drive reports over-temperature and communication faults", + "values": [ + {"key": "fault_codes", "value": ["0x4210", "0x7500"]} + ] +} +``` + +Healthy/cleared: + +```json +{ + "v": 1, + "node": "/xbot/joint/knee_pitch_1/health", + "hw_id": "knee_pitch_1", + "stamp": 1785967305.0, + "level": 0, + "msg": "OK", + "values": [ + {"key": "fault_codes", "value": []} + ] +} +``` + +## Compatibility aliases + +For migration, the tracker currently accepts these aliases: + +- single-code aliases: `fault_code`, `error_code` +- multi-code aliases: `error_codes`, `active_fault_codes`, `active_error_codes` + +New publishers must use `fault_codes`. Compatibility aliases may be removed in a future schema version. + +## Design rationale + +The path suffix provides an explicit namespace boundary so arbitrary telemetry cannot accidentally create or clear faults. A complete active-code set makes updates idempotent and allows the aggregator to compute raises and clears by set difference. It also supports devices that report one current code and devices that report multiple simultaneous codes without changing the storage model. From ac89162ab3083832c23f89c4df8a137bbd8bd2d3 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 00:01:25 +0200 Subject: [PATCH 05/12] use string-native fault report contract --- .../aggregator/fault_tracker.py | 136 ++++++------------ 1 file changed, 45 insertions(+), 91 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py index 0c9967d..6d01ef0 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py +++ b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py @@ -7,33 +7,18 @@ from typing import Any, Iterable -# Fault lifecycle contract: -# - only diagnostic paths ending in /health are considered; -# - fault_codes is the canonical required value key; -# - an empty fault_codes collection means that all previous faults are cleared; -# - levels 1/2 describe active faults, level 0 describes no active faults; -# - level 3 is transport staleness and must not change hardware fault state. -_CANONICAL_CODE_KEY = "fault_codes" -_LEGACY_SINGLE_CODE_KEYS = {"fault_code", "error_code"} -_LEGACY_MULTI_CODE_KEYS = { - "error_codes", - "active_fault_codes", - "active_error_codes", -} - - @dataclass(frozen=True) class FaultKey: - """Stable identity for one device fault.""" + """Stable identity for one standardized fault report.""" hw_id: str node: str - code: str + report: str @dataclass(frozen=True) class FaultState: - """Current lifecycle state for one device fault.""" + """Current lifecycle state for one standardized fault report.""" key: FaultKey active: bool @@ -55,16 +40,11 @@ class FaultTransition: class FaultLifecycleTracker: - """Track coded faults published through the health-message contract. - - Canonical publishers use a node path ending in ``/health`` and include one - ``fault_codes`` value. Its value is the complete set of faults active at the - message timestamp. An empty collection clears faults previously reported by - that health source. + """Track faults published through the string-native health contract. - ``fault_code``, ``error_code``, ``error_codes``, ``active_fault_codes`` and - ``active_error_codes`` are accepted as compatibility aliases, but new - publishers should only emit ``fault_codes``. + Only nodes whose final path segment is ``health`` 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: @@ -72,32 +52,24 @@ def __init__(self) -> None: self._active_by_source: dict[tuple[str, str], set[str]] = {} def update(self, message: Any, recv_time: float) -> list[FaultTransition]: - """Consume one normalized message and return fault transitions.""" if not self._is_health_node(message.node) or message.level == 3: return [] - codes, has_code_key = self._extract_codes(message.values) - if not has_code_key: - # A /health message without the required code set is malformed for - # lifecycle purposes. Ignoring it is safer than clearing state. + reports, declared_count, valid = self._extract_reports(message.values) + if not valid or declared_count != len(reports): return [] - - source = (message.hw_id or "unknown", message.node) - previous_codes = set(self._active_by_source.get(source, set())) - current_codes = codes - - # fault_codes is the source of truth. Level conveys severity only. - # Inconsistent messages are ignored to avoid false raises or clears. - if current_codes and message.level not in (1, 2): + if reports and message.level not in (1, 2): return [] - if not current_codes and message.level != 0: + 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 code in sorted(previous_codes - current_codes): - key = FaultKey(source[0], source[1], code) + for report in sorted(previous_reports - reports): + key = FaultKey(source[0], source[1], report) previous = self.states[key] state = FaultState( key=key, @@ -112,8 +84,8 @@ def update(self, message: Any, recv_time: float) -> list[FaultTransition]: self.states[key] = state transitions.append(FaultTransition("cleared", state, stamp)) - for code in sorted(current_codes - previous_codes): - key = FaultKey(source[0], source[1], code) + for report in sorted(reports - previous_reports): + key = FaultKey(source[0], source[1], report) previous = self.states.get(key) state = FaultState( key=key, @@ -128,8 +100,8 @@ def update(self, message: Any, recv_time: float) -> list[FaultTransition]: self.states[key] = state transitions.append(FaultTransition("raised", state, stamp)) - for code in sorted(current_codes & previous_codes): - key = FaultKey(source[0], source[1], code) + for report in sorted(reports & previous_reports): + key = FaultKey(source[0], source[1], report) previous = self.states[key] self.states[key] = FaultState( key=key, @@ -142,7 +114,7 @@ def update(self, message: Any, recv_time: float) -> list[FaultTransition]: occurrence_count=previous.occurrence_count, ) - self._active_by_source[source] = current_codes + self._active_by_source[source] = reports return transitions @staticmethod @@ -159,51 +131,33 @@ def _event_stamp(source_stamp: Any, recv_time: float) -> float: return stamp if math.isfinite(stamp) and stamp > 0.0 else recv_time @classmethod - def _extract_codes(cls, values: Iterable[Any]) -> tuple[set[str], bool]: - codes: set[str] = set() - has_code_key = False + 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() - value = item.value - if key == _CANONICAL_CODE_KEY or key in _LEGACY_MULTI_CODE_KEYS: - has_code_key = True - for raw_code in cls._iter_codes(value): - code = cls._normalize_code(raw_code) - if code is not None: - codes.add(code) - elif key in _LEGACY_SINGLE_CODE_KEYS: - has_code_key = True - code = cls._normalize_code(value) - if code is not None: - codes.add(code) - - return codes, has_code_key - - @staticmethod - def _iter_codes(value: Any) -> Iterable[Any]: - if isinstance(value, (list, tuple, set)): - return value - if isinstance(value, str): - return [part.strip() for part in value.split(",") if part.strip()] - return [value] + if key == "fault_report": + report = cls._normalize_report(item.value) + if report is None: + return set(), None, False + reports.add(report) + elif key == "fault_count": + try: + count = int(str(item.value).strip(), 10) + except (TypeError, ValueError): + return set(), None, False + if count < 0: + return set(), None, False + counts.append(count) + + if len(counts) != 1: + return set(), None, False + return reports, counts[0], True @staticmethod - def _normalize_code(value: Any) -> str | None: - if value is None or isinstance(value, bool): - return None - if isinstance(value, int): - return None if value == 0 else f"0x{value:X}" - if isinstance(value, float) and value.is_integer(): - integer = int(value) - return None if integer == 0 else f"0x{integer:X}" - - text = str(value).strip() - if not text or text.lower() in {"0", "0x0", "0x0000", "none", "ok"}: - return None - if text.lower().startswith("0x"): - try: - return f"0x{int(text, 16):X}" - except ValueError: - pass - return text + def _normalize_report(value: Any) -> str | None: + report = str(value).strip() + return report if report else None From 71ac8eb6b5dfae353531c7f14c033fef316a9159 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 00:02:00 +0200 Subject: [PATCH 06/12] publish fault transitions before health snapshots --- python/src/pyxbot2_diagnostics/aggregator/aggregator.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/aggregator.py b/python/src/pyxbot2_diagnostics/aggregator/aggregator.py index d19d70c..b1d7ca4 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/aggregator.py +++ b/python/src/pyxbot2_diagnostics/aggregator/aggregator.py @@ -128,7 +128,7 @@ def __init__( @property def fault_states(self) -> dict[Any, FaultState]: - """Return a snapshot of all known coded fault lifecycle states.""" + """Return a snapshot of all known standardized fault-report states.""" return dict(self._fault_tracker.states) @staticmethod @@ -209,9 +209,12 @@ def process_message(self, message: DiagnosticsMessage, now: float | None = None) 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 /health snapshot is serialized. + self._publish_fault_updates(transitions) for sink in self._sinks: sink.handle_message(message) - self._publish_fault_updates(transitions) self._publish_state() return True From 71396dd9445a4ff53207534ef79df330b2c7bf40 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 00:02:21 +0200 Subject: [PATCH 07/12] write grafana friendly health and fault events --- .../aggregator/sinks/influxdb_sink.py | 178 +++++++++++++----- 1 file changed, 130 insertions(+), 48 deletions(-) diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py index 2cc2326..33e929b 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py @@ -1,34 +1,27 @@ -"""InfluxDB sink for diagnostics metric values.""" +"""InfluxDB sink for diagnostics, health snapshots, and fault events.""" from __future__ import annotations import logging +import math import time from typing import Any from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticsMessage LOGGER = logging.getLogger(__name__) - -# Single measurement name for all robot diagnostics. -_MEASUREMENT = "robot_diagnostics" - -# Minimum seconds between batch writes to InfluxDB. _FLUSH_INTERVAL_SEC = 1.0 class InfluxDBSink: """Write diagnostics to InfluxDB v2. - Schema - ------ - measurement : robot_diagnostics - tags : hw_id, path (full status name), name (last path segment) - fields : level (int), one float field per kv-pair in status.values, - message (str, only when non-empty) + Ordinary diagnostics keep the existing path-derived measurement schema. - Points are buffered in handle_message and flushed as a single batch write - at most once per _FLUSH_INTERVAL_SEC to avoid per-message HTTP overhead. + Health messages (paths ending in ``/health``) are written as one periodic + ``health`` snapshot per source. Fault transitions are written separately as + ``fault_event`` points. Standardized ``fault_report`` values are tags so + Grafana can filter and group by report efficiently. """ def __init__( @@ -48,6 +41,7 @@ def __init__( self._write_api = write_api self._pending: list[dict[str, Any]] = [] self._last_flush = 0.0 + self._last_fault_by_source: dict[tuple[str, str], Any] = {} if not enabled: return @@ -70,8 +64,6 @@ def __init__( return self._client = InfluxDBClient(url=url, token=token, org=org) - # SYNCHRONOUS so errors surface immediately rather than being silently - # dropped by the async batch queue. self._write_api = self._client.write_api(write_options=SYNCHRONOUS) LOGGER.info("InfluxDB sink enabled: url=%s bucket=%s org=%s", url, bucket, org) @@ -80,48 +72,139 @@ def handle_message(self, message: DiagnosticsMessage) -> None: return path = message.node - parts = [p for p in path.split("/") if p] - - fields: dict[str, Any] = {"level": message.level} - - # Coerce each kv-value to float; fall back to string for non-numeric ones. - for kv in message.values: - try: - fields[kv.key] = float(kv.value) - except (ValueError, TypeError): - fields[kv.key] = str(kv.value) - - if message.msg: - fields["message"] = message.msg - - """ - node schema is defined as follows: - // - - example: /xbot/joint/knee_pitch_1/pos_ref --> - component = /xbot/joint - name = knee_pitch_1 - measurement = pos_ref - """ - + parts = [part for part in path.split("/") if part] measurement = parts[-1] if parts else "unknown" name = parts[-2] if len(parts) >= 2 else measurement component = "/".join(parts[:-2]) + tags = { + "hw_id": message.hw_id if message.hw_id else "unknown", + "path": path, + "name": name, + "component": component, + } + + if measurement.lower() == "health": + fields = self._health_fields(message) + if fields is None: + # Preserve ordinary diagnostics export while avoiding a + # misleading lifecycle snapshot for malformed health data. + fields = self._generic_fields(message) + else: + fields = self._generic_fields(message) self._pending.append( { "measurement": measurement, - "tags": { - "hw_id": message.hw_id if message.hw_id else "unknown", - "path": path, - "name": name, - "component": component, - }, + "tags": tags, "fields": fields, - "time": int(1e9 * time.time()), + "time": self._timestamp_ns(message.stamp), } ) + def handle_fault_transitions(self, transitions, states) -> None: + del states + if not self._enabled or self._write_api is None: + return + + for transition in transitions: + state = transition.state + source = (state.key.hw_id, state.key.node) + self._last_fault_by_source[source] = state + + parts = [part for part in state.key.node.split("/") if part] + name = parts[-2] if len(parts) >= 2 else "health" + component = "/".join(parts[:-2]) + fields: dict[str, Any] = { + "active": state.active, + "level": state.level, + "message": state.message, + "occurrence_count": state.occurrence_count, + "first_raised_ns": int(1e9 * state.first_raised), + "last_raised_ns": int(1e9 * state.last_raised), + } + if state.last_cleared is not None: + fields["last_cleared_ns"] = int(1e9 * state.last_cleared) + + self._pending.append( + { + "measurement": "fault_event", + "tags": { + "hw_id": state.key.hw_id, + "path": state.key.node, + "name": name, + "component": component, + "fault_report": state.key.report, + "transition": transition.kind, + }, + "fields": fields, + "time": int(1e9 * transition.stamp), + } + ) + + def _health_fields(self, message: DiagnosticsMessage) -> dict[str, Any] | None: + reports: list[str] = [] + counts: list[int] = [] + fields: dict[str, Any] = {"level": message.level} + + for kv in message.values: + key = kv.key.strip().lower() + if key == "fault_report": + report = str(kv.value).strip() + if report: + reports.append(report) + elif key == "fault_count": + try: + counts.append(int(str(kv.value).strip(), 10)) + except ValueError: + return None + else: + self._add_generic_field(fields, kv.key, kv.value) + + if len(counts) != 1 or counts[0] != len(set(reports)): + return None + + fields["fault_count"] = counts[0] + fields["active_fault_reports"] = "; ".join(sorted(set(reports))) + if message.msg: + fields["message"] = message.msg + + source = (message.hw_id if message.hw_id else "unknown", message.node) + last_fault = self._last_fault_by_source.get(source) + if last_fault is not None: + fields["last_fault_report"] = last_fault.key.report + fields["last_fault_active"] = last_fault.active + fields["last_fault_level"] = last_fault.level + fields["last_raised_ns"] = int(1e9 * last_fault.last_raised) + if last_fault.last_cleared is not None: + fields["last_cleared_ns"] = int(1e9 * last_fault.last_cleared) + + return fields + + def _generic_fields(self, message: DiagnosticsMessage) -> dict[str, Any]: + fields: dict[str, Any] = {"level": message.level} + for kv in message.values: + self._add_generic_field(fields, kv.key, kv.value) + if message.msg: + fields["message"] = message.msg + return fields + + @staticmethod + def _add_generic_field(fields: dict[str, Any], key: str, value: Any) -> None: + try: + fields[key] = float(value) + except (ValueError, TypeError): + fields[key] = str(value) + + @staticmethod + def _timestamp_ns(stamp: Any) -> int: + try: + value = float(stamp) + except (TypeError, ValueError): + value = time.time() + if not math.isfinite(value) or value <= 0.0: + value = time.time() + return int(1e9 * value) + def publish_state(self, states: dict[str, DiagnosticsMessage]) -> None: del states self._flush() @@ -142,7 +225,6 @@ def _flush(self) -> None: LOGGER.warning("InfluxDB write failed (%d points dropped): %s", len(points), exc) def close(self) -> None: - # Final flush on shutdown — ignore the rate limit. if self._pending and self._enabled and self._write_api is not None: try: self._write_api.write(bucket=self._bucket, org=self._org, record=self._pending) From b1a79e4fd98c0706da8242db22ef648c492e73e6 Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 00:02:41 +0200 Subject: [PATCH 08/12] test repeated fault report contract --- tests/test_fault_tracker.py | 98 +++++++++++++++++++++++-------------- 1 file changed, 60 insertions(+), 38 deletions(-) diff --git a/tests/test_fault_tracker.py b/tests/test_fault_tracker.py index 60d90db..d359c2b 100644 --- a/tests/test_fault_tracker.py +++ b/tests/test_fault_tracker.py @@ -15,6 +15,14 @@ from pyxbot2_diagnostics.aggregator.fault_tracker import FaultLifecycleTracker +def _values(*reports: str, count: int | None = None) -> tuple[DiagnosticKeyValue, ...]: + declared_count = len(reports) if count is None else count + return ( + DiagnosticKeyValue("fault_count", str(declared_count)), + *(DiagnosticKeyValue("fault_report", report) for report in reports), + ) + + def _message( *, level: int, @@ -36,15 +44,12 @@ def _message( def test_single_fault_raise_duplicate_and_clear() -> None: tracker = FaultLifecycleTracker() - fault = _message( - level=2, - values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),), - ) + fault = _message(level=2, values=_values("motor over temperature")) transitions = tracker.update(fault, recv_time=11.0) assert [item.kind for item in transitions] == ["raised"] state = transitions[0].state - assert state.key.code == "0x4210" + assert state.key.report == "motor over temperature" assert state.active assert state.last_raised == 10.0 assert state.occurrence_count == 1 @@ -53,12 +58,7 @@ def test_single_fault_raise_duplicate_and_clear() -> None: assert next(iter(tracker.states.values())).occurrence_count == 1 cleared = tracker.update( - _message( - level=0, - values=(DiagnosticKeyValue("fault_codes", []),), - stamp=20.0, - msg="OK", - ), + _message(level=0, values=_values(), stamp=20.0, msg="OK"), recv_time=21.0, ) assert [item.kind for item in cleared] == ["cleared"] @@ -66,10 +66,10 @@ def test_single_fault_raise_duplicate_and_clear() -> None: assert cleared[0].state.last_cleared == 20.0 -def test_code_change_clears_old_and_raises_new() -> None: +def test_report_change_clears_old_and_raises_new() -> None: tracker = FaultLifecycleTracker() tracker.update( - _message(level=2, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), + _message(level=2, values=_values("motor over temperature")), recv_time=10.0, ) @@ -77,14 +77,14 @@ def test_code_change_clears_old_and_raises_new() -> None: _message( level=2, stamp=30.0, - values=(DiagnosticKeyValue("fault_codes", ["0x7500"]),), + values=_values("encoder signal lost"), ), recv_time=31.0, ) - assert [(item.kind, item.state.key.code) for item in transitions] == [ - ("cleared", "0x4210"), - ("raised", "0x7500"), + assert [(item.kind, item.state.key.report) for item in transitions] == [ + ("cleared", "motor over temperature"), + ("raised", "encoder signal lost"), ] @@ -93,33 +93,33 @@ def test_multiple_simultaneous_faults_are_diffed_as_sets() -> None: first = tracker.update( _message( level=2, - values=(DiagnosticKeyValue("fault_codes", [0x4210, "0x7500"]),), + values=_values("motor over temperature", "encoder signal lost"), ), recv_time=10.0, ) - assert {(item.kind, item.state.key.code) for item in first} == { - ("raised", "0x4210"), - ("raised", "0x7500"), + assert {(item.kind, item.state.key.report) for item in first} == { + ("raised", "motor over temperature"), + ("raised", "encoder signal lost"), } second = tracker.update( _message( level=2, stamp=40.0, - values=(DiagnosticKeyValue("fault_codes", ["0x7500", "0x8611"]),), + values=_values("encoder signal lost", "dc link over voltage"), ), recv_time=41.0, ) - assert {(item.kind, item.state.key.code) for item in second} == { - ("cleared", "0x4210"), - ("raised", "0x8611"), + assert {(item.kind, item.state.key.report) for item in second} == { + ("cleared", "motor over temperature"), + ("raised", "dc link over voltage"), } def test_stale_does_not_clear_hardware_faults() -> None: tracker = FaultLifecycleTracker() tracker.update( - _message(level=2, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), + _message(level=2, values=_values("motor over temperature")), recv_time=10.0, ) @@ -128,7 +128,7 @@ def test_stale_does_not_clear_hardware_faults() -> None: level=3, stamp=50.0, msg="STALE", - values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),), + values=_values("motor over temperature"), ), recv_time=50.0, ) == [] @@ -141,7 +141,7 @@ def test_non_health_message_is_not_interpreted_as_fault_contract() -> None: _message( level=2, node="/xbot/joint/knee_pitch_1/temperature", - values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),), + values=_values("motor over temperature"), ), recv_time=10.0, ) @@ -149,24 +149,42 @@ def test_non_health_message_is_not_interpreted_as_fault_contract() -> None: assert tracker.states == {} -def test_health_message_requires_fault_codes_key() -> None: +def test_health_message_requires_exactly_one_fault_count() -> None: tracker = FaultLifecycleTracker() - transitions = tracker.update( - _message(level=2, values=(DiagnosticKeyValue("temperature", 90.0),)), - recv_time=10.0, + missing = _message( + level=2, + values=(DiagnosticKeyValue("fault_report", "motor over temperature"),), ) - assert transitions == [] + duplicate = _message( + level=2, + values=( + DiagnosticKeyValue("fault_count", "1"), + DiagnosticKeyValue("fault_count", "1"), + DiagnosticKeyValue("fault_report", "motor over temperature"), + ), + ) + assert tracker.update(missing, recv_time=10.0) == [] + assert tracker.update(duplicate, recv_time=10.0) == [] + assert tracker.states == {} + + +def test_fault_count_must_match_unique_reports() -> None: + tracker = FaultLifecycleTracker() + assert tracker.update( + _message(level=2, values=_values("motor over temperature", count=2)), + recv_time=10.0, + ) == [] assert tracker.states == {} -def test_inconsistent_level_and_fault_codes_is_ignored() -> None: +def test_inconsistent_level_and_reports_is_ignored() -> None: tracker = FaultLifecycleTracker() assert tracker.update( - _message(level=0, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), + _message(level=0, values=_values("motor over temperature")), recv_time=10.0, ) == [] assert tracker.update( - _message(level=2, values=(DiagnosticKeyValue("fault_codes", []),)), + _message(level=2, values=_values()), recv_time=10.0, ) == [] assert tracker.states == {} @@ -183,13 +201,16 @@ def close(self) -> None: @dataclass class FaultSink: + calls: list[str] = field(default_factory=list) transitions: list[object] = field(default_factory=list) state_snapshots: list[dict[object, object]] = field(default_factory=list) def handle_message(self, message) -> None: del message + self.calls.append("message") def handle_fault_transitions(self, transitions, states) -> None: + self.calls.append("transitions") self.transitions.extend(transitions) self.state_snapshots.append(dict(states)) @@ -200,7 +221,7 @@ def close(self) -> None: return -def test_aggregator_publishes_fault_transitions_to_opt_in_sink() -> None: +def test_aggregator_publishes_transitions_before_health_snapshot() -> None: config = AggregatorConfig( aggregator=AggregatorSection( zmq_endpoint="inproc://unused", @@ -213,10 +234,11 @@ def test_aggregator_publishes_fault_transitions_to_opt_in_sink() -> None: aggregator = DiagnosticsAggregator(config, [sink], sources=[NullSource()]) aggregator.process_message( - _message(level=2, values=(DiagnosticKeyValue("fault_codes", ["0x4210"]),)), + _message(level=2, values=_values("motor over temperature")), now=11.0, ) + assert sink.calls[:2] == ["transitions", "message"] assert len(sink.transitions) == 1 assert sink.transitions[0].kind == "raised" assert next(iter(aggregator.fault_states.values())).active From 78468e4cf8f4a97dd4835658b79f4a64f2d1728c Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 00:02:58 +0200 Subject: [PATCH 09/12] document health and influx fault schema --- docs/fault_health_contract.md | 151 ++++++++++++++++++++-------------- 1 file changed, 88 insertions(+), 63 deletions(-) diff --git a/docs/fault_health_contract.md b/docs/fault_health_contract.md index 3abe859..d90f5cf 100644 --- a/docs/fault_health_contract.md +++ b/docs/fault_health_contract.md @@ -2,98 +2,123 @@ The aggregator interprets a diagnostic message as a fault-state source only when its normalized `node` path ends in `/health`. -Examples of health-source paths: +Examples: - `/xbot/joint/knee_pitch_1/health` - `/xbot/power/battery/health` - `host/robot-pc/network/eth0/health` -Messages with other suffixes remain ordinary diagnostics and are not inspected for fault lifecycle information, even if they contain similarly named values. +Messages with other suffixes remain ordinary diagnostics even if they contain similarly named values. -## Required values +## ROS string-native values -Every `/health` message must contain exactly one canonical key-value entry named `fault_codes`. +`diagnostic_msgs/KeyValue.value` is a string. A valid `/health` message therefore contains: -```json -{"key": "fault_codes", "value": ["0x4210", "0x7500"]} -``` - -`fault_codes` is the complete set of fault codes active for that health source at the message timestamp. It is not a delta and must not contain only newly raised faults. +- exactly one `fault_count` entry containing a non-negative decimal integer; +- zero or more repeated `fault_report` entries; +- `fault_count` equal to the number of unique `fault_report` values. -The value should be an array. Each code may be an integer or a stable string identifier. Integer codes and hexadecimal strings are normalized to uppercase hexadecimal strings by the aggregator. Zero-like values (`0`, `0x0000`, `none`, `ok`) are ignored and must not be used as real fault identifiers. +Each `fault_report` is a standardized, stable, human-friendly identifier. It must not contain changing measurements, timestamps, counters, or other occurrence-specific text. -An empty array explicitly means that the source has no active faults: +Active example: -```json -{"key": "fault_codes", "value": []} +```yaml +name: /xbot/joint/knee_pitch_1/health +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 ``` -A `/health` message without `fault_codes` is invalid for lifecycle tracking and is ignored. It does not clear previously active faults. +Healthy example: + +```yaml +name: /xbot/joint/knee_pitch_1/health +hardware_id: knee_pitch_1 +level: 0 +message: OK +values: + - key: fault_count + value: "0" +``` -## Level and message semantics +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. -The ROS diagnostics level must agree with `fault_codes`: +## Level semantics -| `fault_codes` | `level` | Meaning | +| Reports | `level` | Meaning | |---|---:|---| -| empty | `0` | Healthy; clear all faults previously reported by this source | -| non-empty | `1` | One or more warning-level faults are active | -| non-empty | `2` | One or more error-level faults are active | -| unchanged/any | `3` | Source is stale; do not raise or clear hardware faults | +| 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 | -Inconsistent combinations, such as non-empty `fault_codes` with level `0`, are ignored for lifecycle tracking. +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 a human-readable summary for dashboards and logs. It is not part of fault identity and must not be parsed to determine active faults. +`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. -`hw_id` identifies the physical device. A fault lifecycle is keyed by: +The lifecycle identity is: ```text -(hw_id, node, fault_code) +(hw_id, node, fault_report) ``` -## Examples - -Active faults: - -```json -{ - "v": 1, - "node": "/xbot/joint/knee_pitch_1/health", - "hw_id": "knee_pitch_1", - "stamp": 1785967012.0, - "level": 2, - "msg": "Drive reports over-temperature and communication faults", - "values": [ - {"key": "fault_codes", "value": ["0x4210", "0x7500"]} - ] -} -``` +## InfluxDB schema -Healthy/cleared: - -```json -{ - "v": 1, - "node": "/xbot/joint/knee_pitch_1/health", - "hw_id": "knee_pitch_1", - "stamp": 1785967305.0, - "level": 0, - "msg": "OK", - "values": [ - {"key": "fault_codes", "value": []} - ] -} -``` +Every valid health publication, including the heartbeat, produces one `health` 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 health 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`) -## Compatibility aliases +Fields: -For migration, the tracker currently accepts these aliases: +- `active` +- `level` +- `message` +- `occurrence_count` +- `first_raised_ns` +- `last_raised_ns` +- `last_cleared_ns`, when available -- single-code aliases: `fault_code`, `error_code` -- multi-code aliases: `error_codes`, `active_fault_codes`, `active_error_codes` +`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. -New publishers must use `fault_codes`. Compatibility aliases may be removed in a future schema version. +Both health and event points use the diagnostic source timestamp when valid, falling back to aggregator wall-clock time only when necessary. ## Design rationale -The path suffix provides an explicit namespace boundary so arbitrary telemetry cannot accidentally create or clear faults. A complete active-code set makes updates idempotent and allows the aggregator to compute raises and clears by set difference. It also supports devices that report one current code and devices that report multiple simultaneous codes without changing the storage model. +The `/health` 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 `health` 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. From d6e9fb0305b6c9e1882306fb9a78b2805815c9ea Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 00:03:17 +0200 Subject: [PATCH 10/12] test grafana friendly influx fault schema --- tests/test_fault_influx.py | 133 +++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 tests/test_fault_influx.py diff --git a/tests/test_fault_influx.py b/tests/test_fault_influx.py new file mode 100644 index 0000000..3dfafd0 --- /dev/null +++ b/tests/test_fault_influx.py @@ -0,0 +1,133 @@ +from pyxbot2_diagnostics.aggregator.aggregator import ( + DiagnosticKeyValue, + DiagnosticsMessage, +) +from pyxbot2_diagnostics.aggregator.fault_tracker import ( + FaultKey, + FaultState, + FaultTransition, +) +from pyxbot2_diagnostics.aggregator.sinks.influxdb_sink import InfluxDBSink + + +class FakeWriteApi: + def __init__(self) -> None: + self.calls = [] + + def write(self, *, bucket, org, record): + self.calls.append({"bucket": bucket, "org": org, "record": record}) + + +def _sink(): + fake = FakeWriteApi() + sink = InfluxDBSink( + enabled=True, + url="", + token="", + org="xbot2", + bucket="diagnostics", + write_api=fake, + ) + return sink, fake + + +def _health(*reports: str, level: int = 2, stamp: float = 10.0): + values = [DiagnosticKeyValue("fault_count", str(len(reports)))] + values.extend(DiagnosticKeyValue("fault_report", report) for report in reports) + return DiagnosticsMessage( + v=1, + node="/xbot/joint/knee_pitch_1/health", + hw_id="knee_pitch_1", + stamp=stamp, + level=level, + msg="Drive faults active" if reports else "OK", + values=tuple(values), + ) + + +def test_health_snapshot_is_one_grafana_row_per_source() -> None: + sink, fake = _sink() + sink.handle_message( + _health("motor over temperature", "encoder signal lost") + ) + sink._last_flush = 0.0 + sink.publish_state({}) + + point = fake.calls[0]["record"][0] + assert point["measurement"] == "health" + assert point["tags"] == { + "hw_id": "knee_pitch_1", + "path": "/xbot/joint/knee_pitch_1/health", + "name": "knee_pitch_1", + "component": "xbot/joint", + } + assert point["fields"]["fault_count"] == 2 + assert point["fields"]["active_fault_reports"] == ( + "encoder signal lost; motor over temperature" + ) + assert point["time"] == 10_000_000_000 + + +def test_fault_event_uses_standardized_report_as_tag() -> None: + sink, fake = _sink() + state = FaultState( + key=FaultKey( + "knee_pitch_1", + "/xbot/joint/knee_pitch_1/health", + "motor over temperature", + ), + active=True, + level=2, + message="Drive faults active", + first_raised=10.0, + last_raised=10.0, + last_cleared=None, + occurrence_count=1, + ) + sink.handle_fault_transitions( + [FaultTransition("raised", state, 10.0)], + {state.key: state}, + ) + sink.handle_message(_health("motor over temperature")) + sink._last_flush = 0.0 + sink.publish_state({}) + + event, health = fake.calls[0]["record"] + assert event["measurement"] == "fault_event" + assert event["tags"]["fault_report"] == "motor over temperature" + assert event["tags"]["transition"] == "raised" + assert event["fields"]["active"] is True + assert health["fields"]["last_fault_report"] == "motor over temperature" + assert health["fields"]["last_fault_active"] is True + assert health["fields"]["last_raised_ns"] == 10_000_000_000 + + +def test_clear_event_updates_next_health_snapshot() -> None: + sink, fake = _sink() + state = FaultState( + key=FaultKey( + "knee_pitch_1", + "/xbot/joint/knee_pitch_1/health", + "motor over temperature", + ), + active=False, + level=0, + message="OK", + first_raised=10.0, + last_raised=10.0, + last_cleared=20.0, + occurrence_count=1, + ) + sink.handle_fault_transitions( + [FaultTransition("cleared", state, 20.0)], + {state.key: state}, + ) + sink.handle_message(_health(level=0, stamp=20.0)) + sink._last_flush = 0.0 + sink.publish_state({}) + + event, health = fake.calls[0]["record"] + assert event["tags"]["transition"] == "cleared" + assert event["fields"]["last_cleared_ns"] == 20_000_000_000 + assert health["fields"]["last_fault_active"] is False + assert health["fields"]["last_cleared_ns"] == 20_000_000_000 From f482602d8a72f66db77b41e2b92684ac7b8d6b9d Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 00:10:24 +0200 Subject: [PATCH 11/12] summarize immediate child diagnostic levels --- .../aggregator/sinks/ros_diagnostics_sink.py | 40 ++++++++- tests/test_ros_group_summary.py | 85 +++++++++++++++++++ 2 files changed, 122 insertions(+), 3 deletions(-) create mode 100644 tests/test_ros_group_summary.py diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/ros_diagnostics_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/ros_diagnostics_sink.py index 256b124..bdcb234 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/ros_diagnostics_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/ros_diagnostics_sink.py @@ -3,6 +3,7 @@ from __future__ import annotations import time +from collections import Counter from dataclasses import dataclass, field from typing import Any, Callable @@ -93,11 +94,24 @@ def _level_message(level: int) -> str: }.get(level, "UNKNOWN") @classmethod - def _to_group_status(cls, name: str, level: int) -> RosDiagnosticStatus: + def _child_level_summary(cls, child_levels: list[int]) -> str: + """Summarize non-OK immediate children in deterministic severity order.""" + counts = Counter(child_levels) + parts = [ + f"{counts[level]} {cls._level_message(level)}" + for level in (1, 2, 3) + if counts[level] + ] + return ", ".join(parts) if parts else "OK" + + @classmethod + def _to_group_status( + cls, name: str, level: int, child_levels: list[int] + ) -> RosDiagnosticStatus: return RosDiagnosticStatus( level=cls._to_level(level), name=name, - message=cls._level_message(level), + message=cls._child_level_summary(child_levels), hardware_id="", values=[], ) @@ -118,6 +132,13 @@ def _aggregate_segments(self, name: str) -> list[str]: def _aggregate_path(segments: list[str], length: int) -> str: return "/" + "/".join(segments[:length]) + @staticmethod + def _parent_path(path: str) -> str | None: + parent, separator, _ = path.rpartition("/") + if not separator or not parent: + return None + return parent + def _build_aggregated_statuses( self, states: dict[str, DiagnosticsMessage] ) -> list[RosDiagnosticStatus]: @@ -132,9 +153,22 @@ def _build_aggregated_statuses( path = self._aggregate_path(segments, length) group_levels[path] = max(group_levels.get(path, 0), msg.level) + immediate_child_levels: dict[str, list[int]] = {} + for path, level in group_levels.items(): + parent = self._parent_path(path) + if parent is not None and parent in group_levels: + immediate_child_levels.setdefault(parent, []).append(level) + statuses: list[RosDiagnosticStatus] = [] for path in sorted(group_levels): - statuses.append(leaf_statuses.get(path) or self._to_group_status(path, group_levels[path])) + statuses.append( + leaf_statuses.get(path) + or self._to_group_status( + path, + group_levels[path], + immediate_child_levels.get(path, []), + ) + ) return statuses def handle_message(self, message: DiagnosticsMessage) -> None: diff --git a/tests/test_ros_group_summary.py b/tests/test_ros_group_summary.py new file mode 100644 index 0000000..438bcf3 --- /dev/null +++ b/tests/test_ros_group_summary.py @@ -0,0 +1,85 @@ +from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticsMessage +from pyxbot2_diagnostics.aggregator.sinks.ros_diagnostics_sink import RosDiagnosticsSink + + +def _msg(node: str, level: int, message: str) -> DiagnosticsMessage: + return DiagnosticsMessage( + v=1, + node=node, + hw_id="hw", + stamp=1.0, + level=level, + msg=message, + values=(), + ) + + +def _level_value(level): + return level[0] if isinstance(level, bytes) else level + + +def test_group_message_summarizes_immediate_child_levels() -> None: + published = [] + sink = RosDiagnosticsSink( + aggregated_publisher=published.append, + time_fn=lambda: 1.0, + ) + + sink.publish_state( + { + "/xbot/group/warn_a/metric": _msg("/xbot/group/warn_a/metric", 1, "warn"), + "/xbot/group/warn_b/metric": _msg("/xbot/group/warn_b/metric", 1, "warn"), + "/xbot/group/error/metric": _msg("/xbot/group/error/metric", 2, "error"), + "/xbot/group/stale_a/metric": _msg("/xbot/group/stale_a/metric", 3, "stale"), + "/xbot/group/stale_b/metric": _msg("/xbot/group/stale_b/metric", 3, "stale"), + "/xbot/group/stale_c/metric": _msg("/xbot/group/stale_c/metric", 3, "stale"), + "/xbot/group/stale_d/metric": _msg("/xbot/group/stale_d/metric", 3, "stale"), + "/xbot/group/ok/metric": _msg("/xbot/group/ok/metric", 0, "OK"), + } + ) + + statuses = {status.name: status for status in published[0].status} + group = statuses["/Robot/xbot/group"] + assert _level_value(group.level) == 3 + assert group.message == "2 WARN, 1 ERROR, 4 STALE" + + +def test_group_summary_counts_direct_children_not_descendant_leaves() -> None: + published = [] + sink = RosDiagnosticsSink( + aggregated_publisher=published.append, + time_fn=lambda: 1.0, + ) + + sink.publish_state( + { + "/xbot/arm/joint_a/temperature": _msg( + "/xbot/arm/joint_a/temperature", 1, "warn" + ), + "/xbot/arm/joint_a/voltage": _msg( + "/xbot/arm/joint_a/voltage", 2, "error" + ), + "/xbot/arm/joint_b/temperature": _msg( + "/xbot/arm/joint_b/temperature", 0, "OK" + ), + } + ) + + statuses = {status.name: status for status in published[0].status} + assert statuses["/Robot/xbot/arm"].message == "1 ERROR" + assert statuses["/Robot/xbot/arm/joint_a"].message == "1 WARN, 1 ERROR" + assert statuses["/Robot/xbot/arm/joint_b"].message == "OK" + + +def test_leaf_message_is_preserved() -> None: + published = [] + sink = RosDiagnosticsSink( + aggregated_publisher=published.append, + time_fn=lambda: 1.0, + ) + sink.publish_state( + {"/xbot/drive/health": _msg("/xbot/drive/health", 2, "Drive fault")} + ) + + statuses = {status.name: status for status in published[0].status} + assert statuses["/Robot/xbot/drive/health"].message == "Drive fault" From 6091d8f8d0a464343ae20eddb49c14aa78d85c3d Mon Sep 17 00:00:00 2001 From: Arturo Laurenzi Date: Thu, 6 Aug 2026 21:38:15 +0200 Subject: [PATCH 12/12] fault support --- docker/docker-compose.diagnostics.yml | 2 +- docs/fault_health_contract.md | 31 +++++---- .../aggregator/aggregator.py | 2 +- .../aggregator/fault_tracker.py | 42 +++++++++---- .../aggregator/sinks/influxdb_sink.py | 24 +++---- tests/test_fault_influx.py | 36 +++++------ tests/test_fault_tracker.py | 63 +++++++++++++++++-- tests/test_ros_group_summary.py | 4 +- 8 files changed, 142 insertions(+), 62 deletions(-) diff --git a/docker/docker-compose.diagnostics.yml b/docker/docker-compose.diagnostics.yml index ad0946d..de6a25e 100644 --- a/docker/docker-compose.diagnostics.yml +++ b/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: diff --git a/docs/fault_health_contract.md b/docs/fault_health_contract.md index d90f5cf..422f90b 100644 --- a/docs/fault_health_contract.md +++ b/docs/fault_health_contract.md @@ -1,29 +1,34 @@ -# Fault health diagnostic contract +# Fault diagnostic contract -The aggregator interprets a diagnostic message as a fault-state source only when its normalized `node` path ends in `/health`. +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/health` -- `/xbot/power/battery/health` -- `host/robot-pc/network/eth0/health` +- `/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 `/health` message therefore contains: +`diagnostic_msgs/KeyValue.value` is a string. A valid `/fault` message therefore contains: -- exactly one `fault_count` entry containing a non-negative decimal integer; +- 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/health +name: /xbot/joint/knee_pitch_1/fault hardware_id: knee_pitch_1 level: 2 message: Drive faults active @@ -39,7 +44,7 @@ values: Healthy example: ```yaml -name: /xbot/joint/knee_pitch_1/health +name: /xbot/joint/knee_pitch_1/fault hardware_id: knee_pitch_1 level: 0 message: OK @@ -71,7 +76,7 @@ The lifecycle identity is: ## InfluxDB schema -Every valid health publication, including the heartbeat, produces one `health` point. +Every valid fault publication, including the heartbeat, produces one `fault` point. Tags: @@ -92,7 +97,7 @@ Fields: - `last_raised_ns`, when known - `last_cleared_ns`, when known -This gives Grafana one current row per health source using a latest-point query. +This gives Grafana one current row per fault source using a latest-point query. Each raise or clear produces one `fault_event` point. @@ -117,8 +122,8 @@ Fields: `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 health and event points use the diagnostic source timestamp when valid, falling back to aggregator wall-clock time only when necessary. +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 `/health` 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 `health` 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. +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. diff --git a/python/src/pyxbot2_diagnostics/aggregator/aggregator.py b/python/src/pyxbot2_diagnostics/aggregator/aggregator.py index b1d7ca4..6c8aad1 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/aggregator.py +++ b/python/src/pyxbot2_diagnostics/aggregator/aggregator.py @@ -211,7 +211,7 @@ def process_message(self, message: DiagnosticsMessage, now: float | None = None) transitions = self._fault_tracker.update(message, recv_time) # Transition-aware sinks update their per-source summary before the - # corresponding /health snapshot is serialized. + # corresponding /fault snapshot is serialized. self._publish_fault_updates(transitions) for sink in self._sinks: sink.handle_message(message) diff --git a/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py index 6d01ef0..c8a6c2a 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py +++ b/python/src/pyxbot2_diagnostics/aggregator/fault_tracker.py @@ -1,12 +1,30 @@ -"""Fault lifecycle tracking for normalized health diagnostics messages.""" +"""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.""" @@ -40,9 +58,9 @@ class FaultTransition: class FaultLifecycleTracker: - """Track faults published through the string-native health contract. + """Track faults published through the string-native fault contract. - Only nodes whose final path segment is ``health`` participate. A valid + 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. """ @@ -52,7 +70,7 @@ def __init__(self) -> None: self._active_by_source: dict[tuple[str, str], set[str]] = {} def update(self, message: Any, recv_time: float) -> list[FaultTransition]: - if not self._is_health_node(message.node) or message.level == 3: + if not self._is_fault_node(message.node) or message.level == 3: return [] reports, declared_count, valid = self._extract_reports(message.values) @@ -118,9 +136,9 @@ def update(self, message: Any, recv_time: float) -> list[FaultTransition]: return transitions @staticmethod - def _is_health_node(node: Any) -> bool: + def _is_fault_node(node: Any) -> bool: parts = [part for part in str(node).split("/") if part] - return bool(parts) and parts[-1].lower() == "health" + return bool(parts) and parts[-1].lower() == "fault" @staticmethod def _event_stamp(source_stamp: Any, recv_time: float) -> float: @@ -142,14 +160,14 @@ def _extract_reports( if key == "fault_report": report = cls._normalize_report(item.value) if report is None: - return set(), None, False + # 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": - try: - count = int(str(item.value).strip(), 10) - except (TypeError, ValueError): - return set(), None, False - if count < 0: + count = parse_fault_count(item.value) + if count is None: return set(), None, False counts.append(count) diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py index 33e929b..679dc56 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py @@ -1,4 +1,4 @@ -"""InfluxDB sink for diagnostics, health snapshots, and fault events.""" +"""InfluxDB sink for diagnostics, fault snapshots, and fault events.""" from __future__ import annotations @@ -7,6 +7,8 @@ import time from typing import Any +from pyxbot2_diagnostics.aggregator.fault_tracker import parse_fault_count + from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticsMessage LOGGER = logging.getLogger(__name__) @@ -18,8 +20,8 @@ class InfluxDBSink: Ordinary diagnostics keep the existing path-derived measurement schema. - Health messages (paths ending in ``/health``) are written as one periodic - ``health`` snapshot per source. Fault transitions are written separately as + Fault messages (paths ending in ``/fault``) are written as one periodic + ``fault`` snapshot per source. Fault transitions are written separately as ``fault_event`` points. Standardized ``fault_report`` values are tags so Grafana can filter and group by report efficiently. """ @@ -83,11 +85,11 @@ def handle_message(self, message: DiagnosticsMessage) -> None: "component": component, } - if measurement.lower() == "health": - fields = self._health_fields(message) + if measurement.lower() == "fault": + fields = self._fault_fields(message) if fields is None: # Preserve ordinary diagnostics export while avoiding a - # misleading lifecycle snapshot for malformed health data. + # misleading lifecycle snapshot for malformed fault data. fields = self._generic_fields(message) else: fields = self._generic_fields(message) @@ -112,7 +114,7 @@ def handle_fault_transitions(self, transitions, states) -> None: self._last_fault_by_source[source] = state parts = [part for part in state.key.node.split("/") if part] - name = parts[-2] if len(parts) >= 2 else "health" + name = parts[-2] if len(parts) >= 2 else "fault" component = "/".join(parts[:-2]) fields: dict[str, Any] = { "active": state.active, @@ -141,7 +143,7 @@ def handle_fault_transitions(self, transitions, states) -> None: } ) - def _health_fields(self, message: DiagnosticsMessage) -> dict[str, Any] | None: + def _fault_fields(self, message: DiagnosticsMessage) -> dict[str, Any] | None: reports: list[str] = [] counts: list[int] = [] fields: dict[str, Any] = {"level": message.level} @@ -153,10 +155,10 @@ def _health_fields(self, message: DiagnosticsMessage) -> dict[str, Any] | None: if report: reports.append(report) elif key == "fault_count": - try: - counts.append(int(str(kv.value).strip(), 10)) - except ValueError: + count = parse_fault_count(kv.value) + if count is None: return None + counts.append(count) else: self._add_generic_field(fields, kv.key, kv.value) diff --git a/tests/test_fault_influx.py b/tests/test_fault_influx.py index 3dfafd0..220b686 100644 --- a/tests/test_fault_influx.py +++ b/tests/test_fault_influx.py @@ -31,12 +31,12 @@ def _sink(): return sink, fake -def _health(*reports: str, level: int = 2, stamp: float = 10.0): +def _fault(*reports: str, level: int = 2, stamp: float = 10.0): values = [DiagnosticKeyValue("fault_count", str(len(reports)))] values.extend(DiagnosticKeyValue("fault_report", report) for report in reports) return DiagnosticsMessage( v=1, - node="/xbot/joint/knee_pitch_1/health", + node="/xbot/joint/knee_pitch_1/fault", hw_id="knee_pitch_1", stamp=stamp, level=level, @@ -45,19 +45,19 @@ def _health(*reports: str, level: int = 2, stamp: float = 10.0): ) -def test_health_snapshot_is_one_grafana_row_per_source() -> None: +def test_fault_snapshot_is_one_grafana_row_per_source() -> None: sink, fake = _sink() sink.handle_message( - _health("motor over temperature", "encoder signal lost") + _fault("motor over temperature", "encoder signal lost") ) sink._last_flush = 0.0 sink.publish_state({}) point = fake.calls[0]["record"][0] - assert point["measurement"] == "health" + assert point["measurement"] == "fault" assert point["tags"] == { "hw_id": "knee_pitch_1", - "path": "/xbot/joint/knee_pitch_1/health", + "path": "/xbot/joint/knee_pitch_1/fault", "name": "knee_pitch_1", "component": "xbot/joint", } @@ -73,7 +73,7 @@ def test_fault_event_uses_standardized_report_as_tag() -> None: state = FaultState( key=FaultKey( "knee_pitch_1", - "/xbot/joint/knee_pitch_1/health", + "/xbot/joint/knee_pitch_1/fault", "motor over temperature", ), active=True, @@ -88,26 +88,26 @@ def test_fault_event_uses_standardized_report_as_tag() -> None: [FaultTransition("raised", state, 10.0)], {state.key: state}, ) - sink.handle_message(_health("motor over temperature")) + sink.handle_message(_fault("motor over temperature")) sink._last_flush = 0.0 sink.publish_state({}) - event, health = fake.calls[0]["record"] + event, fault = fake.calls[0]["record"] assert event["measurement"] == "fault_event" assert event["tags"]["fault_report"] == "motor over temperature" assert event["tags"]["transition"] == "raised" assert event["fields"]["active"] is True - assert health["fields"]["last_fault_report"] == "motor over temperature" - assert health["fields"]["last_fault_active"] is True - assert health["fields"]["last_raised_ns"] == 10_000_000_000 + assert fault["fields"]["last_fault_report"] == "motor over temperature" + assert fault["fields"]["last_fault_active"] is True + assert fault["fields"]["last_raised_ns"] == 10_000_000_000 -def test_clear_event_updates_next_health_snapshot() -> None: +def test_clear_event_updates_next_fault_snapshot() -> None: sink, fake = _sink() state = FaultState( key=FaultKey( "knee_pitch_1", - "/xbot/joint/knee_pitch_1/health", + "/xbot/joint/knee_pitch_1/fault", "motor over temperature", ), active=False, @@ -122,12 +122,12 @@ def test_clear_event_updates_next_health_snapshot() -> None: [FaultTransition("cleared", state, 20.0)], {state.key: state}, ) - sink.handle_message(_health(level=0, stamp=20.0)) + sink.handle_message(_fault(level=0, stamp=20.0)) sink._last_flush = 0.0 sink.publish_state({}) - event, health = fake.calls[0]["record"] + event, fault = fake.calls[0]["record"] assert event["tags"]["transition"] == "cleared" assert event["fields"]["last_cleared_ns"] == 20_000_000_000 - assert health["fields"]["last_fault_active"] is False - assert health["fields"]["last_cleared_ns"] == 20_000_000_000 + assert fault["fields"]["last_fault_active"] is False + assert fault["fields"]["last_cleared_ns"] == 20_000_000_000 diff --git a/tests/test_fault_tracker.py b/tests/test_fault_tracker.py index d359c2b..fefb5ff 100644 --- a/tests/test_fault_tracker.py +++ b/tests/test_fault_tracker.py @@ -29,7 +29,7 @@ def _message( values: tuple[DiagnosticKeyValue, ...], stamp: float = 10.0, msg: str = "fault", - node: str = "/xbot/joint/knee_pitch_1/health", + node: str = "/xbot/joint/knee_pitch_1/fault", ) -> DiagnosticsMessage: return DiagnosticsMessage( v=1, @@ -135,7 +135,7 @@ def test_stale_does_not_clear_hardware_faults() -> None: assert next(iter(tracker.states.values())).active -def test_non_health_message_is_not_interpreted_as_fault_contract() -> None: +def test_non_fault_message_is_not_interpreted_as_fault_contract() -> None: tracker = FaultLifecycleTracker() transitions = tracker.update( _message( @@ -149,7 +149,21 @@ def test_non_health_message_is_not_interpreted_as_fault_contract() -> None: assert tracker.states == {} -def test_health_message_requires_exactly_one_fault_count() -> None: +def test_legacy_health_message_is_not_interpreted_as_fault_contract() -> None: + tracker = FaultLifecycleTracker() + transitions = tracker.update( + _message( + level=2, + node="/xbot/joint/knee_pitch_1/health", + values=_values("motor over temperature"), + ), + recv_time=10.0, + ) + assert transitions == [] + assert tracker.states == {} + + +def test_fault_message_requires_exactly_one_fault_count() -> None: tracker = FaultLifecycleTracker() missing = _message( level=2, @@ -177,6 +191,47 @@ def test_fault_count_must_match_unique_reports() -> None: assert tracker.states == {} +def test_ros_double_formatted_whole_count_and_empty_clear_report_are_accepted() -> None: + tracker = FaultLifecycleTracker() + raised = tracker.update( + _message( + level=2, + values=( + DiagnosticKeyValue("fault_count", "1.000000"), + DiagnosticKeyValue("fault_report", "motor over temperature"), + ), + ), + recv_time=10.0, + ) + assert [item.kind for item in raised] == ["raised"] + + cleared = tracker.update( + _message( + level=0, + values=( + DiagnosticKeyValue("fault_count", "0.000000"), + DiagnosticKeyValue("fault_report", ""), + ), + ), + recv_time=11.0, + ) + assert [item.kind for item in cleared] == ["cleared"] + + +def test_fractional_fault_count_is_rejected() -> None: + tracker = FaultLifecycleTracker() + assert tracker.update( + _message( + level=2, + values=( + DiagnosticKeyValue("fault_count", "1.5"), + DiagnosticKeyValue("fault_report", "motor over temperature"), + ), + ), + recv_time=10.0, + ) == [] + + def test_inconsistent_level_and_reports_is_ignored() -> None: tracker = FaultLifecycleTracker() assert tracker.update( @@ -221,7 +276,7 @@ def close(self) -> None: return -def test_aggregator_publishes_transitions_before_health_snapshot() -> None: +def test_aggregator_publishes_transitions_before_fault_snapshot() -> None: config = AggregatorConfig( aggregator=AggregatorSection( zmq_endpoint="inproc://unused", diff --git a/tests/test_ros_group_summary.py b/tests/test_ros_group_summary.py index 438bcf3..88bd2e0 100644 --- a/tests/test_ros_group_summary.py +++ b/tests/test_ros_group_summary.py @@ -78,8 +78,8 @@ def test_leaf_message_is_preserved() -> None: time_fn=lambda: 1.0, ) sink.publish_state( - {"/xbot/drive/health": _msg("/xbot/drive/health", 2, "Drive fault")} + {"/xbot/drive/fault": _msg("/xbot/drive/fault", 2, "Drive fault")} ) statuses = {status.name: status for status in published[0].status} - assert statuses["/Robot/xbot/drive/health"].message == "Drive fault" + assert statuses["/Robot/xbot/drive/fault"].message == "Drive fault"