//! State payload encoding + rate limiting (ADR-115 §3.5 / §3.7). //! //! This module owns the translation from internal `sensing-server` //! broadcast messages (`pose_data`, `edge_vitals`, `sensing_update`) //! into the per-entity MQTT state-topic payloads consumed by Home //! Assistant. It is gated behind the `mqtt` feature flag at the call //! site, but the encoders and rate-limiter logic compile without any //! network deps so they're testable under `--no-default-features`. //! //! Per ADR-115 §3.5, state-topic QoS / retain / cadence is: //! //! | Topic kind | QoS | Retain | Cadence | //! |------------------------|-----|--------|------------------------| //! | `sensor/*/state` | 0 | no | rate-limited per §3.7 | //! | `binary_sensor/*/state`| 1 | yes | on change only | //! | `event/*/state` | 1 | no | on event | //! | `*/availability` | 1 | yes | LWT + 30 s heartbeat | //! //! Per ADR-115 §3.7, default rates are: //! //! - presence binary : on change //! - person count : 1.0 Hz //! - vitals (HR / BR) : 0.2 Hz (every 5 s) //! - motion level : 1.0 Hz //! - fall events : on event (no rate limit) //! - RSSI : 0.1 Hz //! - pose : 1.0 Hz when `--mqtt-publish-pose` (off by default) //! - zones : on change use std::collections::HashMap; use std::time::Duration; use serde::Serialize; use serde_json::Value; use super::config::PublishRates; use super::discovery::{DiscoveryComponent, EntityKind}; /// Encoded outbound MQTT publication. `topic` is fully-qualified /// (already prefixed with the discovery namespace + node id). `payload` /// is the UTF-8 string the broker should publish on that topic. #[derive(Debug, Clone, PartialEq, Eq)] pub struct StateMessage { pub topic: String, pub payload: String, pub qos: u8, pub retain: bool, } impl StateMessage { pub fn new(topic: String, payload: String, component: DiscoveryComponent, is_change_only: bool) -> Self { let (qos, retain) = match component { DiscoveryComponent::BinarySensor => (1, is_change_only), DiscoveryComponent::Event => (1, false), DiscoveryComponent::Sensor => (0, false), }; Self { topic, payload, qos, retain } } } /// Sample-rate-limit decisions, per entity. Tracks the last-emitted /// instant per entity and gates further emissions accordingly. Time is /// supplied by the caller so the limiter is testable without a clock. #[derive(Debug, Default)] pub struct RateLimiter { last: HashMap, } impl RateLimiter { /// Build a fresh limiter with no per-entity history. pub fn new() -> Self { Self { last: HashMap::new() } } /// Decide whether a sample for `entity` is allowed to publish at /// `now`, given the configured `rates`. Returns true to publish /// (and updates last-emitted state); false to drop. pub fn allow(&mut self, entity: EntityKind, now: Duration, rates: &PublishRates) -> bool { let min_gap = match rate_hz_for(entity, rates) { // Zero / negative Hz → emit only on change (caller path). // Here we treat it as "always allow" because the caller is // already gating with change detection. rate if rate <= 0.0 => return true, rate => Duration::from_secs_f64(1.0 / rate), }; match self.last.get(&entity) { Some(&prev) if now.saturating_sub(prev) < min_gap => false, _ => { self.last.insert(entity, now); true } } } /// Reset all per-entity history. Used after a reconnect so the first /// post-reconnect sample is emitted promptly. pub fn reset(&mut self) { self.last.clear(); } } /// Look up the configured Hz for an entity. Numerical entities use the /// `rates` struct; non-rate-limited entities (events / change-only) /// return 0.0 to short-circuit limiting. fn rate_hz_for(entity: EntityKind, rates: &PublishRates) -> f64 { match entity { // Change-only / event entities — caller drives them. EntityKind::Presence | EntityKind::ZoneOccupancy | EntityKind::FallDetected | EntityKind::BedExit | EntityKind::MultiRoomTransition | EntityKind::SomeoneSleeping | EntityKind::PossibleDistress | EntityKind::RoomActive | EntityKind::ElderlyInactivityAnomaly | EntityKind::MeetingInProgress | EntityKind::BathroomOccupied | EntityKind::NoMovement => 0.0, // Rate-limited measurements. EntityKind::PersonCount => rates.count_hz, EntityKind::BreathingRate | EntityKind::HeartRate => rates.vitals_hz, EntityKind::MotionLevel | EntityKind::MotionEnergy => rates.motion_hz, EntityKind::PresenceScore => rates.motion_hz, EntityKind::Rssi => rates.rssi_hz, EntityKind::PoseKeypoints => rates.pose_hz, EntityKind::FallRiskElevated => rates.motion_hz, } } // ─── Per-entity state payload encoders ─────────────────────────────────── /// Inputs the encoder accepts. The caller (publisher loop) projects the /// internal server broadcast into this struct so the encoder never /// touches the original `serde_json::Value`s directly. Avoids leaking /// the server's internal schema into ADR-115's wire format. #[derive(Debug, Clone, Default)] pub struct VitalsSnapshot { pub node_id: String, pub timestamp_ms: i64, pub presence: bool, pub fall_detected: bool, pub motion: f64, // 0.0–1.0 pub motion_energy: f64, pub presence_score: f64, // 0.0–1.0 pub breathing_rate_bpm: Option, pub heartrate_bpm: Option, pub n_persons: u32, pub rssi_dbm: Option, pub vital_confidence: f64, // 0.0–1.0 } #[derive(Serialize, Debug)] struct NumberWithConfidence { bpm: f64, confidence: f64, ts: String, } #[derive(Serialize, Debug)] struct MotionStatePayload { level_pct: f64, ts: String, } #[derive(Serialize, Debug)] struct EnergyStatePayload { energy: f64, ts: String, } #[derive(Serialize, Debug)] struct CountStatePayload { n_persons: u32, ts: String, } #[derive(Serialize, Debug)] struct PresenceScorePayload { score_pct: f64, ts: String, } #[derive(Serialize, Debug)] struct RssiPayload { dbm: f64, ts: String, } #[derive(Serialize, Debug)] struct FallEventPayload { event_type: &'static str, ts: String, #[serde(skip_serializing_if = "Option::is_none")] confidence: Option, } /// Encoder bundle that knows how to render each entity's state payload /// from a [`VitalsSnapshot`]. Operates on an existing [`DiscoveryBuilder`] /// so topics are guaranteed to match what was advertised at discovery /// time. pub struct StateEncoder<'a> { pub builder: &'a super::discovery::DiscoveryBuilder<'a>, } impl<'a> StateEncoder<'a> { /// Build the binary state ("ON"/"OFF") topic + payload for the given /// boolean entity. pub fn boolean(&self, entity: EntityKind, on: bool) -> Option { if !matches!(entity.component(), DiscoveryComponent::BinarySensor) { return None; } let topic = format!( "{}/{}/wifi_densepose_{}/{}/state", self.builder.discovery_prefix, entity.component().as_str(), self.builder.node_id, entity.topic_slug(), ); let payload = if on { "ON" } else { "OFF" }.to_string(); Some(StateMessage::new(topic, payload, entity.component(), true)) } /// Numeric/measurement state encoder. pub fn numeric(&self, entity: EntityKind, snap: &VitalsSnapshot) -> Option { if !matches!(entity.component(), DiscoveryComponent::Sensor) { return None; } let ts = iso_ts(snap.timestamp_ms); let payload_value: Value = match entity { EntityKind::PersonCount => serde_json::to_value(CountStatePayload { n_persons: snap.n_persons, ts: ts.clone(), }).ok()?, EntityKind::BreathingRate => { let bpm = snap.breathing_rate_bpm?; serde_json::to_value(NumberWithConfidence { bpm, confidence: snap.vital_confidence, ts: ts.clone(), }).ok()? } EntityKind::HeartRate => { let bpm = snap.heartrate_bpm?; serde_json::to_value(NumberWithConfidence { bpm, confidence: snap.vital_confidence, ts: ts.clone(), }).ok()? } EntityKind::MotionLevel => serde_json::to_value(MotionStatePayload { level_pct: (snap.motion.clamp(0.0, 1.0)) * 100.0, ts: ts.clone(), }).ok()?, EntityKind::MotionEnergy => serde_json::to_value(EnergyStatePayload { energy: snap.motion_energy, ts: ts.clone(), }).ok()?, EntityKind::PresenceScore => serde_json::to_value(PresenceScorePayload { score_pct: snap.presence_score.clamp(0.0, 1.0) * 100.0, ts: ts.clone(), }).ok()?, EntityKind::Rssi => { let dbm = snap.rssi_dbm?; serde_json::to_value(RssiPayload { dbm, ts: ts.clone() }).ok()? } _ => return None, }; let topic = format!( "{}/{}/wifi_densepose_{}/{}/state", self.builder.discovery_prefix, entity.component().as_str(), self.builder.node_id, entity.topic_slug(), ); let payload = serde_json::to_string(&payload_value).ok()?; 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 { if !matches!(entity.component(), DiscoveryComponent::Event) { return None; } let payload_json = FallEventPayload { event_type, ts: iso_ts(ts_ms), confidence }; let payload = serde_json::to_string(&payload_json).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::Event, false)) } } fn iso_ts(ms: i64) -> String { // Avoid pulling chrono into a hot path: format manually as ISO-8601 // UTC. chrono is already in the crate's deps, but we keep this // encoder allocation-light for benchmark numbers. let secs = ms / 1000; let nanos = ((ms % 1000) * 1_000_000) as u32; let dt = chrono::DateTime::::from_timestamp(secs, nanos) .unwrap_or_else(|| chrono::DateTime::::from_timestamp(0, 0).unwrap()); dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true) } #[cfg(test)] mod tests { use super::*; use crate::mqtt::discovery::DiscoveryBuilder; fn builder() -> DiscoveryBuilder<'static> { DiscoveryBuilder { discovery_prefix: "homeassistant", node_id: "aabbccddeeff", node_friendly_name: Some("Bedroom"), sw_version: "v0.7.0", model: "ESP32-S3 CSI node", via_device: None, } } fn rates() -> PublishRates { PublishRates { vitals_hz: 0.2, motion_hz: 1.0, count_hz: 1.0, rssi_hz: 0.1, pose_hz: 1.0, } } fn snap() -> VitalsSnapshot { VitalsSnapshot { node_id: "aabbccddeeff".into(), timestamp_ms: 1779_512_400_000, presence: true, fall_detected: false, motion: 0.35, motion_energy: 1234.5, presence_score: 0.91, breathing_rate_bpm: Some(14.2), heartrate_bpm: Some(68.2), n_persons: 1, rssi_dbm: Some(-52.0), vital_confidence: 0.87, } } // ─── Rate limiter ──────────────────────────────────────────────── #[test] fn rate_limiter_first_sample_always_passes() { let mut rl = RateLimiter::new(); assert!(rl.allow(EntityKind::HeartRate, Duration::ZERO, &rates())); } #[test] fn rate_limiter_drops_within_gap() { let mut rl = RateLimiter::new(); let r = rates(); // 0.2 Hz → 5 s gap. assert!(rl.allow(EntityKind::HeartRate, Duration::from_secs(0), &r)); assert!(!rl.allow(EntityKind::HeartRate, Duration::from_secs(1), &r)); assert!(!rl.allow(EntityKind::HeartRate, Duration::from_secs(4), &r)); } #[test] fn rate_limiter_allows_after_gap() { let mut rl = RateLimiter::new(); let r = rates(); assert!(rl.allow(EntityKind::HeartRate, Duration::from_secs(0), &r)); // 5 s gap met → allow. assert!(rl.allow(EntityKind::HeartRate, Duration::from_secs(5), &r)); } #[test] fn rate_limiter_per_entity_independent() { let mut rl = RateLimiter::new(); let r = rates(); assert!(rl.allow(EntityKind::HeartRate, Duration::from_secs(0), &r)); // Different entity, same instant → independent budget. assert!(rl.allow(EntityKind::MotionLevel, Duration::from_secs(0), &r)); } #[test] fn rate_limiter_change_only_entities_always_allow() { let mut rl = RateLimiter::new(); let r = rates(); // Presence is change-only → rate=0 → unlimited; caller does change detection. for s in 0..3 { assert!(rl.allow(EntityKind::Presence, Duration::from_secs(s), &r)); } } #[test] fn rate_limiter_reset_re_enables_immediate_publish() { let mut rl = RateLimiter::new(); let r = rates(); assert!(rl.allow(EntityKind::HeartRate, Duration::from_secs(0), &r)); assert!(!rl.allow(EntityKind::HeartRate, Duration::from_secs(1), &r)); rl.reset(); // Post-reset: first sample passes. assert!(rl.allow(EntityKind::HeartRate, Duration::from_secs(1), &r)); } // ─── Boolean / binary_sensor encoder ───────────────────────────── #[test] fn boolean_encoder_emits_on_off_payload() { let b = builder(); let enc = StateEncoder { builder: &b }; let on = enc.boolean(EntityKind::Presence, true).unwrap(); assert_eq!(on.payload, "ON"); assert_eq!(on.qos, 1); assert!(on.retain, "binary_sensor state must be retained per §3.5"); let off = enc.boolean(EntityKind::Presence, false).unwrap(); assert_eq!(off.payload, "OFF"); } #[test] fn boolean_encoder_rejects_non_binary_entities() { let b = builder(); let enc = StateEncoder { builder: &b }; assert!(enc.boolean(EntityKind::HeartRate, true).is_none()); assert!(enc.boolean(EntityKind::FallDetected, true).is_none()); } #[test] fn boolean_topic_matches_discovery_state_topic() { let b = builder(); let enc = StateEncoder { builder: &b }; let msg = enc.boolean(EntityKind::Presence, true).unwrap(); assert_eq!( msg.topic, "homeassistant/binary_sensor/wifi_densepose_aabbccddeeff/presence/state" ); } // ─── Numeric / sensor encoder ──────────────────────────────────── #[test] fn numeric_encoder_emits_bpm_payload_for_heart_rate() { let b = builder(); let enc = StateEncoder { builder: &b }; let s = snap(); let msg = enc.numeric(EntityKind::HeartRate, &s).unwrap(); let json: serde_json::Value = serde_json::from_str(&msg.payload).unwrap(); assert_eq!(json["bpm"], 68.2); assert_eq!(json["confidence"], 0.87); assert_eq!(msg.qos, 0, "sensor state is QoS 0 per §3.5"); assert!(!msg.retain); } #[test] fn numeric_encoder_emits_motion_percent_payload() { let b = builder(); let enc = StateEncoder { builder: &b }; let s = snap(); let msg = enc.numeric(EntityKind::MotionLevel, &s).unwrap(); let json: serde_json::Value = serde_json::from_str(&msg.payload).unwrap(); // 0.35 → 35.0% assert_eq!(json["level_pct"], 35.0); } #[test] fn numeric_encoder_returns_none_when_optional_field_missing() { let b = builder(); let enc = StateEncoder { builder: &b }; let mut s = snap(); s.heartrate_bpm = None; assert!(enc.numeric(EntityKind::HeartRate, &s).is_none()); } #[test] fn numeric_encoder_clamps_out_of_range_motion() { let b = builder(); let enc = StateEncoder { builder: &b }; let mut s = snap(); s.motion = 1.7; // pathological — clamp to 1.0 then ×100. let msg = enc.numeric(EntityKind::MotionLevel, &s).unwrap(); let json: serde_json::Value = serde_json::from_str(&msg.payload).unwrap(); assert_eq!(json["level_pct"], 100.0); } #[test] fn numeric_encoder_rejects_non_sensor_entities() { let b = builder(); let enc = StateEncoder { builder: &b }; let s = snap(); assert!(enc.numeric(EntityKind::Presence, &s).is_none()); assert!(enc.numeric(EntityKind::FallDetected, &s).is_none()); } // ─── Event encoder ─────────────────────────────────────────────── #[test] fn event_encoder_emits_fall_payload() { let b = builder(); let enc = StateEncoder { builder: &b }; let msg = enc .event(EntityKind::FallDetected, "fall_detected", 1779_512_400_000, Some(0.87)) .unwrap(); let json: serde_json::Value = serde_json::from_str(&msg.payload).unwrap(); assert_eq!(json["event_type"], "fall_detected"); assert_eq!(json["confidence"], 0.87); assert_eq!(msg.qos, 1); assert!(!msg.retain, "events must never be retained — HA would replay old falls"); } #[test] fn event_encoder_omits_confidence_when_absent() { let b = builder(); let enc = StateEncoder { builder: &b }; let msg = enc .event(EntityKind::BedExit, "bed_exit", 1779_512_400_000, None) .unwrap(); assert!(!msg.payload.contains("confidence")); } #[test] fn event_encoder_rejects_non_event_entities() { let b = builder(); let enc = StateEncoder { builder: &b }; assert!(enc.event(EntityKind::Presence, "x", 0, None).is_none()); assert!(enc.event(EntityKind::HeartRate, "x", 0, None).is_none()); } #[test] fn iso_ts_is_rfc3339_utc_with_millis() { let ts = iso_ts(1779_512_400_000); assert!(ts.ends_with("Z")); assert!(ts.contains("T")); // .000 suffix from `SecondsFormat::Millis`. assert!(ts.contains("."), "want millisecond fraction in: {}", ts); } }