Merge 818c4ac7e2 into 2c136aca74

CHANGELOG.md

@@ -245,6 +245,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   optional Lindblad/Hamiltonian extension if AC magnetometry, MW power
   saturation, hyperfine spectroscopy, or pulsed protocols become required.
 
+### Improved
+- **Vital signs (heart rate + breathing rate) now flow through the 4-stage `wifi-densepose-vitals` pipeline** (IIR + autocorrelation) with confidence-based fallback to the legacy FFT heuristic. When crate confidence \u2265 0.05 the `CrateVitalsPipeline` result is used; below that threshold the existing FFT detector takes over, preserving availability under poor signal conditions. Reduces HR jitter (previously \u00b115 BPM minute-to-minute) and unstable confidence readings.
+
 ### Fixed
 - **WebSocket broadcast handler now handles Lagged events gracefully and sends periodic ping keepalives to prevent dashboard disconnects** —
   `handle_ws_client` and `handle_ws_pose_client` in `wifi-densepose-sensing-server`

v2/crates/wifi-densepose-sensing-server/Cargo.toml

@@ -88,6 +88,9 @@ mqtt = ["dep:rumqttc"]
 # surface small until the SDK choice is validated.
 matter = []
 
+# 4-stage vital signs pipeline (IIR + autocorrelation) — issue #44 / ADR-045.
+wifi-densepose-vitals = { version = "0.3.0", path = "../wifi-densepose-vitals" }
+
 [dev-dependencies]
 tempfile = "3.10"
 # `tower::ServiceExt::oneshot` for in-process Router tests (bearer_auth).

v2/crates/wifi-densepose-sensing-server/src/types.rs

@@ -10,7 +10,7 @@ use tokio::sync::{broadcast, RwLock};
 use crate::adaptive_classifier;
 use crate::rvf_container::RvfContainerInfo;
 use crate::rvf_pipeline::ProgressiveLoader;
-use crate::vital_signs::{VitalSignDetector, VitalSigns};
+use crate::vital_signs::{CrateVitalsPipeline, VitalSigns};
 
 use wifi_densepose_signal::ruvsense::field_model::FieldModel;
 use wifi_densepose_signal::ruvsense::longitudinal::{EmbeddingEntry, EmbeddingHistory};
