vrr/Ruview
2
0
Fork 0
mirror of https://github.com/ruvnet/RuView.git synced 2026-04-28 05:59:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Merge branch 'main' of https://github.com/ruvnet/RuView

Browse source
This commit is contained in:
oscarwangr-creator 2026-03-28 18:18:41 +01:00
commit dc391bea4d
8 changed files with 946 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

10
.github/workflows/firmware-ci.yml vendored
View file

@ -15,7 +15,7 @@ jobs:
name: Build ESP32-S3 Firmware
runs-on: ubuntu-latest
container:
image: espressif/idf:v5.2
image: espressif/idf:v5.4
steps:
- uses: actions/checkout@v4
@ -54,9 +54,10 @@ jobs:
fi
# Check partition table magic (0xAA50 at offset 0).
# Use od instead of xxd (xxd not available in espressif/idf container).
PT=build/partition_table/partition-table.bin
if [ -f "$PT" ]; then
MAGIC=$(xxd -l2 -p "$PT")
MAGIC=$(od -A n -t x1 -N 2 "$PT" | tr -d ' ')
if [ "$MAGIC" != "aa50" ]; then
echo "::warning::Partition table magic mismatch: $MAGIC (expected aa50)"
ERRORS=$((ERRORS + 1))
@ -71,7 +72,7 @@ jobs:
fi
# Verify non-zero data in binary (not all 0xFF padding).
NONZERO=$(xxd -l 1024 -p "$BIN" | tr -d 'f' | wc -c)
NONZERO=$(od -A n -t x1 -N 1024 "$BIN" | tr -d ' f\n' | wc -c)
if [ "$NONZERO" -lt 100 ]; then
echo "::error::Binary appears to be mostly padding (non-zero chars: $NONZERO)"
ERRORS=$((ERRORS + 1))
@ -97,4 +98,5 @@ jobs:
firmware/esp32-csi-node/build/esp32-csi-node.bin
firmware/esp32-csi-node/build/bootloader/bootloader.bin
firmware/esp32-csi-node/build/partition_table/partition-table.bin
retention-days: 30
firmware/esp32-csi-node/build/ota_data_initial.bin
retention-days: 90

11
README.md
View file

