From 418552d8dc7d0e7807df0a978299cf36c0c913da Mon Sep 17 00:00:00 2001 From: ruv Date: Tue, 9 Jun 2026 12:15:01 -0400 Subject: [PATCH] feat(cli): wire ADR-151 enroll / train-room / room-status / room-watch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrates the wifi-densepose-calibration crate into the CLI as four subcommands driving the full Stage 2–5 pipeline against a live ESP32 raw-CSI stream (edge_tier=0): - enroll: walks the guided anchor sequence, gates each capture against the ADR-135 baseline deviation (re-prompts bad anchors), writes labelled features - train-room: fits the SpecialistBank from the enrollment, persists JSON - room-status: prints a trained bank's summary - room-watch: live mixture-of-specialists readout (presence/posture/breathing/ heart/restless) over a rolling window, with anomaly veto + STALE flagging Per-frame scalar is the mean CSI amplitude (carries presence/motion + breathing modulation). Validated end-to-end on the live ESP32 (COM8, edge_tier=0): the real parser → feature extraction → runtime detected breathing (~16–31 BPM) on hardware. Full multi-anchor enrollment accuracy requires the operator to perform the poses; phase-based breathing extraction is a noted refinement. 48 tests pass (29 calibration + 19 CLI). Co-Authored-By: claude-flow --- v2/Cargo.lock | 15 + v2/crates/wifi-densepose-cli/Cargo.toml | 1 + v2/crates/wifi-densepose-cli/src/lib.rs | 14 + v2/crates/wifi-densepose-cli/src/main.rs | 12 + v2/crates/wifi-densepose-cli/src/room.rs | 362 +++++++++++++++++++++++ 5 files changed, 404 insertions(+) create mode 100644 v2/crates/wifi-densepose-cli/src/room.rs diff --git a/v2/Cargo.lock b/v2/Cargo.lock index fe7ff2bb..0c75d899 100644 --- a/v2/Cargo.lock +++ b/v2/Cargo.lock @@ -10811,6 +10811,20 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "wifi-densepose-calibration" +version = "0.3.0" +dependencies = [ + "ndarray 0.17.2", + "num-complex", + "serde", + "serde_json", + "thiserror 2.0.18", + "uuid", + "wifi-densepose-core", + "wifi-densepose-signal", +] + [[package]] name = "wifi-densepose-cli" version = "0.3.0" @@ -10837,6 +10851,7 @@ dependencies = [ "tracing", "tracing-subscriber", "uuid", + "wifi-densepose-calibration", "wifi-densepose-core", "wifi-densepose-mat", "wifi-densepose-signal", diff --git a/v2/crates/wifi-densepose-cli/Cargo.toml b/v2/crates/wifi-densepose-cli/Cargo.toml index 215df58b..c794a50d 100644 --- a/v2/crates/wifi-densepose-cli/Cargo.toml +++ b/v2/crates/wifi-densepose-cli/Cargo.toml @@ -24,6 +24,7 @@ mat = [] wifi-densepose-mat = { version = "0.3.0", path = "../wifi-densepose-mat" } wifi-densepose-signal = { version = "0.3.1", path = "../wifi-densepose-signal", default-features = false } wifi-densepose-core = { version = "0.3.0", path = "../wifi-densepose-core" } +wifi-densepose-calibration = { version = "0.3.0", path = "../wifi-densepose-calibration" } # Linear algebra / complex numbers (used by calibrate.rs to build CsiFrame) ndarray = { workspace = true } diff --git a/v2/crates/wifi-densepose-cli/src/lib.rs b/v2/crates/wifi-densepose-cli/src/lib.rs index bb043f8d..9ce1c8e7 100644 --- a/v2/crates/wifi-densepose-cli/src/lib.rs +++ b/v2/crates/wifi-densepose-cli/src/lib.rs @@ -28,6 +28,7 @@ use clap::{Parser, Subcommand}; pub mod calibrate; pub mod calibrate_api; +pub mod room; pub mod mat; /// WiFi-DensePose Command Line Interface @@ -58,6 +59,19 @@ pub enum Commands { /// endpoints at `/api/v1/calibration/*` (CORS-enabled). CalibrateServe(calibrate_api::CalibrateServeArgs), + /// Guided per-room enrollment (ADR-151 Stage 2) — walk the anchor sequence + /// against a baseline, writing labelled features. + Enroll(room::EnrollArgs), + + /// Train the per-room specialist bank from an enrollment (ADR-151 Stage 4). + TrainRoom(room::TrainRoomArgs), + + /// Show a trained specialist bank's summary. + RoomStatus(room::RoomStatusArgs), + + /// Live mixture-of-specialists readout from the CSI stream (ADR-151 Stage 5). + RoomWatch(room::RoomWatchArgs), + /// Mass Casualty Assessment Tool commands #[command(subcommand)] Mat(mat::MatCommand), diff --git a/v2/crates/wifi-densepose-cli/src/main.rs b/v2/crates/wifi-densepose-cli/src/main.rs index 806bf586..14a782ae 100644 --- a/v2/crates/wifi-densepose-cli/src/main.rs +++ b/v2/crates/wifi-densepose-cli/src/main.rs @@ -24,6 +24,18 @@ async fn main() -> anyhow::Result<()> { Commands::CalibrateServe(args) => { wifi_densepose_cli::calibrate_api::execute(args).await?; } + Commands::Enroll(args) => { + wifi_densepose_cli::room::enroll(args).await?; + } + Commands::TrainRoom(args) => { + wifi_densepose_cli::room::train_room(args).await?; + } + Commands::RoomStatus(args) => { + wifi_densepose_cli::room::room_status(args).await?; + } + Commands::RoomWatch(args) => { + wifi_densepose_cli::room::room_watch(args).await?; + } Commands::Mat(mat_cmd) => { wifi_densepose_cli::mat::execute(mat_cmd).await?; } diff --git a/v2/crates/wifi-densepose-cli/src/room.rs b/v2/crates/wifi-densepose-cli/src/room.rs new file mode 100644 index 00000000..1949a40d --- /dev/null +++ b/v2/crates/wifi-densepose-cli/src/room.rs @@ -0,0 +1,362 @@ +//! `enroll` / `train-room` / `room-status` / `room-watch` — ADR-151 Stages 2–5 CLI. +//! +//! Drives the `wifi-densepose-calibration` pipeline against a live ESP32 CSI +//! stream (requires `edge_tier=0` raw CSI). `enroll` walks the guided anchors and +//! writes labelled features; `train-room` fits the specialist bank; `room-watch` +//! runs the mixture runtime and prints live room state. + +use anyhow::{bail, Result}; +use clap::Args; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use tokio::net::UdpSocket; +use wifi_densepose_calibration::{ + Anchor, AnchorLabel, AnchorQualityGate, AnchorRecorder, EnrollmentEvent, EnrollmentSession, + MixtureOfSpecialists, SpecialistBank, +}; +use wifi_densepose_calibration::extract::{AnchorFeature, Features}; +use wifi_densepose_core::types::CsiFrame; +use wifi_densepose_signal::BaselineCalibration; + +use crate::calibrate::parse_csi_packet; + +const RECV_BUF: usize = 2048; + +// --------------------------------------------------------------------------- +// Shared helpers +// --------------------------------------------------------------------------- + +fn now_unix() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} + +/// One scalar per frame: mean amplitude across all subcarriers/streams. +/// Carries presence/motion energy and the breathing amplitude modulation. +fn frame_scalar(frame: &CsiFrame) -> f32 { + let a = &frame.amplitude; + if a.is_empty() { + return 0.0; + } + (a.sum() / a.len() as f64) as f32 +} + +fn load_baseline(path: &str) -> Result { + let bytes = std::fs::read(path) + .map_err(|e| anyhow::anyhow!("cannot read baseline {path}: {e} — run `calibrate` first"))?; + BaselineCalibration::from_bytes(&bytes) + .map_err(|e| anyhow::anyhow!("invalid baseline {path}: {e}")) +} + +/// Persisted enrollment output (labelled features + audit log). +#[derive(serde::Serialize, serde::Deserialize)] +struct EnrollmentData { + room_id: String, + baseline_id: String, + fs_hz: f32, + anchors: Vec, + session: EnrollmentSession, +} + +// --------------------------------------------------------------------------- +// enroll +// --------------------------------------------------------------------------- + +/// Arguments for `enroll`. +#[derive(Args, Debug, Clone)] +pub struct EnrollArgs { + /// UDP port for ESP32 CSI frames (raw CSI; provision with `--edge-tier 0`). + #[arg(long, default_value_t = 5005)] + pub udp_port: u16, + /// Bind address for the UDP socket. + #[arg(long, default_value = "0.0.0.0")] + pub bind: String, + /// Path to the empty-room baseline produced by `calibrate`. + #[arg(long, default_value = "./baseline.bin")] + pub baseline: String, + /// PHY tier (ht20 / ht40 / he20 / he40). + #[arg(long, default_value = "ht20")] + pub tier: String, + /// Room label. + #[arg(long, default_value = "default")] + pub room_id: String, + /// Output enrollment file. + #[arg(long, default_value = "./enrollment.json")] + pub output: String, + /// CSI sample rate (Hz) used for periodicity extraction. + #[arg(long, default_value_t = 15.0)] + pub fs_hz: f32, + /// Max attempts per anchor before moving on. + #[arg(long, default_value_t = 2)] + pub attempts: u32, +} + +/// Capture one anchor: returns (accepted feature?, anchor verdict, reason). +async fn capture_anchor( + socket: &UdpSocket, + baseline: &BaselineCalibration, + gate: &AnchorQualityGate, + label: AnchorLabel, + tier: &str, + fs_hz: f32, + room_id: &str, +) -> Result<(Option, Anchor, Option)> { + eprintln!("\n[enroll] {} — {}", label.as_str(), label.prompt()); + for c in (1..=3).rev() { + eprintln!("[enroll] starting in {c}…"); + tokio::time::sleep(Duration::from_secs(1)).await; + } + eprintln!("[enroll] capturing {} s…", label.duration_s()); + + let mut recorder = AnchorRecorder::new(label); + let mut series: Vec = Vec::new(); + let mut buf = vec![0u8; RECV_BUF]; + let deadline = Instant::now() + Duration::from_secs(label.duration_s() as u64); + + while Instant::now() < deadline { + let timeout = Duration::from_millis(500); + if let Ok(Ok(n)) = tokio::time::timeout(timeout, socket.recv(&mut buf)).await { + if let Some(frame) = parse_csi_packet(&buf[..n], tier) { + recorder.record_frame(baseline, &frame); + series.push(frame_scalar(&frame)); + } + } + } + + let (anchor, reason) = recorder.finalize(gate, now_unix()); + let feature = if anchor.quality.accepted { + Some(AnchorFeature::from_series(room_id, label, &series, fs_hz)) + } else { + None + }; + Ok((feature, anchor, reason)) +} + +/// Execute `enroll`. +pub async fn enroll(args: EnrollArgs) -> Result<()> { + let baseline = load_baseline(&args.baseline)?; + let baseline_id = baseline.calibration_uuid().to_string(); + let gate = AnchorQualityGate::default(); + + let addr = format!("{}:{}", args.bind, args.udp_port); + let socket = UdpSocket::bind(&addr) + .await + .map_err(|e| anyhow::anyhow!("cannot bind {addr}: {e}"))?; + eprintln!("[enroll] room='{}' baseline={} on udp://{addr}", args.room_id, &baseline_id[..8]); + eprintln!("[enroll] follow each prompt; bad captures are re-prompted."); + + let mut session = EnrollmentSession::new(&args.room_id, &baseline_id, now_unix()); + let mut features: Vec = Vec::new(); + + for label in AnchorLabel::SEQUENCE { + let mut accepted = false; + for attempt in 1..=args.attempts { + let (feat, anchor, reason) = + capture_anchor(&socket, &baseline, &gate, label, &args.tier, args.fs_hz, &args.room_id) + .await?; + if anchor.quality.accepted { + eprintln!( + "[enroll] ✓ accepted (presence_z={:.2} motion={:.0}% frames={})", + anchor.quality.presence_z, + anchor.quality.motion_rate * 100.0, + anchor.quality.frames + ); + if let Some(f) = feat { + features.push(f); + } + session.apply(EnrollmentEvent::AnchorAccepted { anchor }); + accepted = true; + break; + } else { + let why = reason.unwrap_or_default(); + eprintln!("[enroll] ✗ rejected: {why}"); + session.apply(EnrollmentEvent::AnchorRejected { + label, + reason: why, + at: now_unix(), + }); + if attempt < args.attempts { + eprintln!("[enroll] retrying ({}/{})…", attempt + 1, args.attempts); + } + } + } + if !accepted { + eprintln!("[enroll] moving on without '{}'", label.as_str()); + } + } + + if session.is_complete() { + session.apply(EnrollmentEvent::Completed { at: now_unix() }); + } + let (got, total) = session.progress(); + let data = EnrollmentData { + room_id: args.room_id.clone(), + baseline_id, + fs_hz: args.fs_hz, + anchors: features, + session, + }; + std::fs::write( + &args.output, + serde_json::to_string_pretty(&data).map_err(|e| anyhow::anyhow!("serialize: {e}"))?, + ) + .map_err(|e| anyhow::anyhow!("cannot write {}: {e}", args.output))?; + eprintln!( + "\n[enroll] done: {got}/{total} anchors accepted → {} (next: `train-room`)", + args.output + ); + Ok(()) +} + +// --------------------------------------------------------------------------- +// train-room +// --------------------------------------------------------------------------- + +/// Arguments for `train-room`. +#[derive(Args, Debug, Clone)] +pub struct TrainRoomArgs { + /// Enrollment file from `enroll`. + #[arg(long, default_value = "./enrollment.json")] + pub enrollment: String, + /// Output specialist-bank file. + #[arg(long, default_value = "./room-bank.json")] + pub output: String, +} + +/// Execute `train-room`. +pub async fn train_room(args: TrainRoomArgs) -> Result<()> { + let raw = std::fs::read_to_string(&args.enrollment) + .map_err(|e| anyhow::anyhow!("cannot read {}: {e} — run `enroll` first", args.enrollment))?; + let data: EnrollmentData = + serde_json::from_str(&raw).map_err(|e| anyhow::anyhow!("invalid enrollment: {e}"))?; + if data.anchors.is_empty() { + bail!("no accepted anchors in {} — re-run enroll", args.enrollment); + } + + let bank = SpecialistBank::train(&data.room_id, &data.baseline_id, &data.anchors, now_unix()) + .map_err(|e| anyhow::anyhow!("training failed: {e}"))?; + std::fs::write(&args.output, bank.to_json().map_err(|e| anyhow::anyhow!("{e}"))?) + .map_err(|e| anyhow::anyhow!("cannot write {}: {e}", args.output))?; + + eprintln!( + "[train-room] room='{}' trained {} specialists from {} anchors → {}", + bank.room_id, + bank.trained_kinds().len(), + bank.anchor_count, + args.output + ); + for k in bank.trained_kinds() { + eprintln!("[train-room] • {k:?}"); + } + Ok(()) +} + +// --------------------------------------------------------------------------- +// room-status +// --------------------------------------------------------------------------- + +/// Arguments for `room-status`. +#[derive(Args, Debug, Clone)] +pub struct RoomStatusArgs { + /// Specialist-bank file. + #[arg(long, default_value = "./room-bank.json")] + pub bank: String, +} + +/// Execute `room-status`. +pub async fn room_status(args: RoomStatusArgs) -> Result<()> { + 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}"))?; + println!("room: {}", bank.room_id); + println!("baseline: {}", bank.baseline_id); + println!("trained_at: {}", bank.trained_at_unix_s); + println!("anchors: {}", bank.anchor_count); + println!("specialists: {:?}", bank.trained_kinds()); + Ok(()) +} + +// --------------------------------------------------------------------------- +// room-watch +// --------------------------------------------------------------------------- + +/// Arguments for `room-watch`. +#[derive(Args, Debug, Clone)] +pub struct RoomWatchArgs { + /// Specialist-bank file. + #[arg(long, default_value = "./room-bank.json")] + pub bank: String, + /// UDP port for ESP32 CSI frames (raw CSI). + #[arg(long, default_value_t = 5005)] + pub udp_port: u16, + /// Bind address. + #[arg(long, default_value = "0.0.0.0")] + pub bind: String, + /// PHY tier. + #[arg(long, default_value = "ht20")] + pub tier: String, + /// CSI sample rate (Hz). + #[arg(long, default_value_t = 15.0)] + pub fs_hz: f32, + /// Rolling window length (frames) for each inference. + #[arg(long, default_value_t = 200)] + pub window: usize, + /// Seconds to run (0 = until Ctrl-C). + #[arg(long, default_value_t = 0)] + pub seconds: u32, +} + +/// Execute `room-watch` — live mixture-of-specialists readout. +pub async fn room_watch(args: RoomWatchArgs) -> Result<()> { + 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}"))?; + let baseline_id = bank.baseline_id.clone(); + let mix = MixtureOfSpecialists::new(bank); + + 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] inferring on udp://{addr} (window={} frames)", args.window); + + let mut buf = vec![0u8; RECV_BUF]; + let mut win: std::collections::VecDeque = std::collections::VecDeque::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 let Some(frame) = parse_csi_packet(&buf[..n], &args.tier) { + win.push_back(frame_scalar(&frame)); + while win.len() > args.window { + win.pop_front(); + } + } + } + if last_print.elapsed() >= Duration::from_secs(1) && win.len() >= 32 { + let series: Vec = win.iter().copied().collect(); + let f = Features::from_series(&series, args.fs_hz); + let s = mix.infer(&f, &baseline_id); + let pres = s.presence.as_ref().map(|r| r.label.clone().unwrap_or_default()).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 hr = s.heartbeat.as_ref().map(|r| format!("{:.0}bpm", r.value)).unwrap_or("-".into()); + let rest = s.restlessness.as_ref().map(|r| format!("{:.2}", r.value)).unwrap_or("-".into()); + let flags = format!( + "{}{}", + if s.vetoed { " VETO" } else { "" }, + if s.stale { " STALE" } else { "" } + ); + println!( + "presence={pres:<7} posture={post:<8} breathing={br:<8} heart={hr:<7} restless={rest}{flags}" + ); + last_print = Instant::now(); + } + } + Ok(()) +}