@@ -264,7 +264,7 @@ pub struct NodeState {
     pub hr_buffer: VecDeque<f64>,
     pub br_buffer: VecDeque<f64>,
     pub rssi_history: VecDeque<f64>,
-    pub vital_detector: VitalSignDetector,
+    pub vital_detector: CrateVitalsPipeline,
     pub latest_vitals: VitalSigns,
     pub last_frame_time: Option<std::time::Instant>,
     pub edge_vitals: Option<Esp32VitalsPacket>,
@@ -308,7 +308,7 @@ impl NodeState {
             hr_buffer: VecDeque::with_capacity(8),
             br_buffer: VecDeque::with_capacity(8),
             rssi_history: VecDeque::new(),
-            vital_detector: VitalSignDetector::new(10.0),
+            vital_detector: CrateVitalsPipeline::new(10.0, 56),
             latest_vitals: VitalSigns::default(),
             last_frame_time: None,
             edge_vitals: None,
@@ -413,7 +413,7 @@ pub struct AppStateInner {
     pub tx: broadcast::Sender<String>,
     pub total_detections: u64,
     pub start_time: std::time::Instant,
-    pub vital_detector: VitalSignDetector,
+    pub vital_detector: CrateVitalsPipeline,
     pub latest_vitals: VitalSigns,
     pub rvf_info: Option<RvfContainerInfo>,
     pub save_rvf_path: Option<PathBuf>,

v2/crates/wifi-densepose-sensing-server/src/vital_signs.rs

@@ -618,6 +618,149 @@ pub fn run_benchmark(n_frames: usize) -> (std::time::Duration, std::time::Durati
     (total, per_frame)
 }
 
+// ── Crate-backed vitals pipeline (issue #44) ─────────────────────────────
+
+/// Enhanced vital sign pipeline using `wifi-densepose-vitals` crate.
+///
+/// Wraps the crate's 4-stage pipeline (preprocessor -> breathing extractor ->
+/// heart rate extractor -> anomaly detection) and converts output to the
+/// server's `VitalSigns` format. Falls back to `VitalSignDetector` on error.
+#[allow(dead_code)]
+pub struct CrateVitalsPipeline {
+    preprocessor: wifi_densepose_vitals::CsiVitalPreprocessor,
+    breathing: wifi_densepose_vitals::BreathingExtractor,
+    heartrate: wifi_densepose_vitals::HeartRateExtractor,
+    anomaly: wifi_densepose_vitals::VitalAnomalyDetector,
+    sample_index: u64,
+    sample_rate: f64,
+    n_subcarriers: usize,
+    /// FFT-based fallback detector.
+    fallback: VitalSignDetector,
+}
+
+impl CrateVitalsPipeline {
+    /// Create a new pipeline with given sample rate and subcarrier count.
+    pub fn new(sample_rate: f64, n_subcarriers: usize) -> Self {
+        Self {
+            preprocessor: wifi_densepose_vitals::CsiVitalPreprocessor::new(
+                n_subcarriers, 0.05,
+            ),
+            breathing: wifi_densepose_vitals::BreathingExtractor::new(
+                n_subcarriers, sample_rate, 30.0,
+            ),
+            heartrate: wifi_densepose_vitals::HeartRateExtractor::new(
+                n_subcarriers, sample_rate, 15.0,
+            ),
+            anomaly: wifi_densepose_vitals::VitalAnomalyDetector::default_config(),
+            sample_index: 0,
+            sample_rate,
+            n_subcarriers,
+            fallback: VitalSignDetector::new(sample_rate),
+        }
+    }
+
+    /// Process one CSI frame and return vital signs.
+    ///
+    /// Uses the crate's multi-stage pipeline (EMA preprocessing, IIR bandpass,
+    /// phase-coherence weighted HR extraction). Falls back to the FFT-based
+    /// `VitalSignDetector` if the crate pipeline produces no estimates.
+    pub fn process_frame(&mut self, amplitude: &[f64], phase: &[f64]) -> VitalSigns {
+        let fallback_result = self.fallback.process_frame(amplitude, phase);
+
+        let n_sub = amplitude.len().min(phase.len());
+        if n_sub == 0 {
+            return fallback_result;
+        }
+
+        let frame = wifi_densepose_vitals::CsiFrame {
+            amplitudes: amplitude.to_vec(),
+            phases: phase.to_vec(),
+            n_subcarriers: n_sub,
+            sample_index: self.sample_index,
+            sample_rate_hz: self.sample_rate,
+        };
+        self.sample_index += 1;
+
+        let residuals = match self.preprocessor.process(&frame) {
+            Some(r) => r,
+            None => return fallback_result,
+        };
+
+        let uniform_w = 1.0 / n_sub as f64;
+        let weights = vec![uniform_w; n_sub];
+
+        let rr = self.breathing.extract(&residuals, &weights);
+        let hr = self.heartrate.extract(&residuals, &frame.phases);
+
+        // Build a VitalReading for anomaly detection
+        let reading = wifi_densepose_vitals::VitalReading {
+            respiratory_rate: rr
+                .clone()
+                .unwrap_or_else(wifi_densepose_vitals::VitalEstimate::unavailable),
+            heart_rate: hr
+                .clone()
+                .unwrap_or_else(wifi_densepose_vitals::VitalEstimate::unavailable),
+            subcarrier_count: n_sub,
+            signal_quality: fallback_result.signal_quality,
+            timestamp_secs: self.sample_index as f64 / self.sample_rate,
+        };
+
+        let _alerts = self.anomaly.check(&reading);
+
+        // Convert crate output to server's VitalSigns, falling back per-field
+        let breathing_rate_bpm = rr
+            .as_ref()
+            .filter(|e| e.confidence > 0.05)
+            .map(|e| e.value_bpm)
+            .or(fallback_result.breathing_rate_bpm);
+
+        let breathing_confidence = rr
+            .as_ref()
+            .filter(|e| e.confidence > 0.05)
+            .map(|e| e.confidence)
+            .unwrap_or(fallback_result.breathing_confidence);
+
+        let heart_rate_bpm = hr
+            .as_ref()
+            .filter(|e| e.confidence > 0.05)
+            .map(|e| e.value_bpm)
+            .or(fallback_result.heart_rate_bpm);
+
+        let heartbeat_confidence = hr
+            .as_ref()
+            .filter(|e| e.confidence > 0.05)
+            .map(|e| e.confidence)
+            .unwrap_or(fallback_result.heartbeat_confidence);
+
+        // Use crate's signal quality from the reading (backed by fallback's
+        // amplitude-statistics quality for now; the crate's subcarrier-count
+        // and confidence data augment this in future iterations).
+        let signal_quality = fallback_result.signal_quality;
+
+        VitalSigns {
+            breathing_rate_bpm,
+            heart_rate_bpm,
+            breathing_confidence,
+            heartbeat_confidence,
+            signal_quality,
+        }
+    }
+
+    /// Reset all internal state.
+    pub fn reset(&mut self) {
+        self.preprocessor.reset();
+        self.breathing.reset();
+        self.heartrate.reset();
+        self.fallback.reset();
+        self.sample_index = 0;
+    }
+
+    /// Buffer status from the fallback detector (for diagnostics).
+    pub fn buffer_status(&self) -> (usize, usize, usize, usize) {
+        self.fallback.buffer_status()
+    }
+}
+
 // ── Tests ──────────────────────────────────────────────────────────────────
 
 #[cfg(test)]