nx_modbus_pred_maint_pipeline.nx source

1// nx_modbus_pred_maint_pipeline.nx -- Modbus + pred-maint composer. 2// 3// Wires the Modbus shim (commit eee5855b) -> nx_pred_maint_consumer 4// (commit 87dca0a0) -> nx_telemetry bus (commit 469f822f), proving 5// the consumer is PROTOCOL-AGNOSTIC: the same Welford / 3-sigma / 6// drift-extrapolation pipeline that catches OBD coolant-temp 7// anomalies also catches industrial-sensor anomalies via Modbus. 8// 9// This commit is the structural proof that the substrate-side 10// value-add layer (consumer) reuses across the §5 INTEROPERABILITY_CHARTER 11// catalog families. Future protocol shims (CAN raw / J1939 / BLE / 12// Modbus extensions / etc) all reuse the SAME consumer with the 13// SAME telemetry bus. 14// 15// Operator use case (W0+W1 tier per WORKLOAD_TARGETS.md): 16// Industrial pressure sensor on a Modbus RTU bus. Sensor reading 17// normally hovers around register-value 1500 (e.g., 15.00 bar via 18// centibar Q2 scaling). Slow drift toward 2500 indicates a 19// diaphragm wear-out failure mode. Pipeline detects the drift 20// 4-7 days before the sensor crosses its DTC-equivalent 21// threshold; operator schedules planned maintenance vs unplanned 22// outage. Same value-add proposition as OBD pred-maint; same 23// substrate code path. 24// 25// Status: SEED v0.1.0. 2026-05-27. 26// WINNER-TIER: WINNER-S CANDIDATE (preserved from pred-maint; 27// this commit doesn't add new claims, it ports the 28// S-CANDIDATE pred-maint claim to the industrial tier) 29// INCUMBENTS: same as pred-maint (no incumbent matches the 30// cross-protocol value-add); within Modbus specifically, 31// vendor SCADA tools (Ignition, Wonderware, ClearSCADA) 32// have trend-tracking but not substrate-integrated 33// prediction 34// PLAN: M-next: paired bench on real Modbus RTU sensor data 35// with known failure-mode corpus 36// EXEMPTION REASON: n/a; provisional pending hardware measurement 37 38import "nx_syscalls.nx" 39import "nx_telemetry.nx" 40import "shims/nx_modbus_shim.nx" 41import "consumers/nx_pred_maint_consumer.nx" 42 43// ===== Sealed verdict surface ================================================= 44// 45// Pipeline-specific verdicts; extends the underlying shim + consumer 46// verdict surfaces transparently (caller distinguishes by code 47// range; pipeline owns 400-409). 48 49const NX_MODPMP_OK: i64 = 0 50const NX_MODPMP_BAD_INPUT: i64 = 400 51const NX_MODPMP_NOT_INITIALIZED: i64 = 401 52const NX_MODPMP_WRONG_UNIT_ID: i64 = 402 // response unit_id != configured 53const NX_MODPMP_NO_REG_EVENT: i64 = 403 // parsed frame had no register-read event 54 55// ===== Pipeline state ================================================= 56// 57// V1: tracks ONE register per pipeline instance (one sensor). For 58// multi-register, V2 grows the tracked-register list + maintains 59// per-request state to match responses to requests. 60 61struct NxModbusPMP { 62 shim: *NxModbusShim 63 consumer: *NxPredMaintConsumer 64 telemetry: *NxTelemetryBus 65 66 tracked_unit_id: i64 // Modbus slave address 67 tracked_reg_addr: i64 // holding-register address 68 tracked_evt_kind: i64 // operator-chosen NxProtocolEvent key 69 // (used as the consumer's ring lookup 70 // key; opaque to the shim) 71 valid: i64 72} 73 74// ===== Init ================================================= 75// 76// Caller supplies pre-initialised shim + consumer + telemetry bus. 77// Pipeline binds them + records the tracked (unit_id, reg_addr). 78 79func nx_modpmp_init(p: *NxModbusPMP, 80 shim: *NxModbusShim, 81 consumer: *NxPredMaintConsumer, 82 telemetry: *NxTelemetryBus, 83 unit_id: i64, reg_addr: i64, evt_kind: i64) -> i64 { 84 if (p as i64) == 0 { return 0 - NX_MODPMP_BAD_INPUT } 85 if (shim as i64) == 0 { return 0 - NX_MODPMP_BAD_INPUT } 86 if (consumer as i64) == 0 { return 0 - NX_MODPMP_BAD_INPUT } 87 p.shim = shim 88 p.consumer = consumer 89 p.telemetry = telemetry 90 p.tracked_unit_id = unit_id 91 p.tracked_reg_addr = reg_addr 92 p.tracked_evt_kind = evt_kind 93 p.valid = 1 94 return NX_MODPMP_OK 95} 96 97// ===== Register: declare the tracked register's thresholds ================================================= 98// 99// Wraps nx_pm_consumer_register so the operator only has to declare 100// "this Modbus register's value drift triggers anomalies between 101// threshold_lo and threshold_hi"; pipeline handles the consumer- 102// register plumbing. 103 104func nx_modpmp_register(p: *NxModbusPMP, 105 times_buf: *i64, values_buf: *i64, 106 threshold_lo: i64, threshold_hi: i64) -> i64 { 107 if p.valid != 1 { return 0 - NX_MODPMP_NOT_INITIALIZED } 108 let rc: i64 = nx_pm_consumer_register(p.consumer, p.tracked_evt_kind, 109 times_buf, values_buf, 110 threshold_lo, threshold_hi) 111 if rc != NX_PM_OK { return rc } 112 // Also wire telemetry if the consumer doesn't yet have a bus. 113 if (p.telemetry as i64) != 0 { 114 nx_pm_consumer_set_telemetry_bus(p.consumer, p.telemetry) 115 } 116 return NX_MODPMP_OK 117} 118 119// ===== Build a poll request (RTU) ================================================= 120// 121// Convenience: builds an FC 03 read-holding-registers request for 122// the pipeline's tracked (unit_id, reg_addr) with quantity=1. 123// Operator sends the resulting bytes over the wire (out of scope). 124 125func nx_modpmp_build_rtu_poll(p: *NxModbusPMP, out: *u8, out_cap: i64) -> i64 { 126 if p.valid != 1 { return 0 - NX_MODPMP_NOT_INITIALIZED } 127 return nx_modbus_rtu_build_read_holding(p.shim, 128 p.tracked_unit_id, 129 p.tracked_reg_addr, 130 1, 131 out, out_cap) 132} 133 134// ===== Build a poll request (TCP) ================================================= 135 136func nx_modpmp_build_tcp_poll(p: *NxModbusPMP, txn_id: i64, 137 out: *u8, out_cap: i64) -> i64 { 138 if p.valid != 1 { return 0 - NX_MODPMP_NOT_INITIALIZED } 139 return nx_modbus_tcp_build_read_holding(p.shim, txn_id, 140 p.tracked_unit_id, 141 p.tracked_reg_addr, 142 1, 143 out, out_cap) 144} 145 146// ===== Process a response frame (RTU) ================================================= 147// 148// Parses the frame via the shim + dispatches the first register-read 149// event to the consumer. Consumer classifies + emits telemetry per 150// its own wiring (commit d3037668). 151 152func nx_modpmp_on_rtu_response(p: *NxModbusPMP, 153 frame: *u8, frame_len: i64, ts_ns: i64) -> i64 { 154 if p.valid != 1 { return 0 - NX_MODPMP_NOT_INITIALIZED } 155 156 let evt_kinds: *i64 = (sys_mmap(8 * 8)) as *i64 157 let evt_values: *i64 = (sys_mmap(8 * 8)) as *i64 158 let evt_units: *i64 = (sys_mmap(8 * 8)) as *i64 159 160 let n_evts: i64 = nx_modbus_rtu_parse(p.shim, frame, frame_len, 161 evt_kinds, evt_values, evt_units, 8) 162 if n_evts < 0 { return n_evts } // shim verdict passes through 163 if n_evts == 0 { return 0 - NX_MODPMP_NO_REG_EVENT } 164 165 // First event is the holding-register read (assumption: V1 polls 166 // one register, so response has one register event). 167 if evt_units[0] != p.tracked_unit_id { return 0 - NX_MODPMP_WRONG_UNIT_ID } 168 if evt_kinds[0] != NX_MODBUS_EVT_HOLDING_REG_READ { 169 if evt_kinds[0] != NX_MODBUS_EVT_INPUT_REG_READ { 170 return 0 - NX_MODPMP_NO_REG_EVENT 171 } 172 } 173 174 let stats: *NxPidStats = (sys_mmap(64)) as *NxPidStats 175 return nx_pm_consumer_on_event(p.consumer, p.tracked_evt_kind, 176 evt_values[0], ts_ns, stats) 177} 178 179// ===== Process a response frame (TCP) ================================================= 180 181func nx_modpmp_on_tcp_response(p: *NxModbusPMP, 182 frame: *u8, frame_len: i64, ts_ns: i64) -> i64 { 183 if p.valid != 1 { return 0 - NX_MODPMP_NOT_INITIALIZED } 184 185 let evt_kinds: *i64 = (sys_mmap(8 * 8)) as *i64 186 let evt_values: *i64 = (sys_mmap(8 * 8)) as *i64 187 let evt_units: *i64 = (sys_mmap(8 * 8)) as *i64 188 189 let n_evts: i64 = nx_modbus_tcp_parse(p.shim, frame, frame_len, 190 evt_kinds, evt_values, evt_units, 8) 191 if n_evts < 0 { return n_evts } 192 if n_evts == 0 { return 0 - NX_MODPMP_NO_REG_EVENT } 193 194 if evt_units[0] != p.tracked_unit_id { return 0 - NX_MODPMP_WRONG_UNIT_ID } 195 if evt_kinds[0] != NX_MODBUS_EVT_HOLDING_REG_READ { 196 if evt_kinds[0] != NX_MODBUS_EVT_INPUT_REG_READ { 197 return 0 - NX_MODPMP_NO_REG_EVENT 198 } 199 } 200 201 let stats: *NxPidStats = (sys_mmap(64)) as *NxPidStats 202 return nx_pm_consumer_on_event(p.consumer, p.tracked_evt_kind, 203 evt_values[0], ts_ns, stats) 204}