feat: implement ADR-288/289/290 — benchmark harness, wideband CSI ingest, vitals ground-truth rig

ADR-288 (wifi-densepose-train): Widar3.0 Intel-5300 .dat bfee parser
(bounded, panic-free), WidarDataset over the CsiDataset trait, deterministic
SplitProtocol (cross-subject/environment/orientation + leakage-prone random
baseline), LeakageAudit that Errs on subject/environment/recording overlap,
train-only MeanPoseBaseline, and EvidenceGrade where Measured requires an
embedded reproducer. Criterion bench for parser + split + audit.

ADR-289 (wifi-densepose-mat): FeitCSI binary record parser (layout verified
against upstream source) with dimension-vs-buffer validation and bounded
allocation, DeviceType::FeitCsi replay/stream modes, subcarrier-agnostic
frame metadata (bandwidth/band/native->pipeline mapping). Criterion bench at
1992-subcarrier frames.

ADR-290 (wifi-densepose-vitals): reference-series CSV ingest, cross-correlation
time alignment with optional drift fit, Bland-Altman/MAE/RMSE agreement with
mandatory SessionScope, and EvidenceGrade gating Measured on reference device +
coverage + reproducer. Criterion bench for hour-scale alignment.

All three crate suites green; benches compile. Numbers are SYNTHETIC (in-code
fixtures); no hardware claims.

