diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 180ae6f53..01819bb2a 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -18,6 +18,8 @@ mod field_localize; mod model_format; mod multistatic_bridge; mod mediatek_csi; +mod mediatek_csi_ring; +mod mediatek_devices; mod qualcomm_csi; mod realtek_csi; mod realtek_radar; @@ -109,6 +111,12 @@ struct Args { #[arg(long, default_value = "5005")] udp_port: u16, + /// Per-device ring buffer size (frame count) for MediaTek per-subcarrier + /// CSI inspection (`/api/v1/csi/mediatek/devices/:id/frames`). See `mediatek_csi_ring`'s doc + /// comment for the memory bound at this default. + #[arg(long, default_value = "256")] + csi_ring: usize, + /// UDP bind address for the CSI receiver (ADR-296). Defaults to /// `127.0.0.1` (loopback only). Binding to a routable address (`0.0.0.0` /// or a LAN IP) is an explicit operator choice and requires `--udp-allow` @@ -185,7 +193,7 @@ struct Args { #[command(flatten)] mqtt_opts: wifi_densepose_sensing_server::cli::MqttArgs, - /// Data source: auto, wifi, esp32, simulate + /// Data source: auto, wifi, esp32, mediatek, simulate #[arg(long, default_value = "auto")] source: String, @@ -1857,6 +1865,20 @@ struct AppStateInner { latest_mediatek_csi: Option, /// Instant of the last validated MediaTek CSI UDP frame. last_mediatek_frame: Option, + /// Per-device MediaTek CSI, keyed by `device_id`, so multiple MediaTek + /// bridges (e.g. two WN586X3 routers) each keep their own slot instead of + /// overwriting `latest_mediatek_csi`. See `mediatek_devices` module and + /// `/api/v1/csi/mediatek/devices`. + mediatek_csi_by_device: + HashMap, + /// Per-device bounded ring buffer of full per-subcarrier amplitude/phase + /// (`mediatek_csi_ring`), for per-device channel inspection. + /// Sized by `csi_ring_capacity` (the `--csi-ring` flag) when a device's + /// entry is first created. + mediatek_csi_ring_by_device: HashMap, + /// Ring size (frame count) for new entries in `mediatek_csi_ring_by_device`. + /// Set once from `--csi-ring` at startup. + csi_ring_capacity: usize, /// Latest validated Qualcomm CSI summary; raw matrices are not retained here. latest_qualcomm_csi: Option, /// Instant of the last validated Qualcomm CSI UDP frame. @@ -2569,6 +2591,9 @@ impl AppStateInner { last_realtek_frame: None, latest_mediatek_csi: None, last_mediatek_frame: None, + mediatek_csi_by_device: HashMap::new(), + mediatek_csi_ring_by_device: HashMap::new(), + csi_ring_capacity: 16, latest_qualcomm_csi: None, last_qualcomm_frame: None, latest_realtek_csi: None, @@ -4682,6 +4707,19 @@ fn plan_source(requested: &str, esp32_detected: bool, wifi_detected: bool) -> So run_simulator: false, run_wifi: false, }, + // Explicit MediaTek override: bind UDP so MTC1 frames from + // wifi-densepose-mtk-bridge (or the ADR-266 simulator) are received and + // served on /api/v1/csi/mediatek/*, but never run the ESP32-shaped + // simulator — it has no per-device mediatek branch to stand down for + // (see `simulated_data_task`'s `effective_source() == "esp32"` guard) + // and would otherwise fabricate a "simulated" room classification + // forever alongside real MediaTek data. + "mediatek" => SourcePlan { + initial_source: "mediatek".to_string(), + bind_udp: true, + run_simulator: false, + run_wifi: false, + }, "wifi" => SourcePlan { initial_source: "wifi".to_string(), bind_udp: false, @@ -4763,6 +4801,36 @@ mod issue_1004_source_plan_tests { assert_eq!(plan.initial_source, "esp32"); } + // Before this arm existed, `--source mediatek` fell into the `other` + // catch-all below (bind_udp: false) and silently received nothing at all. + // Mirrors `explicit_esp32_binds_udp`: bind UDP for real MTC1 frames, but + // never run the ESP32-shaped simulator (it has no mediatek branch to + // stand down for). + #[test] + fn explicit_mediatek_binds_udp_no_simulator() { + let plan = plan_source("mediatek", false, false); + assert!(plan.bind_udp, "--source mediatek must receive MTC1 frames"); + assert!( + !plan.run_simulator, + "must not fabricate a simulated classification" + ); + assert!(!plan.run_wifi); + assert_eq!(plan.initial_source, "mediatek"); + } + + // The bug `explicit_mediatek_binds_udp_no_simulator` guards against is + // general: any vendor name not explicitly matched still falls into the + // catch-all and binds nothing. Pinned here so the next vendor addition + // remembers to add its own arm rather than relying on this default. + #[test] + fn unmatched_vendor_name_still_binds_nothing_by_default() { + let plan = plan_source("qualcomm", false, false); + assert!(!plan.bind_udp); + assert!(!plan.run_simulator); + assert!(!plan.run_wifi); + assert_eq!(plan.initial_source, "qualcomm"); + } + // Promotion check: the runtime promotes by setting `AppStateInner.source` // to "esp32" on the first real frame; `effective_source()` then reports it // (and reverts to "esp32:offline" after a 5 s gap). This asserts the @@ -9782,11 +9850,26 @@ async fn udp_receiver_task( let snapshot = mediatek_csi::MediatekCsiSnapshot::from_frame(&frame); debug!("MediaTek CSI from {src}: profile={} seq={} dimensions={}x{}x{}", snapshot.chipset, snapshot.sequence, snapshot.tx_count, snapshot.rx_count, snapshot.subcarrier_count); let json = serde_json::to_string(&snapshot).ok(); + let now = std::time::Instant::now(); + let device_id = snapshot.device_id.clone(); + let source_label = snapshot.source.to_string(); let mut s = state.write().await; - s.source = snapshot.source.to_string(); - s.last_mediatek_frame = Some(std::time::Instant::now()); + s.source = source_label.clone(); + s.last_mediatek_frame = Some(now); + s.mediatek_csi_by_device + .insert(device_id.clone(), (snapshot.clone(), now)); s.latest_mediatek_csi = Some(snapshot); if let Some(json) = json { let _ = s.tx.send(json); } + + // Per-device channel inspection: retain full + // per-subcarrier amplitude/phase from this + // frame's raw I/Q, bounded to the last + // `csi_ring_capacity` frames per device. + let ring_capacity = s.csi_ring_capacity; + s.mediatek_csi_ring_by_device + .entry(device_id.clone()) + .or_insert_with(|| mediatek_csi_ring::DeviceRing::new(ring_capacity)) + .push(&frame); } Ok((_, consumed)) => warn!("MediaTek CSI datagram from {src} has trailing bytes: consumed={consumed} received={len}"), Err(error) => warn!("Rejected MediaTek CSI datagram from {src}: {error}"), @@ -12119,6 +12202,9 @@ async fn main() { last_realtek_frame: None, latest_mediatek_csi: None, last_mediatek_frame: None, + mediatek_csi_by_device: HashMap::new(), + mediatek_csi_ring_by_device: HashMap::new(), + csi_ring_capacity: args.csi_ring, latest_qualcomm_csi: None, last_qualcomm_frame: None, latest_realtek_csi: None, @@ -12472,6 +12558,18 @@ async fn main() { .route("/api/v1/sensing/latest", get(latest)) .route("/api/v1/radar/latest", get(latest_realtek_radar)) .route("/api/v1/csi/mediatek/latest", get(latest_mediatek_csi)) + .route( + "/api/v1/csi/mediatek/devices", + get(mediatek_devices::mediatek_csi_devices), + ) + .route( + "/api/v1/csi/mediatek/devices/:device_id/frames", + get(mediatek_csi_ring::mediatek_csi_frames), + ) + .route( + "/api/v1/csi/mediatek/devices/:device_id/summary", + get(mediatek_csi_ring::mediatek_csi_summary), + ) .route("/api/v1/csi/qualcomm/latest", get(latest_qualcomm_csi)) .route("/api/v1/csi/realtek/latest", get(latest_realtek_csi)) .route("/api/v1/rf/vendors", get(vendor_descriptors)) diff --git a/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi.rs b/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi.rs index 581732205..f4bd51751 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi.rs @@ -35,13 +35,23 @@ pub(crate) struct MediatekCsiSnapshot { impl MediatekCsiSnapshot { pub(crate) fn from_frame(frame: &CsiFrame) -> Self { let synthetic = frame.flags.contains(CsiFlags::SYNTHETIC); + let calibrated = frame.flags.contains(CsiFlags::CALIBRATED); let (mean_amplitude, peak_amplitude) = amplitude_summary(frame); Self { event_type: "mediatek_csi", + // Provenance is decided by the frame's own flags, in this order, and + // never by which ingest path delivered it (ADR-267). SYNTHETIC is + // checked first and is not clearable by anything downstream: a + // simulated frame stays labelled simulated even when it also claims + // CALIBRATED, which the ADR-266 simulator does. source: if synthetic { "mediatek:simulated" - } else { + } else if calibrated { "mediatek" + } else { + // Real silicon, but nothing has validated this capture against a + // calibration. `wifi-densepose-mtk-bridge` emits exactly this class. + "mediatek:physical-unvalidated" }, report_kind: match frame.report_kind { ReportKind::Csi => "csi", @@ -60,7 +70,7 @@ impl MediatekCsiSnapshot { ppdu_type: format!("{:?}", frame.ppdu_type).to_ascii_lowercase(), rssi_dbm: frame.payload.rssi_dbm().to_vec(), noise_floor_dbm: frame.noise_floor_dbm, - calibrated: frame.flags.contains(CsiFlags::CALIBRATED), + calibrated, synthetic, saturated: frame.flags.contains(CsiFlags::SATURATED), time_synchronized: frame.flags.contains(CsiFlags::TIME_SYNCHRONIZED), @@ -116,6 +126,79 @@ mod tests { assert!(snapshot.peak_amplitude.unwrap() >= snapshot.mean_amplitude.unwrap()); } + /// A frame off real silicon that has not been calibrated must be + /// distinguishable from both a simulation and a validated capture. + /// `wifi-densepose-mtk-bridge` produces exactly this shape. + #[test] + fn physical_uncalibrated_frame_is_labelled_physical_unvalidated() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); + let mut frame = sim.next_frame(); + frame.flags = CsiFlags(0); + let snapshot = MediatekCsiSnapshot::from_frame(&frame); + assert_eq!(snapshot.source, "mediatek:physical-unvalidated"); + assert!(!snapshot.synthetic); + assert!(!snapshot.calibrated); + } + + /// Plain `"mediatek"` is reserved for calibrated physical frames. + #[test] + fn calibrated_physical_frame_keeps_the_plain_mediatek_label() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); + let mut frame = sim.next_frame(); + frame.flags = CsiFlags(CsiFlags::CALIBRATED); + let snapshot = MediatekCsiSnapshot::from_frame(&frame); + assert_eq!(snapshot.source, "mediatek"); + assert!(snapshot.calibrated); + assert!(!snapshot.synthetic); + } + + /// ADR-267: SYNTHETIC is not clearable by an ingest path. No combination of + /// other flags may promote a simulated frame out of `mediatek:simulated`. + #[test] + fn synthetic_frame_can_never_be_relabelled_as_physical() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); + let base = sim.next_frame(); + for extra in [ + 0, + CsiFlags::CALIBRATED, + CsiFlags::TIME_SYNCHRONIZED, + CsiFlags::SATURATED | CsiFlags::DROPPED_PREDECESSOR, + CsiFlags::CALIBRATED | CsiFlags::TIME_SYNCHRONIZED | CsiFlags::SATURATED, + ] { + let mut frame = base.clone(); + frame.flags = CsiFlags(CsiFlags::SYNTHETIC | extra); + let snapshot = MediatekCsiSnapshot::from_frame(&frame); + assert_eq!( + snapshot.source, "mediatek:simulated", + "flags {:#06x} must stay simulated", + frame.flags.0 + ); + assert!(snapshot.synthetic); + } + } + + /// The three labels are distinct, and all three still start with + /// `"mediatek"`, which is what the server's staleness check keys on. + #[test] + fn the_three_provenance_labels_are_distinct_and_all_prefixed_mediatek() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); + let base = sim.next_frame(); + let label = |flags: u16| { + let mut f = base.clone(); + f.flags = CsiFlags(flags); + MediatekCsiSnapshot::from_frame(&f).source + }; + let simulated = label(CsiFlags::SYNTHETIC | CsiFlags::CALIBRATED); + let calibrated = label(CsiFlags::CALIBRATED); + let physical = label(0); + assert_eq!(simulated, "mediatek:simulated"); + assert_eq!(calibrated, "mediatek"); + assert_eq!(physical, "mediatek:physical-unvalidated"); + for l in [simulated, calibrated, physical] { + assert!(l.starts_with("mediatek"), "{l}"); + } + } + #[test] fn capability_summary_does_not_invent_signal_statistics() { let sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); diff --git a/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi_ring.rs b/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi_ring.rs new file mode 100644 index 000000000..6bd9c926a --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/mediatek_csi_ring.rs @@ -0,0 +1,848 @@ +//! Per-device ring buffer of full per-subcarrier MediaTek CSI, for +//! per-device channel inspection and analysis through the +//! `/api/v1/csi/mediatek/devices/:device_id/{frames,summary}` routes. +//! +//! `MediatekCsiSnapshot` (`mediatek_csi.rs`) deliberately keeps only +//! mean/peak amplitude — enough for a coarse summary, but nothing for a +//! human to actually look at the channel with. This module extracts full per-chain, per-subcarrier +//! amplitude and phase directly from each frame's raw MTC1 I/Q payload at +//! ingest, and retains the last `capacity` frames per device — bounded, so +//! memory stays fixed regardless of how long a device has been streaming. +//! +//! # Memory bound +//! +//! `ring_size * chains * subcarrier_count * 2 (amplitude + phase) * 4 bytes (f32)`. +//! At the defaults this ADR-267 targets (N=256 frames, 2x2 MIMO = 4 chains, +//! 64 subcarriers): `256 * 4 * 64 * 2 * 4 = 524,288 bytes ≈ 0.5 MiB` per +//! device (plus a small, comparatively negligible per-frame Vec for +//! `rssi_dbm`, one byte per Rx chain). + +use crate::SharedState; +use axum::extract::{Path, Query, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Json}; +use serde::Deserialize; +use wifi_densepose_hardware::mediatek_csi::{CsiFrame, CsiPayload}; + +/// One (tx, rx) chain's per-subcarrier amplitude and phase for one frame. +#[derive(Debug, Clone)] +pub(crate) struct ChainSamples { + pub(crate) tx: u8, + pub(crate) rx: u8, + /// Real units (`hypot(i, q) * frame.scale`), one per subcarrier. + pub(crate) amplitude: Vec, + /// Radians, one per subcarrier, unwrapped along the subcarrier axis + /// within this frame only (see [`unwrap_phase`]) — never carried across + /// frames or across chains. + pub(crate) phase: Vec, +} + +/// One retained frame: everything a client needs to plot or replay it. +#[derive(Debug, Clone)] +pub(crate) struct RingFrame { + pub(crate) timestamp_us: u64, + pub(crate) sequence: u32, + /// One signed RSSI byte per Rx chain (ADR-267) — indexed by Rx chain + /// only, NOT parallel to `chains` (which has one entry per Tx/Rx pair). + pub(crate) rssi_dbm: Vec, + pub(crate) chains: Vec, +} + +/// Bounded per-device ring buffer of [`RingFrame`]s, oldest first. +#[derive(Debug, Clone)] +pub(crate) struct DeviceRing { + capacity: usize, + frames: std::collections::VecDeque, + subcarrier_count: u16, +} + +impl DeviceRing { + pub(crate) fn new(capacity: usize) -> Self { + Self { + capacity: capacity.max(1), + frames: std::collections::VecDeque::new(), + subcarrier_count: 0, + } + } + + /// Extract per-chain amplitude/phase from `frame`'s raw I/Q payload and + /// push it into the ring, evicting the oldest frame once over capacity. + /// A `CsiPayload::Bytes` report (e.g. an ADR-267 capabilities frame, + /// which carries no I/Q) is not retained. + pub(crate) fn push(&mut self, frame: &CsiFrame) { + let tx = frame.tx_count as usize; + let rx = frame.rx_count as usize; + let sc = frame.subcarrier_count as usize; + let chains = match &frame.payload { + CsiPayload::ComplexI16 { values, .. } => extract_chains( + values.iter().map(|[i, q]| (*i as f32, *q as f32)), + tx, + rx, + sc, + frame.scale, + ), + CsiPayload::ComplexF32 { values, .. } => extract_chains( + values.iter().map(|[i, q]| (*i, *q)), + tx, + rx, + sc, + frame.scale, + ), + CsiPayload::Bytes(_) => return, + }; + self.subcarrier_count = frame.subcarrier_count; + self.frames.push_back(RingFrame { + timestamp_us: frame.timestamp_us, + sequence: frame.sequence, + rssi_dbm: frame.payload.rssi_dbm().to_vec(), + chains, + }); + while self.frames.len() > self.capacity { + self.frames.pop_front(); + } + } + + pub(crate) fn len(&self) -> usize { + self.frames.len() + } + + pub(crate) fn is_empty(&self) -> bool { + self.frames.is_empty() + } + + pub(crate) fn subcarrier_count(&self) -> u16 { + self.subcarrier_count + } + + /// Oldest first. + pub(crate) fn iter(&self) -> impl DoubleEndedIterator { + self.frames.iter() + } + + pub(crate) fn last(&self) -> Option<&RingFrame> { + self.frames.back() + } + + /// Real elapsed time covered by the ring's retained frames (0 with + /// fewer than 2 frames — there's no interval to measure yet). + pub(crate) fn span_us(&self) -> u64 { + match (self.frames.front(), self.frames.back()) { + (Some(oldest), Some(newest)) if self.frames.len() > 1 => { + newest.timestamp_us.saturating_sub(oldest.timestamp_us) + } + _ => 0, + } + } + + /// Frames per second implied by `span_us()` and the retained frame + /// count. `0.0` with fewer than 2 frames. + pub(crate) fn frame_rate_hz(&self) -> f64 { + let span_us = self.span_us(); + if self.frames.len() < 2 || span_us == 0 { + return 0.0; + } + (self.frames.len() - 1) as f64 / (span_us as f64 / 1_000_000.0) + } + + /// Per-chain amplitude mean and standard deviation across every + /// retained frame that carried that chain, per subcarrier — the "what + /// does the channel usually look like" view for an inspection client. + /// + /// 2026-09-20: this used to take chain layout from only the newest + /// frame and match older frames to it positionally (by index into that + /// frame's chain list). A device that interleaves PPDU types with + /// different dimensions (e.g. legacy 1x2 alongside HT 2x2) legitimately + /// varies chain count frame to frame, so that silently dropped whichever + /// chains the newest frame didn't happen to carry, and matched other + /// frames' chains by position rather than identity. Chain identity here + /// is now the layout: this returns one entry per **distinct (tx, rx) + /// pair seen anywhere in the ring**, in first-seen order, each averaged + /// only over the frames that actually carried it (`frames_with_chain` + /// says how many, out of `self.len()`). + pub(crate) fn amplitude_stats(&self) -> Vec { + let sc = self.subcarrier_count as usize; + chain_union(self.frames.iter()) + .into_iter() + .map(|(tx, rx)| { + let mut sum = vec![0.0f64; sc]; + let mut sumsq = vec![0.0f64; sc]; + let mut n = 0usize; + for frame in &self.frames { + let Some(c) = frame.chains.iter().find(|c| c.tx == tx && c.rx == rx) else { + continue; + }; + n += 1; + for (k, &a) in c.amplitude.iter().enumerate().take(sc) { + sum[k] += a as f64; + sumsq[k] += (a as f64) * (a as f64); + } + } + let n_f = n.max(1) as f64; + let mean: Vec = sum.iter().map(|&s| (s / n_f) as f32).collect(); + let std: Vec = sum + .iter() + .zip(sumsq.iter()) + .map(|(&s, &sq)| { + let m = s / n_f; + (sq / n_f - m * m).max(0.0).sqrt() as f32 + }) + .collect(); + ChainAmplitudeStats { + tx, + rx, + amplitude_mean: mean, + amplitude_std: std, + frames_with_chain: n, + } + }) + .collect() + } + + /// Mean RSSI per Rx chain across every retained frame. + pub(crate) fn rssi_by_rx_chain(&self) -> Vec<(u8, f64)> { + let n_rx = self.frames.back().map(|f| f.rssi_dbm.len()).unwrap_or(0); + (0..n_rx) + .map(|rx| { + let (sum, n) = self + .frames + .iter() + .filter_map(|f| f.rssi_dbm.get(rx)) + .fold((0i64, 0u32), |(s, n), &v| (s + v as i64, n + 1)); + let mean = if n > 0 { sum as f64 / n as f64 } else { 0.0 }; + (rx as u8, mean) + }) + .collect() + } +} + +pub(crate) struct ChainAmplitudeStats { + pub(crate) tx: u8, + pub(crate) rx: u8, + pub(crate) amplitude_mean: Vec, + pub(crate) amplitude_std: Vec, + /// How many of the averaged-over frames actually carried this chain — + /// out of `self.len()` for a whole-ring summary, so the UI can show + /// when a chain's stats are based on a partial subset (e.g. a legacy + /// 1x2 frame interleaved with HT 2x2 ones only contributes to 2 of the + /// 4 possible chains). + pub(crate) frames_with_chain: usize, +} + +/// Every distinct `(tx, rx)` pair present in any of `frames`, in first-seen +/// order. Used both for the whole-ring summary and for a windowed +/// `/frames` response's top-level `chains` field — "all chain positions +/// seen [in this scope]", since a per-frame layout can legitimately vary +/// (interleaved PPDU types with different dimensions). +fn chain_union<'a>(frames: impl Iterator) -> Vec<(u8, u8)> { + let mut seen = Vec::new(); + for frame in frames { + for c in &frame.chains { + if !seen.contains(&(c.tx, c.rx)) { + seen.push((c.tx, c.rx)); + } + } + } + seen +} + +fn extract_chains( + iq: impl Iterator, + tx: usize, + rx: usize, + sc: usize, + scale: f32, +) -> Vec { + let values: Vec<(f32, f32)> = iq.collect(); + let mut chains = Vec::with_capacity(tx * rx); + for t in 0..tx { + for r in 0..rx { + let base = (t * rx + r) * sc; + let mut amplitude = Vec::with_capacity(sc); + let mut phase = Vec::with_capacity(sc); + for k in 0..sc { + let idx = base + k; + let Some(&(i, q)) = values.get(idx) else { + break; + }; + amplitude.push(i.hypot(q) * scale); + phase.push(q.atan2(i)); + } + unwrap_phase(&mut phase); + chains.push(ChainSamples { + tx: t as u8, + rx: r as u8, + amplitude, + phase, + }); + } + } + chains +} + +/// `numpy.unwrap`-style phase unwrapping along one frame's subcarrier +/// axis: whenever a consecutive step exceeds +/- pi, fold it back by the +/// nearest multiple of 2*pi so the sequence reads as a continuous phase +/// ramp across frequency instead of wrapping at +/- pi. Frame-local only — +/// never carries state across frames, and computed independently per chain. +fn unwrap_phase(phase: &mut [f32]) { + const TWO_PI: f32 = std::f32::consts::TAU; + for i in 1..phase.len() { + let mut delta = phase[i] - phase[i - 1]; + while delta > std::f32::consts::PI { + phase[i] -= TWO_PI; + delta -= TWO_PI; + } + while delta < -std::f32::consts::PI { + phase[i] += TWO_PI; + delta += TWO_PI; + } + } +} + +#[derive(Debug, Deserialize)] +pub(crate) struct CsiFramesQuery { + /// Frames to return, newest `n`. Capped at the ring's retained frame + /// count (never an error to ask for more than exists). + n: Option, + /// Comma-separated subset of `"amp"`, `"phase"`. Both when omitted. + fields: Option, +} + +fn wants(fields: &Option, name: &str) -> bool { + fields + .as_deref() + .is_none_or(|f| f.split(',').any(|x| x.trim() == name)) +} + +fn device_not_found(device_id: &str) -> (StatusCode, Json) { + ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "unknown device_id", "device_id": device_id })), + ) +} + +/// `GET /api/v1/csi/mediatek/devices/{device_id}/frames?n=64&fields=amp,phase` +/// +/// `{device_id, subcarrier_count, chains: [{tx, rx}], frames: [{ts_us, seq, +/// rssi, chains: [{tx, rx}], amp: [[..per subcarrier..] per chain], phase: +/// [..]}]}`, oldest first within `frames`. +/// +/// 2026-09-20: a 2x2 device legitimately interleaves PPDU types with +/// different dimensions (e.g. legacy 1x2 alongside HT 2x2), so a single +/// window can contain frames whose `amp`/`phase` row count differs. The +/// top-level `chains` is every chain position seen anywhere **in this +/// returned window** (not just the newest frame) — a discovery list, not an +/// index. Each frame additionally carries its **own** `chains` array, +/// naming exactly what its `amp[i]`/`phase[i]` rows are — a consumer must +/// index by a frame's own `chains`, never by the top-level one, since a +/// short frame's rows don't align with the top-level list's positions. +/// +/// `404` when `device_id` has no ring at all (never reported, or +/// `mediatek_csi_by_device` only ever saw a Bytes/capabilities report). +/// `amp`/`phase` per frame are omitted (not `null`) when excluded by +/// `fields`. +pub(crate) async fn mediatek_csi_frames( + State(state): State, + Path(device_id): Path, + Query(query): Query, +) -> impl IntoResponse { + let s = state.read().await; + let Some(ring) = s.mediatek_csi_ring_by_device.get(&device_id) else { + return device_not_found(&device_id).into_response(); + }; + if ring.is_empty() { + return device_not_found(&device_id).into_response(); + } + let include_amp = wants(&query.fields, "amp"); + let include_phase = wants(&query.fields, "phase"); + let n = query.n.unwrap_or(64).min(ring.len()); + let skip = ring.len() - n; + let window: Vec<&RingFrame> = ring.iter().skip(skip).collect(); + + let frames: Vec = window + .iter() + .map(|f| { + let frame_chains: Vec = f + .chains + .iter() + .map(|c| serde_json::json!({ "tx": c.tx, "rx": c.rx })) + .collect(); + let mut obj = serde_json::json!({ + "ts_us": f.timestamp_us, + "seq": f.sequence, + "rssi": f.rssi_dbm, + "chains": frame_chains, + }); + if let Some(map) = obj.as_object_mut() { + if include_amp { + map.insert( + "amp".to_string(), + serde_json::json!(f + .chains + .iter() + .map(|c| &c.amplitude) + .collect::>()), + ); + } + if include_phase { + map.insert( + "phase".to_string(), + serde_json::json!(f.chains.iter().map(|c| &c.phase).collect::>()), + ); + } + } + obj + }) + .collect(); + + let chains: Vec = chain_union(window.into_iter()) + .into_iter() + .map(|(tx, rx)| serde_json::json!({ "tx": tx, "rx": rx })) + .collect(); + + ( + StatusCode::OK, + Json(serde_json::json!({ + "device_id": device_id, + "subcarrier_count": ring.subcarrier_count(), + "chains": chains, + "frames": frames, + })), + ) + .into_response() +} + +/// `GET /api/v1/csi/mediatek/devices/{device_id}/summary` — one call for the +/// analysis view: per-subcarrier amplitude mean/std per chain +/// over the ring, mean RSSI per Rx chain, and the ring's real frame rate. +/// `404` when `device_id` has no ring at all. +pub(crate) async fn mediatek_csi_summary( + State(state): State, + Path(device_id): Path, +) -> impl IntoResponse { + let s = state.read().await; + let Some(ring) = s.mediatek_csi_ring_by_device.get(&device_id) else { + return device_not_found(&device_id).into_response(); + }; + if ring.is_empty() { + return device_not_found(&device_id).into_response(); + } + + let chains: Vec = ring + .amplitude_stats() + .into_iter() + .map(|c| { + serde_json::json!({ + "tx": c.tx, + "rx": c.rx, + "amplitude_mean": c.amplitude_mean, + "amplitude_std": c.amplitude_std, + "frames_with_chain": c.frames_with_chain, + }) + }) + .collect(); + let rssi_by_rx_chain: Vec = ring + .rssi_by_rx_chain() + .into_iter() + .map(|(rx, mean)| serde_json::json!({ "rx": rx, "mean_rssi_dbm": mean })) + .collect(); + let ring_frame_count = ring.len(); + let ring_span_ms = ring.span_us() / 1000; + let frame_rate_hz = ring.frame_rate_hz(); + let subcarrier_count = ring.subcarrier_count(); + + ( + StatusCode::OK, + Json(serde_json::json!({ + "device_id": device_id, + "subcarrier_count": subcarrier_count, + "ring_frame_count": ring_frame_count, + "ring_span_ms": ring_span_ms, + "frame_rate_hz": frame_rate_hz, + "rssi_by_rx_chain": rssi_by_rx_chain, + "chains": chains, + })), + ) + .into_response() +} + +#[cfg(test)] +mod tests { + use super::*; + use wifi_densepose_hardware::mediatek_csi::simulator::{MediatekCsiSimulator, SimulatorConfig}; + + #[test] + fn ring_retains_frames_in_order_up_to_capacity() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); + let mut ring = DeviceRing::new(3); + let seqs: Vec = (0..5) + .map(|_| { + let frame = sim.next_frame(); + ring.push(&frame); + frame.sequence + }) + .collect(); + assert_eq!(ring.len(), 3, "must evict down to capacity"); + let retained: Vec = ring.iter().map(|f| f.sequence).collect(); + assert_eq!( + retained, + seqs[2..], + "oldest-first, only the last `capacity` frames survive" + ); + } + + #[test] + fn capabilities_frame_is_not_retained() { + let sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); + let mut ring = DeviceRing::new(8); + ring.push(&sim.capabilities_frame()); + assert!( + ring.is_empty(), + "a Bytes-payload capabilities frame carries no I/Q to retain" + ); + } + + #[test] + fn chain_layout_matches_tx_rx_dimensions() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 2, + rx_count: 3, + subcarriers: 8, + ..Default::default() + }) + .unwrap(); + let mut ring = DeviceRing::new(4); + ring.push(&sim.next_frame()); + let frame = ring.last().unwrap(); + assert_eq!(frame.chains.len(), 6, "2 tx x 3 rx = 6 chains"); + for chain in &frame.chains { + assert_eq!(chain.amplitude.len(), 8); + assert_eq!(chain.phase.len(), 8); + } + assert_eq!( + frame.rssi_dbm.len(), + 3, + "one RSSI byte per Rx chain, not per tx/rx pair" + ); + } + + #[test] + fn unwrap_removes_discontinuities_bigger_than_pi() { + // A synthetic phase ramp that would wrap at +/- pi without unwrapping. + let mut phase = vec![3.0, -3.0, 3.0, -3.0]; + unwrap_phase(&mut phase); + for w in phase.windows(2) { + assert!( + (w[1] - w[0]).abs() < std::f32::consts::PI + 1e-3, + "no residual jump > pi: {:?}", + phase + ); + } + } + + #[test] + fn amplitude_stats_and_frame_rate_are_sane_over_a_short_run() { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig::default()).unwrap(); + let mut ring = DeviceRing::new(16); + for _ in 0..10 { + ring.push(&sim.next_frame()); + } + let stats = ring.amplitude_stats(); + assert_eq!( + stats.len(), + 2 * 3, + "default simulator config is 2 tx x 3 rx" + ); + for s in &stats { + assert_eq!(s.amplitude_mean.len(), ring.subcarrier_count() as usize); + assert_eq!(s.amplitude_std.len(), ring.subcarrier_count() as usize); + } + // Default simulator frame_period_us = 20_000 -> 10 frames span 9 periods. + assert!(ring.frame_rate_hz() > 0.0); + let rssi = ring.rssi_by_rx_chain(); + assert_eq!(rssi.len(), 3); + } + + /// 2026-09-20 field bug: a 2x2 device interleaves legacy 1x2 and HT 2x2 + /// PPDUs, so per-frame chain count legitimately varies within one ring. + /// `amplitude_stats` must union chain positions across every frame + /// (not just the newest) and report how many frames actually backed + /// each one. + #[test] + fn interleaved_1x2_and_2x2_frames_are_unioned_correctly() { + let mut sim_1x2 = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 1, + rx_count: 2, + subcarriers: 4, + ..Default::default() + }) + .unwrap(); + let mut sim_2x2 = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 2, + rx_count: 2, + subcarriers: 4, + ..Default::default() + }) + .unwrap(); + let mut ring = DeviceRing::new(8); + // Interleaved: 1x2, 2x2, 1x2, 2x2. + ring.push(&sim_1x2.next_frame()); + ring.push(&sim_2x2.next_frame()); + ring.push(&sim_1x2.next_frame()); + ring.push(&sim_2x2.next_frame()); + assert_eq!(ring.len(), 4); + + let counts: Vec = ring.iter().map(|f| f.chains.len()).collect(); + assert_eq!( + counts, + vec![2, 4, 2, 4], + "per-frame chain count must legitimately vary" + ); + + let stats = ring.amplitude_stats(); + assert_eq!( + stats.len(), + 4, + "union of the 1x2 frames' 2 chains and the 2x2 frames' 4 chains" + ); + let by_pos: std::collections::HashMap<(u8, u8), usize> = stats + .iter() + .map(|s| ((s.tx, s.rx), s.frames_with_chain)) + .collect(); + assert_eq!( + by_pos[&(0, 0)], + 4, + "(0,0) exists in every frame, 1x2 and 2x2 alike" + ); + assert_eq!(by_pos[&(0, 1)], 4); + assert_eq!( + by_pos[&(1, 0)], + 2, + "(1,0) only exists in the two 2x2 frames" + ); + assert_eq!(by_pos[&(1, 1)], 2); + } + + mod endpoint_tests { + //! Exercises `mediatek_csi_frames` / `mediatek_csi_summary` directly + //! against `AppStateInner`, constructing the axum extractors by hand + //! rather than routing a full `Request` through a `Router` — these + //! handlers take no headers/body, so `Path`/`Query`/`State` cover + //! everything a real request would provide. + use super::*; + use crate::AppStateInner; + use axum::response::IntoResponse; + use std::sync::Arc; + use tokio::sync::RwLock; + use wifi_densepose_hardware::mediatek_csi::simulator::{ + MediatekCsiSimulator, SimulatorConfig, + }; + + async fn body_json( + response: axum::response::Response, + ) -> (axum::http::StatusCode, serde_json::Value) { + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), 1_000_000) + .await + .unwrap(); + (status, serde_json::from_slice(&bytes).unwrap()) + } + + fn seed_two_frames(s: &mut AppStateInner, device_id: &str) { + let mut sim = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 1, + rx_count: 1, + subcarriers: 4, + ..Default::default() + }) + .unwrap(); + let mut ring = DeviceRing::new(8); + ring.push(&sim.next_frame()); + ring.push(&sim.next_frame()); + s.mediatek_csi_ring_by_device + .insert(device_id.to_string(), ring); + } + + #[tokio::test] + async fn two_ingested_frames_return_in_order_with_correct_shapes() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + seed_two_frames(&mut *state.write().await, "dev-a"); + + let response = mediatek_csi_frames( + State(state), + Path("dev-a".to_string()), + Query(CsiFramesQuery { + n: None, + fields: None, + }), + ) + .await + .into_response(); + let (status, body) = body_json(response).await; + assert_eq!(status, axum::http::StatusCode::OK); + assert_eq!(body["device_id"], "dev-a"); + assert_eq!(body["subcarrier_count"], 4); + + let frames = body["frames"].as_array().expect("frames array"); + assert_eq!(frames.len(), 2, "both ingested frames must be returned"); + // Oldest first: sequence 0 then 1 (MediatekCsiSimulator starts at 0). + assert_eq!(frames[0]["seq"], 0); + assert_eq!(frames[1]["seq"], 1); + + let chains = body["chains"].as_array().expect("chains array"); + assert_eq!(chains.len(), 1, "1 tx x 1 rx = 1 chain"); + + let amp = frames[0]["amp"].as_array().expect("amp array"); + assert_eq!(amp.len(), 1, "one amplitude vector per chain"); + assert_eq!( + amp[0].as_array().unwrap().len(), + 4, + "one value per subcarrier" + ); + let phase = frames[0]["phase"].as_array().expect("phase array"); + assert_eq!(phase[0].as_array().unwrap().len(), 4); + } + + #[tokio::test] + async fn fields_query_param_limits_which_arrays_are_returned() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + seed_two_frames(&mut *state.write().await, "dev-a"); + + let response = mediatek_csi_frames( + State(state), + Path("dev-a".to_string()), + Query(CsiFramesQuery { + n: Some(1), + fields: Some("amp".to_string()), + }), + ) + .await + .into_response(); + let (_, body) = body_json(response).await; + let frames = body["frames"].as_array().unwrap(); + assert_eq!(frames.len(), 1, "n=1 must cap the returned frame count"); + assert!(frames[0].get("amp").is_some()); + assert!( + frames[0].get("phase").is_none(), + "phase must be omitted, not null, when excluded" + ); + } + + #[tokio::test] + async fn summary_matches_the_seeded_ring() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + seed_two_frames(&mut *state.write().await, "dev-a"); + + let response = mediatek_csi_summary(State(state), Path("dev-a".to_string())) + .await + .into_response(); + let (status, body) = body_json(response).await; + assert_eq!(status, axum::http::StatusCode::OK); + assert_eq!(body["device_id"], "dev-a"); + assert_eq!(body["subcarrier_count"], 4); + assert_eq!(body["ring_frame_count"], 2); + + let chains = body["chains"].as_array().expect("chains array"); + assert_eq!(chains.len(), 1); + assert_eq!(chains[0]["amplitude_mean"].as_array().unwrap().len(), 4); + assert_eq!(chains[0]["amplitude_std"].as_array().unwrap().len(), 4); + + let rssi = body["rssi_by_rx_chain"] + .as_array() + .expect("rssi_by_rx_chain array"); + assert_eq!(rssi.len(), 1, "1 rx chain"); + } + + #[tokio::test] + async fn unknown_device_returns_404_on_both_endpoints() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + + let frames_resp = mediatek_csi_frames( + State(state.clone()), + Path("no-such-device".to_string()), + Query(CsiFramesQuery { + n: None, + fields: None, + }), + ) + .await + .into_response(); + assert_eq!(frames_resp.status(), axum::http::StatusCode::NOT_FOUND); + + let summary_resp = + mediatek_csi_summary(State(state), Path("no-such-device".to_string())) + .await + .into_response(); + assert_eq!(summary_resp.status(), axum::http::StatusCode::NOT_FOUND); + } + + /// 2026-09-20 field bug: a consumer indexing a short frame's + /// `amp`/`phase` rows by the top-level `chains` list read undefined + /// on legacy 1x2 frames interleaved with HT 2x2 ones. Each frame + /// must carry its own `chains`, and the top-level list must be the + /// union across the returned window, not just the newest frame. + #[tokio::test] + async fn interleaved_1x2_and_2x2_frames_get_correct_per_frame_chains() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + { + let mut s = state.write().await; + let mut sim_1x2 = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 1, + rx_count: 2, + subcarriers: 4, + ..Default::default() + }) + .unwrap(); + let mut sim_2x2 = MediatekCsiSimulator::new(SimulatorConfig { + tx_count: 2, + rx_count: 2, + subcarriers: 4, + ..Default::default() + }) + .unwrap(); + let mut ring = DeviceRing::new(8); + ring.push(&sim_1x2.next_frame()); + ring.push(&sim_2x2.next_frame()); + s.mediatek_csi_ring_by_device + .insert("dev-mixed".to_string(), ring); + } + + let response = mediatek_csi_frames( + State(state), + Path("dev-mixed".to_string()), + Query(CsiFramesQuery { + n: None, + fields: None, + }), + ) + .await + .into_response(); + let (status, body) = body_json(response).await; + assert_eq!(status, axum::http::StatusCode::OK); + + let top_chains = body["chains"].as_array().unwrap(); + assert_eq!( + top_chains.len(), + 4, + "union across the window: the 1x2 frame's 2 plus the 2x2 frame's 2 new ones" + ); + + let frames = body["frames"].as_array().unwrap(); + assert_eq!(frames.len(), 2); + + let f0_chains = frames[0]["chains"].as_array().unwrap(); + let f0_amp = frames[0]["amp"].as_array().unwrap(); + assert_eq!(f0_chains.len(), 2, "first frame is 1x2 -> 2 chains"); + assert_eq!( + f0_amp.len(), + 2, + "amp rows must match THIS frame's own chains, not the top-level union" + ); + + let f1_chains = frames[1]["chains"].as_array().unwrap(); + let f1_amp = frames[1]["amp"].as_array().unwrap(); + assert_eq!(f1_chains.len(), 4, "second frame is 2x2 -> 4 chains"); + assert_eq!(f1_amp.len(), 4); + } + } +} diff --git a/v2/crates/wifi-densepose-sensing-server/src/mediatek_devices.rs b/v2/crates/wifi-densepose-sensing-server/src/mediatek_devices.rs new file mode 100644 index 000000000..d4ea519c2 --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/mediatek_devices.rs @@ -0,0 +1,171 @@ +//! Per-device MediaTek CSI listing. +//! +//! Multiple MediaTek bridges (e.g. two WN586X3 routers) each carry a distinct +//! `device_id` in their `MTC1` frames (ADR-267), but the single +//! `AppStateInner::latest_mediatek_csi` slot only ever remembers the most +//! recently arrived one — a second unit silently overwrites the first. This +//! module reads `AppStateInner::mediatek_csi_by_device`, a per-device map the +//! UDP ingest loop in `main.rs` populates alongside (not instead of) that +//! single slot, so every device can be fused as its own sensor. + +use crate::mediatek_csi::MediatekCsiSnapshot; +use crate::SharedState; +use axum::extract::State; +use axum::response::Json; +use std::time::{Duration, Instant}; + +/// Entries older than this are dropped from the `/devices` response. The map +/// itself is not pruned here — a device that starts sending again simply +/// reappears once its snapshot is fresh. `pub(crate)` so `mediatek_activity` +/// can apply the SAME cutoff (a downstream consumer observed `/devices` +/// and `activity.devices` naming different receiver sets after one unit +/// restarted, because they used different staleness windows) rather than a +/// second constant that could drift out of sync with this one. +pub(crate) const MEDIATEK_DEVICE_STALE: Duration = Duration::from_secs(30); + +/// `GET /api/v1/csi/mediatek/devices` — every MediaTek device seen within the +/// staleness window, each snapshot annotated with `age_ms`, sorted by +/// `device_id`. Shape: `{"devices": [...], "total": N}`. +pub(crate) async fn mediatek_csi_devices( + State(state): State, +) -> Json { + let s = state.read().await; + let now = Instant::now(); + + let mut entries: Vec<(&String, &MediatekCsiSnapshot, Duration)> = s + .mediatek_csi_by_device + .iter() + .filter_map(|(device_id, (snapshot, seen))| { + let age = now.saturating_duration_since(*seen); + (age <= MEDIATEK_DEVICE_STALE).then_some((device_id, snapshot, age)) + }) + .collect(); + entries.sort_by(|a, b| a.0.cmp(b.0)); + + let devices: Vec = entries + .into_iter() + .filter_map(|(_, snapshot, age)| { + let mut value = serde_json::to_value(snapshot).ok()?; + if let Some(obj) = value.as_object_mut() { + obj.insert( + "age_ms".to_string(), + serde_json::json!(age.as_millis() as u64), + ); + } + Some(value) + }) + .collect(); + + Json(serde_json::json!({ "total": devices.len(), "devices": devices })) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::AppStateInner; + use std::sync::Arc; + use tokio::sync::RwLock; + use wifi_densepose_hardware::mediatek_csi::simulator::{MediatekCsiSimulator, SimulatorConfig}; + + fn simulator(device_id: u64) -> MediatekCsiSimulator { + MediatekCsiSimulator::new(SimulatorConfig { + device_id, + ..Default::default() + }) + .expect("valid simulator config") + } + + /// Two device_ids arriving interleaved (unit1, unit2, unit1-again) must + /// both be present in `/devices`, and the shared single-slot + /// `latest_mediatek_csi` must equal whichever arrived most recently. + #[tokio::test] + async fn two_devices_interleaved_both_present_latest_is_most_recent() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + + let mut unit1 = simulator(0xa9be_7e5b_b164_4d3a); + let mut unit2 = simulator(0x1122_3344_5566_7788); + + let unit1_frame1 = MediatekCsiSnapshot::from_frame(&unit1.next_frame()); + let unit2_frame1 = MediatekCsiSnapshot::from_frame(&unit2.next_frame()); + let unit1_frame2 = MediatekCsiSnapshot::from_frame(&unit1.next_frame()); + assert_ne!(unit1_frame1.device_id, unit2_frame1.device_id); + assert_eq!(unit1_frame1.device_id, unit1_frame2.device_id); + assert_eq!(unit1_frame2.sequence, unit1_frame1.sequence + 1); + + // Ingest in interleaved order, mirroring the UDP loop in main.rs: + // insert into the per-device map AND overwrite the single slot. + for snapshot in [ + unit1_frame1.clone(), + unit2_frame1.clone(), + unit1_frame2.clone(), + ] { + let mut s = state.write().await; + let now = Instant::now(); + s.mediatek_csi_by_device + .insert(snapshot.device_id.clone(), (snapshot.clone(), now)); + s.latest_mediatek_csi = Some(snapshot); + } + + let Json(value) = mediatek_csi_devices(State(state.clone())).await; + let devices = value["devices"].as_array().expect("devices array"); + assert_eq!( + value["total"], 2, + "both devices must be present, not overwritten" + ); + assert_eq!(devices.len(), 2); + + let ids: Vec<&str> = devices + .iter() + .map(|d| d["device_id"].as_str().unwrap()) + .collect(); + let mut expected_ids = vec![ + unit1_frame1.device_id.as_str(), + unit2_frame1.device_id.as_str(), + ]; + expected_ids.sort(); + assert_eq!(ids, expected_ids, "sorted by device_id"); + + let unit1_entry = devices + .iter() + .find(|d| d["device_id"] == unit1_frame1.device_id.as_str()) + .expect("unit1 present"); + assert_eq!( + unit1_entry["sequence"], unit1_frame2.sequence, + "keeps unit1's most recent frame" + ); + assert!(unit1_entry["age_ms"].as_u64().is_some()); + + let s = state.read().await; + assert_eq!( + s.latest_mediatek_csi + .as_ref() + .map(|snap| snap.device_id.clone()), + Some(unit1_frame1.device_id.clone()), + "single latest slot still reflects whichever device arrived last, unchanged behavior" + ); + assert_eq!( + s.latest_mediatek_csi.as_ref().map(|snap| snap.sequence), + Some(unit1_frame2.sequence) + ); + } + + /// A device older than the staleness window is dropped from `/devices` + /// even though its entry is still in the map. + #[tokio::test] + async fn stale_device_is_excluded_from_response() { + let state: SharedState = Arc::new(RwLock::new(AppStateInner::minimal())); + let mut unit = simulator(0xdead_beef_0000_0001); + let frame = MediatekCsiSnapshot::from_frame(&unit.next_frame()); + + { + let mut s = state.write().await; + let stale_seen = Instant::now() - Duration::from_secs(31); + s.mediatek_csi_by_device + .insert(frame.device_id.clone(), (frame, stale_seen)); + } + + let Json(value) = mediatek_csi_devices(State(state)).await; + assert_eq!(value["total"], 0); + assert!(value["devices"].as_array().unwrap().is_empty()); + } +}