diff --git a/v2/crates/wifi-densepose-sensing-server/benches/mqtt_throughput.rs b/v2/crates/wifi-densepose-sensing-server/benches/mqtt_throughput.rs index da2df6337..047dd6bad 100644 --- a/v2/crates/wifi-densepose-sensing-server/benches/mqtt_throughput.rs +++ b/v2/crates/wifi-densepose-sensing-server/benches/mqtt_throughput.rs @@ -50,9 +50,9 @@ fn snap() -> VitalsSnapshot { node_id: "aabbccddeeff".into(), timestamp_ms: 1779_512_400_000, presence: true, - fall_detected: false, + fall_detected: Some(false), motion: 0.35, - motion_energy: 1234.5, + motion_energy: Some(1234.5), presence_score: 0.91, breathing_rate_bpm: Some(14.2), heartrate_bpm: Some(68.2), diff --git a/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs b/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs index b1a9ee0dd..5ca2c76ad 100644 --- a/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs +++ b/v2/crates/wifi-densepose-sensing-server/examples/mqtt_publisher.rs @@ -126,9 +126,9 @@ async fn main() -> Result<(), Box> { node_id: "example_node".into(), timestamp_ms: chrono::Utc::now().timestamp_millis(), presence: tick % 20 < 10, - fall_detected: tick % 60 == 30, + fall_detected: Some(tick % 60 == 30), motion: 0.10 + ((tick as f64).sin().abs() * 0.30), - motion_energy: 1000.0 + (tick as f64).cos() * 200.0, + motion_energy: Some(1000.0 + (tick as f64).cos() * 200.0), presence_score: 0.85, breathing_rate_bpm: Some(13.0 + ((tick as f64) * 0.05).sin()), heartrate_bpm: Some(68.0 + ((tick as f64) * 0.03).sin() * 5.0), diff --git a/v2/crates/wifi-densepose-sensing-server/src/cli.rs b/v2/crates/wifi-densepose-sensing-server/src/cli.rs index 80e4356bf..12145553c 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/cli.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/cli.rs @@ -29,7 +29,8 @@ pub struct MqttArgs { #[arg(long, default_value = "MQTT_PASSWORD")] pub mqtt_password_env: String, - /// MQTT client ID (default: wifi-densepose-) + /// MQTT client ID; prefixes every Home Assistant device id (default: a + /// `wifi-densepose-` id persisted in `/mqtt_client_id`) #[arg(long, env = "RUVIEW_MQTT_CLIENT_ID")] pub mqtt_client_id: Option, diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 180ae6f53..5f8b88f6b 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -10876,108 +10876,6 @@ async fn broadcast_tick_task(state: SharedState, tick_ms: u64) { } } -/// Map one sensing-broadcast JSON document into the `VitalsSnapshot`(s) to -/// publish over MQTT (issues #872/#898). -/// -/// Multi-node sources carry a `nodes` array where **each node has its own -/// `classification`** (`motion_level`, `presence`, `confidence`) and RSSI — so -/// each node must surface its *own* presence/motion, not the room-level -/// aggregate. Previously the bridge applied the aggregate `classification` to -/// every per-node Home-Assistant device, so a node in an empty corner inherited -/// another node's "present" (and `motion_level: "absent"` was mis-mapped to full -/// motion). Vitals (breathing / heart rate) and the person count are room-level -/// and shared across the per-node devices. Falls back to a single aggregate -/// snapshot when there is no per-node data (e.g. wifi / simulate sources). -#[cfg(feature = "mqtt")] -fn vitals_snapshots_from_sensing_json( - v: &serde_json::Value, - base_id: &str, -) -> Vec { - use wifi_densepose_sensing_server::mqtt::state::VitalsSnapshot; - - // motion_level string -> motion scalar. "absent"/"none"/"still"/"idle"/"" - // are non-moving; anything else (walking, …) is motion. `fallback` is used - // when the field is absent so a partial per-node payload defers to the - // room aggregate rather than silently reading 0. - fn motion_of(level: Option<&str>, fallback: f64) -> f64 { - match level { - Some("none") | Some("still") | Some("idle") | Some("absent") | Some("") => 0.0, - Some(_) => 1.0, - None => fallback, - } - } - - let ts = (v["timestamp"].as_f64().unwrap_or(0.0) * 1000.0) as i64; - let vit = &v["vital_signs"]; - let breathing = vit["breathing_rate_bpm"].as_f64(); - let hr = vit["heart_rate_bpm"].as_f64(); - let n_persons = v["persons"] - .as_array() - .map(|a| a.len() as u32) - .or_else(|| v["estimated_persons"].as_u64().map(|x| x as u32)) - .unwrap_or(0); - - // Room-level aggregate: the no-nodes fallback, and the per-node default for - // any field a node omits. - let acls = &v["classification"]; - let agg_presence = acls["presence"].as_bool().unwrap_or(false); - let agg_motion = motion_of(acls["motion_level"].as_str(), 0.0); - let agg_conf = acls["confidence"].as_f64().unwrap_or(0.0); - - let mk = |node_id: String, presence: bool, motion: f64, conf: f64, rssi: Option| { - VitalsSnapshot { - node_id, - timestamp_ms: ts, - presence, - motion, - presence_score: if presence { conf.max(0.0) } else { 0.0 }, - breathing_rate_bpm: breathing, - heartrate_bpm: hr, - n_persons, - rssi_dbm: rssi, - vital_confidence: conf, - ..Default::default() - } - }; - - match v["nodes"].as_array() { - Some(arr) if !arr.is_empty() => arr - .iter() - .map(|node| { - let n = node["node_id"].as_u64().unwrap_or(0); - // Each node carries its OWN classification under `node_inference` - // (ADR-297) — use it, deferring to the room aggregate only for - // fields the node omits. Issue #1541: this previously read a - // `"classification"` key that does not exist on `NodeInfo`'s - // serialized JSON (the field is `node_inference`), so every - // per-node lookup silently fell through to the room aggregate — - // per-node MQTT topics carried array-global values. - let ninf = &node["node_inference"]; - let presence = ninf["classification"] - .as_str() - .map(|c| c != "absent") - .unwrap_or(agg_presence); - let motion = motion_of(ninf["classification"].as_str(), agg_motion); - let conf = ninf["confidence"].as_f64().unwrap_or(agg_conf); - mk( - format!("{base_id}-node{n}"), - presence, - motion, - conf, - node["rssi_dbm"].as_f64(), - ) - }) - .collect(), - _ => vec![mk( - base_id.to_string(), - agg_presence, - agg_motion, - agg_conf, - v["nodes"][0]["rssi_dbm"].as_f64(), - )], - } -} - /// Build the multistatic guard config from the environment (#1031, #1049). /// /// Three precedence layers, most-specific wins: @@ -12052,7 +11950,10 @@ async fn main() { #[cfg(feature = "mqtt")] { use wifi_densepose_sensing_server::mqtt; - let mcfg = std::sync::Arc::new(mqtt::config::MqttConfig::from_args(&args.mqtt_opts)); + let mcfg = std::sync::Arc::new(mqtt::config::MqttConfig::from_args_with_data_dir( + &args.mqtt_opts, + &data_dir, + )); match mcfg.validate() { Ok(()) => { let node_id = mcfg.client_id.clone(); @@ -12069,16 +11970,16 @@ async fn main() { mqtt::publisher::spawn(mcfg, builder, vrx); let mut jrx = tx.subscribe(); tokio::spawn(async move { + // #898/#872/#2085: one snapshot per physical node from + // each `sensing_update`; `edge_vitals` feeds the fall + // detector; other message types are ignored. + let mut bridge = mqtt::bridge::SensingBridge::new(node_id); + let started = std::time::Instant::now(); while let Ok(json) = jrx.recv().await { let Ok(v) = serde_json::from_str::(&json) else { continue; }; - // #898/#872: emit one snapshot per physical node so - // each surfaces as its own Home-Assistant device with - // its *own* presence/motion/RSSI (see - // vitals_snapshots_from_sensing_json). Falls back to a - // single aggregate snapshot for per-node-less sources. - for snap in vitals_snapshots_from_sensing_json(&v, &node_id) { + for snap in bridge.ingest(&v, started.elapsed()) { let _ = vtx.send(snap); } } @@ -13316,114 +13217,6 @@ mod rolling_p95_tests { } } -#[cfg(all(test, feature = "mqtt"))] -mod mqtt_bridge_tests { - use super::vitals_snapshots_from_sensing_json; - use serde_json::json; - - /// Regression for the per-node presence bug (#872/#898, and its - /// resurgence as #1541): each node must surface its OWN classification, - /// not the room-level aggregate. Node 1 is present+moving; node 2 is - /// absent — node 2 must NOT inherit node 1's "present". - /// - /// The fixture below uses `nodes[].node_inference.classification` — the - /// field `NodeInfo` actually serializes (ADR-297) — not a bare - /// `nodes[].classification`. Issue #1541: an earlier version of this exact - /// test used the latter, non-existent shape, which the reader silently - /// treated as "field omitted" and fell back to the room aggregate for - /// every node. The test therefore passed while the real per-node MQTT - /// output was array-global — 100% line coverage of - /// `vitals_snapshots_from_sensing_json` with a fixture that didn't match - /// what `NodeInfo` actually serializes. Keep this fixture in the real - /// shape so this can't recur silently. - #[test] - fn per_node_presence_uses_each_nodes_own_classification() { - let v = json!({ - "timestamp": 1.0, - "classification": { "presence": true, "motion_level": "walking", "confidence": 0.9 }, - "vital_signs": { "breathing_rate_bpm": 14.0, "heart_rate_bpm": 60.0 }, - "persons": [{}, {}], - "nodes": [ - { "node_id": 1, "rssi_dbm": -40.0, - "node_inference": { "classification": "present_moving", "confidence": 0.8 } }, - { "node_id": 2, "rssi_dbm": -70.0, - "node_inference": { "classification": "absent", "confidence": 0.1 } } - ] - }); - let snaps = vitals_snapshots_from_sensing_json(&v, "ruview"); - assert_eq!(snaps.len(), 2, "one snapshot per node"); - - let n1 = snaps.iter().find(|s| s.node_id == "ruview-node1").unwrap(); - let n2 = snaps.iter().find(|s| s.node_id == "ruview-node2").unwrap(); - - assert!(n1.presence && n1.motion > 0.0, "node1 present + moving"); - assert!( - !n2.presence && n2.motion == 0.0, - "node2 must be absent — not inherit the room aggregate" - ); - // Per-node RSSI preserved. - assert_eq!(n1.rssi_dbm, Some(-40.0)); - assert_eq!(n2.rssi_dbm, Some(-70.0)); - // Vitals + person count are room-level, shared across node devices. - assert_eq!(n1.n_persons, 2); - assert_eq!(n2.n_persons, 2); - assert_eq!(n1.breathing_rate_bpm, Some(14.0)); - assert_eq!(n2.heartrate_bpm, Some(60.0)); - // presence_score is gated on presence. - assert!(n1.presence_score > 0.0); - assert_eq!(n2.presence_score, 0.0); - } - - /// A node that omits a classification field defers to the room aggregate - /// rather than silently reading false/0. - #[test] - fn per_node_missing_fields_fall_back_to_aggregate() { - let v = json!({ - "timestamp": 1.0, - "classification": { "presence": true, "motion_level": "still", "confidence": 0.7 }, - "vital_signs": {}, - "nodes": [ { "node_id": 3, "rssi_dbm": -55.0 } ] // no per-node classification - }); - let snaps = vitals_snapshots_from_sensing_json(&v, "n"); - assert_eq!(snaps.len(), 1); - assert_eq!(snaps[0].node_id, "n-node3"); - assert!(snaps[0].presence, "defers to aggregate presence"); - assert_eq!(snaps[0].motion, 0.0, "aggregate 'still' => no motion"); - } - - /// No `nodes` array (wifi / simulate sources): single aggregate snapshot - /// keyed by the base id. - #[test] - fn falls_back_to_single_aggregate_when_no_nodes() { - let v = json!({ - "timestamp": 2.0, - "classification": { "presence": true, "motion_level": "idle", "confidence": 0.6 }, - "vital_signs": { "breathing_rate_bpm": 12.0 }, - "persons": [{}] - }); - let snaps = vitals_snapshots_from_sensing_json(&v, "ruview"); - assert_eq!(snaps.len(), 1); - assert_eq!(snaps[0].node_id, "ruview"); - assert!(snaps[0].presence); - assert_eq!(snaps[0].motion, 0.0, "idle => no motion"); - assert_eq!(snaps[0].n_persons, 1); - } - - /// `motion_level: "absent"` must map to zero motion (the old aggregate - /// match fell through to `Some(_) => 1.0`, treating absent as full motion). - #[test] - fn absent_motion_level_is_zero_motion() { - let v = json!({ - "timestamp": 0.0, - "classification": { "presence": false, "motion_level": "absent", "confidence": 0.0 }, - "vital_signs": {} - }); - let snaps = vitals_snapshots_from_sensing_json(&v, "x"); - assert_eq!(snaps[0].motion, 0.0); - assert!(!snaps[0].presence); - } -} - #[cfg(test)] mod model_load_diagnostic_tests { use super::{diagnose_model_load_error, load_or_convert_model}; diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/bridge.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/bridge.rs new file mode 100644 index 000000000..4cc264f39 --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/bridge.rs @@ -0,0 +1,417 @@ +//! Sensing-broadcast JSON → [`VitalsSnapshot`] bridge (issues #872, #898, +//! #1541, #2085). +//! +//! `sensing-server` broadcasts several JSON message types on one channel. +//! Only two of them carry data the MQTT publisher can use: +//! +//! - `sensing_update` — one snapshot per physical node (`nodes[]`), or a +//! single aggregate snapshot when the source has no per-node data. +//! - `edge_vitals` — the ESP32 edge tier's per-node frame. Its +//! `fall_detected` flag is the only fall source the server has; the bridge +//! turns its rising edge into a one-shot fall on that node's next snapshot. +//! +//! Every other message type (`edge_fused_vitals`, `wasm_event`, …) is +//! ignored. Before #2085 every message was read as a `sensing_update`, so an +//! `edge_vitals` frame produced a phantom aggregate device whose presence +//! flickered with each frame. +//! +//! Absent fields stay absent (`None`) so the publisher can mark the matching +//! entity unavailable instead of publishing a default value. + +use std::collections::HashMap; +use std::time::Duration; + +use serde_json::Value; + +use super::state::VitalsSnapshot; + +/// A node whose last `edge_vitals` frame is older than this no longer counts +/// as having a fall detector; its fall entity goes unavailable. +pub const EDGE_FALL_STALE_AFTER: Duration = Duration::from_secs(30); + +#[derive(Debug, Default)] +struct EdgeFall { + last_seen: Duration, + prev: bool, + pending: bool, +} + +/// Stateful bridge. Holds the per-node fall edge detector, so one instance +/// lives for the lifetime of the publisher. +#[derive(Debug)] +pub struct SensingBridge { + base_id: String, + edge: HashMap, +} + +impl SensingBridge { + /// `base_id` prefixes every node id (`-node`); it is the MQTT + /// client id, so it must be stable across restarts (#2093). + pub fn new(base_id: impl Into) -> Self { + Self { + base_id: base_id.into(), + edge: HashMap::new(), + } + } + + /// Map one broadcast message into zero or more snapshots. `now` is a + /// monotonic time since the bridge started. + pub fn ingest(&mut self, v: &Value, now: Duration) -> Vec { + match v["type"].as_str() { + Some("sensing_update") => self.sensing_update(v, now), + Some("edge_vitals") => { + self.edge_vitals(v, now); + Vec::new() + } + _ => Vec::new(), + } + } + + fn edge_vitals(&mut self, v: &Value, now: Duration) { + let Some(n) = v["node_id"].as_u64() else { + return; + }; + let fall = v["fall_detected"].as_bool().unwrap_or(false); + let e = self.edge.entry(n).or_default(); + e.last_seen = now; + if fall && !e.prev { + e.pending = true; + } + e.prev = fall; + } + + /// `Some(true)` once per fall rising edge, `Some(false)` while the node's + /// edge tier is reporting, `None` when it has no fall source. + fn take_fall(&mut self, node: u64, now: Duration) -> Option { + let e = self.edge.get_mut(&node)?; + if now.saturating_sub(e.last_seen) >= EDGE_FALL_STALE_AFTER { + return None; + } + Some(std::mem::take(&mut e.pending)) + } + + fn sensing_update(&mut self, v: &Value, now: Duration) -> Vec { + let ts = (v["timestamp"].as_f64().unwrap_or(0.0) * 1000.0) as i64; + let vit = &v["vital_signs"]; + let breathing = vit["breathing_rate_bpm"].as_f64(); + let hr = vit["heart_rate_bpm"].as_f64(); + // Confidence of the vitals actually present, not the presence + // classifier's confidence. + let vital_confidence = [ + breathing.and(vit["breathing_confidence"].as_f64()), + hr.and(vit["heartbeat_confidence"].as_f64()), + ] + .into_iter() + .flatten() + .fold(None, |acc: Option, c| { + Some(acc.map_or(c, |a| a.min(c))) + }) + .unwrap_or(0.0); + let n_persons = v["persons"] + .as_array() + .map(|a| a.len() as u32) + .or_else(|| v["estimated_persons"].as_u64().map(|x| x as u32)) + .unwrap_or(0); + + // Room-level aggregate: the no-nodes fallback, and the per-node default + // for any field a node omits. + let acls = &v["classification"]; + let agg_presence = acls["presence"].as_bool().unwrap_or(false); + let agg_motion = motion_of(acls["motion_level"].as_str(), 0.0); + let agg_conf = acls["confidence"].as_f64().unwrap_or(0.0); + let agg_energy = v["features"]["motion_band_power"].as_f64(); + + let mk = |node_id: String, + presence: bool, + motion: f64, + conf: f64, + rssi: Option, + motion_energy: Option, + fall_detected: Option| VitalsSnapshot { + node_id, + timestamp_ms: ts, + presence, + fall_detected, + motion, + motion_energy, + presence_score: if presence { conf.max(0.0) } else { 0.0 }, + breathing_rate_bpm: breathing, + heartrate_bpm: hr, + n_persons, + rssi_dbm: rssi, + vital_confidence, + }; + + let Some(arr) = v["nodes"].as_array().filter(|a| !a.is_empty()) else { + return vec![mk( + self.base_id.clone(), + agg_presence, + agg_motion, + agg_conf, + None, + agg_energy, + None, + )]; + }; + let node_features = v["node_features"].as_array(); + let mut out = Vec::with_capacity(arr.len()); + for node in arr { + let n = node["node_id"].as_u64().unwrap_or(0); + // Each node carries its OWN classification under `node_inference` + // (ADR-297, issue #1541); defer to the room aggregate only for + // fields the node omits. + let ninf = &node["node_inference"]; + let presence = ninf["classification"] + .as_str() + .map(|c| c != "absent") + .unwrap_or(agg_presence); + let motion = motion_of(ninf["classification"].as_str(), agg_motion); + let conf = ninf["confidence"].as_f64().unwrap_or(agg_conf); + let energy = node_features + .and_then(|nf| nf.iter().find(|f| f["node_id"].as_u64() == Some(n))) + .and_then(|f| f["features"]["motion_band_power"].as_f64()) + .or(agg_energy); + let fall = self.take_fall(n, now); + out.push(mk( + format!("{}-node{n}", self.base_id), + presence, + motion, + conf, + node["rssi_dbm"].as_f64(), + energy, + fall, + )); + } + out + } +} + +/// motion_level string → motion scalar. `absent` / `present_still` and the +/// legacy `none`/`still`/`idle`/`""` are non-moving; anything else (e.g. +/// `present_moving`, `walking`) is motion. `fallback` is used when the field +/// is absent so a partial per-node payload defers to the room aggregate. +fn motion_of(level: Option<&str>, fallback: f64) -> f64 { + match level { + Some("none") + | Some("still") + | Some("idle") + | Some("absent") + | Some("present_still") + | Some("") => 0.0, + Some(_) => 1.0, + None => fallback, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + const T0: Duration = Duration::from_secs(100); + + fn snaps(v: Value, base: &str) -> Vec { + SensingBridge::new(base).ingest(&v, T0) + } + + /// Regression for #872/#898/#1541: each node surfaces its OWN + /// classification from `nodes[].node_inference` (the shape `NodeInfo` + /// actually serializes), not the room aggregate. + #[test] + fn per_node_presence_uses_each_nodes_own_classification() { + let v = json!({ + "type": "sensing_update", + "timestamp": 1.0, + "classification": { "presence": true, "motion_level": "present_moving", "confidence": 0.9 }, + "vital_signs": { "breathing_rate_bpm": 14.0, "heart_rate_bpm": 60.0, + "breathing_confidence": 0.8, "heartbeat_confidence": 0.7 }, + "persons": [{}, {}], + "nodes": [ + { "node_id": 1, "rssi_dbm": -40.0, + "node_inference": { "classification": "present_moving", "confidence": 0.8 } }, + { "node_id": 2, "rssi_dbm": -70.0, + "node_inference": { "classification": "absent", "confidence": 0.1 } } + ] + }); + let s = snaps(v, "ruview"); + assert_eq!(s.len(), 2, "one snapshot per node"); + let n1 = s.iter().find(|s| s.node_id == "ruview-node1").unwrap(); + let n2 = s.iter().find(|s| s.node_id == "ruview-node2").unwrap(); + assert!(n1.presence && n1.motion > 0.0); + assert!( + !n2.presence && n2.motion == 0.0, + "node2 must not inherit the aggregate" + ); + assert_eq!(n1.rssi_dbm, Some(-40.0)); + assert_eq!(n2.rssi_dbm, Some(-70.0)); + // Vitals + person count are room-level, shared across node devices. + assert_eq!((n1.n_persons, n2.n_persons), (2, 2)); + assert_eq!(n1.breathing_rate_bpm, Some(14.0)); + assert_eq!(n2.heartrate_bpm, Some(60.0)); + // Vital confidence is the weakest present vital, not the classifier's. + assert_eq!(n1.vital_confidence, 0.7); + assert!(n1.presence_score > 0.0); + assert_eq!(n2.presence_score, 0.0); + } + + #[test] + fn present_still_is_presence_without_motion() { + let v = json!({ + "type": "sensing_update", "timestamp": 1.0, + "classification": { "presence": true, "motion_level": "present_moving", "confidence": 0.9 }, + "nodes": [ { "node_id": 4, "rssi_dbm": -50.0, + "node_inference": { "classification": "present_still", "confidence": 0.6 } } ] + }); + let s = snaps(v, "b"); + assert!(s[0].presence); + assert_eq!(s[0].motion, 0.0); + } + + #[test] + fn per_node_missing_fields_fall_back_to_aggregate() { + let v = json!({ + "type": "sensing_update", "timestamp": 1.0, + "classification": { "presence": true, "motion_level": "still", "confidence": 0.7 }, + "vital_signs": {}, + "nodes": [ { "node_id": 3, "rssi_dbm": -55.0 } ] + }); + let s = snaps(v, "n"); + assert_eq!(s.len(), 1); + assert_eq!(s[0].node_id, "n-node3"); + assert!(s[0].presence, "defers to aggregate presence"); + assert_eq!(s[0].motion, 0.0); + } + + #[test] + fn falls_back_to_single_aggregate_when_no_nodes() { + let v = json!({ + "type": "sensing_update", "timestamp": 2.0, + "classification": { "presence": true, "motion_level": "idle", "confidence": 0.6 }, + "vital_signs": { "breathing_rate_bpm": 12.0 }, + "persons": [{}] + }); + let s = snaps(v, "ruview"); + assert_eq!(s.len(), 1); + assert_eq!(s[0].node_id, "ruview"); + assert!(s[0].presence); + assert_eq!(s[0].n_persons, 1); + assert_eq!( + s[0].rssi_dbm, None, + "no per-node RSSI on an aggregate frame" + ); + assert_eq!( + s[0].fall_detected, None, + "no fall source on an aggregate frame" + ); + } + + #[test] + fn absent_motion_level_is_zero_motion() { + let v = json!({ + "type": "sensing_update", "timestamp": 0.0, + "classification": { "presence": false, "motion_level": "absent", "confidence": 0.0 } + }); + let s = snaps(v, "x"); + assert_eq!(s[0].motion, 0.0); + assert!(!s[0].presence); + } + + /// #2085: `edge_vitals` and other message types must not be read as a + /// `sensing_update` (that produced a phantom aggregate device). + #[test] + fn non_sensing_messages_produce_no_snapshot() { + let mut b = SensingBridge::new("x"); + for t in [ + "edge_vitals", + "edge_fused_vitals", + "wasm_event", + "pose_data", + ] { + let v = json!({ "type": t, "node_id": 1, "presence": true }); + assert!( + b.ingest(&v, T0).is_empty(), + "{t} must not become a snapshot" + ); + } + assert!( + b.ingest(&json!({ "presence": true }), T0).is_empty(), + "untyped" + ); + } + + #[test] + fn motion_energy_prefers_per_node_band_power() { + let v = json!({ + "type": "sensing_update", "timestamp": 1.0, + "classification": { "presence": true, "motion_level": "present_moving", "confidence": 0.9 }, + "features": { "motion_band_power": 9.0 }, + "nodes": [ { "node_id": 1 }, { "node_id": 2 } ], + "node_features": [ { "node_id": 1, "features": { "motion_band_power": 3.5 } } ] + }); + let s = snaps(v, "e"); + assert_eq!( + s[0].motion_energy, + Some(3.5), + "node 1 uses its own band power" + ); + assert_eq!( + s[1].motion_energy, + Some(9.0), + "node 2 falls back to the aggregate" + ); + + let none = json!({ "type": "sensing_update", "timestamp": 1.0, "classification": {} }); + assert_eq!(snaps(none, "e")[0].motion_energy, None, "never a default 0"); + } + + #[test] + fn edge_fall_rising_edge_fires_once_on_that_node() { + let mut b = SensingBridge::new("f"); + let upd = json!({ + "type": "sensing_update", "timestamp": 1.0, "classification": {}, + "nodes": [ { "node_id": 1 }, { "node_id": 2 } ] + }); + let edge = + |fall: bool| json!({ "type": "edge_vitals", "node_id": 1, "fall_detected": fall }); + + // No edge tier yet: no fall source on either node. + let s = b.ingest(&upd, T0); + assert_eq!((s[0].fall_detected, s[1].fall_detected), (None, None)); + + b.ingest(&edge(false), T0); + assert_eq!(b.ingest(&upd, T0)[0].fall_detected, Some(false)); + + // Rising edge → exactly one Some(true), on node 1 only. + b.ingest(&edge(true), T0); + let s = b.ingest(&upd, T0); + assert_eq!(s[0].fall_detected, Some(true)); + assert_eq!(s[1].fall_detected, None); + assert_eq!(b.ingest(&upd, T0)[0].fall_detected, Some(false), "consumed"); + + // A held flag does not re-fire; a new edge after it clears does. + b.ingest(&edge(true), T0); + assert_eq!(b.ingest(&upd, T0)[0].fall_detected, Some(false)); + b.ingest(&edge(false), T0); + b.ingest(&edge(true), T0); + assert_eq!(b.ingest(&upd, T0)[0].fall_detected, Some(true)); + } + + #[test] + fn edge_fall_source_goes_stale() { + let mut b = SensingBridge::new("f"); + b.ingest( + &json!({ "type": "edge_vitals", "node_id": 1, "fall_detected": false }), + T0, + ); + let upd = json!({ "type": "sensing_update", "timestamp": 1.0, "classification": {}, + "nodes": [ { "node_id": 1 } ] }); + assert_eq!( + b.ingest(&upd, T0 + Duration::from_secs(29))[0].fall_detected, + Some(false) + ); + assert_eq!( + b.ingest(&upd, T0 + EDGE_FALL_STALE_AFTER)[0].fall_detected, + None + ); + } +} diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs index a430125b4..2b0308144 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/config.rs @@ -1,6 +1,9 @@ //! Runtime configuration for the MQTT publisher, built from CLI args. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; + +/// File in `--data-dir` holding the persisted default MQTT client id. +pub const CLIENT_ID_FILE: &str = "mqtt_client_id"; /// All knobs the MQTT publisher needs. Built by [`MqttConfig::from_args`] /// after [`crate::cli::Args`] parsing. @@ -59,10 +62,9 @@ impl MqttConfig { /// Build an [`MqttConfig`] from parsed [`crate::cli::Args`]. /// /// Reads `mqtt_password_env` to resolve the broker password from the - /// environment so secrets never appear on the command line. Reads - /// `hostname()` via the `gethostname` crate if `mqtt_client_id` was - /// not supplied — we don't add a dep here, we let the publisher - /// supply the default lazily. + /// environment so secrets never appear on the command line. Without + /// `--mqtt-client-id` the client id is per-process; the server uses + /// [`MqttConfig::from_args_with_data_dir`] for a stable one. pub fn from_args(args: &crate::cli::MqttArgs) -> Self { let password = std::env::var(&args.mqtt_password_env).ok(); let port = args.mqtt_port.unwrap_or(if args.mqtt_tls { 8883 } else { 1883 }); @@ -70,10 +72,7 @@ impl MqttConfig { let client_id = args .mqtt_client_id .clone() - .unwrap_or_else(|| { - // Avoid a `gethostname` dep in P1 — fallback only. - format!("wifi-densepose-{}", std::process::id()) - }); + .unwrap_or_else(|| format!("wifi-densepose-{}", std::process::id())); Self { host: args.mqtt_host.clone(), @@ -96,6 +95,27 @@ impl MqttConfig { } } + /// [`MqttConfig::from_args`], with the default client id persisted in + /// `data_dir` so it survives restarts (#2093). The client id prefixes + /// every HA device id, so a per-process default created a new set of HA + /// devices on each restart. An explicit `--mqtt-client-id` still wins. + pub fn from_args_with_data_dir(args: &crate::cli::MqttArgs, data_dir: &Path) -> Self { + let mut cfg = Self::from_args(args); + if args.mqtt_client_id.is_none() { + match load_or_create_client_id(data_dir) { + Ok(id) => cfg.client_id = id, + Err(e) => tracing::warn!( + "[mqtt] could not persist a client id in {}: {e}; using {} for this run, \ + so Home Assistant device ids will change on restart. Set --mqtt-client-id \ + to pin them.", + data_dir.display(), + cfg.client_id, + ), + } + } + cfg + } + /// True iff this config is safe to start. Pre-flight validation that /// runs before any network I/O so users get a clean error instead of /// a connect failure 30 s later. @@ -135,6 +155,38 @@ impl MqttConfig { } } +/// Read the persisted client id from `/mqtt_client_id`, or create +/// one (`wifi-densepose-<8 hex>`, 23 bytes — the MQTT 3.1.1 guaranteed +/// maximum). The file is untrusted input: a missing, empty or malformed id +/// is replaced rather than sent to the broker. +pub fn load_or_create_client_id(data_dir: &Path) -> std::io::Result { + let path = data_dir.join(CLIENT_ID_FILE); + match std::fs::read_to_string(&path) { + Ok(s) if is_valid_client_id(s.trim()) => return Ok(s.trim().to_string()), + Ok(_) => tracing::warn!( + "[mqtt] {} does not hold a valid client id; replacing it", + path.display() + ), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(e), + } + let id = format!("wifi-densepose-{:08x}", rand::random::()); + std::fs::create_dir_all(data_dir)?; + let tmp = data_dir.join(format!("{CLIENT_ID_FILE}.tmp")); + std::fs::write(&tmp, format!("{id}\n"))?; + std::fs::rename(&tmp, &path)?; + Ok(id) +} + +/// 1–64 bytes of `[A-Za-z0-9_-]`: safe in a client id, an HA `unique_id` +/// and an MQTT topic level. +fn is_valid_client_id(id: &str) -> bool { + (1..=64).contains(&id.len()) + && id + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') +} + fn build_tls(args: &crate::cli::MqttArgs) -> TlsConfig { if !args.mqtt_tls { return TlsConfig::Off; @@ -287,6 +339,60 @@ mod tests { assert!(matches!(cfg.validate(), Err(MqttConfigError::InvalidRate(_)))); } + #[test] + fn default_client_id_is_stable_across_restarts() { + let dir = tempfile::tempdir().unwrap(); + // Two "restarts": each builds its config from scratch. + let first = MqttConfig::from_args_with_data_dir(&parse(&[]), dir.path()); + let second = MqttConfig::from_args_with_data_dir(&parse(&[]), dir.path()); + assert_eq!(first.client_id, second.client_id); + assert!(first.client_id.starts_with("wifi-densepose-")); + assert_eq!(first.client_id.len(), 23); + let stored = std::fs::read_to_string(dir.path().join(CLIENT_ID_FILE)).unwrap(); + assert_eq!(stored.trim(), first.client_id); + // A different install gets a different id. + let other = tempfile::tempdir().unwrap(); + let third = MqttConfig::from_args_with_data_dir(&parse(&[]), other.path()); + assert_ne!(first.client_id, third.client_id); + } + + #[test] + fn explicit_client_id_wins_and_is_not_persisted() { + let dir = tempfile::tempdir().unwrap(); + let cfg = MqttConfig::from_args_with_data_dir( + &parse(&["--mqtt-client-id", "kitchen"]), + dir.path(), + ); + assert_eq!(cfg.client_id, "kitchen"); + assert!(!dir.path().join(CLIENT_ID_FILE).exists()); + } + + #[test] + fn malformed_persisted_client_id_is_replaced() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(CLIENT_ID_FILE); + for bad in ["", "has space", "a/b", "+", &"x".repeat(65)] { + std::fs::write(&path, bad).unwrap(); + let id = load_or_create_client_id(dir.path()).unwrap(); + assert!(is_valid_client_id(&id), "{bad:?} -> {id:?}"); + assert_eq!(std::fs::read_to_string(&path).unwrap().trim(), id); + } + std::fs::write(&path, "my-node_1\n").unwrap(); + assert_eq!(load_or_create_client_id(dir.path()).unwrap(), "my-node_1"); + } + + #[test] + fn unwritable_data_dir_falls_back_to_a_per_process_id() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("not-a-dir"); + std::fs::write(&file, "x").unwrap(); + let cfg = MqttConfig::from_args_with_data_dir(&parse(&[]), &file); + assert_eq!( + cfg.client_id, + format!("wifi-densepose-{}", std::process::id()) + ); + } + #[test] fn password_env_resolution() { std::env::set_var("RUVIEW_TEST_MQTT_PW", "s3cret"); diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/discovery.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/discovery.rs index e815d6a16..9eba45163 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mqtt/discovery.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/discovery.rs @@ -214,6 +214,28 @@ impl EntityKind { ) } + /// False for entities the sensing broadcast has no source for. They are + /// not announced, so Home Assistant never shows an entity that can only + /// stay "unknown" (issue #2085): + /// + /// - zone occupancy, bathroom occupied, bed exit, room transition: the + /// broadcast carries no zones, and nothing loads a zone-tag map; + /// - meeting in progress: needs a 1–20 % motion level, but per-node motion + /// is a three-level classification (absent / still / moving); + /// - pose: the broadcast's `pose_keypoints` is never populated, and the + /// `persons[].keypoints` skeletons are synthesized, not measured. + pub fn has_server_source(self) -> bool { + !matches!( + self, + EntityKind::ZoneOccupancy + | EntityKind::BathroomOccupied + | EntityKind::BedExit + | EntityKind::MultiRoomTransition + | EntityKind::MeetingInProgress + | EntityKind::PoseKeypoints + ) + } + /// Human-readable HA entity name shown in the UI. pub fn display_name(self) -> &'static str { match self { @@ -467,6 +489,9 @@ impl<'a> DiscoveryBuilder<'a> { all.into_iter() .filter(|e| { + if !e.has_server_source() { + return false; + } if privacy_mode && e.is_biometric() { return false; } @@ -572,12 +597,30 @@ mod tests { } #[test] - fn enabled_entities_default_excludes_pose_and_includes_all_others() { + fn enabled_entities_default_includes_every_sourced_entity() { let entities = DiscoveryBuilder::enabled_entities(false, false, &[]); assert!(!entities.contains(&EntityKind::PoseKeypoints)); assert!(entities.contains(&EntityKind::Presence)); assert!(entities.contains(&EntityKind::HeartRate)); assert!(entities.contains(&EntityKind::SomeoneSleeping)); + assert_eq!(entities.len(), 15); + } + + #[test] + fn unsourced_entities_are_never_announced() { + // #2085: announcing these left them "unknown" in HA forever. + let entities = DiscoveryBuilder::enabled_entities(false, true, &[]); + for e in [ + EntityKind::ZoneOccupancy, + EntityKind::BathroomOccupied, + EntityKind::BedExit, + EntityKind::MultiRoomTransition, + EntityKind::MeetingInProgress, + EntityKind::PoseKeypoints, + ] { + assert!(!e.has_server_source()); + assert!(!entities.contains(&e), "{e:?} announced without a source"); + } } #[test] @@ -588,7 +631,7 @@ mod tests { } // Semantic primitives must remain available (ADR-115 §3.12.3). assert!(entities.contains(&EntityKind::SomeoneSleeping)); - assert!(entities.contains(&EntityKind::BathroomOccupied)); + assert!(entities.contains(&EntityKind::PossibleDistress)); } #[test] diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/mod.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/mod.rs index 8d125e65c..8c26fd633 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mqtt/mod.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/mod.rs @@ -22,24 +22,27 @@ //! //! - [`discovery`] — HA discovery payload generators per entity type //! - [`state`] — per-entity state-message encoders + rate limiter +//! - [`bridge`] — sensing-broadcast JSON → [`state::VitalsSnapshot`] +//! - [`planner`] — entity → state/availability plan + semantic primitives //! - [`publisher`] — connection lifecycle + topic publication //! - [`privacy`] — biometric stripping per `--privacy-mode` //! - [`config`] — `MqttConfig` struct fed by [`crate::cli::Args`] //! //! ## Cross-protocol coupling //! -//! The semantic inference layer (ADR-115 §3.12, future `crate::semantic`) -//! emits primitive state changes onto a `tokio::broadcast` channel that -//! this module also subscribes to. Same channel is consumed by the Matter -//! Bridge (ADR-115 §3.11, future `crate::matter`), so adding a new -//! semantic primitive automatically flows to all surfaces. +//! The semantic inference layer ([`crate::semantic`], ADR-115 §3.12) runs +//! inside [`planner`], one bus per node, fed by the same snapshots as the +//! raw entities (ADR-115 P4.5). The Matter bridge (ADR-115 §3.11) does not +//! consume semantic states yet. +pub mod bridge; pub mod config; pub mod discovery; +pub mod planner; pub mod privacy; pub mod security; -// State encoders + rate limiter compile without rumqttc, so they're -// available for testing under `--no-default-features`. Only the +// State encoders, the bridge and the planner compile without rumqttc, so +// they're available for testing under `--no-default-features`. Only the // publisher itself (which holds the `rumqttc::AsyncClient`) needs the // `mqtt` feature. pub mod state; diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/mod.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/mod.rs new file mode 100644 index 000000000..5872e8087 --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/mod.rs @@ -0,0 +1,321 @@ +//! Entity → state publication plan (ADR-115 §3.5–§3.7, §3.12; issues +//! #2085, #2093). +//! +//! The planner turns each [`VitalsSnapshot`] into the MQTT messages to +//! publish: per-entity availability, change-gated binary states, rate-limited +//! sensor values, one-shot events, and the semantic primitives (HA-MIND, +//! ADR-115 P4.5). It is pure — no broker, no clock — so the whole mapping is +//! unit-testable; the publisher only sends what it returns. +//! +//! An entity is published only while the server has a real source for it. +//! Otherwise its availability topic carries `offline` and no state is sent, +//! so Home Assistant shows "unavailable" rather than a made-up value. + +use std::collections::HashMap; +use std::time::Duration; + +use crate::semantic::{PrimitiveConfig, RawSnapshot, SemanticBus, SemanticKind}; + +use super::config::{MqttConfig, PublishRates}; +use super::discovery::{DiscoveryBuilder, DiscoveryComponent, EntityKind}; +use super::privacy::{decide, PublishDecision}; +use super::state::{RateLimiter, StateEncoder, StateMessage, VitalsSnapshot}; + +/// Unchanged binary states are re-published at this interval (retained), so +/// a broker that lost its retained copy recovers without a state flip. +pub const BINARY_STATE_HEARTBEAT: Duration = Duration::from_secs(60); + +/// An optional input (vitals, RSSI, fall detector, motion energy) not seen +/// for this long no longer counts as a source; its entity goes unavailable. +pub const SOURCE_STALE_AFTER: Duration = Duration::from_secs(30); + +/// Semantic FSMs are sampled at 1 Hz per node. Their constants (e.g. the +/// distress EWMA's "~100-sample memory at 1 Hz") assume that rate, while the +/// sensing broadcast runs at the CSI frame rate. +pub const SEMANTIC_TICK: Duration = Duration::from_secs(1); + +/// On-change gate with heartbeat for retained binary states (#2093). +#[derive(Debug, Default)] +pub struct ChangeGate { + last: HashMap<(String, EntityKind), (bool, Duration)>, +} + +impl ChangeGate { + /// True when `value` differs from the last published value for this + /// `(node, entity)`, or `heartbeat` has elapsed since that publish. + pub fn allow( + &mut self, + node_id: &str, + entity: EntityKind, + value: bool, + now: Duration, + heartbeat: Duration, + ) -> bool { + let key = (node_id.to_string(), entity); + if let Some(&(prev, at)) = self.last.get(&key) { + if prev == value && now.saturating_sub(at) < heartbeat { + return false; + } + } + self.last.insert(key, (value, now)); + true + } + + pub fn reset(&mut self) { + self.last.clear(); + } +} + +struct NodePlan { + bus: SemanticBus, + last_tick: Option, + pending_fall: bool, + sourced: HashMap, + availability: HashMap, +} + +/// Per-publisher planning state, keyed by node. +pub struct PublishPlanner { + entities: Vec, + rates: PublishRates, + privacy_mode: bool, + semantic: PrimitiveConfig, + nodes: HashMap, + rate_limiter: RateLimiter, + gate: ChangeGate, +} + +impl PublishPlanner { + pub fn new(cfg: &MqttConfig) -> Self { + Self::with_semantic_config(cfg, PrimitiveConfig::default()) + } + + pub fn with_semantic_config(cfg: &MqttConfig, semantic: PrimitiveConfig) -> Self { + Self { + entities: DiscoveryBuilder::enabled_entities(cfg.privacy_mode, cfg.publish_pose, &[]), + rates: cfg.rates, + privacy_mode: cfg.privacy_mode, + semantic, + nodes: HashMap::new(), + rate_limiter: RateLimiter::new(), + gate: ChangeGate::default(), + } + } + + /// Entities announced via discovery. Same list drives state + availability. + pub fn entities(&self) -> &[EntityKind] { + &self.entities + } + + /// Forget publish history after a reconnect so every state and + /// availability is re-sent promptly. + pub fn reset(&mut self) { + self.rate_limiter.reset(); + self.gate.reset(); + for np in self.nodes.values_mut() { + np.availability.clear(); + } + } + + /// Messages for one snapshot. `now` is time since the publisher started + /// (it is also the semantic warmup clock). + pub fn on_snapshot( + &mut self, + b: &DiscoveryBuilder<'_>, + snap: &VitalsSnapshot, + now: Duration, + ) -> Vec { + let node = snap.node_id.as_str(); + let np = self + .nodes + .entry(snap.node_id.clone()) + .or_insert_with(|| NodePlan { + bus: SemanticBus::new(self.semantic.clone()), + last_tick: None, + pending_fall: false, + sourced: HashMap::new(), + availability: HashMap::new(), + }); + + for (entity, present) in [ + (EntityKind::MotionEnergy, snap.motion_energy.is_some()), + (EntityKind::Rssi, snap.rssi_dbm.is_some()), + (EntityKind::BreathingRate, snap.breathing_rate_bpm.is_some()), + (EntityKind::HeartRate, snap.heartrate_bpm.is_some()), + (EntityKind::FallDetected, snap.fall_detected.is_some()), + ] { + if present { + np.sourced.insert(entity, now); + } + } + + np.pending_fall |= snap.fall_detected == Some(true); + if np + .last_tick + .is_none_or(|t| now.saturating_sub(t) >= SEMANTIC_TICK) + { + let raw = raw_snapshot(snap, now, np.pending_fall); + np.bus.tick(&raw); + np.last_tick = Some(now); + np.pending_fall = false; + } + + let mut out = self.availability(b, node, true, now, false); + let np = &self.nodes[node]; + let enc = StateEncoder { builder: b }; + for &e in &self.entities { + if decide(e, self.privacy_mode) == PublishDecision::Suppress + || !entity_available(np, e, now, &self.semantic) + { + continue; + } + match e.component() { + DiscoveryComponent::BinarySensor => { + let on = match e { + EntityKind::Presence => Some(snap.presence), + _ => semantic_kind(e).and_then(|k| np.bus.is_active(k)), + }; + if let Some(on) = on { + if self.gate.allow(node, e, on, now, BINARY_STATE_HEARTBEAT) { + out.extend(enc.boolean(e, on)); + } + } + } + DiscoveryComponent::Event => { + if e == EntityKind::FallDetected && snap.fall_detected == Some(true) { + out.extend(enc.event(e, "fall_detected", snap.timestamp_ms, None)); + } + } + DiscoveryComponent::Sensor => { + let msg = match e { + EntityKind::FallRiskElevated => { + enc.score(e, np.bus.fall_risk_score(), snap.timestamp_ms) + } + _ => enc.numeric(e, snap), + }; + if let Some(m) = msg { + if self.rate_limiter.allow(node, e, now, &self.rates) { + out.push(m); + } + } + } + } + } + out + } + + /// Availability messages for one node. With `force`, every entity is + /// re-published (heartbeat); otherwise only transitions are. A node + /// that is not `node_fresh` reports every entity `offline`. + pub fn availability( + &mut self, + b: &DiscoveryBuilder<'_>, + node_id: &str, + node_fresh: bool, + now: Duration, + force: bool, + ) -> Vec { + let Some(np) = self.nodes.get_mut(node_id) else { + return Vec::new(); + }; + let mut out = Vec::new(); + for &e in &self.entities { + let want = node_fresh && entity_available(np, e, now, &self.semantic); + if force || np.availability.get(&e) != Some(&want) { + np.availability.insert(e, want); + out.push(availability_message(b, e, want)); + } + } + out + } + + /// `offline` for every entity of a node (shutdown path). + pub fn offline(&self, b: &DiscoveryBuilder<'_>) -> Vec { + self.entities + .iter() + .map(|&e| availability_message(b, e, false)) + .collect() + } +} + +fn availability_message(b: &DiscoveryBuilder<'_>, e: EntityKind, online: bool) -> StateMessage { + StateMessage { + topic: b.availability_topic(e), + payload: if online { "online" } else { "offline" }.to_string(), + qos: 1, + retain: true, + } +} + +/// Whether the server currently has a real source for `e` on this node. +fn entity_available(np: &NodePlan, e: EntityKind, now: Duration, cfg: &PrimitiveConfig) -> bool { + let fresh = |k: EntityKind| { + np.sourced + .get(&k) + .is_some_and(|t| now.saturating_sub(*t) < SOURCE_STALE_AFTER) + }; + let warm = now >= cfg.warmup; + match e { + EntityKind::Presence + | EntityKind::PersonCount + | EntityKind::MotionLevel + | EntityKind::PresenceScore => true, + EntityKind::MotionEnergy + | EntityKind::Rssi + | EntityKind::BreathingRate + | EntityKind::HeartRate + | EntityKind::FallDetected => fresh(e), + EntityKind::RoomActive | EntityKind::NoMovement | EntityKind::ElderlyInactivityAnomaly => { + warm + } + EntityKind::SomeoneSleeping => warm && fresh(EntityKind::BreathingRate), + EntityKind::PossibleDistress => warm && fresh(EntityKind::HeartRate), + EntityKind::FallRiskElevated => warm && fresh(EntityKind::FallDetected), + // No source on the sensing broadcast (see EntityKind::has_server_source). + EntityKind::ZoneOccupancy + | EntityKind::PoseKeypoints + | EntityKind::MeetingInProgress + | EntityKind::BathroomOccupied + | EntityKind::BedExit + | EntityKind::MultiRoomTransition => false, + } +} + +fn semantic_kind(e: EntityKind) -> Option { + Some(match e { + EntityKind::SomeoneSleeping => SemanticKind::SomeoneSleeping, + EntityKind::PossibleDistress => SemanticKind::PossibleDistress, + EntityKind::RoomActive => SemanticKind::RoomActive, + EntityKind::ElderlyInactivityAnomaly => SemanticKind::ElderlyAnomaly, + EntityKind::MeetingInProgress => SemanticKind::Meeting, + EntityKind::BathroomOccupied => SemanticKind::BathroomOccupied, + EntityKind::NoMovement => SemanticKind::NoMovement, + _ => return None, + }) +} + +fn raw_snapshot(snap: &VitalsSnapshot, now: Duration, fall: bool) -> RawSnapshot { + use chrono::Timelike; + RawSnapshot { + node_id: snap.node_id.clone(), + since_start: now, + timestamp_ms: snap.timestamp_ms, + presence: snap.presence, + fall_detected: fall, + motion: snap.motion, + motion_energy: snap.motion_energy.unwrap_or(0.0), + breathing_rate_bpm: snap.breathing_rate_bpm, + heart_rate_bpm: snap.heartrate_bpm, + n_persons: snap.n_persons, + rssi_dbm: snap.rssi_dbm, + vital_confidence: snap.vital_confidence, + // No zone source on the sensing broadcast; the zone-based primitives + // are not announced (EntityKind::has_server_source). + active_zones: Vec::new(), + bed_zones: Vec::new(), + local_seconds_since_midnight: chrono::Local::now().num_seconds_from_midnight(), + } +} + +#[cfg(test)] +mod tests; diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/tests.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/tests.rs new file mode 100644 index 000000000..e6fb18d18 --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/tests.rs @@ -0,0 +1,374 @@ +//! Unit tests for [`super::PublishPlanner`] and [`super::ChangeGate`]. + +use super::*; + +const NODE: &str = "base-node1"; +const WARM: Duration = Duration::from_secs(61); + +fn cfg(privacy_mode: bool) -> MqttConfig { + let args = { + use clap::Parser; + #[derive(Parser)] + struct W { + #[command(flatten)] + m: crate::cli::MqttArgs, + } + W::parse_from(["t"]).m + }; + let mut c = MqttConfig::from_args(&args); + c.privacy_mode = privacy_mode; + c +} + +fn builder() -> DiscoveryBuilder<'static> { + DiscoveryBuilder { + discovery_prefix: "homeassistant", + node_id: NODE, + node_friendly_name: None, + sw_version: "t", + model: "t", + via_device: None, + } +} + +/// The live ESP32 shape with no vitals, no fall detector, no motion energy. +fn bare(presence: bool) -> VitalsSnapshot { + VitalsSnapshot { + node_id: NODE.into(), + timestamp_ms: 1_779_512_400_000, + presence, + motion: if presence { 1.0 } else { 0.0 }, + presence_score: if presence { 0.8 } else { 0.0 }, + n_persons: u32::from(presence), + rssi_dbm: Some(-50.0), + ..Default::default() + } +} + +fn full(presence: bool) -> VitalsSnapshot { + VitalsSnapshot { + motion_energy: Some(4.2), + breathing_rate_bpm: Some(14.0), + heartrate_bpm: Some(62.0), + fall_detected: Some(false), + vital_confidence: 0.8, + ..bare(presence) + } +} + +fn slug(topic: &str) -> &str { + topic.rsplit('/').nth(1).unwrap() +} + +fn states(msgs: &[StateMessage]) -> Vec<(String, String)> { + msgs.iter() + .filter(|m| m.topic.ends_with("/state")) + .map(|m| (slug(&m.topic).to_string(), m.payload.clone())) + .collect() +} + +fn avail(msgs: &[StateMessage]) -> HashMap { + msgs.iter() + .filter(|m| m.topic.ends_with("/availability")) + .map(|m| (slug(&m.topic).to_string(), m.payload.clone())) + .collect() +} + +// ─── Entity → state mapping ───────────────────────────────────────── + +#[test] +fn announced_entities_are_exactly_the_sourced_ones() { + let p = PublishPlanner::new(&cfg(false)); + let slugs: Vec<_> = p.entities().iter().map(|e| e.topic_slug()).collect(); + assert_eq!( + slugs, + [ + "presence", + "person_count", + "breathing_rate", + "heart_rate", + "motion_level", + "motion_energy", + "fall", + "presence_score", + "rssi", + "someone_sleeping", + "possible_distress", + "room_active", + "elderly_inactivity_anomaly", + "fall_risk_elevated", + "no_movement", + ] + ); +} + +#[test] +fn every_announced_entity_publishes_state_when_its_source_is_present() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + let mut seen: HashMap = HashMap::new(); + let mut last_avail = HashMap::new(); + // Drive past warmup at 1 Hz; inject one fall edge. + for s in 0..=WARM.as_secs() { + let mut snap = full(true); + if s == WARM.as_secs() { + snap.fall_detected = Some(true); + } + let msgs = p.on_snapshot(&b, &snap, Duration::from_secs(s)); + last_avail.extend(avail(&msgs)); + seen.extend(states(&msgs)); + } + for e in p.entities() { + assert!( + seen.contains_key(e.topic_slug()), + "{e:?} never published state" + ); + assert_eq!(last_avail[e.topic_slug()], "online", "{e:?} not online"); + } + assert_eq!(seen["presence"], "ON"); + assert_eq!(seen["room_active"], "ON", "present + moving → room active"); + assert_eq!(seen["someone_sleeping"], "OFF"); + assert!(seen["heart_rate"].contains("\"bpm\":62.0")); + assert!(seen["motion_energy"].contains("\"energy\":4.2")); + assert!(seen["fall"].contains("fall_detected")); + assert!(seen["fall_risk_elevated"].contains("\"score\"")); +} + +#[test] +fn unsourced_entities_are_unavailable_and_never_publish_state() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + let mut seen = HashMap::new(); + let mut last_avail = HashMap::new(); + for s in 0..=WARM.as_secs() { + let msgs = p.on_snapshot(&b, &bare(true), Duration::from_secs(s)); + last_avail.extend(avail(&msgs)); + seen.extend(states(&msgs)); + } + for slug in [ + "breathing_rate", + "heart_rate", + "motion_energy", + "fall", + "someone_sleeping", + "possible_distress", + "fall_risk_elevated", + ] { + assert_eq!(last_avail[slug], "offline", "{slug} must be unavailable"); + assert!( + !seen.contains_key(slug), + "{slug} published a value with no source" + ); + } + for slug in [ + "presence", + "person_count", + "rssi", + "room_active", + "no_movement", + ] { + assert_eq!(last_avail[slug], "online", "{slug}"); + assert!(seen.contains_key(slug), "{slug}"); + } +} + +#[test] +fn semantic_entities_unavailable_during_warmup_then_online() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + let first = p.on_snapshot(&b, &bare(true), Duration::ZERO); + assert_eq!(avail(&first)["room_active"], "offline"); + assert!(!states(&first).iter().any(|(s, _)| s == "room_active")); + let warm = p.on_snapshot(&b, &bare(true), WARM); + assert_eq!( + avail(&warm)["room_active"], + "online", + "transition published" + ); + assert!(states(&warm).contains(&("room_active".into(), "ON".into()))); +} + +#[test] +fn vitals_go_unavailable_when_their_source_stops() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + assert_eq!( + avail(&p.on_snapshot(&b, &full(true), Duration::ZERO))["heart_rate"], + "online" + ); + let later = p.on_snapshot(&b, &bare(true), SOURCE_STALE_AFTER); + assert_eq!(avail(&later)["heart_rate"], "offline"); + assert!(!states(&later).iter().any(|(s, _)| s == "heart_rate")); +} + +#[test] +fn stale_node_heartbeat_reports_everything_offline() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + p.on_snapshot(&b, &bare(true), Duration::ZERO); + let hb = p.availability(&b, NODE, false, Duration::from_secs(40), true); + assert_eq!(hb.len(), p.entities().len()); + assert!(hb + .iter() + .all(|m| m.payload == "offline" && m.retain && m.qos == 1)); + assert!(p + .availability(&b, "unknown-node", true, Duration::ZERO, true) + .is_empty()); +} + +// ─── Rate limit / heartbeat (#2093) ───────────────────────────────── + +#[test] +fn presence_publishes_on_change_and_heartbeat_not_per_frame() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + let count = |p: &mut PublishPlanner, presence: bool, ms: u64| { + states(&p.on_snapshot(&b, &bare(presence), Duration::from_millis(ms))) + .into_iter() + .filter(|(s, _)| s == "presence") + .count() + }; + // ~73 frames/s for 10 s with no change → one publish. + let mut n = 0; + for i in 0..730 { + n += count(&mut p, true, i * 1000 / 73); + } + assert_eq!(n, 1, "unchanged presence must not publish per frame"); + assert_eq!( + count(&mut p, false, 10_100), + 1, + "change publishes immediately" + ); + assert_eq!(count(&mut p, false, 10_200), 0); + assert_eq!( + count(&mut p, false, 10_100 + 60_000), + 1, + "heartbeat after 60 s" + ); +} + +#[test] +fn change_gate_heartbeat_and_reset() { + let mut g = ChangeGate::default(); + let hb = Duration::from_secs(60); + let e = EntityKind::Presence; + assert!(g.allow("a", e, true, Duration::ZERO, hb)); + assert!(!g.allow("a", e, true, Duration::from_secs(59), hb)); + assert!( + g.allow("b", e, true, Duration::from_secs(59), hb), + "per node" + ); + assert!( + g.allow("a", e, false, Duration::from_secs(59), hb), + "change" + ); + assert!( + g.allow("a", e, false, Duration::from_secs(119), hb), + "heartbeat" + ); + g.reset(); + assert!(g.allow("a", e, false, Duration::from_secs(120), hb)); +} + +#[test] +fn numeric_sensors_keep_their_configured_rate() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + let mut rssi = 0; + for i in 0..730u64 { + rssi += states(&p.on_snapshot(&b, &bare(true), Duration::from_millis(i * 1000 / 73))) + .iter() + .filter(|(s, _)| s == "rssi") + .count(); + } + assert_eq!(rssi, 1, "rssi at 0.1 Hz → one sample in 10 s"); +} + +#[test] +fn reset_republishes_state_and_availability() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + p.on_snapshot(&b, &bare(true), Duration::ZERO); + let quiet = p.on_snapshot(&b, &bare(true), Duration::from_millis(10)); + assert!(avail(&quiet).is_empty() && !states(&quiet).iter().any(|(s, _)| s == "presence")); + p.reset(); + let again = p.on_snapshot(&b, &bare(true), Duration::from_millis(20)); + assert_eq!(avail(&again)["presence"], "online"); + assert!(states(&again).iter().any(|(s, _)| s == "presence")); +} + +// ─── Privacy mode ─────────────────────────────────────────────────── + +#[test] +fn privacy_mode_never_announces_or_publishes_biometrics() { + let mut p = PublishPlanner::new(&cfg(true)); + assert!(p.entities().iter().all(|e| !e.is_biometric())); + let b = builder(); + for s in 0..=WARM.as_secs() { + let msgs = p.on_snapshot(&b, &full(true), Duration::from_secs(s)); + for m in &msgs { + for slug in ["heart_rate", "breathing_rate", "pose"] { + assert!( + !m.topic.contains(&format!("/{slug}/")), + "{slug} leaked: {}", + m.topic + ); + } + } + } + // Inferred states stay (ADR-115 §3.12.3): their inputs are used + // server-side, only the state crosses the wire. + let msgs = p.on_snapshot(&b, &full(true), WARM + Duration::from_secs(1)); + let all: Vec<_> = p.availability(&b, NODE, true, WARM, true); + assert_eq!(avail(&all)["someone_sleeping"], "online"); + assert!(msgs.iter().all(|m| !m.payload.contains("bpm"))); +} + +// ─── Semantic wiring ──────────────────────────────────────────────── + +#[test] +fn no_movement_turns_on_after_its_dwell() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + let still = VitalsSnapshot { + motion: 0.0, + ..bare(true) + }; + let mut first_off = None; + let mut first_on = None; + for s in 0..WARM.as_secs() + 30 * 60 + 2 { + for (slug, v) in states(&p.on_snapshot(&b, &still, Duration::from_secs(s))) { + match (slug.as_str(), v.as_str()) { + ("no_movement", "OFF") => _ = first_off.get_or_insert(s), + ("no_movement", "ON") => _ = first_on.get_or_insert(s), + _ => {} + } + } + } + // OFF once warmup (60 s) ends, ON after the 30 min stillness dwell. + assert_eq!(first_off, Some(60)); + assert_eq!(first_on, Some(60 + 30 * 60)); +} + +#[test] +fn semantic_bus_samples_at_one_hz_but_keeps_fall_edges() { + let mut p = PublishPlanner::new(&cfg(false)); + let b = builder(); + // Warm up with the fall detector online. + for s in 0..=WARM.as_secs() { + p.on_snapshot(&b, &full(true), Duration::from_secs(s)); + } + let base = p.nodes[NODE].bus.fall_risk_score(); + // A fall edge between two 1 Hz ticks must still reach the FSM. + let t = WARM + Duration::from_millis(500); + p.on_snapshot( + &b, + &VitalsSnapshot { + fall_detected: Some(true), + ..full(true) + }, + t, + ); + p.on_snapshot(&b, &full(true), WARM + SEMANTIC_TICK); + assert!(p.nodes[NODE].bus.fall_risk_score() >= base + 10.0); +} diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs index 96b469449..a052360c9 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs @@ -11,21 +11,22 @@ //! install LWT on every entity's availability topic, set keepalive. //! 2. **Discovery**: emit one retained discovery `config` topic per //! enabled entity per known node. Re-emit every `refresh_secs`. -//! 3. **Availability heartbeat**: publish `online` retained on every -//! availability topic on connect, and re-publish every 30 s so HA can +//! 3. **Availability**: per entity, `online` only while the server has a +//! source for it (see [`PublishPlanner`]); transitions publish at once +//! and every availability topic is re-published every 30 s so HA can //! detect zombie sessions. -//! 4. **State publication**: subscribe to the broadcast channel; for -//! each inbound message project it into a [`VitalsSnapshot`], pass -//! through the privacy filter, gate by [`RateLimiter`], encode via -//! [`StateEncoder`], publish. +//! 4. **State publication**: for each [`VitalsSnapshot`] from the bridge, +//! [`PublishPlanner`] runs the semantic primitives, applies the privacy +//! filter, the on-change gate (binary states) and the rate limiter +//! (sensors), and returns the messages to publish. //! //! ## Reconnect strategy //! //! `rumqttc::EventLoop` reconnects automatically with backoff. After a //! successful reconnect we re-publish discovery (retained config topics //! survive at the broker, but a fresh HA install that came online after -//! we last refreshed needs them) and reset the rate limiter so the -//! first post-reconnect sample emits promptly. +//! we last refreshed needs them) and reset the planner so every state and +//! availability is re-sent promptly. use std::sync::Arc; use std::time::{Duration, Instant}; @@ -60,7 +61,8 @@ macro_rules! otel_warn { use super::config::{MqttConfig, TlsConfig}; use super::discovery::{DiscoveryBuilder, EntityKind}; -use super::state::{RateLimiter, StateEncoder, StateMessage, VitalsSnapshot}; +use super::planner::PublishPlanner; +use super::state::{StateMessage, VitalsSnapshot}; /// Heartbeat cadence for availability re-publication (per §3.6). const AVAILABILITY_HEARTBEAT: Duration = Duration::from_secs(30); @@ -146,22 +148,6 @@ fn build_transport(tls: &TlsConfig) -> Transport { Transport::tls_with_default_config() } -/// One node's per-entity availability topics, pre-computed at startup so -/// the heartbeat loop doesn't allocate per tick. -struct NodeAvailability { - online_topics: Vec, -} - -impl NodeAvailability { - fn for_builder(b: &DiscoveryBuilder<'_>, entities: &[EntityKind]) -> Self { - let online_topics = entities - .iter() - .map(|e| b.availability_topic(*e)) - .collect(); - Self { online_topics } - } -} - /// Spawn the MQTT publisher background task. Returns the join handle so /// the caller can `await` it on shutdown. Errors during connection are /// retried internally by `rumqttc::EventLoop`. @@ -229,27 +215,21 @@ async fn run( let (client, mut eventloop): (AsyncClient, EventLoop) = AsyncClient::builder(opts).capacity(256).build(); - let entities = DiscoveryBuilder::enabled_entities( - cfg.privacy_mode, - cfg.publish_pose, - &[], // no_semantic — wire from cli::Args in P3.5 - ); + let mut planner = PublishPlanner::new(&cfg); + if cfg.publish_pose { + otel_warn!( + "[mqtt] --mqtt-publish-pose ignored: the sensing broadcast carries no measured pose \ + keypoints, so no pose entity is announced (issue #2085)" + ); + } - // #898: one Home-Assistant device per node. Discovery + availability are - // published lazily the first time a snapshot for a given node_id arrives; - // each node's builder + availability are retained here for heartbeats and - // the offline LWT. (Previously a single hard-coded builder collapsed every - // node into one device.) - // Issue #1555: the third tuple element is the Instant this node's last - // broadcast snapshot arrived, so the heartbeat below can tell a node that - // has genuinely gone quiet from one that's just between publish-rate - // ticks, and stop asserting "online" for it. - let mut nodes: std::collections::HashMap< - String, - (OwnedDiscoveryBuilder, NodeAvailability, Instant), - > = std::collections::HashMap::new(); + // #898: one Home-Assistant device per node. Discovery is published lazily + // the first time a snapshot for a given node_id arrives. The Instant is + // when that node's last snapshot arrived (issue #1555), so the heartbeat + // can tell a node that has gone quiet from one between publish ticks. + let mut nodes: std::collections::HashMap = + std::collections::HashMap::new(); - let mut rate_limiter = RateLimiter::new(); let mut last_heartbeat = Instant::now(); let mut last_refresh = Instant::now(); let start_instant = Instant::now(); @@ -258,7 +238,7 @@ async fn run( host = %cfg.host, port = cfg.port, prefix = %cfg.discovery_prefix, - entities = entities.len(), + entities = planner.entities().len(), privacy = cfg.privacy_mode, "[mqtt] publisher started", ); @@ -274,7 +254,7 @@ async fn run( Ok(_) => {} Err(e) => { otel_error!("[mqtt] event loop error, will reconnect: {e}"); - rate_limiter.reset(); + planner.reset(); // Brief backoff before next poll attempt. tokio::time::sleep(Duration::from_millis(500)).await; } @@ -284,29 +264,24 @@ async fn run( // Periodic heartbeat / discovery refresh. _ = tokio::time::sleep(Duration::from_secs(1)) => { if last_heartbeat.elapsed() >= AVAILABILITY_HEARTBEAT { - for (node_id, (_, na, last_seen)) in &nodes { - // Issue #1555: a node whose snapshots have actually - // stopped arriving must go `offline`, not keep - // reporting `online` on a fixed timer regardless of - // whether its data is still flowing — a frozen HA - // entity that still shows "available" is worse than - // one correctly marked unavailable. - let state = if last_seen.elapsed() < NODE_SNAPSHOT_STALE_AFTER { - "online" - } else { - "offline" - }; - if let Err(e) = publish_availability(&client, na, state).await { + let now = start_instant.elapsed(); + for (node_id, (nb, last_seen)) in &nodes { + // Issue #1555: a node whose snapshots have stopped + // arriving goes `offline` rather than reporting + // `online` on a fixed timer. + let fresh = last_seen.elapsed() < NODE_SNAPSHOT_STALE_AFTER; + let b = nb.as_borrowed(); + let msgs = planner.availability(&b, node_id, fresh, now, true); + if let Err(e) = publish_messages(&client, &msgs).await { otel_warn!("[mqtt] heartbeat publish failed for node {node_id}: {e}"); } } last_heartbeat = Instant::now(); } if last_refresh.elapsed() >= Duration::from_secs(cfg.refresh_secs) { - for (nb, _, _) in nodes.values() { - if let Err(e) = - publish_all_discovery(&client, &nb.as_borrowed(), &entities).await - { + for (nb, _) in nodes.values() { + let b = nb.as_borrowed(); + if let Err(e) = publish_all_discovery(&client, &b, planner.entities()).await { otel_warn!("[mqtt] discovery refresh failed: {e}"); } } @@ -318,32 +293,25 @@ async fn run( recv = state_rx.recv() => { match recv { Ok(snap) => { - let elapsed = start_instant.elapsed(); let now = Instant::now(); // #898: on first sight of a node_id, publish that - // node's discovery + availability; then route its - // state to per-node topics. - if !nodes.contains_key(&snap.node_id) { + // node's discovery; then route its state to per-node + // topics. + if let Some(entry) = nodes.get_mut(&snap.node_id) { + entry.1 = now; + } else { let nb = builder_owned.for_node(&snap.node_id); - let borrowed = nb.as_borrowed(); - if let Err(e) = - publish_all_discovery(&client, &borrowed, &entities).await - { + let b = nb.as_borrowed(); + if let Err(e) = publish_all_discovery(&client, &b, planner.entities()).await { otel_warn!("[mqtt] node {} discovery failed: {e}", snap.node_id); } - let na = NodeAvailability::for_builder(&borrowed, &entities); - if let Err(e) = publish_availability(&client, &na, "online").await { - otel_warn!("[mqtt] node {} availability failed: {e}", snap.node_id); - } - nodes.insert(snap.node_id.clone(), (nb, na, now)); - } else if let Some(entry) = nodes.get_mut(&snap.node_id) { - // Issue #1555: record that this node is still - // alive so the heartbeat above doesn't have to - // guess from a fixed timer. - entry.2 = now; + nodes.insert(snap.node_id.clone(), (nb, now)); } let borrowed = nodes[&snap.node_id].0.as_borrowed(); - publish_snapshot(&client, &borrowed, &snap, &cfg, &mut rate_limiter, elapsed).await; + let msgs = planner.on_snapshot(&borrowed, &snap, start_instant.elapsed()); + if let Err(e) = publish_messages(&client, &msgs).await { + otel_warn!("[mqtt] node {} state publish failed: {e}", snap.node_id); + } } Err(broadcast::error::RecvError::Lagged(n)) => { warn!("[mqtt] lagged behind broadcast by {n} messages — dropped"); @@ -351,8 +319,9 @@ async fn run( Err(broadcast::error::RecvError::Closed) => { info!("[mqtt] broadcast channel closed, draining"); // Publish offline for every known node before exit. - for (_, na, _) in nodes.values() { - let _ = publish_availability(&client, na, "offline").await; + for (nb, _) in nodes.values() { + let msgs = planner.offline(&nb.as_borrowed()); + let _ = publish_messages(&client, &msgs).await; } let _ = client.disconnect().await; return; @@ -384,73 +353,13 @@ async fn publish_all_discovery( Ok(()) } -async fn publish_availability( - client: &AsyncClient, - avail: &NodeAvailability, - state: &str, -) -> Result<(), ClientError> { - for topic in &avail.online_topics { - client - .publish( - topic, - state, - PublishOptions::new(QoS::AtLeastOnce).retained(), - ) - .await?; +async fn publish_messages(client: &AsyncClient, msgs: &[StateMessage]) -> Result<(), ClientError> { + for m in msgs { + publish_state(client, m).await?; } Ok(()) } -async fn publish_snapshot( - client: &AsyncClient, - b: &DiscoveryBuilder<'_>, - snap: &VitalsSnapshot, - cfg: &MqttConfig, - rl: &mut RateLimiter, - elapsed: Duration, -) { - let encoder = StateEncoder { builder: b }; - - // Binary: presence (change-only — caller is responsible for detecting - // change, but we always publish here because broadcast already debounces - // and HA will dedup retained equal values harmlessly). - if let Some(m) = encoder.boolean(EntityKind::Presence, snap.presence) { - let _ = publish_state(client, &m).await; - } - - // Event: fall. - if snap.fall_detected { - if let Some(m) = encoder.event( - EntityKind::FallDetected, - "fall_detected", - snap.timestamp_ms, - Some(snap.vital_confidence), - ) { - let _ = publish_state(client, &m).await; - } - } - - // Numeric rate-limited entities. Rate limiting is per (node, entity) - // (ADR-297, issue #1541) so nodes never starve one another. - let node = snap.node_id.as_str(); - for (entity, allowed) in [ - (EntityKind::PersonCount, rl.allow(node, EntityKind::PersonCount, elapsed, &cfg.rates)), - (EntityKind::HeartRate, !cfg.privacy_mode && rl.allow(node, EntityKind::HeartRate, elapsed, &cfg.rates)), - (EntityKind::BreathingRate, !cfg.privacy_mode && rl.allow(node, EntityKind::BreathingRate, elapsed, &cfg.rates)), - (EntityKind::MotionLevel, rl.allow(node, EntityKind::MotionLevel, elapsed, &cfg.rates)), - (EntityKind::MotionEnergy, rl.allow(node, EntityKind::MotionEnergy, elapsed, &cfg.rates)), - (EntityKind::PresenceScore, rl.allow(node, EntityKind::PresenceScore, elapsed, &cfg.rates)), - (EntityKind::Rssi, rl.allow(node, EntityKind::Rssi, elapsed, &cfg.rates)), - ] { - if !allowed { - continue; - } - if let Some(m) = encoder.numeric(entity, snap) { - let _ = publish_state(client, &m).await; - } - } -} - async fn publish_state(client: &AsyncClient, m: &StateMessage) -> Result<(), ClientError> { let qos = match m.qos { 0 => QoS::AtMostOnce, diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/state.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/state.rs index 86d2ca438..b472a60b5 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mqtt/state.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mqtt/state.rs @@ -155,9 +155,12 @@ pub struct VitalsSnapshot { pub node_id: String, pub timestamp_ms: i64, pub presence: bool, - pub fall_detected: bool, + /// `Some(true)` once per detected fall, `Some(false)` while the node's + /// fall detector reports, `None` when the node has no fall source. + pub fall_detected: Option, pub motion: f64, // 0.0–1.0 - pub motion_energy: f64, + /// `None` when the frame carries no motion band power. + pub motion_energy: Option, pub presence_score: f64, // 0.0–1.0 pub breathing_rate_bpm: Option, pub heartrate_bpm: Option, @@ -203,6 +206,12 @@ struct RssiPayload { ts: String, } +#[derive(Serialize, Debug)] +struct ScorePayload { + score: f64, + ts: String, +} + #[derive(Serialize, Debug)] struct FallEventPayload { event_type: &'static str, @@ -269,7 +278,7 @@ impl<'a> StateEncoder<'a> { ts: ts.clone(), }).ok()?, EntityKind::MotionEnergy => serde_json::to_value(EnergyStatePayload { - energy: snap.motion_energy, + energy: snap.motion_energy?, ts: ts.clone(), }).ok()?, EntityKind::PresenceScore => serde_json::to_value(PresenceScorePayload { @@ -293,6 +302,32 @@ impl<'a> StateEncoder<'a> { Some(StateMessage::new(topic, payload, DiscoveryComponent::Sensor, false)) } + /// 0–100 score sensor (fall risk), matching the `{{ value_json.score }}` + /// discovery template. + pub fn score(&self, entity: EntityKind, value: f64, ts_ms: i64) -> Option { + if !matches!(entity.component(), DiscoveryComponent::Sensor) || !value.is_finite() { + return None; + } + let payload = serde_json::to_string(&ScorePayload { + score: value.clamp(0.0, 100.0), + ts: iso_ts(ts_ms), + }) + .ok()?; + let topic = format!( + "{}/{}/wifi_densepose_{}/{}/state", + self.builder.discovery_prefix, + entity.component().as_str(), + self.builder.node_id, + entity.topic_slug(), + ); + Some(StateMessage::new( + topic, + payload, + DiscoveryComponent::Sensor, + false, + )) + } + /// One-shot event encoder. Used for fall, bed exit, multi-room /// transition. pub fn event(&self, entity: EntityKind, event_type: &'static str, ts_ms: i64, confidence: Option) -> Option { @@ -354,9 +389,9 @@ mod tests { node_id: "aabbccddeeff".into(), timestamp_ms: 1779_512_400_000, presence: true, - fall_detected: false, + fall_detected: Some(false), motion: 0.35, - motion_energy: 1234.5, + motion_energy: Some(1234.5), presence_score: 0.91, breathing_rate_bpm: Some(14.2), heartrate_bpm: Some(68.2), @@ -509,6 +544,32 @@ mod tests { assert!(enc.numeric(EntityKind::HeartRate, &s).is_none()); } + #[test] + fn numeric_encoder_has_no_motion_energy_without_a_source() { + let b = builder(); + let enc = StateEncoder { builder: &b }; + let mut s = snap(); + s.motion_energy = None; + assert!( + enc.numeric(EntityKind::MotionEnergy, &s).is_none(), + "never a default 0" + ); + } + + #[test] + fn score_encoder_matches_discovery_template() { + let b = builder(); + let enc = StateEncoder { builder: &b }; + let msg = enc.score(EntityKind::FallRiskElevated, 142.0, 0).unwrap(); + let json: serde_json::Value = serde_json::from_str(&msg.payload).unwrap(); + assert_eq!(json["score"], 100.0, "clamped to 0..100"); + assert!(msg.topic.ends_with("/fall_risk_elevated/state")); + assert!(enc + .score(EntityKind::FallRiskElevated, f64::NAN, 0) + .is_none()); + assert!(enc.score(EntityKind::Presence, 1.0, 0).is_none()); + } + #[test] fn numeric_encoder_clamps_out_of_range_motion() { let b = builder(); diff --git a/v2/crates/wifi-densepose-sensing-server/src/semantic/bus.rs b/v2/crates/wifi-densepose-sensing-server/src/semantic/bus.rs index b5b168cdd..7502e4499 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/semantic/bus.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/semantic/bus.rs @@ -81,6 +81,29 @@ impl SemanticBus { } } + /// Current state of a boolean primitive, so the publisher can report it + /// on a heartbeat and not only on the tick it changed. `None` for the + /// scalar and event primitives. + pub fn is_active(&self, kind: SemanticKind) -> Option { + Some(match kind { + SemanticKind::SomeoneSleeping => self.sleeping.active, + SemanticKind::PossibleDistress => self.distress.active, + SemanticKind::RoomActive => self.room_active.active, + SemanticKind::ElderlyAnomaly => self.elderly_anomaly.active, + SemanticKind::Meeting => self.meeting.active, + SemanticKind::BathroomOccupied => self.bathroom.active, + SemanticKind::NoMovement => self.no_movement.active, + SemanticKind::FallRisk | SemanticKind::BedExit | SemanticKind::MultiRoom => { + return None + } + }) + } + + /// Latest fall-risk score (0–100). Zero until the first post-warmup tick. + pub fn fall_risk_score(&self) -> f64 { + self.fall_risk.last_score + } + /// Run all primitives on one snapshot. Returns only events that /// emit (Idle states are filtered). pub fn tick(&mut self, snap: &RawSnapshot) -> Vec { @@ -220,6 +243,22 @@ mod tests { } } + #[test] + fn is_active_reflects_boolean_primitive_state() { + let mut bus = SemanticBus::new(cfg()); + assert_eq!(bus.is_active(SemanticKind::RoomActive), Some(false)); + assert_eq!(bus.is_active(SemanticKind::FallRisk), None); + let snap = RawSnapshot { + node_id: "test".into(), + since_start: Duration::from_secs(120), + presence: true, + motion: 0.4, + ..Default::default() + }; + bus.tick(&snap); + assert_eq!(bus.is_active(SemanticKind::RoomActive), Some(true)); + } + #[test] fn _unused_reason_helper_remains_constructible() { // Touch Reason::empty to keep clippy happy when the bus uses diff --git a/v2/crates/wifi-densepose-sensing-server/tests/mqtt_integration.rs b/v2/crates/wifi-densepose-sensing-server/tests/mqtt_integration.rs index c8dd1c8c1..fb40ad7d7 100644 --- a/v2/crates/wifi-densepose-sensing-server/tests/mqtt_integration.rs +++ b/v2/crates/wifi-densepose-sensing-server/tests/mqtt_integration.rs @@ -323,9 +323,9 @@ async fn state_messages_published_on_snapshot_broadcast() { node_id: "inttest3".into(), timestamp_ms: 1779_512_400_000 + (i as i64) * 300, presence: i % 2 == 0, - fall_detected: false, + fall_detected: Some(false), motion: if i % 2 == 0 { 0.40 } else { 0.02 }, - motion_energy: 800.0, + motion_energy: Some(800.0), presence_score: if i % 2 == 0 { 0.95 } else { 0.10 }, breathing_rate_bpm: Some(14.0), heartrate_bpm: Some(72.0),