dearsky/wifi-densepose
1
0
Fork 0
Code Issues 20 Pull Requests 5 Actions Packages Projects Releases 1 Wiki Activity

feat: Add commodity sensing, proof bundle, Three.js viz, mock isolation

Browse Source 
Commodity Sensing Module (ADR-013):
- sensing/rssi_collector.py: Real Linux WiFi RSSI collection from
  /proc/net/wireless and iw commands, with SimulatedCollector for testing
- sensing/feature_extractor.py: FFT-based spectral analysis, CUSUM
  change-point detection, breathing/motion band power extraction
- sensing/classifier.py: Rule-based presence/motion classification
  with confidence scoring and multi-receiver agreement
- sensing/backend.py: Common SensingBackend protocol with honest
  capability reporting (PRESENCE + MOTION only for commodity)

Proof of Reality Bundle (ADR-011):
- data/proof/generate_reference_signal.py: Deterministic synthetic CSI
  with known breathing (0.3 Hz) and walking (1.2 Hz) signals
- data/proof/sample_csi_data.json: Generated reference signal
- data/proof/verify.py: One-command pipeline verification with SHA-256
- data/proof/expected_features.sha256: Expected output hash

Three.js Visualization:
- ui/components/scene.js: 3D scene setup with OrbitControls

Mock Isolation:
- testing/mock_pose_generator.py: Mock pose generation moved out of
  production pose_service.py
- services/pose_service.py: Cleaned mock paths

https://claude.ai/code/session_01Ki7pvEZtJDvqJkmyn6B714
This commit is contained in:
Claude
2026-02-28 06:18:58 +00:00
parent e3f0c7a3fa
commit 2199174cac
12 changed files with 358561 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
v1/data/proof/expected_features.sha256 Normal file
View File

@@ -0,0 +1 @@
7b9ed15a01a2ae49cb32c5a1bb7e41361e0c83d9216f092efe3a3e279c7731ba

324
v1/data/proof/generate_reference_signal.py Normal file
View File

