feat(sensing-server): per-device MediaTek CSI ingest and inspection

The server kept one MediaTek snapshot in total, so two MT7981 receivers
overwrote each other, and `--source mediatek` fell through to the
catch-all source plan and bound no UDP socket at all.

- `--source mediatek` binds the UDP receiver and does not start the
  ESP32-shaped simulator, which has no MediaTek branch and would otherwise
  publish a simulated room classification next to real frames.
- Snapshots are kept per `device_id`.
  `GET /api/v1/csi/mediatek/devices` lists them with `age_ms` and drops
  any device silent for over 30 s.
- A bounded per-device ring (`--csi-ring`, default 256 frames, about
  0.5 MiB for a 2x2 device at 64 subcarriers) keeps per-chain,
  per-subcarrier amplitude and phase. It serves
  `/devices/:device_id/frames` and `/devices/:device_id/summary`. Chains
  are keyed by (tx, rx) identity, so a device that interleaves 1x2 legacy
  and 2x2 HT frames is reported correctly.
- New provenance label `mediatek:physical-unvalidated` for real-silicon
  frames that are not calibrated. `SYNTHETIC` still takes precedence over
  every other flag. Plain `mediatek` stays reserved for calibrated frames.

No classifier changes in this commit.

Developed in whitsentry, a RuView-based project.

Co-Authored-By: claude-flow <ruv@ruv.net>

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<mediatek_csi::MediatekCsiSnapshot>,
     /// Instant of the last validated MediaTek CSI UDP frame.
     last_mediatek_frame: Option<std::time::Instant>,
+    /// 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<String, (mediatek_csi::MediatekCsiSnapshot, std::time::Instant)>,
+    /// 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<String, mediatek_csi_ring::DeviceRing>,
+    /// 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<qualcomm_csi::QualcommCsiSnapshot>,
     /// 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))

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();

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<i8> 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<f32>,
+    /// 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<f32>,
+}
+
+/// 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<i8>,
+    pub(crate) chains: Vec<ChainSamples>,
+}
+
+/// Bounded per-device ring buffer of [`RingFrame`]s, oldest first.
+#[derive(Debug, Clone)]
+pub(crate) struct DeviceRing {
+    capacity: usize,
+    frames: std::collections::VecDeque<RingFrame>,
+    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<Item = &RingFrame> {
+        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<ChainAmplitudeStats> {
+        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<f32> = sum.iter().map(|&s| (s / n_f) as f32).collect();
+                let std: Vec<f32> = 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<f32>,
+    pub(crate) amplitude_std: Vec<f32>,
+    /// 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<Item = &'a RingFrame>) -> 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<Item = (f32, f32)>,
+    tx: usize,
+    rx: usize,
+    sc: usize,
+    scale: f32,
+) -> Vec<ChainSamples> {
+    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<usize>,
+    /// Comma-separated subset of `"amp"`, `"phase"`. Both when omitted.
+    fields: Option<String>,
+}
+
+fn wants(fields: &Option<String>, 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<serde_json::Value>) {
+    (
+        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<SharedState>,
+    Path(device_id): Path<String>,
+    Query(query): Query<CsiFramesQuery>,
+) -> 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<serde_json::Value> = window
+        .iter()
+        .map(|f| {
+            let frame_chains: Vec<serde_json::Value> = 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::<Vec<_>>()),
+                    );
+                }
+                if include_phase {
+                    map.insert(
+                        "phase".to_string(),
+                        serde_json::json!(f.chains.iter().map(|c| &c.phase).collect::<Vec<_>>()),
+                    );
+                }
+            }
+            obj
+        })
+        .collect();
+
+    let chains: Vec<serde_json::Value> = 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<SharedState>,
+    Path(device_id): Path<String>,
+) -> 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<serde_json::Value> = 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<serde_json::Value> = 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<u32> = (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<u32> = 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<usize> = 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);
+        }
+    }
+}

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<SharedState>,
+) -> Json<serde_json::Value> {
+    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<serde_json::Value> = 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());
+    }
+}