mirror of
https://github.com/ruvnet/RuView.git
synced 2026-10-21 22:22:33 -05:00
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 <data-dir>/mqtt_client_id. --mqtt-client-id still overrides it. Refs #2085, #2093 Co-Authored-By: claude-flow <ruv@ruv.net>
This commit is contained in:
Mathew Beaneandclaude-flow committed
1 parent
03a897938f
commit
e351d0fce1
14 files changed
+1461
-394
No files matched your search
@@ -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),
|
||||
|
||||
@@ -126,9 +126,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
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),
|
||||
|
||||
@@ -29,7 +29,8 @@ pub struct MqttArgs {
|
||||
#[arg(long, default_value = "MQTT_PASSWORD")]
|
||||
pub mqtt_password_env: String,
|
||||
|
||||
/// MQTT client ID (default: wifi-densepose-<pid>)
|
||||
/// MQTT client ID; prefixes every Home Assistant device id (default: a
|
||||
/// `wifi-densepose-<hex>` id persisted in `<data-dir>/mqtt_client_id`)
|
||||
#[arg(long, env = "RUVIEW_MQTT_CLIENT_ID")]
|
||||
pub mqtt_client_id: Option<String>,
|
||||
|
||||
|
||||
@@ -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<wifi_densepose_sensing_server::mqtt::state::VitalsSnapshot> {
|
||||
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<f64>| {
|
||||
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::<serde_json::Value>(&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};
|
||||
|
||||
@@ -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<u64, EdgeFall>,
|
||||
}
|
||||
|
||||
impl SensingBridge {
|
||||
/// `base_id` prefixes every node id (`<base_id>-node<N>`); it is the MQTT
|
||||
/// client id, so it must be stable across restarts (#2093).
|
||||
pub fn new(base_id: impl Into<String>) -> 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<VitalsSnapshot> {
|
||||
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<bool> {
|
||||
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<VitalsSnapshot> {
|
||||
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<f64>, 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<f64>,
|
||||
motion_energy: Option<f64>,
|
||||
fall_detected: Option<bool>| 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<VitalsSnapshot> {
|
||||
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
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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 `<data_dir>/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<String> {
|
||||
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::<u32>());
|
||||
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");
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<Duration>,
|
||||
pending_fall: bool,
|
||||
sourced: HashMap<EntityKind, Duration>,
|
||||
availability: HashMap<EntityKind, bool>,
|
||||
}
|
||||
|
||||
/// Per-publisher planning state, keyed by node.
|
||||
pub struct PublishPlanner {
|
||||
entities: Vec<EntityKind>,
|
||||
rates: PublishRates,
|
||||
privacy_mode: bool,
|
||||
semantic: PrimitiveConfig,
|
||||
nodes: HashMap<String, NodePlan>,
|
||||
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<StateMessage> {
|
||||
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<StateMessage> {
|
||||
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<StateMessage> {
|
||||
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<SemanticKind> {
|
||||
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;
|
||||
@@ -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<String, String> {
|
||||
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<String, String> = 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);
|
||||
}
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
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<String, (OwnedDiscoveryBuilder, Instant)> =
|
||||
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,
|
||||
|
||||
@@ -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<bool>,
|
||||
pub motion: f64, // 0.0–1.0
|
||||
pub motion_energy: f64,
|
||||
/// `None` when the frame carries no motion band power.
|
||||
pub motion_energy: Option<f64>,
|
||||
pub presence_score: f64, // 0.0–1.0
|
||||
pub breathing_rate_bpm: Option<f64>,
|
||||
pub heartrate_bpm: Option<f64>,
|
||||
@@ -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<StateMessage> {
|
||||
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<f64>) -> Option<StateMessage> {
|
||||
@@ -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();
|
||||
|
||||
@@ -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<bool> {
|
||||
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<SemanticEvent> {
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
|
||||
Reference in new issue
Block a user