From ffe49d7c66001d7b576bb52722a8b0d19bff2f57 Mon Sep 17 00:00:00 2001 From: ruv Date: Tue, 9 Jun 2026 12:37:41 -0400 Subject: [PATCH] =?UTF-8?q?feat(cli):=20multistatic=20room-watch=20?= =?UTF-8?q?=E2=80=94=20fuse=20co-located=20nodes=20(ADR-029/151)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `room-watch --node-bank N:path` (repeatable) groups live CSI frames by node_id and fuses per-node banks via MultiNodeMixture. Validated live on COM8 (node 9, edge_tier=0): frames grouped + fused end-to-end. True 2-node fusion is covered by unit tests; a second raw-CSI node is the hardware blocker. 54 tests pass. Co-Authored-By: claude-flow --- CHANGELOG.md | 3 +- v2/crates/wifi-densepose-cli/src/room.rs | 98 +++++++++++++++++++++++- 2 files changed, 97 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9872b04a..9ea90138 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,7 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - **ADR-151 per-room calibration & specialist training — full `baseline → enroll → extract → train` pipeline (new `wifi-densepose-calibration` crate).** "Teach the room before you teach the model": a local-first pipeline that turns a few minutes of clean human anchors — layered on the ADR-135 empty-room baseline — into a versioned bank of small, room-calibrated specialists for **presence, posture, breathing, heartbeat, restlessness, and anomaly**. Stages: guided enrollment with an adaptive quality gate (event-sourced `EnrollmentSession`, re-prompts bad anchors); feature extraction (autocorrelation periodicity in breathing/HR bands + variance/motion); six small specialists (learned threshold / nearest-prototype / band-limited periodicity / novelty); a `SpecialistBank` with baseline-drift **STALE** invalidation; and a `MixtureOfSpecialists` runtime with presence short-circuit + anomaly veto + confidence gating. Specialists are statistical heads today (runnable + hardware-validated); the frozen ADR-150 HF RF Foundation Encoder backbone is the documented upgrade path. - **CLI:** `enroll` / `train-room` / `room-status` / `room-watch`, plus the Stage-1 `calibrate-serve` HTTP API (CORS-enabled: `POST /start`, `GET /status`, `POST /stop`, `GET /result`, `GET /baselines`, `GET /health`) and a firewall-free `scripts/csi-udp-relay.py` for local Windows ESP32 testing without admin. - - **Validated on live ESP32-S3 (COM8, `edge_tier=0` raw CSI):** baseline capture (120 frames → 52-subcarrier baseline); and the real parser → feature-extraction → mixture runtime detecting breathing (~16–31 BPM) end-to-end. Full multi-anchor enrollment accuracy requires the operator to perform the poses; phase-based (vs amplitude) breathing extraction + RVF/HNSW storage are noted refinements. 48 tests pass (29 calibration + 19 CLI). + - **Multistatic fusion (ADR-029):** `MultiNodeMixture` fuses several co-located nodes (each with its own room-calibrated bank) into one room state — presence OR'd across nodes, posture/breathing/heartbeat from the highest-confidence node, a single implausible node vetoes the room's vitals. Driven via `room-watch --node-bank N:path` (repeatable), which groups live frames by `node_id` and fuses. Same-room only; cross-room is federation (ADR-105). + - **Validated on live ESP32-S3 (COM8, `edge_tier=0` raw CSI):** baseline capture (120 frames → 52-subcarrier baseline); the real parser → feature-extraction → mixture runtime detecting breathing (~16–31 BPM); and the multistatic ingest grouping/fusing by node-id end-to-end. Full multi-anchor enrollment accuracy requires the operator to perform the poses; true 2-node fusion + phase-based breathing + RVF/HNSW storage are noted follow-ups. 54 tests pass (35 calibration + 19 CLI). - **WiFi-CSI pose: efficiency frontier + per-room calibration service** (ADR-150 §3.2–3.6). Two beyond-SOTA results on the MM-Fi benchmark, plus the deployment mechanism that resolves real-world generalization: - **Efficiency frontier** — a **75 K-param model beats published SOTA** (74.3% vs MultiFormer 72.25% torso-PCK@20); every config from `micro` up is Pareto-dominant (smaller *and* more accurate than prior work). Shipped a deployable **int4 edge model (~20 KB, verified 74.08%, 0.135 ms single-thread CPU)** — published at [`ruvnet/wifi-densepose-mmfi-pose/edge`](https://huggingface.co/ruvnet/wifi-densepose-mmfi-pose). See [`docs/benchmarks/wifi-pose-efficiency-frontier.md`](docs/benchmarks/wifi-pose-efficiency-frontier.md). - **Generalization solved by few-shot calibration** — zero-shot cross-subject (~64%) and cross-environment (~10%) are *not* closeable by algorithms (CORAL, DANN, instance-norm, contrastive foundation-pretraining all tested, all failed) or by more training subjects (saturates ~64%). But **~100–200 labeled in-room samples recover SOTA-level pose**: cross-subject 64→76%, **cross-environment 10→73% (60% from just 5 samples)** — deployable as a **~11 KB per-room LoRA adapter** on a frozen shared base. Full empirical chain in ADR-150 §3.2–3.6. diff --git a/v2/crates/wifi-densepose-cli/src/room.rs b/v2/crates/wifi-densepose-cli/src/room.rs index 4ebbbb2b..1f010eae 100644 --- a/v2/crates/wifi-densepose-cli/src/room.rs +++ b/v2/crates/wifi-densepose-cli/src/room.rs @@ -11,7 +11,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::net::UdpSocket; use wifi_densepose_calibration::{ Anchor, AnchorLabel, AnchorQualityGate, AnchorRecorder, EnrollmentEvent, EnrollmentSession, - MixtureOfSpecialists, SpecialistBank, + MixtureOfSpecialists, MultiNodeMixture, SpecialistBank, }; use wifi_densepose_calibration::extract::{AnchorFeature, Features}; use wifi_densepose_core::types::CsiFrame; @@ -288,9 +288,13 @@ pub async fn room_status(args: RoomStatusArgs) -> Result<()> { /// Arguments for `room-watch`. #[derive(Args, Debug, Clone)] pub struct RoomWatchArgs { - /// Specialist-bank file. + /// Specialist-bank file (single-node mode). #[arg(long, default_value = "./room-bank.json")] pub bank: String, + /// Multistatic mode: map a node id to its bank as `N:path` (repeatable). + /// When supplied, frames are grouped by node id and fused (ADR-029/151). + #[arg(long = "node-bank", value_name = "N:PATH")] + pub node_bank: Vec, /// UDP port for ESP32 CSI frames (raw CSI). #[arg(long, default_value_t = 5005)] pub udp_port: u16, @@ -311,8 +315,11 @@ pub struct RoomWatchArgs { pub seconds: u32, } -/// Execute `room-watch` — live mixture-of-specialists readout. +/// Execute `room-watch` — live (multistatic) mixture-of-specialists readout. pub async fn room_watch(args: RoomWatchArgs) -> Result<()> { + if !args.node_bank.is_empty() { + return room_watch_multi(args).await; + } let raw = std::fs::read_to_string(&args.bank) .map_err(|e| anyhow::anyhow!("cannot read {}: {e}", args.bank))?; let bank = SpecialistBank::from_json(&raw).map_err(|e| anyhow::anyhow!("{e}"))?; @@ -364,3 +371,88 @@ pub async fn room_watch(args: RoomWatchArgs) -> Result<()> { } Ok(()) } + +/// Multistatic `room-watch`: fuse several co-located nodes (ADR-029/151). +async fn room_watch_multi(args: RoomWatchArgs) -> Result<()> { + use std::collections::{BTreeMap, VecDeque}; + + let mut mix = MultiNodeMixture::new(); + let mut node_ids: Vec = Vec::new(); + for spec in &args.node_bank { + let (id_s, path) = spec + .split_once(':') + .ok_or_else(|| anyhow::anyhow!("--node-bank must be N:path (got {spec:?})"))?; + let id: u8 = id_s + .parse() + .map_err(|_| anyhow::anyhow!("bad node id in {spec:?}"))?; + let raw = std::fs::read_to_string(path) + .map_err(|e| anyhow::anyhow!("cannot read {path}: {e}"))?; + let bank = SpecialistBank::from_json(&raw).map_err(|e| anyhow::anyhow!("{e}"))?; + let baseline = bank.baseline_id.clone(); + mix.add_node(id, bank, baseline); + node_ids.push(id); + } + eprintln!("[room-watch] multistatic over nodes {node_ids:?}"); + + let addr = format!("{}:{}", args.bind, args.udp_port); + let socket = UdpSocket::bind(&addr) + .await + .map_err(|e| anyhow::anyhow!("cannot bind {addr}: {e}"))?; + eprintln!("[room-watch] fusing on udp://{addr} (window={} frames)", args.window); + + let mut buf = vec![0u8; RECV_BUF]; + let mut wins: BTreeMap> = BTreeMap::new(); + let start = Instant::now(); + let mut last_print = Instant::now(); + + loop { + if args.seconds > 0 && start.elapsed() >= Duration::from_secs(args.seconds as u64) { + break; + } + if let Ok(Ok(n)) = + tokio::time::timeout(Duration::from_millis(500), socket.recv(&mut buf)).await + { + if n < 5 { + continue; + } + let node_id = buf[4]; + if !node_ids.contains(&node_id) { + continue; + } + if let Some(frame) = parse_csi_packet(&buf[..n], &args.tier) { + let w = wins.entry(node_id).or_default(); + w.push_back(frame_scalar(&frame)); + while w.len() > args.window { + w.pop_front(); + } + } + } + if last_print.elapsed() >= Duration::from_secs(1) { + let per_node: BTreeMap = wins + .iter() + .filter(|(_, w)| w.len() >= 32) + .map(|(id, w)| { + let series: Vec = w.iter().copied().collect(); + (*id, Features::from_series(&series, args.fs_hz)) + }) + .collect(); + if !per_node.is_empty() { + let active: Vec = per_node.keys().copied().collect(); + let s = mix.infer(&per_node); + let pres = s.presence.as_ref().and_then(|r| r.label.clone()).unwrap_or("-".into()); + let post = s.posture.as_ref().and_then(|r| r.label.clone()).unwrap_or("-".into()); + let br = s.breathing.as_ref().map(|r| format!("{:.1}bpm", r.value)).unwrap_or("-".into()); + let flags = format!( + "{}{}", + if s.vetoed { " VETO" } else { "" }, + if s.stale { " STALE" } else { "" } + ); + println!( + "nodes={active:?} presence={pres:<7} posture={post:<8} breathing={br:<8}{flags}" + ); + } + last_print = Instant::now(); + } + } + Ok(()) +}