@@ -0,0 +1,324 @@
#!/usr/bin/env python3
"""
Deterministic Reference CSI Signal Generator for WiFi-DensePose Proof Bundle.
This script generates a SYNTHETIC, DETERMINISTIC CSI (Channel State Information)
reference signal for pipeline verification. It is NOT a real WiFi capture.
The signal models a 3-antenna, 56-subcarrier WiFi system with:
- Human breathing modulation at 0.3 Hz
- Walking motion modulation at 1.2 Hz
- Structured (deterministic) multipath propagation with known delays
- 10 seconds of data at 100 Hz sampling rate (1000 frames total)
Generation Formula
==================
For each frame t (t = 0..999) at time s = t / 100.0:
CSI[antenna_a, subcarrier_k] = sum over P paths of:
A_p * exp(j * (2*pi*f_k*tau_p + phi_p,a))
* (1 + alpha_breathe * sin(2*pi * 0.3 * s + psi_breathe_a))
* (1 + alpha_walk * sin(2*pi * 1.2 * s + psi_walk_a))
Where:
- f_k = center_freq + (k - 28) * subcarrier_spacing [subcarrier frequency]
- tau_p = deterministic path delay for path p
- A_p = deterministic path amplitude for path p
- phi_p,a = deterministic phase offset per path per antenna
- alpha_breathe = 0.02 (breathing modulation depth)
- alpha_walk = 0.08 (walking modulation depth)
- psi_breathe_a, psi_walk_a = deterministic per-antenna phase offsets
All parameters are computed from numpy with seed=42. No randomness is used
at generation time -- the seed is used ONLY to select fixed parameter values
once, which are then documented in the metadata file.
Output:
- sample_csi_data.json: All 1000 CSI frames with amplitude and phase arrays
- sample_csi_meta.json: Complete parameter documentation
Author: WiFi-DensePose Project (synthetic test data)
"""
import json
import os
import sys
import numpy as np
def generate_deterministic_parameters():
"""Generate all fixed parameters using seed=42.
These parameters define the multipath channel model and human motion
modulation. Once generated, they are constants -- no further randomness
is used.
Returns:
dict: All channel and motion parameters.
"""
rng = np.random.RandomState(42)
# System parameters (fixed by design, not random)
num_antennas = 3
num_subcarriers = 56
sampling_rate_hz = 100
duration_s = 10.0
center_freq_hz = 5.21e9 # WiFi 5 GHz channel 42
subcarrier_spacing_hz = 312.5e3 # Standard 802.11n/ac
# Multipath channel: 5 deterministic paths
num_paths = 5
# Path delays in nanoseconds (typical indoor)
path_delays_ns = np.array([0.0, 15.0, 42.0, 78.0, 120.0])
# Path amplitudes (linear scale, decreasing with delay)
path_amplitudes = np.array([1.0, 0.6, 0.35, 0.18, 0.08])
# Phase offsets per path per antenna (from seed=42, then fixed)
path_phase_offsets = rng.uniform(-np.pi, np.pi, size=(num_paths, num_antennas))
# Human motion modulation parameters
breathing_freq_hz = 0.3
walking_freq_hz = 1.2
breathing_depth = 0.02 # 2% amplitude modulation
walking_depth = 0.08 # 8% amplitude modulation
# Per-antenna phase offsets for motion signals (from seed=42, then fixed)
breathing_phase_offsets = rng.uniform(0, 2 * np.pi, size=num_antennas)
walking_phase_offsets = rng.uniform(0, 2 * np.pi, size=num_antennas)
return {
"num_antennas": num_antennas,
"num_subcarriers": num_subcarriers,
"sampling_rate_hz": sampling_rate_hz,
"duration_s": duration_s,
"center_freq_hz": center_freq_hz,
"subcarrier_spacing_hz": subcarrier_spacing_hz,
"num_paths": num_paths,
"path_delays_ns": path_delays_ns,
"path_amplitudes": path_amplitudes,
"path_phase_offsets": path_phase_offsets,
"breathing_freq_hz": breathing_freq_hz,
"walking_freq_hz": walking_freq_hz,
"breathing_depth": breathing_depth,
"walking_depth": walking_depth,
"breathing_phase_offsets": breathing_phase_offsets,
"walking_phase_offsets": walking_phase_offsets,
}
def generate_csi_frames(params):
"""Generate all CSI frames deterministically from the given parameters.
Args:
params: Dictionary of channel/motion parameters.
Returns:
list: List of dicts, each containing amplitude and phase arrays
for one frame, plus timestamp.
"""
num_antennas = params["num_antennas"]
num_subcarriers = params["num_subcarriers"]
sampling_rate = params["sampling_rate_hz"]
duration = params["duration_s"]
center_freq = params["center_freq_hz"]
subcarrier_spacing = params["subcarrier_spacing_hz"]
num_paths = params["num_paths"]
path_delays_ns = params["path_delays_ns"]
path_amplitudes = params["path_amplitudes"]
path_phase_offsets = params["path_phase_offsets"]
breathing_freq = params["breathing_freq_hz"]
walking_freq = params["walking_freq_hz"]
breathing_depth = params["breathing_depth"]
walking_depth = params["walking_depth"]
breathing_phase = params["breathing_phase_offsets"]
walking_phase = params["walking_phase_offsets"]
num_frames = int(duration * sampling_rate)
# Precompute subcarrier frequencies relative to center
k_indices = np.arange(num_subcarriers) - num_subcarriers // 2
subcarrier_freqs = center_freq + k_indices * subcarrier_spacing
# Convert path delays to seconds
path_delays_s = path_delays_ns * 1e-9
frames = []
for frame_idx in range(num_frames):
t = frame_idx / sampling_rate
# Build complex CSI matrix: (num_antennas, num_subcarriers)
csi_complex = np.zeros((num_antennas, num_subcarriers), dtype=complex)
for a in range(num_antennas):
# Human motion modulation for this antenna at this time
breathing_mod = 1.0 + breathing_depth * np.sin(
2.0 * np.pi * breathing_freq * t + breathing_phase[a]
)
walking_mod = 1.0 + walking_depth * np.sin(
2.0 * np.pi * walking_freq * t + walking_phase[a]
)
motion_factor = breathing_mod * walking_mod
for p in range(num_paths):
# Phase shift from path delay across subcarriers
phase_from_delay = 2.0 * np.pi * subcarrier_freqs * path_delays_s[p]
# Add per-path per-antenna offset
total_phase = phase_from_delay + path_phase_offsets[p, a]
# Accumulate path contribution
csi_complex[a, :] += (
path_amplitudes[p] * motion_factor * np.exp(1j * total_phase)
)
amplitude = np.abs(csi_complex)
phase = np.angle(csi_complex) # in [-pi, pi]
frames.append({
"frame_index": frame_idx,
"timestamp_s": round(t, 4),
"amplitude": amplitude.tolist(),
"phase": phase.tolist(),
})
return frames
def save_data(frames, params, output_dir):
"""Save CSI frames and metadata to JSON files.
Args:
frames: List of CSI frame dicts.
params: Generation parameters.
output_dir: Directory to write output files.
"""
# Save CSI data
csi_data = {
"description": (
"SYNTHETIC deterministic CSI reference signal for pipeline verification. "
"This is NOT a real WiFi capture. Generated mathematically with known "
"parameters for reproducibility testing."
),
"generator": "generate_reference_signal.py",
"generator_version": "1.0.0",
"numpy_seed": 42,
"num_frames": len(frames),
"num_antennas": params["num_antennas"],
"num_subcarriers": params["num_subcarriers"],
"sampling_rate_hz": params["sampling_rate_hz"],
"frequency_hz": params["center_freq_hz"],
"bandwidth_hz": params["subcarrier_spacing_hz"] * params["num_subcarriers"],
"frames": frames,
}
data_path = os.path.join(output_dir, "sample_csi_data.json")
with open(data_path, "w") as f:
json.dump(csi_data, f, indent=2)
print(f"Wrote {len(frames)} frames to {data_path}")
# Save metadata
meta = {
"description": (
"Metadata for the SYNTHETIC deterministic CSI reference signal. "
"Documents all generation parameters so the signal can be independently "
"reproduced and verified."
),
"is_synthetic": True,
"is_real_capture": False,
"generator_script": "generate_reference_signal.py",
"numpy_seed": 42,
"system_parameters": {
"num_antennas": params["num_antennas"],
"num_subcarriers": params["num_subcarriers"],
"sampling_rate_hz": params["sampling_rate_hz"],
"duration_s": params["duration_s"],
"center_frequency_hz": params["center_freq_hz"],
"subcarrier_spacing_hz": params["subcarrier_spacing_hz"],
"total_frames": int(params["duration_s"] * params["sampling_rate_hz"]),
},
"multipath_channel": {
"num_paths": params["num_paths"],
"path_delays_ns": params["path_delays_ns"].tolist(),
"path_amplitudes": params["path_amplitudes"].tolist(),
"path_phase_offsets_rad": params["path_phase_offsets"].tolist(),
"description": (
"5-path indoor multipath model with deterministic delays and "
"amplitudes. Path amplitudes decrease with delay (typical indoor)."
),
},
"human_motion_signals": {
"breathing": {
"frequency_hz": params["breathing_freq_hz"],
"modulation_depth": params["breathing_depth"],
"per_antenna_phase_offsets_rad": params["breathing_phase_offsets"].tolist(),
"description": (
"Sinusoidal amplitude modulation at 0.3 Hz modeling human "
"breathing (typical adult resting rate: 12-20 breaths/min = 0.2-0.33 Hz)."
),
},
"walking": {
"frequency_hz": params["walking_freq_hz"],
"modulation_depth": params["walking_depth"],
"per_antenna_phase_offsets_rad": params["walking_phase_offsets"].tolist(),
"description": (
"Sinusoidal amplitude modulation at 1.2 Hz modeling human "
"walking motion (typical stride rate: ~1.0-1.4 Hz)."
),
},
},
"generation_formula": (
"CSI[a,k,t] = sum_p { A_p * exp(j*(2*pi*f_k*tau_p + phi_{p,a})) "
"* (1 + d_breathe * sin(2*pi*0.3*t + psi_breathe_a)) "
"* (1 + d_walk * sin(2*pi*1.2*t + psi_walk_a)) }"
),
"determinism_guarantee": (
"All parameters are derived from numpy.random.RandomState(42) at "
"script initialization. The generation loop itself uses NO randomness. "
"Running this script on any platform with the same numpy version will "
"produce bit-identical output."
),
}
meta_path = os.path.join(output_dir, "sample_csi_meta.json")
with open(meta_path, "w") as f:
json.dump(meta, f, indent=2)
print(f"Wrote metadata to {meta_path}")
def main():
"""Main entry point."""
# Determine output directory
output_dir = os.path.dirname(os.path.abspath(__file__))
print("=" * 70)
print("WiFi-DensePose: Deterministic Reference CSI Signal Generator")
print("=" * 70)
print(f"Output directory: {output_dir}")
print()
# Step 1: Generate deterministic parameters
print("[1/3] Generating deterministic channel parameters (seed=42)...")
params = generate_deterministic_parameters()
print(f" - {params['num_paths']} multipath paths")
print(f" - {params['num_antennas']} antennas, {params['num_subcarriers']} subcarriers")
print(f" - Breathing: {params['breathing_freq_hz']} Hz, depth={params['breathing_depth']}")
print(f" - Walking: {params['walking_freq_hz']} Hz, depth={params['walking_depth']}")
print()
# Step 2: Generate all frames
num_frames = int(params["duration_s"] * params["sampling_rate_hz"])
print(f"[2/3] Generating {num_frames} CSI frames...")
print(f" - Duration: {params['duration_s']}s at {params['sampling_rate_hz']} Hz")
frames = generate_csi_frames(params)
print(f" - Generated {len(frames)} frames")
print()
# Step 3: Save output
print("[3/3] Saving output files...")
save_data(frames, params, output_dir)
print()
print("Done. Reference signal generated successfully.")
print("=" * 70)
if __name__ == "__main__":
main()