@ -1,11 +1,20 @@
# π RuView
<p align="center">
<a href="https://ruvnet.github.io/RuView/">
<a href="https://x.com/rUv/status/2037556932802761004">
<img src="assets/ruview-small-gemini.jpg" alt="RuView - WiFi DensePose" width="100%">
</a>
</p>
> **Alpha Software** — This project is under active development. APIs, firmware behavior, and documentation may change. Known limitations:
> - Multi-node person counting may show identical output regardless of the number of people (#249)
> - Training pipeline on MM-Fi dataset may plateau at low PCK (#318) — hyperparameter tuning in progress
> - No pre-trained model weights are provided; training from scratch is required
> - ESP32-C3 and original ESP32 are not supported (single-core, insufficient for CSI DSP)
> - Single ESP32 deployments have limited spatial resolution
>
> Contributions and bug reports welcome at [Issues](https://github.com/ruvnet/RuView/issues).
## **See through walls with WiFi + Ai** ##
**Perceive the world through signals.** No cameras. No wearables. No Internet. Just physics.

182
docs/adr/ADR-068-per-node-state-pipeline.md Normal file
View file

@ -0,0 +1,182 @@
# ADR-068: Per-Node State Pipeline for Multi-Node Sensing
| Field | Value |
|------------|-------------------------------------|
| Status | Accepted |
| Date | 2026-03-27 |
| Authors | rUv, claude-flow |
| Drivers | #249, #237, #276, #282 |
| Supersedes | — |
## Context
The sensing server (`wifi-densepose-sensing-server`) was originally designed for
single-node operation. When multiple ESP32 nodes send CSI frames simultaneously,
all data is mixed into a single shared pipeline:
- **One** `frame_history` VecDeque for all nodes
- **One** `smoothed_person_score` / `smoothed_motion` / vital sign buffers
- **One** baseline and debounce state
This means the classification, person count, and vital signs reported to the UI
are an uncontrolled aggregate of all nodes' data. The result: the detection
window shows identical output regardless of how many nodes are deployed, where
people stand, or how many people are in the room (#249 — 24 comments, the most
reported issue).
### Root Cause Verified
Investigation of `AppStateInner` (main.rs lines 279-367) confirmed:
| Shared field | Impact |
|---------------------------|--------------------------------------------|
| `frame_history` | Temporal analysis mixes all nodes' CSI data |
| `smoothed_person_score` | Person count aggregates all nodes |
| `smoothed_motion` | Motion classification undifferentiated |
| `smoothed_hr` / `br` | Vital signs are global, not per-node |
| `baseline_motion` | Adaptive baseline learned from mixed data |
| `debounce_counter` | All nodes share debounce state |
## Decision
Introduce **per-node state tracking** via a `HashMap<u8, NodeState>` in
`AppStateInner`. Each ESP32 node (identified by its `node_id` byte) gets an
independent sensing pipeline with its own temporal history, smoothing buffers,
baseline, and classification state.
### Architecture
```
┌─────────────────────────────────────────┐
UDP frames │ AppStateInner │
───────────► │ │
node_id=1 ──► │ node_states: HashMap<u8, NodeState> │
node_id=2 ──► │ ├── 1: NodeState { frame_history, │
node_id=3 ──► │ │ smoothed_motion, vitals, ... }│
│ ├── 2: NodeState { ... } │
│ └── 3: NodeState { ... } │
│ │
│ ┌── Per-Node Pipeline ──┐ │
│ │ extract_features() │ │
│ │ smooth_and_classify() │ │
│ │ smooth_vitals() │ │
│ │ score_to_person_count()│ │
│ └────────────────────────┘ │
│ │
│ ┌── Multi-Node Fusion ──┐ │
│ │ Aggregate person count │ │
│ │ Per-node classification│ │
│ │ All-nodes WebSocket msg│ │
│ └────────────────────────┘ │
│ │
│ ──► WebSocket broadcast (sensing_update) │
└─────────────────────────────────────────┘
```
### NodeState Struct
```rust
struct NodeState {
frame_history: VecDeque<Vec<f64>>,
smoothed_person_score: f64,
prev_person_count: usize,
smoothed_motion: f64,
current_motion_level: String,
debounce_counter: u32,
debounce_candidate: String,
baseline_motion: f64,
baseline_frames: u64,
smoothed_hr: f64,
smoothed_br: f64,
smoothed_hr_conf: f64,
smoothed_br_conf: f64,
hr_buffer: VecDeque<f64>,
br_buffer: VecDeque<f64>,
rssi_history: VecDeque<f64>,
vital_detector: VitalSignDetector,
latest_vitals: VitalSigns,
last_frame_time: Option<std::time::Instant>,
edge_vitals: Option<Esp32VitalsPacket>,
}
```
### Multi-Node Aggregation
- **Person count**: Sum of per-node `prev_person_count` for active nodes
(seen within last 10 seconds).
- **Classification**: Per-node classification included in `SensingUpdate.nodes`.
- **Vital signs**: Per-node vital signs; UI can render per-node or aggregate.
- **Signal field**: Generated from the most-recently-updated node's features.
- **Stale nodes**: Nodes with no frame for >10 seconds are excluded from
aggregation and marked offline (consistent with PR #300).
### Backward Compatibility
- The simulated data path (`simulated_data_task`) continues using global state.
- Single-node deployments behave identically (HashMap has one entry).
- The WebSocket message format (`sensing_update`) remains the same but the
`nodes` array now contains all active nodes, and `estimated_persons` reflects
the cross-node aggregate.
- The edge vitals path (#323 fix) also uses per-node state.
## Scaling Characteristics
| Nodes | Per-Node Memory | Total Overhead | Notes |
|-------|----------------|----------------|-------|
| 1 | ~50 KB | ~50 KB | Identical to current |
| 3 | ~50 KB | ~150 KB | Typical home setup |
| 10 | ~50 KB | ~500 KB | Small office |
| 50 | ~50 KB | ~2.5 MB | Building floor |
| 100 | ~50 KB | ~5 MB | Large deployment |
| 256 | ~50 KB | ~12.8 MB | Max (u8 node_id) |
Memory is dominated by `frame_history` (100 frames x ~500 bytes each = ~50 KB
per node). This scales linearly and fits comfortably in server memory even at
256 nodes.
## QEMU Validation
The existing QEMU swarm infrastructure (ADR-062, `scripts/qemu_swarm.py`)
supports multi-node simulation with configurable topologies:
- `star`: Central coordinator + sensor nodes
- `mesh`: Fully connected peer network
- `line`: Sequential chain
- `ring`: Circular topology
Each QEMU instance runs with a unique `node_id` via NVS provisioning. The
swarm health validator (`scripts/swarm_health.py`) checks per-node UART output.
Validation plan:
1. QEMU swarm with 3-5 nodes in mesh topology
2. Verify server produces distinct per-node classifications
3. Verify aggregate person count reflects multi-node contributions
4. Verify stale-node eviction after timeout
## Consequences
### Positive
- Each node's CSI data is processed independently — no cross-contamination
- Person count scales with the number of deployed nodes
- Vital signs are per-node, enabling room-level health monitoring
- Foundation for spatial localization (per-node positions + triangulation)
- Scales to 256 nodes with <13 MB memory overhead
### Negative
- Slightly more memory per node (~50 KB each)
- `smooth_and_classify_node` function duplicates some logic from global version
- Per-node `VitalSignDetector` instances add CPU cost proportional to node count
### Risks
- Node ID collisions (mitigated by NVS persistence since v0.5.0)
- HashMap growth without cleanup (mitigated by stale-node eviction)
## References
- Issue #249: Detection window same regardless (24 comments)
- Issue #237: Same display for 0/1/2 people (12 comments)
- Issue #276: Only one can be detected (8 comments)
- Issue #282: Detection fail (5 comments)
- PR #295: Hysteresis smoothing (partial mitigation)
- PR #300: ESP32 offline detection after 5s
- ADR-062: QEMU Swarm Configurator

37
firmware/esp32-csi-node/main/edge_processing.c
View file

@ -41,12 +41,14 @@ static const char *TAG = "edge_proc";
* ====================================================================== */
static edge_ring_buf_t s_ring;
static uint32_t s_ring_drops; /* Frames dropped due to full ring buffer. */
static inline bool ring_push(const uint8_t *iq, uint16_t len,
int8_t rssi, uint8_t channel)
{
uint32_t next = (s_ring.head + 1) % EDGE_RING_SLOTS;
if (next == s_ring.tail) {
s_ring_drops++;
return false; /* Full — drop frame. */
}
@ -788,12 +790,13 @@ static void process_frame(const edge_ring_slot_t *slot)
if ((s_frame_count % 200) == 0) {
ESP_LOGI(TAG, "Vitals: br=%.1f hr=%.1f motion=%.4f pres=%s "
"fall=%s persons=%u frames=%lu",
"fall=%s persons=%u frames=%lu drops=%lu",
s_breathing_bpm, s_heartrate_bpm, s_motion_energy,
s_presence_detected ? "YES" : "no",
s_fall_detected ? "YES" : "no",
(unsigned)s_latest_pkt.n_persons,
(unsigned long)s_frame_count);
(unsigned long)s_frame_count,
(unsigned long)s_ring_drops);
}
}
@ -831,18 +834,32 @@ static void edge_task(void *arg)
edge_ring_slot_t slot;
/* Maximum frames to process before a longer yield. On busy LANs
* (corporate networks, many APs), the ring buffer fills continuously.
* Without a batch limit the task processes frames back-to-back with
* only 1-tick yields, which on high frame rates can still starve
* IDLE1 enough to trip the 5-second task watchdog. See #266, #321. */
const uint8_t BATCH_LIMIT = 4;
while (1) {
if (ring_pop(&slot)) {
uint8_t processed = 0;
while (processed < BATCH_LIMIT && ring_pop(&slot)) {
process_frame(&slot);
/* Yield after every frame to feed the Core 1 watchdog.
* process_frame() is CPU-intensive (biquad filters, Welford stats,
* BPM estimation, multi-person vitals) and can take several ms.
* Without this yield, edge_dsp at priority 5 starves IDLE1 at
* priority 0, triggering the task watchdog. See issue #266. */
processed++;
/* 1-tick yield between frames within a batch. */
vTaskDelay(1);
}
if (processed > 0) {
/* Post-batch yield: 2 ticks (~20 ms at 100 Hz) so IDLE1 can
* run and feed the Core 1 watchdog even under sustained load.
* This is intentionally longer than the 1-tick inter-frame yield. */
vTaskDelay(2);
} else {
/* No frames available — yield briefly. */
vTaskDelay(pdMS_TO_TICKS(1));
/* No frames available — sleep one full tick.
* NOTE: pdMS_TO_TICKS(5) == 0 at 100 Hz, which would busy-spin. */
vTaskDelay(1);
}
}
}

2
pyproject.toml
View file

@ -185,7 +185,7 @@ package-dir = {"" = "."}
[tool.setuptools.packages.find]
where = ["."]
include = ["src*"]
include = ["wifi_densepose*", "src*"]
exclude = ["tests*", "docs*", "scripts*"]
[tool.setuptools.package-data]

384
rust-port/wifi-densepose-rs/crates/wifi-densepose-sensing-server/src/main.rs
View file

@ -16,7 +16,7 @@ mod vital_signs;
// Training pipeline modules (exposed via lib.rs)
use wifi_densepose_sensing_server::{graph_transformer, trainer, dataset, embedding};
use std::collections::VecDeque;
use std::collections::{HashMap, VecDeque};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
@ -275,6 +275,59 @@ struct BoundingBox {
height: f64,
}
/// Per-node sensing state for multi-node deployments (issue #249).
/// Each ESP32 node gets its own frame history, smoothing buffers, and vital
/// sign detector so that data from different nodes is never mixed.
struct NodeState {
frame_history: VecDeque<Vec<f64>>,
smoothed_person_score: f64,
prev_person_count: usize,
smoothed_motion: f64,
current_motion_level: String,
debounce_counter: u32,
debounce_candidate: String,
baseline_motion: f64,
baseline_frames: u64,
smoothed_hr: f64,
smoothed_br: f64,
smoothed_hr_conf: f64,
smoothed_br_conf: f64,
hr_buffer: VecDeque<f64>,
br_buffer: VecDeque<f64>,
rssi_history: VecDeque<f64>,
vital_detector: VitalSignDetector,
latest_vitals: VitalSigns,
last_frame_time: Option<std::time::Instant>,
edge_vitals: Option<Esp32VitalsPacket>,
}
impl NodeState {
fn new() -> Self {
Self {
frame_history: VecDeque::new(),
smoothed_person_score: 0.0,
prev_person_count: 0,
smoothed_motion: 0.0,
current_motion_level: "absent".to_string(),
debounce_counter: 0,
debounce_candidate: "absent".to_string(),
baseline_motion: 0.0,
baseline_frames: 0,
smoothed_hr: 0.0,
smoothed_br: 0.0,
smoothed_hr_conf: 0.0,
smoothed_br_conf: 0.0,
hr_buffer: VecDeque::with_capacity(8),
br_buffer: VecDeque::with_capacity(8),
rssi_history: VecDeque::new(),
vital_detector: VitalSignDetector::new(10.0),
latest_vitals: VitalSigns::default(),
last_frame_time: None,
edge_vitals: None,
}
}
}
/// Shared application state
struct AppStateInner {
latest_update: Option<SensingUpdate>,
@ -364,6 +417,10 @@ struct AppStateInner {
// ── Adaptive classifier (environment-tuned) ──────────────────────────
/// Trained adaptive model (loaded from data/adaptive_model.json or trained at runtime).
adaptive_model: Option<adaptive_classifier::AdaptiveModel>,
// ── Per-node state (issue #249) ─────────────────────────────────────
/// Per-node sensing state for multi-node deployments.
/// Keyed by `node_id` from the ESP32 frame header.
node_states: HashMap<u8, NodeState>,
}
/// If no ESP32 frame arrives within this duration, source reverts to offline.
@ -964,6 +1021,44 @@ fn smooth_and_classify(state: &mut AppStateInner, raw: &mut ClassificationInfo,
raw.confidence = (0.4 + sm * 0.6).clamp(0.0, 1.0);
}
/// Per-node variant of `smooth_and_classify` that operates on a `NodeState`
/// instead of `AppStateInner` (issue #249).
fn smooth_and_classify_node(ns: &mut NodeState, raw: &mut ClassificationInfo, raw_motion: f64) {
ns.baseline_frames += 1;
if ns.baseline_frames < BASELINE_WARMUP {
ns.baseline_motion = ns.baseline_motion * 0.9 + raw_motion * 0.1;
} else if raw_motion < ns.smoothed_motion + 0.05 {
ns.baseline_motion = ns.baseline_motion * (1.0 - BASELINE_EMA_ALPHA)
+ raw_motion * BASELINE_EMA_ALPHA;
}
let adjusted = (raw_motion - ns.baseline_motion * 0.7).max(0.0);
ns.smoothed_motion = ns.smoothed_motion * (1.0 - MOTION_EMA_ALPHA)
+ adjusted * MOTION_EMA_ALPHA;
let sm = ns.smoothed_motion;
let candidate = raw_classify(sm);
if candidate == ns.current_motion_level {
ns.debounce_counter = 0;
ns.debounce_candidate = candidate;
} else if candidate == ns.debounce_candidate {
ns.debounce_counter += 1;
if ns.debounce_counter >= DEBOUNCE_FRAMES {
ns.current_motion_level = candidate;
ns.debounce_counter = 0;
}
} else {
ns.debounce_candidate = candidate;
ns.debounce_counter = 1;
}
raw.motion_level = ns.current_motion_level.clone();
raw.presence = sm > 0.03;
raw.confidence = (0.4 + sm * 0.6).clamp(0.0, 1.0);
}
/// If an adaptive model is loaded, override the classification with the
/// model's prediction. Uses the full 15-feature vector for higher accuracy.
fn adaptive_override(state: &AppStateInner, features: &FeatureInfo, classification: &mut ClassificationInfo) {
@ -1064,6 +1159,55 @@ fn smooth_vitals(state: &mut AppStateInner, raw: &VitalSigns) -> VitalSigns {
}
}
/// Per-node variant of `smooth_vitals` that operates on a `NodeState` (issue #249).
fn smooth_vitals_node(ns: &mut NodeState, raw: &VitalSigns) -> VitalSigns {
let raw_hr = raw.heart_rate_bpm.unwrap_or(0.0);
let raw_br = raw.breathing_rate_bpm.unwrap_or(0.0);
let hr_ok = ns.smoothed_hr < 1.0 || (raw_hr - ns.smoothed_hr).abs() < HR_MAX_JUMP;
let br_ok = ns.smoothed_br < 1.0 || (raw_br - ns.smoothed_br).abs() < BR_MAX_JUMP;
if hr_ok && raw_hr > 0.0 {
ns.hr_buffer.push_back(raw_hr);
if ns.hr_buffer.len() > VITAL_MEDIAN_WINDOW { ns.hr_buffer.pop_front(); }
}
if br_ok && raw_br > 0.0 {
ns.br_buffer.push_back(raw_br);
if ns.br_buffer.len() > VITAL_MEDIAN_WINDOW { ns.br_buffer.pop_front(); }
}
let trimmed_hr = trimmed_mean(&ns.hr_buffer);
let trimmed_br = trimmed_mean(&ns.br_buffer);
if trimmed_hr > 0.0 {
if ns.smoothed_hr < 1.0 {
ns.smoothed_hr = trimmed_hr;
} else if (trimmed_hr - ns.smoothed_hr).abs() > HR_DEAD_BAND {
ns.smoothed_hr = ns.smoothed_hr * (1.0 - VITAL_EMA_ALPHA)
+ trimmed_hr * VITAL_EMA_ALPHA;
}
}
if trimmed_br > 0.0 {
if ns.smoothed_br < 1.0 {
ns.smoothed_br = trimmed_br;
} else if (trimmed_br - ns.smoothed_br).abs() > BR_DEAD_BAND {
ns.smoothed_br = ns.smoothed_br * (1.0 - VITAL_EMA_ALPHA)
+ trimmed_br * VITAL_EMA_ALPHA;
}
}
ns.smoothed_hr_conf = ns.smoothed_hr_conf * 0.92 + raw.heartbeat_confidence * 0.08;
ns.smoothed_br_conf = ns.smoothed_br_conf * 0.92 + raw.breathing_confidence * 0.08;
VitalSigns {
breathing_rate_bpm: if ns.smoothed_br > 1.0 { Some(ns.smoothed_br) } else { None },
heart_rate_bpm: if ns.smoothed_hr > 1.0 { Some(ns.smoothed_hr) } else { None },
breathing_confidence: ns.smoothed_br_conf,
heartbeat_confidence: ns.smoothed_hr_conf,
signal_quality: raw.signal_quality,
}
}
/// Trimmed mean: sort, drop top/bottom 25%, average the middle 50%.
/// More robust than median (uses more data) and less noisy than raw mean.
fn trimmed_mean(buf: &VecDeque<f64>) -> f64 {
@ -2820,6 +2964,115 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
})) {
let _ = s.tx.send(json);
}
// Issue #323: Also emit a sensing_update so the UI renders
// detections for ESP32 nodes running the edge DSP pipeline
// (Tier 2+). Without this, vitals arrive but the UI shows
// "no detection" because it only renders sensing_update msgs.
s.source = "esp32".to_string();
s.last_esp32_frame = Some(std::time::Instant::now());
// ── Per-node state for edge vitals (issue #249) ──────
let node_id = vitals.node_id;
let ns = s.node_states.entry(node_id).or_insert_with(NodeState::new);
ns.last_frame_time = Some(std::time::Instant::now());
ns.edge_vitals = Some(vitals.clone());
ns.rssi_history.push_back(vitals.rssi as f64);
if ns.rssi_history.len() > 60 { ns.rssi_history.pop_front(); }
// Store per-node person count from edge vitals.
let node_est = if vitals.presence {
(vitals.n_persons as usize).max(1)
} else {
0
};
ns.prev_person_count = node_est;
s.tick += 1;
let tick = s.tick;
let motion_level = if vitals.motion { "present_moving" }
else if vitals.presence { "present_still" }
else { "absent" };
let motion_score = if vitals.motion { 0.8 }
else if vitals.presence { 0.3 }
else { 0.05 };
// Aggregate person count across all active nodes.
let now = std::time::Instant::now();
let total_persons: usize = s.node_states.values()
.filter(|n| n.last_frame_time.map_or(false, |t| now.duration_since(t).as_secs() < 10))
.map(|n| n.prev_person_count)
.sum();
// Build nodes array with all active nodes.
let active_nodes: Vec<NodeInfo> = s.node_states.iter()
.filter(|(_, n)| n.last_frame_time.map_or(false, |t| now.duration_since(t).as_secs() < 10))
.map(|(&id, n)| NodeInfo {
node_id: id,
rssi_dbm: n.rssi_history.back().copied().unwrap_or(0.0),
position: [2.0, 0.0, 1.5],
amplitude: vec![],
subcarrier_count: 0,
})
.collect();
let features = FeatureInfo {
mean_rssi: vitals.rssi as f64,
variance: vitals.motion_energy as f64,
motion_band_power: vitals.motion_energy as f64,
breathing_band_power: if vitals.presence { 0.5 } else { 0.0 },
dominant_freq_hz: vitals.breathing_rate_bpm / 60.0,
change_points: 0,
spectral_power: vitals.motion_energy as f64,
};
let classification = ClassificationInfo {
motion_level: motion_level.to_string(),
presence: vitals.presence,
confidence: vitals.presence_score as f64,
};
let signal_field = generate_signal_field(
vitals.rssi as f64, motion_score, vitals.breathing_rate_bpm / 60.0,
(vitals.presence_score as f64).min(1.0), &[],
);
let mut update = SensingUpdate {
msg_type: "sensing_update".to_string(),
timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
source: "esp32".to_string(),
tick,
nodes: active_nodes,
features: features.clone(),
classification,
signal_field,
vital_signs: Some(VitalSigns {
breathing_rate_bpm: if vitals.breathing_rate_bpm > 0.0 { Some(vitals.breathing_rate_bpm) } else { None },
heart_rate_bpm: if vitals.heartrate_bpm > 0.0 { Some(vitals.heartrate_bpm) } else { None },
breathing_confidence: if vitals.presence { 0.7 } else { 0.0 },
heartbeat_confidence: if vitals.presence { 0.7 } else { 0.0 },
signal_quality: vitals.presence_score as f64,
}),
enhanced_motion: None,
enhanced_breathing: None,
posture: None,
signal_quality_score: None,
quality_verdict: None,
bssid_count: None,
pose_keypoints: None,
model_status: None,
persons: None,
estimated_persons: if total_persons > 0 { Some(total_persons) } else { None },
};
let persons = derive_pose_from_sensing(&update);
if !persons.is_empty() {
update.persons = Some(persons);
}
if let Ok(json) = serde_json::to_string(&update) {
let _ = s.tx.send(json);
}
s.latest_update = Some(update);
s.edge_vitals = Some(vitals);
continue;
}
@ -2851,24 +3104,90 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
s.source = "esp32".to_string();
s.last_esp32_frame = Some(std::time::Instant::now());
// Append current amplitudes to history before extracting features so
// that temporal analysis includes the most recent frame.
// Also maintain global frame_history for backward compat
// (simulation path, REST endpoints, etc.).
s.frame_history.push_back(frame.amplitudes.clone());
if s.frame_history.len() > FRAME_HISTORY_CAPACITY {
s.frame_history.pop_front();
}
let sample_rate_hz = 1000.0 / 500.0_f64; // default tick; ESP32 frames arrive as fast as they come
let (features, mut classification, breathing_rate_hz, sub_variances, raw_motion) =
extract_features_from_frame(&frame, &s.frame_history, sample_rate_hz);
smooth_and_classify(&mut s, &mut classification, raw_motion);
adaptive_override(&s, &features, &mut classification);
// ── Per-node processing (issue #249) ──────────────────
// Process entirely within per-node state so different
// ESP32 nodes never mix their smoothing/vitals buffers.
// We scope the mutable borrow of node_states so we can
// access other AppStateInner fields afterward.
let node_id = frame.node_id;
let adaptive_model_ref = s.adaptive_model.as_ref().map(|m| m as *const _);
let ns = s.node_states.entry(node_id).or_insert_with(NodeState::new);
ns.last_frame_time = Some(std::time::Instant::now());
ns.frame_history.push_back(frame.amplitudes.clone());
if ns.frame_history.len() > FRAME_HISTORY_CAPACITY {
ns.frame_history.pop_front();
}
let sample_rate_hz = 1000.0 / 500.0_f64;
let (features, mut classification, breathing_rate_hz, sub_variances, raw_motion) =
extract_features_from_frame(&frame, &ns.frame_history, sample_rate_hz);
smooth_and_classify_node(ns, &mut classification, raw_motion);
// SAFETY: adaptive_model_ref points into s which we hold
// via write lock; the model is not mutated here. We use a
// raw pointer to break the borrow-checker deadlock between
// node_states and adaptive_model (both inside s).
if let Some(model_ptr) = adaptive_model_ref {
let model: &adaptive_classifier::AdaptiveModel = unsafe { &*model_ptr };
let amps = ns.frame_history.back()
.map(|v| v.as_slice())
.unwrap_or(&[]);
let feat_arr = adaptive_classifier::features_from_runtime(
&serde_json::json!({
"variance": features.variance,
"motion_band_power": features.motion_band_power,
"breathing_band_power": features.breathing_band_power,
"spectral_power": features.spectral_power,
"dominant_freq_hz": features.dominant_freq_hz,
"change_points": features.change_points,
"mean_rssi": features.mean_rssi,
}),
amps,
);
let (label, conf) = model.classify(&feat_arr);
classification.motion_level = label.to_string();
classification.presence = label != "absent";
classification.confidence = (conf * 0.7 + classification.confidence * 0.3).clamp(0.0, 1.0);
}
ns.rssi_history.push_back(features.mean_rssi);
if ns.rssi_history.len() > 60 {
ns.rssi_history.pop_front();
}
let raw_vitals = ns.vital_detector.process_frame(
&frame.amplitudes,
&frame.phases,
);
let vitals = smooth_vitals_node(ns, &raw_vitals);
ns.latest_vitals = vitals.clone();
let raw_score = compute_person_score(&features);
ns.smoothed_person_score = ns.smoothed_person_score * 0.90 + raw_score * 0.10;
if classification.presence {
let count = score_to_person_count(ns.smoothed_person_score, ns.prev_person_count);
ns.prev_person_count = count;
} else {
ns.prev_person_count = 0;
}
// Done with per-node mutable borrow; now read aggregated
// state from all nodes (the borrow of `ns` ends here).
// (We re-borrow node_states immutably via `s` below.)
// Update RSSI history
s.rssi_history.push_back(features.mean_rssi);
if s.rssi_history.len() > 60 {
s.rssi_history.pop_front();
}
s.latest_vitals = vitals.clone();
s.tick += 1;
let tick = s.tick;
@ -2877,37 +3196,33 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
else if classification.motion_level == "present_still" { 0.3 }
else { 0.05 };
let raw_vitals = s.vital_detector.process_frame(
&frame.amplitudes,
&frame.phases,
);
let vitals = smooth_vitals(&mut s, &raw_vitals);
s.latest_vitals = vitals.clone();
// Aggregate person count across all active nodes.
let now = std::time::Instant::now();
let total_persons: usize = s.node_states.values()
.filter(|n| n.last_frame_time.map_or(false, |t| now.duration_since(t).as_secs() < 10))
.map(|n| n.prev_person_count)
.sum();
// Multi-person estimation with temporal smoothing (EMA α=0.10).
let raw_score = compute_person_score(&features);
s.smoothed_person_score = s.smoothed_person_score * 0.90 + raw_score * 0.10;
let est_persons = if classification.presence {
let count = score_to_person_count(s.smoothed_person_score, s.prev_person_count);
s.prev_person_count = count;
count
} else {
s.prev_person_count = 0;
0
};
// Build nodes array with all active nodes.
let active_nodes: Vec<NodeInfo> = s.node_states.iter()
.filter(|(_, n)| n.last_frame_time.map_or(false, |t| now.duration_since(t).as_secs() < 10))
.map(|(&id, n)| NodeInfo {
node_id: id,
rssi_dbm: n.rssi_history.back().copied().unwrap_or(0.0),
position: [2.0, 0.0, 1.5],
amplitude: n.frame_history.back()
.map(|a| a.iter().take(56).cloned().collect())
.unwrap_or_default(),
subcarrier_count: n.frame_history.back().map_or(0, |a| a.len()),
})
.collect();
let mut update = SensingUpdate {
msg_type: "sensing_update".to_string(),
timestamp: chrono::Utc::now().timestamp_millis() as f64 / 1000.0,
source: "esp32".to_string(),
tick,
nodes: vec![NodeInfo {
node_id: frame.node_id,
rssi_dbm: features.mean_rssi,
position: [2.0, 0.0, 1.5],
amplitude: frame.amplitudes.iter().take(56).cloned().collect(),
subcarrier_count: frame.n_subcarriers as usize,
}],
nodes: active_nodes,
features: features.clone(),
classification,
signal_field: generate_signal_field(
@ -2924,7 +3239,7 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
pose_keypoints: None,
model_status: None,
persons: None,
estimated_persons: if est_persons > 0 { Some(est_persons) } else { None },
estimated_persons: if total_persons > 0 { Some(total_persons) } else { None },
};
let persons = derive_pose_from_sensing(&update);
@ -3676,6 +3991,7 @@ async fn main() {
m.trained_frames, m.training_accuracy * 100.0);
m
}),
node_states: HashMap::new(),
}));
// Start background tasks based on source

