#!/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()