356014
v1/data/proof/sample_csi_data.json Normal file
View File

File diff suppressed because it is too large Load Diff

85
v1/data/proof/sample_csi_meta.json Normal file
View File

@@ -0,0 +1,85 @@
{
"description": "Metadata for the SYNTHETIC deterministic CSI reference signal. Documents all generation parameters so the signal can be independently reproduced and verified.",
"is_synthetic": true,
"is_real_capture": false,
"generator_script": "generate_reference_signal.py",
"numpy_seed": 42,
"system_parameters": {
"num_antennas": 3,
"num_subcarriers": 56,
"sampling_rate_hz": 100,
"duration_s": 10.0,
"center_frequency_hz": 5210000000.0,
"subcarrier_spacing_hz": 312500.0,
"total_frames": 1000
},
"multipath_channel": {
"num_paths": 5,
"path_delays_ns": [
0.0,
15.0,
42.0,
78.0,
120.0
],
"path_amplitudes": [
1.0,
0.6,
0.35,
0.18,
0.08
],
"path_phase_offsets_rad": [
[
-0.788287681898749,
2.8319215077704234,
1.4576609265440963
],
[
0.6198895383354297,
-2.1612986243157413,
-2.1614501754128375
],
[
-2.776642555026645,
2.3007525789727232,
0.6353243561202211
],
[
1.3073585636350948,
-3.012256461474685,
2.952530678803174
],
[
2.088798716157191,
-1.8074266732364683,
-1.9991526911557285
]
],
"description": "5-path indoor multipath model with deterministic delays and amplitudes. Path amplitudes decrease with delay (typical indoor)."
},
"human_motion_signals": {
"breathing": {
"frequency_hz": 0.3,
"modulation_depth": 0.02,
"per_antenna_phase_offsets_rad": [
1.152364521581569,
1.9116103907867292,
3.297141901079666
],
"description": "Sinusoidal amplitude modulation at 0.3 Hz modeling human breathing (typical adult resting rate: 12-20 breaths/min = 0.2-0.33 Hz)."
},
"walking": {
"frequency_hz": 1.2,
"modulation_depth": 0.08,
"per_antenna_phase_offsets_rad": [
2.713990594641554,
1.8298466547148808,
3.844385118274953
],
"description": "Sinusoidal amplitude modulation at 1.2 Hz modeling human walking motion (typical stride rate: ~1.0-1.4 Hz)."
}
},
"generation_formula": "CSI[a,k,t] = sum_p { A_p * exp(j*(2*pi*f_k*tau_p + phi_{p,a})) * (1 + d_breathe * sin(2*pi*0.3*t + psi_breathe_a)) * (1 + d_walk * sin(2*pi*1.2*t + psi_walk_a)) }",
"determinism_guarantee": "All parameters are derived from numpy.random.RandomState(42) at script initialization. The generation loop itself uses NO randomness. Running this script on any platform with the same numpy version will produce bit-identical output."
}

