diff --git a/v2/crates/wifi-densepose-sensing-server/src/adaptive_classifier.rs b/v2/crates/wifi-densepose-sensing-server/src/adaptive_classifier.rs index f5baecbf7..a4db93c2e 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/adaptive_classifier.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/adaptive_classifier.rs @@ -27,6 +27,65 @@ const N_FEATURES: usize = 15; /// Default class names for backward compatibility with old saved models. const DEFAULT_CLASSES: &[&str] = &["absent", "present_still", "present_moving", "active"]; +/// Model version written by the trainer. Version 2 models consume the measured +/// temporal `dominant_freq_hz` that recordings carry (ADR-356); stamping 1 here +/// would make the server swap in the legacy subcarrier proxy at runtime. +pub const TRAINED_MODEL_VERSION: u32 = 2; + +/// Subcarrier amplitudes kept per node in the broadcast `sensing_update`, and +/// therefore in every recording. Runtime classification truncates to the same +/// length so the subcarrier statistics match what training saw. +pub const RECORDED_AMPLITUDE_LEN: usize = 56; + +/// Extract one feature vector per node from a recorded `sensing_update` line. +/// +/// Runtime classification is per node, from that node's own features and +/// amplitudes. Each broadcast carries a snapshot of every node, so a node only +/// contributes when its amplitude vector changed since the previous line +/// (`last_amps`), i.e. when it delivered a new frame. Stale nodes are skipped. +/// Lines without `node_features` (older single-node recordings) fall back to +/// the frame-level features and the first node. +pub fn node_samples_from_frame( + frame: &serde_json::Value, + last_amps: &mut HashMap>, +) -> Vec<[f64; N_FEATURES]> { + let Some(per_node) = frame.get("node_features").and_then(|n| n.as_array()) else { + return vec![features_from_frame(frame)]; + }; + let nodes = frame + .get("nodes") + .and_then(|n| n.as_array()) + .map(Vec::as_slice) + .unwrap_or(&[]); + let mut samples = Vec::new(); + for entry in per_node { + if entry + .get("stale") + .and_then(|v| v.as_bool()) + .unwrap_or(false) + { + continue; + } + let Some(node_id) = entry.get("node_id").and_then(|v| v.as_u64()) else { + continue; + }; + let amps: Vec = nodes + .iter() + .find(|n| n.get("node_id").and_then(|v| v.as_u64()) == Some(node_id)) + .and_then(|n| n.get("amplitude")) + .and_then(|a| a.as_array()) + .map(|arr| arr.iter().filter_map(|v| v.as_f64()).collect()) + .unwrap_or_default(); + if amps.is_empty() || last_amps.get(&node_id) == Some(&s) { + continue; + } + let feat = entry.get("features").cloned().unwrap_or_default(); + samples.push(features_from_runtime(&feat, &s)); + last_amps.insert(node_id, amps); + } + samples +} + /// Extract extended feature vector from a JSONL frame (features + raw amplitudes). pub fn features_from_frame(frame: &serde_json::Value) -> [f64; N_FEATURES] { let feat = frame @@ -118,6 +177,7 @@ pub fn features_from_runtime(feat: &serde_json::Value, amps: &[f64]) -> [f64; N_ .get("mean_rssi") .and_then(|v| v.as_f64()) .unwrap_or(0.0); + let amps = &s[..amps.len().min(RECORDED_AMPLITUDE_LEN)]; let (amp_mean, amp_std, amp_skew, amp_kurt, amp_iqr, amp_entropy, amp_max, amp_range) = subcarrier_stats(amps); [ @@ -349,15 +409,14 @@ fn load_recording(path: &Path, class_idx: usize) -> Vec { Ok(c) => c, Err(_) => return Vec::new(), }; + let mut last_amps = HashMap::new(); content .lines() - .filter_map(|line| { - let v: serde_json::Value = serde_json::from_str(line).ok()?; - // Use extended features (server features + subcarrier stats). - Some(Sample { - features: features_from_frame(&v), - class_idx, - }) + .filter_map(|line| serde_json::from_str::(line).ok()) + .flat_map(|v| node_samples_from_frame(&v, &mut last_amps)) + .map(|features| Sample { + features, + class_idx, }) .collect() } @@ -630,7 +689,11 @@ pub fn train_from_recordings(recordings_dir: &Path) -> Result Result PathBuf { PathBuf::from("data/adaptive_model.json") } +#[cfg(test)] +mod node_sample_tests { + use super::*; + use serde_json::json; + + fn node_features(node_id: u64, variance: f64, stale: bool) -> serde_json::Value { + json!({ + "node_id": node_id, + "features": { "variance": variance, "mean_rssi": -50.0, "change_points": 3 }, + "stale": stale, + }) + } + + fn node(node_id: u64, amplitude: &[f64]) -> serde_json::Value { + json!({ "node_id": node_id, "amplitude": amplitude }) + } + + #[test] + fn one_sample_per_node_using_that_nodes_features() { + let frame = json!({ + "features": { "variance": 999.0 }, + "nodes": [node(1, &[1.0, 2.0]), node(2, &[3.0, 4.0])], + "node_features": [node_features(1, 10.0, false), node_features(2, 20.0, false)], + }); + let samples = node_samples_from_frame(&frame, &mut HashMap::new()); + let mut variances: Vec = samples.iter().map(|s| s[0]).collect(); + variances.sort_by(|a, b| a.partial_cmp(b).unwrap()); + assert_eq!(variances, vec![10.0, 20.0], "per-node, never the fused 999"); + } + + #[test] + fn repeated_snapshot_of_an_unchanged_node_is_not_resampled() { + let mut last = HashMap::new(); + let first = json!({ + "nodes": [node(1, &[1.0, 2.0]), node(2, &[3.0, 4.0])], + "node_features": [node_features(1, 10.0, false), node_features(2, 20.0, false)], + }); + // Node 2 delivered a new frame; node 1's entry is the same snapshot. + let second = json!({ + "nodes": [node(1, &[1.0, 2.0]), node(2, &[5.0, 6.0])], + "node_features": [node_features(1, 10.0, false), node_features(2, 21.0, false)], + }); + assert_eq!(node_samples_from_frame(&first, &mut last).len(), 2); + let again = node_samples_from_frame(&second, &mut last); + assert_eq!(again.len(), 1); + assert_eq!(again[0][0], 21.0); + } + + #[test] + fn stale_and_amplitude_less_nodes_are_skipped() { + let frame = json!({ + "nodes": [node(1, &[1.0, 2.0]), node(2, &[])], + "node_features": [node_features(1, 10.0, true), node_features(2, 20.0, false)], + }); + assert!(node_samples_from_frame(&frame, &mut HashMap::new()).is_empty()); + } + + #[test] + fn recordings_without_node_features_fall_back_to_frame_level() { + let frame = json!({ + "features": { "variance": 7.0 }, + "nodes": [node(1, &[1.0, 2.0])], + }); + let samples = node_samples_from_frame(&frame, &mut HashMap::new()); + assert_eq!(samples.len(), 1); + assert_eq!(samples[0][0], 7.0); + } + + #[test] + fn runtime_uses_the_same_amplitude_window_as_recordings() { + let mut long = vec![1.0; RECORDED_AMPLITUDE_LEN]; + long.extend([100.0; 144]); + let short = vec![1.0; RECORDED_AMPLITUDE_LEN]; + let feat = json!({}); + assert_eq!( + features_from_runtime(&feat, &long), + features_from_runtime(&feat, &short) + ); + } + + // The trainer must stamp a version that consumes measured frequency. + const _: () = assert!(TRAINED_MODEL_VERSION >= 2); +} + #[cfg(test)] mod compatibility_tests { use super::compatible_dominant_frequency; diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 180ae6f53..20b9444b5 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -10464,7 +10464,12 @@ async fn udp_receiver_task( } else { n.frame_history .back() - .map(|a| a.iter().take(56).cloned().collect()) + .map(|a| { + a.iter() + .take(adaptive_classifier::RECORDED_AMPLITUDE_LEN) + .cloned() + .collect() + }) .unwrap_or_default() }, subcarrier_count: if suppress_raw {