From e351d0fce19777d64d388eeaaaa9cd74f141c105 Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 18:46:11 -0400 Subject: [PATCH 1/2] feat(mqtt): wire semantic states into the publisher, announce only sourced entities Discovery announced 20 entities per device, but only 6 ever received state (#2085). presence/state was published on every CSI frame, and the per-process default client id gave Home Assistant new devices on every restart (#2093). - mqtt::planner maps each snapshot to availability, state and events. It runs one SemanticBus per node at 1 Hz (ADR-115 P4.5) and gates binary states on change with a 60 s retained heartbeat. Sensors keep their configured rates. Each entity is `online` only while the server has a source for it; otherwise it is `offline` and gets no state. - Entities with no source on the sensing broadcast are no longer announced: zone_occupancy, bathroom_occupied, bed_exit, multi_room_transition (no zones), meeting_in_progress (motion is three-level) and pose (pose_keypoints is never set). 15 remain, 13 with --privacy-mode. - mqtt::bridge, moved from main.rs, reads only sensing_update. That removes the phantom aggregate device that edge_vitals frames created. It takes fall events from the edge_vitals rising edge and motion_energy from motion_band_power (previously always 0.0), and maps present_still to 0 % motion. - The default client id is wifi-densepose-<8 hex>, persisted in /mqtt_client_id. --mqtt-client-id still overrides it. Refs #2085, #2093 Co-Authored-By: claude-flow --- .../benches/mqtt_throughput.rs | 4 +- .../examples/mqtt_publisher.rs | 4 +- .../wifi-densepose-sensing-server/src/cli.rs | 3 +- .../wifi-densepose-sensing-server/src/main.rs | 227 +--------- .../src/mqtt/bridge.rs | 417 ++++++++++++++++++ .../src/mqtt/config.rs | 124 +++++- .../src/mqtt/discovery.rs | 47 +- .../src/mqtt/mod.rs | 17 +- .../src/mqtt/planner/mod.rs | 321 ++++++++++++++ .../src/mqtt/planner/tests.rs | 374 ++++++++++++++++ .../src/mqtt/publisher.rs | 203 +++------ .../src/mqtt/state.rs | 71 ++- .../src/semantic/bus.rs | 39 ++ .../tests/mqtt_integration.rs | 4 +- 14 files changed, 1461 insertions(+), 394 deletions(-) create mode 100644 v2/crates/wifi-densepose-sensing-server/src/mqtt/bridge.rs create mode 100644 v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/mod.rs create mode 100644 v2/crates/wifi-densepose-sensing-server/src/mqtt/planner/tests.rs 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), From 8c63d7b558eec8cfa5a57556aa57cc90b1c83ba8 Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 18:46:21 -0400 Subject: [PATCH 2/2] docs(mqtt): document the published entity set and add an ADR-115 P4.5 status note MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit home-assistant.md now describes current behaviour: the 15 announced entities (13 with --privacy-mode) with their source and availability rule, the entities that are not announced and why, publish rates and retention, the persisted client id, and what privacy mode does and does not cover. The guide no longer documents --semantic* flags that sensing-server does not accept, and it drops the HIPAA/GDPR/CCPA claims. ADR-115 §11 records what P4.5 wired and what it did not, with a before/after measurement against a real broker (MEASURED, one live node). Refs #2085, #2093 Co-Authored-By: claude-flow --- .../adr/ADR-115-home-assistant-integration.md | 57 ++++++- docs/integrations/home-assistant.md | 161 +++++++++--------- 2 files changed, 140 insertions(+), 78 deletions(-) diff --git a/docs/adr/ADR-115-home-assistant-integration.md b/docs/adr/ADR-115-home-assistant-integration.md index b7a886add..160751678 100644 --- a/docs/adr/ADR-115-home-assistant-integration.md +++ b/docs/adr/ADR-115-home-assistant-integration.md @@ -3,7 +3,7 @@ | Field | Value | |-------|-------| | **Status** | **Accepted** (MQTT track P1–P7 + P8a + P9 + P10 shipped 2026-05-23 in PR #778, 410 lib tests, witness bundle VERIFIED) / **Proposed** (Matter SDK wiring P8b deferred to v0.7.1 per §9.10) | -| **Date** | 2026-05-23 | +| **Date** | 2026-05-23 (amended 2026-10-02: P4.5 status note, §11) | | **Deciders** | ruv | | **Codename** | **HA-DISCO** (MQTT) + **HA-FABRIC** (Matter) + **HA-MIND** (semantic primitives) | | **Relates to** | ADR-018 (CSI binary frame format), ADR-021 (ESP32 vitals), ADR-031 (RuView sensing-first), ADR-039 (edge vitals packet 0xC511_0002), ADR-079 (camera ground-truth), ADR-103 (cog-person-count), ADR-110 (ESP32-C6 firmware), ADR-114 (cog-quantum-vitals) | @@ -440,7 +440,7 @@ This means: **adding a new primitive is one file change**. No MQTT schema rev, n | **P2** | HA discovery message emitter. New module `mqtt_discovery.rs`. Emits all entity `config` topics on connect + every `--mqtt-refresh-secs`. Schema-validated against HA's published JSON schema. | pending | | **P3** | State publication. Subscribe to internal `tokio::broadcast` channel (the one `tx.send(json)` writes to on line 3983 of `main.rs`). Translate `edge_vitals` / `sensing_update` / `pose_data` messages into per-entity state payloads. Apply rate-limit + privacy-mode filters. | pending | | **P4** | Integration tests: dockerised mosquitto in CI (extend `.github/workflows/firmware-qemu.yml` pattern), schema-validate every emitted config against HA's `homeassistant/components/mqtt` JSON schemas (pin to a tested HA version). Add a smoke test that brings up sensing-server in `--source mock --mqtt`, subscribes with `paho-mqtt` test client, asserts on entity creation. | pending | -| **P4.5** | **Semantic inference layer (HA-MIND).** New module `semantic_inference.rs` implementing the 10 v1 primitives from §3.12. Output broadcast channel consumed by both MQTT publisher (P3) and Matter bridge (P8). Per-primitive precision/recall baselines published to `docs/integrations/semantic-primitives-metrics.md`. Unit tests per FSM + integration tests via replay of ADR-079 paired captures. | pending | +| **P4.5** | **Semantic inference layer (HA-MIND).** New module `semantic_inference.rs` implementing the 10 v1 primitives from §3.12. Output broadcast channel consumed by both MQTT publisher (P3) and Matter bridge (P8). Per-primitive precision/recall baselines published to `docs/integrations/semantic-primitives-metrics.md`. Unit tests per FSM + integration tests via replay of ADR-079 paired captures. | partial — 6 of 10 primitives on the live MQTT publisher; see §11 | | **P5** | Docs: new `docs/integrations/home-assistant.md` with screenshots of the HA UI after auto-discovery completes, example HA dashboard YAML (Lovelace card configs), 8 starter blueprints from §3.12.2 (distress notify, wake routine, hallway dim, elderly anomaly alert, meeting lights, bathroom fan, fall-risk escalate, auto-arm security), and the raw-channel example automations: "turn on hall light when presence ON", "send notification on fall_detected event", "log HR/BR to InfluxDB". | pending | | **P6** | Ship `--mqtt` in the next sensing-server release (target: v0.7.0). Demo end-to-end on `cognitum-v0` against a Mosquitto add-on running on a Home Assistant OS install. Update README hardware-options table with "Works with Home Assistant" badge. | pending | | **P7** | Matter Bridge spike: build a throwaway prototype with `matter-rs` exposing one `OccupancySensor` endpoint + one `GenericSwitch` for fall. Pair against Apple Home, Google Home, and HA's Matter integration. Decision gate: if pairing works on all three, proceed to P8; if blocked, switch to chip-tool FFI and re-spike. | pending | @@ -667,4 +667,57 @@ Empty as of 2026-05-23. New questions discovered during implementation will be f --- +## 11. Status note: P4.5 semantic states on the live publisher (2026-10-02) + +Issues [#2085](https://github.com/ruvnet/RuView/issues/2085) and [#2093](https://github.com/ruvnet/RuView/issues/2093) found that discovery announced 20 entities per device while only 6 ever received state, that `presence/state` was published on every CSI frame, and that HA device ids changed on every restart. The FSMs in `crate::semantic` existed but had no caller outside their tests. This note records what is now wired and what is not. + +### 11.1 What changed + +- **Semantic primitives run on the live path.** `mqtt::planner` holds one `SemanticBus` per node, sampled at 1 Hz (the rate the FSM constants assume). The bus feeds only the MQTT publisher. Matter does not consume it yet, so the "output broadcast channel shared with Matter" in P4.5 is not built. +- **Announce only what has a source.** Six primitives are announced: `room_active`, `no_movement`, `elderly_inactivity_anomaly`, `someone_sleeping`, `possible_distress` and `fall_risk_elevated`. Four are not: `meeting_in_progress` needs a 1–20 % motion level, but per-node motion is a three-level classification; `bathroom_occupied`, `bed_exit` and `multi_room_transition` need zones, and the broadcast carries none (the §3.12.5 zone file is not implemented). For the same reason `zone_occupancy` (§3.1) and `pose` are not announced: `pose_keypoints` is never populated on the broadcast, and `persons[].keypoints` are synthesized. +- **Per-entity availability (§3.6).** An announced entity is `online` only while its input exists: vitals while the calibration-gated estimate is published, `fall` while the node sends `edge_vitals`, `motion_energy` and `rssi` while present, and semantic primitives after the 60 s warmup (sleeping and distress also need breathing or heart rate, fall risk needs the fall detector). Otherwise it is `offline` and no state is sent. Transitions publish at once; all availability topics are re-sent every 30 s. +- **No default values.** `motion_energy` was always `0.0` (MEASURED: 176 of 176 samples) because the bridge never set it. It now carries the node's `motion_band_power`, or the room value, or nothing. `fall` is driven by the rising edge of `edge_vitals.fall_detected`. Only `sensing_update` produces snapshots, which removes the phantom aggregate device that `edge_vitals` frames created. `present_still` now maps to 0 % motion instead of 100 %. +- **Rates (§3.7).** Binary states publish on change, plus a 60 s retained heartbeat, as §3.5 already specified. Sensors keep their configured rates. +- **Stable device identity (§3.8).** The default client id is now `wifi-densepose-<8 hex>`, persisted in `/mqtt_client_id`. §3.8 specified `wifi-densepose-`; a persisted random id avoids a hostname dependency and survives hostname changes. `--mqtt-client-id` still overrides it. + +### 11.2 Evidence + +Unit tests (CODE-DERIVED, run with `cargo test -p wifi-densepose-sensing-server --features mqtt --lib -- mqtt:: semantic::`): `mqtt::planner::tests` covers the entity → state mapping, unavailable-without-source, warmup, on-change with heartbeat at a 73 frame/s input, sensor rates, reset and privacy mode. `mqtt::bridge::tests` covers message-type filtering, fall edges and motion energy. `mqtt::config::tests` covers client-id persistence across two "restarts" in a temp dir, override and malformed-file recovery. + +Live measurement (MEASURED, 2026-10-02): one ESP32 node streaming CSI and `edge_vitals` over UDP to a macOS host; `sensing-server --source esp32 --mqtt` against `eclipse-mosquitto:2` in Docker on a non-default port; every topic captured with a `paho-mqtt` subscriber on `#`. "Before" is `03a89793` (main), "after" is this change, built `--release --features mqtt`, same command and window. + +```bash +# Broker: mosquitto.conf = "listener 1883 0.0.0.0 / allow_anonymous true / persistence false" +docker run -d --rm -p 127.0.0.1:18830:1883 \ + -v "$PWD/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro" eclipse-mosquitto:2 +# Run A (180 s, then SIGTERM) and run B (restart, 45 s) share one --data-dir. +# A paho-mqtt subscriber on '#' starts 2 s before each run. +RUVIEW_API_TOKEN= sensing-server --source esp32 --udp-port 5006 \ + --udp-bind 0.0.0.0 --udp-allow --http-port 3000 --ws-port 3001 \ + --no-mdns --data-dir --mqtt --mqtt-host 127.0.0.1 --mqtt-port 18830 +``` + +| Measure (180 s window unless noted) | Before (`03a89793`) | After | +|---|---|---| +| HA devices announced in run A | 2 (one phantom aggregate + `-node1`) | 1 (`-node1`) | +| Entities announced per device | 20 | 15 | +| Entities that received state (node device) | 6 | 10 | +| Announced entities without state that still reported `online` | 14 | 1 (`fall`: an event, and no fall occurred) | +| `presence/state` messages (node device) | 6,814 (37.9 / s) | 4 (one per change, plus 60 s heartbeat) | +| `motion_energy` values | `0.0` in 176 of 176 | node band power, 177 samples | +| Device id, run A → run B (restart, same `--data-dir`) | `wifi-densepose-` → `` (4 devices on the broker after B) | same persisted id in both runs (1 device) | +| `offline` published on SIGTERM | no | no (unchanged; see 11.3) | +| Same, plus `--privacy-mode` (75 s, fresh `--data-dir`) | — | 13 entities announced; no `heart_rate`, `breathing_rate` or `pose` topic for that device | + +After the change, five announced entities had no state: `heart_rate` and `breathing_rate`, because no calibration had been run; `someone_sleeping` and `possible_distress`, which depend on them; and `fall`, which is an event and no fall occurred. The first four were `offline`. `fall` was `online`, because the node sent `edge_vitals`. Before the change all 14 entities without state still reported `online`. Only one node was streaming during this measurement. #2085 used five nodes and measured about 73 `presence` messages per second per node. + +### 11.3 Not done + +- Semantic precision/recall on real captures. The figures in `semantic-primitives-metrics.md` have no reproducer in the repository and stay CLAIMED. +- Zone and threshold files, `--semantic*` / `--no-semantic` flags on `sensing-server`, the `reason` attribute (§3.12.4), and `semantic_events.jsonl`. +- MQTT last-will and offline-on-shutdown (§3.6). The server registers no LWT and SIGTERM does not publish `offline`, so availability stays as last published after the server stops. +- Matter consumption of the semantic states. + +--- + *ADR-115 is the integration story that turns RuView from "another sensing platform" into "drop-in upgrade for any HA install **and** any Matter-controller home." MQTT carries the rich, differentiated telemetry; Matter carries the standardised subset across every controller ecosystem. Numbers 111 and 112 remain reserved per the project ADR-numbering policy.* diff --git a/docs/integrations/home-assistant.md b/docs/integrations/home-assistant.md index 8498a36f9..d34de923f 100644 --- a/docs/integrations/home-assistant.md +++ b/docs/integrations/home-assistant.md @@ -33,59 +33,82 @@ cargo run --release -p wifi-densepose-sensing-server \ --mqtt-username homeassistant ``` -Within ~5 seconds of starting, Home Assistant should auto-create: +Within ~5 seconds of the first CSI frame, Home Assistant should auto-create: -- One **device** per RuView node (named after the MAC or the `friendly_name` from your zones config) -- 17+ **entities** per device (presence, person count, heart rate, breathing rate, motion, fall events, signal strength, zones, and the 10 semantic primitives) +- One **device** per RuView node, named `RuView node -node` (see [Device identity](#device-identity)) +- **15 entities** per device (13 with `--privacy-mode`): the raw signals and six semantic primitives listed in the [Entity reference](#entity-reference) If nothing appears in HA's Settings → Devices, see [Troubleshooting](#troubleshooting). ### 3. Stop the publisher cleanly -Ctrl-C — the publisher pushes `offline` to every availability topic before disconnect so HA marks all entities unavailable instantly. A `kill -9` triggers MQTT LWT, which has the same effect within ~30 s. +Ctrl-C (SIGINT/SIGTERM) stops the server. It does **not** publish `offline` first, and no MQTT last-will is registered, so every availability topic keeps its last retained value until the server starts again (MEASURED: no availability message in the last 6 s of a SIGTERM'd run, see [ADR-115 §11](../adr/ADR-115-home-assistant-integration.md#11-status-note-p45-semantic-states-on-the-live-publisher-2026-10-02)). While the server runs, a node that stops sending frames goes `offline` at the next 30 s availability heartbeat. --- ## Entity reference -RuView publishes three classes of entity. Names below are the `unique_id` slugs — Home Assistant assigns friendly names automatically. +Names below are the `unique_id` slugs; Home Assistant assigns friendly names automatically. Every entity has its own availability topic (`…//availability`, retained). An entity is `online` only while the server has a real source for it. Otherwise it is `offline`, no state is published, and Home Assistant shows **unavailable** rather than a made-up value. -### Raw signals (11 entities) +### Raw signals (9 entities) -| HA entity | Slug | HA component | Unit | Source field | +| HA entity | Slug | HA component | Unit | Source | Available when | +|---|---|---|---|---|---| +| Presence | `presence` | `binary_sensor` | — | `nodes[].node_inference.classification != "absent"` | always | +| Person count | `person_count` | `sensor` | persons | `persons[]` length, else `estimated_persons` (room-level, same on every node) | always | +| Motion level | `motion_level` | `sensor` | % | node classification: `present_moving` = 100, `present_still`/`absent` = 0 | always | +| Presence score | `presence_score` | `sensor` | % | node classification confidence × 100 while present, else 0 | always | +| Motion energy | `motion_energy` | `sensor` | (dimensionless) | `node_features[].features.motion_band_power`, else the room `features.motion_band_power` | a band-power value was seen in the last 30 s | +| Signal strength | `rssi` | `sensor` | dBm | `nodes[].rssi_dbm` | an RSSI was seen in the last 30 s | +| Heart rate | `heart_rate` | `sensor` | bpm | `vital_signs.heart_rate_bpm` (room-level) | a published estimate in the last 30 s | +| Breathing rate | `breathing_rate` | `sensor` | bpm | `vital_signs.breathing_rate_bpm` (room-level) | a published estimate in the last 30 s | +| Fall detected | `fall` | `event` | — | rising edge of the node's `edge_vitals.fall_detected` | the node sent `edge_vitals` in the last 30 s | + +The server publishes heart and breathing rate only after explicit calibration, with exactly one occupant and a passing quality gate. Until then both entities stay unavailable. Heart rate and breathing rate are **biometric**: `--privacy-mode` removes them from discovery and state (see [Privacy](#privacy)). + +### Semantic automation primitives (6 entities) + +Small finite-state machines run inside the server, one set per node, sampled once per second ([ADR-115 §3.12](../adr/ADR-115-home-assistant-integration.md#312-semantic-automation-primitives-ha-mind), P4.5). All of them are unavailable for the first 60 s after start (warmup). After that each publishes its current state, then on every change. + +| HA entity | Slug | HA component | What it fires on | Available when | |---|---|---|---|---| -| Presence | `presence` | `binary_sensor` | — | `edge_vitals.presence` | -| Person count | `person_count` | `sensor` | persons | `edge_vitals.n_persons` | -| Heart rate | `heart_rate` | `sensor` | bpm | `edge_vitals.heartrate_bpm` | -| Breathing rate | `breathing_rate` | `sensor` | bpm | `edge_vitals.breathing_rate_bpm` | -| Motion level | `motion_level` | `sensor` | % | `edge_vitals.motion` × 100 | -| Motion energy | `motion_energy` | `sensor` | (dimensionless) | `edge_vitals.motion_energy` | -| Fall detected | `fall` | `event` | — | `edge_vitals.fall_detected` | -| Presence score | `presence_score` | `sensor` | % | `edge_vitals.presence_score` × 100 | -| Signal strength | `rssi` | `sensor` | dBm | `edge_vitals.rssi` | -| Zone occupancy | `zone_occupancy` | `binary_sensor` | — | `sensing_update.zones` | -| Pose keypoints | `pose` | `sensor` (attrs) | — | `pose_data.keypoints` (opt-in via `--mqtt-publish-pose`) | +| Room active | `room_active` | `binary_sensor` | presence + motion >10% within a 30 s window; off after 10 min idle | after warmup | +| No movement (safety) | `no_movement` | `binary_sensor` | presence + motion <1% for 30 min | after warmup | +| Elderly inactivity anomaly | `elderly_inactivity_anomaly` | `binary_sensor` | stillness > 2× the longest stillness seen since start (floor 30 min) | after warmup | +| Someone sleeping | `someone_sleeping` | `binary_sensor` | presence + motion <5% + BR ∈ [8,20] bpm for 5 min | after warmup, while breathing rate is available | +| Possible distress | `possible_distress` | `binary_sensor` | HR ≥ 1.5× resting baseline + motion >20% + no fall, for 60 s | after warmup, while heart rate is available | +| Fall risk elevated | `fall_risk_elevated` | `sensor` | 0–100 score: 10 per fall in 24 h + 50 × motion variance over 60 s | after warmup, while the fall detector is available | -Heart rate, breathing rate, and pose are **biometric** entities — they are stripped from MQTT (and never published over Matter) when `--privacy-mode` is set. See [Privacy](#privacy) below. +Per-node motion is a three-level classification, so the motion thresholds above effectively mean "moving" or "not moving". The precision and recall figures in [`semantic-primitives-metrics.md`](./semantic-primitives-metrics.md) are CLAIMED: the replay tooling they cite is not in the repository. No `reason` attribute is published yet; the explanation tags stay server-side. -### Semantic automation primitives (10 entities) +### Not announced -These are the inferred high-level states that customer automations actually use. Each one is a small finite-state machine running server-side with explicit warmup, hysteresis, and refractory windows. Per-primitive precision/recall is published in [`semantic-primitives-metrics.md`](./semantic-primitives-metrics.md). +These entities from ADR-115 §3.1 and §3.12 are not announced, because the sensing broadcast has nothing to drive them. Announcing them only produced entities stuck at "unknown" ([#2085](https://github.com/ruvnet/RuView/issues/2085)). -| HA entity | Slug | HA component | What it fires on | -|---|---|---|---| -| Someone sleeping | `someone_sleeping` | `binary_sensor` | presence + motion<5% + BR ∈ [8,20] bpm sustained for 5 min | -| Possible distress | `possible_distress` | `binary_sensor` | HR > 1.5× baseline + motion >20% + no fall, sustained 60 s | -| Room active | `room_active` | `binary_sensor` | motion >10% in a 30-s rolling window | -| Elderly inactivity anomaly | `elderly_inactivity_anomaly` | `binary_sensor` | idle > 2× observed-max-idle baseline | -| Meeting in progress | `meeting_in_progress` | `binary_sensor` | ≥2 persons + low-amplitude motion for 10 min | -| Bathroom occupied | `bathroom_occupied` | `binary_sensor` | presence + active zone tagged `bathroom` | -| Fall risk elevated | `fall_risk_elevated` | `sensor` | 0–100 score; event fires on ≥70 crossing | -| Bed exit (overnight) | `bed_exit` | `event` | sleeping → presence leaves bed zone between 22:00–06:00 | -| No movement (safety) | `no_movement` | `binary_sensor` | presence + motion <1% for 30 min | -| Multi-room transition | `multi_room_transition` | `event` | zone X exit + zone Y enter within 10 s | +| Slug | Why not | +|---|---| +| `zone_occupancy`, `bathroom_occupied`, `bed_exit`, `multi_room_transition` | the broadcast carries no zones, and the server loads no zone-tag map | +| `meeting_in_progress` | needs a 1–20 % motion level; per-node motion is three-level | +| `pose` | `pose_keypoints` is never populated on the broadcast, and the `persons[].keypoints` skeletons are synthesized. `--mqtt-publish-pose` is accepted but logs a warning and announces nothing | -Every state change carries a `reason` attribute (e.g. `["motion<5%", "br=12bpm", "presence=true"]`) so you can template against it in HA automations to understand why an automation triggered. +### Publish rates and retention + +| Entity kind | When it publishes | QoS / retain | +|---|---|---| +| `presence` and semantic `binary_sensor`s | on change, plus a heartbeat every 60 s if unchanged | 1 / retained | +| `person_count`, `motion_level`, `motion_energy`, `presence_score`, `fall_risk_elevated` | at most `--mqtt-rate-motion` / `--mqtt-rate-count` (default 1 Hz) | 0 / not retained | +| `heart_rate`, `breathing_rate` | at most `--mqtt-rate-vitals` (default 0.2 Hz) | 0 / not retained | +| `rssi` | at most `--mqtt-rate-rssi` (default 0.1 Hz) | 0 / not retained | +| `fall` | once per detected fall | 1 / not retained | +| availability | on every transition, and all topics every 30 s | 1 / retained | + +Before [#2093](https://github.com/ruvnet/RuView/issues/2093), `presence` was published on every CSI frame (MEASURED 37.9 msg/s with one node, 1 msg per ~45 s after the fix; see [ADR-115 §11](../adr/ADR-115-home-assistant-integration.md#11-status-note-p45-semantic-states-on-the-live-publisher-2026-10-02)). + +### Device identity + +Each node's HA device id is `wifi_densepose_-node`, where `` is the node id from the CSI frames. Without `--mqtt-client-id`, the server creates `wifi-densepose-<8 hex digits>` on first start and keeps it in `/mqtt_client_id`, so device ids survive restarts. Keep `--data-dir` on persistent storage (a Docker volume, for example). If the file cannot be written, the server logs a warning and falls back to a per-process id for that run. `--mqtt-client-id` overrides the persisted id. Use it to pin ids across reinstalls or to run two servers against one broker. + +Releases before this change used `wifi-densepose-`, so every restart created a new set of devices. To remove the leftovers, publish an empty retained message to each stale `homeassistant/+/wifi_densepose_…/+/config` topic, or delete the devices in Home Assistant. ### Matter device-type mapping @@ -115,7 +138,7 @@ Per ADR-115 §3.11.1, the Matter Bridge exposes a subset on standard clusters so | `--mqtt-port ` | 1883 (8883 with TLS) | Broker port | | `--mqtt-username ` | — | Username for broker auth | | `--mqtt-password-env ` | `MQTT_PASSWORD` | Env var holding the password | -| `--mqtt-client-id ` | `wifi-densepose-` | MQTT client ID | +| `--mqtt-client-id ` | persisted `wifi-densepose-` in `/mqtt_client_id` | MQTT client ID; prefixes every HA device id | | `--mqtt-prefix ` | `homeassistant` | Discovery topic prefix | | `--mqtt-tls` | off | Encrypt connection | | `--mqtt-ca-file ` | — | Pinned CA for TLS / mTLS | @@ -126,43 +149,21 @@ Per ADR-115 §3.11.1, the Matter Bridge exposes a subset on standard clusters so | `--mqtt-rate-motion ` | 1.0 | Motion publish rate (Hz) | | `--mqtt-rate-count ` | 1.0 | Person-count publish rate (Hz) | | `--mqtt-rate-rssi ` | 0.1 | RSSI publish rate (Hz) | -| `--mqtt-publish-pose` | off | Enable pose-keypoint publication | -| `--mqtt-rate-pose ` | 1.0 | Pose publish rate when enabled | +| `--mqtt-publish-pose` | off | Accepted; no pose source exists, so it only logs a warning | +| `--mqtt-rate-pose ` | 1.0 | Unused while no pose entity is announced | | `--privacy-mode` | off | Strip HR/BR/pose from MQTT and Matter | +| `--data-dir ` | `data` | Holds `mqtt_client_id` (and other server state) | | `--matter` | off | Enable the HA-FABRIC Matter Bridge | | `--matter-setup-file ` | — | Where to write the QR + manual code | | `--matter-reset` | off | Wipe fabric credentials and re-commission | | `--matter-vendor-id ` | `0xFFF1` (dev) | CSA-assigned vendor ID | | `--matter-product-id ` | `0x8001` | Product ID | -| `--semantic` | on | Enable inference layer | -| `--semantic-thresholds-file ` | — | Per-primitive threshold overrides | -| `--semantic-zones-file ` | — | Zone-tag map (`bathroom`, `bedroom`, …) | -| `--no-semantic ` | — | Disable a specific primitive (repeatable) | -### Zone tag file format +The `--semantic`, `--semantic-thresholds-file`, `--semantic-zones-file` and `--no-semantic` flags from ADR-115 §3.12.5 are **not accepted by `sensing-server` yet**. The primitives always run with the defaults in `PrimitiveConfig`, and no zone map is loaded. -```yaml -# semantic-zones.yaml — passed to --semantic-zones-file -zones: - bathroom: ["zone_3", "zone_7"] - bedroom: ["zone_1"] - kitchen: ["zone_2"] - living: ["zone_5"] -bed_zones: ["zone_1"] -``` +### Zone tags and threshold overrides (not implemented) -### Threshold overrides - -```yaml -# semantic-thresholds.yaml — passed to --semantic-thresholds-file -sleep_dwell_secs: 300 -distress_hr_multiple: 1.5 -room_active_motion_threshold: 0.10 -elderly_anomaly_multiple: 2.0 -meeting_min_persons: 2 -no_movement_dwell_secs: 1800 -fall_risk_event_threshold: 70.0 -``` +ADR-115 §3.12.5 specifies a zone-tag file and a threshold-override file. Neither is read by `sensing-server` today, which is why the zone-based primitives are not announced. --- @@ -170,9 +171,11 @@ fall_risk_event_threshold: 70.0 When deploying in **healthcare**, **AAL (aging-in-place)**, or **commercial** settings, set `--privacy-mode`. This: -- **Strips** heart rate, breathing rate, and pose keypoints from every outbound MQTT publication. -- **Suppresses discovery** for those entities entirely — HA never even sees they exist. -- **Keeps every semantic primitive enabled.** Sleeping / distress / room-active / etc are *inferred* states. The inference happens server-side and only the boolean or score crosses the wire. This is the architectural win that makes the platform deployable in regulated contexts. +- **Strips** heart rate and breathing rate from every outbound MQTT publication (pose is never announced; see [Not announced](#not-announced)). +- **Suppresses discovery** for those entities entirely: Home Assistant never sees them, and no availability or state topic is published for them. +- **Keeps the semantic primitives enabled.** `someone_sleeping` and `possible_distress` still use breathing and heart rate *inside the server*; only the ON/OFF state crosses the wire (ADR-115 §3.12.3). If that is too much for your deployment, drop those two entities in Home Assistant. + +With `--privacy-mode`, 13 entities are announced per node. Privacy mode reduces what leaves the server. It does not by itself make a deployment compliant with any regulation. Always pair `--privacy-mode` with `--mqtt-tls` on non-localhost brokers. @@ -210,7 +213,6 @@ action: title: "Possible distress detected" message: > RuView flagged sustained elevated heart rate + agitated motion. - Reason: {{ state_attr(trigger.entity_id, 'reason') }}. ``` ### 2. Dim hallway when someone is sleeping @@ -256,6 +258,8 @@ action: ### 3. Wake-up routine on bed exit +> `bed_exit` is not published by `sensing-server` today (it needs a bed-zone map; see [Not announced](#not-announced)). This blueprint will not fire until that lands. + ```yaml blueprint: name: RuView — wake-up routine on bed exit @@ -335,8 +339,8 @@ cards: entity: binary_sensor.ruview_kitchen_presence name: Kitchen - type: tile - entity: binary_sensor.ruview_bathroom_occupied - name: Bathroom + entity: binary_sensor.ruview_bedroom_room_active + name: Bedroom active ``` --- @@ -373,6 +377,15 @@ All three accept the same HA discovery topics RuView publishes. Performance and wscat -c ws://localhost:8765/ws/sensing ``` 3. Confirm rate limits aren't dropping everything: `--mqtt-rate-vitals 1.0` for diagnosis (default 0.2 Hz = every 5 s). +4. Unchanged binary states (presence, the semantic primitives) publish once per change and then every 60 s. A flat line in HA history is expected while nothing changes. + +### Entity shows "unavailable" + +The server has no source for it right now: heart and breathing rate need explicit calibration and a single occupant, `fall` needs a node that sends `edge_vitals`, and the semantic primitives wait 60 s after start (`someone_sleeping` and `possible_distress` also need breathing or heart rate). Check the availability topic with `mosquitto_sub -t 'homeassistant/+/+//availability' -v`. + +### Duplicate devices after a restart + +Older releases derived device ids from the process id. See [Device identity](#device-identity) for how to remove the stale devices; the current default id is persisted in `--data-dir`. ### "Plaintext MQTT on non-localhost broker" WARN @@ -391,7 +404,7 @@ Per [ADR-115 §3.9](../adr/ADR-115-home-assistant-integration.md#39-tls--auth), ## Applications — what people actually do with this -The 21 entities per node — 11 raw signals (presence, person count, breathing, heart rate, motion, RSSI, etc.) and 10 inferred semantic states (someone-sleeping, possible-distress, room-active, elderly-inactivity-anomaly, meeting-in-progress, bathroom-occupied, fall-risk-elevated, bed-exit, no-movement, multi-room-transition) — slot into Home Assistant like any other sensor. The list below groups real-world uses so you can pick the ones that match your space. +The 15 entities per node (9 raw signals and 6 semantic states, see the [Entity reference](#entity-reference)) slot into Home Assistant like any other sensor. Some uses below rely on entities that are not announced yet (zones, bathroom, bed exit, meeting); those wait on a zone source or a finer motion level. The list below groups real-world uses so you can pick the ones that match your space. ### Personal & home @@ -495,13 +508,9 @@ A few patterns appear over and over; if you understand these you can build most ### What about regulated environments? -Run RuView with `--privacy-mode` and only the 10 inferred semantic states reach Home Assistant — heart rate, breathing rate, and pose values are stripped at the MQTT wire. Per ADR-115 §6, this passes: +With `--privacy-mode`, heart rate and breathing rate never reach the broker; their entities are not even announced. The semantic states still reach Home Assistant, and two of them (`someone_sleeping`, `possible_distress`) are derived from those vitals inside the server, so their ON/OFF values are themselves health-related information. -- **HIPAA-style minimum-necessary** (no biometric numbers leave the device) -- **GDPR purpose-limitation** (the inferred states are the smallest dataset that supports the automation) -- **CCPA "sensitive personal information"** (no health data crosses the wire) - -The fall-risk-elevated / possible-distress / someone-sleeping flags still work — they're computed *inside* the sensor pipeline and only the boolean outputs are published. That's the architectural win that makes RuView deployable in care homes, hospitals, schools, and shared-housing scenarios where raw biometrics would be a non-starter. +Privacy mode narrows what crosses the MQTT wire. It is not a compliance certification, and RuView makes no HIPAA, GDPR or CCPA compliance claim. Whether a deployment meets a regulation depends on the whole system: broker access control and TLS, Home Assistant's recorder retention, who can see the dashboards, consent, and local law. Assess that with whoever is responsible for compliance in your setting. ## References