233
rust-port/wifi-densepose-rs/crates/wifi-densepose-sensing-server/tests/multi_node_test.rs Normal file
View file

@ -0,0 +1,233 @@
//! Integration test: multi-node per-node state isolation (ADR-068, #249).
//!
//! Sends simulated ESP32 CSI frames from multiple node IDs to the server's
//! UDP port and verifies that:
//! 1. Each node gets independent state (no cross-contamination)
//! 2. Person count aggregates across active nodes
//! 3. Stale nodes are excluded from aggregation
//!
//! This does NOT require QEMU — it sends raw UDP packets directly.
use std::net::UdpSocket;
use std::time::Duration;
/// Build a minimal valid ESP32 CSI frame (magic 0xC511_0001).
///
/// Format (ADR-018):
/// [0..3] magic: 0xC511_0001 (LE)
/// [4] node_id
/// [5] n_antennas (1)
/// [6] n_subcarriers (e.g., 32)
/// [7] reserved
/// [8..9] freq_mhz (2437 = channel 6)
/// [10..13] sequence (LE u32)
/// [14] rssi (signed)
/// [15] noise_floor
/// [16..19] reserved
/// [20..] I/Q pairs (n_antennas * n_subcarriers * 2 bytes)
fn build_csi_frame(node_id: u8, seq: u32, rssi: i8, n_sub: u8) -> Vec<u8> {
let n_pairs = n_sub as usize;
let mut buf = vec![0u8; 20 + n_pairs * 2];
// Magic
let magic: u32 = 0xC511_0001;
buf[0..4].copy_from_slice(&magic.to_le_bytes());
buf[4] = node_id;
buf[5] = 1; // n_antennas
buf[6] = n_sub;
buf[7] = 0;
// freq = 2437 MHz (channel 6)
let freq: u16 = 2437;
buf[8..10].copy_from_slice(&freq.to_le_bytes());
// sequence
buf[10..14].copy_from_slice(&seq.to_le_bytes());
buf[14] = rssi as u8;
buf[15] = (-90i8) as u8; // noise floor
// Generate I/Q pairs with node-specific patterns.
// Different nodes produce different amplitude patterns so the server
// computes different features for each.
for i in 0..n_pairs {
let phase = (i as f64 + node_id as f64 * 0.5) * 0.3;
let amplitude = 20.0 + (node_id as f64) * 5.0 + (phase.sin() * 10.0);
let i_val = (amplitude * phase.cos()) as i8;
let q_val = (amplitude * phase.sin()) as i8;
buf[20 + i * 2] = i_val as u8;
buf[20 + i * 2 + 1] = q_val as u8;
}
buf
}
/// Build an edge vitals packet (magic 0xC511_0002).
fn build_vitals_packet(node_id: u8, presence: bool, n_persons: u8, rssi: i8) -> Vec<u8> {
let mut buf = vec![0u8; 32];
let magic: u32 = 0xC511_0002;
buf[0..4].copy_from_slice(&magic.to_le_bytes());
buf[4] = node_id;
buf[5] = if presence { 0x01 } else { 0x00 }; // flags
// breathing_rate (u16 LE) = 15.0 * 100 = 1500
buf[6..8].copy_from_slice(&1500u16.to_le_bytes());
// heartrate (u32 LE) = 72.0 * 10000 = 720000
buf[8..12].copy_from_slice(&720000u32.to_le_bytes());
buf[12] = rssi as u8;
buf[13] = n_persons;
// bytes 14-15: reserved
// motion_energy (f32 LE)
let me: f32 = if presence { 0.5 } else { 0.0 };
buf[16..20].copy_from_slice(&me.to_le_bytes());
// presence_score (f32 LE)
let ps: f32 = if presence { 0.8 } else { 0.0 };
buf[20..24].copy_from_slice(&ps.to_le_bytes());
// timestamp_ms (u32 LE)
buf[24..28].copy_from_slice(&1000u32.to_le_bytes());
buf
}
#[test]
fn test_csi_frame_builder_valid() {
let frame = build_csi_frame(1, 0, -50, 32);
assert_eq!(frame.len(), 20 + 32 * 2);
assert_eq!(u32::from_le_bytes([frame[0], frame[1], frame[2], frame[3]]), 0xC511_0001);
assert_eq!(frame[4], 1); // node_id
assert_eq!(frame[5], 1); // n_antennas
assert_eq!(frame[6], 32); // n_subcarriers
}
#[test]
fn test_vitals_packet_builder_valid() {
let pkt = build_vitals_packet(2, true, 1, -45);
assert_eq!(pkt.len(), 32);
assert_eq!(u32::from_le_bytes([pkt[0], pkt[1], pkt[2], pkt[3]]), 0xC511_0002);
assert_eq!(pkt[4], 2); // node_id
assert_eq!(pkt[5], 0x01); // flags: presence
assert_eq!(pkt[13], 1); // n_persons
}
#[test]
fn test_different_nodes_produce_different_frames() {
let frame1 = build_csi_frame(1, 0, -50, 32);
let frame2 = build_csi_frame(2, 0, -50, 32);
// I/Q data should differ due to node_id-based amplitude offset
assert_ne!(&frame1[20..], &frame2[20..]);
}
/// Send multiple frames from different nodes to a UDP port.
/// This test verifies the packet format is accepted by a real server
/// if one is running, but doesn't fail if no server is available.
#[test]
fn test_multi_node_udp_send() {
// Try to bind to a random port and send to localhost:5005
// This is a smoke test — it verifies frames can be sent without panic.
let sock = UdpSocket::bind("0.0.0.0:0").expect("bind");
sock.set_write_timeout(Some(Duration::from_millis(100))).ok();
let n_sub = 32u8;
let node_ids = [1u8, 2, 3, 5, 7];
for &nid in &node_ids {
for seq in 0..10u32 {
let frame = build_csi_frame(nid, seq, -50 + nid as i8, n_sub);
// Send to localhost:5005 (won't fail even if nothing is listening)
let _ = sock.send_to(&frame, "127.0.0.1:5005");
}
}
// Also send vitals packets
for &nid in &node_ids {
let pkt = build_vitals_packet(nid, true, 1, -45);
let _ = sock.send_to(&pkt, "127.0.0.1:5005");
}
// If we get here without panic, the frame builders work correctly
assert!(true, "Multi-node UDP send completed without errors");
}
/// Verify that the frame builder produces frames of the correct minimum
/// size for various subcarrier counts (boundary testing).
#[test]
fn test_frame_sizes() {
for n_sub in [1u8, 16, 32, 52, 56, 64, 128] {
let frame = build_csi_frame(1, 0, -50, n_sub);
let expected = 20 + (n_sub as usize) * 2;
assert_eq!(frame.len(), expected, "wrong size for n_sub={n_sub}");
}
}
/// Simulate a mesh of N nodes sending frames at different rates.
/// Nodes 1-3 send every "tick", node 4 sends every other tick,
/// node 5 stops after 5 ticks (simulating going offline).
#[test]
fn test_mesh_simulation_pattern() {
let sock = UdpSocket::bind("0.0.0.0:0").expect("bind");
sock.set_write_timeout(Some(Duration::from_millis(50))).ok();
let mut total_sent = 0u32;
for tick in 0..20u32 {
// Nodes 1-3: every tick
for nid in 1..=3u8 {
let frame = build_csi_frame(nid, tick, -50, 32);
let _ = sock.send_to(&frame, "127.0.0.1:5005");
total_sent += 1;
}
// Node 4: every other tick
if tick % 2 == 0 {
let frame = build_csi_frame(4, tick / 2, -55, 32);
let _ = sock.send_to(&frame, "127.0.0.1:5005");
total_sent += 1;
}
// Node 5: stops after tick 5
if tick < 5 {
let frame = build_csi_frame(5, tick, -60, 32);
let _ = sock.send_to(&frame, "127.0.0.1:5005");
total_sent += 1;
}
}
// Expected: 3*20 + 10 + 5 = 75 frames
assert_eq!(total_sent, 75, "unexpected frame count");
}
/// Large mesh: simulate 100 nodes each sending 10 frames.
/// Verifies the frame builder scales without issues.
#[test]
fn test_large_mesh_100_nodes() {
let sock = UdpSocket::bind("0.0.0.0:0").expect("bind");
sock.set_write_timeout(Some(Duration::from_millis(50))).ok();
let mut total = 0u32;
for nid in 1..=100u8 {
for seq in 0..10u32 {
let frame = build_csi_frame(nid, seq, -50 + (nid % 30) as i8, 32);
let _ = sock.send_to(&frame, "127.0.0.1:5005");
total += 1;
}
}
assert_eq!(total, 1000);
}
/// Max mesh: simulate 255 nodes (max u8 node_id) with 1 frame each.
#[test]
fn test_max_nodes_255() {
let sock = UdpSocket::bind("0.0.0.0:0").expect("bind");
sock.set_write_timeout(Some(Duration::from_millis(100))).ok();
for nid in 1..=255u8 {
let frame = build_csi_frame(nid, 0, -50, 16);
let _ = sock.send_to(&frame, "127.0.0.1:5005");
}
// 255 unique node_ids — the HashMap should handle this fine
assert!(true);
}

