#![allow(dead_code)]
mod adaptive_classifier;
pub mod cli;
pub mod csi;
mod engine_bridge;
mod field_bridge;
mod multistatic_bridge;
pub mod pose;
mod rvf_container;
mod rvf_pipeline;
mod tracker_bridge;
pub mod types;
mod vital_signs;
use wifi_densepose_sensing_server::{dataset, embedding, graph_transformer, trainer};
use ruvector_mincut::{DynamicMinCut, MinCutBuilder};
use std::collections::{HashMap, VecDeque};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use axum::{
extract::{
ws::{Message, WebSocket, WebSocketUpgrade},
Path, Query, State,
},
http::StatusCode,
response::{Html, IntoResponse, Json},
routing::{delete, get, post},
Extension, Router,
};
use clap::Parser;
use axum::http::HeaderValue;
use serde::{Deserialize, Serialize};
use tokio::net::UdpSocket;
use tokio::sync::{broadcast, RwLock};
use tower_http::services::ServeDir;
use tower_http::set_header::SetResponseHeaderLayer;
use tracing::{debug, error, info, warn};
use rvf_container::{RvfBuilder, RvfContainerInfo, RvfReader, VitalSignConfig};
use rvf_pipeline::ProgressiveLoader;
use vital_signs::{VitalSignDetector, VitalSigns};
use wifi_densepose_wifiscan::parse_netsh_output as parse_netsh_bssid_output;
use wifi_densepose_wifiscan::{BssidRegistry, WindowsWifiPipeline};
use wifi_densepose_signal::ruvsense::field_model::{CalibrationStatus, FieldModel};
use wifi_densepose_signal::ruvsense::multistatic::{MultistaticConfig, MultistaticFuser};
use wifi_densepose_signal::ruvsense::pose_tracker::PoseTracker;
#[derive(Parser, Debug)]
#[command(name = "sensing-server", about = "WiFi-DensePose sensing server")]
struct Args {
#[arg(long, default_value = "8080")]
http_port: u16,
#[arg(long, default_value = "8765")]
ws_port: u16,
#[arg(long, default_value = "5005")]
udp_port: u16,
#[arg(long, default_value = "../ui")]
ui_path: PathBuf,
#[arg(long, default_value = "100")]
tick_ms: u64,
#[arg(long, default_value = "127.0.0.1", env = "SENSING_BIND_ADDR")]
bind_addr: String,
#[arg(long = "allowed-host", value_name = "HOST")]
allowed_hosts: Vec<String>,
#[arg(long)]
disable_host_validation: bool,
#[command(flatten)]
mqtt_opts: wifi_densepose_sensing_server::cli::MqttArgs,
#[arg(long, default_value = "auto")]
source: String,
#[arg(long)]
benchmark: bool,
#[arg(long, value_name = "PATH")]
load_rvf: Option<PathBuf>,
#[arg(long, value_name = "PATH")]
save_rvf: Option<PathBuf>,
#[arg(long, value_name = "PATH")]
model: Option<PathBuf>,
#[arg(long)]
progressive: bool,
#[arg(long, value_name = "PATH")]
export_rvf: Option<PathBuf>,
#[arg(long)]
train: bool,
#[arg(long, value_name = "PATH")]
dataset: Option<PathBuf>,
#[arg(long, value_name = "TYPE", default_value = "mmfi")]
dataset_type: String,
#[arg(long, default_value = "100")]
epochs: usize,
#[arg(long, value_name = "DIR")]
checkpoint_dir: Option<PathBuf>,
#[arg(long)]
pretrain: bool,
#[arg(long, default_value = "50")]
pretrain_epochs: usize,
#[arg(long)]
embed: bool,
#[arg(long, value_name = "TYPE")]
build_index: Option<String>,
#[arg(long, env = "SENSING_NODE_POSITIONS")]
node_positions: Option<String>,
#[arg(long)]
calibrate: bool,
#[arg(
long,
value_name = "URL",
env = "RUVIEW_EDGE_REGISTRY_URL",
default_value = "https://storage.googleapis.com/cognitum-apps/app-registry.json"
)]
edge_registry_url: String,
#[arg(
long,
value_name = "SECS",
env = "RUVIEW_EDGE_REGISTRY_TTL_SECS",
default_value = "3600"
)]
edge_registry_ttl_secs: u64,
#[arg(long, env = "RUVIEW_NO_EDGE_REGISTRY")]
no_edge_registry: bool,
}
#[derive(Debug, Clone)]
#[allow(dead_code)]
struct Esp32Frame {
magic: u32,
node_id: u8,
n_antennas: u8,
n_subcarriers: u16,
freq_mhz: u16,
sequence: u32,
rssi: i8,
noise_floor: i8,
ppdu_type: wifi_densepose_hardware::PpduType,
amplitudes: Vec<f64>,
phases: Vec<f64>,
}
impl Esp32Frame {
fn grid(&self) -> (u16, wifi_densepose_hardware::PpduType) {
(self.n_subcarriers, self.ppdu_type)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SensingUpdate {
#[serde(rename = "type")]
msg_type: String,
timestamp: f64,
source: String,
tick: u64,
nodes: Vec<NodeInfo>,
features: FeatureInfo,
classification: ClassificationInfo,
signal_field: SignalField,
#[serde(skip_serializing_if = "Option::is_none")]
vital_signs: Option<VitalSigns>,
#[serde(skip_serializing_if = "Option::is_none")]
enhanced_motion: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
enhanced_breathing: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
posture: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
signal_quality_score: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
quality_verdict: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
bssid_count: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pose_keypoints: Option<Vec<[f64; 4]>>,
#[serde(skip_serializing_if = "Option::is_none")]
model_status: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
persons: Option<Vec<PersonDetection>>,
#[serde(skip_serializing_if = "Option::is_none")]
estimated_persons: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
node_features: Option<Vec<PerNodeFeatureInfo>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct NodeInfo {
node_id: u8,
rssi_dbm: f64,
position: [f64; 3],
amplitude: Vec<f64>,
subcarrier_count: usize,
#[serde(skip_serializing_if = "Option::is_none")]
sync: Option<NodeSyncSnapshot>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct NodeSyncSnapshot {
offset_us: i64,
is_leader: bool,
is_valid: bool,
smoothed: bool,
sequence: u32,
csi_fps_ema: f64,
csi_fps_samples: u32,
#[serde(skip_serializing_if = "Option::is_none")]
staleness_ms: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct FeatureInfo {
mean_rssi: f64,
variance: f64,
motion_band_power: f64,
breathing_band_power: f64,
dominant_freq_hz: f64,
change_points: usize,
spectral_power: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ClassificationInfo {
motion_level: String,
presence: bool,
confidence: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SignalField {
grid_size: [usize; 3],
values: Vec<f64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PoseKeypoint {
name: String,
x: f64,
y: f64,
z: f64,
confidence: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PersonDetection {
id: u32,
confidence: f64,
keypoints: Vec<PoseKeypoint>,
bbox: BoundingBox,
zone: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct BoundingBox {
x: f64,
y: f64,
width: f64,
height: f64,
}
struct NodeState {
pub(crate) frame_history: VecDeque<Vec<f64>>,
smoothed_person_score: f64,
pub(crate) prev_person_count: usize,
smoothed_motion: f64,
current_motion_level: String,
debounce_counter: u32,
debounce_candidate: String,
baseline_motion: f64,
baseline_frames: u64,
smoothed_hr: f64,
smoothed_br: f64,
smoothed_hr_conf: f64,
smoothed_br_conf: f64,
hr_buffer: VecDeque<f64>,
br_buffer: VecDeque<f64>,
rssi_history: VecDeque<f64>,
vital_detector: VitalSignDetector,
latest_vitals: VitalSigns,
pub(crate) last_frame_time: Option<std::time::Instant>,
edge_vitals: Option<Esp32VitalsPacket>,
latest_sync: Option<wifi_densepose_hardware::SyncPacket>,
latest_sync_at: Option<std::time::Instant>,
csi_fps_ema: f64,
csi_fps_samples: u32,
latest_features: Option<FeatureInfo>,
prev_keypoints: Option<Vec<[f64; 3]>>,
motion_energy_history: VecDeque<f64>,
coherence_score: f64,
feature_history: Option<wifi_densepose_signal::ruvsense::longitudinal::EmbeddingHistory>,
pub(crate) last_novelty_score: Option<f32>,
active_grid: Option<(u16, wifi_densepose_hardware::PpduType)>,
}
const TEMPORAL_EMA_ALPHA_DEFAULT: f64 = 0.15;
const TEMPORAL_EMA_ALPHA_LOW_COHERENCE: f64 = 0.05;
const COHERENCE_LOW_THRESHOLD: f64 = 0.3;
const MAX_BONE_CHANGE_RATIO: f64 = 0.20;
const COHERENCE_WINDOW: usize = 20;
const NOVELTY_VECTOR_DIM: usize = 56;
const NOVELTY_HISTORY_CAPACITY: usize = 64;
const NOVELTY_SKETCH_VERSION: u16 = 1;
pub(crate) fn update_csi_fps_ema(prev_fps: f64, dt_sec: f64) -> Option<f64> {
if !(dt_sec > 0.0 && dt_sec < 1.0) {
return None;
}
let instantaneous = 1.0 / dt_sec;
Some(prev_fps + (instantaneous - prev_fps) / 8.0)
}
#[cfg(test)]
mod fps_ema_tests {
use super::update_csi_fps_ema;
#[test]
fn steady_10hz_converges_toward_10() {
let mut fps = 20.0;
for _ in 0..40 {
fps = update_csi_fps_ema(fps, 0.100).unwrap();
}
assert!((fps - 10.0).abs() < 0.1,
"expected ~10 Hz after 40 samples at 100 ms intervals, got {fps}");
}
#[test]
fn steady_20hz_stays_near_20() {
let mut fps = 20.0;
for _ in 0..20 {
fps = update_csi_fps_ema(fps, 0.050).unwrap();
}
assert!((fps - 20.0).abs() < 0.05, "expected ~20 Hz, got {fps}");
}
#[test]
fn nonpositive_dt_rejected() {
assert!(update_csi_fps_ema(15.0, 0.0).is_none());
assert!(update_csi_fps_ema(15.0, -0.1).is_none());
}
#[test]
fn long_gap_rejected_as_implausible() {
assert!(update_csi_fps_ema(20.0, 2.0).is_none());
}
}
impl NodeState {
pub(crate) fn mesh_aligned_us(&self, local_at_frame_us: u64) -> Option<u64> {
let sync = self.latest_sync.as_ref()?;
let seen_at = self.latest_sync_at?;
if seen_at.elapsed() > std::time::Duration::from_secs(9) {
return None;
}
Some(sync.apply_to_local(local_at_frame_us))
}
pub(crate) fn mesh_aligned_us_for_csi_frame(&self, frame_sequence: u32) -> Option<u64> {
let sync = self.latest_sync.as_ref()?;
let seen_at = self.latest_sync_at?;
if seen_at.elapsed() > std::time::Duration::from_secs(9) {
return None;
}
let fps = if self.csi_fps_samples >= 5 { self.csi_fps_ema } else { 20.0 };
Some(sync.mesh_aligned_us_for_sequence(frame_sequence, fps))
}
pub(crate) fn apply_sync_packet(
&mut self,
pkt: wifi_densepose_hardware::SyncPacket,
now: std::time::Instant,
) {
self.latest_sync = Some(pkt);
self.latest_sync_at = Some(now);
}
pub(crate) fn sync_snapshot(&self) -> Option<NodeSyncSnapshot> {
let sync = self.latest_sync.as_ref()?;
Some(NodeSyncSnapshot {
offset_us: sync.local_minus_epoch_us(),
is_leader: sync.flags.is_leader,
is_valid: sync.flags.is_valid,
smoothed: sync.flags.smoothed_used,
sequence: sync.sequence,
csi_fps_ema: self.csi_fps_ema,
csi_fps_samples: self.csi_fps_samples,
staleness_ms: self.latest_sync_at.map(|t| t.elapsed().as_millis() as u64),
})
}
pub(crate) fn observe_csi_frame_arrival(&mut self, now: std::time::Instant) {
if let Some(prev) = self.last_frame_time {
let dt = now.duration_since(prev).as_secs_f64();
if let Some(new_ema) = update_csi_fps_ema(self.csi_fps_ema, dt) {
self.csi_fps_ema = new_ema;
self.csi_fps_samples = self.csi_fps_samples.saturating_add(1);
}
}
self.last_frame_time = Some(now);
}
pub(crate) fn new() -> Self {
Self {
frame_history: VecDeque::new(),
smoothed_person_score: 0.0,
prev_person_count: 0,
smoothed_motion: 0.0,
current_motion_level: "absent".to_string(),
debounce_counter: 0,
debounce_candidate: "absent".to_string(),
baseline_motion: 0.0,
baseline_frames: 0,
smoothed_hr: 0.0,
smoothed_br: 0.0,
smoothed_hr_conf: 0.0,
smoothed_br_conf: 0.0,
hr_buffer: VecDeque::with_capacity(8),
br_buffer: VecDeque::with_capacity(8),
rssi_history: VecDeque::new(),
vital_detector: VitalSignDetector::new(10.0),
latest_vitals: VitalSigns::default(),
last_frame_time: None,
edge_vitals: None,
latest_sync: None,
latest_sync_at: None,
csi_fps_ema: 20.0,
csi_fps_samples: 0,
latest_features: None,
prev_keypoints: None,
motion_energy_history: VecDeque::with_capacity(COHERENCE_WINDOW),
coherence_score: 1.0, feature_history: Some(
wifi_densepose_signal::ruvsense::longitudinal::EmbeddingHistory::with_sketch(
NOVELTY_VECTOR_DIM,
NOVELTY_HISTORY_CAPACITY,
NOVELTY_SKETCH_VERSION,
),
),
last_novelty_score: None,
active_grid: None,
}
}
fn accept_grid(&mut self, grid: (u16, wifi_densepose_hardware::PpduType)) -> bool {
match self.active_grid {
None => {
self.active_grid = Some(grid);
true
}
Some(active) if active == grid => true,
Some((active_n, _)) if grid.0 > active_n => {
self.active_grid = Some(grid);
self.frame_history.clear();
self.baseline_motion = 0.0;
self.baseline_frames = 0;
true
}
Some(_) => false,
}
}
pub(crate) fn update_novelty(&mut self, amplitudes: &[f64]) {
let history = match &mut self.feature_history {
Some(h) => h,
None => return,
};
let mut feature: Vec<f32> = amplitudes
.iter()
.take(NOVELTY_VECTOR_DIM)
.map(|&v| v as f32)
.collect();
feature.resize(NOVELTY_VECTOR_DIM, 0.0);
self.last_novelty_score = history.novelty(&feature);
let _ = history.push(
wifi_densepose_signal::ruvsense::longitudinal::EmbeddingEntry {
person_id: 0,
day_us: 0,
embedding: feature,
},
);
}
fn update_coherence(&mut self, motion_energy: f64) {
if self.motion_energy_history.len() >= COHERENCE_WINDOW {
self.motion_energy_history.pop_front();
}
self.motion_energy_history.push_back(motion_energy);
let n = self.motion_energy_history.len();
if n < 2 {
self.coherence_score = 1.0;
return;
}
let mean: f64 = self.motion_energy_history.iter().sum::<f64>() / n as f64;
let variance: f64 = self
.motion_energy_history
.iter()
.map(|v| (v - mean) * (v - mean))
.sum::<f64>()
/ (n - 1) as f64;
self.coherence_score = (1.0 / (1.0 + variance)).clamp(0.0, 1.0);
}
fn ema_alpha(&self) -> f64 {
if self.coherence_score < COHERENCE_LOW_THRESHOLD {
TEMPORAL_EMA_ALPHA_LOW_COHERENCE
} else {
TEMPORAL_EMA_ALPHA_DEFAULT
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PerNodeFeatureInfo {
node_id: u8,
features: FeatureInfo,
classification: ClassificationInfo,
rssi_dbm: f64,
last_seen_ms: u64,
frame_rate_hz: f64,
stale: bool,
#[serde(skip_serializing_if = "Option::is_none")]
novelty_score: Option<f32>,
}
fn build_node_features(
node_states: &std::collections::HashMap<u8, NodeState>,
now: std::time::Instant,
) -> Option<Vec<PerNodeFeatureInfo>> {
if node_states.is_empty() {
return None;
}
let entries: Vec<PerNodeFeatureInfo> = node_states
.iter()
.map(|(&node_id, ns)| {
let last_seen_ms = ns
.last_frame_time
.map(|t| now.saturating_duration_since(t).as_millis() as u64)
.unwrap_or(u64::MAX);
let stale = ns
.last_frame_time
.map(|t| now.saturating_duration_since(t) > ESP32_OFFLINE_TIMEOUT)
.unwrap_or(true);
let features = ns.latest_features.clone().unwrap_or(FeatureInfo {
mean_rssi: 0.0,
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,
});
PerNodeFeatureInfo {
node_id,
features,
classification: ClassificationInfo {
motion_level: ns.current_motion_level.clone(),
presence: !matches!(ns.current_motion_level.as_str(), "absent"),
confidence: ns.smoothed_person_score.clamp(0.0, 1.0),
},
rssi_dbm: ns.rssi_history.back().copied().unwrap_or(0.0),
last_seen_ms,
frame_rate_hz: 0.0, stale,
novelty_score: ns.last_novelty_score,
}
})
.collect();
Some(entries)
}
pub struct RollingP95 {
buf: std::collections::VecDeque<f64>,
window: usize,
min_samples: usize,
}
impl RollingP95 {
pub fn new(window: usize, min_samples: usize) -> Self {
Self {
buf: std::collections::VecDeque::with_capacity(window),
window,
min_samples,
}
}
pub fn push(&mut self, v: f64) {
if self.buf.len() == self.window {
self.buf.pop_front();
}
self.buf.push_back(v);
}
pub fn current(&self) -> Option<f64> {
if self.buf.len() < self.min_samples {
return None;
}
let mut sorted: Vec<f64> = self.buf.iter().copied().collect();
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let idx = ((sorted.len() as f64) * 0.95).ceil() as usize;
Some(sorted[idx.saturating_sub(1).min(sorted.len() - 1)])
}
#[allow(dead_code)]
pub fn len(&self) -> usize {
self.buf.len()
}
#[allow(dead_code)]
pub fn is_empty(&self) -> bool {
self.buf.is_empty()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct RuntimeConfig {
pub dedup_factor: f64,
}
impl Default for RuntimeConfig {
fn default() -> Self {
Self { dedup_factor: 3.0 }
}
}
pub(crate) fn load_runtime_config(data_dir: &std::path::Path) -> RuntimeConfig {
let path = data_dir.join("config.json");
match std::fs::read_to_string(&path) {
Ok(json) => serde_json::from_str(&json).unwrap_or_default(),
Err(_) => RuntimeConfig::default(),
}
}
pub(crate) fn save_runtime_config(data_dir: &std::path::Path, config: &RuntimeConfig) {
let path = data_dir.join("config.json");
if let Ok(json) = serde_json::to_string_pretty(config) {
if let Err(e) = std::fs::write(&path, json) {
warn!("Failed to save runtime config to {}: {e}", path.display());
} else {
info!("Runtime config saved to {}", path.display());
}
}
}
struct AppStateInner {
latest_update: Option<SensingUpdate>,
rssi_history: VecDeque<f64>,
frame_history: VecDeque<Vec<f64>>,
tick: u64,
source: String,
last_esp32_frame: Option<std::time::Instant>,
tx: broadcast::Sender<String>,
intro: wifi_densepose_sensing_server::introspection::IntrospectionState,
intro_tx: broadcast::Sender<String>,
total_detections: u64,
start_time: std::time::Instant,
vital_detector: VitalSignDetector,
latest_vitals: VitalSigns,
rvf_info: Option<RvfContainerInfo>,
save_rvf_path: Option<PathBuf>,
progressive_loader: Option<ProgressiveLoader>,
active_sona_profile: Option<String>,
model_loaded: bool,
smoothed_person_score: f64,
prev_person_count: usize,
smoothed_motion: f64,
current_motion_level: String,
debounce_counter: u32,
debounce_candidate: String,
baseline_motion: f64,
baseline_frames: u64,
smoothed_hr: f64,
smoothed_br: f64,
smoothed_hr_conf: f64,
smoothed_br_conf: f64,
hr_buffer: VecDeque<f64>,
br_buffer: VecDeque<f64>,
edge_vitals: Option<Esp32VitalsPacket>,
latest_wasm_events: Option<WasmOutputPacket>,
discovered_models: Vec<serde_json::Value>,
active_model_id: Option<String>,
recordings: Vec<serde_json::Value>,
recording_active: bool,
recording_start_time: Option<std::time::Instant>,
recording_current_id: Option<String>,
recording_stop_tx: Option<tokio::sync::watch::Sender<bool>>,
training_status: String,
training_config: Option<serde_json::Value>,
adaptive_model: Option<adaptive_classifier::AdaptiveModel>,
node_states: HashMap<u8, NodeState>,
pose_tracker: PoseTracker,
last_tracker_instant: Option<std::time::Instant>,
multistatic_fuser: MultistaticFuser,
engine_bridge: engine_bridge::EngineBridge,
field_model: Option<FieldModel>,
pub(crate) p95_variance: RollingP95,
pub(crate) p95_motion_band_power: RollingP95,
pub(crate) p95_spectral_power: RollingP95,
pub(crate) dedup_factor: f64,
pub(crate) data_dir: std::path::PathBuf,
}
const ESP32_OFFLINE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
impl AppStateInner {
fn person_count(&self) -> usize {
match self.field_model.as_ref() {
Some(fm) => {
let history = if !self.frame_history.is_empty() {
&self.frame_history
} else {
self.node_states
.values()
.filter(|ns| !ns.frame_history.is_empty())
.max_by_key(|ns| ns.last_frame_time)
.map(|ns| &ns.frame_history)
.unwrap_or(&self.frame_history)
};
field_bridge::occupancy_or_fallback(
fm,
history,
self.smoothed_person_score,
self.prev_person_count,
)
}
None => score_to_person_count(self.smoothed_person_score, self.prev_person_count),
}
}
fn effective_source(&self) -> String {
if self.source == "esp32" {
if let Some(last) = self.last_esp32_frame {
if last.elapsed() > ESP32_OFFLINE_TIMEOUT {
return "esp32:offline".to_string();
}
}
}
self.source.clone()
}
}
const FRAME_HISTORY_CAPACITY: usize = 100;
type SharedState = Arc<RwLock<AppStateInner>>;
#[derive(Debug, Clone, Serialize)]
struct Esp32VitalsPacket {
node_id: u8,
presence: bool,
fall_detected: bool,
motion: bool,
breathing_rate_bpm: f64,
heartrate_bpm: f64,
rssi: i8,
n_persons: u8,
motion_energy: f32,
presence_score: f32,
timestamp_ms: u32,
}
fn parse_esp32_vitals(buf: &[u8]) -> Option<Esp32VitalsPacket> {
if buf.len() < 32 {
return None;
}
let magic = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
if magic != 0xC511_0002 {
return None;
}
let node_id = buf[4];
let flags = buf[5];
let breathing_raw = u16::from_le_bytes([buf[6], buf[7]]);
let heartrate_raw = u32::from_le_bytes([buf[8], buf[9], buf[10], buf[11]]);
let rssi = buf[12] as i8;
let n_persons = buf[13];
let motion_energy = f32::from_le_bytes([buf[16], buf[17], buf[18], buf[19]]);
let presence_score = f32::from_le_bytes([buf[20], buf[21], buf[22], buf[23]]);
let timestamp_ms = u32::from_le_bytes([buf[24], buf[25], buf[26], buf[27]]);
Some(Esp32VitalsPacket {
node_id,
presence: (flags & 0x01) != 0,
fall_detected: (flags & 0x02) != 0,
motion: (flags & 0x04) != 0,
breathing_rate_bpm: breathing_raw as f64 / 100.0,
heartrate_bpm: heartrate_raw as f64 / 10000.0,
rssi,
n_persons,
motion_energy,
presence_score,
timestamp_ms,
})
}
#[derive(Debug, Clone, Serialize)]
struct WasmEvent {
event_type: u8,
value: f32,
}
#[derive(Debug, Clone, Serialize)]
struct WasmOutputPacket {
node_id: u8,
module_id: u8,
events: Vec<WasmEvent>,
}
fn parse_wasm_output(buf: &[u8]) -> Option<WasmOutputPacket> {
if buf.len() < 8 {
return None;
}
let magic = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
if magic != 0xC511_0007 {
return None;
}
let node_id = buf[4];
let module_id = buf[5];
let event_count = u16::from_le_bytes([buf[6], buf[7]]) as usize;
let mut events = Vec::with_capacity(event_count);
let mut offset = 8;
for _ in 0..event_count {
if offset + 5 > buf.len() {
break;
}
let event_type = buf[offset];
let value = f32::from_le_bytes([
buf[offset + 1],
buf[offset + 2],
buf[offset + 3],
buf[offset + 4],
]);
events.push(WasmEvent { event_type, value });
offset += 5;
}
Some(WasmOutputPacket {
node_id,
module_id,
events,
})
}
#[derive(Debug, Clone, Serialize)]
struct EdgeFusedVitalsPacket {
node_id: u8,
flags: u8,
breathing_rate_bpm: f32,
heartrate_bpm: f32,
rssi: i8,
n_persons: u8,
mmwave_type: u8,
fusion_confidence: u8,
motion_energy: f32,
presence_score: f32,
timestamp_ms: u32,
mmwave_hr_bpm: f32,
mmwave_br_bpm: f32,
mmwave_distance_cm: f32,
mmwave_targets: u8,
mmwave_confidence: u8,
}
fn parse_edge_fused_vitals(buf: &[u8]) -> Option<EdgeFusedVitalsPacket> {
if buf.len() < 48 {
return None;
}
let magic = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
if magic != 0xC511_0004 {
return None;
}
let node_id = buf[4];
let flags = buf[5];
let breathing_raw = u16::from_le_bytes([buf[6], buf[7]]);
let heartrate_raw = u32::from_le_bytes([buf[8], buf[9], buf[10], buf[11]]);
let rssi = buf[12] as i8;
let n_persons = buf[13];
let mmwave_type = buf[14];
let fusion_confidence = buf[15];
let motion_energy = f32::from_le_bytes([buf[16], buf[17], buf[18], buf[19]]);
let presence_score = f32::from_le_bytes([buf[20], buf[21], buf[22], buf[23]]);
let timestamp_ms = u32::from_le_bytes([buf[24], buf[25], buf[26], buf[27]]);
let mmwave_hr_bpm = f32::from_le_bytes([buf[28], buf[29], buf[30], buf[31]]);
let mmwave_br_bpm = f32::from_le_bytes([buf[32], buf[33], buf[34], buf[35]]);
let mmwave_distance_cm = f32::from_le_bytes([buf[36], buf[37], buf[38], buf[39]]);
let mmwave_targets = buf[40];
let mmwave_confidence = buf[41];
Some(EdgeFusedVitalsPacket {
node_id,
flags,
breathing_rate_bpm: breathing_raw as f32 / 100.0,
heartrate_bpm: heartrate_raw as f32 / 10000.0,
rssi,
n_persons,
mmwave_type,
fusion_confidence,
motion_energy,
presence_score,
timestamp_ms,
mmwave_hr_bpm,
mmwave_br_bpm,
mmwave_distance_cm,
mmwave_targets,
mmwave_confidence,
})
}
#[cfg(test)]
mod issue_928_magic_collision_tests {
use super::*;
fn build_fused_vitals_packet() -> Vec<u8> {
let mut buf = vec![0u8; 48];
buf[0..4].copy_from_slice(&0xC511_0004u32.to_le_bytes());
buf[4] = 9; buf[5] = 0b0000_1001; buf[6..8].copy_from_slice(&1600u16.to_le_bytes()); buf[8..12].copy_from_slice(&720_000u32.to_le_bytes()); buf[12] = (-55i8) as u8; buf[13] = 1; buf[14] = 2; buf[15] = 85; buf[16..20].copy_from_slice(&0.42f32.to_le_bytes()); buf[20..24].copy_from_slice(&0.95f32.to_le_bytes()); buf[24..28].copy_from_slice(&1_234_567u32.to_le_bytes()); buf[28..32].copy_from_slice(&71.5f32.to_le_bytes()); buf[32..36].copy_from_slice(&15.8f32.to_le_bytes()); buf[36..40].copy_from_slice(&182.0f32.to_le_bytes()); buf[40] = 1; buf[41] = 90; buf
}
#[test]
fn parse_edge_fused_vitals_extracts_fields_correctly() {
let buf = build_fused_vitals_packet();
let pkt = parse_edge_fused_vitals(&buf).expect("must parse a well-formed packet");
assert_eq!(pkt.node_id, 9);
assert_eq!(pkt.flags, 0b0000_1001);
assert!((pkt.breathing_rate_bpm - 16.0).abs() < 1e-3, "breathing scale 100");
assert!((pkt.heartrate_bpm - 72.0).abs() < 1e-3, "heartrate scale 10000");
assert_eq!(pkt.rssi, -55);
assert_eq!(pkt.n_persons, 1);
assert_eq!(pkt.mmwave_type, 2);
assert_eq!(pkt.fusion_confidence, 85);
assert!((pkt.motion_energy - 0.42).abs() < 1e-6);
assert!((pkt.presence_score - 0.95).abs() < 1e-6);
assert_eq!(pkt.timestamp_ms, 1_234_567);
assert!((pkt.mmwave_hr_bpm - 71.5).abs() < 1e-6);
assert!((pkt.mmwave_br_bpm - 15.8).abs() < 1e-3);
assert!((pkt.mmwave_distance_cm - 182.0).abs() < 1e-6);
assert_eq!(pkt.mmwave_targets, 1);
assert_eq!(pkt.mmwave_confidence, 90);
}
#[test]
fn parse_edge_fused_vitals_rejects_short_buffer() {
let buf = build_fused_vitals_packet();
assert!(parse_edge_fused_vitals(&buf[..47]).is_none());
}
#[test]
fn parse_edge_fused_vitals_rejects_wrong_magic() {
let mut buf = build_fused_vitals_packet();
buf[0..4].copy_from_slice(&0xC511_0007u32.to_le_bytes()); assert!(parse_edge_fused_vitals(&buf).is_none());
}
#[test]
fn parse_wasm_output_rejects_legacy_0004_magic() {
let buf = build_fused_vitals_packet();
assert!(parse_wasm_output(&buf).is_none(),
"issue #928: WASM parser must NOT accept 0xC511_0004");
}
#[test]
fn parse_wasm_output_accepts_new_0007_magic() {
let mut buf = vec![0u8; 8];
buf[0..4].copy_from_slice(&0xC511_0007u32.to_le_bytes());
buf[4] = 5; buf[5] = 1; buf[6..8].copy_from_slice(&0u16.to_le_bytes()); let pkt = parse_wasm_output(&buf).expect("0xC511_0007 must parse");
assert_eq!(pkt.node_id, 5);
assert_eq!(pkt.module_id, 1);
assert!(pkt.events.is_empty());
}
}
fn parse_esp32_frame(buf: &[u8]) -> Option<Esp32Frame> {
if buf.len() < 20 {
return None;
}
let magic = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
if magic != 0xC511_0001 {
return None;
}
let node_id = buf[4];
let n_antennas = buf[5];
let n_subcarriers = u16::from_le_bytes([buf[6], buf[7]]);
let freq_mhz =
u16::try_from(u32::from_le_bytes([buf[8], buf[9], buf[10], buf[11]])).unwrap_or(0);
let sequence = u32::from_le_bytes([buf[12], buf[13], buf[14], buf[15]]);
let rssi_raw = buf[16] as i8;
let rssi = if rssi_raw > 0 {
rssi_raw.saturating_neg()
} else {
rssi_raw
};
let noise_floor = buf[17] as i8;
let ppdu_type = wifi_densepose_hardware::PpduType::from_byte(buf[18]);
let iq_start = 20;
let n_pairs = n_antennas as usize * n_subcarriers as usize;
let expected_len = iq_start + n_pairs * 2;
if buf.len() < expected_len {
return None;
}
let mut amplitudes = Vec::with_capacity(n_pairs);
let mut phases = Vec::with_capacity(n_pairs);
for k in 0..n_pairs {
let i_val = buf[iq_start + k * 2] as i8 as f64;
let q_val = buf[iq_start + k * 2 + 1] as i8 as f64;
amplitudes.push((i_val * i_val + q_val * q_val).sqrt());
phases.push(q_val.atan2(i_val));
}
Some(Esp32Frame {
magic,
node_id,
n_antennas,
n_subcarriers,
freq_mhz,
sequence,
rssi,
noise_floor,
ppdu_type,
amplitudes,
phases,
})
}
fn generate_signal_field(
_mean_rssi: f64,
motion_score: f64,
breathing_rate_hz: f64,
signal_quality: f64,
subcarrier_variances: &[f64],
) -> SignalField {
let grid = 20usize;
let mut values = vec![0.0f64; grid * grid];
let center = (grid as f64 - 1.0) / 2.0;
let max_var = subcarrier_variances.iter().cloned().fold(0.0f64, f64::max);
let norm_factor = if max_var > 1e-9 { max_var } else { 1.0 };
let n_sub = subcarrier_variances.len().max(1);
for (k, &var) in subcarrier_variances.iter().enumerate() {
let weight = (var / norm_factor) * motion_score;
if weight < 1e-6 {
continue;
}
let angle = (k as f64 / n_sub as f64) * 2.0 * std::f64::consts::PI;
let radius = center * 0.8 * weight.sqrt();
let hx = center + radius * angle.cos();
let hz = center + radius * angle.sin();
for z in 0..grid {
for x in 0..grid {
let dx = x as f64 - hx;
let dz = z as f64 - hz;
let dist2 = dx * dx + dz * dz;
let spread = (0.5 + weight * 2.0).max(0.5);
values[z * grid + x] += weight * (-dist2 / (2.0 * spread * spread)).exp();
}
}
}
for z in 0..grid {
for x in 0..grid {
let dx = x as f64 - center;
let dz = z as f64 - center;
let dist = (dx * dx + dz * dz).sqrt();
let base = signal_quality * (-dist * 0.12).exp();
values[z * grid + x] += base * 0.3;
}
}
if breathing_rate_hz > 0.05 {
let ring_r = center * 0.55;
let ring_width = 1.8f64;
for z in 0..grid {
for x in 0..grid {
let dx = x as f64 - center;
let dz = z as f64 - center;
let dist = (dx * dx + dz * dz).sqrt();
let ring_val =
0.08 * (-(dist - ring_r).powi(2) / (2.0 * ring_width * ring_width)).exp();
values[z * grid + x] += ring_val;
}
}
}
let field_max = values.iter().cloned().fold(0.0f64, f64::max);
let scale = if field_max > 1e-9 {
1.0 / field_max
} else {
1.0
};
for v in &mut values {
*v = (*v * scale).clamp(0.0, 1.0);
}
SignalField {
grid_size: [grid, 1, grid],
values,
}
}
fn estimate_breathing_rate_hz(frame_history: &VecDeque<Vec<f64>>, sample_rate_hz: f64) -> f64 {
let n = frame_history.len();
if n < 6 {
return 0.0;
}
let series: Vec<f64> = frame_history
.iter()
.map(|amps| {
if amps.is_empty() {
0.0
} else {
amps.iter().sum::<f64>() / amps.len() as f64
}
})
.collect();
let mean_s = series.iter().sum::<f64>() / n as f64;
let detrended: Vec<f64> = series.iter().map(|x| x - mean_s).collect();
let n_candidates = 9usize;
let f_low = 0.1f64;
let f_high = 0.5f64;
let mut best_freq = 0.0f64;
let mut best_power = 0.0f64;
for i in 0..n_candidates {
let freq = f_low + (f_high - f_low) * i as f64 / (n_candidates - 1).max(1) as f64;
let omega = 2.0 * std::f64::consts::PI * freq / sample_rate_hz;
let coeff = 2.0 * omega.cos();
let mut s_prev2 = 0.0f64;
let mut s_prev1 = 0.0f64;
for &x in &detrended {
let s = x + coeff * s_prev1 - s_prev2;
s_prev2 = s_prev1;
s_prev1 = s;
}
let power = s_prev2 * s_prev2 + s_prev1 * s_prev1 - coeff * s_prev1 * s_prev2;
if power > best_power {
best_power = power;
best_freq = freq;
}
}
let avg_power = {
let mut total = 0.0f64;
for i in 0..n_candidates {
let freq = f_low + (f_high - f_low) * i as f64 / (n_candidates - 1).max(1) as f64;
let omega = 2.0 * std::f64::consts::PI * freq / sample_rate_hz;
let coeff = 2.0 * omega.cos();
let mut s_prev2 = 0.0f64;
let mut s_prev1 = 0.0f64;
for &x in &detrended {
let s = x + coeff * s_prev1 - s_prev2;
s_prev2 = s_prev1;
s_prev1 = s;
}
total += s_prev2 * s_prev2 + s_prev1 * s_prev1 - coeff * s_prev1 * s_prev2;
}
total / n_candidates as f64
};
if best_power > avg_power * 3.0 {
best_freq.clamp(f_low, f_high)
} else {
0.0
}
}
fn compute_subcarrier_importance_weights(sensitivity: &[f64]) -> Vec<f64> {
let n = sensitivity.len();
if n == 0 {
return vec![];
}
let max_sens = sensitivity
.iter()
.cloned()
.fold(f64::NEG_INFINITY, f64::max)
.max(1e-9);
let mut sorted = sensitivity.to_vec();
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let median = if n % 2 == 0 {
(sorted[n / 2 - 1] + sorted[n / 2]) / 2.0
} else {
sorted[n / 2]
};
sensitivity
.iter()
.map(|&s| {
if s >= median {
1.0 + (s / max_sens).min(1.0)
} else {
0.5
}
})
.collect()
}
fn compute_subcarrier_variances(frame_history: &VecDeque<Vec<f64>>, n_sub: usize) -> Vec<f64> {
if frame_history.is_empty() || n_sub == 0 {
return vec![0.0; n_sub];
}
let n_frames = frame_history.len() as f64;
let mut means = vec![0.0f64; n_sub];
let mut sq_means = vec![0.0f64; n_sub];
for frame in frame_history.iter() {
for k in 0..n_sub {
let a = if k < frame.len() { frame[k] } else { 0.0 };
means[k] += a;
sq_means[k] += a * a;
}
}
(0..n_sub)
.map(|k| {
let mean = means[k] / n_frames;
let sq_mean = sq_means[k] / n_frames;
(sq_mean - mean * mean).max(0.0)
})
.collect()
}
fn extract_features_from_frame(
frame: &Esp32Frame,
frame_history: &VecDeque<Vec<f64>>,
sample_rate_hz: f64,
) -> (FeatureInfo, ClassificationInfo, f64, Vec<f64>, f64) {
let n_sub = frame.amplitudes.len().max(1);
let n = n_sub as f64;
let mean_rssi = frame.rssi as f64;
let sub_sensitivity: Vec<f64> = frame.amplitudes.iter().map(|a| a.abs()).collect();
let importance_weights = compute_subcarrier_importance_weights(&sub_sensitivity);
let weight_sum: f64 = importance_weights.iter().sum::<f64>();
let mean_amp: f64 = if weight_sum > 0.0 {
frame
.amplitudes
.iter()
.zip(importance_weights.iter())
.map(|(a, w)| a * w)
.sum::<f64>()
/ weight_sum
} else {
frame.amplitudes.iter().sum::<f64>() / n
};
let intra_variance: f64 = if weight_sum > 0.0 {
frame
.amplitudes
.iter()
.zip(importance_weights.iter())
.map(|(a, w)| w * (a - mean_amp).powi(2))
.sum::<f64>()
/ weight_sum
} else {
frame
.amplitudes
.iter()
.map(|a| (a - mean_amp).powi(2))
.sum::<f64>()
/ n
};
let sub_variances = compute_subcarrier_variances(frame_history, n_sub);
let temporal_variance: f64 = if sub_variances.is_empty() {
intra_variance
} else {
sub_variances.iter().sum::<f64>() / sub_variances.len() as f64
};
let variance = intra_variance.max(temporal_variance);
let spectral_power: f64 = frame.amplitudes.iter().map(|a| a * a).sum::<f64>() / n;
let half = frame.amplitudes.len() / 2;
let motion_band_power = if half > 0 {
frame.amplitudes[half..]
.iter()
.map(|a| (a - mean_amp).powi(2))
.sum::<f64>()
/ (frame.amplitudes.len() - half) as f64
} else {
0.0
};
let breathing_band_power = if half > 0 {
frame.amplitudes[..half]
.iter()
.map(|a| (a - mean_amp).powi(2))
.sum::<f64>()
/ half as f64
} else {
0.0
};
let peak_idx = frame
.amplitudes
.iter()
.enumerate()
.max_by(|a, b| a.1.partial_cmp(b.1).unwrap_or(std::cmp::Ordering::Equal))
.map(|(i, _)| i)
.unwrap_or(0);
let dominant_freq_hz = peak_idx as f64 * 0.05;
let threshold = mean_amp * 1.2;
let change_points = frame
.amplitudes
.windows(2)
.filter(|w| (w[0] < threshold) != (w[1] < threshold))
.count();
let temporal_motion_score = if let Some(prev_frame) = frame_history.back() {
let n_cmp = n_sub.min(prev_frame.len());
if n_cmp > 0 {
let diff_energy: f64 = (0..n_cmp)
.map(|k| (frame.amplitudes[k] - prev_frame[k]).powi(2))
.sum::<f64>()
/ n_cmp as f64;
let ref_energy = mean_amp * mean_amp + 1e-9;
(diff_energy / ref_energy).sqrt().clamp(0.0, 1.0)
} else {
0.0
}
} else {
(intra_variance / (mean_amp * mean_amp + 1e-9))
.sqrt()
.clamp(0.0, 1.0)
};
let variance_motion = (temporal_variance / 10.0).clamp(0.0, 1.0);
let mbp_motion = (motion_band_power / 25.0).clamp(0.0, 1.0);
let cp_motion = (change_points as f64 / 15.0).clamp(0.0, 1.0);
let motion_score = (temporal_motion_score * 0.4
+ variance_motion * 0.2
+ mbp_motion * 0.25
+ cp_motion * 0.15)
.clamp(0.0, 1.0);
let snr_db = (frame.rssi as f64 - frame.noise_floor as f64).max(0.0);
let snr_quality = (snr_db / 40.0).clamp(0.0, 1.0); let stability =
(1.0 - (temporal_variance / (mean_amp * mean_amp + 1e-9)).clamp(0.0, 1.0)).max(0.0);
let signal_quality = (snr_quality * 0.6 + stability * 0.4).clamp(0.0, 1.0);
let breathing_rate_hz = estimate_breathing_rate_hz(frame_history, sample_rate_hz);
let features = FeatureInfo {
mean_rssi,
variance,
motion_band_power,
breathing_band_power,
dominant_freq_hz,
change_points,
spectral_power,
};
let raw_classification = ClassificationInfo {
motion_level: raw_classify(motion_score),
presence: motion_score > 0.04,
confidence: (0.4 + signal_quality * 0.3 + motion_score * 0.3).clamp(0.0, 1.0),
};
(
features,
raw_classification,
breathing_rate_hz,
sub_variances,
motion_score,
)
}
fn raw_classify(score: f64) -> String {
if score > 0.25 {
"active".into()
} else if score > 0.12 {
"present_moving".into()
} else if score > 0.04 {
"present_still".into()
} else {
"absent".into()
}
}
const DEBOUNCE_FRAMES: u32 = 4;
const MOTION_EMA_ALPHA: f64 = 0.15;
const BASELINE_EMA_ALPHA: f64 = 0.003;
const BASELINE_WARMUP: u64 = 50;
fn smooth_and_classify(state: &mut AppStateInner, raw: &mut ClassificationInfo, raw_motion: f64) {
state.baseline_frames += 1;
if state.baseline_frames < BASELINE_WARMUP {
state.baseline_motion = state.baseline_motion * 0.9 + raw_motion * 0.1;
} else if raw_motion < state.smoothed_motion + 0.05 {
state.baseline_motion =
state.baseline_motion * (1.0 - BASELINE_EMA_ALPHA) + raw_motion * BASELINE_EMA_ALPHA;
}
let adjusted = (raw_motion - state.baseline_motion * 0.7).max(0.0);
state.smoothed_motion =
state.smoothed_motion * (1.0 - MOTION_EMA_ALPHA) + adjusted * MOTION_EMA_ALPHA;
let sm = state.smoothed_motion;
let candidate = raw_classify(sm);
if candidate == state.current_motion_level {
state.debounce_counter = 0;
state.debounce_candidate = candidate;
} else if candidate == state.debounce_candidate {
state.debounce_counter += 1;
if state.debounce_counter >= DEBOUNCE_FRAMES {
state.current_motion_level = candidate;
state.debounce_counter = 0;
}
} else {
state.debounce_candidate = candidate;
state.debounce_counter = 1;
}
raw.motion_level = state.current_motion_level.clone();
raw.presence = sm > 0.03;
raw.confidence = (0.4 + sm * 0.6).clamp(0.0, 1.0);
}
fn smooth_and_classify_node(ns: &mut NodeState, raw: &mut ClassificationInfo, raw_motion: f64) {
ns.baseline_frames += 1;
if ns.baseline_frames < BASELINE_WARMUP {
ns.baseline_motion = ns.baseline_motion * 0.9 + raw_motion * 0.1;
} else if raw_motion < ns.smoothed_motion + 0.05 {
ns.baseline_motion =
ns.baseline_motion * (1.0 - BASELINE_EMA_ALPHA) + raw_motion * BASELINE_EMA_ALPHA;
}
let adjusted = (raw_motion - ns.baseline_motion * 0.7).max(0.0);
ns.smoothed_motion =
ns.smoothed_motion * (1.0 - MOTION_EMA_ALPHA) + adjusted * MOTION_EMA_ALPHA;
let sm = ns.smoothed_motion;
let candidate = raw_classify(sm);
if candidate == ns.current_motion_level {
ns.debounce_counter = 0;
ns.debounce_candidate = candidate;
} else if candidate == ns.debounce_candidate {
ns.debounce_counter += 1;
if ns.debounce_counter >= DEBOUNCE_FRAMES {
ns.current_motion_level = candidate;
ns.debounce_counter = 0;
}
} else {
ns.debounce_candidate = candidate;
ns.debounce_counter = 1;
}
raw.motion_level = ns.current_motion_level.clone();
raw.presence = sm > 0.03;
raw.confidence = (0.4 + sm * 0.6).clamp(0.0, 1.0);
}
fn adaptive_override(
state: &AppStateInner,
features: &FeatureInfo,
classification: &mut ClassificationInfo,
) {
if let Some(ref model) = state.adaptive_model {
let amps = state
.frame_history
.back()
.map(|v| v.as_slice())
.unwrap_or(&[]);
let feat_arr = adaptive_classifier::features_from_runtime(
&serde_json::json!({
"variance": features.variance,
"motion_band_power": features.motion_band_power,
"breathing_band_power": features.breathing_band_power,
"spectral_power": features.spectral_power,
"dominant_freq_hz": features.dominant_freq_hz,
"change_points": features.change_points,
"mean_rssi": features.mean_rssi,
}),
amps,
);
let (label, conf) = model.classify(&feat_arr);
classification.motion_level = label.to_string();
classification.presence = label != "absent";
classification.confidence = (conf * 0.7 + classification.confidence * 0.3).clamp(0.0, 1.0);
}
}
const VITAL_MEDIAN_WINDOW: usize = 21;
const VITAL_EMA_ALPHA: f64 = 0.02;
const HR_MAX_JUMP: f64 = 8.0;
const BR_MAX_JUMP: f64 = 2.0;
const HR_DEAD_BAND: f64 = 2.0;
const BR_DEAD_BAND: f64 = 0.5;
fn smooth_vitals(state: &mut AppStateInner, raw: &VitalSigns) -> VitalSigns {
let raw_hr = raw.heart_rate_bpm.unwrap_or(0.0);
let raw_br = raw.breathing_rate_bpm.unwrap_or(0.0);
let hr_ok = state.smoothed_hr < 1.0 || (raw_hr - state.smoothed_hr).abs() < HR_MAX_JUMP;
let br_ok = state.smoothed_br < 1.0 || (raw_br - state.smoothed_br).abs() < BR_MAX_JUMP;
if hr_ok && raw_hr > 0.0 {
state.hr_buffer.push_back(raw_hr);
if state.hr_buffer.len() > VITAL_MEDIAN_WINDOW {
state.hr_buffer.pop_front();
}
}
if br_ok && raw_br > 0.0 {
state.br_buffer.push_back(raw_br);
if state.br_buffer.len() > VITAL_MEDIAN_WINDOW {
state.br_buffer.pop_front();
}
}
let trimmed_hr = trimmed_mean(&state.hr_buffer);
let trimmed_br = trimmed_mean(&state.br_buffer);
if trimmed_hr > 0.0 {
if state.smoothed_hr < 1.0 {
state.smoothed_hr = trimmed_hr;
} else if (trimmed_hr - state.smoothed_hr).abs() > HR_DEAD_BAND {
state.smoothed_hr =
state.smoothed_hr * (1.0 - VITAL_EMA_ALPHA) + trimmed_hr * VITAL_EMA_ALPHA;
}
}
if trimmed_br > 0.0 {
if state.smoothed_br < 1.0 {
state.smoothed_br = trimmed_br;
} else if (trimmed_br - state.smoothed_br).abs() > BR_DEAD_BAND {
state.smoothed_br =
state.smoothed_br * (1.0 - VITAL_EMA_ALPHA) + trimmed_br * VITAL_EMA_ALPHA;
}
}
state.smoothed_hr_conf = state.smoothed_hr_conf * 0.92 + raw.heartbeat_confidence * 0.08;
state.smoothed_br_conf = state.smoothed_br_conf * 0.92 + raw.breathing_confidence * 0.08;
VitalSigns {
breathing_rate_bpm: if state.smoothed_br > 1.0 {
Some(state.smoothed_br)
} else {
None
},
heart_rate_bpm: if state.smoothed_hr > 1.0 {
Some(state.smoothed_hr)
} else {
None
},
breathing_confidence: state.smoothed_br_conf,
heartbeat_confidence: state.smoothed_hr_conf,
signal_quality: raw.signal_quality,
}
}
fn smooth_vitals_node(ns: &mut NodeState, raw: &VitalSigns) -> VitalSigns {
let raw_hr = raw.heart_rate_bpm.unwrap_or(0.0);
let raw_br = raw.breathing_rate_bpm.unwrap_or(0.0);
let hr_ok = ns.smoothed_hr < 1.0 || (raw_hr - ns.smoothed_hr).abs() < HR_MAX_JUMP;
let br_ok = ns.smoothed_br < 1.0 || (raw_br - ns.smoothed_br).abs() < BR_MAX_JUMP;
if hr_ok && raw_hr > 0.0 {
ns.hr_buffer.push_back(raw_hr);
if ns.hr_buffer.len() > VITAL_MEDIAN_WINDOW {
ns.hr_buffer.pop_front();
}
}
if br_ok && raw_br > 0.0 {
ns.br_buffer.push_back(raw_br);
if ns.br_buffer.len() > VITAL_MEDIAN_WINDOW {
ns.br_buffer.pop_front();
}
}
let trimmed_hr = trimmed_mean(&ns.hr_buffer);
let trimmed_br = trimmed_mean(&ns.br_buffer);
if trimmed_hr > 0.0 {
if ns.smoothed_hr < 1.0 {
ns.smoothed_hr = trimmed_hr;
} else if (trimmed_hr - ns.smoothed_hr).abs() > HR_DEAD_BAND {
ns.smoothed_hr =
ns.smoothed_hr * (1.0 - VITAL_EMA_ALPHA) + trimmed_hr * VITAL_EMA_ALPHA;
}
}
if trimmed_br > 0.0 {
if ns.smoothed_br < 1.0 {
ns.smoothed_br = trimmed_br;
} else if (trimmed_br - ns.smoothed_br).abs() > BR_DEAD_BAND {
ns.smoothed_br =
ns.smoothed_br * (1.0 - VITAL_EMA_ALPHA) + trimmed_br * VITAL_EMA_ALPHA;
}
}
ns.smoothed_hr_conf = ns.smoothed_hr_conf * 0.92 + raw.heartbeat_confidence * 0.08;
ns.smoothed_br_conf = ns.smoothed_br_conf * 0.92 + raw.breathing_confidence * 0.08;
VitalSigns {
breathing_rate_bpm: if ns.smoothed_br > 1.0 {
Some(ns.smoothed_br)
} else {
None
},
heart_rate_bpm: if ns.smoothed_hr > 1.0 {
Some(ns.smoothed_hr)
} else {
None
},
breathing_confidence: ns.smoothed_br_conf,
heartbeat_confidence: ns.smoothed_hr_conf,
signal_quality: raw.signal_quality,
}
}
fn trimmed_mean(buf: &VecDeque<f64>) -> f64 {
if buf.is_empty() {
return 0.0;
}
let mut sorted: Vec<f64> = buf.iter().copied().collect();
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let n = sorted.len();
let trim = n / 4; let middle = &sorted[trim..n - trim.max(0)];
if middle.is_empty() {
sorted[n / 2] } else {
middle.iter().sum::<f64>() / middle.len() as f64
}
}
fn parse_netsh_interfaces_output(output: &str) -> Option<(f64, f64, String)> {
let mut rssi = None;
let mut signal = None;
let mut ssid = None;
for line in output.lines() {
let line = line.trim();
if line.starts_with("Signal") {
if let Some(pct) = line.split(':').nth(1) {
let pct = pct.trim().trim_end_matches('%');
if let Ok(v) = pct.parse::<f64>() {
signal = Some(v);
rssi = Some(-100.0 + v * 0.6);
}
}
}
if line.starts_with("SSID") && !line.starts_with("BSSID") {
if let Some(s) = line.split(':').nth(1) {
ssid = Some(s.trim().to_string());
}
}
}
match (rssi, signal, ssid) {
(Some(r), Some(_s), Some(name)) => Some((r, _s, name)),
(Some(r), Some(_s), None) => Some((r, _s, "Unknown".into())),
_ => None,
}
}
async fn windows_wifi_task(state: SharedState, tick_ms: u64) {
let mut interval = tokio::time::interval(Duration::from_millis(tick_ms));
let mut seq: u32 = 0;
let mut registry = BssidRegistry::new(32, 30);
let mut pipeline = WindowsWifiPipeline::new();
info!(
"Windows WiFi multi-BSSID pipeline active (tick={}ms, max_bssids=32)",
tick_ms
);
loop {
interval.tick().await;
seq += 1;
let bssid_scan_result = tokio::task::spawn_blocking(|| {
let output = std::process::Command::new("netsh")
.args(["wlan", "show", "networks", "mode=bssid"])
.output()
.map_err(|e| format!("netsh bssid scan failed: {e}"))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
return Err(format!(
"netsh exited with {}: {}",
output.status,
stderr.trim()
));
}
let stdout = String::from_utf8_lossy(&output.stdout);
parse_netsh_bssid_output(&stdout).map_err(|e| format!("parse error: {e}"))
})
.await;
let observations = match bssid_scan_result {
Ok(Ok(obs)) if !obs.is_empty() => obs,
Ok(Ok(_empty)) => {
debug!("Multi-BSSID scan returned 0 observations, falling back");
windows_wifi_fallback_tick(&state, seq).await;
continue;
}
Ok(Err(e)) => {
warn!("Multi-BSSID scan error: {e}, falling back");
windows_wifi_fallback_tick(&state, seq).await;
continue;
}
Err(join_err) => {
error!("spawn_blocking panicked: {join_err}");
continue;
}
};
let obs_count = observations.len();
let ssid = observations
.first()
.map(|o| o.ssid.clone())
.unwrap_or_else(|| "Unknown".into());
registry.update(&observations);
let multi_ap_frame = registry.to_multi_ap_frame();
let enhanced = pipeline.process(&multi_ap_frame);
let first_rssi = observations.first().map(|o| o.rssi_dbm).unwrap_or(-80.0);
let _first_signal_pct = observations.first().map(|o| o.signal_pct).unwrap_or(40.0);
let frame = Esp32Frame {
magic: 0xC511_0001,
node_id: 0,
n_antennas: 1,
n_subcarriers: obs_count.min(u16::MAX as usize) as u16,
freq_mhz: 2437,
sequence: seq,
rssi: first_rssi.clamp(-128.0, 127.0) as i8,
noise_floor: -90,
ppdu_type: wifi_densepose_hardware::PpduType::HtLegacy,
amplitudes: multi_ap_frame.amplitudes.clone(),
phases: multi_ap_frame.phases.clone(),
};
let mut s_write_pre = state.write().await;
s_write_pre
.frame_history
.push_back(frame.amplitudes.clone());
if s_write_pre.frame_history.len() > FRAME_HISTORY_CAPACITY {
s_write_pre.frame_history.pop_front();
}
let sample_rate_hz = 1000.0 / tick_ms as f64;
let (features, mut classification, breathing_rate_hz, sub_variances, raw_motion) =
extract_features_from_frame(&frame, &s_write_pre.frame_history, sample_rate_hz);
smooth_and_classify(&mut s_write_pre, &mut classification, raw_motion);
adaptive_override(&s_write_pre, &features, &mut classification);
drop(s_write_pre);
let enhanced_motion = Some(serde_json::json!({
"score": enhanced.motion.score,
"level": format!("{:?}", enhanced.motion.level),
"contributing_bssids": enhanced.motion.contributing_bssids,
}));
let enhanced_breathing = enhanced.breathing.as_ref().map(|b| {
serde_json::json!({
"rate_bpm": b.rate_bpm,
"confidence": b.confidence,
"bssid_count": b.bssid_count,
})
});
let posture_str = enhanced.posture.map(|p| format!("{p:?}"));
let sig_quality_score = Some(enhanced.signal_quality.score);
let verdict_str = Some(format!("{:?}", enhanced.verdict));
let bssid_n = Some(enhanced.bssid_count);
let mut s = state.write().await;
s.source = format!("wifi:{ssid}");
s.rssi_history.push_back(first_rssi);
if s.rssi_history.len() > 60 {
s.rssi_history.pop_front();
}
s.tick += 1;
let tick = s.tick;
let motion_score = if classification.motion_level == "active" {
0.8
} else if classification.motion_level == "present_still" {
0.3
} else {
0.05
};
let raw_vitals = s
.vital_detector
.process_frame(&frame.amplitudes, &frame.phases);
let vitals = smooth_vitals(&mut s, &raw_vitals);
s.latest_vitals = vitals.clone();
let feat_variance = features.variance;
s.p95_variance.push(features.variance);
s.p95_motion_band_power.push(features.motion_band_power);
s.p95_spectral_power.push(features.spectral_power);
let raw_score = compute_person_score(&s, &features);
s.smoothed_person_score = s.smoothed_person_score * 0.90 + raw_score * 0.10;
let est_persons = if classification.presence {
let count = s.person_count();
s.prev_person_count = count;
count
} else {
s.prev_person_count = 0;
0
};
let mut update = SensingUpdate {
msg_type: "sensing_update".to_string(),
timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
source: format!("wifi:{ssid}"),
tick,
nodes: vec![NodeInfo {
node_id: 0,
rssi_dbm: first_rssi,
position: [0.0, 0.0, 0.0],
amplitude: multi_ap_frame.amplitudes,
subcarrier_count: obs_count,
sync: None, }],
features,
classification,
signal_field: generate_signal_field(
first_rssi,
motion_score,
breathing_rate_hz,
feat_variance.min(1.0),
&sub_variances,
),
vital_signs: Some(vitals),
enhanced_motion,
enhanced_breathing,
posture: posture_str,
signal_quality_score: sig_quality_score,
quality_verdict: verdict_str,
bssid_count: bssid_n,
pose_keypoints: None,
model_status: None,
persons: None,
estimated_persons: if est_persons > 0 {
Some(est_persons)
} else {
None
},
node_features: None,
};
let raw_persons = derive_pose_from_sensing(&update);
let mut last_tracker_instant = s.last_tracker_instant.take();
let tracked = tracker_bridge::tracker_update(
&mut s.pose_tracker,
&mut last_tracker_instant,
raw_persons,
);
s.last_tracker_instant = last_tracker_instant;
if !tracked.is_empty() {
update.persons = Some(tracked);
}
if let Ok(json) = serde_json::to_string(&update) {
let _ = s.tx.send(json);
}
s.latest_update = Some(update);
debug!(
"Multi-BSSID tick #{tick}: {obs_count} BSSIDs, quality={:.2}, verdict={:?}",
enhanced.signal_quality.score, enhanced.verdict
);
}
}
async fn windows_wifi_fallback_tick(state: &SharedState, seq: u32) {
let output = match tokio::process::Command::new("netsh")
.args(["wlan", "show", "interfaces"])
.output()
.await
{
Ok(o) => String::from_utf8_lossy(&o.stdout).to_string(),
Err(e) => {
warn!("netsh interfaces fallback failed: {e}");
return;
}
};
let (rssi_dbm, signal_pct, ssid) = match parse_netsh_interfaces_output(&output) {
Some(v) => v,
None => {
debug!("Fallback: no WiFi interface connected");
return;
}
};
let frame = Esp32Frame {
magic: 0xC511_0001,
node_id: 0,
n_antennas: 1,
n_subcarriers: 1,
freq_mhz: 2437,
sequence: seq,
rssi: rssi_dbm as i8,
noise_floor: -90,
ppdu_type: wifi_densepose_hardware::PpduType::HtLegacy,
amplitudes: vec![signal_pct],
phases: vec![0.0],
};
let mut s = state.write().await;
s.frame_history.push_back(frame.amplitudes.clone());
if s.frame_history.len() > FRAME_HISTORY_CAPACITY {
s.frame_history.pop_front();
}
let sample_rate_hz = 2.0_f64; let (features, mut classification, breathing_rate_hz, sub_variances, raw_motion) =
extract_features_from_frame(&frame, &s.frame_history, sample_rate_hz);
smooth_and_classify(&mut s, &mut classification, raw_motion);
adaptive_override(&s, &features, &mut classification);
s.source = format!("wifi:{ssid}");
s.rssi_history.push_back(rssi_dbm);
if s.rssi_history.len() > 60 {
s.rssi_history.pop_front();
}
s.tick += 1;
let tick = s.tick;
let motion_score = if classification.motion_level == "active" {
0.8
} else if classification.motion_level == "present_still" {
0.3
} else {
0.05
};
let raw_vitals = s
.vital_detector
.process_frame(&frame.amplitudes, &frame.phases);
let vitals = smooth_vitals(&mut s, &raw_vitals);
s.latest_vitals = vitals.clone();
let feat_variance = features.variance;
s.p95_variance.push(features.variance);
s.p95_motion_band_power.push(features.motion_band_power);
s.p95_spectral_power.push(features.spectral_power);
let raw_score = compute_person_score(&s, &features);
s.smoothed_person_score = s.smoothed_person_score * 0.90 + raw_score * 0.10;
let est_persons = if classification.presence {
let count = s.person_count();
s.prev_person_count = count;
count
} else {
s.prev_person_count = 0;
0
};
let mut update = SensingUpdate {
msg_type: "sensing_update".to_string(),
timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
source: format!("wifi:{ssid}"),
tick,
nodes: vec![NodeInfo {
node_id: 0,
rssi_dbm,
position: [0.0, 0.0, 0.0],
amplitude: vec![signal_pct],
subcarrier_count: 1,
sync: None, }],
features,
classification,
signal_field: generate_signal_field(
rssi_dbm,
motion_score,
breathing_rate_hz,
feat_variance.min(1.0),
&sub_variances,
),
vital_signs: Some(vitals),
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,
estimated_persons: if est_persons > 0 {
Some(est_persons)
} else {
None
},
node_features: None,
};
let raw_persons = derive_pose_from_sensing(&update);
let mut last_tracker_instant = s.last_tracker_instant.take();
let tracked =
tracker_bridge::tracker_update(&mut s.pose_tracker, &mut last_tracker_instant, raw_persons);
s.last_tracker_instant = last_tracker_instant;
if !tracked.is_empty() {
update.persons = Some(tracked);
}
if let Ok(json) = serde_json::to_string(&update) {
let _ = s.tx.send(json);
}
s.latest_update = Some(update);
}
async fn probe_windows_wifi() -> bool {
match tokio::process::Command::new("netsh")
.args(["wlan", "show", "interfaces"])
.output()
.await
{
Ok(o) => {
let out = String::from_utf8_lossy(&o.stdout);
parse_netsh_interfaces_output(&out).is_some()
}
Err(_) => false,
}
}
async fn probe_esp32(port: u16) -> bool {
let addr = format!("0.0.0.0:{port}");
match UdpSocket::bind(&addr).await {
Ok(sock) => {
let mut buf = [0u8; 2048];
match tokio::time::timeout(Duration::from_secs(2), sock.recv_from(&mut buf)).await {
Ok(Ok((len, _))) => parse_esp32_frame(&buf[..len]).is_some(),
_ => false,
}
}
Err(_) => false,
}
}
fn generate_simulated_frame(tick: u64) -> Esp32Frame {
let t = tick as f64 * 0.1;
let n_sub = 56usize;
let mut amplitudes = Vec::with_capacity(n_sub);
let mut phases = Vec::with_capacity(n_sub);
for i in 0..n_sub {
let base = 15.0 + 5.0 * (i as f64 * 0.1 + t * 0.3).sin();
let noise = (i as f64 * 7.3 + t * 13.7).sin() * 2.0;
amplitudes.push((base + noise).max(0.1));
phases.push((i as f64 * 0.2 + t * 0.5).sin() * std::f64::consts::PI);
}
Esp32Frame {
magic: 0xC511_0001,
node_id: 1,
n_antennas: 1,
n_subcarriers: n_sub as u16,
freq_mhz: 2437,
sequence: tick as u32,
rssi: (-40.0 + 5.0 * (t * 0.2).sin()) as i8,
noise_floor: -90,
ppdu_type: wifi_densepose_hardware::PpduType::HtLegacy,
amplitudes,
phases,
}
}
async fn ws_sensing_handler(
ws: WebSocketUpgrade,
State(state): State<SharedState>,
) -> impl IntoResponse {
ws.on_upgrade(|socket| handle_ws_client(socket, state))
}
async fn handle_ws_client(mut socket: WebSocket, state: SharedState) {
let mut rx = {
let s = state.read().await;
s.tx.subscribe()
};
info!("WebSocket client connected (sensing)");
let mut ping_interval = tokio::time::interval(std::time::Duration::from_secs(30));
ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tokio::select! {
msg = rx.recv() => {
match msg {
Ok(json) => {
if socket.send(Message::Text(json)).await.is_err() {
break;
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
tracing::debug!("WS client lagged by {n} frames, skipping");
continue;
}
Err(_) => break, }
}
_ = ping_interval.tick() => {
if socket.send(Message::Ping(vec![])).await.is_err() {
break;
}
}
msg = socket.recv() => {
match msg {
Some(Ok(Message::Close(_))) | None => break,
Some(Ok(Message::Pong(_))) => {} _ => {} }
}
}
}
info!("WebSocket client disconnected (sensing)");
}
async fn ws_introspection_handler(
ws: WebSocketUpgrade,
State(state): State<SharedState>,
) -> impl IntoResponse {
ws.on_upgrade(|socket| handle_ws_introspection_client(socket, state))
}
async fn handle_ws_introspection_client(mut socket: WebSocket, state: SharedState) {
let mut rx = {
let s = state.read().await;
s.intro_tx.subscribe()
};
info!("WebSocket client connected (introspection)");
loop {
tokio::select! {
msg = rx.recv() => {
match msg {
Ok(json) => {
if socket.send(Message::Text(json)).await.is_err() {
break;
}
}
Err(_) => break,
}
}
msg = socket.recv() => {
match msg {
Some(Ok(Message::Close(_))) | None => break,
_ => {} }
}
}
}
info!("WebSocket client disconnected (introspection)");
}
async fn api_introspection_snapshot(State(state): State<SharedState>) -> impl IntoResponse {
let s = state.read().await;
Json(s.intro.snapshot().clone())
}
async fn ws_pose_handler(
ws: WebSocketUpgrade,
State(state): State<SharedState>,
) -> impl IntoResponse {
ws.on_upgrade(|socket| handle_ws_pose_client(socket, state))
}
async fn handle_ws_pose_client(mut socket: WebSocket, state: SharedState) {
let mut rx = {
let s = state.read().await;
s.tx.subscribe()
};
info!("WebSocket client connected (pose)");
let conn_msg = serde_json::json!({
"type": "connection_established",
"payload": { "status": "connected", "backend": "rust+ruvector" }
});
let _ = socket.send(Message::Text(conn_msg.to_string())).await;
loop {
tokio::select! {
msg = rx.recv() => {
match msg {
Ok(json) => {
if let Ok(sensing) = serde_json::from_str::<SensingUpdate>(&json) {
if sensing.msg_type == "sensing_update" {
let model_loaded = {
let s = state.read().await;
s.model_loaded
};
let pose_source = if model_loaded {
"model_inference"
} else {
"signal_derived"
};
let persons = if model_loaded {
sensing.pose_keypoints.as_ref().map(|kps| {
let kp_names = [
"nose","left_eye","right_eye","left_ear","right_ear",
"left_shoulder","right_shoulder","left_elbow","right_elbow",
"left_wrist","right_wrist","left_hip","right_hip",
"left_knee","right_knee","left_ankle","right_ankle",
];
let keypoints: Vec<PoseKeypoint> = kps.iter()
.enumerate()
.map(|(i, kp)| PoseKeypoint {
name: kp_names.get(i).unwrap_or(&"unknown").to_string(),
x: kp[0], y: kp[1], z: kp[2], confidence: kp[3],
})
.collect();
vec![PersonDetection {
id: 1,
confidence: sensing.classification.confidence,
bbox: BoundingBox { x: 260.0, y: 150.0, width: 120.0, height: 220.0 },
keypoints,
zone: "zone_1".into(),
}]
}).unwrap_or_else(|| {
sensing.persons.clone().unwrap_or_else(|| derive_pose_from_sensing(&sensing))
})
} else {
sensing.persons.clone().unwrap_or_else(|| derive_pose_from_sensing(&sensing))
};
let pose_msg = serde_json::json!({
"type": "pose_data",
"zone_id": "zone_1",
"timestamp": sensing.timestamp,
"payload": {
"pose": {
"persons": persons,
},
"confidence": if sensing.classification.presence { sensing.classification.confidence } else { 0.0 },
"activity": sensing.classification.motion_level,
"pose_source": pose_source,
"metadata": {
"frame_id": format!("rust_frame_{}", sensing.tick),
"processing_time_ms": 1,
"source": sensing.source,
"tick": sensing.tick,
"signal_strength": sensing.features.mean_rssi,
"motion_band_power": sensing.features.motion_band_power,
"breathing_band_power": sensing.features.breathing_band_power,
"estimated_persons": persons.len(),
}
}
});
if socket.send(Message::Text(pose_msg.to_string())).await.is_err() {
break;
}
}
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
tracing::debug!("WS pose client lagged by {n} frames, skipping");
continue;
}
Err(_) => break, }
}
msg = socket.recv() => {
match msg {
Some(Ok(Message::Text(text))) => {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&text) {
if v.get("type").and_then(|t| t.as_str()) == Some("ping") {
let pong = serde_json::json!({"type": "pong"});
let _ = socket.send(Message::Text(pong.to_string())).await;
}
}
}
Some(Ok(Message::Close(_))) | None => break,
Some(Ok(Message::Pong(_))) => {} _ => {}
}
}
}
}
info!("WebSocket client disconnected (pose)");
}
async fn health(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"status": "ok",
"source": s.effective_source(),
"tick": s.tick,
"clients": s.tx.receiver_count(),
}))
}
async fn latest(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.latest_update {
Some(update) => Json(serde_json::to_value(update).unwrap_or_default()),
None => Json(serde_json::json!({"status": "no data yet"})),
}
}
fn fuse_multi_node_features(
current_features: &FeatureInfo,
node_states: &HashMap<u8, NodeState>,
) -> FeatureInfo {
let now = std::time::Instant::now();
let active: Vec<(&FeatureInfo, f64)> = node_states
.values()
.filter(|ns| {
ns.last_frame_time
.is_some_and(|t| now.duration_since(t).as_secs() < 10)
})
.filter_map(|ns| {
let feat = ns.latest_features.as_ref()?;
let rssi = ns.rssi_history.back().copied().unwrap_or(-80.0);
Some((feat, rssi))
})
.collect();
if active.len() <= 1 {
return current_features.clone();
}
let max_rssi = active
.iter()
.map(|(_, r)| *r)
.fold(f64::NEG_INFINITY, f64::max);
let weights: Vec<f64> = active
.iter()
.map(|(_, r)| (1.0 + (r - max_rssi + 20.0) / 20.0).clamp(0.1, 1.0))
.collect();
let w_sum: f64 = weights.iter().sum::<f64>().max(1e-9);
FeatureInfo {
variance: active
.iter()
.zip(&weights)
.map(|((f, _), w)| f.variance * w)
.sum::<f64>()
/ w_sum,
motion_band_power: active
.iter()
.zip(&weights)
.map(|((f, _), w)| f.motion_band_power * w)
.sum::<f64>()
/ w_sum,
breathing_band_power: active
.iter()
.zip(&weights)
.map(|((f, _), w)| f.breathing_band_power * w)
.sum::<f64>()
/ w_sum,
spectral_power: active
.iter()
.zip(&weights)
.map(|((f, _), w)| f.spectral_power * w)
.sum::<f64>()
/ w_sum,
dominant_freq_hz: active
.iter()
.zip(&weights)
.map(|((f, _), w)| f.dominant_freq_hz * w)
.sum::<f64>()
/ w_sum,
change_points: current_features.change_points, mean_rssi: active
.iter()
.map(|(f, _)| f.mean_rssi)
.fold(f64::NEG_INFINITY, f64::max),
}
}
fn compute_person_score(state: &AppStateInner, feat: &FeatureInfo) -> f64 {
let var_denom = state
.p95_variance
.current()
.map(|p| p.max(50.0))
.unwrap_or(300.0);
let motion_denom = state
.p95_motion_band_power
.current()
.map(|p| p.max(50.0))
.unwrap_or(250.0);
let sp_denom = state
.p95_spectral_power
.current()
.map(|p| p.max(100.0))
.unwrap_or(500.0);
let var_norm = (feat.variance / var_denom).clamp(0.0, 1.0);
let cp_norm = (feat.change_points as f64 / 30.0).clamp(0.0, 1.0);
let motion_norm = (feat.motion_band_power / motion_denom).clamp(0.0, 1.0);
let sp_norm = (feat.spectral_power / sp_denom).clamp(0.0, 1.0);
var_norm * 0.40 + cp_norm * 0.20 + motion_norm * 0.25 + sp_norm * 0.15
}
fn estimate_persons_from_correlation(frame_history: &VecDeque<Vec<f64>>) -> usize {
let n_frames = frame_history.len();
if n_frames < 10 {
return 1;
}
let window: Vec<&Vec<f64>> = frame_history.iter().rev().take(20).collect();
let n_sub = window[0].len().min(56);
if n_sub < 4 {
return 1;
}
let k = window.len() as f64;
let mut means = vec![0.0f64; n_sub];
let mut variances = vec![0.0f64; n_sub];
for frame in &window {
for sc in 0..n_sub.min(frame.len()) {
means[sc] += frame[sc] / k;
}
}
for frame in &window {
for sc in 0..n_sub.min(frame.len()) {
variances[sc] += (frame[sc] - means[sc]).powi(2) / k;
}
}
let noise_floor = 1.0;
let active: Vec<usize> = (0..n_sub)
.filter(|&sc| variances[sc] > noise_floor)
.collect();
let m = active.len();
if m < 3 {
return if m == 0 { 0 } else { 1 };
}
let mut edges: Vec<(u64, u64, f64)> = Vec::new();
let source = m as u64;
let sink = (m + 1) as u64;
let stds: Vec<f64> = active
.iter()
.map(|&sc| variances[sc].sqrt().max(1e-9))
.collect();
for i in 0..m {
for j in (i + 1)..m {
let mut cov = 0.0f64;
for frame in &window {
let si = active[i];
let sj = active[j];
if si < frame.len() && sj < frame.len() {
cov += (frame[si] - means[si]) * (frame[sj] - means[sj]) / k;
}
}
let corr = (cov / (stds[i] * stds[j])).abs();
if corr > 0.1 {
let weight = corr * 10.0; edges.push((i as u64, j as u64, weight));
edges.push((j as u64, i as u64, weight));
}
}
}
let (max_var_idx, _) = active
.iter()
.enumerate()
.max_by(|(_, &a), (_, &b)| {
variances[a]
.partial_cmp(&variances[b])
.unwrap_or(std::cmp::Ordering::Equal)
})
.unwrap_or((0, &0));
let (min_var_idx, _) = active
.iter()
.enumerate()
.min_by(|(_, &a), (_, &b)| {
variances[a]
.partial_cmp(&variances[b])
.unwrap_or(std::cmp::Ordering::Equal)
})
.unwrap_or((0, &0));
if max_var_idx == min_var_idx {
return 1;
}
edges.push((source, max_var_idx as u64, 100.0));
edges.push((min_var_idx as u64, sink, 100.0));
let mc: DynamicMinCut = match MinCutBuilder::new()
.exact()
.with_edges(edges.clone())
.build()
{
Ok(mc) => mc,
Err(_) => return 1,
};
let cut_value = mc.min_cut_value();
let total_edge_weight: f64 = edges
.iter()
.filter(|(s, t, _)| *s != source && *s != sink && *t != source && *t != sink)
.map(|(_, _, w)| w)
.sum::<f64>()
/ 2.0;
if total_edge_weight < 1e-9 {
return 1;
}
let cut_ratio = cut_value / total_edge_weight;
if cut_ratio > 0.4 {
1 } else if cut_ratio > 0.15 {
2 } else {
3 }
}
fn corr_persons_to_score(corr_persons: usize) -> f64 {
match corr_persons {
0 => 0.20,
1 => 0.40,
2 => 0.74,
_ => 0.96,
}
}
#[cfg(test)]
mod corr_persons_round_trip_tests {
use super::*;
fn converge(corr_persons: usize) -> usize {
let mut score = 0.0f64;
let mut count = 1usize;
for _ in 0..400 {
let target = corr_persons_to_score(corr_persons);
score = score * 0.92 + target * 0.08;
count = score_to_person_count(score, count);
}
count
}
#[test]
fn sustained_one_person_estimate_reports_one() {
assert_eq!(converge(1), 1);
}
#[test]
fn sustained_two_person_estimate_reports_two() {
assert_eq!(converge(2), 2, "#803: min-cut=2 must round-trip to count 2");
}
#[test]
fn sustained_three_person_estimate_reports_three() {
assert_eq!(converge(3), 3);
}
#[test]
fn old_div3_mapping_would_pin_two_people_to_one() {
let mut score = 0.0f64;
let mut count = 1usize;
for _ in 0..400 {
score = score * 0.92 + (2.0 / 3.0) * 0.08;
count = score_to_person_count(score, count);
}
assert_eq!(count, 1, "old corr_persons/3.0 mapping was the #803 bug");
}
}
fn score_to_person_count(smoothed_score: f64, prev_count: usize) -> usize {
match prev_count {
0 | 1 => {
if smoothed_score > 0.85 {
3
} else if smoothed_score > 0.70 {
2
} else {
1
}
}
2 => {
if smoothed_score > 0.92 {
3
} else if smoothed_score < 0.55 {
1
} else {
2 }
}
_ => {
if smoothed_score < 0.55 {
1
} else if smoothed_score < 0.78 {
2
} else {
3 }
}
}
}
fn aggregate_person_count(
activity_count: usize,
node_states: &std::collections::HashMap<u8, NodeState>,
) -> usize {
let node_max = node_states
.values()
.map(|n| n.prev_person_count)
.max()
.unwrap_or(0);
activity_count.max(node_max)
}
#[cfg(test)]
mod aggregate_person_count_tests {
use super::*;
use std::collections::HashMap;
fn node_with_count(c: usize) -> NodeState {
let mut n = NodeState::new();
n.prev_person_count = c;
n
}
#[test]
fn empty_nodes_fall_back_to_activity_count() {
let nodes: HashMap<u8, NodeState> = HashMap::new();
assert_eq!(aggregate_person_count(1, &nodes), 1);
assert_eq!(aggregate_person_count(0, &nodes), 0);
}
#[test]
fn node_estimate_raises_a_saturated_activity_count() {
let mut nodes = HashMap::new();
nodes.insert(1u8, node_with_count(2));
assert_eq!(
aggregate_person_count(1, &nodes),
2,
"a node reporting 2 must not be discarded by the activity count"
);
}
#[test]
fn activity_count_wins_when_higher_than_nodes() {
let mut nodes = HashMap::new();
nodes.insert(1u8, node_with_count(1));
assert_eq!(aggregate_person_count(3, &nodes), 3);
}
#[test]
fn takes_max_across_multiple_nodes() {
let mut nodes = HashMap::new();
nodes.insert(1u8, node_with_count(1));
nodes.insert(2u8, node_with_count(3));
nodes.insert(3u8, node_with_count(2));
assert_eq!(aggregate_person_count(1, &nodes), 3);
}
#[test]
fn single_occupant_is_never_inflated() {
let mut nodes = HashMap::new();
nodes.insert(1u8, node_with_count(1));
nodes.insert(2u8, node_with_count(1));
assert_eq!(aggregate_person_count(1, &nodes), 1);
}
}
fn derive_single_person_pose(
update: &SensingUpdate,
person_idx: usize,
total_persons: usize,
) -> PersonDetection {
let cls = &update.classification;
let feat = &update.features;
let phase_offset = person_idx as f64 * 2.094;
let half = (total_persons as f64 - 1.0) / 2.0;
let person_x_offset = (person_idx as f64 - half) * 120.0;
let conf_decay = 1.0 - person_idx as f64 * 0.15;
let motion_score = (feat.motion_band_power / 15.0).clamp(0.0, 1.0);
let is_walking = motion_score > 0.55;
let breath_amp = (feat.breathing_band_power * 4.0).clamp(0.0, 12.0);
let breath_phase = if let Some(ref vs) = update.vital_signs {
let bpm = vs.breathing_rate_bpm.unwrap_or(15.0);
let freq = (bpm / 60.0).clamp(0.1, 0.5);
(update.tick as f64 * freq * 0.02 * std::f64::consts::TAU + phase_offset).sin()
} else {
(update.tick as f64 * 0.02 + phase_offset).sin()
};
let lean_x = (feat.dominant_freq_hz / 5.0 - 1.0).clamp(-1.0, 1.0) * 18.0;
let stride_x = if is_walking {
let stride_phase =
(feat.motion_band_power * 0.7 + update.tick as f64 * 0.06 + phase_offset).sin();
stride_phase * 20.0 * motion_score
} else {
0.0
};
let burst = (feat.change_points as f64 / 20.0).clamp(0.0, 0.3);
let noise_seed = person_idx as f64 * 97.1; let noise_val = (noise_seed.sin() * 43758.545).fract();
let snr_factor = ((feat.variance - 0.5) / 10.0).clamp(0.0, 1.0);
let base_confidence = cls.confidence * (0.6 + 0.4 * snr_factor) * conf_decay;
let base_x = 320.0 + stride_x + lean_x * 0.5 + person_x_offset;
let base_y = 240.0 - motion_score * 8.0;
let kp_names = [
"nose",
"left_eye",
"right_eye",
"left_ear",
"right_ear",
"left_shoulder",
"right_shoulder",
"left_elbow",
"right_elbow",
"left_wrist",
"right_wrist",
"left_hip",
"right_hip",
"left_knee",
"right_knee",
"left_ankle",
"right_ankle",
];
let kp_offsets: [(f64, f64); 17] = [
(0.0, -80.0), (-8.0, -88.0), (8.0, -88.0), (-16.0, -82.0), (16.0, -82.0), (-30.0, -50.0), (30.0, -50.0), (-45.0, -15.0), (45.0, -15.0), (-50.0, 20.0), (50.0, 20.0), (-20.0, 20.0), (20.0, 20.0), (-22.0, 70.0), (22.0, 70.0), (-24.0, 120.0), (24.0, 120.0), ];
const TORSO_KP: [usize; 4] = [5, 6, 11, 12];
const EXTREMITY_KP: [usize; 4] = [9, 10, 15, 16];
let keypoints: Vec<PoseKeypoint> = kp_names
.iter()
.zip(kp_offsets.iter())
.enumerate()
.map(|(i, (name, (dx, dy)))| {
let breath_dx = if TORSO_KP.contains(&i) {
let sign = if *dx < 0.0 { -1.0 } else { 1.0 };
sign * breath_amp * breath_phase * 0.5
} else {
0.0
};
let breath_dy = if TORSO_KP.contains(&i) {
let sign = if *dy < 0.0 { -1.0 } else { 1.0 };
sign * breath_amp * breath_phase * 0.3
} else {
0.0
};
let extremity_jitter = if EXTREMITY_KP.contains(&i) {
let phase = noise_seed + i as f64 * 2.399;
(
phase.sin() * burst * motion_score * 4.0,
(phase * 1.31).cos() * burst * motion_score * 3.0,
)
} else {
(0.0, 0.0)
};
let kp_noise_x = ((noise_seed + i as f64 * 1.618).sin() * 43758.545).fract()
* feat.variance.sqrt().clamp(0.0, 3.0)
* motion_score;
let kp_noise_y = ((noise_seed + i as f64 * std::f64::consts::E).cos() * 31415.926)
.fract()
* feat.variance.sqrt().clamp(0.0, 3.0)
* motion_score
* 0.6;
let swing_dy = if is_walking {
let stride_phase =
(feat.motion_band_power * 0.7 + update.tick as f64 * 0.12 + phase_offset).sin();
match i {
7 | 9 => -stride_phase * 20.0 * motion_score,
8 | 10 => stride_phase * 20.0 * motion_score,
13 | 15 => stride_phase * 25.0 * motion_score,
14 | 16 => -stride_phase * 25.0 * motion_score,
_ => 0.0,
}
} else {
0.0
};
let final_x = base_x + dx + breath_dx + extremity_jitter.0 + kp_noise_x;
let final_y = base_y + dy + breath_dy + extremity_jitter.1 + kp_noise_y + swing_dy;
let kp_conf = if EXTREMITY_KP.contains(&i) {
base_confidence * (0.7 + 0.3 * snr_factor) * (0.85 + 0.15 * noise_val)
} else {
base_confidence * (0.88 + 0.12 * ((i as f64 * 0.7 + noise_seed).cos()))
};
PoseKeypoint {
name: name.to_string(),
x: final_x,
y: final_y,
z: lean_x * 0.02,
confidence: kp_conf.clamp(0.1, 1.0),
}
})
.collect();
let xs: Vec<f64> = keypoints.iter().map(|k| k.x).collect();
let ys: Vec<f64> = keypoints.iter().map(|k| k.y).collect();
let min_x = xs.iter().cloned().fold(f64::MAX, f64::min) - 10.0;
let min_y = ys.iter().cloned().fold(f64::MAX, f64::min) - 10.0;
let max_x = xs.iter().cloned().fold(f64::MIN, f64::max) + 10.0;
let max_y = ys.iter().cloned().fold(f64::MIN, f64::max) + 10.0;
PersonDetection {
id: (person_idx + 1) as u32,
confidence: cls.confidence * conf_decay,
keypoints,
bbox: BoundingBox {
x: min_x,
y: min_y,
width: (max_x - min_x).max(80.0),
height: (max_y - min_y).max(160.0),
},
zone: format!("zone_{}", person_idx + 1),
}
}
fn derive_pose_from_sensing(update: &SensingUpdate) -> Vec<PersonDetection> {
let cls = &update.classification;
if !cls.presence {
return vec![];
}
let person_count = update.estimated_persons.unwrap_or(1).max(1);
(0..person_count)
.map(|idx| derive_single_person_pose(update, idx, person_count))
.collect()
}
const POSE_BONE_PAIRS: &[(usize, usize)] = &[
(5, 7),
(7, 9),
(6, 8),
(8, 10), (5, 11),
(6, 12), (11, 13),
(13, 15),
(12, 14),
(14, 16), (5, 6),
(11, 12), ];
fn apply_temporal_smoothing(persons: &mut [PersonDetection], ns: &mut NodeState) {
if persons.is_empty() {
return;
}
let alpha = ns.ema_alpha();
let person = &mut persons[0];
let current_kps: Vec<[f64; 3]> = person
.keypoints
.iter()
.map(|kp| [kp.x, kp.y, kp.z])
.collect();
let smoothed = if let Some(ref prev) = ns.prev_keypoints {
let mut out = Vec::with_capacity(current_kps.len());
for (cur, prv) in current_kps.iter().zip(prev.iter()) {
out.push([
alpha * cur[0] + (1.0 - alpha) * prv[0],
alpha * cur[1] + (1.0 - alpha) * prv[1],
alpha * cur[2] + (1.0 - alpha) * prv[2],
]);
}
clamp_bone_lengths_f64(&mut out, prev);
out
} else {
current_kps.clone()
};
for (kp, s) in person.keypoints.iter_mut().zip(smoothed.iter()) {
kp.x = s[0];
kp.y = s[1];
kp.z = s[2];
}
ns.prev_keypoints = Some(smoothed);
}
fn clamp_bone_lengths_f64(pose: &mut [[f64; 3]], prev: &[[f64; 3]]) {
for &(p, c) in POSE_BONE_PAIRS {
if p >= pose.len() || c >= pose.len() {
continue;
}
let prev_len = dist_f64(&prev[p], &prev[c]);
if prev_len < 1e-6 {
continue;
}
let cur_len = dist_f64(&pose[p], &pose[c]);
if cur_len < 1e-6 {
continue;
}
let ratio = cur_len / prev_len;
let lo = 1.0 - MAX_BONE_CHANGE_RATIO;
let hi = 1.0 + MAX_BONE_CHANGE_RATIO;
if ratio < lo || ratio > hi {
let target = prev_len * ratio.clamp(lo, hi);
let scale = target / cur_len;
for dim in 0..3 {
let diff = pose[c][dim] - pose[p][dim];
pose[c][dim] = pose[p][dim] + diff * scale;
}
}
}
}
fn dist_f64(a: &[f64; 3], b: &[f64; 3]) -> f64 {
let dx = b[0] - a[0];
let dy = b[1] - a[1];
let dz = b[2] - a[2];
(dx * dx + dy * dy + dz * dz).sqrt()
}
async fn health_live(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"status": "alive",
"uptime": s.start_time.elapsed().as_secs(),
}))
}
fn witness_hex(w: [u8; 32]) -> String {
use std::fmt::Write;
w.iter().fold(String::with_capacity(64), |mut acc, b| {
let _ = write!(acc, "{b:02x}");
acc
})
}
async fn health_ready(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"status": "ready",
"source": s.effective_source(),
"trust": {
"last_witness": s.engine_bridge.last_trust_witness().map(witness_hex),
"effective_class": s.engine_bridge.effective_class().map(|c| format!("{c:?}")),
"demoted": s.engine_bridge.demoted(),
"recalibration_recommended": s.engine_bridge.recalibration_recommended(),
"engine_error_count": s.engine_bridge.engine_error_count(),
"raw_outputs_suppressed": s.engine_bridge.suppress_raw_outputs(),
},
}))
}
async fn health_system(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
let uptime = s.start_time.elapsed().as_secs();
Json(serde_json::json!({
"status": "healthy",
"components": {
"api": { "status": "healthy", "message": "Rust Axum server" },
"hardware": {
"status": if s.effective_source().ends_with(":offline") { "degraded" } else { "healthy" },
"message": format!("Source: {}", s.effective_source())
},
"pose": { "status": "healthy", "message": "WiFi-derived pose estimation" },
"stream": { "status": if s.tx.receiver_count() > 0 { "healthy" } else { "idle" },
"message": format!("{} client(s)", s.tx.receiver_count()) },
},
"metrics": {
"cpu_percent": 2.5,
"memory_percent": 1.8,
"disk_percent": 15.0,
"uptime_seconds": uptime,
}
}))
}
async fn health_version() -> Json<serde_json::Value> {
Json(serde_json::json!({
"version": env!("CARGO_PKG_VERSION"),
"name": "wifi-densepose-sensing-server",
"backend": "rust+axum+ruvector",
}))
}
async fn health_metrics(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"system_metrics": {
"cpu": { "percent": 2.5 },
"memory": { "percent": 1.8, "used_mb": 5 },
"disk": { "percent": 15.0 },
},
"tick": s.tick,
}))
}
async fn api_info(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"version": env!("CARGO_PKG_VERSION"),
"environment": "production",
"backend": "rust",
"source": s.effective_source(),
"features": {
"wifi_sensing": true,
"pose_estimation": true,
"signal_processing": true,
"ruvector": true,
"streaming": true,
}
}))
}
async fn pose_current(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
let persons = match &s.latest_update {
Some(update) => update
.persons
.clone()
.unwrap_or_else(|| derive_pose_from_sensing(update)),
None => vec![],
};
Json(serde_json::json!({
"timestamp": chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
"persons": persons,
"total_persons": persons.len(),
"source": s.effective_source(),
}))
}
async fn pose_stats(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"total_detections": s.total_detections,
"average_confidence": 0.87,
"frames_processed": s.tick,
"source": s.effective_source(),
}))
}
async fn pose_zones_summary(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
let presence = s
.latest_update
.as_ref()
.map(|u| u.classification.presence)
.unwrap_or(false);
Json(serde_json::json!({
"zones": {
"zone_1": { "person_count": if presence { 1 } else { 0 }, "status": "monitored" },
"zone_2": { "person_count": 0, "status": "clear" },
"zone_3": { "person_count": 0, "status": "clear" },
"zone_4": { "person_count": 0, "status": "clear" },
}
}))
}
async fn stream_status(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"active": true,
"clients": s.tx.receiver_count(),
"fps": if s.tick > 1 { 10u64 } else { 0u64 },
"source": s.effective_source(),
}))
}
async fn list_models(State(state): State<SharedState>) -> Json<serde_json::Value> {
let models = scan_model_files();
let total = models.len();
{
let mut s = state.write().await;
s.discovered_models = models.clone();
}
Json(serde_json::json!({ "models": models, "total": total }))
}
async fn get_active_model(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.active_model_id {
Some(id) => {
let model = s
.discovered_models
.iter()
.find(|m| m.get("id").and_then(|v| v.as_str()) == Some(id.as_str()));
Json(serde_json::json!({
"active": model.cloned().unwrap_or_else(|| serde_json::json!({ "id": id })),
}))
}
None => Json(serde_json::json!({ "active": serde_json::Value::Null })),
}
}
async fn load_model(
State(state): State<SharedState>,
Json(body): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
let model_id = body
.get("id")
.or_else(|| body.get("model_id"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if model_id.is_empty() {
return Json(serde_json::json!({ "error": "missing 'id' field", "success": false }));
}
let mut s = state.write().await;
s.active_model_id = Some(model_id.clone());
s.model_loaded = true;
info!("Model loaded: {model_id}");
Json(serde_json::json!({ "success": true, "model_id": model_id }))
}
async fn unload_model(State(state): State<SharedState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
let prev = s.active_model_id.take();
s.model_loaded = false;
info!("Model unloaded (was: {:?})", prev);
Json(serde_json::json!({ "success": true, "previous": prev }))
}
async fn delete_model(
State(state): State<SharedState>,
Path(id): Path<String>,
) -> Json<serde_json::Value> {
let safe_id = std::path::Path::new(&id)
.file_name()
.and_then(|f| f.to_str())
.unwrap_or("");
if safe_id.is_empty() || safe_id != id {
return Json(serde_json::json!({ "error": "invalid model id", "success": false }));
}
let path = effective_models_dir().join(format!("{}.rvf", safe_id));
if path.exists() {
if let Err(e) = std::fs::remove_file(&path) {
warn!("Failed to delete model file {:?}: {}", path, e);
return Json(
serde_json::json!({ "error": format!("delete failed: {e}"), "success": false }),
);
}
let mut s = state.write().await;
if s.active_model_id.as_deref() == Some(id.as_str()) {
s.active_model_id = None;
s.model_loaded = false;
}
s.discovered_models
.retain(|m| m.get("id").and_then(|v| v.as_str()) != Some(id.as_str()));
info!("Model deleted: {id}");
Json(serde_json::json!({ "success": true, "deleted": id }))
} else {
Json(serde_json::json!({ "error": "model not found", "success": false }))
}
}
async fn list_lora_profiles() -> Json<serde_json::Value> {
let profiles = scan_lora_profiles();
Json(serde_json::json!({ "profiles": profiles }))
}
async fn activate_lora_profile(Json(body): Json<serde_json::Value>) -> Json<serde_json::Value> {
let profile = body
.get("profile")
.or_else(|| body.get("name"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if profile.is_empty() {
return Json(serde_json::json!({ "error": "missing 'profile' field", "success": false }));
}
info!("LoRA profile activated: {profile}");
Json(serde_json::json!({ "success": true, "profile": profile }))
}
fn effective_models_dir() -> PathBuf {
PathBuf::from(std::env::var("MODELS_DIR").unwrap_or_else(|_| "data/models".to_string()))
}
fn scan_model_files() -> Vec<serde_json::Value> {
let dir = effective_models_dir();
let mut models = Vec::new();
if let Ok(entries) = std::fs::read_dir(&dir) {
for entry in entries.flatten() {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) == Some("rvf") {
let name = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown")
.to_string();
let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
let modified = entry
.metadata()
.ok()
.and_then(|m| m.modified().ok())
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs())
.unwrap_or(0);
models.push(serde_json::json!({
"id": name,
"name": name,
"path": path.display().to_string(),
"size_bytes": size,
"format": "rvf",
"modified_epoch": modified,
}));
}
}
}
models
}
fn scan_lora_profiles() -> Vec<serde_json::Value> {
let dir = effective_models_dir();
let mut profiles = Vec::new();
if let Ok(entries) = std::fs::read_dir(&dir) {
for entry in entries.flatten() {
let path = entry.path();
let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
if name.ends_with(".lora.json") {
let profile_name = name.trim_end_matches(".lora.json").to_string();
let config = std::fs::read_to_string(&path)
.ok()
.and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
.unwrap_or_else(|| serde_json::json!({}));
profiles.push(serde_json::json!({
"name": profile_name,
"path": path.display().to_string(),
"config": config,
}));
}
}
}
profiles
}
async fn list_recordings() -> Json<serde_json::Value> {
let recordings = scan_recording_files();
Json(serde_json::json!({ "recordings": recordings }))
}
async fn start_recording(
State(state): State<SharedState>,
Json(body): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
let mut s = state.write().await;
if s.recording_active {
return Json(serde_json::json!({
"error": "recording already in progress",
"success": false,
"recording_id": s.recording_current_id,
}));
}
let id = body
.get("id")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
.unwrap_or_else(|| format!("rec_{}", chrono_timestamp()));
let rec_path = PathBuf::from("data/recordings").join(format!("{}.jsonl", id));
let file = match std::fs::File::create(&rec_path) {
Ok(f) => f,
Err(e) => {
warn!("Failed to create recording file {:?}: {}", rec_path, e);
return Json(serde_json::json!({
"error": format!("cannot create file: {e}"),
"success": false,
}));
}
};
let (stop_tx, mut stop_rx) = tokio::sync::watch::channel(false);
s.recording_active = true;
s.recording_start_time = Some(std::time::Instant::now());
s.recording_current_id = Some(id.clone());
s.recording_stop_tx = Some(stop_tx);
let mut rx = s.tx.subscribe();
s.recordings.push(serde_json::json!({
"id": id,
"path": rec_path.display().to_string(),
"status": "recording",
"started_at": chrono_timestamp(),
"frames": 0,
}));
let rec_id = id.clone();
tokio::spawn(async move {
use std::io::Write;
let mut writer = std::io::BufWriter::new(file);
let mut frame_count: u64 = 0;
loop {
tokio::select! {
result = rx.recv() => {
match result {
Ok(frame_json) => {
if writeln!(writer, "{}", frame_json).is_err() {
warn!("Recording {rec_id}: write error, stopping");
break;
}
frame_count += 1;
if frame_count % 100 == 0 {
let _ = writer.flush();
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
debug!("Recording {rec_id}: lagged {n} frames");
}
Err(broadcast::error::RecvError::Closed) => {
info!("Recording {rec_id}: broadcast closed, stopping");
break;
}
}
}
_ = stop_rx.changed() => {
if *stop_rx.borrow() {
info!("Recording {rec_id}: stop signal received ({frame_count} frames)");
break;
}
}
}
}
let _ = writer.flush();
info!("Recording {rec_id} finished: {frame_count} frames written");
});
info!("Recording started: {id}");
Json(serde_json::json!({ "success": true, "recording_id": id }))
}
async fn stop_recording(State(state): State<SharedState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
if !s.recording_active {
return Json(serde_json::json!({
"error": "no recording in progress",
"success": false,
}));
}
if let Some(tx) = s.recording_stop_tx.take() {
let _ = tx.send(true);
}
let duration_secs = s
.recording_start_time
.map(|t| t.elapsed().as_secs())
.unwrap_or(0);
let rec_id = s.recording_current_id.take().unwrap_or_default();
s.recording_active = false;
s.recording_start_time = None;
for rec in s.recordings.iter_mut() {
if rec.get("id").and_then(|v| v.as_str()) == Some(rec_id.as_str()) {
rec["status"] = serde_json::json!("completed");
rec["duration_secs"] = serde_json::json!(duration_secs);
}
}
info!("Recording stopped: {rec_id} ({duration_secs}s)");
Json(serde_json::json!({
"success": true,
"recording_id": rec_id,
"duration_secs": duration_secs,
}))
}
async fn delete_recording(
State(state): State<SharedState>,
Path(id): Path<String>,
) -> Json<serde_json::Value> {
let safe_id = std::path::Path::new(&id)
.file_name()
.and_then(|f| f.to_str())
.unwrap_or("");
if safe_id.is_empty() || safe_id != id {
return Json(serde_json::json!({ "error": "invalid recording id", "success": false }));
}
let path = PathBuf::from("data/recordings").join(format!("{}.jsonl", safe_id));
if path.exists() {
if let Err(e) = std::fs::remove_file(&path) {
warn!("Failed to delete recording {:?}: {}", path, e);
return Json(
serde_json::json!({ "error": format!("delete failed: {e}"), "success": false }),
);
}
let mut s = state.write().await;
s.recordings
.retain(|r| r.get("id").and_then(|v| v.as_str()) != Some(id.as_str()));
info!("Recording deleted: {id}");
Json(serde_json::json!({ "success": true, "deleted": id }))
} else {
Json(serde_json::json!({ "error": "recording not found", "success": false }))
}
}
fn scan_recording_files() -> Vec<serde_json::Value> {
let dir = PathBuf::from("data/recordings");
let mut recordings = Vec::new();
if let Ok(entries) = std::fs::read_dir(&dir) {
for entry in entries.flatten() {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
let name = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown")
.to_string();
let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
let modified = entry
.metadata()
.ok()
.and_then(|m| m.modified().ok())
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_secs())
.unwrap_or(0);
let frame_count = std::fs::read_to_string(&path)
.map(|s| s.lines().count())
.unwrap_or(0);
recordings.push(serde_json::json!({
"id": name,
"name": name,
"path": path.display().to_string(),
"size_bytes": size,
"frames": frame_count,
"modified_epoch": modified,
"status": "completed",
}));
}
}
}
recordings
}
async fn train_status(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"status": s.training_status,
"config": s.training_config,
}))
}
async fn train_start(
State(state): State<SharedState>,
Json(body): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
let mut s = state.write().await;
if s.training_status == "running" {
return Json(serde_json::json!({
"error": "training already running",
"success": false,
}));
}
s.training_status = "running".to_string();
s.training_config = Some(body.clone());
info!("Training started with config: {}", body);
Json(serde_json::json!({
"success": true,
"status": "running",
"message": "Training pipeline started. Use GET /api/v1/train/status to monitor.",
}))
}
async fn train_stop(State(state): State<SharedState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
if s.training_status != "running" {
return Json(serde_json::json!({
"error": "no training in progress",
"success": false,
}));
}
s.training_status = "idle".to_string();
info!("Training stopped");
Json(serde_json::json!({
"success": true,
"status": "idle",
}))
}
async fn adaptive_train(State(state): State<SharedState>) -> Json<serde_json::Value> {
let rec_dir = PathBuf::from("data/recordings");
eprintln!("=== Adaptive Classifier Training ===");
match adaptive_classifier::train_from_recordings(&rec_dir) {
Ok(model) => {
let accuracy = model.training_accuracy;
let frames = model.trained_frames;
let stats: Vec<_> = model
.class_stats
.iter()
.map(|cs| {
serde_json::json!({
"class": cs.label,
"samples": cs.count,
"feature_means": cs.mean,
})
})
.collect();
if let Err(e) = model.save(&adaptive_classifier::model_path()) {
warn!("Failed to save adaptive model: {e}");
} else {
info!(
"Adaptive model saved to {}",
adaptive_classifier::model_path().display()
);
}
let mut s = state.write().await;
s.adaptive_model = Some(model);
Json(serde_json::json!({
"success": true,
"trained_frames": frames,
"accuracy": accuracy,
"class_stats": stats,
}))
}
Err(e) => Json(serde_json::json!({
"success": false,
"error": e,
})),
}
}
async fn adaptive_status(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.adaptive_model {
Some(model) => Json(serde_json::json!({
"loaded": true,
"trained_frames": model.trained_frames,
"accuracy": model.training_accuracy,
"version": model.version,
"classes": model.class_names,
"class_stats": model.class_stats,
})),
None => Json(serde_json::json!({
"loaded": false,
"message": "No adaptive model. POST /api/v1/adaptive/train to train one.",
})),
}
}
async fn adaptive_unload(State(state): State<SharedState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
s.adaptive_model = None;
Json(serde_json::json!({ "success": true, "message": "Adaptive model unloaded." }))
}
async fn calibration_start(State(state): State<SharedState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
if let Some(ref fm) = s.field_model {
match fm.status() {
CalibrationStatus::Collecting => {
return Json(serde_json::json!({
"success": false,
"error": "Calibration already in progress. Call /calibration/stop first.",
"frame_count": fm.calibration_frame_count(),
}));
}
CalibrationStatus::Fresh => {
return Json(serde_json::json!({
"success": false,
"error": "A fresh calibration already exists. Call /calibration/stop or wait for expiry.",
}));
}
_ => {} }
}
match FieldModel::new(field_bridge::single_link_config()) {
Ok(fm) => {
s.field_model = Some(fm);
Json(serde_json::json!({
"success": true,
"message": "Calibration started — keep room empty while frames accumulate.",
}))
}
Err(e) => Json(serde_json::json!({
"success": false,
"error": format!("{e}"),
})),
}
}
async fn calibration_stop(State(state): State<SharedState>) -> Json<serde_json::Value> {
let mut s = state.write().await;
if let Some(ref mut fm) = s.field_model {
let ts = chrono::Utc::now().timestamp_micros() as u64;
match fm.finalize_calibration(ts, 0) {
Ok(modes) => {
let baseline = modes.baseline_eigenvalue_count;
let variance_explained = modes.variance_explained;
info!("Field model calibrated: baseline_eigenvalues={baseline}, variance_explained={variance_explained:.2}");
Json(serde_json::json!({
"success": true,
"baseline_eigenvalue_count": baseline,
"variance_explained": variance_explained,
"frame_count": fm.calibration_frame_count(),
}))
}
Err(e) => Json(serde_json::json!({
"success": false,
"error": format!("{e}"),
})),
}
} else {
Json(serde_json::json!({
"success": false,
"error": "No field model active — call /calibration/start first.",
}))
}
}
async fn calibration_status(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match s.field_model.as_ref() {
Some(fm) => Json(serde_json::json!({
"active": true,
"status": format!("{:?}", fm.status()),
"frame_count": fm.calibration_frame_count(),
})),
None => Json(serde_json::json!({
"active": false,
"status": "none",
})),
}
}
fn chrono_timestamp() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
async fn vital_signs_endpoint(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
let vs = &s.latest_vitals;
let (br_len, br_cap, hb_len, hb_cap) = s.vital_detector.buffer_status();
Json(serde_json::json!({
"vital_signs": {
"breathing_rate_bpm": vs.breathing_rate_bpm,
"heart_rate_bpm": vs.heart_rate_bpm,
"breathing_confidence": vs.breathing_confidence,
"heartbeat_confidence": vs.heartbeat_confidence,
"signal_quality": vs.signal_quality,
},
"buffer_status": {
"breathing_samples": br_len,
"breathing_capacity": br_cap,
"heartbeat_samples": hb_len,
"heartbeat_capacity": hb_cap,
},
"source": s.effective_source(),
"tick": s.tick,
}))
}
#[derive(Debug, Deserialize)]
struct EdgeRegistryParams {
#[serde(default)]
refresh: Option<String>,
}
async fn edge_registry_endpoint(
Extension(reg): Extension<
Option<Arc<wifi_densepose_sensing_server::edge_registry::EdgeRegistry>>,
>,
Query(params): Query<EdgeRegistryParams>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> {
let Some(reg) = reg else {
return Err((
StatusCode::NOT_FOUND,
Json(serde_json::json!({
"error": "edge_registry_disabled",
"detail": "This sensing-server was started with --no-edge-registry."
})),
));
};
let force_refresh = matches!(params.refresh.as_deref(), Some("1") | Some("true"));
if force_refresh {
tracing::debug!(
event = "edge_registry.refresh_requested",
"?refresh=1 bypassed the cache; verify this isn't being abused"
);
}
match tokio::task::spawn_blocking(move || reg.get(force_refresh)).await {
Ok(Ok(resp)) => Ok(Json(
serde_json::to_value(resp).unwrap_or(serde_json::json!({})),
)),
Ok(Err(err)) => {
tracing::warn!(error = %err, "edge_registry upstream fetch failed and no cache");
Err((
StatusCode::SERVICE_UNAVAILABLE,
Json(serde_json::json!({
"error": "edge_registry_upstream_unavailable",
"detail": err.to_string()
})),
))
}
Err(join_err) => {
tracing::error!(error = %join_err, "edge_registry spawn_blocking task panicked");
Err((
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({
"error": "edge_registry_internal_error",
"detail": join_err.to_string()
})),
))
}
}
}
async fn edge_vitals_endpoint(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.edge_vitals {
Some(v) => Json(serde_json::json!({
"status": "ok",
"edge_vitals": v,
})),
None => Json(serde_json::json!({
"status": "no_data",
"edge_vitals": null,
"message": "No edge vitals packet received yet. Ensure ESP32 edge_tier >= 1.",
})),
}
}
async fn wasm_events_endpoint(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.latest_wasm_events {
Some(w) => Json(serde_json::json!({
"status": "ok",
"wasm_events": w,
})),
None => Json(serde_json::json!({
"status": "no_data",
"wasm_events": null,
"message": "No WASM output packet received yet. Upload and start a .wasm module on the ESP32.",
})),
}
}
async fn model_info(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.rvf_info {
Some(info) => Json(serde_json::json!({
"status": "loaded",
"container": info,
})),
None => Json(serde_json::json!({
"status": "no_model",
"message": "No RVF container loaded. Use --load-rvf <path> to load one.",
})),
}
}
async fn model_layers(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.progressive_loader {
Some(loader) => {
let (a, b, c) = loader.layer_status();
Json(serde_json::json!({
"layer_a": a,
"layer_b": b,
"layer_c": c,
"progress": loader.loading_progress(),
}))
}
None => Json(serde_json::json!({
"layer_a": false,
"layer_b": false,
"layer_c": false,
"progress": 0.0,
"message": "No model loaded with progressive loading",
})),
}
}
async fn model_segments(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
match &s.progressive_loader {
Some(loader) => Json(serde_json::json!({ "segments": loader.segment_list() })),
None => Json(serde_json::json!({ "segments": [] })),
}
}
async fn sona_profiles(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
let names = s
.progressive_loader
.as_ref()
.map(|l| l.sona_profile_names())
.unwrap_or_default();
let active = s.active_sona_profile.clone().unwrap_or_default();
Json(serde_json::json!({ "profiles": names, "active": active }))
}
async fn sona_activate(
State(state): State<SharedState>,
Json(body): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
let profile = body
.get("profile")
.and_then(|p| p.as_str())
.unwrap_or("")
.to_string();
let mut s = state.write().await;
let available = s
.progressive_loader
.as_ref()
.map(|l| l.sona_profile_names())
.unwrap_or_default();
if available.contains(&profile) {
s.active_sona_profile = Some(profile.clone());
Json(serde_json::json!({ "status": "activated", "profile": profile }))
} else {
Json(serde_json::json!({
"status": "error",
"message": format!("Profile '{}' not found. Available: {:?}", profile, available),
}))
}
}
async fn node_sync_endpoint(
State(state): State<SharedState>,
Path(id): Path<u8>,
) -> Result<Json<NodeSyncSnapshot>, (StatusCode, Json<serde_json::Value>)> {
let s = state.read().await;
let ns = s.node_states.get(&id).ok_or_else(|| {
(StatusCode::NOT_FOUND, Json(serde_json::json!({
"error": "unknown_node", "node_id": id,
})))
})?;
ns.sync_snapshot().map(Json).ok_or_else(|| {
(StatusCode::NOT_FOUND, Json(serde_json::json!({
"error": "no_sync", "node_id": id,
"hint": "node hasn't emitted a sync packet yet (no mesh peer or not v0.6.9+)",
})))
})
}
async fn mesh_metrics_endpoint(State(state): State<SharedState>) -> impl IntoResponse {
use std::fmt::Write;
let s = state.read().await;
let mut body = String::with_capacity(1024);
let metrics: &[(&str, &str, &str)] = &[
("wifi_densepose_mesh_offset_us",
"Cross-board mesh-aligned offset, microseconds (signed)", "gauge"),
("wifi_densepose_mesh_is_leader",
"1 if this node is the elected mesh leader, else 0", "gauge"),
("wifi_densepose_mesh_is_valid",
"1 if this node has heard a fresh leader beacon, else 0", "gauge"),
("wifi_densepose_mesh_smoothed",
"1 once the firmware-side EMA filter has seeded, else 0", "gauge"),
("wifi_densepose_mesh_sequence",
"High-water CSI sequence at sync emit time", "gauge"),
("wifi_densepose_mesh_csi_fps",
"Per-node measured CSI frame rate (Hz)", "gauge"),
("wifi_densepose_mesh_csi_fps_samples",
"How many inter-frame deltas the fps EMA has seen", "gauge"),
("wifi_densepose_mesh_staleness_ms",
"Milliseconds since the host last received this node's sync packet", "gauge"),
];
let snaps: Vec<(u8, NodeSyncSnapshot)> = s.node_states.iter()
.filter_map(|(&id, ns)| ns.sync_snapshot().map(|snap| (id, snap)))
.collect();
let (leaders, followers) = fleet_role_counts(&snaps);
let no_sync = s.node_states.len().saturating_sub(snaps.len()) as u64;
let _ = writeln!(body,
"# HELP wifi_densepose_mesh_node_total Per-state node count across the fleet");
let _ = writeln!(body, "# TYPE wifi_densepose_mesh_node_total gauge");
let _ = writeln!(body, "wifi_densepose_mesh_node_total{{state=\"leader\"}} {leaders}");
let _ = writeln!(body, "wifi_densepose_mesh_node_total{{state=\"follower\"}} {followers}");
let _ = writeln!(body, "wifi_densepose_mesh_node_total{{state=\"no_sync\"}} {no_sync}");
for (name, help, kind) in metrics {
let _ = writeln!(body, "# HELP {name} {help}");
let _ = writeln!(body, "# TYPE {name} {kind}");
for (id, snap) in &snaps {
let value = match *name {
"wifi_densepose_mesh_offset_us" => snap.offset_us.to_string(),
"wifi_densepose_mesh_is_leader" => bool_metric(snap.is_leader),
"wifi_densepose_mesh_is_valid" => bool_metric(snap.is_valid),
"wifi_densepose_mesh_smoothed" => bool_metric(snap.smoothed),
"wifi_densepose_mesh_sequence" => snap.sequence.to_string(),
"wifi_densepose_mesh_csi_fps" => format!("{:.3}", snap.csi_fps_ema),
"wifi_densepose_mesh_csi_fps_samples" => snap.csi_fps_samples.to_string(),
"wifi_densepose_mesh_staleness_ms" =>
snap.staleness_ms.map(|n| n.to_string()).unwrap_or_else(|| "0".into()),
_ => continue,
};
let _ = writeln!(body, "{name}{{node=\"{id}\"}} {value}");
}
}
([(axum::http::header::CONTENT_TYPE, "text/plain; version=0.0.4")], body)
}
fn bool_metric(b: bool) -> String { (if b { 1 } else { 0 }).to_string() }
pub(crate) fn fleet_role_counts(snaps: &[(u8, NodeSyncSnapshot)]) -> (u64, u64) {
let leaders = snaps.iter().filter(|(_, s)| s.is_leader).count() as u64;
let followers = (snaps.len() as u64).saturating_sub(leaders);
(leaders, followers)
}
async fn mesh_endpoint(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
let mut nodes = serde_json::Map::new();
for (&id, ns) in s.node_states.iter() {
if let Some(snap) = ns.sync_snapshot() {
nodes.insert(id.to_string(), serde_json::to_value(snap).unwrap());
}
}
let total = nodes.len();
Json(serde_json::json!({
"nodes": serde_json::Value::Object(nodes),
"total": total,
}))
}
async fn nodes_endpoint(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
let now = std::time::Instant::now();
let nodes: Vec<serde_json::Value> = s
.node_states
.iter()
.map(|(&id, ns)| {
let elapsed_ms = ns
.last_frame_time
.map(|t| now.duration_since(t).as_millis() as u64)
.unwrap_or(999999);
let stale = elapsed_ms > 5000;
let status = if stale { "stale" } else { "active" };
let rssi = ns.rssi_history.back().copied().unwrap_or(-90.0);
serde_json::json!({
"node_id": id,
"status": status,
"last_seen_ms": elapsed_ms,
"rssi_dbm": rssi,
"motion_level": &ns.current_motion_level,
"person_count": ns.prev_person_count,
})
})
.collect();
Json(serde_json::json!({
"nodes": nodes,
"total": nodes.len(),
}))
}
async fn info_page() -> Html<String> {
Html(
"<html><body>\
<h1>WiFi-DensePose Sensing Server</h1>\
<p>Rust + Axum + RuVector</p>\
<ul>\
<li><a href='/health'>/health</a> — Server health</li>\
<li><a href='/api/v1/sensing/latest'>/api/v1/sensing/latest</a> — Latest sensing data</li>\
<li><a href='/api/v1/vital-signs'>/api/v1/vital-signs</a> — Vital sign estimates (HR/RR)</li>\
<li><a href='/api/v1/model/info'>/api/v1/model/info</a> — RVF model container info</li>\
<li>ws://localhost:8765/ws/sensing — WebSocket stream</li>\
</ul>\
</body></html>"
.to_string()
)
}
async fn udp_receiver_task(state: SharedState, udp_port: u16) {
let addr = format!("0.0.0.0:{udp_port}");
let socket = match UdpSocket::bind(&addr).await {
Ok(s) => {
info!("UDP listening on {addr} for ESP32 CSI frames");
s
}
Err(e) => {
error!("Failed to bind UDP {addr}: {e}");
return;
}
};
let mut buf = [0u8; 2048];
loop {
match socket.recv_from(&mut buf).await {
Ok((len, src)) => {
if let Some(vitals) = parse_esp32_vitals(&buf[..len]) {
debug!(
"ESP32 vitals from {src}: node={} br={:.1} hr={:.1} pres={}",
vitals.node_id,
vitals.breathing_rate_bpm,
vitals.heartrate_bpm,
vitals.presence
);
let mut s = state.write().await;
if let Ok(json) = serde_json::to_string(&serde_json::json!({
"type": "edge_vitals",
"node_id": vitals.node_id,
"presence": vitals.presence,
"fall_detected": vitals.fall_detected,
"motion": vitals.motion,
"breathing_rate_bpm": vitals.breathing_rate_bpm,
"heartrate_bpm": vitals.heartrate_bpm,
"n_persons": vitals.n_persons,
"motion_energy": vitals.motion_energy,
"presence_score": vitals.presence_score,
"rssi": vitals.rssi,
})) {
let _ = s.tx.send(json);
}
s.source = "esp32".to_string();
s.last_esp32_frame = Some(std::time::Instant::now());
let node_id = vitals.node_id;
let ns = s.node_states.entry(node_id).or_insert_with(NodeState::new);
ns.last_frame_time = Some(std::time::Instant::now());
ns.edge_vitals = Some(vitals.clone());
ns.rssi_history.push_back(vitals.rssi as f64);
if ns.rssi_history.len() > 60 {
ns.rssi_history.pop_front();
}
let node_est = if vitals.presence {
(vitals.n_persons as usize).max(1)
} else {
0
};
ns.prev_person_count = node_est;
s.tick += 1;
let tick = s.tick;
let motion_level = if vitals.motion {
"present_moving"
} else if vitals.presence {
"present_still"
} else {
"absent"
};
let motion_score = if vitals.motion {
0.8
} else if vitals.presence {
0.3
} else {
0.05
};
let now = std::time::Instant::now();
let total_persons = if vitals.presence {
let dedup = s.dedup_factor;
let (fused, fallback_count) = multistatic_bridge::fuse_or_fallback(
&s.multistatic_fuser,
&s.node_states,
dedup,
);
match fused {
Some(ref f) => {
let score =
multistatic_bridge::compute_person_score_from_amplitudes(
&f.fused_amplitude,
);
s.smoothed_person_score =
s.smoothed_person_score * 0.90 + score * 0.10;
let count =
aggregate_person_count(s.person_count(), &s.node_states);
s.prev_person_count = count;
count.max(1) }
None => {
aggregate_person_count(fallback_count.unwrap_or(0), &s.node_states)
.max(1)
}
}
} else {
s.prev_person_count = 0;
0
};
{
let sref: &mut AppStateInner = &mut s;
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
sref.engine_bridge.observe_cycle(&sref.node_states, now_ms);
}
if let Some(frame_history) = s
.node_states
.get(&node_id)
.map(|ns| ns.frame_history.clone())
{
if let Some(ref mut fm) = s.field_model {
field_bridge::maybe_feed_calibration(fm, &frame_history);
}
}
let active_nodes: Vec<NodeInfo> = s
.node_states
.iter()
.filter(|(_, n)| {
n.last_frame_time
.is_some_and(|t| now.duration_since(t).as_secs() < 10)
})
.map(|(&id, n)| NodeInfo {
node_id: id,
rssi_dbm: n.rssi_history.back().copied().unwrap_or(0.0),
position: [2.0, 0.0, 1.5],
amplitude: vec![],
subcarrier_count: 0,
sync: n.sync_snapshot(),
})
.collect();
let features = FeatureInfo {
mean_rssi: vitals.rssi as f64,
variance: vitals.motion_energy as f64,
motion_band_power: vitals.motion_energy as f64,
breathing_band_power: if vitals.presence { 0.5 } else { 0.0 },
dominant_freq_hz: vitals.breathing_rate_bpm / 60.0,
change_points: 0,
spectral_power: vitals.motion_energy as f64,
};
if let Some(ns) = s.node_states.get_mut(&node_id) {
ns.latest_features = Some(features.clone());
}
let fused_features = fuse_multi_node_features(&features, &s.node_states);
let mut classification = ClassificationInfo {
motion_level: motion_level.to_string(),
presence: vitals.presence,
confidence: vitals.presence_score as f64,
};
let n_active = s
.node_states
.values()
.filter(|ns| {
ns.last_frame_time
.is_some_and(|t| now.duration_since(t).as_secs() < 10)
})
.count();
if n_active > 1 {
classification.confidence = (classification.confidence
* (1.0 + 0.15 * (n_active as f64 - 1.0)))
.clamp(0.0, 1.0);
}
let signal_field = generate_signal_field(
fused_features.mean_rssi,
motion_score,
vitals.breathing_rate_bpm / 60.0,
(vitals.presence_score as f64).min(1.0),
&[],
);
let mut update = SensingUpdate {
msg_type: "sensing_update".to_string(),
timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
source: "esp32".to_string(),
tick,
nodes: active_nodes,
features: fused_features.clone(),
classification,
signal_field,
vital_signs: Some(VitalSigns {
breathing_rate_bpm: if vitals.breathing_rate_bpm > 0.0 {
Some(vitals.breathing_rate_bpm)
} else {
None
},
heart_rate_bpm: if vitals.heartrate_bpm > 0.0 {
Some(vitals.heartrate_bpm)
} else {
None
},
breathing_confidence: if vitals.presence { 0.7 } else { 0.0 },
heartbeat_confidence: if vitals.presence { 0.7 } else { 0.0 },
signal_quality: vitals.presence_score as f64,
}),
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,
estimated_persons: if total_persons > 0 {
Some(total_persons)
} else {
None
},
node_features: build_node_features(&s.node_states, now),
};
let raw_persons = derive_pose_from_sensing(&update);
let mut last_tracker_instant = s.last_tracker_instant.take();
let tracked = tracker_bridge::tracker_update(
&mut s.pose_tracker,
&mut last_tracker_instant,
raw_persons,
);
s.last_tracker_instant = last_tracker_instant;
if !tracked.is_empty() {
update.persons = Some(tracked);
}
if let Ok(json) = serde_json::to_string(&update) {
let _ = s.tx.send(json);
}
s.latest_update = Some(update);
s.edge_vitals = Some(vitals);
continue;
}
if len >= wifi_densepose_hardware::SYNC_PACKET_SIZE {
let magic = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
if magic == wifi_densepose_hardware::SYNC_PACKET_MAGIC {
match wifi_densepose_hardware::SyncPacket::from_bytes(&buf[..len]) {
Ok(sync) => {
debug!("ESP32 sync from {src}: node={} leader={} valid={} smoothed={} \
seq={} offset_us={}",
sync.node_id, sync.flags.is_leader, sync.flags.is_valid,
sync.flags.smoothed_used, sync.sequence,
sync.local_minus_epoch_us());
let mut s = state.write().await;
let node_id = sync.node_id;
let ns = s.node_states.entry(node_id)
.or_insert_with(NodeState::new);
ns.apply_sync_packet(sync, std::time::Instant::now());
continue;
}
Err(e) => {
debug!("Sync packet decode error from {src}: {e}");
continue;
}
}
}
}
if let Some(fused) = parse_edge_fused_vitals(&buf[..len]) {
debug!(
"Edge fused vitals from {src}: node={} br={:.1} hr={:.1} \
mmwave_targets={} fusion_conf={}",
fused.node_id, fused.breathing_rate_bpm, fused.heartrate_bpm,
fused.mmwave_targets, fused.fusion_confidence,
);
let s = state.write().await;
if let Ok(json) = serde_json::to_string(&serde_json::json!({
"type": "edge_fused_vitals",
"node_id": fused.node_id,
"breathing_rate_bpm": fused.breathing_rate_bpm,
"heartrate_bpm": fused.heartrate_bpm,
"n_persons": fused.n_persons,
"fusion_confidence": fused.fusion_confidence,
"mmwave": {
"hr_bpm": fused.mmwave_hr_bpm,
"br_bpm": fused.mmwave_br_bpm,
"distance_cm": fused.mmwave_distance_cm,
"targets": fused.mmwave_targets,
"confidence": fused.mmwave_confidence,
"type": fused.mmwave_type,
},
"motion_energy": fused.motion_energy,
"presence_score": fused.presence_score,
"timestamp_ms": fused.timestamp_ms,
})) {
let _ = s.tx.send(json);
}
continue;
}
if let Some(wasm_output) = parse_wasm_output(&buf[..len]) {
debug!(
"WASM output from {src}: node={} module={} events={}",
wasm_output.node_id,
wasm_output.module_id,
wasm_output.events.len()
);
let mut s = state.write().await;
if let Ok(json) = serde_json::to_string(&serde_json::json!({
"type": "wasm_event",
"node_id": wasm_output.node_id,
"module_id": wasm_output.module_id,
"events": wasm_output.events,
})) {
let _ = s.tx.send(json);
}
s.latest_wasm_events = Some(wasm_output);
continue;
}
if let Some(frame) = parse_esp32_frame(&buf[..len]) {
debug!(
"ESP32 frame from {src}: node={}, subs={}, seq={}",
frame.node_id, frame.n_subcarriers, frame.sequence
);
let mut s = state.write().await;
s.source = "esp32".to_string();
s.last_esp32_frame = Some(std::time::Instant::now());
let grid_accepted = s
.node_states
.entry(frame.node_id)
.or_insert_with(NodeState::new)
.accept_grid(frame.grid());
if !grid_accepted {
debug!(
"node {}: skipping {}-subcarrier {:?} frame (active grid {:?})",
frame.node_id,
frame.n_subcarriers,
frame.ppdu_type,
s.node_states.get(&frame.node_id).and_then(|ns| ns.active_grid),
);
if let Some(ns) = s.node_states.get_mut(&frame.node_id) {
ns.observe_csi_frame_arrival(std::time::Instant::now());
}
continue;
}
s.frame_history.push_back(frame.amplitudes.clone());
if s.frame_history.len() > FRAME_HISTORY_CAPACITY {
s.frame_history.pop_front();
}
{
let intro_feature = if frame.amplitudes.is_empty() {
0.0
} else {
frame.amplitudes.iter().copied().sum::<f64>()
/ frame.amplitudes.len() as f64
};
let intro_ts_ns = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
let _ = s.intro.update(intro_ts_ns, intro_feature);
if let Ok(intro_json) = serde_json::to_string(s.intro.snapshot()) {
let _ = s.intro_tx.send(intro_json);
}
}
let node_id = frame.node_id;
let adaptive_model_clone = s.adaptive_model.clone();
let ns = s.node_states.entry(node_id).or_insert_with(NodeState::new);
ns.observe_csi_frame_arrival(std::time::Instant::now());
ns.update_novelty(&frame.amplitudes);
ns.frame_history.push_back(frame.amplitudes.clone());
if ns.frame_history.len() > FRAME_HISTORY_CAPACITY {
ns.frame_history.pop_front();
}
let sample_rate_hz = 1000.0 / 500.0_f64;
let (
features,
mut classification,
breathing_rate_hz,
sub_variances,
raw_motion,
) = extract_features_from_frame(&frame, &ns.frame_history, sample_rate_hz);
smooth_and_classify_node(ns, &mut classification, raw_motion);
if let Some(ref model) = adaptive_model_clone {
let amps = ns.frame_history.back().map(|v| v.as_slice()).unwrap_or(&[]);
let feat_arr = adaptive_classifier::features_from_runtime(
&serde_json::json!({
"variance": features.variance,
"motion_band_power": features.motion_band_power,
"breathing_band_power": features.breathing_band_power,
"spectral_power": features.spectral_power,
"dominant_freq_hz": features.dominant_freq_hz,
"change_points": features.change_points,
"mean_rssi": features.mean_rssi,
}),
amps,
);
let (label, conf) = model.classify(&feat_arr);
classification.motion_level = label.to_string();
classification.presence = label != "absent";
classification.confidence =
(conf * 0.7 + classification.confidence * 0.3).clamp(0.0, 1.0);
}
ns.rssi_history.push_back(features.mean_rssi);
if ns.rssi_history.len() > 60 {
ns.rssi_history.pop_front();
}
let raw_vitals = ns
.vital_detector
.process_frame(&frame.amplitudes, &frame.phases);
let vitals = smooth_vitals_node(ns, &raw_vitals);
ns.latest_vitals = vitals.clone();
let corr_persons = estimate_persons_from_correlation(&ns.frame_history);
let raw_score = corr_persons_to_score(corr_persons);
ns.smoothed_person_score = ns.smoothed_person_score * 0.92 + raw_score * 0.08;
if classification.presence {
let count =
score_to_person_count(ns.smoothed_person_score, ns.prev_person_count);
ns.prev_person_count = count;
} else {
ns.prev_person_count = 0;
}
ns.latest_features = Some(features.clone());
s.rssi_history.push_back(features.mean_rssi);
if s.rssi_history.len() > 60 {
s.rssi_history.pop_front();
}
s.latest_vitals = vitals.clone();
let fused_features = fuse_multi_node_features(&features, &s.node_states);
s.tick += 1;
let tick = s.tick;
let motion_score = if classification.motion_level == "active" {
0.8
} else if classification.motion_level == "present_still" {
0.3
} else {
0.05
};
let now = std::time::Instant::now();
let total_persons = if classification.presence {
let dedup = s.dedup_factor;
let (fused, fallback_count) = multistatic_bridge::fuse_or_fallback(
&s.multistatic_fuser,
&s.node_states,
dedup,
);
match fused {
Some(ref f) => {
let score =
multistatic_bridge::compute_person_score_from_amplitudes(
&f.fused_amplitude,
);
s.smoothed_person_score =
s.smoothed_person_score * 0.90 + score * 0.10;
let count =
aggregate_person_count(s.person_count(), &s.node_states);
s.prev_person_count = count;
count.max(1)
}
None => {
aggregate_person_count(fallback_count.unwrap_or(0), &s.node_states)
.max(1)
}
}
} else {
s.prev_person_count = 0;
0
};
{
let sref: &mut AppStateInner = &mut s;
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
sref.engine_bridge.observe_cycle(&sref.node_states, now_ms);
}
if let Some(frame_history) = s
.node_states
.get(&node_id)
.map(|ns| ns.frame_history.clone())
{
if let Some(ref mut fm) = s.field_model {
field_bridge::maybe_feed_calibration(fm, &frame_history);
}
}
let suppress_raw = s.engine_bridge.suppress_raw_outputs();
let active_nodes: Vec<NodeInfo> = s
.node_states
.iter()
.filter(|(_, n)| {
n.last_frame_time
.is_some_and(|t| now.duration_since(t).as_secs() < 10)
})
.map(|(&id, n)| NodeInfo {
node_id: id,
rssi_dbm: n.rssi_history.back().copied().unwrap_or(0.0),
position: [2.0, 0.0, 1.5],
amplitude: if suppress_raw {
vec![]
} else {
n.frame_history
.back()
.map(|a| a.iter().take(56).cloned().collect())
.unwrap_or_default()
},
subcarrier_count: if suppress_raw {
0
} else {
n.frame_history.back().map_or(0, |a| a.len())
},
sync: n.sync_snapshot(),
})
.collect();
let mut update = SensingUpdate {
msg_type: "sensing_update".to_string(),
timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
source: "esp32".to_string(),
tick,
nodes: active_nodes,
features: fused_features.clone(),
classification,
signal_field: generate_signal_field(
fused_features.mean_rssi,
motion_score,
breathing_rate_hz,
fused_features.variance.min(1.0),
&sub_variances,
),
vital_signs: Some(vitals),
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,
estimated_persons: if total_persons > 0 {
Some(total_persons)
} else {
None
},
node_features: build_node_features(&s.node_states, now),
};
let raw_persons = derive_pose_from_sensing(&update);
let mut last_tracker_instant = s.last_tracker_instant.take();
let tracked = tracker_bridge::tracker_update(
&mut s.pose_tracker,
&mut last_tracker_instant,
raw_persons,
);
s.last_tracker_instant = last_tracker_instant;
if !tracked.is_empty() {
update.persons = Some(tracked);
}
if let Ok(json) = serde_json::to_string(&update) {
let _ = s.tx.send(json);
}
s.latest_update = Some(update);
if tick % 100 == 0 {
let stale = Duration::from_secs(60);
let before = s.node_states.len();
s.node_states.retain(|_id, ns| {
ns.last_frame_time
.is_some_and(|t| now.duration_since(t) < stale)
});
let evicted = before - s.node_states.len();
if evicted > 0 {
info!(
"Evicted {} stale node(s), {} active",
evicted,
s.node_states.len()
);
}
}
}
}
Err(e) => {
warn!("UDP recv error: {e}");
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
}
}
async fn simulated_data_task(state: SharedState, tick_ms: u64) {
let mut interval = tokio::time::interval(Duration::from_millis(tick_ms));
info!("Simulated data source active (tick={}ms)", tick_ms);
loop {
interval.tick().await;
let mut s = state.write().await;
s.tick += 1;
let tick = s.tick;
let frame = generate_simulated_frame(tick);
s.frame_history.push_back(frame.amplitudes.clone());
if s.frame_history.len() > FRAME_HISTORY_CAPACITY {
s.frame_history.pop_front();
}
let sample_rate_hz = 1000.0 / tick_ms as f64;
let (features, mut classification, breathing_rate_hz, sub_variances, raw_motion) =
extract_features_from_frame(&frame, &s.frame_history, sample_rate_hz);
smooth_and_classify(&mut s, &mut classification, raw_motion);
adaptive_override(&s, &features, &mut classification);
s.rssi_history.push_back(features.mean_rssi);
if s.rssi_history.len() > 60 {
s.rssi_history.pop_front();
}
let motion_score = if classification.motion_level == "active" {
0.8
} else if classification.motion_level == "present_still" {
0.3
} else {
0.05
};
let raw_vitals = s
.vital_detector
.process_frame(&frame.amplitudes, &frame.phases);
let vitals = smooth_vitals(&mut s, &raw_vitals);
s.latest_vitals = vitals.clone();
let frame_amplitudes = frame.amplitudes.clone();
let frame_n_sub = frame.n_subcarriers;
s.p95_variance.push(features.variance);
s.p95_motion_band_power.push(features.motion_band_power);
s.p95_spectral_power.push(features.spectral_power);
let raw_score = compute_person_score(&s, &features);
s.smoothed_person_score = s.smoothed_person_score * 0.90 + raw_score * 0.10;
let est_persons = if classification.presence {
let count = s.person_count();
s.prev_person_count = count;
count
} else {
s.prev_person_count = 0;
0
};
let mut update = SensingUpdate {
msg_type: "sensing_update".to_string(),
timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
source: "simulated".to_string(),
tick,
nodes: vec![NodeInfo {
node_id: 1,
rssi_dbm: features.mean_rssi,
position: [2.0, 0.0, 1.5],
amplitude: frame_amplitudes,
subcarrier_count: frame_n_sub as usize,
sync: None, }],
features: features.clone(),
classification,
signal_field: generate_signal_field(
features.mean_rssi,
motion_score,
breathing_rate_hz,
features.variance.min(1.0),
&sub_variances,
),
vital_signs: Some(vitals),
enhanced_motion: None,
enhanced_breathing: None,
posture: None,
signal_quality_score: None,
quality_verdict: None,
bssid_count: None,
pose_keypoints: None,
model_status: if s.model_loaded {
Some(serde_json::json!({
"loaded": true,
"layers": s.progressive_loader.as_ref()
.map(|l| { let (a,b,c) = l.layer_status(); a as u8 + b as u8 + c as u8 })
.unwrap_or(0),
"sona_profile": s.active_sona_profile.as_deref().unwrap_or("default"),
}))
} else {
None
},
persons: None,
estimated_persons: if est_persons > 0 {
Some(est_persons)
} else {
None
},
node_features: None,
};
let raw_persons = derive_pose_from_sensing(&update);
let mut last_tracker_instant = s.last_tracker_instant.take();
let tracked = tracker_bridge::tracker_update(
&mut s.pose_tracker,
&mut last_tracker_instant,
raw_persons,
);
s.last_tracker_instant = last_tracker_instant;
if !tracked.is_empty() {
update.persons = Some(tracked);
}
if update.classification.presence {
s.total_detections += 1;
}
if let Ok(json) = serde_json::to_string(&update) {
let _ = s.tx.send(json);
}
s.latest_update = Some(update);
}
}
async fn broadcast_tick_task(state: SharedState, tick_ms: u64) {
let mut interval = tokio::time::interval(Duration::from_millis(tick_ms));
loop {
interval.tick().await;
let s = state.read().await;
if let Some(ref update) = s.latest_update {
if s.tx.receiver_count() > 0 {
let mut tagged = update.clone();
tagged.source = s.effective_source();
if let Ok(json) = serde_json::to_string(&tagged) {
let _ = s.tx.send(json);
}
}
}
}
}
#[cfg(feature = "mqtt")]
fn vitals_snapshots_from_sensing_json(
v: &serde_json::Value,
base_id: &str,
) -> Vec<wifi_densepose_sensing_server::mqtt::state::VitalsSnapshot> {
use wifi_densepose_sensing_server::mqtt::state::VitalsSnapshot;
fn motion_of(level: Option<&str>, fallback: f64) -> f64 {
match level {
Some("none") | Some("still") | Some("idle") | Some("absent") | Some("") => 0.0,
Some(_) => 1.0,
None => fallback,
}
}
let ts = (v["timestamp"].as_f64().unwrap_or(0.0) * 1000.0) as i64;
let vit = &v["vital_signs"];
let breathing = vit["breathing_rate_bpm"].as_f64();
let hr = vit["heart_rate_bpm"].as_f64();
let n_persons = v["persons"]
.as_array()
.map(|a| a.len() as u32)
.or_else(|| v["estimated_persons"].as_u64().map(|x| x as u32))
.unwrap_or(0);
let acls = &v["classification"];
let agg_presence = acls["presence"].as_bool().unwrap_or(false);
let agg_motion = motion_of(acls["motion_level"].as_str(), 0.0);
let agg_conf = acls["confidence"].as_f64().unwrap_or(0.0);
let mk = |node_id: String, presence: bool, motion: f64, conf: f64, rssi: Option<f64>| {
VitalsSnapshot {
node_id,
timestamp_ms: ts,
presence,
motion,
presence_score: if presence { conf.max(0.0) } else { 0.0 },
breathing_rate_bpm: breathing,
heartrate_bpm: hr,
n_persons,
rssi_dbm: rssi,
vital_confidence: conf,
..Default::default()
}
};
match v["nodes"].as_array() {
Some(arr) if !arr.is_empty() => arr
.iter()
.map(|node| {
let n = node["node_id"].as_u64().unwrap_or(0);
let ncls = &node["classification"];
let presence = ncls["presence"].as_bool().unwrap_or(agg_presence);
let motion = motion_of(ncls["motion_level"].as_str(), agg_motion);
let conf = ncls["confidence"].as_f64().unwrap_or(agg_conf);
mk(
format!("{base_id}-node{n}"),
presence,
motion,
conf,
node["rssi_dbm"].as_f64(),
)
})
.collect(),
_ => vec![mk(
base_id.to_string(),
agg_presence,
agg_motion,
agg_conf,
v["nodes"][0]["rssi_dbm"].as_f64(),
)],
}
}
fn diagnose_model_load_error(path: &std::path::Path, data: &[u8], err: &str) -> String {
let name = path
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("")
.to_ascii_lowercase();
let ext = path
.extension()
.and_then(|e| e.to_str())
.unwrap_or("")
.to_ascii_lowercase();
let looks_safetensors = ext == "safetensors" || (data.len() > 9 && data[8] == b'{');
let looks_jsonl =
ext == "jsonl" || name.ends_with(".rvf.jsonl") || data.first() == Some(&b'{');
let looks_quant_bin = ext == "bin" || name.contains("-q");
let kind = if looks_safetensors {
"a safetensors weight file"
} else if looks_jsonl {
"a JSONL manifest, not the binary container"
} else if looks_quant_bin {
"a quantized weight blob (e.g. HuggingFace model-q4.bin)"
} else {
"not an RVF binary container"
};
format!(
"model `{}` could not be loaded: it is {kind}. The --model flag expects an \
RVF binary container (`RVFS` magic 0x52564653) produced by the \
wifi-densepose-train pipeline. The HuggingFace ruvnet/wifi-densepose-pretrained \
files are a different format and encoder architecture, so they do not load \
here directly (issue #894). Continuing with signal heuristics. (loader: {err})",
path.display()
)
}
fn export_emits_placeholder_demo(export_set: bool, train: bool, pretrain: bool) -> bool {
export_set && !train && !pretrain
}
fn coalesce_ui_path(initial: std::path::PathBuf) -> std::path::PathBuf {
if initial.is_dir() {
return initial;
}
for rel in &["../ui", "./ui", "../../ui"] {
let p = std::path::PathBuf::from(rel);
if p.is_dir() {
warn!(
"UI path {} not found; using {} (set --ui-path explicitly if wrong)",
initial.display(),
p.display()
);
return p;
}
}
initial
}
#[tokio::main]
async fn main() {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| "info,tower_http=debug".into()),
)
.init();
let mut args = Args::parse();
args.ui_path = coalesce_ui_path(args.ui_path);
if args.benchmark {
eprintln!("Running vital sign detection benchmark (1000 frames)...");
let (total, per_frame) = vital_signs::run_benchmark(1000);
eprintln!();
eprintln!("Summary: {total:?} total, {per_frame:?} per frame");
return;
}
if export_emits_placeholder_demo(args.export_rvf.is_some(), args.train, args.pretrain) {
let rvf_path = args
.export_rvf
.as_ref()
.expect("export_emits_placeholder_demo implies export_rvf is set");
eprintln!(
"WARNING: --export-rvf writes a CONTAINER-FORMAT DEMO with placeholder \
weights — it is NOT a trained model. Train one with \
`--train --dataset <DIR>` (which exports a calibrated .rvf to the \
models/ directory), or download a pretrained encoder. See issue #894."
);
eprintln!("Exporting RVF container package (placeholder weights)...");
use rvf_pipeline::RvfModelBuilder;
let mut builder = RvfModelBuilder::new("wifi-densepose", "1.0.0");
builder.set_vital_config(0.1, 0.5, 0.8, 2.0);
builder.set_model_profile(
"56-subcarrier CSI amplitude/phase @ 10-100 Hz",
"17 COCO keypoints + body part UV + vital signs",
"ESP32-S3 or Windows WiFi RSSI, Rust 1.85+",
);
let placeholder_weights: Vec<f32> = (0..2856).map(|i| (i as f32 * 0.001).sin()).collect();
builder.set_weights(&placeholder_weights);
builder.set_training_proof(
"wifi-densepose-rs-v1.0.0",
serde_json::json!({
"pipeline": "ADR-023 8-phase",
"test_count": 229,
"benchmark_fps": 9520,
"framework": "wifi-densepose-rs",
}),
);
let default_lora: Vec<f32> = vec![0.0; 64];
builder.add_sona_profile("default", &default_lora, &default_lora);
match builder.build() {
Ok(rvf_bytes) => {
if let Err(e) = std::fs::write(rvf_path, &rvf_bytes) {
eprintln!("Error writing RVF: {e}");
std::process::exit(1);
}
eprintln!("Wrote {} bytes to {}", rvf_bytes.len(), rvf_path.display());
eprintln!("RVF container exported successfully.");
}
Err(e) => {
eprintln!("Error building RVF: {e}");
std::process::exit(1);
}
}
return;
} else if args.export_rvf.is_some() {
eprintln!(
"Note: --export-rvf is ignored in training mode — the trained model \
is exported by the training pipeline to the models/ directory."
);
}
if args.pretrain {
eprintln!("=== WiFi-DensePose Contrastive Pretraining (ADR-024) ===");
let ds_path = args
.dataset
.clone()
.unwrap_or_else(|| PathBuf::from("data"));
let source = match args.dataset_type.as_str() {
"wipose" => dataset::DataSource::WiPose(ds_path.clone()),
_ => dataset::DataSource::MmFi(ds_path.clone()),
};
let pipeline = dataset::DataPipeline::new(dataset::DataConfig {
source,
..Default::default()
});
let generate_synthetic_windows = || -> Vec<Vec<Vec<f32>>> {
(0..50)
.map(|i| {
(0..4)
.map(|a| {
(0..56)
.map(|s| ((i * 7 + a * 13 + s) as f32 * 0.31).sin() * 0.5)
.collect()
})
.collect()
})
.collect()
};
let csi_windows: Vec<Vec<Vec<f32>>> = match pipeline.load() {
Ok(s) if !s.is_empty() => {
eprintln!("Loaded {} samples from {}", s.len(), ds_path.display());
s.into_iter().map(|s| s.csi_window).collect()
}
_ => {
eprintln!("Using synthetic data for pretraining.");
generate_synthetic_windows()
}
};
let n_subcarriers = csi_windows
.first()
.and_then(|w| w.first())
.map(|f| f.len())
.unwrap_or(56);
let tf_config = graph_transformer::TransformerConfig {
n_subcarriers,
n_keypoints: 17,
d_model: 64,
n_heads: 4,
n_gnn_layers: 2,
};
let transformer = graph_transformer::CsiToPoseTransformer::new(tf_config);
eprintln!("Transformer params: {}", transformer.param_count());
let trainer_config = trainer::TrainerConfig {
epochs: args.pretrain_epochs,
batch_size: 8,
lr: 0.001,
warmup_epochs: 2,
min_lr: 1e-6,
early_stop_patience: args.pretrain_epochs + 1,
pretrain_temperature: 0.07,
..Default::default()
};
let mut t = trainer::Trainer::with_transformer(trainer_config, transformer);
let e_config = embedding::EmbeddingConfig {
d_model: 64,
d_proj: 128,
temperature: 0.07,
normalize: true,
};
let mut projection = embedding::ProjectionHead::new(e_config.clone());
let augmenter = embedding::CsiAugmenter::new();
eprintln!(
"Starting contrastive pretraining for {} epochs...",
args.pretrain_epochs
);
let start = std::time::Instant::now();
for epoch in 0..args.pretrain_epochs {
let loss = t.pretrain_epoch(&csi_windows, &augmenter, &mut projection, 0.07, epoch);
if epoch % 10 == 0 || epoch == args.pretrain_epochs - 1 {
eprintln!(" Epoch {epoch}: contrastive loss = {loss:.4}");
}
}
let elapsed = start.elapsed().as_secs_f64();
eprintln!("Pretraining complete in {elapsed:.1}s");
if let Some(ref save_path) = args.save_rvf {
eprintln!("Saving pretrained model to RVF: {}", save_path.display());
t.sync_transformer_weights();
let weights = t.params().to_vec();
let mut proj_weights = Vec::new();
projection.flatten_into(&mut proj_weights);
let mut builder = RvfBuilder::new();
builder.add_manifest(
"wifi-densepose-pretrained",
env!("CARGO_PKG_VERSION"),
"WiFi DensePose contrastive pretrained model (ADR-024)",
);
builder.add_weights(&weights);
builder.add_embedding(
&serde_json::json!({
"d_model": e_config.d_model,
"d_proj": e_config.d_proj,
"temperature": e_config.temperature,
"normalize": e_config.normalize,
"pretrain_epochs": args.pretrain_epochs,
}),
&proj_weights,
);
match builder.write_to_file(save_path) {
Ok(()) => eprintln!(
"RVF saved ({} transformer + {} projection params)",
weights.len(),
proj_weights.len()
),
Err(e) => eprintln!("Failed to save RVF: {e}"),
}
}
return;
}
if args.embed {
eprintln!("=== WiFi-DensePose Embedding Extraction (ADR-024) ===");
let model_path = match &args.model {
Some(p) => p.clone(),
None => {
eprintln!("Error: --embed requires --model <path> to a pretrained .rvf file");
std::process::exit(1);
}
};
let reader = match RvfReader::from_file(&model_path) {
Ok(r) => r,
Err(e) => {
eprintln!("Failed to load model: {e}");
std::process::exit(1);
}
};
let weights = reader.weights().unwrap_or_default();
let (embed_config_json, proj_weights) = reader.embedding().unwrap_or_else(|| {
eprintln!("Warning: no embedding segment in RVF, using defaults");
(
serde_json::json!({"d_model":64,"d_proj":128,"temperature":0.07,"normalize":true}),
Vec::new(),
)
});
let d_model = embed_config_json["d_model"].as_u64().unwrap_or(64) as usize;
let d_proj = embed_config_json["d_proj"].as_u64().unwrap_or(128) as usize;
let tf_config = graph_transformer::TransformerConfig {
n_subcarriers: 56,
n_keypoints: 17,
d_model,
n_heads: 4,
n_gnn_layers: 2,
};
let e_config = embedding::EmbeddingConfig {
d_model,
d_proj,
temperature: 0.07,
normalize: true,
};
let mut extractor = embedding::EmbeddingExtractor::new(tf_config, e_config.clone());
if !weights.is_empty() {
if let Err(e) = extractor.transformer.unflatten_weights(&weights) {
eprintln!("Warning: failed to load transformer weights: {e}");
}
}
if !proj_weights.is_empty() {
let (proj, _) = embedding::ProjectionHead::unflatten_from(&proj_weights, &e_config);
extractor.projection = proj;
}
let _ds_path = args
.dataset
.clone()
.unwrap_or_else(|| PathBuf::from("data"));
let csi_windows: Vec<Vec<Vec<f32>>> = (0..10)
.map(|i| {
(0..4)
.map(|a| {
(0..56)
.map(|s| ((i * 7 + a * 13 + s) as f32 * 0.31).sin() * 0.5)
.collect()
})
.collect()
})
.collect();
eprintln!(
"Extracting embeddings from {} CSI windows...",
csi_windows.len()
);
let embeddings = extractor.extract_batch(&csi_windows);
for (i, emb) in embeddings.iter().enumerate() {
let norm: f32 = emb.iter().map(|x| x * x).sum::<f32>().sqrt();
eprintln!(" Window {i}: {d_proj}-dim embedding, ||e|| = {norm:.4}");
}
eprintln!(
"Extracted {} embeddings of dimension {d_proj}",
embeddings.len()
);
return;
}
if let Some(ref index_type_str) = args.build_index {
eprintln!("=== WiFi-DensePose Fingerprint Index Builder (ADR-024) ===");
let index_type = match index_type_str.as_str() {
"env" | "environment" => embedding::IndexType::EnvironmentFingerprint,
"activity" => embedding::IndexType::ActivityPattern,
"temporal" => embedding::IndexType::TemporalBaseline,
"person" => embedding::IndexType::PersonTrack,
_ => {
eprintln!(
"Unknown index type '{}'. Use: env, activity, temporal, person",
index_type_str
);
std::process::exit(1);
}
};
let tf_config = graph_transformer::TransformerConfig::default();
let e_config = embedding::EmbeddingConfig::default();
let mut extractor = embedding::EmbeddingExtractor::new(tf_config, e_config);
let csi_windows: Vec<Vec<Vec<f32>>> = (0..20)
.map(|i| {
(0..4)
.map(|a| {
(0..56)
.map(|s| ((i * 7 + a * 13 + s) as f32 * 0.31).sin() * 0.5)
.collect()
})
.collect()
})
.collect();
let mut index = embedding::FingerprintIndex::new(index_type);
for (i, window) in csi_windows.iter().enumerate() {
let emb = extractor.extract(window);
index.insert(emb, format!("window_{i}"), i as u64 * 100);
}
eprintln!("Built {:?} index with {} entries", index_type, index.len());
let query_emb = extractor.extract(&csi_windows[0]);
let results = index.search(&query_emb, 5);
eprintln!("Top-5 nearest to window_0:");
for r in &results {
eprintln!(
" entry={}, distance={:.4}, metadata={}",
r.entry, r.distance, r.metadata
);
}
return;
}
if args.train {
eprintln!("=== WiFi-DensePose Training Mode ===");
let ds_path = args
.dataset
.clone()
.unwrap_or_else(|| PathBuf::from("data"));
let source = match args.dataset_type.as_str() {
"wipose" => dataset::DataSource::WiPose(ds_path.clone()),
_ => dataset::DataSource::MmFi(ds_path.clone()),
};
let pipeline = dataset::DataPipeline::new(dataset::DataConfig {
source,
..Default::default()
});
let generate_synthetic = || -> Vec<dataset::TrainingSample> {
(0..50)
.map(|i| {
let csi: Vec<Vec<f32>> = (0..4)
.map(|a| {
(0..56)
.map(|s| ((i * 7 + a * 13 + s) as f32 * 0.31).sin() * 0.5)
.collect()
})
.collect();
let mut kps = [(0.0f32, 0.0f32, 1.0f32); 17];
for (k, kp) in kps.iter_mut().enumerate() {
kp.0 = (k as f32 * 0.1 + i as f32 * 0.02).sin() * 100.0 + 320.0;
kp.1 = (k as f32 * 0.15 + i as f32 * 0.03).cos() * 80.0 + 240.0;
}
dataset::TrainingSample {
csi_window: csi,
pose_label: dataset::PoseLabel {
keypoints: kps,
body_parts: Vec::new(),
confidence: 1.0,
},
source: "synthetic",
}
})
.collect()
};
let samples = match pipeline.load() {
Ok(s) if !s.is_empty() => {
eprintln!("Loaded {} samples from {}", s.len(), ds_path.display());
s
}
Ok(_) => {
eprintln!(
"No samples found at {}. Using synthetic data.",
ds_path.display()
);
generate_synthetic()
}
Err(e) => {
eprintln!("Failed to load dataset: {e}. Using synthetic data.");
generate_synthetic()
}
};
let trainer_samples: Vec<trainer::TrainingSample> =
samples.iter().map(trainer::from_dataset_sample).collect();
let split = (trainer_samples.len() * 4) / 5;
let (train_data, val_data) = trainer_samples.split_at(split.max(1));
eprintln!(
"Train: {} samples, Val: {} samples",
train_data.len(),
val_data.len()
);
let n_subcarriers = train_data
.first()
.and_then(|s| s.csi_features.first())
.map(|f| f.len())
.unwrap_or(56);
let tf_config = graph_transformer::TransformerConfig {
n_subcarriers,
n_keypoints: 17,
d_model: 64,
n_heads: 4,
n_gnn_layers: 2,
};
let transformer = graph_transformer::CsiToPoseTransformer::new(tf_config);
eprintln!("Transformer params: {}", transformer.param_count());
let trainer_config = trainer::TrainerConfig {
epochs: args.epochs,
batch_size: 8,
lr: 0.001,
warmup_epochs: 5,
min_lr: 1e-6,
early_stop_patience: 20,
checkpoint_every: 10,
..Default::default()
};
let mut t = trainer::Trainer::with_transformer(trainer_config, transformer);
eprintln!("Starting training for {} epochs...", args.epochs);
let result = t.run_training(train_data, val_data);
eprintln!("Training complete in {:.1}s", result.total_time_secs);
eprintln!(
" Best epoch: {}, PCK@0.2: {:.4}, OKS mAP: {:.4}",
result.best_epoch, result.best_pck, result.best_oks
);
if let Some(ref ckpt_dir) = args.checkpoint_dir {
let _ = std::fs::create_dir_all(ckpt_dir);
let ckpt_path = ckpt_dir.join("best_checkpoint.json");
let ckpt = t.checkpoint();
match ckpt.save_to_file(&ckpt_path) {
Ok(()) => eprintln!("Checkpoint saved to {}", ckpt_path.display()),
Err(e) => eprintln!("Failed to save checkpoint: {e}"),
}
}
t.sync_transformer_weights();
if let Some(ref save_path) = args.save_rvf {
eprintln!("Saving trained model to RVF: {}", save_path.display());
let weights = t.params().to_vec();
let mut builder = RvfBuilder::new();
builder.add_manifest(
"wifi-densepose-trained",
env!("CARGO_PKG_VERSION"),
"WiFi DensePose trained model weights",
);
builder.add_metadata(&serde_json::json!({
"training": {
"epochs": args.epochs,
"best_epoch": result.best_epoch,
"best_pck": result.best_pck,
"best_oks": result.best_oks,
"n_train_samples": train_data.len(),
"n_val_samples": val_data.len(),
"n_subcarriers": n_subcarriers,
"param_count": weights.len(),
},
}));
builder.add_vital_config(&VitalSignConfig::default());
builder.add_weights(&weights);
match builder.write_to_file(save_path) {
Ok(()) => eprintln!(
"RVF saved ({} params, {} bytes)",
weights.len(),
weights.len() * 4
),
Err(e) => eprintln!("Failed to save RVF: {e}"),
}
}
return;
}
info!("WiFi-DensePose Sensing Server (Rust + Axum + RuVector)");
info!(" HTTP: http://localhost:{}", args.http_port);
info!(" WebSocket: ws://localhost:{}/ws/sensing", args.ws_port);
info!(" UDP: 0.0.0.0:{} (ESP32 CSI)", args.udp_port);
info!(" UI path: {}", args.ui_path.display());
info!(" Source: {}", args.source);
let source = match args.source.as_str() {
"auto" => {
info!("Auto-detecting data source...");
if probe_esp32(args.udp_port).await {
info!(" ESP32 CSI detected on UDP :{}", args.udp_port);
"esp32"
} else if probe_windows_wifi().await {
info!(" Windows WiFi detected");
"wifi"
} else {
error!(
"No real CSI source detected. Auto-detection refuses to silently \
fall back to synthetic data because that would expose downstream \
consumers (/api/v1/sensing/latest, /ws/sensing) to fake telemetry \
tagged as production. To run with synthetic data, set the source \
explicitly: --source simulated (or CSI_SOURCE=simulated in Docker). \
To use real hardware: provision an ESP32 to emit CSI on UDP :{} or \
install the Windows WiFi capture driver. See \
https://github.com/ruvnet/RuView/issues/937 for context.",
args.udp_port
);
std::process::exit(78); }
}
"simulate" => "simulated",
other => other,
};
info!("Data source: {source}");
let vital_sample_rate = 1000.0 / args.tick_ms as f64;
info!("Vital sign detector sample rate: {vital_sample_rate:.1} Hz");
let rvf_info = if let Some(ref rvf_path) = args.load_rvf {
info!("Loading RVF container from {}", rvf_path.display());
match RvfReader::from_file(rvf_path) {
Ok(reader) => {
let info = reader.info();
info!(
" RVF loaded: {} segments, {} bytes",
info.segment_count, info.total_size
);
if let Some(ref manifest) = info.manifest {
if let Some(model_id) = manifest.get("model_id") {
info!(" Model ID: {model_id}");
}
if let Some(version) = manifest.get("version") {
info!(" Version: {version}");
}
}
if info.has_weights {
if let Some(w) = reader.weights() {
info!(" Weights: {} parameters", w.len());
}
}
if info.has_vital_config {
info!(" Vital sign config: present");
}
if info.has_quant_info {
info!(" Quantization info: present");
}
if info.has_witness {
info!(" Witness/proof: present");
}
Some(info)
}
Err(e) => {
error!("Failed to load RVF container: {e}");
None
}
}
} else {
None
};
let model_path = args.model.as_ref().or(args.load_rvf.as_ref());
let mut progressive_loader: Option<ProgressiveLoader> = None;
let mut model_loaded = false;
if let Some(mp) = model_path {
if args.progressive || args.model.is_some() {
info!("Loading trained model (progressive) from {}", mp.display());
match std::fs::read(mp) {
Ok(data) => match ProgressiveLoader::new(&data) {
Ok(mut loader) => {
if let Ok(la) = loader.load_layer_a() {
info!(
" Layer A ready: model={} v{} ({} segments)",
la.model_name, la.version, la.n_segments
);
}
model_loaded = true;
progressive_loader = Some(loader);
}
Err(e) => {
error!("{}", diagnose_model_load_error(mp, &data, &e.to_string()))
}
},
Err(e) => error!("Failed to read model file: {e}"),
}
}
}
let models_dir = effective_models_dir();
let _ = std::fs::create_dir_all(&models_dir);
let _ = std::fs::create_dir_all("data/recordings");
let initial_models = scan_model_files();
let initial_recordings = scan_recording_files();
info!(
"Discovered {} model files, {} recording files",
initial_models.len(),
initial_recordings.len()
);
let data_dir = std::path::PathBuf::from("data");
let runtime_config = load_runtime_config(&data_dir);
info!(
"Loaded runtime config: dedup_factor={:.2}",
runtime_config.dedup_factor
);
let edge_registry: Option<
std::sync::Arc<wifi_densepose_sensing_server::edge_registry::EdgeRegistry>,
> = if args.no_edge_registry || args.edge_registry_url.is_empty() {
info!("Edge module registry: DISABLED (--no-edge-registry or empty URL)");
None
} else {
info!(
"Edge module registry: enabled — upstream={} ttl={}s",
args.edge_registry_url, args.edge_registry_ttl_secs
);
Some(std::sync::Arc::new(
wifi_densepose_sensing_server::edge_registry::EdgeRegistry::new(
args.edge_registry_url.clone(),
std::time::Duration::from_secs(args.edge_registry_ttl_secs),
),
))
};
let (tx, _) = broadcast::channel::<String>(256);
let (intro_tx, _) = broadcast::channel::<String>(256);
if args.mqtt_opts.mqtt {
#[cfg(feature = "mqtt")]
{
use wifi_densepose_sensing_server::mqtt;
let mcfg = std::sync::Arc::new(mqtt::config::MqttConfig::from_args(&args.mqtt_opts));
match mcfg.validate() {
Ok(()) => {
let node_id = mcfg.client_id.clone();
let builder = mqtt::publisher::OwnedDiscoveryBuilder {
discovery_prefix: mcfg.discovery_prefix.clone(),
node_id: node_id.clone(),
node_friendly_name: Some("RuView".to_string()),
sw_version: env!("CARGO_PKG_VERSION").to_string(),
model: "RuView WiFi Sensing".to_string(),
via_device: None,
};
let (vtx, vrx) = broadcast::channel::<mqtt::state::VitalsSnapshot>(64);
let (host, port) = (mcfg.host.clone(), mcfg.port);
mqtt::publisher::spawn(mcfg, builder, vrx);
let mut jrx = tx.subscribe();
tokio::spawn(async move {
while let Ok(json) = jrx.recv().await {
let Ok(v) = serde_json::from_str::<serde_json::Value>(&json) else {
continue;
};
for snap in vitals_snapshots_from_sensing_json(&v, &node_id) {
let _ = vtx.send(snap);
}
}
});
tracing::info!("MQTT publisher started -> {host}:{port}");
}
Err(e) => tracing::error!("MQTT config invalid: {e}; publisher not started"),
}
}
#[cfg(not(feature = "mqtt"))]
tracing::warn!(
"--mqtt set but this binary was built without the `mqtt` feature; the publisher is a \
no-op. Use the official Docker image (built `--features mqtt`) or rebuild with \
`cargo build -p wifi-densepose-sensing-server --features mqtt`."
);
}
let state: SharedState = Arc::new(RwLock::new(AppStateInner {
latest_update: None,
rssi_history: VecDeque::new(),
frame_history: VecDeque::new(),
tick: 0,
source: source.into(),
last_esp32_frame: None,
tx,
intro: wifi_densepose_sensing_server::introspection::IntrospectionState::new(),
intro_tx,
total_detections: 0,
start_time: std::time::Instant::now(),
vital_detector: VitalSignDetector::new(vital_sample_rate),
latest_vitals: VitalSigns::default(),
rvf_info,
save_rvf_path: args.save_rvf.clone(),
progressive_loader,
active_sona_profile: None,
model_loaded,
smoothed_person_score: 0.0,
prev_person_count: 0,
smoothed_motion: 0.0,
current_motion_level: "absent".to_string(),
debounce_counter: 0,
debounce_candidate: "absent".to_string(),
baseline_motion: 0.0,
baseline_frames: 0,
smoothed_hr: 0.0,
smoothed_br: 0.0,
smoothed_hr_conf: 0.0,
smoothed_br_conf: 0.0,
hr_buffer: VecDeque::with_capacity(8),
br_buffer: VecDeque::with_capacity(8),
edge_vitals: None,
latest_wasm_events: None,
discovered_models: initial_models,
active_model_id: None,
recordings: initial_recordings,
recording_active: false,
recording_start_time: None,
recording_current_id: None,
recording_stop_tx: None,
training_status: "idle".to_string(),
training_config: None,
adaptive_model:
adaptive_classifier::AdaptiveModel::load(&adaptive_classifier::model_path())
.ok()
.inspect(|m| {
info!(
"Loaded adaptive classifier: {} frames, {:.1}% accuracy",
m.trained_frames,
m.training_accuracy * 100.0
);
}),
node_states: HashMap::new(),
pose_tracker: PoseTracker::new(),
last_tracker_instant: None,
multistatic_fuser: {
let mut fuser = MultistaticFuser::with_config(MultistaticConfig {
min_nodes: 1, ..Default::default()
});
if let Some(ref pos_str) = args.node_positions {
let positions = field_bridge::parse_node_positions(pos_str);
if !positions.is_empty() {
info!(
"Configured {} node positions for multistatic fusion",
positions.len()
);
fuser.set_node_positions(positions);
}
}
fuser
},
engine_bridge: engine_bridge::EngineBridge::new(
wifi_densepose_bfld::PrivacyMode::PrivateHome,
1,
"default",
"Default Room",
),
field_model: if args.calibrate {
info!("Field model calibration enabled — room should be empty during startup");
FieldModel::new(field_bridge::single_link_config()).ok()
} else {
None
},
p95_variance: RollingP95::new(600, 60),
p95_motion_band_power: RollingP95::new(600, 60),
p95_spectral_power: RollingP95::new(600, 60),
dedup_factor: runtime_config.dedup_factor,
data_dir: data_dir.clone(),
}));
match source {
"esp32" => {
tokio::spawn(udp_receiver_task(state.clone(), args.udp_port));
tokio::spawn(broadcast_tick_task(state.clone(), args.tick_ms));
}
"wifi" => {
tokio::spawn(windows_wifi_task(state.clone(), args.tick_ms));
}
_ => {
tokio::spawn(simulated_data_task(state.clone(), args.tick_ms));
}
}
let bind_ip: std::net::IpAddr = args
.bind_addr
.parse()
.expect("Invalid --bind-addr (use 127.0.0.1 or 0.0.0.0)");
let bearer_auth_state = wifi_densepose_sensing_server::bearer_auth::AuthState::from_env();
if bearer_auth_state.is_enabled() {
info!("API auth: bearer-token enforcement ON for /api/v1/* (RUVIEW_API_TOKEN set)");
if bind_ip.is_unspecified() {
warn!(
"API auth ON but bind-addr is {} — consider --bind-addr 127.0.0.1 for LAN-only deployments",
bind_ip
);
}
} else {
info!(
"API auth: OFF — /api/v1/* is unauthenticated. Set RUVIEW_API_TOKEN=<token> to enforce bearer auth."
);
}
let host_allowlist = if args.disable_host_validation {
warn!(
"Host-header validation DISABLED — server is reachable via any Host. \
Only use this behind a reverse proxy that pins Host."
);
wifi_densepose_sensing_server::host_validation::HostAllowlist::disabled()
} else {
let allowlist =
wifi_densepose_sensing_server::host_validation::HostAllowlist::from_cli_and_env(
args.allowed_hosts.iter().cloned(),
);
info!(
"Host-header validation ON ({} entries; loopback names always included)",
allowlist.entries_for_test().len()
);
allowlist
};
let ws_state = state.clone();
let ws_app = Router::new()
.route("/ws/sensing", get(ws_sensing_handler))
.route("/health", get(health))
.layer(axum::middleware::from_fn_with_state(
host_allowlist.clone(),
wifi_densepose_sensing_server::host_validation::require_allowed_host,
))
.with_state(ws_state);
let ws_addr = SocketAddr::from((bind_ip, args.ws_port));
let ws_listener = tokio::net::TcpListener::bind(ws_addr)
.await
.expect("Failed to bind WebSocket port");
info!("WebSocket server listening on {ws_addr}");
tokio::spawn(async move {
axum::serve(ws_listener, ws_app).await.unwrap();
});
let ui_path = args.ui_path.clone();
let http_app = Router::new()
.route("/", get(info_page))
.route("/health", get(health))
.route("/health/health", get(health_system))
.route("/health/live", get(health_live))
.route("/health/ready", get(health_ready))
.route("/health/version", get(health_version))
.route("/health/metrics", get(health_metrics))
.route("/api/v1/info", get(api_info))
.route("/api/v1/status", get(health_ready))
.route("/api/v1/metrics", get(health_metrics))
.route("/api/v1/sensing/latest", get(latest))
.route("/api/v1/nodes", get(nodes_endpoint))
.route("/api/v1/nodes/:id/sync", get(node_sync_endpoint))
.route("/api/v1/mesh", get(mesh_endpoint))
.route("/api/v1/mesh/metrics", get(mesh_metrics_endpoint))
.route("/api/v1/vital-signs", get(vital_signs_endpoint))
.route("/api/v1/edge-vitals", get(edge_vitals_endpoint))
.route("/api/v1/edge/registry", get(edge_registry_endpoint))
.route("/api/v1/wasm-events", get(wasm_events_endpoint))
.route("/api/v1/model/info", get(model_info))
.route("/api/v1/model/layers", get(model_layers))
.route("/api/v1/model/segments", get(model_segments))
.route("/api/v1/model/sona/profiles", get(sona_profiles))
.route("/api/v1/model/sona/activate", post(sona_activate))
.route("/api/v1/pose/current", get(pose_current))
.route("/api/v1/pose/stats", get(pose_stats))
.route("/api/v1/pose/zones/summary", get(pose_zones_summary))
.route("/api/v1/stream/status", get(stream_status))
.route("/api/v1/stream/pose", get(ws_pose_handler))
.route("/ws/sensing", get(ws_sensing_handler))
.route("/ws/introspection", get(ws_introspection_handler))
.route(
"/api/v1/introspection/snapshot",
get(api_introspection_snapshot),
)
.route("/api/v1/models", get(list_models))
.route("/api/v1/models/active", get(get_active_model))
.route("/api/v1/models/load", post(load_model))
.route("/api/v1/models/unload", post(unload_model))
.route("/api/v1/models/{id}", delete(delete_model))
.route("/api/v1/models/lora/profiles", get(list_lora_profiles))
.route("/api/v1/models/lora/activate", post(activate_lora_profile))
.route("/api/v1/recording/list", get(list_recordings))
.route("/api/v1/recording/start", post(start_recording))
.route("/api/v1/recording/stop", post(stop_recording))
.route("/api/v1/recording/{id}", delete(delete_recording))
.route("/api/v1/train/status", get(train_status))
.route("/api/v1/train/start", post(train_start))
.route("/api/v1/train/stop", post(train_stop))
.route("/api/v1/adaptive/train", post(adaptive_train))
.route("/api/v1/adaptive/status", get(adaptive_status))
.route("/api/v1/adaptive/unload", post(adaptive_unload))
.route("/api/v1/calibration/start", post(calibration_start))
.route("/api/v1/calibration/stop", post(calibration_stop))
.route("/api/v1/calibration/status", get(calibration_status))
.route(
"/api/v1/config/dedup-factor",
get(config_get_dedup_factor).post(config_set_dedup_factor),
)
.route("/api/v1/config/ground-truth", post(config_set_ground_truth))
.nest_service("/ui", ServeDir::new(&ui_path))
.layer(Extension(edge_registry.clone()))
.layer(SetResponseHeaderLayer::overriding(
axum::http::header::CACHE_CONTROL,
HeaderValue::from_static("no-cache, no-store, must-revalidate"),
))
.layer(axum::middleware::from_fn_with_state(
bearer_auth_state.clone(),
wifi_densepose_sensing_server::bearer_auth::require_bearer,
))
.layer(axum::middleware::from_fn_with_state(
host_allowlist.clone(),
wifi_densepose_sensing_server::host_validation::require_allowed_host,
))
.with_state(state.clone());
let http_addr = SocketAddr::from((bind_ip, args.http_port));
let http_listener = tokio::net::TcpListener::bind(http_addr)
.await
.expect("Failed to bind HTTP port");
info!("HTTP server listening on {http_addr}");
info!(
"Open http://localhost:{}/ui/index.html in your browser",
args.http_port
);
let shutdown_state = state.clone();
let server = axum::serve(http_listener, http_app).with_graceful_shutdown(async {
tokio::signal::ctrl_c()
.await
.expect("failed to install CTRL+C handler");
info!("Shutdown signal received");
});
server.await.unwrap();
let s = shutdown_state.read().await;
if let Some(ref save_path) = s.save_rvf_path {
info!("Saving RVF container to {}", save_path.display());
let mut builder = RvfBuilder::new();
builder.add_manifest(
"wifi-densepose-sensing",
env!("CARGO_PKG_VERSION"),
"WiFi DensePose sensing model state",
);
builder.add_metadata(&serde_json::json!({
"source": s.effective_source(),
"total_ticks": s.tick,
"total_detections": s.total_detections,
"uptime_secs": s.start_time.elapsed().as_secs(),
}));
builder.add_vital_config(&VitalSignConfig::default());
let weights: Vec<f32> = if s.model_loaded {
let tf = graph_transformer::CsiToPoseTransformer::new(Default::default());
tf.flatten_weights()
} else {
Vec::new()
};
builder.add_weights(&weights);
match builder.write_to_file(save_path) {
Ok(()) => info!(" RVF saved ({} weight params)", weights.len()),
Err(e) => error!(" Failed to save RVF: {e}"),
}
}
info!("Server shut down cleanly");
}
#[cfg(test)]
mod node_sync_snapshot_serialization_tests {
use super::*;
fn sample_sync() -> NodeSyncSnapshot {
NodeSyncSnapshot {
offset_us: 1_163_565,
is_leader: false,
is_valid: true,
smoothed: true,
sequence: 20,
csi_fps_ema: 10.0,
csi_fps_samples: 47,
staleness_ms: Some(120),
}
}
fn sample_node(sync: Option<NodeSyncSnapshot>) -> NodeInfo {
NodeInfo {
node_id: 9,
rssi_dbm: -38.0,
position: [2.0, 0.0, 1.5],
amplitude: vec![],
subcarrier_count: 0,
sync,
}
}
#[test]
fn sync_present_serializes_all_seven_fields() {
let v = serde_json::to_value(sample_node(Some(sample_sync()))).unwrap();
let s = v.get("sync").expect("sync key must be present");
for key in ["offset_us", "is_leader", "is_valid", "smoothed",
"sequence", "csi_fps_ema", "csi_fps_samples",
"staleness_ms"] {
assert!(s.get(key).is_some(),
"sync object missing field `{}` — UI contract broken", key);
}
assert_eq!(s["offset_us"], 1_163_565);
assert_eq!(s["is_leader"], false);
assert_eq!(s["sequence"], 20);
assert_eq!(s["csi_fps_samples"], 47);
}
#[test]
fn sync_absent_omits_the_key_entirely() {
let v = serde_json::to_value(sample_node(None)).unwrap();
assert!(v.get("sync").is_none(),
"expected `sync` key omitted when None, got {:?}", v.get("sync"));
assert_eq!(v["node_id"], 9);
assert_eq!(v["rssi_dbm"], -38.0);
}
#[test]
fn sync_round_trips_through_serde() {
let original = sample_node(Some(sample_sync()));
let json = serde_json::to_string(&original).unwrap();
let parsed: NodeInfo = serde_json::from_str(&json).unwrap();
let s_orig = original.sync.unwrap();
let s_parsed = parsed.sync.expect("sync should survive round-trip");
assert_eq!(s_parsed.offset_us, s_orig.offset_us);
assert_eq!(s_parsed.is_leader, s_orig.is_leader);
assert_eq!(s_parsed.is_valid, s_orig.is_valid);
assert_eq!(s_parsed.smoothed, s_orig.smoothed);
assert_eq!(s_parsed.sequence, s_orig.sequence);
assert!((s_parsed.csi_fps_ema - s_orig.csi_fps_ema).abs() < 1e-9);
assert_eq!(s_parsed.csi_fps_samples, s_orig.csi_fps_samples);
}
}
#[cfg(test)]
mod sync_snapshot_helper_tests {
use super::*;
use wifi_densepose_hardware::{SyncPacket, SyncPacketFlags};
fn populated_sync(node_id: u8) -> SyncPacket {
SyncPacket {
node_id,
proto_ver: 1,
flags: SyncPacketFlags { is_leader: false, is_valid: true, smoothed_used: true },
local_us: 28_798_450,
epoch_us: 27_634_885,
sequence: 20,
}
}
#[test]
fn fresh_node_with_no_sync_returns_none() {
let ns = NodeState::new();
assert!(ns.sync_snapshot().is_none());
}
#[test]
fn node_with_latest_sync_produces_correct_snapshot() {
let mut ns = NodeState::new();
ns.latest_sync = Some(populated_sync(9));
ns.latest_sync_at = Some(std::time::Instant::now());
ns.csi_fps_ema = 10.5;
ns.csi_fps_samples = 42;
let snap = ns.sync_snapshot().expect("populated state must produce a snapshot");
assert_eq!(snap.offset_us, 1_163_565); assert!(!snap.is_leader);
assert!(snap.is_valid);
assert!(snap.smoothed);
assert_eq!(snap.sequence, 20);
assert!((snap.csi_fps_ema - 10.5).abs() < 1e-9);
assert_eq!(snap.csi_fps_samples, 42);
}
#[test]
fn apply_sync_packet_populates_a_fresh_node() {
let mut ns = NodeState::new();
assert!(ns.latest_sync.is_none());
assert!(ns.latest_sync_at.is_none());
let now = std::time::Instant::now();
ns.apply_sync_packet(populated_sync(9), now);
let sync = ns.latest_sync.as_ref().expect("must be populated");
assert_eq!(sync.node_id, 9);
assert_eq!(sync.sequence, 20);
assert_eq!(ns.latest_sync_at, Some(now));
assert!(ns.sync_snapshot().is_some());
}
#[test]
fn apply_sync_packet_overwrites_older_data() {
let mut ns = NodeState::new();
let t0 = std::time::Instant::now();
ns.apply_sync_packet(populated_sync(9), t0);
let mut second = populated_sync(9);
second.sequence = 40;
second.local_us = 30_000_000;
second.epoch_us = 28_834_900;
let t1 = t0 + std::time::Duration::from_secs(2);
ns.apply_sync_packet(second, t1);
let cur = ns.latest_sync.as_ref().unwrap();
assert_eq!(cur.sequence, 40); assert_eq!(cur.local_us, 30_000_000); assert_eq!(ns.latest_sync_at, Some(t1)); }
#[test]
fn snapshot_staleness_ms_tracks_apply_time() {
let mut ns = NodeState::new();
ns.latest_sync = Some(populated_sync(9));
ns.latest_sync_at = std::time::Instant::now()
.checked_sub(std::time::Duration::from_millis(750));
let snap = ns.sync_snapshot().unwrap();
let st = snap.staleness_ms.expect("staleness_ms must be present");
assert!(st >= 740 && st < 1250,
"expected ~750 ms staleness, got {} ms", st);
}
#[test]
fn fleet_role_counts_classifies_correctly() {
fn snap(is_leader: bool) -> NodeSyncSnapshot {
NodeSyncSnapshot {
offset_us: 0, is_leader, is_valid: true, smoothed: true,
sequence: 0, csi_fps_ema: 10.0, csi_fps_samples: 10,
staleness_ms: Some(0),
}
}
assert_eq!(super::fleet_role_counts(&[]), (0, 0));
let snaps = vec![(12u8, snap(true)), (9, snap(false)), (3, snap(false))];
assert_eq!(super::fleet_role_counts(&snaps), (1, 2));
assert_eq!(super::fleet_role_counts(&[(1u8, snap(true)), (2, snap(true))]), (2, 0));
}
#[test]
fn bool_metric_returns_zero_or_one_as_text() {
assert_eq!(super::bool_metric(true), "1");
assert_eq!(super::bool_metric(false), "0");
}
#[test]
fn mesh_aligned_us_honors_9s_staleness_gate() {
let mut ns = NodeState::new();
let now = std::time::Instant::now();
ns.latest_sync = Some(populated_sync(9));
ns.latest_sync_at = now.checked_sub(std::time::Duration::from_secs(1));
assert!(ns.mesh_aligned_us_for_csi_frame(20).is_some(),
"1 s old sync must produce a mesh-aligned timestamp");
ns.latest_sync_at = now.checked_sub(std::time::Duration::from_secs(8));
assert!(ns.mesh_aligned_us_for_csi_frame(20).is_some(),
"8 s old sync must still be inside the 9 s gate");
ns.latest_sync_at = now.checked_sub(std::time::Duration::from_secs(10));
assert!(ns.mesh_aligned_us_for_csi_frame(20).is_none(),
"10 s old sync must trigger the 9 s staleness gate");
}
#[test]
fn snapshot_reflects_leader_state() {
let mut ns = NodeState::new();
let mut s = populated_sync(12);
s.flags = SyncPacketFlags { is_leader: true, is_valid: true, smoothed_used: false };
s.local_us = 28_864_932;
s.epoch_us = 28_864_939; ns.latest_sync = Some(s);
ns.latest_sync_at = Some(std::time::Instant::now());
let snap = ns.sync_snapshot().unwrap();
assert!(snap.is_leader);
assert_eq!(snap.offset_us, -7); assert!(!snap.smoothed);
}
}
#[cfg(test)]
mod novelty_tests {
use super::*;
#[test]
fn first_frame_yields_max_novelty_then_zero_on_repeat() {
let mut ns = NodeState::new();
let amplitudes: Vec<f64> = (0..NOVELTY_VECTOR_DIM).map(|i| (i as f64).sin()).collect();
ns.update_novelty(&litudes);
let first = ns.last_novelty_score.expect("sketch bank initialised");
assert!(
(first - 1.0).abs() < 1e-6,
"empty bank → max novelty 1.0, got {first}"
);
ns.update_novelty(&litudes);
let second = ns.last_novelty_score.expect("score stays Some");
assert_eq!(second, 0.0, "exact-repeat frame → novelty 0.0");
}
#[test]
fn handles_short_and_long_amplitude_vectors() {
let mut ns = NodeState::new();
ns.update_novelty(&[1.0, 2.0]); assert!(ns.last_novelty_score.is_some());
let too_long: Vec<f64> = (0..NOVELTY_VECTOR_DIM * 2).map(|i| i as f64).collect();
ns.update_novelty(&too_long); assert!(ns.last_novelty_score.is_some());
}
}
async fn config_get_dedup_factor(State(state): State<SharedState>) -> Json<serde_json::Value> {
let s = state.read().await;
Json(serde_json::json!({
"dedup_factor": s.dedup_factor,
"description": "Divisor for multi-node person count deduplication (sum / factor). Range: 1.0–10.0."
}))
}
async fn config_set_dedup_factor(
State(state): State<SharedState>,
Json(body): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
let value = body.get("value").and_then(|v| v.as_f64()).unwrap_or(3.0);
let clamped = value.clamp(1.0, 10.0);
let mut s = state.write().await;
s.dedup_factor = clamped;
let data_dir = s.data_dir.clone();
drop(s);
save_runtime_config(
&data_dir,
&RuntimeConfig {
dedup_factor: clamped,
},
);
Json(serde_json::json!({
"status": "ok",
"dedup_factor": clamped,
}))
}
async fn config_set_ground_truth(
State(state): State<SharedState>,
Json(body): Json<serde_json::Value>,
) -> Json<serde_json::Value> {
let ground_truth = match body.get("count").and_then(|v| v.as_u64()) {
Some(n) if n > 0 => n as usize,
_ => return Json(serde_json::json!({"error": "count must be a positive integer"})),
};
let mut s = state.write().await;
let raw_sum: usize = s
.node_states
.values()
.filter(|ns| {
ns.last_frame_time
.map(|t| t.elapsed() < std::time::Duration::from_secs(10))
.unwrap_or(false)
})
.map(|ns| ns.prev_person_count)
.sum();
let optimal = if raw_sum > 0 {
(raw_sum as f64) / (ground_truth as f64)
} else {
3.0
};
let clamped = optimal.clamp(1.0, 10.0);
s.dedup_factor = clamped;
let data_dir = s.data_dir.clone();
drop(s);
save_runtime_config(
&data_dir,
&RuntimeConfig {
dedup_factor: clamped,
},
);
Json(serde_json::json!({
"status": "ok",
"ground_truth": ground_truth,
"raw_sum": raw_sum,
"computed_dedup_factor": clamped,
}))
}
#[cfg(test)]
mod rolling_p95_tests {
use super::RollingP95;
#[test]
fn cold_start_returns_none() {
let p = RollingP95::new(100, 10);
assert!(p.current().is_none(), "empty buffer must return None");
}
#[test]
fn below_min_samples_returns_none() {
let mut p = RollingP95::new(100, 10);
for i in 1..=9 {
p.push(i as f64);
}
assert!(
p.current().is_none(),
"fewer than min_samples must return None"
);
}
#[test]
fn p95_of_ramp_is_near_95() {
let mut p = RollingP95::new(100, 10);
for i in 1..=100 {
p.push(i as f64);
}
let p95 = p.current().expect("should have value after 100 samples");
assert!(
(94.0..=96.0).contains(&p95),
"P95 of 1..=100 should be ~95, got {p95}"
);
}
#[test]
fn window_slides_evicts_oldest() {
let mut p = RollingP95::new(5, 3);
for i in 1..=5 {
p.push(i as f64);
}
p.push(100.0); let p95 = p.current().expect("6 pushes, window=5 → 5 samples");
assert_eq!(
p95, 100.0,
"largest value should dominate p95 after eviction"
);
}
#[test]
fn len_reports_buffer_size() {
let mut p = RollingP95::new(10, 5);
assert_eq!(p.len(), 0);
p.push(1.0);
assert_eq!(p.len(), 1);
}
}
#[cfg(all(test, feature = "mqtt"))]
mod mqtt_bridge_tests {
use super::vitals_snapshots_from_sensing_json;
use serde_json::json;
#[test]
fn per_node_presence_uses_each_nodes_own_classification() {
let v = json!({
"timestamp": 1.0,
"classification": { "presence": true, "motion_level": "walking", "confidence": 0.9 },
"vital_signs": { "breathing_rate_bpm": 14.0, "heart_rate_bpm": 60.0 },
"persons": [{}, {}],
"nodes": [
{ "node_id": 1, "rssi_dbm": -40.0,
"classification": { "presence": true, "motion_level": "walking", "confidence": 0.8 } },
{ "node_id": 2, "rssi_dbm": -70.0,
"classification": { "presence": false, "motion_level": "absent", "confidence": 0.1 } }
]
});
let snaps = vitals_snapshots_from_sensing_json(&v, "ruview");
assert_eq!(snaps.len(), 2, "one snapshot per node");
let n1 = snaps.iter().find(|s| s.node_id == "ruview-node1").unwrap();
let n2 = snaps.iter().find(|s| s.node_id == "ruview-node2").unwrap();
assert!(n1.presence && n1.motion > 0.0, "node1 present + moving");
assert!(
!n2.presence && n2.motion == 0.0,
"node2 must be absent — not inherit the room aggregate"
);
assert_eq!(n1.rssi_dbm, Some(-40.0));
assert_eq!(n2.rssi_dbm, Some(-70.0));
assert_eq!(n1.n_persons, 2);
assert_eq!(n2.n_persons, 2);
assert_eq!(n1.breathing_rate_bpm, Some(14.0));
assert_eq!(n2.heartrate_bpm, Some(60.0));
assert!(n1.presence_score > 0.0);
assert_eq!(n2.presence_score, 0.0);
}
#[test]
fn per_node_missing_fields_fall_back_to_aggregate() {
let v = json!({
"timestamp": 1.0,
"classification": { "presence": true, "motion_level": "still", "confidence": 0.7 },
"vital_signs": {},
"nodes": [ { "node_id": 3, "rssi_dbm": -55.0 } ] });
let snaps = vitals_snapshots_from_sensing_json(&v, "n");
assert_eq!(snaps.len(), 1);
assert_eq!(snaps[0].node_id, "n-node3");
assert!(snaps[0].presence, "defers to aggregate presence");
assert_eq!(snaps[0].motion, 0.0, "aggregate 'still' => no motion");
}
#[test]
fn falls_back_to_single_aggregate_when_no_nodes() {
let v = json!({
"timestamp": 2.0,
"classification": { "presence": true, "motion_level": "idle", "confidence": 0.6 },
"vital_signs": { "breathing_rate_bpm": 12.0 },
"persons": [{}]
});
let snaps = vitals_snapshots_from_sensing_json(&v, "ruview");
assert_eq!(snaps.len(), 1);
assert_eq!(snaps[0].node_id, "ruview");
assert!(snaps[0].presence);
assert_eq!(snaps[0].motion, 0.0, "idle => no motion");
assert_eq!(snaps[0].n_persons, 1);
}
#[test]
fn absent_motion_level_is_zero_motion() {
let v = json!({
"timestamp": 0.0,
"classification": { "presence": false, "motion_level": "absent", "confidence": 0.0 },
"vital_signs": {}
});
let snaps = vitals_snapshots_from_sensing_json(&v, "x");
assert_eq!(snaps[0].motion, 0.0);
assert!(!snaps[0].presence);
}
}
#[cfg(test)]
mod model_load_diagnostic_tests {
use super::diagnose_model_load_error;
use std::path::Path;
#[test]
fn safetensors_is_named_and_points_at_894() {
let data = [0x10, 0, 0, 0, 0, 0, 0, 0, b'{', b'"'];
let msg = diagnose_model_load_error(
Path::new("models/wifi-densepose-pretrained/model.safetensors"),
&data,
"invalid magic at offset 0",
);
assert!(msg.contains("safetensors"), "{msg}");
assert!(msg.contains("#894"), "{msg}");
assert!(msg.contains("signal heuristics"), "{msg}");
}
#[test]
fn quantized_bin_is_identified() {
let data = [0x35, 0x57, 0x45, 0x77]; let msg = diagnose_model_load_error(Path::new("model-q4.bin"), &data, "bad magic");
assert!(msg.contains("quantized weight blob"), "{msg}");
assert!(msg.contains("RVFS") || msg.contains("0x52564653"), "{msg}");
}
#[test]
fn jsonl_manifest_is_identified() {
let data = *b"{\"seg\":0}";
let msg = diagnose_model_load_error(Path::new("model.rvf.jsonl"), &data, "x");
assert!(msg.contains("JSONL manifest"), "{msg}");
}
#[test]
fn unknown_format_still_gives_guidance() {
let data = [0u8, 1, 2, 3];
let msg = diagnose_model_load_error(Path::new("weird.dat"), &data, "x");
assert!(msg.contains("RVF binary container"), "{msg}");
assert!(msg.contains("wifi-densepose-train"), "{msg}");
}
}
#[cfg(test)]
mod export_rvf_mode_tests {
use super::export_emits_placeholder_demo;
#[test]
fn standalone_export_emits_placeholder() {
assert!(export_emits_placeholder_demo(true, false, false));
}
#[test]
fn export_with_train_does_not_short_circuit() {
assert!(!export_emits_placeholder_demo(true, true, false));
assert!(!export_emits_placeholder_demo(true, false, true));
assert!(!export_emits_placeholder_demo(true, true, true));
}
#[test]
fn no_export_flag_never_emits() {
assert!(!export_emits_placeholder_demo(false, false, false));
assert!(!export_emits_placeholder_demo(false, true, false));
}
}