From 8ecac18978f9e7050e982702eaa08c4c3e914d0c Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 18:28:26 -0400 Subject: [PATCH] feat(sensing-server): labelled MediaTek amplitude heuristic and activity index Real MediaTek frames now produce a room update, but from a heuristic that is labelled as one on every output. The validated ESP32 pipeline needs per-subcarrier phase history this path does not feed it, so it is not used here and is never impersonated. - `mediatek-amplitude-heuristic-v0` (CLAIMED, no labelled evaluation): presence and motion from per-device mean and peak amplitude over a 10 s time window (not a sample count, because arrivals are bursty). Hysteresis means one dip cannot flip the room to absent. Below a confidence of 0.3, a device reports `unknown` and abstains from the room vote rather than asserting "absent". `SensingUpdate.classifier` names it; `nodes[].device_id` and `nodes[].mediatek_diagnostics` say why each device reads as it does. - `activity-index-v0` (CLAIMED): a fast/slow fluctuation index against each device's own baseline, plus an absolute level against a learned quiet floor. It is served in `SensingUpdate.activity` and `GET /api/v1/csi/mediatek/activity`, with `is_occupancy_estimate: false` on every block. `--activity-state ` persists learned floors across restarts. - The summary endpoint adds the heuristic's window statistics. - The MQTT per-node topics key on `device_id` when it is present, so two 64-bit receiver ids cannot collide on `node_id` 0. Developed in whitsentry, a RuView-based project. Co-Authored-By: claude-flow --- .../wifi-densepose-sensing-server/src/main.rs | 885 +++++++++++++- .../src/mediatek_activity.rs | 1024 +++++++++++++++++ .../src/mediatek_csi_ring.rs | 142 ++- .../src/mediatek_heuristic.rs | 688 +++++++++++ 4 files changed, 2729 insertions(+), 10 deletions(-) create mode 100644 v2/crates/wifi-densepose-sensing-server/src/mediatek_activity.rs create mode 100644 v2/crates/wifi-densepose-sensing-server/src/mediatek_heuristic.rs diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 01819bb2a..a8a70f6e1 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -17,9 +17,11 @@ mod field_bridge; mod field_localize; mod model_format; mod multistatic_bridge; +mod mediatek_activity; mod mediatek_csi; mod mediatek_csi_ring; mod mediatek_devices; +mod mediatek_heuristic; mod qualcomm_csi; mod realtek_csi; mod realtek_radar; @@ -117,6 +119,15 @@ struct Args { #[arg(long, default_value = "256")] csi_ring: usize, + /// Path to persist each MediaTek device's learned `activity` floor + /// (JSON `{device_id: floor}`), written every + /// `mediatek_activity::ACTIVITY_STATE_PERSIST_INTERVAL` and loaded at + /// startup, so a restart doesn't reset the quiet-room baseline it took + /// up to FLOOR_INIT_WINDOW (or longer) to learn. In-memory only + /// (no persistence) when omitted. + #[arg(long)] + activity_state: Option, + /// UDP bind address for the CSI receiver (ADR-296). Defaults to /// `127.0.0.1` (loopback only). Binding to a routable address (`0.0.0.0` /// or a LAN IP) is an explicit operator choice and requires `--udp-allow` @@ -488,6 +499,24 @@ struct SensingUpdate { /// fresh node backs the room rather than a frozen online value. #[serde(skip_serializing_if = "Option::is_none")] room_inference: Option, + /// Which detector produced `classification`/`nodes[].node_inference`, when + /// it isn't the validated ESP32 `extract_features_from_frame` / + /// `VitalSignDetector` pipeline (that path leaves this `None`, matching + /// its pre-existing wire format). Set to + /// `mediatek_heuristic::CLASSIFIER_ID` on frames derived from the + /// unvalidated MediaTek amplitude heuristic so consumers never mistake + /// one for the other. + #[serde(skip_serializing_if = "Option::is_none")] + classifier: Option, + /// Relative fast/slow activity index (2026-09-20, `mediatek_activity`, + /// `activity-index-v0`) — `{room:{level,index,peak_30s}, + /// devices:{device_id:{level,index,fast,slow,peak_30s,frames_2s}}}`. + /// Answers "is the channel fluctuating more than it typically does + /// here", NOT an absolute presence call — see the module's doc comment + /// for why a room that's always busy trends back toward the middle of + /// the scale. `None` on every non-MediaTek update. + #[serde(skip_serializing_if = "Option::is_none")] + activity: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -497,6 +526,15 @@ struct NodeInfo { position: [f64; 3], amplitude: Vec, subcarrier_count: usize, + /// Stable receiver identity for sources whose real identity isn't a small + /// mesh index, e.g. a MediaTek MTC1 `device_id` (64-bit, serialized as a + /// 16-hex-char string). `None` on ESP32/simulated nodes, which keep using + /// `node_id`. When present, the MQTT mapper + /// (`vitals_snapshots_from_sensing_json`) keys per-node topics on this + /// instead of `node_id`, since collapsing a 64-bit id into `node_id: u8` + /// risks two receivers colliding onto the same topic. + #[serde(skip_serializing_if = "Option::is_none")] + device_id: Option, /// ADR-110 iter 23 — cross-board sync snapshot for this node. /// `None` when no fresh sync packet has been observed (no mesh peer /// reachable, or this node is a singleton). Populated from @@ -506,9 +544,30 @@ struct NodeInfo { /// ADR-297 — this node's *own* inference (classification + confidence + /// freshness). Distinct from the room aggregate; a node reports what it /// sees, with no silent fallback to the room value. `None` on synthetic / - /// placeholder frames that carry no per-node classification. + /// placeholder frames that carry no per-node classification, AND on a + /// MediaTek device whose confidence is below + /// `mediatek_heuristic::UNKNOWN_CONFIDENCE_THRESHOLD` (it abstains from + /// the room vote rather than asserting "absent" on weak evidence — see + /// `mediatek_diagnostics.presence_state` for that case). #[serde(skip_serializing_if = "Option::is_none")] node_inference: Option, + /// MediaTek-heuristic-specific per-device diagnostics (added with the + /// hysteresis fix), so a client can show why a device reads + /// the way it does even when it abstained (`node_inference: None`). + /// `None` on every non-MediaTek node. + #[serde(skip_serializing_if = "Option::is_none")] + mediatek_diagnostics: Option, +} + +/// See `NodeInfo::mediatek_diagnostics`. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct MediatekNodeDiagnostics { + /// "present" | "absent" | "unknown" — see `mediatek_heuristic::PresenceState`. + presence_state: String, + coefficient_of_variation: f64, + baseline_deviation: f64, + sample_count: usize, + span_ms: u64, } /// ADR-110 iter 23 — per-node mesh-sync snapshot embedded in NodeInfo. @@ -1871,6 +1930,12 @@ struct AppStateInner { /// `/api/v1/csi/mediatek/devices`. mediatek_csi_by_device: HashMap, + /// Per-device rolling amplitude history backing the MediaTek heuristic + /// classifier (`mediatek_heuristic`), keyed by `device_id`. Separate from + /// `mediatek_csi_by_device` above: that map holds the last raw snapshot + /// for the `/devices` route; this one accumulates the window the + /// heuristic judges presence/motion from. + mediatek_heuristic_by_device: HashMap, /// Per-device bounded ring buffer of full per-subcarrier amplitude/phase /// (`mediatek_csi_ring`), for per-device channel inspection. /// Sized by `csi_ring_capacity` (the `--csi-ring` flag) when a device's @@ -1879,6 +1944,31 @@ struct AppStateInner { /// Ring size (frame count) for new entries in `mediatek_csi_ring_by_device`. /// Set once from `--csi-ring` at startup. csi_ring_capacity: usize, + /// Per-device fast/slow relative-activity EMA + peak-hold state + /// (`mediatek_activity`, activity-index-v0), keyed by `device_id`. + mediatek_activity_by_device: HashMap, + /// Floors loaded from `--activity-state` at startup (empty if the flag + /// was omitted or the file didn't exist yet), consulted once when a + /// device_id's `ActivityTracker` is first created so a restart doesn't + /// reset its learned quiet-room baseline. Read-only after startup — + /// never mutated once `main()` seeds it. + activity_floor_seed: HashMap, + /// Room-level 30s peak-hold history for the room activity index — + /// separate from each device's own (`ActivityTracker` carries that + /// itself), since the room index is a per-tick aggregate rather than + /// something with its own EMA state. + room_activity_peak_history: std::collections::VecDeque<(std::time::Instant, f64)>, + /// Room-level 30s peak-hold history for the ABSOLUTE (`abs_level`) + /// activity index — separate from `room_activity_peak_history` above, + /// which tracks the relative `level`'s own peak. + room_abs_activity_peak_history: std::collections::VecDeque<(std::time::Instant, f64)>, + /// Decimated (~1 Hz) room + per-device activity history for + /// `GET /api/v1/csi/mediatek/activity`, retained for + /// `mediatek_activity::HISTORY_RETENTION`. + activity_history: std::collections::VecDeque, + /// Gate for `activity_history`'s decimation — the `Instant` the last + /// sample was actually appended (not every ingest tick). + activity_history_last_sampled: Option, /// Latest validated Qualcomm CSI summary; raw matrices are not retained here. latest_qualcomm_csi: Option, /// Instant of the last validated Qualcomm CSI UDP frame. @@ -2592,8 +2682,15 @@ impl AppStateInner { latest_mediatek_csi: None, last_mediatek_frame: None, mediatek_csi_by_device: HashMap::new(), + mediatek_heuristic_by_device: HashMap::new(), mediatek_csi_ring_by_device: HashMap::new(), csi_ring_capacity: 16, + mediatek_activity_by_device: HashMap::new(), + activity_floor_seed: HashMap::new(), + room_activity_peak_history: std::collections::VecDeque::new(), + room_abs_activity_peak_history: std::collections::VecDeque::new(), + activity_history: std::collections::VecDeque::new(), + activity_history_last_sampled: None, latest_qualcomm_csi: None, last_qualcomm_frame: None, latest_realtek_csi: None, @@ -4357,12 +4454,14 @@ async fn wifi_task(state: SharedState, tick_ms: u64) { tick, nodes: vec![NodeInfo { node_id: 0, + device_id: None, rssi_dbm: first_rssi, position: [0.0, 0.0, 0.0], amplitude: multi_ap_frame.amplitudes, subcarrier_count: obs_count, sync: None, // multi-BSSID scan path — no mesh peer node_inference: None, // single aggregate frame; no per-node split + mediatek_diagnostics: None, }], features, classification, @@ -4391,6 +4490,8 @@ async fn wifi_task(state: SharedState, tick_ms: u64) { }, node_features: None, room_inference: None, + classifier: None, + activity: None, }; // Populate persons from the sensing update (Kalman-smoothed via tracker). @@ -4523,12 +4624,14 @@ async fn windows_wifi_fallback_tick(state: &SharedState, seq: u32) { tick, nodes: vec![NodeInfo { node_id: 0, + device_id: None, rssi_dbm, position: [0.0, 0.0, 0.0], amplitude: vec![signal_pct], subcarrier_count: 1, sync: None, // synthetic-RSSI fallback path — no mesh peer node_inference: None, // synthetic fallback; no per-node inference + mediatek_diagnostics: None, }], features, classification, @@ -4557,6 +4660,8 @@ async fn windows_wifi_fallback_tick(state: &SharedState, seq: u32) { }, node_features: None, room_inference: None, + classifier: None, + activity: None, }; let raw_persons = derive_pose_from_sensing(&update); @@ -9770,6 +9875,619 @@ mod node_positions_by_active_id_tests { } } +/// Build the MediaTek-heuristic `SensingUpdate` from every currently-fresh +/// device in `mediatek_csi_by_device` / `mediatek_heuristic_by_device`, after +/// the caller has already folded the arriving frame's amplitude into its own +/// device's history via `DeviceHistory::observe`. Devices whose last frame is +/// older than `stale_after` are dropped from the vote, mirroring the same +/// per-vendor freshness pattern `effective_source` already uses — a receiver +/// that goes quiet takes only itself out of the room. +/// +/// `classifier` is always `Some(mediatek_heuristic::CLASSIFIER_ID)`: this +/// path never produces the validated ESP32 pipeline's output. +fn mediatek_room_update( + s: &mut AppStateInner, + now: std::time::Instant, + stale_after: std::time::Duration, + tick: u64, + source: String, +) -> SensingUpdate { + let mut verdicts: HashMap = HashMap::new(); + let fresh_ids: Vec = s + .mediatek_csi_by_device + .iter() + .filter(|(_, (_, seen))| now.saturating_duration_since(*seen) <= stale_after) + .map(|(id, _)| id.clone()) + .collect(); + for id in fresh_ids { + if let Some(hist) = s.mediatek_heuristic_by_device.get_mut(&id) { + verdicts.insert(id, hist.verdict(now)); + } + } + let confidences = mediatek_heuristic::apply_agreement(&verdicts); + + let mut ids: Vec<&String> = verdicts.keys().collect(); + ids.sort(); + + let nodes: Vec = ids + .iter() + .map(|&id| { + let v = &verdicts[id]; + let confidence = confidences.get(id).copied().unwrap_or(0.0); + let rssi = s + .mediatek_csi_by_device + .get(id) + .and_then(|(snap, _)| snap.rssi_dbm.first().copied()) + .unwrap_or(0) as f64; + // A device below the confidence threshold abstains from the room + // vote (node_inference: None) rather than asserting "absent" on + // weak evidence — the 2026-09-19 "standing person read as + // absent 0.09" bug. Its diagnostics are still reported so the + // client can show why. + let node_inference = v + .presence + .map(|_| NodeInference::new(v.motion_level.as_str(), confidence, Some(0))); + NodeInfo { + node_id: 0, + device_id: Some(id.clone()), + rssi_dbm: rssi, + position: [0.0, 0.0, 0.0], + amplitude: vec![], + subcarrier_count: 0, + sync: None, + node_inference, + mediatek_diagnostics: Some(MediatekNodeDiagnostics { + presence_state: v.presence_state.as_str().to_string(), + coefficient_of_variation: v.window.coefficient_of_variation, + baseline_deviation: v.window.baseline_deviation, + sample_count: v.window.sample_count, + span_ms: v.window.span_ms, + }), + } + }) + .collect(); + + let node_inferences: Vec = nodes + .iter() + .filter_map(|n| n.node_inference.clone()) + .collect(); + let room = fuse_room(node_inferences.iter(), stale_after.as_millis() as u64); + let classification = classification_from_room(&room); + + let mean_rssi = if nodes.is_empty() { + 0.0 + } else { + nodes.iter().map(|n| n.rssi_dbm).sum::() / nodes.len() as f64 + }; + + // Fast/slow activity index (activity-index-v0, plus an absolute + // measure — a purely relative index sags to mid-scale while a crowd + // simply stays, so a "maxed out" reading would decay within minutes + // even with the room still full). Independent of the presence/absent/unknown heuristic above. + // Room aggregates exclude devices stale past ROOM_STALE_AFTER; the + // per-device listing shows every tracked device's last known verdict + // regardless, same convention as /api/v1/csi/mediatek/devices. + let room_activity_index = mediatek_activity::room_index( + &s.mediatek_activity_by_device, + now, + mediatek_activity::ROOM_STALE_AFTER, + ); + let room_activity_level = mediatek_activity::room_level(room_activity_index); + let room_activity_peak_30s = mediatek_activity::room_peak_30s( + &mut s.room_activity_peak_history, + now, + room_activity_level, + ); + // abs_level's room aggregate is a MAX, not a weighted mean — a crowd + // anywhere in any receiver's path counts. + let room_activity_abs_level = mediatek_activity::room_abs_level( + &s.mediatek_activity_by_device, + now, + mediatek_activity::ROOM_STALE_AFTER, + ); + let room_activity_abs_peak_30s = mediatek_activity::room_peak_30s( + &mut s.room_abs_activity_peak_history, + now, + room_activity_abs_level, + ); + mediatek_activity::maybe_sample_history( + &mut s.activity_history, + &mut s.activity_history_last_sampled, + now, + room_activity_level, + room_activity_abs_level, + &s.mediatek_activity_by_device, + ); + // 2026-09-20: apply the SAME staleness cutoff /api/v1/csi/mediatek/devices + // uses (mediatek_activity::ROOM_STALE_AFTER is literally that constant), + // so activity.devices can never list a receiver /devices has already + // dropped — a downstream consumer caught the two listings disagreeing after a + // unit restart when this filter didn't exist. + let activity_devices: serde_json::Map = s + .mediatek_activity_by_device + .iter() + .filter_map(|(id, tracker)| { + let last_update = tracker.last_update()?; + let age = now.saturating_duration_since(last_update); + if age > mediatek_activity::ROOM_STALE_AFTER { + return None; + } + let v = tracker.last_verdict()?; + Some(( + id.clone(), + serde_json::json!({ + "level": v.level, + "index": v.index, + "fast": v.fast, + "slow": v.slow, + "peak_30s": v.peak_30s, + "frames_2s": v.frames_2s, + "floor": v.floor, + "abs_level": v.abs_level, + "abs_peak_30s": v.abs_peak_30s, + "age_ms": age.as_millis() as u64, + "is_occupancy_estimate": false, + }), + )) + }) + .collect(); + let activity = serde_json::json!({ + "room": { + "level": room_activity_level, + "index": room_activity_index, + "peak_30s": room_activity_peak_30s, + "abs_level": room_activity_abs_level, + "abs_peak_30s": room_activity_abs_peak_30s, + "is_occupancy_estimate": false, + }, + "devices": activity_devices, + }); + + SensingUpdate { + msg_type: "sensing_update".to_string(), + timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0, + source, + tick, + nodes, + // The remaining ESP32-shaped spectral features (motion/breathing band + // power, dominant frequency, spectral power, change points) require + // per-subcarrier phase history this heuristic doesn't have — left at + // 0.0 rather than fabricated. `mean_rssi` is real, averaged over the + // contributing devices' own reported RSSI. + features: FeatureInfo { + mean_rssi, + variance: 0.0, + motion_band_power: 0.0, + breathing_band_power: 0.0, + dominant_freq_hz: 0.0, + change_points: 0, + spectral_power: 0.0, + }, + classification, + // No spatial field: this heuristic has no per-subcarrier or + // multi-antenna geometry to place a heat map from. + signal_field: SignalField { + grid_size: [0, 0, 0], + values: vec![], + }, + vital_signs: None, + calibrated_presence_evidence: None, + enhanced_motion: None, + enhanced_breathing: None, + posture: None, + signal_quality_score: None, + quality_verdict: None, + bssid_count: None, + pose_keypoints: None, + model_status: None, + persons: None, + // Presence/motion, not a person count this heuristic has no basis to + // claim — left absent rather than inventing a number from receiver + // count. + estimated_persons: None, + node_features: None, + room_inference: Some(room), + classifier: Some(mediatek_heuristic::CLASSIFIER_ID.to_string()), + activity: Some(activity), + } +} + +#[cfg(test)] +mod mediatek_room_update_tests { + //! Exercises `mediatek_room_update` directly against `AppStateInner`, + //! without binding a socket — mirrors the style of + //! `mediatek_devices::tests` and `issue_1004_source_plan_tests`. Uses a + //! synthetic, monotonically-advancing `Instant` for both the device's + //! amplitude history AND `mediatek_csi_by_device`'s "last seen" stamp, + //! matching production (both are set from the same `now` per incoming + //! frame) — mixing a synthetic clock with a fresh real `Instant::now()` + //! would silently defeat hysteresis (a future synthetic timestamp reads + //! as "0ms old" against a real `now` that hasn't caught up to it). + use super::*; + use std::time::Duration; + + /// Feeds `samples` 500ms apart starting at `start`, returns the `Instant` + /// of the last sample (use this as the `now` passed to + /// `mediatek_room_update`). + fn seed_device( + s: &mut AppStateInner, + device_id: &str, + start: std::time::Instant, + samples: &[f64], + rssi: i8, + ) -> std::time::Instant { + use wifi_densepose_hardware::mediatek_csi::{ + ChipsetProfile, CsiFlags, CsiPayload, PpduType, ReportKind, + }; + let mut hist = mediatek_heuristic::DeviceHistory::default(); + let mut t = start; + for (i, &sample) in samples.iter().enumerate() { + if i > 0 { + t += Duration::from_millis(500); + } + hist.observe(t, sample, sample); + } + s.mediatek_heuristic_by_device + .insert(device_id.to_string(), hist); + let frame = wifi_densepose_hardware::mediatek_csi::CsiFrame { + report_kind: ReportKind::Csi, + sequence: samples.len() as u32, + timestamp_us: 0, + device_id: u64::from_str_radix(device_id, 16).unwrap_or(0), + chipset: ChipsetProfile::Mt7981Mt7976, + bandwidth_mhz: 80, + center_freq_khz: 5_210_000, + flags: CsiFlags(CsiFlags::CALIBRATED), + tx_count: 2, + rx_count: 1, + ppdu_type: PpduType::HeSu, + subcarrier_count: 4, + noise_floor_dbm: -95, + scale: 1.0, + subcarrier_spacing_hz: 312_500.0, + calibration_id: 1, + payload: CsiPayload::ComplexI16 { + rssi_dbm: vec![rssi], + values: vec![[0, 0]; 8], + }, + }; + let snapshot = mediatek_csi::MediatekCsiSnapshot::from_frame(&frame); + s.mediatek_csi_by_device + .insert(device_id.to_string(), (snapshot, t)); + t + } + + /// 40 samples spaced 500ms = ~19.5s span: comfortably past WINDOW_DURATION, + /// HOLD_DURATION, and CONFIRM_ABSENT_DURATION (all <= 10s) so confidence + /// clears UNKNOWN_CONFIDENCE_THRESHOLD and hysteresis has settled. + const SETTLE_SAMPLES: usize = 40; + + #[test] + fn empty_room_flat_amplitudes_is_absent() { + let mut s = AppStateInner::minimal(); + let flat: Vec = std::iter::repeat_n(100.0, SETTLE_SAMPLES).collect(); + let now = seed_device( + &mut s, + "a9be7e5bb1644d3a", + std::time::Instant::now(), + &flat, + -50, + ); + + let update = mediatek_room_update( + &mut s, + now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + assert!(!update.classification.presence); + assert_eq!(update.classification.motion_level, "absent"); + assert_eq!( + update.classifier.as_deref(), + Some(mediatek_heuristic::CLASSIFIER_ID) + ); + assert_eq!(update.nodes.len(), 1); + assert_eq!( + update.nodes[0].device_id.as_deref(), + Some("a9be7e5bb1644d3a") + ); + assert_eq!( + update.nodes[0] + .mediatek_diagnostics + .as_ref() + .map(|d| d.presence_state.as_str()), + Some("absent") + ); + } + + #[test] + fn injected_variance_is_present_moving() { + let mut s = AppStateInner::minimal(); + let swinging: Vec = (0..SETTLE_SAMPLES) + .map(|i| if i % 2 == 0 { 100.0 } else { 130.0 }) + .collect(); + let now = seed_device( + &mut s, + "a9be7e5bb1644d3a", + std::time::Instant::now(), + &swinging, + -50, + ); + + let update = mediatek_room_update( + &mut s, + now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + assert!(update.classification.presence); + assert_eq!(update.classification.motion_level, "present_moving"); + } + + /// A device below the confidence threshold abstains (node_inference: + /// None) rather than asserting "absent" — the exact 2026-09-19 bug + /// ("standing person read as absent 0.09"). + #[test] + fn low_confidence_device_abstains_instead_of_asserting_absent() { + let mut s = AppStateInner::minimal(); + // A single sample: window barely covers any time, so confidence is + // near zero regardless of amplitude. + let now = seed_device( + &mut s, + "a9be7e5bb1644d3a", + std::time::Instant::now(), + &[1800.0], + -49, + ); + + let update = mediatek_room_update( + &mut s, + now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + assert_eq!(update.nodes.len(), 1); + assert_eq!( + update.nodes[0] + .mediatek_diagnostics + .as_ref() + .map(|d| d.presence_state.as_str()), + Some("unknown") + ); + assert!( + update.nodes[0].node_inference.is_none(), + "an unknown device must abstain, not vote" + ); + // No fresh contributor -> room is honestly unavailable-derived absent, not a guess. + assert_eq!(update.classification.motion_level, "absent"); + assert_eq!(update.classification.confidence, 0.0); + } + + /// A single transient dip must not flip the room to absent — the core + /// 2026-09-19 field bug (walk between routers flickered present <-> absent). + #[test] + fn a_single_transient_dip_does_not_flip_the_room_to_absent() { + let mut s = AppStateInner::minimal(); + let swinging: Vec = (0..SETTLE_SAMPLES) + .map(|i| if i % 2 == 0 { 100.0 } else { 130.0 }) + .collect(); + let mut samples = swinging.clone(); + samples.push(115.0); // one flat-ish sample right after — a real transient fade + let now = seed_device( + &mut s, + "a9be7e5bb1644d3a", + std::time::Instant::now(), + &samples, + -50, + ); + + let update = mediatek_room_update( + &mut s, + now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + assert!( + update.classification.presence, + "a single dip must not flip the room to absent" + ); + } + + #[test] + fn two_devices_disagreeing_lowers_room_confidence_vs_single_device() { + let swinging: Vec = (0..SETTLE_SAMPLES) + .map(|i| if i % 2 == 0 { 100.0 } else { 130.0 }) + .collect(); + let flat: Vec = std::iter::repeat_n(100.0, SETTLE_SAMPLES).collect(); + + let mut solo = AppStateInner::minimal(); + let solo_now = seed_device( + &mut solo, + "a9be7e5bb1644d3a", + std::time::Instant::now(), + &swinging, + -50, + ); + let solo_update = mediatek_room_update( + &mut solo, + solo_now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + + let mut pair = AppStateInner::minimal(); + let start = std::time::Instant::now(); + seed_device(&mut pair, "a9be7e5bb1644d3a", start, &swinging, -50); + let pair_now = seed_device(&mut pair, "a9be7d5bb1644b87", start, &flat, -55); + let pair_update = mediatek_room_update( + &mut pair, + pair_now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + + assert_eq!( + pair_update.nodes.len(), + 2, + "both devices must vote, not overwrite each other" + ); + let moving_node = pair_update + .nodes + .iter() + .find(|n| n.device_id.as_deref() == Some("a9be7e5bb1644d3a")) + .expect("moving device present"); + let moving_confidence = moving_node.node_inference.as_ref().unwrap().confidence; + let solo_confidence = solo_update.classification.confidence; + assert!( + moving_confidence < solo_confidence, + "disagreement should lower confidence: solo={solo_confidence} paired={moving_confidence}" + ); + } + + /// A bursty two-device timeline (dump-loop-style gaps up to a few + /// seconds between arrivals per device), interleaved. Both devices must + /// still vote correctly and the room must not flicker on the gaps alone. + #[test] + fn bursty_interleaved_two_device_timeline_stays_stable() { + let mut s = AppStateInner::minimal(); + let start = std::time::Instant::now(); + let mut t = start; + let mut hist_a = mediatek_heuristic::DeviceHistory::default(); + let mut hist_b = mediatek_heuristic::DeviceHistory::default(); + + // Both devices arrive at the SAME irregular times (0.3-4s gaps, + // mirroring the real dump-loop cadence) so this isolates "does a + // bursty arrival pattern break the time-based window" from "did one + // device merely go stale relative to the other" (a separate, + // legitimate concern already covered by ESP32_OFFLINE_TIMEOUT). + // A moves (swinging amplitude), B is steady (flat). + let gaps_ms = [300u64, 4000, 500, 3500, 300, 4000, 500, 3000, 300, 4000]; + for (i, &gap) in gaps_ms.iter().enumerate() { + t += Duration::from_millis(gap); + let sample_a = if i % 2 == 0 { 100.0 } else { 130.0 }; + hist_a.observe(t, sample_a, sample_a + 2.0); + hist_b.observe(t, 1600.0, 1650.0); + } + let (t_a, t_b) = (t, t); + + let now = t; + s.mediatek_heuristic_by_device + .insert("dev-a".to_string(), hist_a); + s.mediatek_heuristic_by_device + .insert("dev-b".to_string(), hist_b); + // Minimal snapshots just to satisfy the freshness join in + // mediatek_room_update; RSSI/content beyond device_id is unused here. + for (id, seen) in [("dev-a", t_a), ("dev-b", t_b)] { + use wifi_densepose_hardware::mediatek_csi::{ + ChipsetProfile, CsiFlags, CsiPayload, PpduType, ReportKind, + }; + let frame = wifi_densepose_hardware::mediatek_csi::CsiFrame { + report_kind: ReportKind::Csi, + sequence: 0, + timestamp_us: 0, + device_id: 0, + chipset: ChipsetProfile::Mt7981Mt7976, + bandwidth_mhz: 80, + center_freq_khz: 5_210_000, + flags: CsiFlags(CsiFlags::CALIBRATED), + tx_count: 2, + rx_count: 1, + ppdu_type: PpduType::HeSu, + subcarrier_count: 4, + noise_floor_dbm: -95, + scale: 1.0, + subcarrier_spacing_hz: 312_500.0, + calibration_id: 1, + payload: CsiPayload::ComplexI16 { + rssi_dbm: vec![-50], + values: vec![[0, 0]; 8], + }, + }; + let snapshot = mediatek_csi::MediatekCsiSnapshot::from_frame(&frame); + s.mediatek_csi_by_device + .insert(id.to_string(), (snapshot, seen)); + } + + let update = mediatek_room_update( + &mut s, + now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + assert_eq!( + update.nodes.len(), + 2, + "both bursty devices must still be represented" + ); + let dev_a = update + .nodes + .iter() + .find(|n| n.device_id.as_deref() == Some("dev-a")) + .unwrap(); + // dev-a's time-based window must have accumulated more than the + // handful of samples closest to `now` — a count-based window with + // this arrival pattern would badly underfill given the gaps. + let diag = dev_a.mediatek_diagnostics.as_ref().unwrap(); + assert!( + diag.sample_count >= 3, + "sample_count was {}", + diag.sample_count + ); + } + + /// `/devices` and `activity.devices` must + /// never name different receiver sets. A device stale past + /// ROOM_STALE_AFTER — the same cutoff `/devices` uses — must be dropped + /// from `activity.devices` (and so excluded from `room.level`/ + /// `room.abs_level` too), while a fresh device reports `age_ms`. + #[test] + fn stale_device_is_excluded_from_activity_but_fresh_one_reports_age_ms() { + let mut s = AppStateInner::minimal(); + let stale_seen = std::time::Instant::now(); + let now = + stale_seen + mediatek_activity::ROOM_STALE_AFTER + std::time::Duration::from_secs(1); + + let mut fresh = mediatek_activity::ActivityTracker::default(); + fresh.observe(now, 0.5, 10); + s.mediatek_activity_by_device + .insert("dev-fresh".to_string(), fresh); + + let mut stale = mediatek_activity::ActivityTracker::default(); + stale.observe(stale_seen, 0.5, 10); + s.mediatek_activity_by_device + .insert("dev-stale".to_string(), stale); + + let update = mediatek_room_update( + &mut s, + now, + ESP32_OFFLINE_TIMEOUT, + 1, + "mediatek:physical-unvalidated".to_string(), + ); + let activity = update.activity.expect("activity block present"); + let devices = activity["devices"].as_object().expect("devices object"); + assert!( + devices.contains_key("dev-fresh"), + "the fresh device must be listed" + ); + assert!( + !devices.contains_key("dev-stale"), + "a device stale past ROOM_STALE_AFTER must be dropped, matching /devices" + ); + assert!( + devices["dev-fresh"]["age_ms"].as_u64().is_some(), + "a listed device must report age_ms" + ); + } +} + // ── UDP receiver task ──────────────────────────────────────────────────────── async fn udp_receiver_task( @@ -9853,6 +10571,8 @@ async fn udp_receiver_task( let now = std::time::Instant::now(); let device_id = snapshot.device_id.clone(); let source_label = snapshot.source.to_string(); + let mean_amplitude = snapshot.mean_amplitude; + let peak_amplitude = snapshot.peak_amplitude; let mut s = state.write().await; s.source = source_label.clone(); s.last_mediatek_frame = Some(now); @@ -9862,14 +10582,81 @@ async fn udp_receiver_task( if let Some(json) = json { let _ = s.tx.send(json); } // Per-device channel inspection: retain full - // per-subcarrier amplitude/phase from this - // frame's raw I/Q, bounded to the last - // `csi_ring_capacity` frames per device. + // per-subcarrier + // amplitude/phase from this frame's raw I/Q, + // bounded to the last `csi_ring_capacity` frames + // per device. Independent of the heuristic + // classifier below — this runs even on a frame + // whose mean_amplitude is somehow absent. let ring_capacity = s.csi_ring_capacity; s.mediatek_csi_ring_by_device .entry(device_id.clone()) .or_insert_with(|| mediatek_csi_ring::DeviceRing::new(ring_capacity)) .push(&frame); + + // Tier-2 heuristic classifier: fold this frame + // into its device's + // rolling amplitude window, then rebuild the room + // picture from every currently-fresh MediaTek + // device so latest_update reflects real CSI + // instead of simulated_data_task's fabricated + // number. Skipped when the snapshot carries no + // mean_amplitude (defensive; from_frame always + // sets it for a valid ComplexI16/F32 payload) — + // silently falling back would mean no update at + // all this tick, which is honest, not a bug. + if let Some(mean_amp) = mean_amplitude { + let peak_amp = peak_amplitude.unwrap_or(mean_amp); + + // Relative fast/slow activity index + // (2026-09-20, activity-index-v0): the + // ring already has this frame (pushed + // above), so compute "fast" from it now + // and fold it into this device's EMA + // baseline. Independent of the + // presence/absent heuristic below. + if let Some(ring) = s.mediatek_csi_ring_by_device.get(&device_id) { + let (fast, frames_2s) = ring.fast_activity(mediatek_activity::FAST_WINDOW); + // Seed a brand-new tracker's floor from + // --activity-state (2026-09-20, + // floor-bootstrap fix) so a restart + // doesn't re-run the ~30 s bootstrap + // window and re-saturate abs_level in + // the meantime. Read out of + // activity_floor_seed before touching + // mediatek_activity_by_device to keep + // the borrows disjoint. + let floor_seed = s.activity_floor_seed.get(&device_id).copied(); + s.mediatek_activity_by_device + .entry(device_id.clone()) + .or_insert_with(|| { + let mut tracker = mediatek_activity::ActivityTracker::default(); + if let Some(seed) = floor_seed { + tracker.seed_floor(seed); + } + tracker + }) + .observe(now, fast, frames_2s); + } + + s.mediatek_heuristic_by_device + .entry(device_id) + .or_default() + .observe(now, mean_amp as f64, peak_amp as f64); + s.tick += 1; + let tick = s.tick; + let update = mediatek_room_update( + &mut s, + now, + ESP32_OFFLINE_TIMEOUT, + tick, + source_label, + ); + if let Ok(json) = serde_json::to_string(&update) { + let _ = s.tx.send(json); + } + s.latest_update = Some(update); + } } Ok((_, consumed)) => warn!("MediaTek CSI datagram from {src} has trailing bytes: consumed={consumed} received={len}"), Err(error) => warn!("Rejected MediaTek CSI datagram from {src}: {error}"), @@ -10015,6 +10802,7 @@ async fn udp_receiver_task( .filter(|(_, n)| node_is_fresh(n, now)) .map(|(&id, n)| NodeInfo { node_id: id, + device_id: None, rssi_dbm: n.rssi_history.back().copied().unwrap_or(0.0), position: resolved_positions .get(&id) @@ -10027,6 +10815,7 @@ async fn udp_receiver_task( sync: n.sync_snapshot(), // ADR-297 — each node carries its own inference. node_inference: Some(node_inference_for(n, now)), + mediatek_diagnostics: None, }) .collect(); @@ -10153,6 +10942,8 @@ async fn udp_receiver_task( // tripping back to the server. node_features: build_node_features(&s.node_states, now), room_inference: Some(room_inference), + classifier: None, + activity: None, }; let raw_persons = derive_pose_from_sensing(&update); @@ -10537,6 +11328,7 @@ async fn udp_receiver_task( .filter(|(_, n)| node_is_fresh(n, now)) .map(|(&id, n)| NodeInfo { node_id: id, + device_id: None, rssi_dbm: n.rssi_history.back().copied().unwrap_or(0.0), position: resolved_positions .get(&id) @@ -10559,6 +11351,7 @@ async fn udp_receiver_task( sync: n.sync_snapshot(), // ADR-297 — each node carries its own inference. node_inference: Some(node_inference_for(n, now)), + mediatek_diagnostics: None, }) .collect(); @@ -10638,6 +11431,8 @@ async fn udp_receiver_task( // tripping back to the server. node_features: build_node_features(&s.node_states, now), room_inference: Some(room_inference), + classifier: None, + activity: None, }; let raw_persons = derive_pose_from_sensing(&update); @@ -10855,12 +11650,14 @@ async fn simulated_data_task(state: SharedState, tick_ms: u64) { tick, nodes: vec![NodeInfo { node_id: 1, + device_id: None, rssi_dbm: features.mean_rssi, position: [2.0, 0.0, 1.5], amplitude: frame_amplitudes, subcarrier_count: frame_n_sub as usize, sync: None, // simulated frame path — no mesh peer node_inference: None, // simulated frame; source is synthetic + mediatek_diagnostics: None, }], features: features.clone(), classification, @@ -10899,6 +11696,8 @@ async fn simulated_data_task(state: SharedState, tick_ms: u64) { }, node_features: None, room_inference: None, + classifier: None, + activity: None, }; // Populate persons from the sensing update (Kalman-smoothed via tracker). @@ -11027,7 +11826,17 @@ fn vitals_snapshots_from_sensing_json( Some(arr) if !arr.is_empty() => arr .iter() .map(|node| { - let n = node["node_id"].as_u64().unwrap_or(0); + // Prefer a string `device_id` (MediaTek MTC1 and other + // vendor receivers whose real identity is a 64-bit id, not a + // small mesh index) over the numeric `node_id`. Collapsing a + // device_id into `node_id: u8` would risk two receivers + // colliding onto the same topic; ESP32/simulated nodes never + // set `device_id`, so they fall through to the pre-existing + // numeric-suffix topic naming unchanged. + let node_key = node["device_id"] + .as_str() + .map(str::to_string) + .unwrap_or_else(|| node["node_id"].as_u64().unwrap_or(0).to_string()); // 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 @@ -11043,7 +11852,7 @@ fn vitals_snapshots_from_sensing_json( 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}"), + format!("{base_id}-node{node_key}"), presence, motion, conf, @@ -12191,6 +13000,14 @@ async fn main() { // WDP_TDM_SLOTS/WDP_GUARD_INTERVAL_US-derived guard (#1049/#1057). let mut engine_bridge_multistatic_cfg: Option = None; let mut node_positions_config: HashMap = HashMap::new(); + // Loaded once at startup from --activity-state (if given) so a restart + // doesn't reset each device's learned quiet-room activity floor; see + // AppStateInner::activity_floor_seed and mediatek_activity::load_activity_state. + let activity_floor_seed: HashMap = args + .activity_state + .as_ref() + .map(|p| mediatek_activity::load_activity_state(p)) + .unwrap_or_default(); let state: SharedState = Arc::new(RwLock::new(AppStateInner { latest_update: None, rssi_history: VecDeque::new(), @@ -12203,8 +13020,15 @@ async fn main() { latest_mediatek_csi: None, last_mediatek_frame: None, mediatek_csi_by_device: HashMap::new(), + mediatek_heuristic_by_device: HashMap::new(), mediatek_csi_ring_by_device: HashMap::new(), csi_ring_capacity: args.csi_ring, + mediatek_activity_by_device: HashMap::new(), + activity_floor_seed, + room_activity_peak_history: std::collections::VecDeque::new(), + room_abs_activity_peak_history: std::collections::VecDeque::new(), + activity_history: std::collections::VecDeque::new(), + activity_history_last_sampled: None, latest_qualcomm_csi: None, last_qualcomm_frame: None, latest_realtek_csi: None, @@ -12423,6 +13247,17 @@ async fn main() { if plan.run_simulator { tokio::spawn(simulated_data_task(state.clone(), args.tick_ms)); } + // Floor persistence (2026-09-20, floor-bootstrap fix): only when + // --activity-state was given. Writes every + // mediatek_activity::ACTIVITY_STATE_PERSIST_INTERVAL so a restart's + // freshly loaded activity_floor_seed reflects the last run's learned + // quiet-room baseline, not an empty map. + if let Some(activity_state_path) = args.activity_state.clone() { + tokio::spawn(mediatek_activity::activity_state_persist_task( + state.clone(), + activity_state_path, + )); + } // ADR-166: Parse bind address once, use for all listeners let bind_ip: std::net::IpAddr = args @@ -12570,6 +13405,10 @@ async fn main() { "/api/v1/csi/mediatek/devices/:device_id/summary", get(mediatek_csi_ring::mediatek_csi_summary), ) + .route( + "/api/v1/csi/mediatek/activity", + get(mediatek_activity::mediatek_activity_history), + ) .route("/api/v1/csi/qualcomm/latest", get(latest_qualcomm_csi)) .route("/api/v1/csi/realtek/latest", get(latest_realtek_csi)) .route("/api/v1/rf/vendors", get(vendor_descriptors)) @@ -12893,12 +13732,14 @@ mod node_sync_snapshot_serialization_tests { fn sample_node(sync: Option) -> NodeInfo { NodeInfo { node_id: 9, + device_id: None, rssi_dbm: -38.0, position: [2.0, 0.0, 1.5], amplitude: vec![], subcarrier_count: 0, sync, node_inference: None, + mediatek_diagnostics: None, } } @@ -13472,6 +14313,36 @@ mod mqtt_bridge_tests { assert_eq!(n2.presence_score, 0.0); } + /// A `device_id` string (MediaTek MTC1 and other vendor receivers whose + /// real identity isn't a small mesh index) takes the MQTT topic slot + /// instead of `node_id`, and two devices never collide onto one topic. + #[test] + fn per_node_device_id_string_takes_priority_over_node_id() { + let v = json!({ + "timestamp": 1.0, + "classification": { "presence": true, "motion_level": "present_moving", "confidence": 0.7 }, + "vital_signs": {}, + "nodes": [ + { "node_id": 0, "device_id": "a9be7e5bb1644d3a", "rssi_dbm": -42.0, + "node_inference": { "classification": "present_moving", "confidence": 0.8 } }, + { "node_id": 0, "device_id": "a9be7d5bb1644b87", "rssi_dbm": -55.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 device, despite identical node_id: 0" + ); + assert!(snaps + .iter() + .any(|s| s.node_id == "ruview-nodea9be7e5bb1644d3a")); + assert!(snaps + .iter() + .any(|s| s.node_id == "ruview-nodea9be7d5bb1644b87")); + } + /// A node that omits a classification field defers to the room aggregate /// rather than silently reading false/0. #[test] @@ -13673,6 +14544,8 @@ mod observatory_persons_field_position_tests { estimated_persons: Some(1), node_features: None, room_inference: None, + classifier: None, + activity: None, } } diff --git a/v2/crates/wifi-densepose-sensing-server/src/mediatek_activity.rs b/v2/crates/wifi-densepose-sensing-server/src/mediatek_activity.rs new file mode 100644 index 000000000..a1453e2b0 --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/mediatek_activity.rs @@ -0,0 +1,1024 @@ +//! "How much is happening right now, relative to how much usually happens +//! here" — a fast/slow relative-activity index per MediaTek device and for +//! the room as a whole. The use case is an at-a-glance activity dial: it +//! should drop when nobody is moving and rise when a room full of people all +//! stand up at once. +//! +//! This is deliberately NOT an absolute presence/motion classifier — see +//! `mediatek_heuristic` for that (presence/absent/unknown against fixed +//! thresholds). This module answers a different question: is the channel +//! fluctuating more, right now, than it *typically* does at this location? +//! +//! # Relative to its own baseline, on purpose +//! +//! The index compares a fast signal (`fast`, the last [`FAST_WINDOW`]) to a +//! slow self-baseline (`slow`, an EMA with [`SLOW_TIME_CONSTANT`]). A room +//! that is ALWAYS busy trends back toward the middle of the scale — its own +//! "busy" becomes the new normal — and that is intended, not a bug. A step +//! change (everyone standing up at once) shows as a spike against the +//! baseline; sustained elevated activity decays back toward index 0 as the +//! baseline catches up. Label: [`CLASSIFIER_ID`] = `activity-index-v0`. +//! +//! # Floors, and why `fast` and `slow` share one +//! +//! `fast` is floored to [`SLOW_FLOOR`] before it ever reaches the EMA or the +//! ratio — the SAME floor `slow` itself is clamped to. A perfectly flat +//! channel (fast == 0 forever) then converges `slow` to exactly that same +//! floor too, giving `index = log2(floor / floor) = 0` — "flat means index +//! 0", not a large negative number from an arbitrary epsilon mismatch +//! between two differently-floored quantities. + +use crate::SharedState; +use axum::extract::{Query, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Json}; +use serde::Deserialize; +use std::collections::{BTreeMap, HashMap, VecDeque}; +use std::path::Path; +use std::time::{Duration, Instant}; +use tracing::warn; + +pub(crate) const CLASSIFIER_ID: &str = "activity-index-v0"; +/// "Fast" signal window: how much the channel is fluctuating right now. +pub(crate) const FAST_WINDOW: Duration = Duration::from_secs(2); +/// "Slow" baseline time constant. Applied as a genuine time-constant EMA +/// (`alpha = 1 - exp(-dt/tau)`, not a fixed per-sample alpha), so irregular +/// (bursty) arrival timing doesn't change how quickly the baseline actually +/// adapts in wall-clock time. +pub(crate) const SLOW_TIME_CONSTANT: Duration = Duration::from_secs(300); +/// Shared floor for both `fast` and `slow` — see the module doc's "Floors" +/// section for why this must be the SAME constant for both. +pub(crate) const SLOW_FLOOR: f64 = 1e-4; +/// Peak-hold window for both per-device and room `peak_30s`. +pub(crate) const PEAK_HOLD_WINDOW: Duration = Duration::from_secs(30); +pub(crate) const INDEX_MIN: f64 = -2.0; +pub(crate) const INDEX_MAX: f64 = 3.0; +/// How much history `/api/v1/csi/mediatek/activity` can serve. +pub(crate) const HISTORY_RETENTION: Duration = Duration::from_secs(300); +/// Retained-history decimation — a strip chart doesn't need every ingest +/// tick, just ~1 Hz. +pub(crate) const HISTORY_SAMPLE_INTERVAL: Duration = Duration::from_secs(1); +/// A device counts toward the room aggregate — and appears in +/// `activity.devices` at all — only while its last update is within this +/// of `now`. Deliberately the SAME constant `/api/v1/csi/mediatek/devices` +/// uses (`mediatek_devices::MEDIATEK_DEVICE_STALE`), not an independent +/// value: a downstream consumer observed `/devices` and `activity.devices` +/// naming two different receiver sets after one unit restarted, because this +/// used to be its own 5s cutoff. The two listings must never disagree about +/// which receivers currently exist. +pub(crate) const ROOM_STALE_AFTER: Duration = crate::mediatek_devices::MEDIATEK_DEVICE_STALE; + +/// Hard minimum for `floor` (the quiet-room baseline) — same value as +/// `SLOW_FLOOR`, for the same reason: a channel that's never shown any +/// measurable fluctuation shouldn't produce an undefined or exploding +/// `abs_level` either. +pub(crate) const FLOOR_HARD_MIN: f64 = SLOW_FLOOR; +/// How quickly `floor` is pulled DOWN toward a new, lower `fast` reading. +/// This approximates a running low-percentile of `fast` over the last tens +/// of minutes without the cost of a real quantile sketch: one low sample +/// nudges the floor a little, a SUSTAINED quiet stretch pulls it most of +/// the way down within a few multiples of this constant. Deliberately +/// asymmetric with `SLOW_TIME_CONSTANT` — the floor is meant to track "how +/// quiet does this room get", which should move slower to reach than the +/// change-detector's baseline, but doesn't need the full 30-minute window +/// a literal percentile would use to be a reasonable proxy for it. +pub(crate) const FLOOR_DOWN_TIME_CONSTANT: Duration = Duration::from_secs(600); +/// Upper bound on how fast `floor` may drift UP on its own (compounding, +/// applied every observation regardless of the downward pull above), so a +/// one-off quiet dip doesn't permanently anchor the floor after the room's +/// true quiet baseline has genuinely risen (furniture moved, sensor +/// relocated, persistent new noise source, etc.) — it re-learns rather +/// than getting stuck. +pub(crate) const FLOOR_UPWARD_CREEP_PER_MINUTE: f64 = 0.01; +/// How long a brand-new device (no persisted floor — see +/// `ActivityTracker::seed_floor`) collects `fast` readings before +/// committing an initial `floor`, instead of seeding it from a single +/// reading. 2026-09-20: without this, the very first observation (often +/// near-zero — a ring with only 1-2 frames has no meaningful variance yet) +/// pinned `floor` at `FLOOR_HARD_MIN`, and the ≤1%/min creep meant it would +/// stay saturated at `abs_level` 1.0 for hours regardless of what the room +/// actually did. +pub(crate) const FLOOR_INIT_WINDOW: Duration = Duration::from_secs(30); +/// Percentile (nearest-rank) of the init-window samples used as the +/// committed initial floor — low enough to represent "how quiet this +/// window got", not its typical level, but not the single minimum sample +/// (more robust to one anomalously-low reading). +pub(crate) const FLOOR_INIT_PERCENTILE: f64 = 0.20; +/// How often `--activity-state` is rewritten to disk. +pub(crate) const ACTIVITY_STATE_PERSIST_INTERVAL: Duration = Duration::from_secs(60); + +fn index_to_level(index: f64) -> f64 { + ((index - INDEX_MIN) / (INDEX_MAX - INDEX_MIN)).clamp(0.0, 1.0) +} + +/// Nearest-rank percentile of `samples` (sorted in place). `p` is a +/// fraction in `[0, 1]`; `p=0.0` is the min, `p=1.0` is the max. Returns +/// `0.0` for an empty slice — callers `.max(FLOOR_HARD_MIN)` the result, so +/// this never surfaces as a usable floor on its own. +fn percentile(samples: &mut [f64], p: f64) -> f64 { + if samples.is_empty() { + return 0.0; + } + samples.sort_by(f64::total_cmp); + let idx = ((samples.len() - 1) as f64 * p.clamp(0.0, 1.0)).round() as usize; + samples[idx.min(samples.len() - 1)] +} + +/// One device's activity verdict at the moment it was last observed. +/// Carries TWO independent measures, because a purely-relative index sags +/// back to mid-scale while a crowd simply stays, so a "maxed out" reading +/// would decay within minutes even with the room still full: +/// +/// - `index`/`level`: the CHANGE detector (fast vs. the 300s `slow` +/// baseline) — spikes on a step change like a stand-up, decays back +/// toward 0 as the new activity level becomes the norm. Good for "did +/// something just happen", bad for "is the room currently full". +/// - `floor`/`abs_level`: an ABSOLUTE measure against a slow-moving +/// QUIET-ROOM baseline (`floor`) instead of a fast-moving one — a full +/// room stays high for as long as it stays full; an empty room reads +/// near 0; `floor` itself only re-learns "how quiet is quiet" over tens +/// of minutes, so occupancy doesn't have to decay just because it's +/// sustained. See `is_occupancy_estimate` at the API layer: this is +/// still a channel-fluctuation proxy, not a validated occupancy count. +#[derive(Debug, Clone, Copy, PartialEq)] +pub(crate) struct DeviceVerdict { + pub(crate) index: f64, + pub(crate) level: f64, + pub(crate) fast: f64, + pub(crate) slow: f64, + pub(crate) peak_30s: f64, + pub(crate) frames_2s: usize, + pub(crate) floor: f64, + pub(crate) abs_level: f64, + pub(crate) abs_peak_30s: f64, +} + +/// Per-device EMA/peak-hold state, persisted across ingest ticks. Only +/// mutated when a NEW frame arrives for this specific device — other +/// devices' trackers are read-only from the room aggregate's point of view. +#[derive(Debug, Clone, Default)] +pub(crate) struct ActivityTracker { + slow: Option, + /// `None` until either `seed_floor` was called (persisted-state restore) + /// or the `FLOOR_INIT_WINDOW` bootstrap below committed a value. + floor: Option, + /// `fast` readings collected since `floor_init_start`, while `floor` is + /// still `None` and no seed was provided. Cleared once the window + /// commits a floor. + floor_init_samples: Vec, + floor_init_start: Option, + last_update: Option, + level_history: VecDeque<(Instant, f64)>, + abs_level_history: VecDeque<(Instant, f64)>, + last_verdict: Option, +} + +impl ActivityTracker { + /// Restore a previously-learned floor (from `--activity-state`) for a + /// device seen in an earlier run, so a restart doesn't re-saturate + /// `abs_level` while the bootstrap window re-collects — an empty-room + /// floor learned in one run should survive to the next. Skips the + /// `FLOOR_INIT_WINDOW` bootstrap entirely: a persisted floor is already + /// better-informed than a fresh 30s sample. Only meaningful before the + /// first `observe()` call; a no-op with a warning-free overwrite if + /// called after (harmless, just wasted — callers only do this once, at + /// tracker creation). + pub(crate) fn seed_floor(&mut self, floor: f64) { + self.floor = Some(floor.max(FLOOR_HARD_MIN)); + self.floor_init_samples.clear(); + self.floor_init_start = None; + } + + /// The tracker's current floor, for periodic persistence + /// (`--activity-state`). `None` while still in the init-window + /// bootstrap (nothing durable to save yet). + pub(crate) fn floor(&self) -> Option { + self.floor + } + + /// Fold in a new `fast` reading observed at `now` (already computed + /// from the device's CSI ring via `DeviceRing::fast_activity`), update + /// both the relative (slow EMA) and absolute (floor) baselines plus + /// their 30s peak-holds, and return (and cache) the verdict. + pub(crate) fn observe(&mut self, now: Instant, fast: f64, frames_2s: usize) -> DeviceVerdict { + let fast = fast.max(SLOW_FLOOR); + let dt = self + .last_update + .map(|t| now.saturating_duration_since(t).as_secs_f64()) + .unwrap_or(0.0); + + // Relative: fast vs. a 300s EMA baseline. A change detector. + let tau = SLOW_TIME_CONSTANT.as_secs_f64(); + self.slow = Some(match self.slow { + None => fast, + Some(s) => { + let alpha = if tau > 0.0 { + 1.0 - (-dt / tau).exp() + } else { + 1.0 + }; + (s + alpha * (fast - s)).max(SLOW_FLOOR) + } + }); + let slow = self.slow.unwrap(); + let index = (fast / slow).log2().clamp(INDEX_MIN, INDEX_MAX); + let level = index_to_level(index); + + // Absolute: fast vs. a slow-to-rise, quick-to-fall quiet-room + // floor. An occupancy proxy, not a change detector. Bootstrapped + // from a FLOOR_INIT_WINDOW of samples (a percentile, not the first + // single reading) unless `seed_floor` already restored one. + let floor = match self.floor { + Some(f) => { + let floor_down_tau = FLOOR_DOWN_TIME_CONSTANT.as_secs_f64(); + let pulled_down = if fast < f { + let alpha = if floor_down_tau > 0.0 { + 1.0 - (-dt / floor_down_tau).exp() + } else { + 1.0 + }; + f + alpha * (fast - f) + } else { + f + }; + let dt_minutes = dt / 60.0; + let creep = (1.0 + FLOOR_UPWARD_CREEP_PER_MINUTE).powf(dt_minutes); + let updated = (pulled_down * creep).max(FLOOR_HARD_MIN); + self.floor = Some(updated); + updated + } + None => { + let start = *self.floor_init_start.get_or_insert(now); + self.floor_init_samples.push(fast); + if now.saturating_duration_since(start) >= FLOOR_INIT_WINDOW { + let committed = percentile(&mut self.floor_init_samples, FLOOR_INIT_PERCENTILE) + .max(FLOOR_HARD_MIN); + self.floor = Some(committed); + self.floor_init_samples.clear(); + committed + } else { + // Provisional: the lowest reading seen so far in the + // still-open window, so abs_level is defined from the + // first sample rather than reading as a fixed 0 or 1 + // for up to 30s. Not stored in `self.floor` — the + // window isn't committed yet, so a later, genuinely + // lower sample can still pull this down before commit. + self.floor_init_samples + .iter() + .copied() + .fold(f64::INFINITY, f64::min) + .max(FLOOR_HARD_MIN) + } + } + }; + let abs_level = ((fast / floor).log2() / 5.0).clamp(0.0, 1.0); + + self.last_update = Some(now); + + self.level_history.push_back((now, level)); + while self + .level_history + .front() + .is_some_and(|(t, _)| now.saturating_duration_since(*t) > PEAK_HOLD_WINDOW) + { + self.level_history.pop_front(); + } + let peak_30s = self + .level_history + .iter() + .map(|(_, l)| *l) + .fold(0.0, f64::max); + + self.abs_level_history.push_back((now, abs_level)); + while self + .abs_level_history + .front() + .is_some_and(|(t, _)| now.saturating_duration_since(*t) > PEAK_HOLD_WINDOW) + { + self.abs_level_history.pop_front(); + } + let abs_peak_30s = self + .abs_level_history + .iter() + .map(|(_, l)| *l) + .fold(0.0, f64::max); + + let verdict = DeviceVerdict { + index, + level, + fast, + slow, + peak_30s, + frames_2s, + floor, + abs_level, + abs_peak_30s, + }; + self.last_verdict = Some(verdict); + verdict + } + + pub(crate) fn last_verdict(&self) -> Option { + self.last_verdict + } + + pub(crate) fn last_update(&self) -> Option { + self.last_update + } +} + +/// Room activity index: the `frames_2s`-weighted mean of every device +/// that's updated within `stale_after` of `now`. `INDEX_MIN` (no activity) +/// when no device is fresh. A running `peak_30s` of this value is tracked +/// separately by the caller (`room_peak_30s`) — "weighted mean... plus a +/// max" is the peak-hold on the aggregate, not a per-tick blend with the +/// single highest device (a weighted mean can never exceed its inputs' +/// max, so blending those two per-tick would be a no-op). +pub(crate) fn room_index( + trackers: &HashMap, + now: Instant, + stale_after: Duration, +) -> f64 { + let fresh: Vec = trackers + .values() + .filter(|t| { + t.last_update() + .is_some_and(|u| now.saturating_duration_since(u) <= stale_after) + }) + .filter_map(|t| t.last_verdict()) + .collect(); + if fresh.is_empty() { + return INDEX_MIN; + } + let total_weight: f64 = fresh.iter().map(|v| v.frames_2s as f64).sum(); + let mean = if total_weight > 0.0 { + fresh + .iter() + .map(|v| v.index * v.frames_2s as f64) + .sum::() + / total_weight + } else { + // No device has any frames within its own last 2s to weight by + // (e.g. everyone just barely reconnected) — fall back to a plain + // mean across the fresh set rather than reporting no activity. + fresh.iter().map(|v| v.index).sum::() / fresh.len() as f64 + }; + mean.clamp(INDEX_MIN, INDEX_MAX) +} + +pub(crate) fn room_level(index: f64) -> f64 { + index_to_level(index) +} + +/// Room `abs_level`: the MAX over every fresh device's `abs_level` — unlike +/// the relative `room_index`'s weighted mean, this is deliberately a max: +/// a crowd anywhere in any receiver's path counts. +/// `0.0` when no device is fresh. +pub(crate) fn room_abs_level( + trackers: &HashMap, + now: Instant, + stale_after: Duration, +) -> f64 { + trackers + .values() + .filter(|t| { + t.last_update() + .is_some_and(|u| now.saturating_duration_since(u) <= stale_after) + }) + .filter_map(|t| t.last_verdict()) + .map(|v| v.abs_level) + .fold(0.0, f64::max) +} + +/// Advances a 30s peak-hold and returns the current peak. Used for both the +/// room's relative `peak_30s` (`room_activity_peak_history`) and its +/// absolute `abs_peak_30s` (`room_abs_activity_peak_history`) on +/// `AppStateInner` — same pattern as `ActivityTracker`'s per-device ones. +pub(crate) fn room_peak_30s( + peak_history: &mut VecDeque<(Instant, f64)>, + now: Instant, + value: f64, +) -> f64 { + peak_history.push_back((now, value)); + while peak_history + .front() + .is_some_and(|(t, _)| now.saturating_duration_since(*t) > PEAK_HOLD_WINDOW) + { + peak_history.pop_front(); + } + peak_history.iter().map(|(_, l)| *l).fold(0.0, f64::max) +} + +/// One decimated (~1 Hz) history sample for the `/activity` strip-chart +/// route. Carries both measures per the 2026-09-20 amendment: `level` +/// (relative, spikes-and-decays) and `abs_level` (absolute, stays high +/// while a crowd stays) — see `DeviceVerdict`'s doc comment for what each +/// answers. +#[derive(Debug, Clone)] +pub(crate) struct ActivitySample { + pub(crate) at: Instant, + pub(crate) room_level: f64, + pub(crate) room_abs_level: f64, + pub(crate) device_levels: BTreeMap, + pub(crate) device_abs_levels: BTreeMap, +} + +/// Appends a new history sample if at least `HISTORY_SAMPLE_INTERVAL` has +/// passed since the last one (decimation), then prunes anything older than +/// `HISTORY_RETENTION`. A no-op call this often is cheap; the ingest path +/// calls it on every MediaTek frame and lets this function decide whether +/// it's actually time to sample. +#[allow(clippy::too_many_arguments)] +pub(crate) fn maybe_sample_history( + history: &mut VecDeque, + last_sampled: &mut Option, + now: Instant, + room_level: f64, + room_abs_level: f64, + trackers: &HashMap, +) { + if last_sampled.is_some_and(|t| now.saturating_duration_since(t) < HISTORY_SAMPLE_INTERVAL) { + return; + } + *last_sampled = Some(now); + // Same staleness cutoff as activity.devices / /api/v1/csi/mediatek/devices + // (see ROOM_STALE_AFTER's doc comment) — a device the live listings have + // dropped shouldn't linger in freshly-recorded history samples either. + let fresh = |t: &&ActivityTracker| { + t.last_update() + .is_some_and(|u| now.saturating_duration_since(u) <= ROOM_STALE_AFTER) + }; + let device_levels: BTreeMap = trackers + .iter() + .filter(|(_, t)| fresh(t)) + .filter_map(|(id, t)| t.last_verdict().map(|v| (id.clone(), v.level))) + .collect(); + let device_abs_levels: BTreeMap = trackers + .iter() + .filter(|(_, t)| fresh(t)) + .filter_map(|(id, t)| t.last_verdict().map(|v| (id.clone(), v.abs_level))) + .collect(); + history.push_back(ActivitySample { + at: now, + room_level, + room_abs_level, + device_levels, + device_abs_levels, + }); + while history + .front() + .is_some_and(|s| now.saturating_duration_since(s.at) > HISTORY_RETENTION) + { + history.pop_front(); + } +} + +#[derive(Debug, Deserialize)] +pub(crate) struct ActivityHistoryQuery { + seconds: Option, +} + +/// `GET /api/v1/csi/mediatek/activity?seconds=300` — 1 Hz(-ish) samples of +/// room + per-device activity level, oldest first, for a strip chart. Each +/// sample's `age_ms` is relative to the moment of THIS request (there is no +/// meaningful absolute timestamp to report — retained samples are keyed by +/// a monotonic `Instant`, matching this crate's existing `age_ms` +/// convention elsewhere, e.g. `mediatek_devices`). `seconds` is capped at +/// `HISTORY_RETENTION` (300s) — asking for more just returns everything +/// retained. Never 404s: an empty `samples` array is a valid answer before +/// the first MediaTek frame has ever arrived. +pub(crate) async fn mediatek_activity_history( + State(state): State, + Query(query): Query, +) -> impl IntoResponse { + let s = state.read().await; + let now = Instant::now(); + let window = query + .seconds + .map(Duration::from_secs) + .unwrap_or(HISTORY_RETENTION) + .min(HISTORY_RETENTION); + let samples: Vec = s + .activity_history + .iter() + .filter(|sample| now.saturating_duration_since(sample.at) <= window) + .map(|sample| { + let devices: serde_json::Map = sample + .device_levels + .iter() + .map(|(id, level)| { + let abs_level = sample.device_abs_levels.get(id).copied().unwrap_or(0.0); + ( + id.clone(), + serde_json::json!({ "level": level, "abs_level": abs_level }), + ) + }) + .collect(); + serde_json::json!({ + "age_ms": now.saturating_duration_since(sample.at).as_millis() as u64, + "room_level": sample.room_level, + "room_abs_level": sample.room_abs_level, + "devices": devices, + }) + }) + .collect(); + ( + StatusCode::OK, + Json(serde_json::json!({ + "seconds": window.as_secs(), + "sample_count": samples.len(), + "samples": samples, + })), + ) +} + +/// Loads a previously-persisted `--activity-state` file (JSON +/// `{device_id: floor}`) at startup, for seeding new `ActivityTracker`s via +/// `ActivityTracker::seed_floor` so a restart doesn't reset the quiet-room +/// baseline. Never fails the caller: a missing file returns an empty map +/// (first run), and a missing/corrupt file logs a warning and also returns +/// empty (start fresh rather than refuse to boot over a stale-state file). +pub(crate) fn load_activity_state(path: &Path) -> HashMap { + match std::fs::read_to_string(path) { + Ok(contents) => serde_json::from_str(&contents).unwrap_or_else(|e| { + warn!( + "activity state at {} is not valid JSON ({e}) — starting with no seeded floors", + path.display() + ); + HashMap::new() + }), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => HashMap::new(), + Err(e) => { + warn!( + "could not read activity state at {} ({e}) — starting with no seeded floors", + path.display() + ); + HashMap::new() + } + } +} + +fn write_activity_state(path: &Path, floors: &HashMap) -> std::io::Result<()> { + let json = serde_json::to_string_pretty(floors)?; + // Write-then-rename so a crash mid-write can never leave a truncated, + // unparseable state file for the next `load_activity_state` to choke on. + let tmp = path.with_extension("tmp"); + std::fs::write(&tmp, json)?; + std::fs::rename(&tmp, path)?; + Ok(()) +} + +/// Background task: every `ACTIVITY_STATE_PERSIST_INTERVAL`, snapshots +/// every device's COMMITTED floor (`ActivityTracker::floor()` — a device +/// still in its `FLOOR_INIT_WINDOW` bootstrap has nothing durable yet) and +/// writes it to `path`. The write itself runs on a blocking thread so a +/// slow or contended disk never stalls the async runtime; only spawned at +/// all when `--activity-state` was given. +pub(crate) async fn activity_state_persist_task(state: SharedState, path: std::path::PathBuf) { + let mut interval = tokio::time::interval(ACTIVITY_STATE_PERSIST_INTERVAL); + loop { + interval.tick().await; + let snapshot: HashMap = { + let s = state.read().await; + s.mediatek_activity_by_device + .iter() + .filter_map(|(id, t)| t.floor().map(|f| (id.clone(), f))) + .collect() + }; + let write_path = path.clone(); + match tokio::task::spawn_blocking(move || write_activity_state(&write_path, &snapshot)) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => warn!( + "failed to persist activity state to {}: {e}", + path.display() + ), + Err(e) => warn!("activity state persist task panicked: {e}"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// 2026-09-20 live regression: floor used to seed from a single first + /// reading (`None => fast.max(FLOOR_HARD_MIN)`), which was often + /// near-zero on a fresh ring — pinning `abs_level` at 1.0 for hours + /// regardless of what the room actually did, since the floor could + /// only creep up at most 1%/min from there. The floor must instead + /// bootstrap from a percentile of the first FLOOR_INIT_WINDOW of + /// readings, so a device reading only its own steady, unremarkable + /// baseline does NOT read as maximally active. + #[test] + fn floor_bootstraps_from_the_init_window_not_a_single_first_reading() { + let mut tracker = ActivityTracker::default(); + let start = Instant::now(); + let mut t = start; + let quiet_fast = 0.01; // representative "nothing happening" reading + + let mut verdict = tracker.observe(t, quiet_fast, 10); + while t.duration_since(start) < FLOOR_INIT_WINDOW { + t += Duration::from_secs(2); + verdict = tracker.observe(t, quiet_fast, 10); + } + + assert!( + verdict.floor > quiet_fast * 0.5, + "floor committed far below the actual quiet reading: floor={} quiet_fast={quiet_fast}", + verdict.floor + ); + assert!( + verdict.abs_level < 0.3, + "a device reading only its own steady baseline must not saturate abs_level, got {}", + verdict.abs_level + ); + } + + /// Before the init window has fully elapsed, `abs_level` must still be + /// well-defined (from the running minimum seen so far), not a + /// placeholder 0 or 1 — a live dashboard shouldn't show a meaningless + /// number for the first 30s of a device's life. + #[test] + fn provisional_floor_during_the_bootstrap_window_tracks_the_running_minimum() { + let mut tracker = ActivityTracker::default(); + let start = Instant::now(); + // A dip partway through the window: the provisional floor must + // follow it down (not stay pinned at the first sample), since a + // later, genuinely lower reading should still be eligible to win + // the window's percentile once it commits. + let first = tracker.observe(start, 0.02, 10); + assert!( + first.floor > 0.0, + "provisional floor must be defined from the very first sample" + ); + let dip = tracker.observe(start + Duration::from_secs(5), 0.005, 10); + assert!( + dip.floor <= 0.005 + 1e-12, + "provisional floor must track a new lower reading: {}", + dip.floor + ); + } + + /// `--activity-state` restore: a device with a persisted floor must + /// skip the FLOOR_INIT_WINDOW bootstrap entirely — its very first + /// `observe()` call already reflects the seeded value. + #[test] + fn seed_floor_skips_the_bootstrap_window() { + let mut tracker = ActivityTracker::default(); + tracker.seed_floor(0.02); + let verdict = tracker.observe(Instant::now(), 0.02, 10); + assert!( + (verdict.floor - 0.02).abs() < 1e-9, + "seeded floor must be used immediately: {}", + verdict.floor + ); + assert!( + verdict.abs_level < 0.1, + "fast == seeded floor must read a near-zero abs_level right away" + ); + } + + #[test] + fn flat_amplitude_signal_converges_to_index_zero() { + let mut tracker = ActivityTracker::default(); + let start = Instant::now(); + let mut t = start; + let mut verdict = tracker.observe(t, 0.0, 10); + // Feed a perfectly flat `fast` reading for a while. + for _ in 0..20 { + t += Duration::from_millis(500); + verdict = tracker.observe(t, 0.0, 10); + } + assert!( + (verdict.index - 0.0).abs() < 1e-9, + "index was {}", + verdict.index + ); + // The [-2, +3] range isn't centered on 0: level = (index - (-2)) / 5, + // so index 0 sits at 2/5 = 0.4 up the scale, not the midpoint 0.5. + assert!( + (verdict.level - 0.4).abs() < 1e-9, + "index 0 must map to level 0.4, got {}", + verdict.level + ); + } + + #[test] + fn a_10x_burst_spikes_to_the_index_ceiling_then_decays() { + let mut tracker = ActivityTracker::default(); + let start = Instant::now(); + let mut t = start; + let baseline_fast = 0.05; + // Establish a settled baseline (slow ~= baseline_fast) over several + // slow-time-constant spans, well past what a naive fixed alpha + // would need — the time-aware EMA converges in wall-clock time, + // not sample count. + let mut verdict = tracker.observe(t, baseline_fast, 20); + for _ in 0..5 { + t += SLOW_TIME_CONSTANT; + verdict = tracker.observe(t, baseline_fast, 20); + } + assert!( + (verdict.slow - baseline_fast).abs() < baseline_fast * 0.01, + "baseline should have settled: slow={}", + verdict.slow + ); + assert!( + verdict.index.abs() < 0.05, + "settled baseline should read ~index 0, got {}", + verdict.index + ); + + // A sudden 10x std burst: fast jumps 10x, slow hasn't moved yet. + t += Duration::from_millis(100); + let burst = tracker.observe(t, baseline_fast * 10.0, 20); + assert!( + (burst.index - INDEX_MAX).abs() < 0.1, + "log2(10) ~= 3.32 should clamp to +3, got {}", + burst.index + ); + + // Hold the burst level and let the baseline start catching up — + // the index must decay back down, not stay pinned at the ceiling. + let mut sustained = burst; + for _ in 0..5 { + t += Duration::from_secs(60); + sustained = tracker.observe(t, baseline_fast * 10.0, 20); + } + assert!( + sustained.index < burst.index, + "index must decay as slow catches up to a sustained burst: {} -> {}", + burst.index, + sustained.index + ); + } + + /// While the relative + /// index decays back toward 0 as `slow` catches up to a SUSTAINED + /// burst, `abs_level` must stay high — it's judged against `floor` + /// (the quiet-room baseline), which barely moves while `fast` stays + /// well above it (floor only pulls DOWN toward a lower reading; a + /// sustained high reading only lets it creep up by at most 1%/min). + #[test] + fn a_sustained_burst_keeps_abs_level_high_while_relative_index_decays() { + let mut tracker = ActivityTracker::default(); + let start = Instant::now(); + let mut t = start; + let baseline_fast = 0.05; + + let mut verdict = tracker.observe(t, baseline_fast, 20); + for _ in 0..5 { + t += SLOW_TIME_CONSTANT; + verdict = tracker.observe(t, baseline_fast, 20); + } + assert!( + verdict.index.abs() < 0.05, + "settled baseline should read ~index 0, got {}", + verdict.index + ); + assert!( + verdict.abs_level < 0.05, + "a settled quiet baseline should read ~abs_level 0, got {}", + verdict.abs_level + ); + + // A sustained 10x std burst — held, not transient. + t += Duration::from_millis(100); + let burst = tracker.observe(t, baseline_fast * 10.0, 20); + assert!( + burst.abs_level > 0.3, + "a real burst must read a meaningfully high abs_level, got {}", + burst.abs_level + ); + + let mut sustained = burst; + for _ in 0..5 { + t += Duration::from_secs(60); + sustained = tracker.observe(t, baseline_fast * 10.0, 20); + } + assert!( + sustained.index < burst.index, + "sanity check (already covered elsewhere): relative index must have decayed, {} -> {}", + burst.index, + sustained.index + ); + assert!( + sustained.abs_level > burst.abs_level * 0.8, + "abs_level must stay high while the burst is sustained, not decay like the relative index: {} -> {}", + burst.abs_level, + sustained.abs_level + ); + } + + /// 2026-09-20 amendment: once activity genuinely drops back to (or + /// below) the quiet-room floor and stays there, `abs_level` must + /// return to ~0 — the floor's downward pull is what lets it "re-learn" + /// a quiet room after a crowd leaves, on the order of a few + /// `FLOOR_DOWN_TIME_CONSTANT`s. + #[test] + fn abs_level_returns_to_zero_once_activity_drops_back_to_the_floor() { + let mut tracker = ActivityTracker::default(); + let mut t = Instant::now(); + let baseline_fast = 0.05; + + tracker.observe(t, baseline_fast, 20); + for _ in 0..5 { + t += SLOW_TIME_CONSTANT; + tracker.observe(t, baseline_fast, 20); + } + + // Elevated for a while (people in the room). + t += Duration::from_millis(100); + for _ in 0..5 { + t += Duration::from_secs(60); + tracker.observe(t, baseline_fast * 10.0, 20); + } + + // Everyone leaves: fast drops straight back to the original quiet + // level, sustained long enough for the floor to re-learn it. + let mut after_leave = tracker.observe(t + Duration::from_secs(1), baseline_fast, 20); + for _ in 0..10 { + t += FLOOR_DOWN_TIME_CONSTANT; + after_leave = tracker.observe(t, baseline_fast, 20); + } + assert!( + after_leave.abs_level < 0.05, + "abs_level must return to ~0 once the floor re-learns the quiet level, got {} (floor={})", + after_leave.abs_level, + after_leave.floor + ); + } + + #[test] + fn two_devices_room_index_is_the_weighted_mean() { + let mut trackers = HashMap::new(); + let baseline_t = Instant::now(); + let now = baseline_t + Duration::from_secs(1); + + // A tracker's FIRST observation always seeds slow = fast (no prior + // baseline to compare against), so index is always exactly 0 on + // that call regardless of the fast value. Establish a real + // baseline first, then feed a different reading to get a genuinely + // non-zero, non-equal index per device. + let mut a = ActivityTracker::default(); + a.observe(baseline_t, 0.05, 30); + a.observe(now, 0.5, 30); // well above its own baseline -> positive index + + let mut b = ActivityTracker::default(); + b.observe(baseline_t, 0.05, 10); + b.observe(now, 0.06, 10); // barely above its own baseline -> small positive index + + trackers.insert("dev-a".to_string(), a); + trackers.insert("dev-b".to_string(), b); + + let idx_a = trackers["dev-a"].last_verdict().unwrap().index; + let idx_b = trackers["dev-b"].last_verdict().unwrap().index; + assert!( + idx_a > idx_b, + "test needs two genuinely different indices: a={idx_a} b={idx_b}" + ); + let expected = (idx_a * 30.0 + idx_b * 10.0) / 40.0; + + let room = room_index(&trackers, now, ROOM_STALE_AFTER); + assert!( + (room - expected).abs() < 1e-9, + "room index {} != weighted mean {}", + room, + expected + ); + // Sanity: the weighted mean must lie strictly between the two + // inputs (never at either extreme) for genuinely different values. + assert!( + room > idx_b && room < idx_a, + "room={room} must lie strictly between idx_b={idx_b} and idx_a={idx_a}" + ); + } + + #[test] + fn a_device_stale_past_room_stale_after_does_not_vote() { + let mut trackers = HashMap::new(); + let old = Instant::now(); + let now = old + ROOM_STALE_AFTER + Duration::from_secs(1); + let mut a = ActivityTracker::default(); + a.observe(old, 5.0, 30); + trackers.insert("dev-a".to_string(), a); + + assert_eq!( + room_index(&trackers, now, ROOM_STALE_AFTER), + INDEX_MIN, + "a stale-only room has no evidence of activity" + ); + } + + /// Room abs_level is a MAX over devices, not a weighted mean — a crowd + /// anywhere in any receiver's path counts, + /// so one busy device must win even against a much heavier-weighted + /// quiet one. + #[test] + fn room_abs_level_is_the_max_over_devices_not_a_weighted_mean() { + let mut trackers = HashMap::new(); + let baseline_t = Instant::now(); + let now = baseline_t + Duration::from_secs(1); + + // A tracker's FIRST observation always seeds floor = fast (no + // quiet-room baseline to compare against yet), so abs_level is 0 on + // that call regardless of the fast value — establish a real floor + // first, same as the relative-index tests above. + let mut quiet = ActivityTracker::default(); + quiet.observe(baseline_t, SLOW_FLOOR, 100); + quiet.observe(now, SLOW_FLOOR, 100); // heavy weight, stays at its own quiet floor + + let mut busy = ActivityTracker::default(); + busy.observe(baseline_t, SLOW_FLOOR, 1); + busy.observe(now, 0.5, 1); // light weight, but a real spike above its own floor + + trackers.insert("dev-quiet".to_string(), quiet); + trackers.insert("dev-busy".to_string(), busy); + + let abs_busy = trackers["dev-busy"].last_verdict().unwrap().abs_level; + let abs_quiet = trackers["dev-quiet"].last_verdict().unwrap().abs_level; + assert!( + abs_busy > abs_quiet, + "test setup needs a genuinely busier device" + ); + + let room_abs = room_abs_level(&trackers, now, ROOM_STALE_AFTER); + assert_eq!( + room_abs, abs_busy, + "room abs_level must be the max, not diluted by the heavier-weighted quiet device" + ); + } + + #[test] + fn room_peak_30s_holds_the_highest_recent_room_level() { + let mut history = VecDeque::new(); + let start = Instant::now(); + assert_eq!(room_peak_30s(&mut history, start, 0.2), 0.2); + assert_eq!( + room_peak_30s(&mut history, start + Duration::from_secs(5), 0.9), + 0.9 + ); + // A later, lower reading must not erase the still-recent peak + // (the t=5 sample, 0.9, is only 5s old here). + assert_eq!( + room_peak_30s(&mut history, start + Duration::from_secs(10), 0.3), + 0.9 + ); + // At t=40: t=0 (age 40>30, pruned), t=5 (age 35>30, pruned), t=10 + // (age exactly 30, kept — the peak sample has aged out, not this + // one), t=40 (new). Remaining max is the t=10 sample, 0.3. + assert_eq!( + room_peak_30s(&mut history, start + Duration::from_secs(40), 0.1), + 0.3 + ); + } + + #[test] + fn history_is_decimated_and_pruned() { + let mut history = VecDeque::new(); + let mut last_sampled = None; + let trackers = HashMap::new(); + let start = Instant::now(); + + maybe_sample_history(&mut history, &mut last_sampled, start, 0.1, 0.1, &trackers); + maybe_sample_history( + &mut history, + &mut last_sampled, + start + Duration::from_millis(200), + 0.2, + 0.2, + &trackers, + ); + assert_eq!( + history.len(), + 1, + "a sample within HISTORY_SAMPLE_INTERVAL must be dropped, not appended" + ); + + maybe_sample_history( + &mut history, + &mut last_sampled, + start + HISTORY_SAMPLE_INTERVAL, + 0.3, + 0.3, + &trackers, + ); + assert_eq!(history.len(), 2); + + maybe_sample_history( + &mut history, + &mut last_sampled, + start + HISTORY_RETENTION + Duration::from_secs(10), + 0.4, + 0.4, + &trackers, + ); + assert_eq!( + history.len(), + 1, + "samples older than HISTORY_RETENTION must be pruned" + ); + } +} diff --git a/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi_ring.rs b/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi_ring.rs index 6bd9c926a..b02089967 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi_ring.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi_ring.rs @@ -3,8 +3,9 @@ //! `/api/v1/csi/mediatek/devices/:device_id/{frames,summary}` routes. //! //! `MediatekCsiSnapshot` (`mediatek_csi.rs`) deliberately keeps only -//! mean/peak amplitude — enough for a coarse summary, but nothing for a -//! human to actually look at the channel with. This module extracts full per-chain, per-subcarrier +//! mean/peak amplitude — that's enough for the heuristic classifier +//! (`mediatek_heuristic`) but nothing for a human to actually look at the +//! channel with. This module extracts full per-chain, per-subcarrier //! amplitude and phase directly from each frame's raw MTC1 I/Q payload at //! ingest, and retains the last `capacity` frames per device — bounded, so //! memory stays fixed regardless of how long a device has been streaming. @@ -22,6 +23,7 @@ use axum::extract::{Path, Query, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Json}; use serde::Deserialize; +use std::time::Duration; use wifi_densepose_hardware::mediatek_csi::{CsiFrame, CsiPayload}; /// One (tx, rx) chain's per-subcarrier amplitude and phase for one frame. @@ -144,6 +146,66 @@ impl DeviceRing { (self.frames.len() - 1) as f64 / (span_us as f64 / 1_000_000.0) } + /// "Fast" activity signal for `mediatek_activity`: the mean, over every + /// `(chain, subcarrier)` pair observed in the last `window` of ring + /// time (by frame timestamp, relative to the newest retained frame — + /// not wall-clock `Instant::now()`), of that pair's coefficient of + /// variation (`std / mean`) — a gain-normalized measure of how much the + /// channel is fluctuating right now. Returns `(fast, frame_count)`; + /// `frame_count` is also the module's `frames_2s`-style weight. `(0.0, + /// n)` when fewer than 2 frames fall in the window (a std needs at + /// least 2 samples). + pub(crate) fn fast_activity(&self, window: Duration) -> (f64, usize) { + let Some(newest) = self.frames.back() else { + return (0.0, 0); + }; + let window_us = window.as_micros() as u64; + let recent: Vec<&RingFrame> = self + .frames + .iter() + .rev() + .take_while(|f| newest.timestamp_us.saturating_sub(f.timestamp_us) <= window_us) + .collect(); + let n = recent.len(); + if n < 2 { + return (0.0, n); + } + let sc = self.subcarrier_count as usize; + let mut cv_sum = 0.0f64; + let mut cv_count = 0usize; + for (tx, rx) in chain_union(recent.iter().copied()) { + let mut sum = vec![0.0f64; sc]; + let mut sumsq = vec![0.0f64; sc]; + let mut m = 0usize; + for frame in &recent { + let Some(c) = frame.chains.iter().find(|c| c.tx == tx && c.rx == rx) else { + continue; + }; + m += 1; + for (k, &a) in c.amplitude.iter().enumerate().take(sc) { + sum[k] += a as f64; + sumsq[k] += (a as f64) * (a as f64); + } + } + if m < 2 { + continue; + } + let m_f = m as f64; + for k in 0..sc { + let mean = sum[k] / m_f; + let variance = (sumsq[k] / m_f - mean * mean).max(0.0); + cv_sum += variance.sqrt() / mean.abs().max(1e-6); + cv_count += 1; + } + } + let fast = if cv_count > 0 { + cv_sum / cv_count as f64 + } else { + 0.0 + }; + (fast, n) + } + /// Per-chain amplitude mean and standard deviation across every /// retained frame that carried that chain, per subcarrier — the "what /// does the channel usually look like" view for an inspection client. @@ -414,13 +476,14 @@ pub(crate) async fn mediatek_csi_frames( /// `GET /api/v1/csi/mediatek/devices/{device_id}/summary` — one call for the /// analysis view: per-subcarrier amplitude mean/std per chain -/// over the ring, mean RSSI per Rx chain, and the ring's real frame rate. +/// over the ring, mean RSSI per Rx chain, the ring's real frame rate, and +/// the heuristic classifier's current window stats for the same device. /// `404` when `device_id` has no ring at all. pub(crate) async fn mediatek_csi_summary( State(state): State, Path(device_id): Path, ) -> impl IntoResponse { - let s = state.read().await; + let mut s = state.write().await; let Some(ring) = s.mediatek_csi_ring_by_device.get(&device_id) else { return device_not_found(&device_id).into_response(); }; @@ -451,6 +514,24 @@ pub(crate) async fn mediatek_csi_summary( let frame_rate_hz = ring.frame_rate_hz(); let subcarrier_count = ring.subcarrier_count(); + // Separate borrow: the heuristic's history lives in a different map on + // the same AppStateInner, and `verdict()` needs `&mut self` to prune by + // age — done after the (now-dropped) immutable `ring` borrow above ends. + let now = std::time::Instant::now(); + let heuristic = s + .mediatek_heuristic_by_device + .get_mut(&device_id) + .map(|hist| hist.verdict(now)) + .map(|v| { + serde_json::json!({ + "presence_state": v.presence_state.as_str(), + "coefficient_of_variation": v.window.coefficient_of_variation, + "baseline_deviation": v.window.baseline_deviation, + "sample_count": v.window.sample_count, + "span_ms": v.window.span_ms, + }) + }); + ( StatusCode::OK, Json(serde_json::json!({ @@ -461,6 +542,7 @@ pub(crate) async fn mediatek_csi_summary( "frame_rate_hz": frame_rate_hz, "rssi_by_rx_chain": rssi_by_rx_chain, "chains": chains, + "heuristic": heuristic, })), ) .into_response() @@ -623,6 +705,53 @@ mod tests { assert_eq!(by_pos[&(1, 1)], 2); } + #[test] + fn fast_activity_is_zero_for_a_flat_signal_and_positive_for_a_varying_one() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 1, + rx_count: 1, + subcarriers: 4, + ..Default::default() + }) + .unwrap(); + let mut flat_ring = DeviceRing::new(16); + for _ in 0..8 { + flat_ring.push(&sim.next_frame()); + } + // The ADR-266 simulator's amplitude has a slow motion phase, not + // frame-to-frame noise, so back-to-back frames read as effectively + // flat over a short window: fast_activity should be near zero. + let (fast_flat, n) = flat_ring.fast_activity(Duration::from_secs(2)); + assert_eq!(n, 8); + assert!( + fast_flat < 0.05, + "expected a near-flat signal, got fast={fast_flat}" + ); + } + + #[test] + fn fast_activity_ignores_frames_outside_the_window() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 1, + rx_count: 1, + subcarriers: 4, + frame_period_us: 500_000, // 0.5s apart + ..Default::default() + }) + .unwrap(); + let mut ring = DeviceRing::new(16); + for _ in 0..8 { + ring.push(&sim.next_frame()); // spans 3.5s total + } + let (_, n_full) = ring.fast_activity(Duration::from_secs(10)); + assert_eq!(n_full, 8, "a wide window covers every retained frame"); + let (_, n_narrow) = ring.fast_activity(Duration::from_secs(1)); + assert!( + n_narrow < 8, + "a 1s window must exclude older frames from a 3.5s-spanning ring, got {n_narrow}" + ); + } + mod endpoint_tests { //! Exercises `mediatek_csi_frames` / `mediatek_csi_summary` directly //! against `AppStateInner`, constructing the axum extractors by hand @@ -751,6 +880,11 @@ mod tests { .as_array() .expect("rssi_by_rx_chain array"); assert_eq!(rssi.len(), 1, "1 rx chain"); + + // heuristic is present but null when this device never went + // through mediatek_heuristic_by_device (this test only seeds the + // ring, mirroring a device whose heuristic hasn't observed yet). + assert!(body["heuristic"].is_null()); } #[tokio::test] diff --git a/v2/crates/wifi-densepose-sensing-server/src/mediatek_heuristic.rs b/v2/crates/wifi-densepose-sensing-server/src/mediatek_heuristic.rs new file mode 100644 index 000000000..6c4764820 --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/mediatek_heuristic.rs @@ -0,0 +1,688 @@ +//! A clearly-labelled heuristic presence/motion detector for real MediaTek +//! MTC1 CSI, used only because the validated ESP32 pipeline +//! (`extract_features_from_frame` / `VitalSignDetector`) needs per-subcarrier +//! amplitude and phase arrays that `MediatekCsiSnapshot` does not retain (see +//! its doc comment on `AppStateInner::latest_mediatek_csi`: "raw matrices are +//! not retained here"). All this module has to work with, per frame, is +//! `mean_amplitude`, `peak_amplitude`, and per-chain `rssi_dbm`. +//! +//! Every [`DeviceVerdict`] carries [`CLASSIFIER_ID`] (via +//! `mediatek_room_update` in main.rs) so nothing downstream can mistake it +//! for the validated ESP32 detector. Per repository policy (RuView +//! `CLAUDE.md`: "Never present WiFi sensing as camera-grade... tag +//! MEASURED/CLAIMED/SYNTHETIC"), treat this output as `CLAIMED`, not +//! `MEASURED` — it has not been validated against ground truth. +//! +//! # 2026-09-19 field data and the hysteresis/unknown-state fix +//! +//! In a multi-person field run with one person walking between the two +//! receivers (CLAIMED: from field notes, not a labelled evaluation), the v0 +//! design (a fixed 64-sample count window, single-sample threshold crossing) +//! flickered `present_moving` -> `absent` on 5+ of ~90 +//! two-second samples, sometimes for 10 s runs, with confidence collapsing to +//! 0.01-0.04 while both receivers were fresh (age < 1 s) and clearly +//! non-flat. Two causes, both addressed below: +//! +//! - The dump-loop clients are bursty (0.25-6 s gaps between arrivals per +//! receiver), so a **count**-based window mixed old and new activity +//! unpredictably depending on arrival timing rather than elapsed time. The +//! window is now **time**-based ([`WINDOW_DURATION`]): samples older than +//! the window are pruned by age on every observation, not by count, and a +//! gap does not clear or reset the window — it just means fewer samples +//! remain in it until fresh ones arrive. +//! - A single transient dip below threshold (a real one — e.g. a deep RF +//! fade mid-walk, not a bug) instantly flipped `presence` from `true` to +//! `false`. Presence now has hysteresis: fresh evidence sets presence +//! immediately, but leaving presence requires BOTH the metric being +//! continuously below threshold for [`CONFIRM_ABSENT_DURATION`] AND being +//! past [`HOLD_DURATION`] since the last positive evidence. +//! - Confidence still legitimately collapses sometimes (e.g. a genuinely +//! sparse window right after a receiver has been quiet). Reading a +//! low-confidence sample as a confident "absent" was itself the bug ("one +//! person stood up" was reported as `absent 0.09`). Below +//! [`UNKNOWN_CONFIDENCE_THRESHOLD`] the device now reports +//! [`PresenceState::Unknown`] (`presence: None`) instead of asserting +//! either way, and abstains from the room-level vote in +//! `mediatek_room_update` rather than getting to assert "absent" on weak +//! evidence. + +use std::collections::{HashMap, VecDeque}; +use std::time::{Duration, Instant}; + +/// Tags every output of this module so it is never mistaken for the +/// validated ESP32 `extract_features_from_frame` / `VitalSignDetector` +/// pipeline. Bump the trailing version if the method below changes in a way +/// that would invalidate a prior comparison. +pub(crate) const CLASSIFIER_ID: &str = "mediatek-amplitude-heuristic-v0"; + +/// Time-based amplitude window. Replaces a fixed sample count (v0) so a +/// bursty arrival pattern (dump-loop clients see 0.25-6 s gaps) doesn't make +/// the window's real time coverage depend on how lucky recent arrivals were. +/// Samples are pruned by age, not count; a gap does not reset the window. +pub(crate) const WINDOW_DURATION: Duration = Duration::from_secs(10); + +/// How long presence persists after the last positive evidence before a +/// flip to absent is even considered. A transient dip inside this window +/// never flips presence on its own. +pub(crate) const HOLD_DURATION: Duration = Duration::from_secs(10); + +/// Once past `HOLD_DURATION`, the metric must additionally read +/// continuously below threshold for this long before presence actually +/// commits to absent — one low sample is not enough. +pub(crate) const CONFIRM_ABSENT_DURATION: Duration = Duration::from_secs(3); + +/// Below this confidence, a device reports [`PresenceState::Unknown`] +/// (`presence: None`) rather than asserting either "present" or "absent", +/// and abstains from the room-level vote. Raw CSI abstains rather than +/// guesses, same convention the ADR-267 provenance gate already uses +/// elsewhere in this crate. +pub(crate) const UNKNOWN_CONFIDENCE_THRESHOLD: f64 = 0.3; + +/// Coefficient of variation (stdev / |mean|) of `mean_amplitude` over the +/// window above which motion is declared. Loosely tuned against the ADR-266 +/// simulator's synthetic motion phase; revisit as more real WN586X3 capture +/// accumulates. +pub(crate) const MOTION_CV_THRESHOLD: f64 = 0.05; + +/// Fractional deviation of the latest `mean_amplitude` from the slow +/// baseline above which "present but still" is declared even with low +/// frame-to-frame variance (a body blocking/reflecting the channel without +/// moving). +pub(crate) const BASELINE_DEVIATION_THRESHOLD: f64 = 0.08; + +/// EMA smoothing factor for the slow-moving empty-room baseline, applied +/// per-*sample* (not per elapsed time) — a known simplification: during a +/// dense burst the baseline drifts faster than during a quiet gap. Small +/// enough that a person entering the room doesn't drag the baseline toward +/// "occupied is normal" within a few samples either way. +pub(crate) const BASELINE_EMA_ALPHA: f64 = 0.02; + +/// Coarse motion label, matching the string vocabulary `RoomInference` / +/// `ClassificationInfo` already use elsewhere in this crate. This is the +/// *raw*, instantaneous reading for this observation, before hysteresis; +/// see [`PresenceState`] for the debounced, room-vote-eligible state. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum MotionLevel { + Absent, + PresentStill, + PresentMoving, +} + +impl MotionLevel { + pub(crate) fn as_str(self) -> &'static str { + match self { + MotionLevel::Absent => "absent", + MotionLevel::PresentStill => "present_still", + MotionLevel::PresentMoving => "present_moving", + } + } +} + +/// The hysteresis-committed, confidence-gated presence call. Distinct from +/// the raw `MotionLevel` above: this is what a device is allowed to assert +/// to the room vote. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PresenceState { + Present, + Absent, + /// Confidence below `UNKNOWN_CONFIDENCE_THRESHOLD`. Not a vote either + /// way — see `mediatek_room_update`'s exclusion of Unknown devices from + /// `fuse_room`. + Unknown, +} + +impl PresenceState { + pub(crate) fn as_str(self) -> &'static str { + match self { + PresenceState::Present => "present", + PresenceState::Absent => "absent", + PresenceState::Unknown => "unknown", + } + } +} + +/// Per-device amplitude-window statistics, exposed in `nodes[]` (as +/// `MediatekNodeDiagnostics` in main.rs) so a client can show why a device +/// reads the way it does. +#[derive(Debug, Clone, Copy, PartialEq)] +pub(crate) struct WindowStats { + pub(crate) coefficient_of_variation: f64, + pub(crate) baseline_deviation: f64, + pub(crate) sample_count: usize, + pub(crate) span_ms: u64, +} + +/// One device's verdict for the current moment. +#[derive(Debug, Clone, Copy, PartialEq)] +pub(crate) struct DeviceVerdict { + /// Raw, instantaneous motion reading (not hysteresis-debounced) — kept + /// for diagnostics; `presence_state`/`presence` are what should drive + /// any decision. + pub(crate) motion_level: MotionLevel, + pub(crate) presence_state: PresenceState, + /// `Some(true)`/`Some(false)` mirror `presence_state`'s Present/Absent; + /// `None` when `presence_state` is `Unknown`. Nullable specifically so a + /// low-confidence reading is never silently read as a confident + /// "absent" (2026-09-19: "one person stood up" was reported as + /// `absent 0.09` before this fix). + pub(crate) presence: Option, + /// This device's own confidence: window fill x how far past (or safely + /// under, for an absent-leaning reading) the relevant threshold the + /// signal sits. Cross-device agreement is applied separately by + /// [`apply_agreement`], since a single device has no visibility into + /// its peers. + pub(crate) raw_confidence: f64, + pub(crate) window: WindowStats, +} + +/// Time-windowed per-device amplitude history plus the slow baseline and +/// hysteresis state it's judged against. One instance lives per `device_id` +/// for the life of the process (or until the caller decides to drop a +/// long-silent device). +#[derive(Debug, Clone, Default)] +pub(crate) struct DeviceHistory { + /// (arrival time, mean_amplitude), oldest first. Pruned by age against + /// `WINDOW_DURATION` on every observation — never by count. + samples: VecDeque<(Instant, f64)>, + baseline: Option, + /// Last time raw evidence (motion or sustained deviation) was seen. + /// Presence cannot commit to absent within `HOLD_DURATION` of this. + last_present_evidence: Option, + /// When the metric most recently became continuously below threshold. + /// `None` whenever the metric is currently at/above threshold. Presence + /// cannot commit to absent until `now - this >= CONFIRM_ABSENT_DURATION`. + below_threshold_since: Option, + /// The last hysteresis-committed presence call (pre-confidence-gate). + /// Defaults to `false`: a device with no evidence yet has nothing to + /// hold onto, and starts out below the confidence threshold anyway (an + /// empty window), so it reports Unknown until it has real data — + /// this default is never observable as a false "present". + committed_presence: bool, +} + +impl DeviceHistory { + /// Fold in one new frame's `mean_amplitude`, observed at `now`, and + /// return this device's verdict. `peak_amplitude` is accepted for + /// future use (e.g. a saturation guard) but not yet part of the + /// decision. + pub(crate) fn observe( + &mut self, + now: Instant, + mean_amplitude: f64, + _peak_amplitude: f64, + ) -> DeviceVerdict { + self.samples.push_back((now, mean_amplitude)); + self.prune(now); + + self.baseline = Some(match self.baseline { + None => mean_amplitude, + Some(b) => b + BASELINE_EMA_ALPHA * (mean_amplitude - b), + }); + + self.verdict_at(now) + } + + /// Recompute this device's verdict at `now` without folding in a new + /// sample — used for devices that aren't the one whose frame just + /// arrived this tick, so the room picture still reflects every + /// currently-fresh receiver. Still prunes the window by age (a gap + /// since the last real sample must age the window normally). + pub(crate) fn verdict(&mut self, now: Instant) -> DeviceVerdict { + self.prune(now); + self.verdict_at(now) + } + + fn prune(&mut self, now: Instant) { + while let Some(&(t, _)) = self.samples.front() { + if now.saturating_duration_since(t) > WINDOW_DURATION { + self.samples.pop_front(); + } else { + break; + } + } + } + + fn verdict_at(&mut self, now: Instant) -> DeviceVerdict { + let n = self.samples.len(); + if n == 0 { + return DeviceVerdict { + motion_level: MotionLevel::Absent, + presence_state: PresenceState::Unknown, + presence: None, + raw_confidence: 0.0, + window: WindowStats { + coefficient_of_variation: 0.0, + baseline_deviation: 0.0, + sample_count: 0, + span_ms: 0, + }, + }; + } + + let span_ms = self + .samples + .back() + .zip(self.samples.front()) + .map(|((newest, _), (oldest, _))| { + newest.saturating_duration_since(*oldest).as_millis() as u64 + }) + .unwrap_or(0); + let window_fill = (span_ms as f64 / WINDOW_DURATION.as_millis() as f64).min(1.0); + + let mean: f64 = self.samples.iter().map(|(_, v)| v).sum::() / n as f64; + let variance = if n > 1 { + self.samples + .iter() + .map(|(_, v)| (v - mean).powi(2)) + .sum::() + / n as f64 + } else { + 0.0 + }; + let cv = if mean.abs() > f64::EPSILON { + variance.sqrt() / mean.abs() + } else { + 0.0 + }; + + let latest = self.samples.back().map(|(_, v)| *v).unwrap_or(mean); + let baseline = self.baseline.unwrap_or(latest); + let deviation = if baseline.abs() > f64::EPSILON { + (latest - baseline).abs() / baseline.abs() + } else { + 0.0 + }; + + let motion = cv > MOTION_CV_THRESHOLD; + let sustained = deviation > BASELINE_DEVIATION_THRESHOLD; + let raw_evidence = motion || sustained; + let motion_level = if motion { + MotionLevel::PresentMoving + } else if sustained { + MotionLevel::PresentStill + } else { + MotionLevel::Absent + }; + + // Hysteresis: feed the evidence timers, then decide whether a flip + // to absent is actually allowed yet. + if raw_evidence { + self.last_present_evidence = Some(now); + self.below_threshold_since = None; + self.committed_presence = true; + } else { + if self.below_threshold_since.is_none() { + self.below_threshold_since = Some(now); + } + let held_by_recency = self + .last_present_evidence + .is_some_and(|t| now.saturating_duration_since(t) < HOLD_DURATION); + let confirmed_absent = self + .below_threshold_since + .is_some_and(|t| now.saturating_duration_since(t) >= CONFIRM_ABSENT_DURATION); + if !held_by_recency && confirmed_absent { + self.committed_presence = false; + } + // Otherwise: keep whatever `committed_presence` already was — a + // transient dip inside the hold/confirm window never flips it. + } + + // Signal margin: how far past (present) or safely under (absent) + // threshold the stronger signal sits, normalized into roughly + // [0, 1] either way — based on the *committed* call, not the raw + // instantaneous one, so confidence doesn't itself flicker on a + // single-sample blip that hysteresis is about to override anyway. + let motion_margin = if MOTION_CV_THRESHOLD > 0.0 { + cv / MOTION_CV_THRESHOLD + } else { + 0.0 + }; + let deviation_margin = if BASELINE_DEVIATION_THRESHOLD > 0.0 { + deviation / BASELINE_DEVIATION_THRESHOLD + } else { + 0.0 + }; + let signal_margin = if self.committed_presence { + (motion_margin.max(deviation_margin) / 2.0).clamp(0.0, 1.0) + } else { + (1.0 - motion_margin.max(deviation_margin).min(1.0)).clamp(0.0, 1.0) + }; + + let raw_confidence = (window_fill * signal_margin).clamp(0.0, 1.0); + let presence_state = if raw_confidence < UNKNOWN_CONFIDENCE_THRESHOLD { + PresenceState::Unknown + } else if self.committed_presence { + PresenceState::Present + } else { + PresenceState::Absent + }; + let presence = match presence_state { + PresenceState::Present => Some(true), + PresenceState::Absent => Some(false), + PresenceState::Unknown => None, + }; + + DeviceVerdict { + motion_level, + presence_state, + presence, + raw_confidence, + window: WindowStats { + coefficient_of_variation: cv, + baseline_deviation: deviation, + sample_count: n, + span_ms, + }, + } + } +} + +/// Apply cross-device agreement to a set of already-computed verdicts, keyed +/// by `device_id`. Only devices with a definite `presence: Some(_)` +/// participate — an `Unknown` device neither votes nor is voted against, and +/// its own confidence is left unscaled (it is already below the unknown +/// threshold; scaling it further changes nothing observable). +/// +/// A lone definite device (or all-but-one Unknown) is never penalized — +/// there is nothing to disagree with. Otherwise each definite device's +/// confidence is scaled by `0.5 + 0.5 * agreement`, where `agreement` is the +/// fraction of *other* definite devices sharing its `presence` call: full +/// agreement leaves confidence unchanged, full disagreement halves it. A +/// hard zero on any disagreement was considered and rejected — two routers +/// legitimately cover different zones of a room (a doorway vs. a couch), so +/// routine partial disagreement should dampen confidence, not erase it. +pub(crate) fn apply_agreement(verdicts: &HashMap) -> HashMap { + let definite: Vec<(&String, bool)> = verdicts + .iter() + .filter_map(|(id, v)| v.presence.map(|p| (id, p))) + .collect(); + let total = definite.len(); + + verdicts + .iter() + .map(|(id, v)| { + let confidence = match v.presence { + None => v.raw_confidence, // Unknown: left as-is. + Some(_) if total <= 1 => v.raw_confidence, + Some(p) => { + let others = total - 1; + let agreeing_others = definite.iter().filter(|(_, o)| *o == p).count() - 1; + let agreement = agreeing_others as f64 / others as f64; + v.raw_confidence * (0.5 + 0.5 * agreement) + } + }; + (id.clone(), confidence.clamp(0.0, 1.0)) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn flat_amplitude_stream_is_absent_once_confidently_settled() { + let mut h = DeviceHistory::default(); + let start = Instant::now(); + let mut verdict = h.observe(start, 100.0, 102.0); + // Immediately after the first sample the window barely covers any + // time, so confidence should read Unknown, not a confident absent. + assert_eq!(verdict.presence_state, PresenceState::Unknown); + + // Feed flat samples spanning past HOLD_DURATION + CONFIRM_ABSENT_DURATION + // so the window fills and any (nonexistent) presence hold expires. + let mut t = start; + for _ in 0..40 { + t += Duration::from_millis(500); + verdict = h.observe(t, 100.0, 102.0); + } + assert_eq!(verdict.motion_level, MotionLevel::Absent); + assert_eq!(verdict.presence_state, PresenceState::Absent); + assert_eq!(verdict.presence, Some(false)); + assert!( + verdict.raw_confidence > 0.7, + "confidence was {}", + verdict.raw_confidence + ); + } + + #[test] + fn injected_variance_is_present_moving() { + let mut h = DeviceHistory::default(); + let start = Instant::now(); + let mut t = start; + let mut verdict = h.observe(t, 100.0, 102.0); + for i in 0..40 { + t += Duration::from_millis(500); + let sample = if i % 2 == 0 { 100.0 } else { 130.0 }; + verdict = h.observe(t, sample, sample + 2.0); + } + assert_eq!(verdict.motion_level, MotionLevel::PresentMoving); + assert_eq!(verdict.presence_state, PresenceState::Present); + assert_eq!(verdict.presence, Some(true)); + } + + /// The exact 2026-09-19 field bug: presence is well-established, then one + /// sample dips below threshold (a real transient fade, not a bug in the + /// underlying signal) — hysteresis must not flip presence to absent on + /// that single sample. + #[test] + fn a_single_transient_dip_does_not_flip_presence_to_absent() { + let mut h = DeviceHistory::default(); + let start = Instant::now(); + let mut t = start; + let mut verdict = h.observe(t, 100.0, 102.0); + // Establish confident, sustained motion. + for i in 0..40 { + t += Duration::from_millis(500); + let sample = if i % 2 == 0 { 100.0 } else { 130.0 }; + verdict = h.observe(t, sample, sample + 2.0); + } + assert_eq!( + verdict.presence, + Some(true), + "must be confidently present before the dip" + ); + + // One flat sample right after (a fade) — presence must hold. + t += Duration::from_millis(500); + verdict = h.observe(t, 115.0, 117.0); + assert_eq!( + verdict.presence, + Some(true), + "a single dip within HOLD_DURATION/CONFIRM_ABSENT_DURATION must not flip presence" + ); + } + + /// If the dip is real and sustained (not a blip), presence must + /// eventually commit to absent — hysteresis dampens, it doesn't hide a + /// genuine departure forever. + #[test] + fn a_sustained_dip_past_hold_and_confirm_duration_flips_to_absent() { + let mut h = DeviceHistory::default(); + let start = Instant::now(); + let mut t = start; + let mut verdict = h.observe(t, 100.0, 102.0); + for i in 0..40 { + t += Duration::from_millis(500); + let sample = if i % 2 == 0 { 100.0 } else { 130.0 }; + verdict = h.observe(t, sample, sample + 2.0); + } + assert_eq!(verdict.presence, Some(true)); + + // Flat for well past HOLD_DURATION + CONFIRM_ABSENT_DURATION. + for _ in 0..40 { + t += Duration::from_millis(500); + verdict = h.observe(t, 115.0, 117.0); + } + assert_eq!( + verdict.presence, + Some(false), + "a genuinely sustained dip must eventually commit to absent" + ); + } + + /// A bursty two-device timeline: one device reports in dense bursts with + /// multi-second gaps between them (per the real dump-loop behaviour). + /// The time-based window must not starve during a gap the way a + /// count-based window with a low arrival rate effectively would. + #[test] + fn bursty_arrivals_still_fill_a_time_based_window() { + let mut h = DeviceHistory::default(); + let start = Instant::now(); + let mut t = start; + let mut verdict = h.observe(t, 100.0, 102.0); + // Three bursts of 5 samples each, 3 s apart (a gap well inside + // WINDOW_DURATION, matching observed dump-loop gaps). + for _burst in 0..3 { + for i in 0..5 { + t += Duration::from_millis(100); + let sample = if i % 2 == 0 { 100.0 } else { 130.0 }; + verdict = h.observe(t, sample, sample + 2.0); + } + t += Duration::from_secs(3); + } + // All three bursts are within WINDOW_DURATION (10s) of the last one, + // so the window should have accumulated all 16 samples, not just + // the most recent burst. + assert_eq!(verdict.window.sample_count, 16); + assert!( + verdict.window.span_ms > 6_000, + "span was {}", + verdict.window.span_ms + ); + } + + /// A gap longer than WINDOW_DURATION must age out the old burst rather + /// than let it linger forever. + #[test] + fn a_gap_longer_than_the_window_ages_out_old_samples() { + let mut h = DeviceHistory::default(); + let start = Instant::now(); + let mut t = start; + for i in 0..5 { + t += Duration::from_millis(100); + let sample = if i % 2 == 0 { 100.0 } else { 130.0 }; + let _ = h.observe(t, sample, sample + 2.0); + } + t += WINDOW_DURATION + Duration::from_secs(1); + let verdict = h.observe(t, 100.0, 102.0); + assert_eq!( + verdict.window.sample_count, 1, + "the stale burst must have aged out" + ); + } + + #[test] + fn low_confidence_reads_as_unknown_not_absent() { + // A single sample: window barely covers any time, so window_fill is + // near zero regardless of the amplitude value — must read Unknown, + // never a confident "absent" (2026-09-19: "one person stood up" was + // reported as `absent 0.09` before this fix). + let mut h = DeviceHistory::default(); + let verdict = h.observe(Instant::now(), 1800.0, 9000.0); + assert_eq!(verdict.presence_state, PresenceState::Unknown); + assert_eq!(verdict.presence, None); + assert!(verdict.raw_confidence < UNKNOWN_CONFIDENCE_THRESHOLD); + } + + #[test] + fn lone_definite_device_is_never_penalized_by_agreement() { + let mut verdicts = HashMap::new(); + verdicts.insert( + "only-device".to_string(), + DeviceVerdict { + motion_level: MotionLevel::PresentMoving, + presence_state: PresenceState::Present, + presence: Some(true), + raw_confidence: 0.9, + window: WindowStats { + coefficient_of_variation: 0.1, + baseline_deviation: 0.0, + sample_count: 20, + span_ms: 9000, + }, + }, + ); + let adjusted = apply_agreement(&verdicts); + assert_eq!(adjusted["only-device"], 0.9); + } + + #[test] + fn unknown_device_neither_votes_nor_is_voted_against() { + let mut verdicts = HashMap::new(); + verdicts.insert( + "dev-present".to_string(), + DeviceVerdict { + motion_level: MotionLevel::PresentMoving, + presence_state: PresenceState::Present, + presence: Some(true), + raw_confidence: 0.9, + window: WindowStats { + coefficient_of_variation: 0.1, + baseline_deviation: 0.0, + sample_count: 20, + span_ms: 9000, + }, + }, + ); + verdicts.insert( + "dev-unknown".to_string(), + DeviceVerdict { + motion_level: MotionLevel::Absent, + presence_state: PresenceState::Unknown, + presence: None, + raw_confidence: 0.05, + window: WindowStats { + coefficient_of_variation: 0.01, + baseline_deviation: 0.0, + sample_count: 2, + span_ms: 200, + }, + }, + ); + let adjusted = apply_agreement(&verdicts); + // Only one *definite* device -> no disagreement possible, full confidence kept. + assert_eq!(adjusted["dev-present"], 0.9); + // Unknown device's own confidence is left untouched. + assert_eq!(adjusted["dev-unknown"], 0.05); + } + + #[test] + fn two_definite_devices_disagreeing_have_lowered_confidence() { + let mut verdicts = HashMap::new(); + verdicts.insert( + "dev-moving".to_string(), + DeviceVerdict { + motion_level: MotionLevel::PresentMoving, + presence_state: PresenceState::Present, + presence: Some(true), + raw_confidence: 0.9, + window: WindowStats { + coefficient_of_variation: 0.1, + baseline_deviation: 0.0, + sample_count: 20, + span_ms: 9000, + }, + }, + ); + verdicts.insert( + "dev-absent".to_string(), + DeviceVerdict { + motion_level: MotionLevel::Absent, + presence_state: PresenceState::Absent, + presence: Some(false), + raw_confidence: 0.9, + window: WindowStats { + coefficient_of_variation: 0.01, + baseline_deviation: 0.0, + sample_count: 20, + span_ms: 9000, + }, + }, + ); + let adjusted = apply_agreement(&verdicts); + assert!((adjusted["dev-moving"] - 0.45).abs() < 1e-9); + assert!((adjusted["dev-absent"] - 0.45).abs() < 1e-9); + } +}