From 559528155f66f73b5b254283cf0afff750a525e1 Mon Sep 17 00:00:00 2001 From: Mathew Beane Date: Fri, 2 Oct 2026 03:14:33 -0400 Subject: [PATCH] fix(sensing-server): train the adaptive classifier per node, as it runs Runtime classification is per node, from that node's own features and amplitudes. The trainer instead built one sample per recorded line from the fused frame-level features plus the first node's amplitudes, so a model trained on a multi-node recording learned a different input distribution from the one it is later asked to classify. - load_recording now emits one sample per node per new frame, from that node's node_features entry and amplitude vector. A node contributes only when its amplitude vector changed since the previous line (every broadcast snapshots all nodes), and stale nodes are skipped. Lines without node_features (older single-node recordings) fall back to the frame-level features. - Runtime subcarrier statistics use the same 56-subcarrier window that sensing_update, and therefore every recording, stores (RECORDED_AMPLITUDE_LEN, now shared with the broadcast in main.rs). - The trainer stamps model version 2 (TRAINED_MODEL_VERSION), so a freshly trained model consumes the measured temporal dominant frequency (ADR-356) instead of being switched to the legacy subcarrier proxy at runtime. - The training accuracy log line is labelled in-sample and printed as a percentage (it printed the 0..1 fraction with a % sign). Tests: cargo test -p wifi-densepose-sensing-server --no-default-features: bin 337 passed (5 new node_sample_tests), lib 539 passed, integration tests pass. Clippy reports nothing on the changed lines. Co-Authored-By: claude-flow --- .../src/adaptive_classifier.rs | 165 +++++++++++++++++- .../wifi-densepose-sensing-server/src/main.rs | 7 +- 2 files changed, 162 insertions(+), 10 deletions(-) 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 {