263
v1/data/proof/verify.py Normal file
View File

@@ -0,0 +1,263 @@
#!/usr/bin/env python3
"""
Proof-of-Reality Verification Script for WiFi-DensePose Pipeline.
This script verifies that the signal processing pipeline produces
DETERMINISTIC, REPRODUCIBLE output from a known reference signal.
Steps:
1. Load the synthetic reference CSI signal from sample_csi_data.json
2. Feed each frame through the actual CSI processor feature extraction
3. Collect all feature outputs into a canonical byte representation
4. Compute SHA-256 hash of the full feature output
5. Compare against the expected hash in expected_features.sha256
6. Print PASS or FAIL
The reference signal is SYNTHETIC (generated by generate_reference_signal.py)
and is used purely for pipeline determinism verification.
Usage:
python verify.py # Run verification against stored hash
python verify.py --generate-hash # Generate and print the expected hash
"""
import hashlib
import json
import os
import struct
import sys
import argparse
from datetime import datetime, timezone
import numpy as np
# Add the v1 directory to sys.path so we can import the actual modules
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
V1_DIR = os.path.abspath(os.path.join(SCRIPT_DIR, "..", "..")) # v1/data/proof -> v1/
if V1_DIR not in sys.path:
sys.path.insert(0, V1_DIR)
# Import the actual pipeline modules
from src.hardware.csi_extractor import CSIData
from src.core.csi_processor import CSIProcessor, CSIFeatures
# -- Configuration for the CSI processor (matches production defaults) --
PROCESSOR_CONFIG = {
"sampling_rate": 100,
"window_size": 56,
"overlap": 0.5,
"noise_threshold": -60,
"human_detection_threshold": 0.8,
"smoothing_factor": 0.9,
"max_history_size": 500,
"enable_preprocessing": True,
"enable_feature_extraction": True,
"enable_human_detection": True,
}
# Number of frames to process for the feature hash
# We process a representative subset to keep verification fast while
# still covering temporal dynamics (Doppler requires history)
VERIFICATION_FRAME_COUNT = 100 # First 100 frames = 1 second
def load_reference_signal(data_path):
"""Load the reference CSI signal from JSON.
Args:
data_path: Path to sample_csi_data.json.
Returns:
dict: Parsed JSON data.
Raises:
FileNotFoundError: If the data file doesn't exist.
json.JSONDecodeError: If the data is malformed.
"""
with open(data_path, "r") as f:
data = json.load(f)
return data
def frame_to_csi_data(frame, signal_meta):
"""Convert a JSON frame dict into a CSIData dataclass instance.
Args:
frame: Dict with 'amplitude', 'phase', 'timestamp_s', 'frame_index'.
signal_meta: Top-level signal metadata (num_antennas, frequency, etc).
Returns:
CSIData instance.
"""
amplitude = np.array(frame["amplitude"], dtype=np.float64)
phase = np.array(frame["phase"], dtype=np.float64)
timestamp = datetime.fromtimestamp(frame["timestamp_s"], tz=timezone.utc)
return CSIData(
timestamp=timestamp,
amplitude=amplitude,
phase=phase,
frequency=signal_meta["frequency_hz"],
bandwidth=signal_meta["bandwidth_hz"],
num_subcarriers=signal_meta["num_subcarriers"],
num_antennas=signal_meta["num_antennas"],
snr=15.0, # Fixed SNR for synthetic signal
metadata={
"source": "synthetic_reference",
"frame_index": frame["frame_index"],
},
)
def features_to_bytes(features):
"""Convert CSIFeatures to a deterministic byte representation.
We serialize each numpy array to bytes in a canonical order
using little-endian float64 representation. This ensures the
hash is platform-independent for IEEE 754 compliant systems.
Args:
features: CSIFeatures instance.
Returns:
bytes: Canonical byte representation.
"""
parts = []
# Serialize each feature array in declaration order
for array in [
features.amplitude_mean,
features.amplitude_variance,
features.phase_difference,
features.correlation_matrix,
features.doppler_shift,
features.power_spectral_density,
]:
flat = np.asarray(array, dtype=np.float64).ravel()
# Pack as little-endian double (8 bytes each)
parts.append(struct.pack(f"<{len(flat)}d", *flat))
return b"".join(parts)
def compute_pipeline_hash(data_path):
"""Run the full pipeline and compute the SHA-256 hash of all features.
Args:
data_path: Path to sample_csi_data.json.
Returns:
str: Hex-encoded SHA-256 hash of the feature output.
"""
# Load reference signal
signal_data = load_reference_signal(data_path)
frames = signal_data["frames"][:VERIFICATION_FRAME_COUNT]
# Create processor
processor = CSIProcessor(PROCESSOR_CONFIG)
# Process all frames and accumulate feature bytes
hasher = hashlib.sha256()
features_count = 0
for frame in frames:
csi_data = frame_to_csi_data(frame, signal_data)
# Run through the actual pipeline: preprocess -> extract features
preprocessed = processor.preprocess_csi_data(csi_data)
features = processor.extract_features(preprocessed)
if features is not None:
feature_bytes = features_to_bytes(features)
hasher.update(feature_bytes)
features_count += 1
# Add to history for Doppler computation in subsequent frames
processor.add_to_history(csi_data)
print(f" Processed {features_count} frames through pipeline")
return hasher.hexdigest()
def main():
"""Main verification entry point."""
parser = argparse.ArgumentParser(
description="WiFi-DensePose pipeline verification"
)
parser.add_argument(
"--generate-hash",
action="store_true",
help="Generate and print the expected hash (do not verify)",
)
args = parser.parse_args()
print("=" * 70)
print("WiFi-DensePose: Pipeline Verification")
print("=" * 70)
print()
# Locate data file
data_path = os.path.join(SCRIPT_DIR, "sample_csi_data.json")
hash_path = os.path.join(SCRIPT_DIR, "expected_features.sha256")
if not os.path.exists(data_path):
print(f"FAIL: Reference data not found at {data_path}")
print(" Run generate_reference_signal.py first.")
sys.exit(1)
# Compute hash
print("[1/2] Processing reference signal through pipeline...")
computed_hash = compute_pipeline_hash(data_path)
print(f" SHA-256: {computed_hash}")
print()
if args.generate_hash:
# Write the hash file
with open(hash_path, "w") as f:
f.write(computed_hash + "\n")
print(f"[2/2] Wrote expected hash to {hash_path}")
print()
print("HASH GENERATED - run without --generate-hash to verify")
print("=" * 70)
return
# Verify against expected hash
print("[2/2] Verifying against expected hash...")
if not os.path.exists(hash_path):
print(f" WARNING: No expected hash file at {hash_path}")
print(f" Computed hash: {computed_hash}")
print()
print(" Run with --generate-hash to create the expected hash file.")
print()
print("SKIP (no expected hash to compare against)")
print("=" * 70)
sys.exit(2)
with open(hash_path, "r") as f:
expected_hash = f.read().strip()
print(f" Expected: {expected_hash}")
print(f" Computed: {computed_hash}")
print()
if computed_hash == expected_hash:
print("PASS - Pipeline output is deterministic and matches expected hash.")
print("=" * 70)
sys.exit(0)
else:
print("FAIL - Pipeline output does NOT match expected hash.")
print()
print("Possible causes:")
print(" - Numpy/scipy version mismatch (check requirements-lock.txt)")
print(" - Code change in CSI processor that alters numerical output")
print(" - Platform floating-point differences (unlikely for IEEE 754)")
print()
print("To update the expected hash after intentional changes:")
print(" python verify.py --generate-hash")
print("=" * 70)
sys.exit(1)
if __name__ == "__main__":
main()