137
wifi_densepose/__init__.py Normal file
View file

@ -0,0 +1,137 @@
"""
WiFi-DensePose — WiFi-based human pose estimation using CSI data.
Usage:
from wifi_densepose import WiFiDensePose
system = WiFiDensePose()
system.start()
poses = system.get_latest_poses()
system.stop()
"""
__version__ = "1.2.0"
import sys
import os
import logging
logger = logging.getLogger(__name__)
# Allow importing the v1 src package when installed from the repo
_v1_src = os.path.join(os.path.dirname(os.path.dirname(__file__)), "v1")
if os.path.isdir(_v1_src) and _v1_src not in sys.path:
sys.path.insert(0, _v1_src)
class WiFiDensePose:
"""High-level facade for the WiFi-DensePose sensing system.
This is the primary entry point documented in the README Quick Start.
It wraps the underlying ServiceOrchestrator and exposes a simple
start / get_latest_poses / stop interface.
"""
def __init__(self, host: str = "0.0.0.0", port: int = 3000, **kwargs):
self.host = host
self.port = port
self._config = kwargs
self._orchestrator = None
self._server_task = None
self._poses = []
self._running = False
# ------------------------------------------------------------------
# Public API (matches README Quick Start)
# ------------------------------------------------------------------
def start(self):
"""Start the sensing system (blocking until ready)."""
import asyncio
loop = _get_or_create_event_loop()
loop.run_until_complete(self._async_start())
async def _async_start(self):
try:
from src.config.settings import get_settings
from src.services.orchestrator import ServiceOrchestrator
settings = get_settings()
self._orchestrator = ServiceOrchestrator(settings)
await self._orchestrator.initialize()
await self._orchestrator.start()
self._running = True
logger.info("WiFiDensePose system started on %s:%s", self.host, self.port)
except ImportError:
raise ImportError(
"Core dependencies not found. Make sure you installed "
"from the repository root:\n"
" cd wifi-densepose && pip install -e .\n"
"Or install the v1 package:\n"
" cd wifi-densepose/v1 && pip install -e ."
)
def stop(self):
"""Stop the sensing system."""
import asyncio
if self._orchestrator is not None:
loop = _get_or_create_event_loop()
loop.run_until_complete(self._orchestrator.shutdown())
self._running = False
logger.info("WiFiDensePose system stopped")
def get_latest_poses(self):
"""Return the most recent list of detected pose dicts."""
if self._orchestrator is None:
return []
try:
import asyncio
loop = _get_or_create_event_loop()
return loop.run_until_complete(self._fetch_poses())
except Exception:
return []
async def _fetch_poses(self):
try:
pose_svc = self._orchestrator.pose_service
if pose_svc and hasattr(pose_svc, "get_latest"):
return await pose_svc.get_latest()
except Exception:
pass
return []
# ------------------------------------------------------------------
# Context-manager support
# ------------------------------------------------------------------
def __enter__(self):
self.start()
return self
def __exit__(self, *exc):
self.stop()
# ------------------------------------------------------------------
# Convenience re-exports
# ------------------------------------------------------------------
@staticmethod
def version():
return __version__
def _get_or_create_event_loop():
import asyncio
try:
return asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop
__all__ = ["WiFiDensePose", "__version__"]