nx_pred_maint_consumer.nx
buildroot/runtime/consumers/nx_pred_maint_consumer.nx
about
nx_pred_maint_consumer.nx -- predictive-maintenance consumer.
Subscribes to OBD-II shim events (nx_obd2_shim.nx, commit be03adfa)
and produces "anomaly flagged" or "predicted-DTC-in-N-days" events
AHEAD of OBD's reactive DTC firing. Concretises the
INTEROPERABILITY_CHARTER.md ยง12 worked example:
"4-day-early P0117 prediction via aggregated coolant-temp anomaly".
This is THE bring-the-most-out-of-it proof per the operator's
cardinal: OBD-native tools react to DTCs after they fire; Nishi
pulls predictive insight out of the SAME OBD data that OBD's spec
itself doesn't define.
Status: SEED v0.1.0. 2026-05-26.
WINNER-TIER: WINNER-S CANDIDATE (no incumbent matches the
value-add; OBD-native tools are reactive-only)
INCUMBENTS: python-obd (reactive; no anomaly), obdlib (reactive),
ELM327 tools (passthrough), Vector CANalyzer
(commercial; has trend tracking but no
substrate-integrated prediction)
NUMBERS: V1 ships the algorithm; measured "days-early" metric
vs the reactive baseline requires real CAN-logger
data + a DTC corpus. Provisional.
GAP: no incumbent solves this; V1 establishes the
baseline. M-next: real OBD trace replay against
recorded fleet data; confirm S rating.
EXEMPTION REASON: n/a
Algorithm (V1):
Per PID:
- ring buffer of last N samples (timestamp + value)
- Welford's online algorithm: rolling mean + variance
(numerically stable for streaming data; Welford 1962)
- linear regression: slope estimate over the window
(Pearson r^2 fit; for "is value drifting?" detection)
Anomaly flagging:
- 3-sigma rule: current value > mean + 3*sigma (or < mean - 3*sigma)
- configurable per-PID thresholds (per ZERO_TO_ADVANCED.md
data-driven approach; threshold lives in config not code)
dependencies 2 imports · 2 importers
imports: nx_syscalls.nxnx_telemetry.nx
imported by: _obd_pipeline_smoke.nxnx_modbus_pred_maint_pipeline.nx
structs
| 76 | struct NxPidRing |
| 93 | struct NxPidStats |
| 294 | struct NxPredMaintConsumer |
consts
| 59 | const NX_PM_OK: i64 = 0 |
| 60 | const NX_PM_PID_NOT_REGISTERED: i64 = 1 |
| 61 | const NX_PM_INSUFFICIENT_DATA: i64 = 2 // <MIN_SAMPLES samples; can't estimate |
| 62 | const NX_PM_BUFFER_OVERFLOW: i64 = 3 |
| 63 | const NX_PM_BAD_INPUT: i64 = 4 |
| 66 | const NX_PM_MAX_PIDS: i64 = 32 // max tracked PIDs per vehicle |
| 67 | const NX_PM_WINDOW_SIZE: i64 = 256 // samples per PID ring buffer |
| 68 | const NX_PM_MIN_SAMPLES: i64 = 16 // min before stats are emitted |
| 103 | const NX_PM_ANOM_NONE: i64 = 0 |
| 104 | const NX_PM_ANOM_HIGH_3SIGMA: i64 = 1 // value > mean + 3*sigma |
| 105 | const NX_PM_ANOM_LOW_3SIGMA: i64 = 2 // value < mean - 3*sigma |
| 106 | const NX_PM_ANOM_DRIFT_UP: i64 = 3 // slope crossing threshold_hi within window |
| 107 | const NX_PM_ANOM_DRIFT_DOWN: i64 = 4 // slope crossing threshold_lo within window |
functions
| 113 | func nx_pm_isqrt(x: i64) -> i64 called by 1: nx_pm_compute_stats |
| 129 | func nx_pm_ring_init(ring: *NxPidRing, pid: i64, called by 1: nx_pm_consumer_register |
| 148 | func nx_pm_ring_push(ring: *NxPidRing, ts_ns: i64, value: i64) -> i64 called by 1: nx_pm_consumer_on_event |
| 164 | func nx_pm_compute_stats(ring: *NxPidRing, stats: *NxPidStats) -> i64 |
| 244 | func nx_pm_classify_latest(ring: *NxPidRing, stats: *NxPidStats) -> i64 called by 1: nx_pm_consumer_on_event |
| 301 | func nx_pm_consumer_init(c: *NxPredMaintConsumer, rings: *NxPidRing) -> i64 called by 1: main |
| 314 | func nx_pm_consumer_set_telemetry_bus(c: *NxPredMaintConsumer, |
| 322 | func nx_pm_consumer_register(c: *NxPredMaintConsumer, pid: i64, |
| 334 | func nx_pm_consumer_find_ring(c: *NxPredMaintConsumer, pid: i64) -> *NxPidRing called by 1: nx_pm_consumer_on_event |
| 352 | func nx_pm_consumer_on_event(c: *NxPredMaintConsumer, |