Co-Authored-By: claude-flow <ruv@ruv.net>
Claude-Session: https://claude.ai/code/session_015TcKegTS7QqhWPC2L2SzaS
This commit is contained in:
Claude
2026-08-11 00:06:39 +00:00
parent 01c42d0900
commit 79d1fff99a
18 changed files with 6215 additions and 3 deletions
+5
View File
@@ -101,6 +101,11 @@ approx = "0.5"
name = "detection_bench"
harness = false
# FeitCSI record parse throughput at wideband 802.11ax shapes (ADR-289).
[[bench]]
name = "feitcsi_bench"
harness = false
[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs"]
@@ -0,0 +1,69 @@
//! Criterion benchmark for FeitCSI record parse throughput (ADR-289).
//!
//! Measures `parse_record` over synthetic in-code fixtures at the wideband
//! 802.11ax shapes: 20 MHz (242 tones), 80 MHz (996) and the headline
//! 160 MHz / 1992-subcarrier frames an AX210 delivers. Fixtures are
//! deterministic; no wall-clock or randomness feeds the parsed bytes.
use criterion::{black_box, criterion_group, criterion_main, Criterion, Throughput};
use wifi_densepose_mat::integration::feitcsi::{parse_record, synth, FeitCsiStreamReader};
fn bench_parse(c: &mut Criterion) {
let mut group = c.benchmark_group("feitcsi_parse");
// (label, chan_width_val, HE tone count)
let shapes = [
("he20_242sc", 0u32, 242u32),
("he80_996sc", 2u32, 996u32),
("he160_1992sc", 3u32, 1992u32),
];
for (label, cw, sc) in shapes {
// 2x1 MIMO, HE modulation, fixed timestamp: deterministic bytes.
let bytes = synth::record_bytes(2, 1, sc, cw, 4, 1_000_000);
group.throughput(Throughput::Bytes(bytes.len() as u64));
group.bench_function(label, |b| {
b.iter(|| {
let (record, consumed) =
parse_record(black_box(&bytes)).expect("valid synthetic record");
black_box((record.csi.len(), consumed))
})
});
}
group.finish();
}
/// Stream-reader throughput over an in-memory multi-record capture at the
/// headline 160 MHz / 1992-subcarrier shape. Exercises the reusable payload
/// scratch buffer in `FeitCsiStreamReader` (one raw-byte allocation per
/// stream, not per record).
fn bench_stream(c: &mut Criterion) {
let mut group = c.benchmark_group("feitcsi_stream");
const RECORDS: u64 = 16;
let mut capture = Vec::new();
for ts in 0..RECORDS {
// 2x1 MIMO, 160 MHz HE, deterministic device timestamps.
capture.extend_from_slice(&synth::record_bytes(2, 1, 1992, 3, 4, ts * 1_000));
}
group.throughput(Throughput::Bytes(capture.len() as u64));
group.bench_function("he160_1992sc_x16_records", |b| {
b.iter(|| {
let mut reader = FeitCsiStreamReader::new(std::io::Cursor::new(black_box(&capture[..])));
let mut records = 0u64;
while let Some(rec) = reader.read_next().expect("valid synthetic capture") {
black_box(rec.csi.len());
records += 1;
}
assert_eq!(records, RECORDS);
black_box(records)
})
});
group.finish();
}
criterion_group!(benches, bench_parse, bench_stream);
criterion_main!(benches);
@@ -1293,6 +1293,9 @@ impl From<CsiPacket> for CsiReadings {
rssi: Some(packet.rssi as f64),
noise_floor: Some(packet.noise_floor as f64),
fc_type: FrameControlType::Data,
// Narrowband receiver formats predate wideband provenance
// metadata; FeitCSI ingest attaches it in feitcsi.rs.
wideband: None,
},
}
}
@@ -0,0 +1,983 @@
//! Validated parser for FeitCSI binary CSI records (ADR-289).
//!
//! [FeitCSI](https://feitcsi.kuskosoft.com) is an open-source tool
//! (<https://github.com/KuskoSoft/FeitCSI>) that extracts 802.11ax channel
//! state information from Intel AX200/AX210 NICs at 20/40/80/160 MHz,
//! including the 6 GHz band. FeitCSI is GPL and is used strictly as an
//! *external* capture tool: RuView never links it, never configures the NIC,
//! and only parses the record files/streams its tooling produces.
//!
//! # Record layout (verified against FeitCSI source, `master` @ 2026-08-10)
//!
//! Each record is a packed 272-byte header followed by `csi_data_size` bytes
//! of raw CSI. Layout per `include/Csi.h` (`struct __attribute__((__packed__))
//! RawHeaderData`) and `Csi::save()` in `src/Csi.cpp`, which writes the raw
//! struct memory followed by the CSI buffer. FeitCSI runs on little-endian
//! x86 hosts and dumps native struct memory, so all fields are little-endian.
//!
//! | Offset | Size | Field | Notes |
//! |-------:|-----:|------------------|-----------------------------------------|
//! | 0 | 4 | `csiDataSize` | u32, bytes of CSI payload after header |
//! | 4 | 4 | reserved | (`space4`) |
//! | 8 | 4 | `ftmClock` | u32 |
//! | 12 | 8 | `timestamp` | u64, device timestamp (microseconds) |
//! | 20 | 26 | reserved | (`space20`) |
//! | 46 | 1 | `numRx` | u8, receive antennas |
//! | 47 | 1 | `numTx` | u8, transmit streams |
//! | 48 | 4 | reserved | (`space48`) |
//! | 52 | 4 | `numSubCarriers` | u32 |
//! | 56 | 4 | reserved | (`space54`; upstream field name lags |
//! | | | | the actual packed offset) |
//! | 60 | 4 | `rssi1` | u32, antenna A RSSI |
//! | 64 | 4 | `rssi2` | u32, antenna B RSSI |
//! | 68 | 6 | `srcMac` | source MAC address |
//! | 74 | 18 | reserved | (`space75`) |
//! | 92 | 4 | `rateNflag` | u32, iwlwifi rate flags (see below) |
//! | 96 | 176 | reserved | (`space96`, 44 × u32) |
//!
//! CSI payload: interleaved little-endian `i16` I/Q pairs, iterated
//! `for rx { for tx { for subcarrier { i16 real, i16 imag } } }` (per the
//! processing loops in `src/Csi.cpp`), i.e. 4 bytes per complex sample and
//! `csiDataSize == numRx * numTx * numSubCarriers * 4`.
//!
//! `rateNflag` uses the iwlwifi rate/flags encoding vendored by FeitCSI in
//! `lib/include/rs.h`: modulation type in bits 8..11 (`RATE_MCS_MOD_TYPE`,
//! 0=CCK, 1=legacy OFDM, 2=HT, 3=VHT, 4=HE, 5=EHT) and channel width in bits
//! 11..14 (`RATE_MCS_CHAN_WIDTH`, 0=20 MHz, 1=40, 2=80, 3=160, 4=320).
//!
//! # Failing loudly on format drift
//!
//! The on-disk format carries **no magic number or version field** (it is the
//! raw iwlwifi notification header), so the "version check" required by
//! ADR-289 is structural and strict:
//!
//! - the declared dimensions must agree exactly with the declared buffer
//! length ([`FeitCsiError::DimensionMismatch`]);
//! - dimensions are hard-capped ([`MAX_SUBCARRIERS`], [`MAX_ANTENNAS`]) so a
//! corrupt length can never cause unbounded allocation
//! ([`FeitCsiError::CapExceeded`]);
//! - `rateNflag` values outside the vendored `rs.h` encoding (unknown
//! modulation type, or a channel width this parser does not support, e.g.
//! 320 MHz EHT) are rejected as [`FeitCsiError::UnsupportedFormat`] instead
//! of being misparsed.
//!
//! All input is untrusted: every read is length-checked, allocation is
//! bounded before it happens, and malformed input yields structured errors,
//! never a panic.
use super::hardware_adapter::{
Bandwidth, CsiMetadata, CsiReadings, DeviceType, FrameControlType, SensorCsiReading,
SubcarrierMapping, WidebandMeta, WifiBand,
};
use super::AdapterError;
use chrono::{DateTime, Utc};
use num_complex::Complex64;
use std::io::Read;
/// Size of the packed FeitCSI record header in bytes.
pub const HEADER_LEN: usize = 272;
/// Hard cap on the declared subcarrier count (802.11ax 160 MHz HE is 1992;
/// 4096 leaves headroom for future 802.11bf truncated-CIR shapes without
/// permitting unbounded allocation from a corrupt length field).
pub const MAX_SUBCARRIERS: u32 = 4096;
/// Hard cap on declared antenna/stream counts (AX210 is 2x2; 8 is generous).
pub const MAX_ANTENNAS: u8 = 8;
/// Bytes per complex CSI sample (i16 real + i16 imag).
const BYTES_PER_SAMPLE: usize = 4;
/// Upper bound on a single record's total size (header + max payload).
/// Used to bound stream-mode buffering.
pub const MAX_RECORD_BYTES: usize = HEADER_LEN
+ MAX_ANTENNAS as usize * MAX_ANTENNAS as usize * MAX_SUBCARRIERS as usize * BYTES_PER_SAMPLE;
// iwlwifi rate flag encoding, per FeitCSI `lib/include/rs.h`.
const RATE_MCS_MOD_TYPE_POS: u32 = 8;
const RATE_MCS_MOD_TYPE_MSK: u32 = 0x7 << RATE_MCS_MOD_TYPE_POS;
const RATE_MCS_CHAN_WIDTH_POS: u32 = 11;
const RATE_MCS_CHAN_WIDTH_MSK: u32 = 0x7 << RATE_MCS_CHAN_WIDTH_POS;
/// Structured errors for FeitCSI record parsing. Malformed input always
/// yields one of these — the parser never panics on untrusted bytes.
#[derive(Debug, thiserror::Error)]
pub enum FeitCsiError {
/// The buffer ends before the declared record does.
#[error("truncated FeitCSI record: need {needed} bytes, got {got}")]
Truncated {
/// Bytes required to complete the header or record.
needed: usize,
/// Bytes actually available.
got: usize,
},
/// The declared CSI buffer length disagrees with the declared dimensions.
#[error(
"FeitCSI dimension mismatch: header declares csi_data_size={declared} \
but num_rx={num_rx} * num_tx={num_tx} * num_subcarriers={num_subcarriers} \
* 4 = {expected}"
)]
DimensionMismatch {
/// `csiDataSize` from the header.
declared: u32,
/// Size implied by the dimension fields.
expected: usize,
/// Declared receive antenna count.
num_rx: u8,
/// Declared transmit stream count.
num_tx: u8,
/// Declared subcarrier count.
num_subcarriers: u32,
},
/// A dimension field is zero — the record cannot contain CSI.
#[error("FeitCSI record declares zero-sized dimension: {field}")]
ZeroDimension {
/// Which header field was zero.
field: &'static str,
},
/// A dimension exceeds its hard cap; parsing stops before any
/// allocation sized from the corrupt value.
#[error("FeitCSI {field}={value} exceeds hard cap {cap} (corrupt or hostile length)")]
CapExceeded {
/// Which header field exceeded its cap.
field: &'static str,
/// The declared value.
value: u64,
/// The enforced cap.
cap: u64,
},
/// `rateNflag` encodes a modulation type or channel width outside the
/// layout this parser was written against — fail loudly instead of
/// misparsing a newer/unknown format revision.
#[error(
"unsupported FeitCSI rate flags {rate_n_flags:#010x}: {reason} \
(layout per FeitCSI master @ 2026-08-10; refusing to guess)"
)]
UnsupportedFormat {
/// Raw `rateNflag` value.
rate_n_flags: u32,
/// Which sub-field was unrecognized.
reason: &'static str,
},
/// I/O error while reading a record from a file or stream.
#[error("FeitCSI I/O error: {0}")]
Io(#[from] std::io::Error),
}
impl From<FeitCsiError> for AdapterError {
fn from(e: FeitCsiError) -> Self {
match e {
FeitCsiError::UnsupportedFormat { .. } => AdapterError::UnsupportedAdapter(e.to_string()),
FeitCsiError::Io(io) => AdapterError::Io(io),
_ => AdapterError::DataFormat(e.to_string()),
}
}
}
/// Modulation type decoded from `rateNflag` (iwlwifi `RATE_MCS_MOD_TYPE`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FeitCsiModType {
/// Legacy CCK (802.11b)
Cck,
/// Legacy OFDM (802.11a/g)
LegacyOfdm,
/// HT (802.11n)
Ht,
/// VHT (802.11ac)
Vht,
/// HE (802.11ax)
He,
/// EHT (802.11be)
Eht,
}
/// Channel bandwidth decoded from `rateNflag` (iwlwifi `RATE_MCS_CHAN_WIDTH`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FeitCsiBandwidth {
/// 20 MHz
Bw20,
/// 40 MHz
Bw40,
/// 80 MHz
Bw80,
/// 160 MHz
Bw160,
}
impl FeitCsiBandwidth {
/// Bandwidth in MHz.
pub fn mhz(&self) -> u16 {
match self {
Self::Bw20 => 20,
Self::Bw40 => 40,
Self::Bw80 => 80,
Self::Bw160 => 160,
}
}
/// Map to the adapter-level [`Bandwidth`] enum.
pub fn to_bandwidth(&self) -> Bandwidth {
match self {
Self::Bw20 => Bandwidth::HT20,
Self::Bw40 => Bandwidth::HT40,
Self::Bw80 => Bandwidth::VHT80,
Self::Bw160 => Bandwidth::VHT160,
}
}
}
/// Validated header fields of one FeitCSI record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FeitCsiHeader {
/// Declared CSI payload size in bytes (already validated against dims).
pub csi_data_size: u32,
/// FTM clock value.
pub ftm_clock: u32,
/// Device timestamp (microseconds, per upstream usage).
pub timestamp_us: u64,
/// Number of receive antennas.
pub num_rx: u8,
/// Number of transmit streams.
pub num_tx: u8,
/// Native subcarrier count of this frame.
pub num_subcarriers: u32,
/// Antenna A RSSI (raw u32 as stored on disk).
pub rssi1: u32,
/// Antenna B RSSI (raw u32 as stored on disk).
pub rssi2: u32,
/// Source MAC address.
pub source_mac: [u8; 6],
/// Raw iwlwifi rate flags.
pub rate_n_flags: u32,
/// Modulation type decoded from `rate_n_flags`.
pub mod_type: FeitCsiModType,
/// Channel bandwidth decoded from `rate_n_flags`.
pub bandwidth: FeitCsiBandwidth,
}
/// One parsed FeitCSI record: validated header plus complex CSI, kept at
/// native dimensionality (`num_rx * num_tx * num_subcarriers` samples,
/// iterated rx-major, then tx, then subcarrier).
#[derive(Debug, Clone, PartialEq)]
pub struct FeitCsiRecord {
/// Validated header.
pub header: FeitCsiHeader,
/// Complex CSI samples, flattened `[rx][tx][subcarrier]`.
pub csi: Vec<Complex64>,
}
impl FeitCsiRecord {
/// CSI for one (rx, tx) antenna pair as a subcarrier slice, or `None`
/// when the indices are out of range.
pub fn antenna_pair(&self, rx: u8, tx: u8) -> Option<&[Complex64]> {
if rx >= self.header.num_rx || tx >= self.header.num_tx {
return None;
}
let sc = self.header.num_subcarriers as usize;
let start = (rx as usize * self.header.num_tx as usize + tx as usize) * sc;
self.csi.get(start..start + sc)
}
/// Convert to adapter-level [`CsiReadings`], one [`SensorCsiReading`] per
/// (rx, tx) antenna pair, carrying native subcarrier count, bandwidth and
/// band as first-class frame metadata (ADR-289).
///
/// The FeitCSI header does not record channel/band (the capture
/// configuration owns that), so both are supplied by the caller. The
/// timestamp is derived deterministically from the record's own device
/// timestamp, never from wall-clock, so file replay is reproducible.
pub fn to_readings(&self, band: WifiBand, channel: u8) -> CsiReadings {
let sc = self.header.num_subcarriers as usize;
let mut readings = Vec::with_capacity(self.header.num_rx as usize * self.header.num_tx as usize);
// Interpret the on-disk u32 RSSI as two's-complement dBm (captures
// store negative dBm values in the raw register field).
let rssi_dbm = |raw: u32| raw as i32 as f64;
let rssi = rssi_dbm(self.header.rssi1).max(rssi_dbm(self.header.rssi2));
let mac = self.header.source_mac;
let tx_mac = format!(
"{:02X}:{:02X}:{:02X}:{:02X}:{:02X}:{:02X}",
mac[0], mac[1], mac[2], mac[3], mac[4], mac[5]
);
for rx in 0..self.header.num_rx {
for tx in 0..self.header.num_tx {
let pair = self
.antenna_pair(rx, tx)
.expect("indices bounded by validated header dims");
let mut amplitudes = Vec::with_capacity(sc);
let mut phases = Vec::with_capacity(sc);
for c in pair {
amplitudes.push(c.norm());
phases.push(c.im.atan2(c.re));
}
readings.push(SensorCsiReading {
sensor_id: format!("feitcsi_rx{rx}_tx{tx}"),
amplitudes,
phases,
rssi,
noise_floor: -92.0,
tx_mac: Some(tx_mac.clone()),
rx_mac: None,
sequence_num: None,
});
}
}
// Deterministic timestamp from the device clock (microseconds since
// capture epoch); replay of the same bytes yields the same output.
let timestamp = DateTime::<Utc>::from_timestamp_micros(self.header.timestamp_us as i64)
.unwrap_or_else(|| DateTime::<Utc>::from_timestamp(0, 0).expect("epoch is valid"));
CsiReadings {
timestamp,
readings,
metadata: CsiMetadata {
device_type: DeviceType::FeitCsi,
channel,
bandwidth: self.header.bandwidth.to_bandwidth(),
num_subcarriers: sc,
rssi: Some(rssi),
noise_floor: None,
fc_type: FrameControlType::Data,
wideband: Some(WidebandMeta {
band,
bandwidth_mhz: self.header.bandwidth.mhz(),
native_subcarriers: sc,
mapping: None,
}),
},
}
}
}
/// Validate the fixed-size header. Enforces caps and dimension/length
/// consistency BEFORE any allocation is sized from untrusted fields.
fn validate_header(h: &[u8; HEADER_LEN]) -> Result<FeitCsiHeader, FeitCsiError> {
let u32_at = |off: usize| u32::from_le_bytes([h[off], h[off + 1], h[off + 2], h[off + 3]]);
let csi_data_size = u32_at(0);
let ftm_clock = u32_at(8);
let timestamp_us = u64::from_le_bytes([
h[12], h[13], h[14], h[15], h[16], h[17], h[18], h[19],
]);
let num_rx = h[46];
let num_tx = h[47];
let num_subcarriers = u32_at(52);
let rssi1 = u32_at(60);
let rssi2 = u32_at(64);
let mut source_mac = [0u8; 6];
source_mac.copy_from_slice(&h[68..74]);
let rate_n_flags = u32_at(92);
// Zero dimensions cannot carry CSI.
if num_rx == 0 {
return Err(FeitCsiError::ZeroDimension { field: "num_rx" });
}
if num_tx == 0 {
return Err(FeitCsiError::ZeroDimension { field: "num_tx" });
}
if num_subcarriers == 0 {
return Err(FeitCsiError::ZeroDimension {
field: "num_subcarriers",
});
}
// Hard caps: corrupt lengths must not size any allocation.
if num_rx > MAX_ANTENNAS {
return Err(FeitCsiError::CapExceeded {
field: "num_rx",
value: num_rx as u64,
cap: MAX_ANTENNAS as u64,
});
}
if num_tx > MAX_ANTENNAS {
return Err(FeitCsiError::CapExceeded {
field: "num_tx",
value: num_tx as u64,
cap: MAX_ANTENNAS as u64,
});
}
if num_subcarriers > MAX_SUBCARRIERS {
return Err(FeitCsiError::CapExceeded {
field: "num_subcarriers",
value: num_subcarriers as u64,
cap: MAX_SUBCARRIERS as u64,
});
}
// Dimensions vs declared buffer length. Capped dims bound this product
// at 8 * 8 * 4096 * 4 = 1 MiB, so the arithmetic cannot overflow usize.
let expected =
num_rx as usize * num_tx as usize * num_subcarriers as usize * BYTES_PER_SAMPLE;
if csi_data_size as usize != expected {
return Err(FeitCsiError::DimensionMismatch {
declared: csi_data_size,
expected,
num_rx,
num_tx,
num_subcarriers,
});
}
// Format-revision check on the rate flags: reject encodings outside the
// vendored rs.h layout this parser was written against.
let mod_type = match (rate_n_flags & RATE_MCS_MOD_TYPE_MSK) >> RATE_MCS_MOD_TYPE_POS {
0 => FeitCsiModType::Cck,
1 => FeitCsiModType::LegacyOfdm,
2 => FeitCsiModType::Ht,
3 => FeitCsiModType::Vht,
4 => FeitCsiModType::He,
5 => FeitCsiModType::Eht,
_ => {
return Err(FeitCsiError::UnsupportedFormat {
rate_n_flags,
reason: "unknown modulation type (bits 8..11)",
})
}
};
let bandwidth = match (rate_n_flags & RATE_MCS_CHAN_WIDTH_MSK) >> RATE_MCS_CHAN_WIDTH_POS {
0 => FeitCsiBandwidth::Bw20,
1 => FeitCsiBandwidth::Bw40,
2 => FeitCsiBandwidth::Bw80,
3 => FeitCsiBandwidth::Bw160,
// 4 = 320 MHz (EHT); not supported by this ingest revision.
_ => {
return Err(FeitCsiError::UnsupportedFormat {
rate_n_flags,
reason: "unsupported channel width (bits 11..14; 320 MHz+ not supported)",
})
}
};
Ok(FeitCsiHeader {
csi_data_size,
ftm_clock,
timestamp_us,
num_rx,
num_tx,
num_subcarriers,
rssi1,
rssi2,
source_mac,
rate_n_flags,
mod_type,
bandwidth,
})
}
/// Decode a validated CSI payload (interleaved little-endian i16 I/Q pairs)
/// into complex samples. The caller has already validated `payload.len()`
/// against the header dimensions, so this performs exactly one bounded
/// allocation (`chunks_exact` is an exact-size iterator, so `collect`
/// reserves the final length up front) and the conversion loop itself is
/// allocation-free.
#[inline]
fn decode_csi(payload: &[u8]) -> Vec<Complex64> {
payload
.chunks_exact(BYTES_PER_SAMPLE)
.map(|sample| {
let re = i16::from_le_bytes([sample[0], sample[1]]) as f64;
let im = i16::from_le_bytes([sample[2], sample[3]]) as f64;
Complex64::new(re, im)
})
.collect()
}
/// Parse one record from the front of `buf`.
///
/// Returns the record and the number of bytes consumed, so callers can walk
/// a multi-record capture. All validation happens before any allocation is
/// sized from untrusted fields; malformed input yields a structured
/// [`FeitCsiError`], never a panic. The header is read in place (no copy)
/// and the payload is converted directly from the input slice, so the CSI
/// bytes are traversed exactly once.
pub fn parse_record(buf: &[u8]) -> Result<(FeitCsiRecord, usize), FeitCsiError> {
if buf.len() < HEADER_LEN {
return Err(FeitCsiError::Truncated {
needed: HEADER_LEN,
got: buf.len(),
});
}
let header_bytes: &[u8; HEADER_LEN] = buf[..HEADER_LEN]
.try_into()
.expect("slice length checked above");
let header = validate_header(header_bytes)?;
let payload_len = header.csi_data_size as usize;
let total = HEADER_LEN + payload_len;
if buf.len() < total {
return Err(FeitCsiError::Truncated {
needed: total,
got: buf.len(),
});
}
// payload_len == validated expected size <= 1 MiB: bounded allocation.
let csi = decode_csi(&buf[HEADER_LEN..total]);
Ok((FeitCsiRecord { header, csi }, total))
}
/// Read one record from a byte stream (file, pipe, socket wrapper).
///
/// Returns `Ok(None)` on clean EOF (no bytes before end-of-stream); a
/// mid-record EOF is a [`FeitCsiError::Truncated`] error. `read_exact`
/// semantics mean a blocking pipe simply waits for the writer, so the same
/// code path serves file replay and streaming mode.
pub fn read_one_record<R: Read>(
reader: &mut R,
) -> Result<Option<(FeitCsiRecord, usize)>, FeitCsiError> {
let mut scratch = Vec::new();
read_one_record_with_scratch(reader, &mut scratch)
}
/// [`read_one_record`] with a caller-owned scratch buffer for the raw
/// payload, so long-running replay/stream loops reuse one allocation across
/// records instead of allocating per record. The scratch is only ever
/// resized to the header-validated payload length (<= 1 MiB), never to an
/// untrusted value.
fn read_one_record_with_scratch<R: Read>(
reader: &mut R,
scratch: &mut Vec<u8>,
) -> Result<Option<(FeitCsiRecord, usize)>, FeitCsiError> {
let mut header_bytes = [0u8; HEADER_LEN];
let mut filled = 0usize;
while filled < HEADER_LEN {
let n = reader.read(&mut header_bytes[filled..])?;
if n == 0 {
if filled == 0 {
return Ok(None); // clean EOF between records
}
return Err(FeitCsiError::Truncated {
needed: HEADER_LEN,
got: filled,
});
}
filled += n;
}
let header = validate_header(&header_bytes)?;
let payload_len = header.csi_data_size as usize; // validated, <= 1 MiB
scratch.resize(payload_len, 0);
reader.read_exact(scratch).map_err(|e| {
if e.kind() == std::io::ErrorKind::UnexpectedEof {
FeitCsiError::Truncated {
needed: HEADER_LEN + payload_len,
got: HEADER_LEN, // header complete, payload short
}
} else {
FeitCsiError::Io(e)
}
})?;
let csi = decode_csi(scratch);
Ok(Some((FeitCsiRecord { header, csi }, HEADER_LEN + payload_len)))
}
/// Streaming reader over any [`Read`] source (recorded capture file, or a
/// path/pipe an external FeitCSI process writes to). RuView never configures
/// the NIC — FeitCSI's own tooling owns capture, per least-authority.
///
/// Holds a reusable payload scratch buffer so a long-running stream performs
/// one bounded raw-byte allocation total (plus the per-record `Vec<Complex64>`
/// output), rather than one raw-byte allocation per record.
pub struct FeitCsiStreamReader<R: Read> {
inner: R,
scratch: Vec<u8>,
}
impl<R: Read> FeitCsiStreamReader<R> {
/// Wrap a byte source.
pub fn new(inner: R) -> Self {
Self {
inner,
scratch: Vec::new(),
}
}
/// Read the next record; `Ok(None)` on clean end-of-stream.
pub fn read_next(&mut self) -> Result<Option<FeitCsiRecord>, FeitCsiError> {
Ok(read_one_record_with_scratch(&mut self.inner, &mut self.scratch)?.map(|(rec, _)| rec))
}
}
/// Deterministic file-replay reader for recorded FeitCSI captures.
pub struct FeitCsiFileReader {
stream: FeitCsiStreamReader<std::io::BufReader<std::fs::File>>,
}
impl FeitCsiFileReader {
/// Open a recorded capture for sequential replay.
pub fn open(path: &str) -> Result<Self, FeitCsiError> {
let file = std::fs::File::open(path)?;
Ok(Self {
stream: FeitCsiStreamReader::new(std::io::BufReader::new(file)),
})
}
/// Read the next record; `Ok(None)` at end of capture.
pub fn read_next(&mut self) -> Result<Option<FeitCsiRecord>, FeitCsiError> {
self.stream.read_next()
}
}
/// Convert wideband readings to the pipeline's subcarrier width via the
/// existing interpolation path (`wifi-densepose-signal`'s Catmull-Rom cubic
/// resampler from ADR-027), recording the native → pipeline mapping in frame
/// metadata so downstream consumers know the true spectral resolution
/// (ADR-289 §3).
///
/// This is the ONLY sanctioned native→pipeline conversion: it is explicit,
/// and the mapping is auditable in `metadata.wideband.mapping`.
pub fn resample_readings_to_pipeline(
readings: &CsiReadings,
pipeline_subcarriers: usize,
) -> Result<CsiReadings, AdapterError> {
let normalizer =
wifi_densepose_signal::HardwareNormalizer::with_canonical_subcarriers(pipeline_subcarriers)
.map_err(|e| AdapterError::Config(format!("invalid pipeline width: {e}")))?;
let native = readings.metadata.num_subcarriers;
let mut out = readings.clone();
for reading in &mut out.readings {
reading.amplitudes = normalizer.resample_to_canonical(&reading.amplitudes);
reading.phases = normalizer.resample_to_canonical(&reading.phases);
}
out.metadata.num_subcarriers = pipeline_subcarriers;
let mapping = SubcarrierMapping {
native,
pipeline: pipeline_subcarriers,
method: "catmull-rom-cubic",
};
match &mut out.metadata.wideband {
Some(wb) => wb.mapping = Some(mapping),
None => {
// Preserve provenance even for frames that arrived without
// wideband metadata: native resolution is still recorded.
out.metadata.wideband = Some(WidebandMeta {
band: WifiBand::Band5GHz,
bandwidth_mhz: readings.metadata.bandwidth.mhz(),
native_subcarriers: native,
mapping: Some(mapping),
});
}
}
Ok(out)
}
/// Deterministic synthetic-fixture generation for tests and benchmarks.
///
/// FeitCSI capture fixtures are always generated in code (never checked in
/// as binary files, per repo policy). CSI samples are a pure function of the
/// sample index, so round-trips are checkable and replay is reproducible.
pub mod synth {
use super::{BYTES_PER_SAMPLE, HEADER_LEN, RATE_MCS_CHAN_WIDTH_POS, RATE_MCS_MOD_TYPE_POS};
/// Build the bytes of one synthetic FeitCSI record with the given
/// dimensions, `rateNflag` channel-width value (0=20 MHz .. 3=160 MHz),
/// modulation-type value (4=HE), and device timestamp.
pub fn record_bytes(
num_rx: u8,
num_tx: u8,
num_subcarriers: u32,
chan_width_val: u32,
mod_type_val: u32,
timestamp_us: u64,
) -> Vec<u8> {
let samples = num_rx as usize * num_tx as usize * num_subcarriers as usize;
let csi_data_size = (samples * BYTES_PER_SAMPLE) as u32;
let mut h = vec![0u8; HEADER_LEN];
h[0..4].copy_from_slice(&csi_data_size.to_le_bytes());
h[8..12].copy_from_slice(&0xAABBCCDDu32.to_le_bytes()); // ftm_clock
h[12..20].copy_from_slice(&timestamp_us.to_le_bytes());
h[46] = num_rx;
h[47] = num_tx;
h[52..56].copy_from_slice(&num_subcarriers.to_le_bytes());
h[60..64].copy_from_slice(&(-42i32 as u32).to_le_bytes()); // rssi1
h[64..68].copy_from_slice(&(-45i32 as u32).to_le_bytes()); // rssi2
h[68..74].copy_from_slice(&[0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]);
let rate_n_flags =
(mod_type_val << RATE_MCS_MOD_TYPE_POS) | (chan_width_val << RATE_MCS_CHAN_WIDTH_POS);
h[92..96].copy_from_slice(&rate_n_flags.to_le_bytes());
for i in 0..samples {
let re = (i as i64 % 200 - 100) as i16;
let im = (i as i64 % 97 - 48) as i16;
h.extend_from_slice(&re.to_le_bytes());
h.extend_from_slice(&im.to_le_bytes());
}
h
}
}
#[cfg(test)]
mod tests {
use super::synth::record_bytes as make_record_bytes;
use super::*;
/// HE (802.11ax) records at 20/80/160 MHz shapes parse with correct
/// native dimensions, decoded bandwidth, and sample round-trip.
#[test]
fn test_parse_valid_he_shapes() {
// (chan_width_val, expected bandwidth, HE tone count)
let shapes = [
(0u32, FeitCsiBandwidth::Bw20, 242u32),
(2u32, FeitCsiBandwidth::Bw80, 996u32),
(3u32, FeitCsiBandwidth::Bw160, 1992u32),
];
for (cw, expected_bw, sc) in shapes {
let bytes = make_record_bytes(2, 1, sc, cw, 4, 1_000_000);
let (rec, consumed) = parse_record(&bytes).expect("valid record must parse");
assert_eq!(consumed, bytes.len());
assert_eq!(rec.header.num_subcarriers, sc);
assert_eq!(rec.header.bandwidth, expected_bw);
assert_eq!(rec.header.mod_type, FeitCsiModType::He);
assert_eq!(rec.header.num_rx, 2);
assert_eq!(rec.header.num_tx, 1);
assert_eq!(rec.csi.len(), 2 * sc as usize);
// Deterministic sample round-trip: index 5 → re = 5-100... check
// the generator formula directly.
let i = 5usize;
assert_eq!(rec.csi[i].re, (i as i64 % 200 - 100) as f64);
assert_eq!(rec.csi[i].im, (i as i64 % 97 - 48) as f64);
// Antenna-pair accessor yields native-width slices.
assert_eq!(rec.antenna_pair(0, 0).unwrap().len(), sc as usize);
assert_eq!(rec.antenna_pair(1, 0).unwrap().len(), sc as usize);
assert!(rec.antenna_pair(2, 0).is_none());
}
}
/// A truncated buffer (header or payload cut short) is a structured
/// error, not a panic.
#[test]
fn test_truncated_buffer() {
let bytes = make_record_bytes(1, 1, 242, 0, 4, 0);
// Header cut short.
let r = parse_record(&bytes[..100]);
assert!(matches!(
r,
Err(FeitCsiError::Truncated { needed, got: 100 }) if needed == HEADER_LEN
));
// Payload cut short.
let r = parse_record(&bytes[..bytes.len() - 1]);
assert!(matches!(r, Err(FeitCsiError::Truncated { .. })));
// Empty buffer.
assert!(matches!(
parse_record(&[]),
Err(FeitCsiError::Truncated { .. })
));
}
/// csiDataSize that disagrees with the declared dimensions is rejected.
#[test]
fn test_dimension_mismatch() {
let mut bytes = make_record_bytes(1, 1, 242, 0, 4, 0);
// Corrupt the declared size (off by 4 bytes).
let bad = (242 * BYTES_PER_SAMPLE as u32) + 4;
bytes[0..4].copy_from_slice(&bad.to_le_bytes());
let r = parse_record(&bytes);
assert!(matches!(
r,
Err(FeitCsiError::DimensionMismatch {
declared,
expected,
..
}) if declared == bad && expected == 242 * BYTES_PER_SAMPLE
));
}
/// Unknown rate-flag encodings fail loudly (the format has no magic, so
/// this is the version/format check): 320 MHz width and out-of-range
/// modulation types are refused rather than misparsed.
#[test]
fn test_unsupported_format_fails_loudly() {
// Channel width value 4 = 320 MHz (EHT) — unsupported.
let bytes = make_record_bytes(1, 1, 242, 4, 5, 0);
assert!(matches!(
parse_record(&bytes),
Err(FeitCsiError::UnsupportedFormat { .. })
));
// Modulation type 7 — outside the vendored rs.h encoding.
let bytes = make_record_bytes(1, 1, 242, 0, 7, 0);
assert!(matches!(
parse_record(&bytes),
Err(FeitCsiError::UnsupportedFormat { .. })
));
}
/// Corrupt dimension fields beyond the hard caps are rejected BEFORE any
/// allocation is sized from them — a hostile length cannot cause
/// unbounded allocation.
#[test]
fn test_allocation_cap_enforcement() {
// Subcarrier count over the cap, with a consistent (huge) size field.
let mut h = vec![0u8; HEADER_LEN];
let huge_sc: u32 = 100_000;
h[46] = 1;
h[47] = 1;
h[52..56].copy_from_slice(&huge_sc.to_le_bytes());
h[0..4].copy_from_slice(&(huge_sc * 4).to_le_bytes());
h[92..96].copy_from_slice(&(4u32 << RATE_MCS_MOD_TYPE_POS).to_le_bytes());
let r = parse_record(&h);
assert!(matches!(
r,
Err(FeitCsiError::CapExceeded {
field: "num_subcarriers",
value,
cap,
}) if value == huge_sc as u64 && cap == MAX_SUBCARRIERS as u64
));
// Antenna count over the cap.
let mut h = vec![0u8; HEADER_LEN];
h[46] = 9; // num_rx > MAX_ANTENNAS
h[47] = 1;
h[52..56].copy_from_slice(&242u32.to_le_bytes());
h[0..4].copy_from_slice(&(9 * 242 * 4u32).to_le_bytes());
assert!(matches!(
parse_record(&h),
Err(FeitCsiError::CapExceeded { field: "num_rx", .. })
));
// Zero dimension.
let mut h = vec![0u8; HEADER_LEN];
h[46] = 0;
h[47] = 1;
h[52..56].copy_from_slice(&242u32.to_le_bytes());
assert!(matches!(
parse_record(&h),
Err(FeitCsiError::ZeroDimension { field: "num_rx" })
));
}
/// Streaming reader over an in-memory multi-record capture: reads all
/// records in order, then clean EOF.
#[test]
fn test_stream_reader_multi_record() {
let mut capture = Vec::new();
for ts in [10u64, 20, 30] {
capture.extend_from_slice(&make_record_bytes(1, 1, 242, 0, 4, ts));
}
let mut reader = FeitCsiStreamReader::new(std::io::Cursor::new(capture));
let mut timestamps = Vec::new();
while let Some(rec) = reader.read_next().expect("stream parse") {
timestamps.push(rec.header.timestamp_us);
}
assert_eq!(timestamps, vec![10, 20, 30]);
}
/// A stream that ends mid-record reports Truncated, not clean EOF.
#[test]
fn test_stream_reader_mid_record_eof() {
let bytes = make_record_bytes(1, 1, 242, 0, 4, 0);
let cut = &bytes[..bytes.len() - 10];
let mut reader = FeitCsiStreamReader::new(std::io::Cursor::new(cut.to_vec()));
assert!(matches!(
reader.read_next(),
Err(FeitCsiError::Truncated { .. })
));
}
/// File replay is deterministic: two independent reads of the same
/// synthetic capture yield byte-identical record sequences and identical
/// converted readings (timestamps derive from the record, not wall-clock).
#[test]
fn test_replay_determinism() {
let mut capture = Vec::new();
for ts in [1_000u64, 2_000, 3_000] {
capture.extend_from_slice(&make_record_bytes(2, 1, 996, 2, 4, ts));
}
let path = std::env::temp_dir().join(format!(
"feitcsi_replay_test_{}.dat",
std::process::id()
));
std::fs::write(&path, &capture).unwrap();
let read_all = || -> Vec<FeitCsiRecord> {
let mut reader = FeitCsiFileReader::open(path.to_str().unwrap()).unwrap();
let mut out = Vec::new();
while let Some(rec) = reader.read_next().unwrap() {
out.push(rec);
}
out
};
let first = read_all();
let second = read_all();
assert_eq!(first.len(), 3);
assert_eq!(first, second, "replay must be deterministic");
// Converted readings are also identical, including timestamps.
let r1 = first[0].to_readings(WifiBand::Band6GHz, 37);
let r2 = second[0].to_readings(WifiBand::Band6GHz, 37);
assert_eq!(r1.timestamp, r2.timestamp);
assert_eq!(r1.readings[0].amplitudes, r2.readings[0].amplitudes);
let _ = std::fs::remove_file(&path);
}
/// Native → pipeline conversion goes through the explicit interpolation
/// path and records the mapping in frame metadata.
#[test]
fn test_native_to_pipeline_mapping_recorded() {
let bytes = make_record_bytes(1, 1, 1992, 3, 4, 500);
let (rec, _) = parse_record(&bytes).unwrap();
let native = rec.to_readings(WifiBand::Band6GHz, 37);
// Native metadata is first-class.
assert_eq!(native.metadata.num_subcarriers, 1992);
let wb = native.metadata.wideband.as_ref().expect("wideband meta");
assert_eq!(wb.band, WifiBand::Band6GHz);
assert_eq!(wb.bandwidth_mhz, 160);
assert_eq!(wb.native_subcarriers, 1992);
assert!(wb.mapping.is_none(), "no mapping before conversion");
// Explicit conversion to pipeline width.
let converted = resample_readings_to_pipeline(&native, 56).unwrap();
assert_eq!(converted.metadata.num_subcarriers, 56);
assert_eq!(converted.readings[0].amplitudes.len(), 56);
assert_eq!(converted.readings[0].phases.len(), 56);
let wb = converted.metadata.wideband.as_ref().unwrap();
assert_eq!(wb.native_subcarriers, 1992, "true resolution preserved");
let mapping = wb.mapping.as_ref().expect("mapping recorded");
assert_eq!(mapping.native, 1992);
assert_eq!(mapping.pipeline, 56);
assert_eq!(mapping.method, "catmull-rom-cubic");
// Original frame is untouched.
assert_eq!(native.metadata.num_subcarriers, 1992);
}
/// to_readings emits one reading per (rx, tx) pair at native width.
#[test]
fn test_to_readings_antenna_pairs() {
let bytes = make_record_bytes(2, 2, 242, 0, 4, 0);
let (rec, _) = parse_record(&bytes).unwrap();
let readings = rec.to_readings(WifiBand::Band5GHz, 36);
assert_eq!(readings.readings.len(), 4);
for r in &readings.readings {
assert_eq!(r.amplitudes.len(), 242);
assert_eq!(r.phases.len(), 242);
}
assert_eq!(
readings.readings[0].tx_mac.as_deref(),
Some("AA:BB:CC:DD:EE:FF")
);
assert!(matches!(readings.metadata.device_type, DeviceType::FeitCsi));
assert_eq!(readings.metadata.bandwidth, Bandwidth::HT20);
}
}
@@ -129,6 +129,42 @@ impl HardwareConfig {
}
}
/// Create configuration for deterministic FeitCSI capture replay
/// (wideband 802.11ax records from Intel AX200/AX210, ADR-289).
pub fn feitcsi_replay(file_path: &str) -> Self {
Self::feitcsi(file_path, FeitCsiMode::FileReplay)
}
/// Create configuration for streaming FeitCSI ingest from a path/pipe an
/// external FeitCSI process writes to (no NIC configuration in-crate).
pub fn feitcsi_stream(path: &str) -> Self {
Self::feitcsi(path, FeitCsiMode::Stream)
}
fn feitcsi(path: &str, mode: FeitCsiMode) -> Self {
Self {
device_type: DeviceType::FeitCsi,
device_settings: DeviceSettings::FeitCsi(FeitCsiSettings {
path: path.to_string(),
mode,
band: WifiBand::Band5GHz,
channel: 36,
loop_playback: false,
pipeline_subcarriers: None,
}),
buffer_size: 8192,
raw_mode: false,
sample_rate_override: 0,
channel_config: ChannelConfig {
channel: 36,
bandwidth: Bandwidth::VHT160,
// Native width travels with each frame; this is only the
// configured expectation (802.11ax HE 160 MHz = 1992 tones).
num_subcarriers: 1992,
},
}
}
/// Create configuration for UDP receiver (generic CSI)
pub fn udp_receiver(bind_addr: &str, port: u16) -> Self {
Self {
@@ -160,6 +196,11 @@ pub enum DeviceType {
UdpReceiver,
/// PCAP file replay
PcapFile,
/// FeitCSI wideband 802.11ax records from Intel AX200/AX210 (ADR-289):
/// file replay of a recorded capture, or a path/pipe an external FeitCSI
/// process writes to. RuView never configures the NIC — FeitCSI's own
/// tooling owns capture, per least-authority.
FeitCsi,
/// Simulated device (for testing)
Simulated,
}
@@ -186,10 +227,46 @@ pub enum DeviceSettings {
Udp(UdpSettings),
/// PCAP file settings
Pcap(PcapSettings),
/// FeitCSI capture replay / stream settings
FeitCsi(FeitCsiSettings),
/// Simulated device (no real hardware)
Simulated,
}
/// FeitCSI ingest mode (ADR-289).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FeitCsiMode {
/// Deterministic replay of a recorded capture file.
FileReplay,
/// Live stream read from a path/pipe an external FeitCSI process writes
/// to. RuView performs no NIC configuration; the external tool owns it.
Stream,
}
/// FeitCSI source settings (ADR-289).
///
/// The FeitCSI record header carries bandwidth (via the iwlwifi rate flags)
/// but not channel/band — the capture configuration owns those — so band and
/// channel are supplied here and stamped into frame metadata.
#[derive(Debug, Clone)]
pub struct FeitCsiSettings {
/// Path to the recorded capture (FileReplay) or the file/FIFO the
/// external FeitCSI process appends records to (Stream).
pub path: String,
/// Ingest mode.
pub mode: FeitCsiMode,
/// Radio band the capture was taken on (2.4/5/6 GHz).
pub band: WifiBand,
/// WiFi channel the capture was taken on.
pub channel: u8,
/// Restart from the beginning when file replay reaches the end.
pub loop_playback: bool,
/// When `Some(n)`, frames are explicitly converted from their native
/// subcarrier count to `n` via the interpolation path, and the mapping is
/// recorded in `CsiMetadata::wideband`. `None` keeps native width.
pub pipeline_subcarriers: Option<usize>,
}
/// Serial port configuration
#[derive(Debug, Clone)]
pub struct SerialSettings {
@@ -264,6 +341,55 @@ impl Bandwidth {
Bandwidth::VHT160 => 484,
}
}
/// Channel bandwidth in MHz.
pub fn mhz(&self) -> u16 {
match self {
Bandwidth::HT20 => 20,
Bandwidth::HT40 => 40,
Bandwidth::VHT80 => 80,
Bandwidth::VHT160 => 160,
}
}
}
/// WiFi radio band (first-class frame metadata per ADR-289).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WifiBand {
/// 2.4 GHz ISM band
Band2_4GHz,
/// 5 GHz band
Band5GHz,
/// 6 GHz band (802.11ax/Wi-Fi 6E and later)
Band6GHz,
}
/// Record of an explicit native → pipeline subcarrier conversion, so
/// downstream consumers know the true spectral resolution of a frame and
/// how it was resampled (ADR-289 §3).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SubcarrierMapping {
/// Native subcarrier count as captured.
pub native: usize,
/// Pipeline subcarrier count after conversion.
pub pipeline: usize,
/// Interpolation/decimation method used (e.g. "catmull-rom-cubic").
pub method: &'static str,
}
/// Wideband spectral provenance metadata (ADR-289): band, bandwidth, native
/// subcarrier dimensionality, and any native → pipeline mapping applied.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WidebandMeta {
/// Radio band the frame was captured on.
pub band: WifiBand,
/// Channel bandwidth in MHz (20160).
pub bandwidth_mhz: u16,
/// Native subcarrier count of the capture (true spectral resolution).
pub native_subcarriers: usize,
/// Native → pipeline conversion record; `None` while the frame is still
/// at native width.
pub mapping: Option<SubcarrierMapping>,
}
/// Antenna configuration for MIMO
@@ -376,6 +502,13 @@ enum DeviceSpecificState {
driver: AtherosDriver,
csi_buf_ptr: Option<u64>,
},
FeitCsi {
/// Byte offset into the capture for deterministic file replay.
replay_offset: u64,
/// Open handle for streaming mode (path/pipe written by an external
/// FeitCSI process); opened lazily on first read.
stream: Option<std::fs::File>,
},
Other,
}
@@ -457,6 +590,7 @@ impl HardwareAdapter {
DeviceType::Atheros(driver) => self.initialize_atheros(*driver).await?,
DeviceType::UdpReceiver => self.initialize_udp().await?,
DeviceType::PcapFile => self.initialize_pcap().await?,
DeviceType::FeitCsi => self.initialize_feitcsi().await?,
DeviceType::Simulated => self.initialize_simulated().await?,
}
@@ -662,6 +796,57 @@ impl HardwareAdapter {
Ok(())
}
/// Initialize FeitCSI file-replay / stream ingest (ADR-289).
///
/// No privileged operations: RuView does not configure the NIC; the
/// external FeitCSI tooling owns capture. This only validates the
/// configured path.
async fn initialize_feitcsi(&mut self) -> Result<(), AdapterError> {
let settings = match &self.config.device_settings {
DeviceSettings::FeitCsi(s) => s,
_ => {
return Err(AdapterError::Config(
"FeitCSI requires FeitCSI settings".into(),
))
}
};
tracing::info!(
"Initializing FeitCSI ingest ({:?}) from {}",
settings.mode,
settings.path
);
match settings.mode {
FeitCsiMode::FileReplay => {
if !std::path::Path::new(&settings.path).exists() {
return Err(AdapterError::Hardware(format!(
"FeitCSI capture file not found: {}",
settings.path
)));
}
}
FeitCsiMode::Stream => {
// The external process may create the pipe/file later; warn
// rather than fail so start order is not constrained.
if !std::path::Path::new(&settings.path).exists() {
tracing::warn!(
"FeitCSI stream path {} does not exist yet; will retry on read",
settings.path
);
}
}
}
let mut state = self.state.write().await;
state.device_state = DeviceSpecificState::FeitCsi {
replay_offset: 0,
stream: None,
};
Ok(())
}
/// Initialize simulated device
async fn initialize_simulated(&mut self) -> Result<(), AdapterError> {
tracing::info!("Initializing simulated CSI device");
@@ -764,7 +949,7 @@ impl HardwareAdapter {
/// Read a single CSI packet from the device
async fn read_csi_packet(
config: &HardwareConfig,
_state: &Arc<RwLock<DeviceState>>,
state: &Arc<RwLock<DeviceState>>,
) -> Result<CsiReadings, AdapterError> {
match &config.device_type {
DeviceType::Esp32 => Self::read_esp32_csi(config).await,
@@ -772,10 +957,141 @@ impl HardwareAdapter {
DeviceType::Atheros(driver) => Self::read_atheros_csi(config, *driver).await,
DeviceType::UdpReceiver => Self::read_udp_csi(config).await,
DeviceType::PcapFile => Self::read_pcap_csi(config).await,
DeviceType::FeitCsi => Self::read_feitcsi_csi(config, state).await,
DeviceType::Simulated => Self::generate_simulated_csi(config).await,
}
}
/// Read one wideband CSI frame from a FeitCSI capture or stream (ADR-289).
///
/// Frames carry their native subcarrier count, bandwidth (20160 MHz) and
/// band (2.4/5/6 GHz) as metadata. When `pipeline_subcarriers` is
/// configured, conversion to pipeline width happens explicitly via the
/// interpolation path and the native → pipeline mapping is recorded in
/// `CsiMetadata::wideband`.
async fn read_feitcsi_csi(
config: &HardwareConfig,
state: &Arc<RwLock<DeviceState>>,
) -> Result<CsiReadings, AdapterError> {
let settings = match &config.device_settings {
DeviceSettings::FeitCsi(s) => s,
_ => return Err(AdapterError::Config("Invalid settings for FeitCSI".into())),
};
let record = match settings.mode {
FeitCsiMode::FileReplay => Self::read_feitcsi_replay(settings, state).await?,
FeitCsiMode::Stream => Self::read_feitcsi_stream(settings, state).await?,
};
let readings = record.to_readings(settings.band, settings.channel);
match settings.pipeline_subcarriers {
Some(n) if n != readings.metadata.num_subcarriers => {
super::feitcsi::resample_readings_to_pipeline(&readings, n)
}
_ => Ok(readings),
}
}
/// Deterministic file replay: reads the record at the current byte offset
/// and advances it, so the capture is walked once from start to end
/// (looping when configured). Same input file ⇒ same record sequence.
async fn read_feitcsi_replay(
settings: &FeitCsiSettings,
state: &Arc<RwLock<DeviceState>>,
) -> Result<super::feitcsi::FeitCsiRecord, AdapterError> {
let offset = {
let st = state.read().await;
match &st.device_state {
DeviceSpecificState::FeitCsi { replay_offset, .. } => *replay_offset,
_ => 0,
}
};
let path = settings.path.clone();
let loop_playback = settings.loop_playback;
let (record, new_offset) = tokio::task::spawn_blocking(
move || -> Result<(super::feitcsi::FeitCsiRecord, u64), AdapterError> {
use std::io::{Seek, SeekFrom};
let mut file = std::fs::File::open(&path).map_err(|e| {
AdapterError::Hardware(format!("Failed to open FeitCSI capture {path}: {e}"))
})?;
file.seek(SeekFrom::Start(offset))
.map_err(AdapterError::Io)?;
match super::feitcsi::read_one_record(&mut file)? {
Some((rec, consumed)) => Ok((rec, offset + consumed as u64)),
None if loop_playback && offset != 0 => {
file.seek(SeekFrom::Start(0)).map_err(AdapterError::Io)?;
match super::feitcsi::read_one_record(&mut file)? {
Some((rec, consumed)) => Ok((rec, consumed as u64)),
None => Err(AdapterError::DataFormat(format!(
"FeitCSI capture {path} contains no records"
))),
}
}
None => Err(AdapterError::HardwareUnavailable(format!(
"End of FeitCSI capture {path} (offset {offset})"
))),
}
},
)
.await
.map_err(|e| AdapterError::Hardware(format!("FeitCSI replay task failed: {e}")))??;
let mut st = state.write().await;
if let DeviceSpecificState::FeitCsi { replay_offset, .. } = &mut st.device_state {
*replay_offset = new_offset;
}
Ok(record)
}
/// Streaming mode: hold the open handle across reads (a FIFO cannot be
/// reopened per record) and block until one full record arrives. The
/// blocking read runs on the blocking pool; if the surrounding stream
/// loop is shut down mid-read, the orphaned task finishes on its own and
/// the handle is reopened on the next read.
async fn read_feitcsi_stream(
settings: &FeitCsiSettings,
state: &Arc<RwLock<DeviceState>>,
) -> Result<super::feitcsi::FeitCsiRecord, AdapterError> {
let existing = {
let mut st = state.write().await;
match &mut st.device_state {
DeviceSpecificState::FeitCsi { stream, .. } => stream.take(),
_ => None,
}
};
let path = settings.path.clone();
let result = tokio::task::spawn_blocking(
move || -> Result<(std::fs::File, super::feitcsi::FeitCsiRecord), AdapterError> {
let mut file = match existing {
Some(f) => f,
None => std::fs::File::open(&path).map_err(|e| {
AdapterError::HardwareUnavailable(format!(
"FeitCSI stream {path} unavailable: {e}"
))
})?,
};
match super::feitcsi::read_one_record(&mut file) {
Ok(Some((rec, _consumed))) => Ok((file, rec)),
Ok(None) => Err(AdapterError::HardwareUnavailable(format!(
"FeitCSI stream {path} closed (EOF)"
))),
Err(e) => Err(e.into()),
}
},
)
.await
.map_err(|e| AdapterError::Hardware(format!("FeitCSI stream task failed: {e}")))?;
let (file, record) = result?;
let mut st = state.write().await;
if let DeviceSpecificState::FeitCsi { stream, .. } = &mut st.device_state {
*stream = Some(file);
}
Ok(record)
}
/// Read CSI from ESP32 via serial.
///
/// The ESP-CSI firmware emits newline-delimited `CSI_DATA,...` CSV records.
@@ -1023,6 +1339,7 @@ impl HardwareAdapter {
rssi: Some(-45.0),
noise_floor: Some(-92.0),
fc_type: FrameControlType::Data,
wideband: None,
},
})
}
@@ -1039,6 +1356,7 @@ impl HardwareAdapter {
DeviceType::Intel5300 | DeviceType::Atheros(_) => self.discover_nic_sensors().await,
DeviceType::UdpReceiver => Ok(vec![]),
DeviceType::PcapFile => Ok(vec![]),
DeviceType::FeitCsi => Ok(vec![]),
DeviceType::Simulated => self.discover_simulated_sensors().await,
}
}
@@ -1165,6 +1483,7 @@ impl HardwareAdapter {
rssi: None,
noise_floor: None,
fc_type: FrameControlType::Data,
wideband: None,
},
})
}
@@ -1320,6 +1639,10 @@ pub struct CsiMetadata {
pub noise_floor: Option<f64>,
/// Frame control type
pub fc_type: FrameControlType,
/// Wideband spectral provenance (ADR-289): band, native subcarrier count
/// and any native → pipeline mapping applied. `None` for legacy
/// narrowband sources that predate wideband metadata.
pub wideband: Option<WidebandMeta>,
}
/// WiFi frame control types
@@ -1640,6 +1963,124 @@ mod tests {
let _ = std::fs::remove_file(&path);
}
#[test]
fn test_feitcsi_config() {
let config = HardwareConfig::feitcsi_replay("/tmp/capture.dat");
assert!(matches!(config.device_type, DeviceType::FeitCsi));
match &config.device_settings {
DeviceSettings::FeitCsi(s) => {
assert_eq!(s.mode, FeitCsiMode::FileReplay);
assert!(s.pipeline_subcarriers.is_none());
}
other => panic!("unexpected settings: {other:?}"),
}
let config = HardwareConfig::feitcsi_stream("/tmp/feitcsi.fifo");
match &config.device_settings {
DeviceSettings::FeitCsi(s) => assert_eq!(s.mode, FeitCsiMode::Stream),
other => panic!("unexpected settings: {other:?}"),
}
}
/// End-to-end FeitCSI file replay through the adapter read path:
/// initialize, then read the capture record-by-record. Two adapters over
/// the same synthetic capture see identical, order-preserving sequences
/// (replay determinism), and native+wideband metadata is carried.
#[tokio::test]
async fn test_feitcsi_replay_end_to_end_deterministic() {
use crate::integration::feitcsi::synth;
// Synthetic 3-record HE 80 MHz capture, generated in code.
let mut capture = Vec::new();
for ts in [100u64, 200, 300] {
capture.extend_from_slice(&synth::record_bytes(2, 1, 996, 2, 4, ts));
}
let path = std::env::temp_dir().join(format!(
"feitcsi_adapter_test_{}.dat",
std::process::id()
));
std::fs::write(&path, &capture).unwrap();
let run = || async {
let mut config = HardwareConfig::feitcsi_replay(path.to_str().unwrap());
if let DeviceSettings::FeitCsi(s) = &mut config.device_settings {
s.band = WifiBand::Band6GHz;
s.channel = 37;
}
let mut adapter = HardwareAdapter::with_config(config.clone());
adapter.initialize().await.unwrap();
let mut frames = Vec::new();
for _ in 0..3 {
let readings = HardwareAdapter::read_csi_packet(&config, &adapter.state)
.await
.expect("replay read");
frames.push(readings);
}
// Capture exhausted: typed error, not fabricated data.
let end = HardwareAdapter::read_csi_packet(&config, &adapter.state).await;
assert!(matches!(end, Err(AdapterError::HardwareUnavailable(_))));
frames
};
let first = run().await;
let second = run().await;
assert_eq!(first.len(), 3);
for (a, b) in first.iter().zip(&second) {
assert_eq!(a.timestamp, b.timestamp, "replay must be deterministic");
assert_eq!(a.readings[0].amplitudes, b.readings[0].amplitudes);
}
// Native wideband metadata is first-class on every frame.
let meta = &first[0].metadata;
assert!(matches!(meta.device_type, DeviceType::FeitCsi));
assert_eq!(meta.num_subcarriers, 996);
assert_eq!(meta.bandwidth, Bandwidth::VHT80);
let wb = meta.wideband.as_ref().expect("wideband metadata");
assert_eq!(wb.band, WifiBand::Band6GHz);
assert_eq!(wb.bandwidth_mhz, 80);
assert_eq!(wb.native_subcarriers, 996);
assert!(wb.mapping.is_none(), "native width: no mapping");
// 2 rx * 1 tx = 2 antenna-pair readings per frame.
assert_eq!(first[0].readings.len(), 2);
let _ = std::fs::remove_file(&path);
}
/// FeitCSI replay with a configured pipeline width converts explicitly
/// through the interpolation path and records the mapping in metadata.
#[tokio::test]
async fn test_feitcsi_replay_pipeline_conversion() {
use crate::integration::feitcsi::synth;
let capture = synth::record_bytes(1, 1, 1992, 3, 4, 42);
let path = std::env::temp_dir().join(format!(
"feitcsi_pipeline_test_{}.dat",
std::process::id()
));
std::fs::write(&path, &capture).unwrap();
let mut config = HardwareConfig::feitcsi_replay(path.to_str().unwrap());
if let DeviceSettings::FeitCsi(s) = &mut config.device_settings {
s.pipeline_subcarriers = Some(56);
}
let mut adapter = HardwareAdapter::with_config(config.clone());
adapter.initialize().await.unwrap();
let readings = HardwareAdapter::read_csi_packet(&config, &adapter.state)
.await
.expect("replay read");
assert_eq!(readings.metadata.num_subcarriers, 56);
assert_eq!(readings.readings[0].amplitudes.len(), 56);
let wb = readings.metadata.wideband.as_ref().unwrap();
assert_eq!(wb.native_subcarriers, 1992, "true resolution preserved");
let mapping = wb.mapping.as_ref().expect("mapping recorded");
assert_eq!((mapping.native, mapping.pipeline), (1992, 56));
let _ = std::fs::remove_file(&path);
}
/// Honest hardware gating: Intel 5300 / Atheros return typed
/// HardwareUnavailable (no device/driver), never fabricated CSI.
#[tokio::test]
@@ -13,6 +13,9 @@
//! - **Intel 5300 NIC**: Using Linux CSI Tool (iwlwifi driver)
//! - **Atheros NICs**: Using ath9k/ath10k/ath11k CSI patches
//! - **Nexmon**: For Broadcom chips with CSI firmware
//! - **FeitCSI (Intel AX200/AX210)**: Wideband 802.11ax CSI up to 160 MHz /
//! 1992 subcarriers including 6 GHz, ingested from recorded captures or a
//! stream written by the external FeitCSI tool (ADR-289)
//!
//! # Example Usage
//!
@@ -37,6 +40,7 @@
//! ```
pub mod csi_receiver;
pub mod feitcsi;
mod hardware_adapter;
mod neural_adapter;
mod signal_adapter;
@@ -52,6 +56,9 @@ pub use hardware_adapter::{
CsiStream,
DeviceSettings,
DeviceType,
// FeitCSI wideband ingest settings (ADR-289)
FeitCsiMode,
FeitCsiSettings,
FlowControl,
FrameControlType,
// Main adapter
@@ -73,8 +80,18 @@ pub use hardware_adapter::{
// Serial settings
SerialSettings,
StreamingStats,
// Wideband spectral provenance (ADR-289)
SubcarrierMapping,
// UDP settings
UdpSettings,
WidebandMeta,
WifiBand,
};
pub use feitcsi::{
parse_record as parse_feitcsi_record, resample_readings_to_pipeline, FeitCsiBandwidth,
FeitCsiError, FeitCsiFileReader, FeitCsiHeader, FeitCsiModType, FeitCsiRecord,
FeitCsiStreamReader,
};
pub use neural_adapter::NeuralAdapter;
pub use signal_adapter::SignalAdapter;
@@ -107,3 +107,9 @@ ndarray-npy.workspace = true
[[bench]]
name = "training_bench"
harness = false
# ADR-288 — bfee-parser throughput and split-assignment benchmarks on
# synthetic, code-generated corpora (no dataset files).
[[bench]]
name = "benchmark_harness"
harness = false
@@ -0,0 +1,127 @@
//! ADR-288 benchmarks: bfee parser throughput and split assignment over
//! synthetic, code-generated corpora (no dataset files are read or written).
use criterion::{black_box, criterion_group, criterion_main, Criterion, Throughput};
use wifi_densepose_train::dataset::widar::{encode_bfee_frame, parse_bfee_bytes, WIDAR_SUBCARRIERS};
use wifi_densepose_train::protocols::leakage::LeakageAudit;
use wifi_densepose_train::protocols::{SampleMeta, SplitPlan, SplitProtocol, SplitSide};
/// Deterministic synthetic bfee log: `num_records` framed 3×3 records.
fn synthetic_log(num_records: usize) -> Vec<u8> {
let (n_rx, n_tx) = (3u8, 3u8);
let pairs = WIDAR_SUBCARRIERS * n_rx as usize * n_tx as usize;
let mut bytes = Vec::new();
for t in 0..num_records {
let csi: Vec<(i16, i16)> = (0..pairs)
.map(|i| {
let re = ((t * 37 + i * 13) % 1024) as i16 - 512;
let im = ((t * 17 + i * 7) % 1024) as i16 - 512;
(re, im)
})
.collect();
bytes.extend_from_slice(&encode_bfee_frame(t as u32, t as u16, n_rx, n_tx, &csi));
}
bytes
}
/// Deterministic synthetic metadata corpus.
fn synthetic_metas(n: usize) -> Vec<SampleMeta> {
(0..n)
.map(|i| SampleMeta {
subject_id: 1 + (i % 17) as u32,
environment_id: 1 + (i % 3) as u32,
orientation_id: 1 + (i % 5) as u32,
gesture_id: 1 + (i % 6) as u32,
recording_id: (i / 50) as u64,
window_index: (i % 50) as u64,
})
.collect()
}
/// Deterministic synthetic corpus whose domain attributes are constant per
/// recording (as real datasets are), so protocol splits keep recordings whole
/// and the leakage audit exercises its full passing path.
fn synthetic_recording_metas(n: usize, windows_per_recording: usize) -> Vec<SampleMeta> {
(0..n)
.map(|i| {
let recording = (i / windows_per_recording) as u64;
SampleMeta {
subject_id: 1 + (recording % 17) as u32,
environment_id: 1 + (recording % 3) as u32,
orientation_id: 1 + (recording % 5) as u32,
gesture_id: 1 + (recording % 6) as u32,
recording_id: recording,
window_index: (i % windows_per_recording) as u64,
}
})
.collect()
}
fn bench_bfee_parser(c: &mut Criterion) {
let bytes = synthetic_log(500);
let mut group = c.benchmark_group("widar_bfee_parse");
group.throughput(Throughput::Bytes(bytes.len() as u64));
group.bench_function("500_records_3x3", |b| {
b.iter(|| {
let parse = parse_bfee_bytes(black_box(&bytes));
assert_eq!(parse.records.len(), 500);
parse
})
});
group.finish();
}
fn bench_split_assignment(c: &mut Criterion) {
let metas = synthetic_metas(10_000);
let mut group = c.benchmark_group("split_assignment");
group.throughput(Throughput::Elements(metas.len() as u64));
for protocol in [
SplitProtocol::CrossSubject,
SplitProtocol::CrossEnvironment,
SplitProtocol::CrossOrientation,
SplitProtocol::RandomBaseline,
] {
let plan = SplitPlan::new(protocol, 42, 0.3).expect("valid fraction");
group.bench_function(protocol.tag(), |b| {
b.iter(|| plan.partition(black_box(&metas)))
});
}
group.finish();
}
fn bench_leakage_audit(c: &mut Criterion) {
// ~10k windows in 200 recordings; a clean cross-subject split so the
// audit runs every check (recording crossing + claimed disjointness) to
// completion instead of failing fast.
let metas = synthetic_recording_metas(10_000, 50);
let plan = SplitPlan::new(SplitProtocol::CrossSubject, 42, 0.3).expect("valid fraction");
let mut train = Vec::new();
let mut test = Vec::new();
for meta in &metas {
match plan.assign(meta) {
SplitSide::Train => train.push(*meta),
SplitSide::Test => test.push(*meta),
}
}
assert!(!train.is_empty() && !test.is_empty(), "degenerate corpus");
let audit = LeakageAudit::for_protocol(SplitProtocol::CrossSubject);
let mut group = c.benchmark_group("leakage_audit");
group.throughput(Throughput::Elements(metas.len() as u64));
group.bench_function("cross_subject_10k_windows", |b| {
b.iter(|| {
audit
.audit(black_box(&train), black_box(&test))
.expect("clean split must pass")
})
});
group.finish();
}
criterion_group!(
benches,
bench_bfee_parser,
bench_split_assignment,
bench_leakage_audit
);
criterion_main!(benches);
@@ -40,6 +40,10 @@
//! assert_eq!(sample.amplitude.shape(), &[100, 3, 3, 56]);
//! ```
/// Widar3.0 ingest — Intel 5300 `.dat` "bfee" parser and [`CsiDataset`]
/// adapter with split-protocol metadata (ADR-288 §1).
pub mod widar;
use ndarray::{Array1, Array2, Array4};
use ruvector_temporal_tensor::segment as tt_segment;
use ruvector_temporal_tensor::{TemporalTensorCompressor, TierPolicy};
File diff suppressed because it is too large Load Diff
+101 -1
View File
@@ -12,7 +12,8 @@
//! ├── ConfigError (config validation / file loading)
//! ├── DatasetError (data loading, I/O, format)
//! ├── SubcarrierError (frequency-axis resampling)
//! ── MaeError (MAE patchify / masking — ADR-152 §2.3)
//! ── MaeError (MAE patchify / masking — ADR-152 §2.3)
//! └── ProtocolError (split protocols / leakage audit — ADR-288)
//! ```
use std::path::PathBuf;
@@ -49,6 +50,10 @@ pub enum TrainError {
#[error("MAE pretraining error: {0}")]
Mae(#[from] MaeError),
/// A split-protocol / leakage-audit error (ADR-288).
#[error("Protocol error: {0}")]
Protocol(#[from] ProtocolError),
/// JSON (de)serialization error.
#[error("JSON error: {0}")]
Json(#[from] serde_json::Error),
@@ -466,3 +471,98 @@ pub enum MaeError {
value: f32,
},
}
// ---------------------------------------------------------------------------
// ProtocolError
// ---------------------------------------------------------------------------
/// Errors produced by the public-benchmark split protocols and leakage guards
/// ([`crate::protocols`], ADR-288).
///
/// Every leakage-audit failure is an `Err`, never a warning: a split that
/// leaks subjects, environments, or windows of a continuous recording across
/// the train/test boundary must not be usable for reporting.
#[derive(Debug, Error)]
pub enum ProtocolError {
/// The requested held-out fraction is not a finite value strictly inside
/// `(0, 1)`.
#[error("Invalid test fraction {value}: must be finite and strictly inside (0, 1)")]
InvalidTestFraction {
/// The offending fraction.
value: f64,
},
/// A split side contains no samples — a degenerate split cannot support
/// any claim.
#[error("The {side} partition is empty")]
EmptyPartition {
/// Which side is empty (`"train"` or `"test"`).
side: &'static str,
},
/// A subject appears on both sides of a split that claims
/// subject-disjointness.
#[error("Subject {subject_id} appears in both train and test (subject leakage)")]
SubjectOverlap {
/// The leaked subject id.
subject_id: u32,
},
/// An environment/room appears on both sides of a split that claims
/// environment-disjointness.
#[error("Environment {environment_id} appears in both train and test (environment leakage)")]
EnvironmentOverlap {
/// The leaked environment id.
environment_id: u32,
},
/// An orientation appears on both sides of a split that claims
/// orientation-disjointness.
#[error("Orientation {orientation_id} appears in both train and test (orientation leakage)")]
OrientationOverlap {
/// The leaked orientation id.
orientation_id: u32,
},
/// Two windows cut from the same continuous recording ended up on
/// opposite sides of the split. Overlapping/adjacent windows are
/// near-identical, so this is window-level leakage regardless of the
/// protocol (the 20242025 leakage reckoning; ADR-288 §Context).
#[error(
"Recording {recording_id} has windows on both sides of the split \
(window-level leakage from a continuous recording)"
)]
RecordingCrossesSplit {
/// The recording whose windows straddle the boundary.
recording_id: u64,
},
/// The mean-pose baseline cannot be fitted because the training split
/// contributed no poses.
#[error("Cannot fit mean-pose baseline: the training split contains no poses")]
EmptyTrainingPoses,
/// A pose array has a different shape from the first pose seen.
#[error("Pose shape mismatch: expected {expected:?}, got {actual:?}")]
PoseShapeMismatch {
/// Shape established by the first pose.
expected: Vec<usize>,
/// Offending shape.
actual: Vec<usize>,
},
/// A `MEASURED` evidence grade was requested without a reproducer
/// command. CLAUDE.md: accuracy statements tagged `MEASURED` require a
/// reproducer; anything else must be `SYNTHETIC` or `CLAIMED`.
#[error("MEASURED evidence requires a non-empty reproducer command string")]
MissingReproducer,
/// A reported metric is NaN or ±inf.
#[error("Metric `{name}` is not finite: {value}")]
NonFiniteMetric {
/// Name of the offending metric.
name: String,
/// The non-finite value.
value: f64,
},
}
+13 -1
View File
@@ -59,6 +59,11 @@ pub mod mae;
/// `oks_canonical`, available **without** the `tch-backend` feature so the
/// single metric definition is reachable from the workspace test gate.
pub mod metrics_core;
/// Public-benchmark split protocols and leakage guards (ADR-288 §23) —
/// deterministic cross-subject / cross-environment / cross-orientation
/// assignment plus the structural [`protocols::leakage::LeakageAudit`],
/// mean-pose baseline, and evidence-graded evaluation reports.
pub mod protocols;
pub mod rapid_adapt;
pub mod ruview_metrics;
pub mod signal_features;
@@ -103,7 +108,14 @@ pub use config::TrainingConfig;
pub use dataset::{
CsiDataset, CsiSample, DataLoader, MmFiDataset, SyntheticConfig, SyntheticCsiDataset,
};
pub use error::{ConfigError, DatasetError, MaeError, SubcarrierError, TrainError};
// ADR-288 — Widar3.0 ingest, split protocols, and leakage guards.
pub use dataset::widar::{parse_bfee_bytes, BfeeParse, BfeeRecord, WidarDataset, WidarFileMeta};
pub use protocols::leakage::{
EvaluationReport, EvidenceGrade, LeakageAudit, LeakageClaims, MeanPoseBaseline,
};
pub use protocols::{SampleMeta, SplitPlan, SplitProtocol, SplitSide};
pub use error::{ConfigError, DatasetError, MaeError, ProtocolError, SubcarrierError, TrainError};
// TrainResult<T> is the generic Result alias from error.rs; the concrete
// TrainResult struct from trainer.rs is accessed via trainer::TrainResult.
pub use error::TrainResult as TrainResultAlias;
@@ -0,0 +1,388 @@
//! Standard public-benchmark split protocols (ADR-288 §2).
//!
//! The field's documented leakage failure is the window-level random split:
//! adjacent windows cut from one continuous recording are near-identical, so
//! splitting them across train/test inflates accuracy (one dataset's F1
//! collapsed from ~90% to ~22% under subject-disjoint splits — ADR-288
//! §Context). This module expresses the standard leaderboard evaluations as a
//! [`SplitProtocol`] whose assignment is a **pure function of sample metadata
//! plus a seed** — no RNG state, no iteration-order dependence, byte-identical
//! across runs and platforms.
//!
//! - [`SplitProtocol::CrossSubject`] — MM-Fi-style: held-out subjects.
//! - [`SplitProtocol::CrossEnvironment`] — held-out rooms/environments.
//! - [`SplitProtocol::CrossOrientation`] — Widar-style: held-out orientations.
//! - [`SplitProtocol::RandomBaseline`] — window-level random split, kept
//! **only** as the explicitly leakage-prone comparison point; it makes no
//! disjointness claim and will normally fail the
//! [`leakage::LeakageAudit`].
//!
//! Structural verification of a produced split lives in [`leakage`].
pub mod leakage;
use serde::{Deserialize, Serialize};
use crate::error::ProtocolError;
// ---------------------------------------------------------------------------
// SampleMeta
// ---------------------------------------------------------------------------
/// Loader-agnostic per-window metadata consumed by split assignment and the
/// leakage audit. Produced by e.g.
/// [`WidarDataset::sample_meta`](crate::dataset::widar::WidarDataset::sample_meta).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct SampleMeta {
/// Subject/user id.
pub subject_id: u32,
/// Environment/room id (`0` when the dataset tree does not encode one).
pub environment_id: u32,
/// Orientation id (Widar face orientation; `0` when unknown).
pub orientation_id: u32,
/// Gesture/action id.
pub gesture_id: u32,
/// Identifier of the continuous recording this window was cut from.
/// Windows sharing a `recording_id` are temporally correlated and must
/// never straddle a train/test boundary.
pub recording_id: u64,
/// Window offset within the recording.
pub window_index: u64,
}
// ---------------------------------------------------------------------------
// SplitProtocol
// ---------------------------------------------------------------------------
/// Which side of a train/test split a sample is assigned to.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum SplitSide {
/// Training partition.
Train,
/// Held-out test partition.
Test,
}
/// A standard evaluation protocol determining *what* is held out.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum SplitProtocol {
/// Hold out whole subjects (MM-Fi cross-subject protocol).
CrossSubject,
/// Hold out whole environments/rooms (MM-Fi cross-environment protocol).
CrossEnvironment,
/// Hold out whole orientations (Widar3.0 cross-orientation protocol).
CrossOrientation,
/// Window-level random split. **Leakage-prone by construction** — kept
/// only so leaderboard-style numbers can be contrasted against a leaky
/// baseline; it claims no disjointness and normally fails the audit.
RandomBaseline,
}
impl SplitProtocol {
/// Stable lowercase tag for logs/reports.
#[must_use]
pub fn tag(self) -> &'static str {
match self {
SplitProtocol::CrossSubject => "cross-subject",
SplitProtocol::CrossEnvironment => "cross-environment",
SplitProtocol::CrossOrientation => "cross-orientation",
SplitProtocol::RandomBaseline => "random-baseline-leaky",
}
}
/// Disjointness this protocol claims and the audit must verify.
#[must_use]
pub fn claims(self) -> leakage::LeakageClaims {
match self {
SplitProtocol::CrossSubject => leakage::LeakageClaims {
subject_disjoint: true,
environment_disjoint: false,
orientation_disjoint: false,
},
SplitProtocol::CrossEnvironment => leakage::LeakageClaims {
subject_disjoint: false,
environment_disjoint: true,
orientation_disjoint: false,
},
SplitProtocol::CrossOrientation => leakage::LeakageClaims {
subject_disjoint: false,
environment_disjoint: false,
orientation_disjoint: true,
},
SplitProtocol::RandomBaseline => leakage::LeakageClaims {
subject_disjoint: false,
environment_disjoint: false,
orientation_disjoint: false,
},
}
}
}
// ---------------------------------------------------------------------------
// SplitPlan
// ---------------------------------------------------------------------------
/// A concrete, seeded instantiation of a [`SplitProtocol`].
///
/// [`SplitPlan::assign`] is a pure function: the same `(protocol, seed,
/// test_fraction, meta)` always yields the same side, independent of call
/// order, thread, or platform.
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub struct SplitPlan {
/// The evaluation protocol.
pub protocol: SplitProtocol,
/// Seed mixed into every assignment hash.
pub seed: u64,
/// Target fraction of held-out *units* (subjects / environments /
/// orientations / windows, per protocol), strictly inside `(0, 1)`.
pub test_fraction: f64,
}
impl SplitPlan {
/// Create a plan, validating `test_fraction`.
///
/// # Errors
///
/// [`ProtocolError::InvalidTestFraction`] when the fraction is not finite
/// or not strictly inside `(0, 1)`.
pub fn new(
protocol: SplitProtocol,
seed: u64,
test_fraction: f64,
) -> Result<Self, ProtocolError> {
if !test_fraction.is_finite() || test_fraction <= 0.0 || test_fraction >= 1.0 {
return Err(ProtocolError::InvalidTestFraction {
value: test_fraction,
});
}
Ok(SplitPlan {
protocol,
seed,
test_fraction,
})
}
/// Assign one sample to a side — pure, deterministic, stateless.
///
/// The protocol's held-out *unit* (subject, environment, orientation, or
/// individual window) is hashed together with a protocol-specific domain
/// tag and the seed; the unit lands in the test set when its hash falls
/// below `test_fraction` of the hash space. All windows of one unit
/// therefore always land on the same side (except under
/// [`SplitProtocol::RandomBaseline`], which hashes per window — that is
/// its documented leak).
#[must_use]
pub fn assign(&self, meta: &SampleMeta) -> SplitSide {
// Distinct domain tags keep e.g. subject 3 and orientation 3 from
// sharing a hash under the same seed.
const DOMAIN_SUBJECT: u64 = 0x5355424a; // "SUBJ"
const DOMAIN_ENVIRONMENT: u64 = 0x454e5652; // "ENVR"
const DOMAIN_ORIENTATION: u64 = 0x4f524e54; // "ORNT"
const DOMAIN_RANDOM: u64 = 0x524e444d; // "RNDM"
let unit = match self.protocol {
SplitProtocol::CrossSubject => {
mix2(DOMAIN_SUBJECT, meta.subject_id as u64)
}
SplitProtocol::CrossEnvironment => {
mix2(DOMAIN_ENVIRONMENT, meta.environment_id as u64)
}
SplitProtocol::CrossOrientation => {
mix2(DOMAIN_ORIENTATION, meta.orientation_id as u64)
}
SplitProtocol::RandomBaseline => mix2(
mix2(DOMAIN_RANDOM, meta.recording_id),
meta.window_index,
),
};
let h = splitmix64(unit ^ splitmix64(self.seed));
// Integer threshold comparison — no float accumulation, identical on
// every platform.
let threshold = (self.test_fraction * (1u128 << 64) as f64) as u128;
if (h as u128) < threshold {
SplitSide::Test
} else {
SplitSide::Train
}
}
/// Partition metadata into `(train_indices, test_indices)` by
/// [`Self::assign`], preserving input order within each side.
#[must_use]
pub fn partition(&self, metas: &[SampleMeta]) -> (Vec<usize>, Vec<usize>) {
let mut train = Vec::new();
let mut test = Vec::new();
for (i, meta) in metas.iter().enumerate() {
match self.assign(meta) {
SplitSide::Train => train.push(i),
SplitSide::Test => test.push(i),
}
}
(train, test)
}
}
/// SplitMix64 finalizer — a well-distributed 64-bit mixing function.
fn splitmix64(mut x: u64) -> u64 {
x = x.wrapping_add(0x9E37_79B9_7F4A_7C15);
x = (x ^ (x >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
x = (x ^ (x >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
x ^ (x >> 31)
}
/// Order-sensitive combination of two words through SplitMix64.
fn mix2(a: u64, b: u64) -> u64 {
splitmix64(splitmix64(a) ^ b.rotate_left(32))
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
/// 4 subjects × 2 environments × 4 orientations, 2 recordings each with
/// 5 windows — a deterministic synthetic corpus.
fn corpus() -> Vec<SampleMeta> {
let mut metas = Vec::new();
let mut recording = 0u64;
for subject in 1..=4u32 {
for environment in 1..=2u32 {
for orientation in 1..=4u32 {
for _ in 0..2 {
for window in 0..5u64 {
metas.push(SampleMeta {
subject_id: subject,
environment_id: environment,
orientation_id: orientation,
gesture_id: 1 + (recording % 6) as u32,
recording_id: recording,
window_index: window,
});
}
recording += 1;
}
}
}
}
metas
}
#[test]
fn plan_rejects_bad_fractions() {
for bad in [0.0, 1.0, -0.2, 1.7, f64::NAN, f64::INFINITY] {
assert!(matches!(
SplitPlan::new(SplitProtocol::CrossSubject, 1, bad),
Err(ProtocolError::InvalidTestFraction { .. })
));
}
assert!(SplitPlan::new(SplitProtocol::CrossSubject, 1, 0.25).is_ok());
}
#[test]
fn assignment_is_deterministic_across_calls_and_order() {
let metas = corpus();
let plan = SplitPlan::new(SplitProtocol::CrossSubject, 42, 0.3).unwrap();
let (tr1, te1) = plan.partition(&metas);
let (tr2, te2) = plan.partition(&metas);
assert_eq!(tr1, tr2);
assert_eq!(te1, te2);
// Pure per-sample function: reversing iteration order changes nothing.
let reversed: Vec<SampleMeta> = metas.iter().rev().copied().collect();
for (meta, rev) in metas.iter().zip(reversed.iter().rev()) {
assert_eq!(plan.assign(meta), plan.assign(rev));
}
}
#[test]
fn different_seeds_change_the_split() {
let metas = corpus();
let a = SplitPlan::new(SplitProtocol::CrossSubject, 1, 0.5).unwrap();
let b = SplitPlan::new(SplitProtocol::CrossSubject, 2, 0.5).unwrap();
// With 4 subjects at 50% some seed pair must differ; these two do —
// and if the hash ever changes this test flags the compat break.
let (_, te_a) = a.partition(&metas);
let (_, te_b) = b.partition(&metas);
assert_ne!(te_a, te_b, "seeds 1 and 2 should hold out different subjects");
}
#[test]
fn cross_subject_keeps_subjects_whole() {
let metas = corpus();
let plan = SplitPlan::new(SplitProtocol::CrossSubject, 7, 0.4).unwrap();
let mut side_by_subject = std::collections::BTreeMap::new();
for meta in &metas {
let side = plan.assign(meta);
let prev = side_by_subject.insert(meta.subject_id, side);
if let Some(prev) = prev {
assert_eq!(prev, side, "subject {} split across sides", meta.subject_id);
}
}
}
#[test]
fn cross_environment_keeps_environments_whole() {
let metas = corpus();
let plan = SplitPlan::new(SplitProtocol::CrossEnvironment, 11, 0.5).unwrap();
let mut side_by_env = std::collections::BTreeMap::new();
for meta in &metas {
let side = plan.assign(meta);
if let Some(prev) = side_by_env.insert(meta.environment_id, side) {
assert_eq!(prev, side);
}
}
}
#[test]
fn cross_orientation_keeps_orientations_whole() {
let metas = corpus();
let plan = SplitPlan::new(SplitProtocol::CrossOrientation, 13, 0.5).unwrap();
let mut side_by_orient = std::collections::BTreeMap::new();
for meta in &metas {
let side = plan.assign(meta);
if let Some(prev) = side_by_orient.insert(meta.orientation_id, side) {
assert_eq!(prev, side);
}
}
}
#[test]
fn random_baseline_splits_within_recordings() {
// The leaky baseline must (for some recording) place windows of the
// same recording on both sides — that is the leak it demonstrates.
let metas = corpus();
let plan = SplitPlan::new(SplitProtocol::RandomBaseline, 3, 0.5).unwrap();
let mut crossing = false;
let mut side_by_recording = std::collections::BTreeMap::new();
for meta in &metas {
let side = plan.assign(meta);
if let Some(prev) = side_by_recording.insert(meta.recording_id, side) {
if prev != side {
crossing = true;
}
}
}
assert!(crossing, "window-level split should cross recordings");
}
#[test]
fn protocol_claims_match_semantics() {
assert!(SplitProtocol::CrossSubject.claims().subject_disjoint);
assert!(SplitProtocol::CrossEnvironment.claims().environment_disjoint);
assert!(SplitProtocol::CrossOrientation.claims().orientation_disjoint);
let random = SplitProtocol::RandomBaseline.claims();
assert!(!random.subject_disjoint);
assert!(!random.environment_disjoint);
assert!(!random.orientation_disjoint);
}
#[test]
fn tags_are_stable() {
assert_eq!(SplitProtocol::CrossSubject.tag(), "cross-subject");
assert_eq!(SplitProtocol::RandomBaseline.tag(), "random-baseline-leaky");
}
}
@@ -0,0 +1,703 @@
//! Structural leakage guards, mean-pose baseline, and evidence-graded
//! evaluation reports (ADR-288 §3).
//!
//! Three enforcement points, all `Err`-on-failure (never a warning):
//!
//! 1. [`LeakageAudit`] verifies a proposed train/test split structurally:
//! subject-disjointness and environment/orientation-disjointness **where
//! the protocol claims them**, and — unconditionally — that no two windows
//! cut from the same continuous recording straddle the boundary.
//! 2. [`MeanPoseBaseline`] is fitted from the *training* split only; PCK /
//! MPJPE numbers are meaningless without it (CLAUDE.md: pose PCK requires
//! the mean-pose baseline).
//! 3. [`EvaluationReport`] pairs the model metric with the baseline metric
//! and carries an [`EvidenceGrade`]; `MEASURED` cannot be constructed
//! without an embedded reproducer command string.
use ndarray::Array2;
use serde::{Deserialize, Serialize};
use std::collections::BTreeSet;
use crate::error::ProtocolError;
use crate::protocols::{SampleMeta, SplitProtocol};
// ---------------------------------------------------------------------------
// LeakageAudit
// ---------------------------------------------------------------------------
/// Disjointness properties a protocol claims; the audit verifies each claimed
/// one. Obtained from [`SplitProtocol::claims`], or constructed directly for
/// custom protocols.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeakageClaims {
/// Train and test must share no subject.
pub subject_disjoint: bool,
/// Train and test must share no environment/room.
pub environment_disjoint: bool,
/// Train and test must share no orientation.
pub orientation_disjoint: bool,
}
/// Summary returned by a **passing** audit — counts for reporting, no claim
/// stronger than what was structurally checked.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LeakageAuditPass {
/// Claims that were verified.
pub claims: LeakageClaims,
/// Number of training windows.
pub train_windows: usize,
/// Number of test windows.
pub test_windows: usize,
/// Distinct subjects in train.
pub train_subjects: usize,
/// Distinct subjects in test.
pub test_subjects: usize,
/// Distinct continuous recordings in train.
pub train_recordings: usize,
/// Distinct continuous recordings in test.
pub test_recordings: usize,
}
/// Structural train/test-split auditor (ADR-288 §3).
///
/// A failed audit is an [`Err`], not a warning: leaky splits must be unusable
/// for reporting, not merely frowned upon.
#[derive(Debug, Clone, Copy)]
pub struct LeakageAudit {
claims: LeakageClaims,
}
impl LeakageAudit {
/// Auditor for an explicit set of claims.
#[must_use]
pub fn new(claims: LeakageClaims) -> Self {
LeakageAudit { claims }
}
/// Auditor for the claims a standard protocol makes.
#[must_use]
pub fn for_protocol(protocol: SplitProtocol) -> Self {
LeakageAudit {
claims: protocol.claims(),
}
}
/// Verify a proposed split.
///
/// Checks, in order:
/// 1. both partitions are non-empty;
/// 2. no continuous recording has windows on both sides (unconditional —
/// overlapping windows of one recording are near-duplicates);
/// 3. subject-disjointness, when claimed;
/// 4. environment-disjointness, when claimed;
/// 5. orientation-disjointness, when claimed.
///
/// # Errors
///
/// The [`ProtocolError`] variant describing the **first** violation found.
pub fn audit(
&self,
train: &[SampleMeta],
test: &[SampleMeta],
) -> Result<LeakageAuditPass, ProtocolError> {
if train.is_empty() {
return Err(ProtocolError::EmptyPartition { side: "train" });
}
if test.is_empty() {
return Err(ProtocolError::EmptyPartition { side: "test" });
}
// (2) Recording windows must never cross the boundary.
let train_recordings: BTreeSet<u64> = train.iter().map(|m| m.recording_id).collect();
let test_recordings: BTreeSet<u64> = test.iter().map(|m| m.recording_id).collect();
if let Some(&recording_id) = train_recordings.intersection(&test_recordings).next() {
return Err(ProtocolError::RecordingCrossesSplit { recording_id });
}
// (35) Claimed disjointness.
let train_subjects: BTreeSet<u32> = train.iter().map(|m| m.subject_id).collect();
let test_subjects: BTreeSet<u32> = test.iter().map(|m| m.subject_id).collect();
if self.claims.subject_disjoint {
if let Some(&subject_id) = train_subjects.intersection(&test_subjects).next() {
return Err(ProtocolError::SubjectOverlap { subject_id });
}
}
if self.claims.environment_disjoint {
let train_envs: BTreeSet<u32> = train.iter().map(|m| m.environment_id).collect();
let test_envs: BTreeSet<u32> = test.iter().map(|m| m.environment_id).collect();
if let Some(&environment_id) = train_envs.intersection(&test_envs).next() {
return Err(ProtocolError::EnvironmentOverlap { environment_id });
}
}
if self.claims.orientation_disjoint {
let train_orients: BTreeSet<u32> = train.iter().map(|m| m.orientation_id).collect();
let test_orients: BTreeSet<u32> = test.iter().map(|m| m.orientation_id).collect();
if let Some(&orientation_id) = train_orients.intersection(&test_orients).next() {
return Err(ProtocolError::OrientationOverlap { orientation_id });
}
}
Ok(LeakageAuditPass {
claims: self.claims,
train_windows: train.len(),
test_windows: test.len(),
train_subjects: train_subjects.len(),
test_subjects: test_subjects.len(),
train_recordings: train_recordings.len(),
test_recordings: test_recordings.len(),
})
}
}
// ---------------------------------------------------------------------------
// MeanPoseBaseline
// ---------------------------------------------------------------------------
/// The mean-pose baseline: predicts the per-joint mean of the **training**
/// poses for every test sample (CLAUDE.md: pose PCK requires this baseline —
/// a model must beat "always predict the average pose" before any number
/// means anything).
#[derive(Debug, Clone, PartialEq)]
pub struct MeanPoseBaseline {
mean_pose: Array2<f32>,
num_train_poses: usize,
}
impl MeanPoseBaseline {
/// Fit the baseline from training-split poses only. Each pose is
/// `[num_joints, 2]` (normalised x, y); all poses must share one shape.
///
/// # Errors
///
/// - [`ProtocolError::EmptyTrainingPoses`] when `train_poses` is empty.
/// - [`ProtocolError::PoseShapeMismatch`] when poses disagree in shape.
pub fn fit(train_poses: &[Array2<f32>]) -> Result<Self, ProtocolError> {
let first = train_poses.first().ok_or(ProtocolError::EmptyTrainingPoses)?;
let shape = first.dim();
let mut mean_pose = Array2::<f32>::zeros(shape);
for pose in train_poses {
if pose.dim() != shape {
return Err(ProtocolError::PoseShapeMismatch {
expected: vec![shape.0, shape.1],
actual: pose.shape().to_vec(),
});
}
mean_pose += pose;
}
mean_pose /= train_poses.len() as f32;
Ok(MeanPoseBaseline {
mean_pose,
num_train_poses: train_poses.len(),
})
}
/// The fitted mean pose, `[num_joints, 2]`.
#[must_use]
pub fn mean_pose(&self) -> &Array2<f32> {
&self.mean_pose
}
/// Number of training poses the baseline was fitted on.
#[must_use]
pub fn num_train_poses(&self) -> usize {
self.num_train_poses
}
/// Mean per-joint position error (MPJPE) of the baseline over test-split
/// poses: the mean Euclidean distance between each test joint and the
/// corresponding mean-pose joint.
///
/// # Errors
///
/// - [`ProtocolError::EmptyTrainingPoses`] when `test_poses` is empty
/// (nothing to evaluate).
/// - [`ProtocolError::PoseShapeMismatch`] when a test pose does not match
/// the fitted shape.
pub fn mpjpe(&self, test_poses: &[Array2<f32>]) -> Result<f64, ProtocolError> {
if test_poses.is_empty() {
return Err(ProtocolError::EmptyTrainingPoses);
}
let shape = self.mean_pose.dim();
let mut total = 0.0f64;
let mut joints = 0usize;
for pose in test_poses {
if pose.dim() != shape {
return Err(ProtocolError::PoseShapeMismatch {
expected: vec![shape.0, shape.1],
actual: pose.shape().to_vec(),
});
}
for j in 0..shape.0 {
let dx = (pose[[j, 0]] - self.mean_pose[[j, 0]]) as f64;
let dy = (pose[[j, 1]] - self.mean_pose[[j, 1]]) as f64;
total += (dx * dx + dy * dy).sqrt();
joints += 1;
}
}
Ok(total / joints as f64)
}
/// Baseline PCK@`threshold` over test poses: fraction of joints whose
/// distance to the mean-pose joint is `< threshold` (same units as the
/// pose coordinates).
///
/// # Errors
///
/// Same conditions as [`Self::mpjpe`].
pub fn pck_at(
&self,
test_poses: &[Array2<f32>],
threshold: f32,
) -> Result<f64, ProtocolError> {
if test_poses.is_empty() {
return Err(ProtocolError::EmptyTrainingPoses);
}
let shape = self.mean_pose.dim();
let mut correct = 0usize;
let mut joints = 0usize;
for pose in test_poses {
if pose.dim() != shape {
return Err(ProtocolError::PoseShapeMismatch {
expected: vec![shape.0, shape.1],
actual: pose.shape().to_vec(),
});
}
for j in 0..shape.0 {
let dx = pose[[j, 0]] - self.mean_pose[[j, 0]];
let dy = pose[[j, 1]] - self.mean_pose[[j, 1]];
if (dx * dx + dy * dy).sqrt() < threshold {
correct += 1;
}
joints += 1;
}
}
Ok(correct as f64 / joints as f64)
}
}
// ---------------------------------------------------------------------------
// EvaluationReport
// ---------------------------------------------------------------------------
/// Evidence grade of a reported number (CLAUDE.md tagging rules).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum EvidenceGrade {
/// Measured on real data with a leak-free split; carries the exact
/// command that reproduces the number. Constructible only through
/// [`EvaluationReport::measured`], which rejects an empty reproducer.
Measured {
/// Command line that reproduces this result.
reproducer: String,
},
/// Computed on synthetic/generated data.
Synthetic,
/// Quoted from elsewhere; not reproduced in this repository.
Claimed,
}
impl EvidenceGrade {
/// Stable uppercase tag (`MEASURED` / `SYNTHETIC` / `CLAIMED`).
#[must_use]
pub fn tag(&self) -> &'static str {
match self {
EvidenceGrade::Measured { .. } => "MEASURED",
EvidenceGrade::Synthetic => "SYNTHETIC",
EvidenceGrade::Claimed => "CLAIMED",
}
}
}
/// An evaluation result that structurally pairs the model metric with the
/// mean-pose (or other) baseline metric and an [`EvidenceGrade`] — a model
/// number can never be reported without its baseline (ADR-288 §3).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EvaluationReport {
/// Protocol the split followed.
pub protocol: SplitProtocol,
/// Metric name, e.g. `"pck@0.2"` or `"mpjpe"`.
pub metric_name: String,
/// The model's metric on the audited test split.
pub model_metric: f64,
/// The baseline's metric on the same split (e.g.
/// [`MeanPoseBaseline::mpjpe`]).
pub baseline_metric: f64,
/// Evidence grade; `MEASURED` embeds its reproducer.
pub evidence: EvidenceGrade,
}
impl EvaluationReport {
/// Build a `MEASURED` report. The reproducer command is mandatory and
/// must be non-blank — a measured number without a reproducer is not
/// measured (CLAUDE.md).
///
/// # Errors
///
/// - [`ProtocolError::MissingReproducer`] when `reproducer` is blank.
/// - [`ProtocolError::NonFiniteMetric`] when either metric is NaN/±inf.
pub fn measured(
protocol: SplitProtocol,
metric_name: impl Into<String>,
model_metric: f64,
baseline_metric: f64,
reproducer: impl Into<String>,
) -> Result<Self, ProtocolError> {
let reproducer = reproducer.into();
if reproducer.trim().is_empty() {
return Err(ProtocolError::MissingReproducer);
}
Self::build(
protocol,
metric_name.into(),
model_metric,
baseline_metric,
EvidenceGrade::Measured { reproducer },
)
}
/// Build a `SYNTHETIC` report (synthetic/generated data).
///
/// # Errors
///
/// [`ProtocolError::NonFiniteMetric`] when either metric is NaN/±inf.
pub fn synthetic(
protocol: SplitProtocol,
metric_name: impl Into<String>,
model_metric: f64,
baseline_metric: f64,
) -> Result<Self, ProtocolError> {
Self::build(
protocol,
metric_name.into(),
model_metric,
baseline_metric,
EvidenceGrade::Synthetic,
)
}
/// Build a `CLAIMED` report (quoted, not reproduced here).
///
/// # Errors
///
/// [`ProtocolError::NonFiniteMetric`] when either metric is NaN/±inf.
pub fn claimed(
protocol: SplitProtocol,
metric_name: impl Into<String>,
model_metric: f64,
baseline_metric: f64,
) -> Result<Self, ProtocolError> {
Self::build(
protocol,
metric_name.into(),
model_metric,
baseline_metric,
EvidenceGrade::Claimed,
)
}
fn build(
protocol: SplitProtocol,
metric_name: String,
model_metric: f64,
baseline_metric: f64,
evidence: EvidenceGrade,
) -> Result<Self, ProtocolError> {
if !model_metric.is_finite() {
return Err(ProtocolError::NonFiniteMetric {
name: format!("{metric_name} (model)"),
value: model_metric,
});
}
if !baseline_metric.is_finite() {
return Err(ProtocolError::NonFiniteMetric {
name: format!("{metric_name} (baseline)"),
value: baseline_metric,
});
}
Ok(EvaluationReport {
protocol,
metric_name,
model_metric,
baseline_metric,
evidence,
})
}
/// `model baseline` (positive is better for higher-is-better metrics
/// such as PCK; interpret per metric).
#[must_use]
pub fn margin_over_baseline(&self) -> f64 {
self.model_metric - self.baseline_metric
}
/// One-line evidence-tagged summary, e.g.
/// `"[MEASURED] cross-subject pck@0.2: model 0.6100 vs mean-pose baseline 0.4100"`.
#[must_use]
pub fn summary(&self) -> String {
format!(
"[{}] {} {}: model {:.4} vs mean-pose baseline {:.4}",
self.evidence.tag(),
self.protocol.tag(),
self.metric_name,
self.model_metric,
self.baseline_metric
)
}
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
use approx::assert_abs_diff_eq;
use ndarray::array;
fn meta(
subject: u32,
environment: u32,
orientation: u32,
recording: u64,
window: u64,
) -> SampleMeta {
SampleMeta {
subject_id: subject,
environment_id: environment,
orientation_id: orientation,
gesture_id: 1,
recording_id: recording,
window_index: window,
}
}
// ----- LeakageAudit -----------------------------------------------------
#[test]
fn audit_passes_clean_cross_subject_split() {
let train = vec![meta(1, 1, 1, 0, 0), meta(1, 1, 1, 0, 1), meta(2, 1, 2, 1, 0)];
let test = vec![meta(3, 1, 1, 2, 0), meta(3, 1, 1, 2, 1)];
let pass = LeakageAudit::for_protocol(SplitProtocol::CrossSubject)
.audit(&train, &test)
.expect("clean split must pass");
assert_eq!(pass.train_windows, 3);
assert_eq!(pass.test_windows, 2);
assert_eq!(pass.train_subjects, 2);
assert_eq!(pass.test_subjects, 1);
assert_eq!(pass.train_recordings, 2);
assert_eq!(pass.test_recordings, 1);
}
#[test]
fn audit_rejects_subject_overlap() {
let train = vec![meta(1, 1, 1, 0, 0), meta(2, 1, 1, 1, 0)];
let test = vec![meta(2, 2, 2, 2, 0)]; // subject 2 on both sides
let err = LeakageAudit::for_protocol(SplitProtocol::CrossSubject)
.audit(&train, &test)
.unwrap_err();
assert!(matches!(err, ProtocolError::SubjectOverlap { subject_id: 2 }));
}
#[test]
fn audit_rejects_environment_overlap_when_claimed() {
let train = vec![meta(1, 1, 1, 0, 0)];
let test = vec![meta(2, 1, 2, 1, 0)]; // environment 1 on both sides
let err = LeakageAudit::for_protocol(SplitProtocol::CrossEnvironment)
.audit(&train, &test)
.unwrap_err();
assert!(matches!(
err,
ProtocolError::EnvironmentOverlap { environment_id: 1 }
));
// The same split passes a protocol that does not claim env-disjointness.
assert!(LeakageAudit::for_protocol(SplitProtocol::CrossSubject)
.audit(&train, &test)
.is_ok());
}
#[test]
fn audit_rejects_orientation_overlap_when_claimed() {
let train = vec![meta(1, 1, 3, 0, 0)];
let test = vec![meta(2, 2, 3, 1, 0)];
let err = LeakageAudit::for_protocol(SplitProtocol::CrossOrientation)
.audit(&train, &test)
.unwrap_err();
assert!(matches!(
err,
ProtocolError::OrientationOverlap { orientation_id: 3 }
));
}
#[test]
fn audit_always_rejects_recording_crossing() {
// Even a protocol claiming nothing (RandomBaseline) must fail when a
// continuous recording straddles the boundary.
let train = vec![meta(1, 1, 1, 5, 0)];
let test = vec![meta(2, 2, 2, 5, 1)]; // same recording 5
let err = LeakageAudit::for_protocol(SplitProtocol::RandomBaseline)
.audit(&train, &test)
.unwrap_err();
assert!(matches!(
err,
ProtocolError::RecordingCrossesSplit { recording_id: 5 }
));
}
#[test]
fn audit_rejects_empty_partitions() {
let some = vec![meta(1, 1, 1, 0, 0)];
let audit = LeakageAudit::for_protocol(SplitProtocol::CrossSubject);
assert!(matches!(
audit.audit(&[], &some),
Err(ProtocolError::EmptyPartition { side: "train" })
));
assert!(matches!(
audit.audit(&some, &[]),
Err(ProtocolError::EmptyPartition { side: "test" })
));
}
#[test]
fn random_baseline_partition_fails_audit_end_to_end() {
// Wire a real RandomBaseline SplitPlan into the audit: the leaky
// window-level split must be rejected, which is exactly its purpose.
use crate::protocols::SplitPlan;
let mut metas = Vec::new();
for recording in 0..8u64 {
for window in 0..6u64 {
metas.push(meta(1 + (recording % 3) as u32, 1, 1, recording, window));
}
}
let plan = SplitPlan::new(SplitProtocol::RandomBaseline, 9, 0.5).unwrap();
let (train_idx, test_idx) = plan.partition(&metas);
let train: Vec<SampleMeta> = train_idx.iter().map(|&i| metas[i]).collect();
let test: Vec<SampleMeta> = test_idx.iter().map(|&i| metas[i]).collect();
assert!(matches!(
LeakageAudit::for_protocol(SplitProtocol::RandomBaseline).audit(&train, &test),
Err(ProtocolError::RecordingCrossesSplit { .. })
));
}
// ----- MeanPoseBaseline -------------------------------------------------
#[test]
fn mean_pose_is_elementwise_mean_of_training_poses() {
let train = vec![
array![[0.0f32, 0.0], [1.0, 1.0]],
array![[0.2f32, 0.4], [0.6, 0.0]],
];
let baseline = MeanPoseBaseline::fit(&train).unwrap();
assert_eq!(baseline.num_train_poses(), 2);
let mean = baseline.mean_pose();
assert_abs_diff_eq!(mean[[0, 0]], 0.1, epsilon = 1e-6);
assert_abs_diff_eq!(mean[[0, 1]], 0.2, epsilon = 1e-6);
assert_abs_diff_eq!(mean[[1, 0]], 0.8, epsilon = 1e-6);
assert_abs_diff_eq!(mean[[1, 1]], 0.5, epsilon = 1e-6);
}
#[test]
fn mean_pose_mpjpe_math() {
// Baseline fitted on a single pose ⇒ mean equals it exactly.
let train = vec![array![[0.0f32, 0.0], [1.0, 0.0]]];
let baseline = MeanPoseBaseline::fit(&train).unwrap();
// Test pose offset by (0.3, 0.4) on both joints ⇒ distance 0.5 each.
let test = vec![array![[0.3f32, 0.4], [1.3, 0.4]]];
let mpjpe = baseline.mpjpe(&test).unwrap();
assert_abs_diff_eq!(mpjpe, 0.5, epsilon = 1e-6);
// Distances are ~0.5 up to f32 rounding, so probe strictly either
// side: PCK@0.6 counts both joints, PCK@0.49 counts neither.
assert_abs_diff_eq!(baseline.pck_at(&test, 0.6).unwrap(), 1.0, epsilon = 1e-9);
assert_abs_diff_eq!(baseline.pck_at(&test, 0.49).unwrap(), 0.0, epsilon = 1e-9);
}
#[test]
fn mean_pose_rejects_empty_and_mismatched() {
assert!(matches!(
MeanPoseBaseline::fit(&[]),
Err(ProtocolError::EmptyTrainingPoses)
));
let train = vec![
array![[0.0f32, 0.0], [1.0, 1.0]],
array![[0.0f32, 0.0]], // 1 joint vs 2
];
assert!(matches!(
MeanPoseBaseline::fit(&train),
Err(ProtocolError::PoseShapeMismatch { .. })
));
let baseline = MeanPoseBaseline::fit(&[array![[0.0f32, 0.0]]]).unwrap();
assert!(baseline.mpjpe(&[]).is_err());
assert!(baseline
.mpjpe(&[array![[0.0f32, 0.0], [1.0, 1.0]]])
.is_err());
}
// ----- EvaluationReport -------------------------------------------------
#[test]
fn measured_requires_reproducer() {
let err = EvaluationReport::measured(
SplitProtocol::CrossSubject,
"pck@0.2",
0.61,
0.41,
" ",
)
.unwrap_err();
assert!(matches!(err, ProtocolError::MissingReproducer));
let report = EvaluationReport::measured(
SplitProtocol::CrossSubject,
"pck@0.2",
0.61,
0.41,
"cargo run -p wifi-densepose-train --bin train -- eval --protocol cross-subject --seed 42",
)
.unwrap();
assert_eq!(report.evidence.tag(), "MEASURED");
assert_abs_diff_eq!(report.margin_over_baseline(), 0.2, epsilon = 1e-9);
assert!(report.summary().starts_with("[MEASURED] cross-subject pck@0.2"));
}
#[test]
fn synthetic_and_claimed_tags() {
let s =
EvaluationReport::synthetic(SplitProtocol::CrossOrientation, "mpjpe", 0.1, 0.3)
.unwrap();
assert_eq!(s.evidence.tag(), "SYNTHETIC");
let c = EvaluationReport::claimed(SplitProtocol::CrossSubject, "pck@0.5", 0.9, 0.5)
.unwrap();
assert_eq!(c.evidence.tag(), "CLAIMED");
}
#[test]
fn report_rejects_non_finite_metrics() {
assert!(matches!(
EvaluationReport::synthetic(SplitProtocol::CrossSubject, "pck", f64::NAN, 0.5),
Err(ProtocolError::NonFiniteMetric { .. })
));
assert!(matches!(
EvaluationReport::synthetic(SplitProtocol::CrossSubject, "pck", 0.5, f64::INFINITY),
Err(ProtocolError::NonFiniteMetric { .. })
));
}
#[test]
fn report_serializes_roundtrip() {
let report = EvaluationReport::measured(
SplitProtocol::CrossEnvironment,
"mpjpe",
0.07,
0.19,
"cargo test -p wifi-densepose-train",
)
.unwrap();
let json = serde_json::to_string(&report).unwrap();
let back: EvaluationReport = serde_json::from_str(&json).unwrap();
assert_eq!(back, report);
}
}
@@ -23,6 +23,10 @@ criterion = { version = "0.5", features = ["html_reports"] }
name = "vitals_bench"
harness = false
[[bench]]
name = "groundtruth_bench"
harness = false
[features]
default = ["serde"]
serde = ["dep:serde"]
@@ -0,0 +1,136 @@
//! Benchmark for ground-truth time alignment (ADR-290).
//!
//! Aligns an hour-scale synthetic session (3600 s) against a reference
//! series with a known 12 s clock offset, over the default ±30 s lag
//! window. Variants: 1 Hz estimate vs 1 Hz reference (same-rate), 0.5 Hz
//! estimate vs 1 Hz reference (rate-mismatched, the realistic CSI case),
//! and same-rate with the windowed drift fit enabled. The lag search is
//! O(lags × grid points) with no per-lag allocation; this bench tracks that
//! cost at realistic session length. All input is generated in code and
//! fully deterministic; measurement time is kept short deliberately.
//!
//! Reproduce:
//! cargo bench -p wifi-densepose-vitals --bench groundtruth_bench
//! Compile-only:
//! cargo bench -p wifi-densepose-vitals --bench groundtruth_bench --no-run
use criterion::{black_box, criterion_group, criterion_main, Criterion};
use std::time::Duration;
use wifi_densepose_vitals::groundtruth::{
align, AlignmentConfig, EstimateSeries, Measurand, MeasurementPrinciple, ReferenceDevice,
ReferenceSample, ReferenceSeries,
};
/// Session length in seconds (one hour).
const SESSION_SECS: usize = 3600;
/// Known clock offset injected into the estimate series, milliseconds.
const OFFSET_MS: i64 = 12_000;
/// Deterministic aperiodic heart-rate-like signal (incommensurate periods).
fn synth(t_secs: f64) -> f64 {
70.0 + 5.0 * (2.0 * std::f64::consts::PI * t_secs / 47.0).sin()
+ 3.0 * (2.0 * std::f64::consts::PI * t_secs / 113.0).sin()
}
/// Reference series: `SESSION_SECS` samples at 1 Hz on the reference clock.
fn reference_1hz() -> ReferenceSeries {
ReferenceSeries::new(
Measurand::HeartRateBpm,
ReferenceDevice {
make: "Synthetic".to_string(),
model: "bench".to_string(),
principle: MeasurementPrinciple::Other,
},
(0..SESSION_SECS)
.map(|i| ReferenceSample {
timestamp_ms: (i as i64) * 1000,
value: synth(i as f64),
})
.collect(),
)
.expect("valid reference")
}
/// Estimate series at `period_ms` sampling, shifted `OFFSET_MS` earlier.
fn estimate(period_ms: i64) -> EstimateSeries {
let n = (SESSION_SECS as i64 * 1000) / period_ms;
EstimateSeries::new(
Measurand::HeartRateBpm,
(0..n)
.map(|i| {
let t_ms = i * period_ms;
ReferenceSample {
timestamp_ms: t_ms - OFFSET_MS,
value: synth(t_ms as f64 / 1000.0),
}
})
.collect(),
)
.expect("valid estimate")
}
fn bench_align_hour_session(c: &mut Criterion) {
let reference = reference_1hz();
let est_1hz = estimate(1000);
let est_half_hz = estimate(2000);
let cfg = AlignmentConfig::default();
// 1 Hz estimate vs 1 Hz reference: exact offset recovery expected.
c.bench_function("groundtruth_align_1h_est1hz_ref1hz_pm30s", |b| {
b.iter(|| {
let result = align(black_box(&est_1hz), black_box(&reference), black_box(&cfg))
.expect("alignment succeeds");
assert_eq!(result.offset_ms, OFFSET_MS);
black_box(result);
});
});
// 0.5 Hz estimate vs 1 Hz reference: the realistic CSI-pipeline case.
// Nearest-sample resampling quantizes, so allow one grid step of slack.
c.bench_function("groundtruth_align_1h_est0p5hz_ref1hz_pm30s", |b| {
b.iter(|| {
let result = align(
black_box(&est_half_hz),
black_box(&reference),
black_box(&cfg),
)
.expect("alignment succeeds");
assert!((result.offset_ms - OFFSET_MS).abs() <= cfg.grid_step_ms);
black_box(result);
});
});
// Same-rate alignment with the windowed linear drift fit enabled.
let cfg_drift = AlignmentConfig {
fit_drift: true,
..AlignmentConfig::default()
};
c.bench_function("groundtruth_align_1h_est1hz_ref1hz_pm30s_drift", |b| {
b.iter(|| {
let result = align(
black_box(&est_1hz),
black_box(&reference),
black_box(&cfg_drift),
)
.expect("alignment succeeds");
black_box(result);
});
});
}
/// Short measurement window: each iteration is an hour-scale alignment, so
/// default criterion settings would make the suite needlessly slow.
fn short_config() -> Criterion {
Criterion::default()
.warm_up_time(Duration::from_millis(500))
.measurement_time(Duration::from_secs(3))
.sample_size(10)
}
criterion_group! {
name = benches;
config = short_config();
targets = bench_align_hour_session
}
criterion_main!(benches);
File diff suppressed because it is too large Load Diff
@@ -23,6 +23,12 @@
//! Results are stored in a [`VitalSignStore`] with configurable
//! retention for historical analysis.
//!
//! Ground-truth evaluation ([`groundtruth`], ADR-290) ingests a
//! reference-device series (CSV export), time-aligns it against a store
//! session, and produces evidence-graded agreement statistics
//! (MAE/RMSE/bias, Bland-Altman limits, percent-within-tolerance) with a
//! mandatory session scope.
//!
//! # Example
//!
//! ```
@@ -67,6 +73,7 @@
pub mod anomaly;
pub mod breathing;
pub mod groundtruth;
pub mod heartrate;
pub mod preprocessor;
pub mod store;
@@ -74,6 +81,12 @@ pub mod types;
pub use anomaly::{AnomalyAlert, VitalAnomalyDetector};
pub use breathing::BreathingExtractor;
pub use groundtruth::{
align, evaluate_session, AgreementConfig, AgreementReport, AlignmentConfig, AlignmentResult,
DistanceBand, DriftFit, EstimateSeries, EvidenceGrade, GradedAgreementReport,
GroundTruthError, Measurand, MeasurementPrinciple, MotionState, Propagation, ReferenceDevice,
ReferenceSample, ReferenceSeries, SessionEvaluation, SessionScope,
};
pub use heartrate::HeartRateExtractor;
pub use preprocessor::CsiVitalPreprocessor;
pub use store::{VitalSignStore, VitalStats};