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 new file mode 100644 index 0000000..422f90b --- /dev/null +++ b/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. diff --git a/python/src/pyxbot2_diagnostics/aggregator/aggregator.py b/python/src/pyxbot2_diagnostics/aggregator/aggregator.py index 119d83d..6c8aad1 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 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() 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..c8a6c2a --- /dev/null +++ b/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 diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py index 2cc2326..679dc56 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py @@ -1,34 +1,29 @@ -"""InfluxDB sink for diagnostics metric values.""" +"""InfluxDB sink for diagnostics, fault snapshots, and fault events.""" from __future__ import annotations import logging +import math 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__) - -# 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. + 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. """ def __init__( @@ -48,6 +43,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 +66,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 +74,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() == "fault": + fields = self._fault_fields(message) + if fields is None: + # Preserve ordinary diagnostics export while avoiding a + # misleading lifecycle snapshot for malformed fault 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 "fault" + 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 _fault_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": + 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) + + 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 +227,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) 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_fault_influx.py b/tests/test_fault_influx.py new file mode 100644 index 0000000..220b686 --- /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 _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/fault", + hw_id="knee_pitch_1", + stamp=stamp, + level=level, + msg="Drive faults active" if reports else "OK", + values=tuple(values), + ) + + +def test_fault_snapshot_is_one_grafana_row_per_source() -> None: + sink, fake = _sink() + sink.handle_message( + _fault("motor over temperature", "encoder signal lost") + ) + sink._last_flush = 0.0 + sink.publish_state({}) + + point = fake.calls[0]["record"][0] + assert point["measurement"] == "fault" + assert point["tags"] == { + "hw_id": "knee_pitch_1", + "path": "/xbot/joint/knee_pitch_1/fault", + "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/fault", + "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(_fault("motor over temperature")) + sink._last_flush = 0.0 + sink.publish_state({}) + + 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 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_fault_snapshot() -> None: + sink, fake = _sink() + state = FaultState( + key=FaultKey( + "knee_pitch_1", + "/xbot/joint/knee_pitch_1/fault", + "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(_fault(level=0, stamp=20.0)) + sink._last_flush = 0.0 + sink.publish_state({}) + + event, fault = fake.calls[0]["record"] + assert event["tags"]["transition"] == "cleared" + assert event["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 new file mode 100644 index 0000000..fefb5ff --- /dev/null +++ b/tests/test_fault_tracker.py @@ -0,0 +1,300 @@ +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 _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, + values: tuple[DiagnosticKeyValue, ...], + stamp: float = 10.0, + msg: str = "fault", + node: str = "/xbot/joint/knee_pitch_1/fault", +) -> DiagnosticsMessage: + return DiagnosticsMessage( + v=1, + node=node, + 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=_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.report == "motor over temperature" + 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=_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_report_change_clears_old_and_raises_new() -> None: + tracker = FaultLifecycleTracker() + tracker.update( + _message(level=2, values=_values("motor over temperature")), + recv_time=10.0, + ) + + transitions = tracker.update( + _message( + level=2, + stamp=30.0, + values=_values("encoder signal lost"), + ), + recv_time=31.0, + ) + + assert [(item.kind, item.state.key.report) for item in transitions] == [ + ("cleared", "motor over temperature"), + ("raised", "encoder signal lost"), + ] + + +def test_multiple_simultaneous_faults_are_diffed_as_sets() -> None: + tracker = FaultLifecycleTracker() + first = tracker.update( + _message( + level=2, + values=_values("motor over temperature", "encoder signal lost"), + ), + recv_time=10.0, + ) + 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=_values("encoder signal lost", "dc link over voltage"), + ), + recv_time=41.0, + ) + 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=_values("motor over temperature")), + recv_time=10.0, + ) + + assert tracker.update( + _message( + level=3, + stamp=50.0, + msg="STALE", + values=_values("motor over temperature"), + ), + recv_time=50.0, + ) == [] + assert next(iter(tracker.states.values())).active + + +def test_non_fault_message_is_not_interpreted_as_fault_contract() -> None: + tracker = FaultLifecycleTracker() + transitions = tracker.update( + _message( + level=2, + node="/xbot/joint/knee_pitch_1/temperature", + values=_values("motor over temperature"), + ), + recv_time=10.0, + ) + assert transitions == [] + assert tracker.states == {} + + +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, + values=(DiagnosticKeyValue("fault_report", "motor over temperature"),), + ) + 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_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( + _message(level=0, values=_values("motor over temperature")), + recv_time=10.0, + ) == [] + assert tracker.update( + _message(level=2, values=_values()), + recv_time=10.0, + ) == [] + assert tracker.states == {} + + +class NullSource: + def poll(self, timeout_ms: int = 100): + del timeout_ms + return [] + + def close(self) -> None: + return + + +@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)) + + def publish_state(self, states) -> None: + del states + + def close(self) -> None: + return + + +def test_aggregator_publishes_transitions_before_fault_snapshot() -> 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=_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 + aggregator.close() diff --git a/tests/test_ros_group_summary.py b/tests/test_ros_group_summary.py new file mode 100644 index 0000000..88bd2e0 --- /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/fault": _msg("/xbot/drive/fault", 2, "Drive fault")} + ) + + statuses = {status.name: status for status in published[0].status} + assert statuses["/Robot/xbot/drive/fault"].message == "Drive fault"