diff --git a/docs/device_health_schema.md b/docs/device_health_schema.md new file mode 100644 index 0000000..4c5881b --- /dev/null +++ b/docs/device_health_schema.md @@ -0,0 +1,130 @@ +# Device health diagnostic contract + +A diagnostic is treated as a device-health message when the final segment of its +`DiagnosticStatus.name` / aggregator `node` is either `health` or `health_status`. +The preceding path identifies the logical device. `hardware_id` must contain the +non-empty physical device identifier. + +Example: + +```text +name: /xbot/joint/left_knee/motor/health_status +device path: /xbot/joint/left_knee/motor +hardware_id: SN-0028417 +``` + +## Required key-value fields + +Values may arrive as JSON strings, as they do in ROS +`diagnostic_msgs/KeyValue`. Native JSON values are also accepted by the ZMQ input. + +| Key | Type | Meaning | +|---|---|---| +| `device.boot_id` | non-empty string | Counter epoch identifier | +| `faults.active` | array of unique strings | Complete currently active fault set | +| `faults.raise_count_total` | object: fault code to non-negative integer | Monotonic raise count within `device.boot_id` | +| `faults.last_raised` | object: fault code to timestamp or null | Latest raise time | +| `faults.last_cleared` | object: fault code to timestamp or null | Latest clear time | + +Timestamps may be non-negative Unix epoch seconds or timezone-aware ISO-8601 +strings. Internally they are normalized to nanoseconds. InfluxDB fields exposed to +Grafana use Unix epoch milliseconds (`last_raised_ms`, `last_cleared_ms`). + +Optional device-specific key-value fields are retained on the `device_health` +InfluxDB point. Scalar values remain scalar fields; structured values are serialized +as compact JSON strings. + +## Validation rules + +- Diagnostic keys must be unique. +- Every active or timestamped fault code must exist in + `faults.raise_count_total`. +- A positive raise counter requires a non-null last-raise time. +- A zero raise counter requires a null or absent last-raise time. +- An active fault must have a positive raise counter. +- For an active fault, `last_cleared` must precede `last_raised` when both exist. +- For an inactive fault, `last_cleared` must not precede `last_raised` when both + exist. + +Malformed health messages are logged and omitted from InfluxDB health measurements. +Other diagnostic messages continue through the generic InfluxDB path unchanged. + +## InfluxDB measurements + +### `device_health` + +One point is written per valid health snapshot. + +Tags: + +- `hw_id` +- `device_path` + +Fields: + +- `level` +- `active_fault_count` +- `boot_id` +- `message`, when non-empty +- optional device-specific health values + +The point timestamp is the diagnostic source timestamp. + +### `fault_counter` + +One point is written per known fault code in each valid health snapshot. + +Tags: + +- `hw_id` +- `device_path` +- `fault_code` + +Fields: + +- `active` +- `raise_count_total` +- `boot_id` +- `last_raised_ms`, when known +- `last_cleared_ms`, when known + +The point timestamp is the diagnostic source timestamp. + +### `fault_occurrence` + +A sparse point is written when a cumulative raise counter increases. + +Tags: + +- `hw_id` +- `device_path` +- `fault_code` + +Fields: + +- `occurrences`: counter delta since the previous received sample +- `counter_before` +- `counter_after` +- `boot_id` +- `last_raised_ms`, when known + +The point timestamp is the current health diagnostic source timestamp. When the +counter delta is greater than one, the exact timestamps of all raises are not known; +`last_raised_ms` records the latest raise supplied by the producer. + +The first sample for a fault establishes a baseline. A changed `boot_id` also +establishes a new baseline. A counter decrease within the same boot is logged as a +warning and establishes a new baseline. None of these baseline cases emits a +`fault_occurrence` point. + +`boot_id` is stored as an Influx field, not a tag, to avoid creating a new series on +every reboot. + +## Open decisions + +1. Whether the canonical suffix should be only `/health_status`, only `/health`, + or whether both aliases should remain supported. +2. Whether `device.boot_id` remains mandatory and whether counters are boot-scoped + or persisted for the lifetime of the device. +3. Whether optional device-specific health values should share the + `device_health` measurement or use the generic diagnostic measurement. diff --git a/python/src/pyxbot2_diagnostics/aggregator/health.py b/python/src/pyxbot2_diagnostics/aggregator/health.py new file mode 100644 index 0000000..d9b3918 --- /dev/null +++ b/python/src/pyxbot2_diagnostics/aggregator/health.py @@ -0,0 +1,291 @@ +"""Validation and normalization for device health diagnostics.""" + +from __future__ import annotations + +import json +import math +from dataclasses import dataclass +from datetime import datetime, timezone +from decimal import Decimal +from typing import Any, Iterable + +from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticKeyValue, DiagnosticsMessage + +HEALTH_STATUS_SUFFIXES = frozenset({"health", "health_status"}) + +_BOOT_ID_KEY = "device.boot_id" +_ACTIVE_KEY = "faults.active" +_RAISE_COUNT_KEY = "faults.raise_count_total" +_LAST_RAISED_KEY = "faults.last_raised" +_LAST_CLEARED_KEY = "faults.last_cleared" + +REQUIRED_HEALTH_KEYS = frozenset( + { + _BOOT_ID_KEY, + _ACTIVE_KEY, + _RAISE_COUNT_KEY, + _LAST_RAISED_KEY, + _LAST_CLEARED_KEY, + } +) + + +class HealthMessageValidationError(ValueError): + """Raised when a health diagnostic does not satisfy the health contract.""" + + +@dataclass(frozen=True) +class FaultHealthRecord: + """Normalized state for one fault code.""" + + code: str + active: bool + raise_count_total: int + last_raised_ns: int | None + last_cleared_ns: int | None + + +@dataclass(frozen=True) +class HealthStatus: + """Normalized device health snapshot.""" + + device_path: str + boot_id: str + faults: tuple[FaultHealthRecord, ...] + extra_values: tuple[DiagnosticKeyValue, ...] + + @property + def active_fault_count(self) -> int: + return sum(fault.active for fault in self.faults) + + +def is_health_message(message: DiagnosticsMessage) -> bool: + """Return whether *message* is identified as a device health status.""" + + parts = _path_parts(message.node) + return bool(parts) and parts[-1] in HEALTH_STATUS_SUFFIXES + + +def parse_health_message(message: DiagnosticsMessage) -> HealthStatus: + """Validate and normalize a ``/health`` or ``/health_status`` message.""" + + parts = _path_parts(message.node) + if not parts or parts[-1] not in HEALTH_STATUS_SUFFIXES: + raise HealthMessageValidationError( + "health status name must end with '/health' or '/health_status'" + ) + if len(parts) < 2: + raise HealthMessageValidationError("health status name must include a device path") + if not message.hw_id.strip(): + raise HealthMessageValidationError("health status requires a non-empty hardware_id") + if not math.isfinite(message.stamp) or message.stamp < 0: + raise HealthMessageValidationError("health status stamp must be finite and non-negative") + + values = _unique_value_map(message.values) + missing = sorted(REQUIRED_HEALTH_KEYS - values.keys()) + if missing: + raise HealthMessageValidationError( + "health status is missing required keys: " + ", ".join(missing) + ) + + boot_id = _require_non_empty_string(values[_BOOT_ID_KEY], _BOOT_ID_KEY) + active_codes = _parse_active_faults(values[_ACTIVE_KEY]) + counts = _parse_counter_map(values[_RAISE_COUNT_KEY]) + last_raised = _parse_timestamp_map(values[_LAST_RAISED_KEY], _LAST_RAISED_KEY) + last_cleared = _parse_timestamp_map(values[_LAST_CLEARED_KEY], _LAST_CLEARED_KEY) + + referenced_codes = set(active_codes) | set(last_raised) | set(last_cleared) + unknown_codes = sorted(referenced_codes - counts.keys()) + if unknown_codes: + raise HealthMessageValidationError( + f"{_RAISE_COUNT_KEY} is missing referenced fault codes: " + + ", ".join(unknown_codes) + ) + + faults: list[FaultHealthRecord] = [] + active_set = set(active_codes) + for code in sorted(counts): + count = counts[code] + raised_ns = last_raised.get(code) + cleared_ns = last_cleared.get(code) + active = code in active_set + + if count == 0 and raised_ns is not None: + raise HealthMessageValidationError( + f"fault '{code}' has zero raises but a non-null last-raised timestamp" + ) + if count > 0 and raised_ns is None: + raise HealthMessageValidationError( + f"fault '{code}' has a positive raise counter but no last-raised timestamp" + ) + if active and count == 0: + raise HealthMessageValidationError( + f"active fault '{code}' must have a positive raise counter" + ) + if raised_ns is not None and cleared_ns is not None: + if active and cleared_ns >= raised_ns: + raise HealthMessageValidationError( + f"active fault '{code}' has last-cleared >= last-raised" + ) + if not active and raised_ns > cleared_ns: + raise HealthMessageValidationError( + f"inactive fault '{code}' has last-raised > last-cleared" + ) + + faults.append( + FaultHealthRecord( + code=code, + active=active, + raise_count_total=count, + last_raised_ns=raised_ns, + last_cleared_ns=cleared_ns, + ) + ) + + extra_values = tuple( + entry for entry in message.values if entry.key not in REQUIRED_HEALTH_KEYS + ) + device_path = "/" + "/".join(parts[:-1]) + return HealthStatus( + device_path=device_path, + boot_id=boot_id, + faults=tuple(faults), + extra_values=extra_values, + ) + + +def timestamp_seconds_to_ns(value: int | float) -> int: + """Convert finite non-negative epoch seconds to integer nanoseconds.""" + + seconds = Decimal(str(value)) + if not seconds.is_finite() or seconds < 0: + raise HealthMessageValidationError( + "timestamp seconds must be finite and non-negative" + ) + return int(seconds * Decimal(1_000_000_000)) + + +def _path_parts(path: str) -> list[str]: + return [part for part in path.split("/") if part] + + +def _unique_value_map(values: Iterable[DiagnosticKeyValue]) -> dict[str, Any]: + result: dict[str, Any] = {} + duplicates: list[str] = [] + for entry in values: + if entry.key in result: + duplicates.append(entry.key) + else: + result[entry.key] = entry.value + if duplicates: + raise HealthMessageValidationError( + "health status contains duplicate keys: " + ", ".join(sorted(set(duplicates))) + ) + return result + + +def _require_non_empty_string(value: Any, key: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise HealthMessageValidationError(f"{key} must be a non-empty string") + return value.strip() + + +def _decode_json(value: Any, key: str) -> Any: + if isinstance(value, str): + try: + return json.loads(value) + except json.JSONDecodeError as exc: + raise HealthMessageValidationError(f"{key} contains invalid JSON: {exc.msg}") from exc + return value + + +def _parse_active_faults(value: Any) -> tuple[str, ...]: + decoded = _decode_json(value, _ACTIVE_KEY) + if not isinstance(decoded, list): + raise HealthMessageValidationError(f"{_ACTIVE_KEY} must be a JSON array") + + result: list[str] = [] + for index, code in enumerate(decoded): + if not isinstance(code, str) or not code.strip(): + raise HealthMessageValidationError( + f"{_ACTIVE_KEY}[{index}] must be a non-empty string" + ) + result.append(code.strip()) + + if len(result) != len(set(result)): + raise HealthMessageValidationError(f"{_ACTIVE_KEY} must not contain duplicates") + return tuple(result) + + +def _parse_counter_map(value: Any) -> dict[str, int]: + decoded = _decode_json(value, _RAISE_COUNT_KEY) + if not isinstance(decoded, dict): + raise HealthMessageValidationError(f"{_RAISE_COUNT_KEY} must be a JSON object") + + result: dict[str, int] = {} + for raw_code, count in decoded.items(): + code = _validate_fault_code(raw_code, _RAISE_COUNT_KEY) + if isinstance(count, bool) or not isinstance(count, int) or count < 0: + raise HealthMessageValidationError( + f"{_RAISE_COUNT_KEY}['{code}'] must be a non-negative integer" + ) + result[code] = count + return result + + +def _parse_timestamp_map(value: Any, key: str) -> dict[str, int | None]: + decoded = _decode_json(value, key) + if not isinstance(decoded, dict): + raise HealthMessageValidationError(f"{key} must be a JSON object") + + result: dict[str, int | None] = {} + for raw_code, timestamp in decoded.items(): + code = _validate_fault_code(raw_code, key) + result[code] = _parse_timestamp(timestamp, f"{key}['{code}']") + return result + + +def _validate_fault_code(value: Any, key: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise HealthMessageValidationError(f"{key} fault codes must be non-empty strings") + return value.strip() + + +def _parse_timestamp(value: Any, location: str) -> int | None: + if value is None: + return None + if isinstance(value, bool): + raise HealthMessageValidationError( + f"{location} must be null, epoch seconds, or an ISO-8601 timestamp" + ) + if isinstance(value, (int, float)): + seconds = float(value) + if not math.isfinite(seconds) or seconds < 0: + raise HealthMessageValidationError( + f"{location} epoch seconds must be finite and non-negative" + ) + return timestamp_seconds_to_ns(seconds) + if not isinstance(value, str) or not value.strip(): + raise HealthMessageValidationError( + f"{location} must be null, epoch seconds, or an ISO-8601 timestamp" + ) + + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise HealthMessageValidationError( + f"{location} must be a valid ISO-8601 timestamp" + ) from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise HealthMessageValidationError(f"{location} must include a timezone") + utc = parsed.astimezone(timezone.utc) + epoch = datetime(1970, 1, 1, tzinfo=timezone.utc) + delta = utc - epoch + return ( + delta.days * 86_400 * 1_000_000_000 + + delta.seconds * 1_000_000_000 + + delta.microseconds * 1_000 + ) diff --git a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py index 2cc2326..d58dee0 100644 --- a/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py +++ b/python/src/pyxbot2_diagnostics/aggregator/sinks/influxdb_sink.py @@ -7,28 +7,58 @@ from typing import Any from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticsMessage +from pyxbot2_diagnostics.aggregator.health import ( + HealthMessageValidationError, + HealthStatus, + is_health_message, + parse_health_message, + timestamp_seconds_to_ns, +) LOGGER = logging.getLogger(__name__) -# Single measurement name for all robot diagnostics. +# Generic measurement name for ordinary robot diagnostics. _MEASUREMENT = "robot_diagnostics" +_DEVICE_HEALTH_MEASUREMENT = "device_health" +_FAULT_COUNTER_MEASUREMENT = "fault_counter" +_FAULT_OCCURRENCE_MEASUREMENT = "fault_occurrence" # Minimum seconds between batch writes to InfluxDB. _FLUSH_INTERVAL_SEC = 1.0 +# (hardware id, device path, fault code) -> (boot id, last total counter) +FaultCounterKey = tuple[str, str, str] +FaultCounterState = tuple[str, int] + 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 retain the existing generic representation. Messages whose + path ends in ``/health`` or ``/health_status`` are validated and normalized into + three measurements with a deliberately small tag set: + + ``device_health`` + tags: hw_id, device_path + fields: level, active_fault_count, boot_id, message, optional values + + ``fault_counter`` + tags: hw_id, device_path, fault_code + fields: active, raise_count_total, boot_id, last_raised_ms, + last_cleared_ms + + ``fault_occurrence`` + tags: hw_id, device_path, fault_code + fields: occurrences, counter_before, counter_after, boot_id, + last_raised_ms + + ``boot_id`` is a field rather than a tag to avoid creating a new series on every + reboot. Transition timestamps are Unix epoch milliseconds so Grafana can format + them directly as date/time fields. InfluxDB point timestamps remain nanoseconds. - 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. + The first sample for a source/fault or a new boot establishes a baseline and + does not emit an occurrence. A counter decrease within the same boot is logged + and also establishes a new baseline. """ def __init__( @@ -48,6 +78,7 @@ def __init__( self._write_api = write_api self._pending: list[dict[str, Any]] = [] self._last_flush = 0.0 + self._fault_counter_state: dict[FaultCounterKey, FaultCounterState] = {} if not enabled: return @@ -70,8 +101,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) @@ -79,12 +108,30 @@ def handle_message(self, message: DiagnosticsMessage) -> None: if not self._enabled or self._write_api is None: return + if is_health_message(message): + self._handle_health_message(message) + return + + self._pending.append(self._generic_point(message)) + + def _handle_health_message(self, message: DiagnosticsMessage) -> None: + try: + health = parse_health_message(message) + except HealthMessageValidationError as exc: + LOGGER.warning("Rejecting invalid health message %s: %s", message.node, exc) + return + + sample_time_ns = timestamp_seconds_to_ns(message.stamp) + self._pending.append(self._device_health_point(message, health, sample_time_ns)) + self._pending.extend(self._fault_counter_points(message, health, sample_time_ns)) + self._pending.extend(self._fault_occurrence_points(message, health, sample_time_ns)) + + @staticmethod + def _generic_point(message: DiagnosticsMessage) -> dict[str, Any]: 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) @@ -94,33 +141,143 @@ def handle_message(self, message: DiagnosticsMessage) -> None: 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 - """ - measurement = parts[-1] if parts else "unknown" name = parts[-2] if len(parts) >= 2 else measurement component = "/".join(parts[:-2]) - self._pending.append( - { - "measurement": measurement, - "tags": { - "hw_id": message.hw_id if message.hw_id else "unknown", - "path": path, - "name": name, - "component": component, - }, - "fields": fields, - "time": int(1e9 * time.time()), + return { + "measurement": measurement, + "tags": { + "hw_id": message.hw_id if message.hw_id else "unknown", + "path": path, + "name": name, + "component": component, + }, + "fields": fields, + "time": int(1e9 * time.time()), + } + + @staticmethod + def _device_health_point( + message: DiagnosticsMessage, + health: HealthStatus, + sample_time_ns: int, + ) -> dict[str, Any]: + fields: dict[str, Any] = { + "level": message.level, + "active_fault_count": health.active_fault_count, + "boot_id": health.boot_id, + } + if message.msg: + fields["message"] = message.msg + + for entry in health.extra_values: + fields[entry.key] = _coerce_health_field(entry.value) + + return { + "measurement": _DEVICE_HEALTH_MEASUREMENT, + "tags": { + "hw_id": message.hw_id, + "device_path": health.device_path, + }, + "fields": fields, + "time": sample_time_ns, + } + + @staticmethod + def _fault_counter_points( + message: DiagnosticsMessage, + health: HealthStatus, + sample_time_ns: int, + ) -> list[dict[str, Any]]: + points: list[dict[str, Any]] = [] + for fault in health.faults: + fields: dict[str, Any] = { + "active": fault.active, + "raise_count_total": fault.raise_count_total, + "boot_id": health.boot_id, + } + if fault.last_raised_ns is not None: + fields["last_raised_ms"] = _timestamp_ns_to_ms(fault.last_raised_ns) + if fault.last_cleared_ns is not None: + fields["last_cleared_ms"] = _timestamp_ns_to_ms(fault.last_cleared_ns) + + points.append( + { + "measurement": _FAULT_COUNTER_MEASUREMENT, + "tags": { + "hw_id": message.hw_id, + "device_path": health.device_path, + "fault_code": fault.code, + }, + "fields": fields, + "time": sample_time_ns, + } + ) + return points + + def _fault_occurrence_points( + self, + message: DiagnosticsMessage, + health: HealthStatus, + sample_time_ns: int, + ) -> list[dict[str, Any]]: + points: list[dict[str, Any]] = [] + for fault in health.faults: + key: FaultCounterKey = (message.hw_id, health.device_path, fault.code) + previous = self._fault_counter_state.get(key) + self._fault_counter_state[key] = (health.boot_id, fault.raise_count_total) + + if previous is None: + continue + + previous_boot_id, previous_total = previous + if previous_boot_id != health.boot_id: + LOGGER.info( + "Fault counter epoch changed for %s %s (%s -> %s); establishing new baseline", + health.device_path, + fault.code, + previous_boot_id, + health.boot_id, + ) + continue + + if fault.raise_count_total < previous_total: + LOGGER.warning( + "Fault counter decreased within boot for %s %s: %d -> %d; establishing new baseline", + health.device_path, + fault.code, + previous_total, + fault.raise_count_total, + ) + continue + + occurrences = fault.raise_count_total - previous_total + if occurrences == 0: + continue + + fields: dict[str, Any] = { + "occurrences": occurrences, + "counter_before": previous_total, + "counter_after": fault.raise_count_total, + "boot_id": health.boot_id, } - ) + if fault.last_raised_ns is not None: + fields["last_raised_ms"] = _timestamp_ns_to_ms(fault.last_raised_ns) + + points.append( + { + "measurement": _FAULT_OCCURRENCE_MEASUREMENT, + "tags": { + "hw_id": message.hw_id, + "device_path": health.device_path, + "fault_code": fault.code, + }, + "fields": fields, + "time": sample_time_ns, + } + ) + return points def publish_state(self, states: dict[str, DiagnosticsMessage]) -> None: del states @@ -142,7 +299,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) @@ -150,3 +306,18 @@ def close(self) -> None: LOGGER.warning("InfluxDB final flush failed: %s", exc) if self._client is not None: self._client.close() + + +def _timestamp_ns_to_ms(value: int) -> int: + return value // 1_000_000 + + +def _coerce_health_field(value: Any) -> Any: + if isinstance(value, (bool, int, float, str)): + return value + try: + import json + + return json.dumps(value, separators=(",", ":"), sort_keys=True) + except (TypeError, ValueError): + return str(value) diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..06a1f0e --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,340 @@ +import logging + +import pytest + +from pyxbot2_diagnostics.aggregator.aggregator import DiagnosticKeyValue, DiagnosticsMessage +from pyxbot2_diagnostics.aggregator.health import ( + HealthMessageValidationError, + is_health_message, + parse_health_message, +) +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 _health_msg( + *, + node: str = "/xbot/joint/knee/motor/health_status", + hw_id: str = "SN-1", + stamp: float = 1785614401.25, + values: tuple[DiagnosticKeyValue, ...] | None = None, +) -> DiagnosticsMessage: + return DiagnosticsMessage( + v=1, + node=node, + hw_id=hw_id, + stamp=stamp, + level=2, + msg="OVERCURRENT active", + values=values + or ( + DiagnosticKeyValue("device.boot_id", "boot-a"), + DiagnosticKeyValue("faults.active", '["OVERCURRENT"]'), + DiagnosticKeyValue( + "faults.raise_count_total", + '{"OVERCURRENT":4,"ENCODER_CRC":2}', + ), + DiagnosticKeyValue( + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:00Z",' + '"ENCODER_CRC":"2026-08-01T19:00:00+00:00"}', + ), + DiagnosticKeyValue( + "faults.last_cleared", + '{"OVERCURRENT":"2026-08-01T18:00:00Z",' + '"ENCODER_CRC":"2026-08-01T19:00:01Z"}', + ), + DiagnosticKeyValue("thermal.temperature", 72.5), + DiagnosticKeyValue("communication.degraded", False), + ), + ) + + +def _replace_value(message: DiagnosticsMessage, key: str, value) -> DiagnosticsMessage: + values = tuple( + DiagnosticKeyValue(entry.key, value if entry.key == key else entry.value) + for entry in message.values + ) + return DiagnosticsMessage( + v=message.v, + node=message.node, + hw_id=message.hw_id, + stamp=message.stamp, + level=message.level, + msg=message.msg, + values=values, + ) + + +def _remove_value(message: DiagnosticsMessage, key: str) -> DiagnosticsMessage: + return DiagnosticsMessage( + v=message.v, + node=message.node, + hw_id=message.hw_id, + stamp=message.stamp, + level=message.level, + msg=message.msg, + values=tuple(entry for entry in message.values if entry.key != key), + ) + + +def test_parse_valid_health_message() -> None: + health = parse_health_message(_health_msg()) + + assert health.device_path == "/xbot/joint/knee/motor" + assert health.boot_id == "boot-a" + assert health.active_fault_count == 1 + assert [fault.code for fault in health.faults] == ["ENCODER_CRC", "OVERCURRENT"] + assert health.faults[0].active is False + assert health.faults[1].active is True + assert [entry.key for entry in health.extra_values] == [ + "thermal.temperature", + "communication.degraded", + ] + + +def test_accepts_native_json_values_and_health_alias() -> None: + message = _health_msg(node="xbot/motor/health") + message = _replace_value(message, "faults.active", ["OVERCURRENT"]) + message = _replace_value( + message, + "faults.raise_count_total", + {"OVERCURRENT": 4, "ENCODER_CRC": 2}, + ) + message = _replace_value( + message, + "faults.last_raised", + {"OVERCURRENT": 1785614400.0, "ENCODER_CRC": 1785610800.0}, + ) + message = _replace_value( + message, + "faults.last_cleared", + {"OVERCURRENT": 1785607200.0, "ENCODER_CRC": 1785610801.0}, + ) + + assert is_health_message(message) + assert parse_health_message(message).device_path == "/xbot/motor" + + +@pytest.mark.parametrize( + "key,value,match", + [ + ("faults.active", '["OVERCURRENT","OVERCURRENT"]', "duplicates"), + ("faults.raise_count_total", '{"OVERCURRENT":-1}', "non-negative"), + ( + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:00"}', + "timezone", + ), + ], +) +def test_rejects_invalid_fault_payloads(key, value, match) -> None: + with pytest.raises(HealthMessageValidationError, match=match): + parse_health_message(_replace_value(_health_msg(), key, value)) + + +def test_rejects_duplicate_diagnostic_keys() -> None: + original = _health_msg() + message = _health_msg( + values=original.values + (DiagnosticKeyValue("faults.active", "[]"),) + ) + with pytest.raises(HealthMessageValidationError, match="duplicate keys"): + parse_health_message(message) + + +def test_rejects_missing_required_key() -> None: + with pytest.raises(HealthMessageValidationError, match="device.boot_id"): + parse_health_message(_remove_value(_health_msg(), "device.boot_id")) + + +def test_rejects_active_fault_missing_from_counter_map() -> None: + message = _replace_value( + _health_msg(), "faults.raise_count_total", '{"ENCODER_CRC":2}' + ) + with pytest.raises( + HealthMessageValidationError, match="missing referenced fault codes" + ): + parse_health_message(message) + + +def test_rejects_positive_counter_without_last_raise() -> None: + message = _replace_value( + _health_msg(), + "faults.last_raised", + '{"OVERCURRENT":null,"ENCODER_CRC":"2026-08-01T19:00:00Z"}', + ) + with pytest.raises(HealthMessageValidationError, match="positive raise counter"): + parse_health_message(message) + + +def test_requires_health_suffix_and_device_path() -> None: + assert not is_health_message(_health_msg(node="/xbot/motor/temperature")) + with pytest.raises(HealthMessageValidationError, match="must end"): + parse_health_message(_health_msg(node="/xbot/motor/temperature")) + with pytest.raises(HealthMessageValidationError, match="device path"): + parse_health_message(_health_msg(node="/health")) + + +def _sink(fake: FakeWriteApi) -> InfluxDBSink: + return InfluxDBSink( + enabled=True, + url="", + token="", + org="xbot2", + bucket="diagnostics", + write_api=fake, + ) + + +def _flush(sink: InfluxDBSink) -> None: + sink._last_flush = 0.0 + sink.publish_state({}) + + +def test_influx_sink_normalizes_health_message() -> None: + fake = FakeWriteApi() + sink = _sink(fake) + + sink.handle_message(_health_msg()) + _flush(sink) + + points = fake.calls[0]["record"] + assert [point["measurement"] for point in points] == [ + "device_health", + "fault_counter", + "fault_counter", + ] + + health_point = points[0] + assert health_point["tags"] == { + "hw_id": "SN-1", + "device_path": "/xbot/joint/knee/motor", + } + assert health_point["fields"]["boot_id"] == "boot-a" + assert health_point["fields"]["level"] == 2 + assert health_point["fields"]["active_fault_count"] == 1 + assert health_point["fields"]["thermal.temperature"] == 72.5 + assert health_point["fields"]["communication.degraded"] is False + assert health_point["time"] == 1785614401250000000 + + fault_points = {point["tags"]["fault_code"]: point for point in points[1:]} + assert fault_points["OVERCURRENT"]["tags"] == { + "hw_id": "SN-1", + "device_path": "/xbot/joint/knee/motor", + "fault_code": "OVERCURRENT", + } + assert fault_points["OVERCURRENT"]["fields"]["active"] is True + assert fault_points["OVERCURRENT"]["fields"]["raise_count_total"] == 4 + assert fault_points["OVERCURRENT"]["fields"]["boot_id"] == "boot-a" + assert fault_points["ENCODER_CRC"]["fields"]["active"] is False + assert fault_points["ENCODER_CRC"]["fields"]["last_cleared_ms"] > 0 + + +def test_influx_sink_emits_fault_occurrence_from_counter_delta() -> None: + fake = FakeWriteApi() + sink = _sink(fake) + + sink.handle_message(_health_msg(stamp=1785614401.0)) + updated = _replace_value( + _health_msg(stamp=1785614402.0), + "faults.raise_count_total", + '{"OVERCURRENT":7,"ENCODER_CRC":2}', + ) + updated = _replace_value( + updated, + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:01.500Z",' + '"ENCODER_CRC":"2026-08-01T19:00:00Z"}', + ) + sink.handle_message(updated) + _flush(sink) + + occurrences = [ + point + for point in fake.calls[0]["record"] + if point["measurement"] == "fault_occurrence" + ] + assert len(occurrences) == 1 + point = occurrences[0] + assert point["tags"] == { + "hw_id": "SN-1", + "device_path": "/xbot/joint/knee/motor", + "fault_code": "OVERCURRENT", + } + assert point["fields"]["boot_id"] == "boot-a" + assert point["fields"]["occurrences"] == 3 + assert point["fields"]["counter_before"] == 4 + assert point["fields"]["counter_after"] == 7 + assert point["fields"]["last_raised_ms"] == 1785614401500 + assert point["time"] == 1785614402000000000 + + +def test_influx_sink_reboot_establishes_new_counter_baseline() -> None: + fake = FakeWriteApi() + sink = _sink(fake) + + sink.handle_message(_health_msg(stamp=1785614401.0)) + restarted = _replace_value(_health_msg(stamp=1785614402.0), "device.boot_id", "boot-b") + restarted = _replace_value( + restarted, + "faults.raise_count_total", + '{"OVERCURRENT":1,"ENCODER_CRC":0}', + ) + restarted = _replace_value( + restarted, + "faults.last_raised", + '{"OVERCURRENT":"2026-08-01T20:00:01Z","ENCODER_CRC":null}', + ) + restarted = _replace_value( + restarted, + "faults.last_cleared", + '{"OVERCURRENT":null,"ENCODER_CRC":null}', + ) + sink.handle_message(restarted) + _flush(sink) + + assert not any( + point["measurement"] == "fault_occurrence" + for point in fake.calls[0]["record"] + ) + + +def test_influx_sink_counter_decrease_establishes_new_baseline(caplog) -> None: + fake = FakeWriteApi() + sink = _sink(fake) + + sink.handle_message(_health_msg(stamp=1785614401.0)) + decreased = _replace_value( + _health_msg(stamp=1785614402.0), + "faults.raise_count_total", + '{"OVERCURRENT":3,"ENCODER_CRC":2}', + ) + with caplog.at_level(logging.WARNING): + sink.handle_message(decreased) + _flush(sink) + + assert not any( + point["measurement"] == "fault_occurrence" + for point in fake.calls[0]["record"] + ) + assert "Fault counter decreased within boot" in caplog.text + + +def test_influx_sink_omits_invalid_health_message(caplog) -> None: + fake = FakeWriteApi() + sink = _sink(fake) + invalid = _remove_value(_health_msg(), "faults.active") + + with caplog.at_level(logging.WARNING): + sink.handle_message(invalid) + _flush(sink) + + assert fake.calls == [] + assert "missing required keys" in caplog.text