wifi-ruview/archive/v1/data/proof/verify.py

#!/usr/bin/env python3
"""
Proof-of-Reality Verification Script for WiFi-DensePose Pipeline.

TRUST KILL SWITCH: A one-command proof replay that makes "it is mocked"
a falsifiable, measurable claim that fails against evidence.

This script verifies that the signal processing pipeline produces
DETERMINISTIC, REPRODUCIBLE output from a known reference signal.

Steps:
  1. Load the published 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 published 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. The point is not
that the signal is real -- the point is that the PIPELINE CODE is real.
The same code that processes this reference also processes live captures.

If someone claims "it is mocked":
  1. Run: ./verify
  2. If PASS: the pipeline code is the same code that produced the published hash
  3. If FAIL: something changed -- investigate

Usage:
  python verify.py                  # Run verification against stored hash
  python verify.py --verbose        # Show detailed feature statistics
  python verify.py --audit          # Scan codebase for mock/random patterns
  python verify.py --generate-hash  # Generate and print the expected hash
"""

import hashlib
import inspect
import json
import os
import struct
import sys
import argparse
import time
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 -- these are the PRODUCTION modules,
# not test doubles. The source paths are printed below for verification.
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 print_banner():
    """Print the verification banner."""
    print("=" * 72)
    print("  WiFi-DensePose: Trust Kill Switch -- Pipeline Proof Replay")
    print("=" * 72)
    print()
    print('  "If the public demo is a one-command replay that produces a matching')
    print('   hash from a published real capture, \'it is mocked\' becomes a')
    print('   measurable claim that fails."')
    print()


def print_source_provenance():
    """Print the actual source file paths used by this verification.

    This lets anyone confirm that the imported modules are the production
    code, not test doubles or mocks.
    """
    csi_processor_file = inspect.getfile(CSIProcessor)
    csi_data_file = inspect.getfile(CSIData)
    csi_features_file = inspect.getfile(CSIFeatures)

    print("  SOURCE PROVENANCE (verify these are production modules):")
    print(f"    CSIProcessor : {os.path.abspath(csi_processor_file)}")
    print(f"    CSIData      : {os.path.abspath(csi_data_file)}")
    print(f"    CSIFeatures  : {os.path.abspath(csi_features_file)}")
    print(f"    numpy        : {np.__file__}")
    print(f"    numpy version: {np.__version__}")

    try:
        import scipy
        print(f"    scipy        : {scipy.__file__}")
        print(f"    scipy version: {scipy.__version__}")
    except ImportError:
        print("    scipy        : NOT AVAILABLE")

    print()


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"],
        },
    )


# Quantization precision for cross-platform hash stability (issue #560).
#
# The bytes packed below feed SHA-256. Without quantization, the hash diverges
# across SIMD backends (Intel AVX2/AVX-512 vs ARM NEON vs different x86 micro-
# architectures in the same CI pool) because scipy.fft's pocketfft kernels
# reorder vectorized FP operations differently per build. IEEE 754 guarantees
# per-operation determinism, not associativity under reordering.
#
# Empirically: 9 decimals was NOT enough to collapse the divergence — two
# back-to-back Ubuntu 24.04 / Python 3.11 / scipy 1.17 CI runs landed on
# different Azure VM microarchitectures (likely Skylake vs Cascade Lake)
# and produced two different SHA-256s even after np.round(.., 9). The DSP
# pipeline (preprocess → biquad bandpass → FFT → PSD → variance accumulation)
# amplifies the ~1e-14 raw FFT divergence by several orders of magnitude
# downstream — the actual drift at features_to_bytes() input can reach 1e-7
# or worse.
#
# 6 decimals (parts per million) gives ~6 orders of magnitude headroom over
# observed pipeline-amplified ULP drift and is still far below any meaningful
# signal change (CSI phase precision is ~1e-3 rad; PSD bins differ by orders
# of magnitude). Round to this precision, then hash.
HASH_QUANTIZATION_DECIMALS = 6


def features_to_bytes(features):
    """Convert CSIFeatures to a deterministic byte representation.

    Each feature array is quantized to ``HASH_QUANTIZATION_DECIMALS`` decimal
    places before being packed as little-endian float64. The quantization is
    what makes the resulting SHA-256 hash actually platform-independent — the
    raw float values diverge at ULP precision across scipy.fft SIMD backends
    (issue #560), even though all platforms compute the "correct" answer.

    Args:
        features: CSIFeatures instance.

    Returns:
        bytes: Canonical, quantized 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()
        # Quantize before packing so SIMD-level FP reordering across
        # Intel AVX vs Apple Silicon NEON pocketfft kernels does not
        # leak into the SHA-256 input.
        flat = np.round(flat, HASH_QUANTIZATION_DECIMALS)
        # 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, verbose=False):
    """Run the full pipeline and compute the SHA-256 hash of all features.

    Args:
        data_path: Path to sample_csi_data.json.
        verbose: If True, print detailed feature statistics.

    Returns:
        tuple: (hex_hash, stats_dict) where stats_dict contains metrics.
    """
    # Load reference signal
    signal_data = load_reference_signal(data_path)
    frames = signal_data["frames"][:VERIFICATION_FRAME_COUNT]

    print(f"    Reference signal: {os.path.basename(data_path)}")
    print(f"    Signal description: {signal_data.get('description', 'N/A')}")
    print(f"    Generator: {signal_data.get('generator', 'N/A')} v{signal_data.get('generator_version', '?')}")
    print(f"    Numpy seed used: {signal_data.get('numpy_seed', 'N/A')}")
    print(f"    Total frames in file: {signal_data.get('num_frames', len(signal_data['frames']))}")
    print(f"    Frames to process: {len(frames)}")
    print(f"    Subcarriers: {signal_data.get('num_subcarriers', 'N/A')}")
    print(f"    Antennas: {signal_data.get('num_antennas', 'N/A')}")
    print(f"    Frequency: {signal_data.get('frequency_hz', 0) / 1e9:.3f} GHz")
    print(f"    Bandwidth: {signal_data.get('bandwidth_hz', 0) / 1e6:.1f} MHz")
    print(f"    Sampling rate: {signal_data.get('sampling_rate_hz', 'N/A')} Hz")
    print()

    # Create processor with production config
    print("    Configuring CSIProcessor with production parameters...")
    processor = CSIProcessor(PROCESSOR_CONFIG)
    print(f"    Window size: {processor.window_size}")
    print(f"    Overlap: {processor.overlap}")
    print(f"    Noise threshold: {processor.noise_threshold} dB")
    print(f"    Preprocessing: {'ENABLED' if processor.enable_preprocessing else 'DISABLED'}")
    print(f"    Feature extraction: {'ENABLED' if processor.enable_feature_extraction else 'DISABLED'}")
    print()

    # Process all frames and accumulate feature bytes
    hasher = hashlib.sha256()
    features_count = 0
    total_feature_bytes = 0
    last_features = None
    doppler_nonzero_count = 0
    doppler_shape = None
    psd_shape = None

    t_start = time.perf_counter()

    for i, frame in enumerate(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
            total_feature_bytes += len(feature_bytes)
            last_features = features

            # Track Doppler statistics
            doppler_shape = features.doppler_shift.shape
            doppler_nonzero_count = int(np.count_nonzero(features.doppler_shift))
            psd_shape = features.power_spectral_density.shape

        # Add to history for Doppler computation in subsequent frames
        processor.add_to_history(csi_data)

        if verbose and (i + 1) % 25 == 0:
            print(f"      ... processed frame {i + 1}/{len(frames)}")

    t_elapsed = time.perf_counter() - t_start

    print(f"    Processing complete.")
    print(f"    Frames processed: {len(frames)}")
    print(f"    Feature vectors extracted: {features_count}")
    print(f"    Total feature bytes hashed: {total_feature_bytes:,}")
    print(f"    Processing time: {t_elapsed:.4f}s ({len(frames) / t_elapsed:.0f} frames/sec)")
    print()

    # Print feature vector details
    if last_features is not None:
        print("    FEATURE VECTOR DETAILS (from last frame):")
        print(f"      amplitude_mean      : shape={last_features.amplitude_mean.shape}, "
              f"min={np.min(last_features.amplitude_mean):.6f}, "
              f"max={np.max(last_features.amplitude_mean):.6f}, "
              f"mean={np.mean(last_features.amplitude_mean):.6f}")
        print(f"      amplitude_variance   : shape={last_features.amplitude_variance.shape}, "
              f"min={np.min(last_features.amplitude_variance):.6f}, "
              f"max={np.max(last_features.amplitude_variance):.6f}")
        print(f"      phase_difference     : shape={last_features.phase_difference.shape}, "
              f"mean={np.mean(last_features.phase_difference):.6f}")
        print(f"      correlation_matrix   : shape={last_features.correlation_matrix.shape}")
        print(f"      doppler_shift        : shape={doppler_shape}, "
              f"non-zero bins={doppler_nonzero_count}/{doppler_shape[0] if doppler_shape else 0}")
        print(f"      power_spectral_density: shape={psd_shape}")
        print()

        if verbose:
            print("    DOPPLER SPECTRUM (proves real FFT, not random):")
            ds = last_features.doppler_shift
            print(f"      First 8 bins: {ds[:8]}")
            print(f"      Sum: {np.sum(ds):.6f}")
            print(f"      Max bin index: {np.argmax(ds)}")
            print(f"      Spectral entropy: {-np.sum(ds[ds > 0] * np.log2(ds[ds > 0] + 1e-15)):.4f}")
            print()

            print("    PSD DETAILS (proves scipy.fft, not random):")
            psd = last_features.power_spectral_density
            print(f"      First 8 bins: {psd[:8]}")
            print(f"      Total power: {np.sum(psd):.4f}")
            print(f"      Peak frequency bin: {np.argmax(psd)}")
            print()

    stats = {
        "frames_processed": len(frames),
        "features_extracted": features_count,
        "total_bytes_hashed": total_feature_bytes,
        "elapsed_seconds": t_elapsed,
        "doppler_shape": doppler_shape,
        "doppler_nonzero": doppler_nonzero_count,
        "psd_shape": psd_shape,
    }

    return hasher.hexdigest(), stats


def audit_codebase(base_dir=None):
    """Scan the production codebase for mock/random patterns.

    Looks for:
      - np.random.rand / np.random.randn calls (outside testing/)
      - mock/Mock imports (outside testing/)
      - random.random() calls (outside testing/)

    Args:
        base_dir: Root directory to scan. Defaults to v1/src/.

    Returns:
        list of (filepath, line_number, line_text, pattern_type) tuples.
    """
    if base_dir is None:
        base_dir = os.path.join(V1_DIR, "src")

    suspicious_patterns = [
        ("np.random.rand", "RANDOM_GENERATOR"),
        ("np.random.randn", "RANDOM_GENERATOR"),
        ("np.random.random", "RANDOM_GENERATOR"),
        ("np.random.uniform", "RANDOM_GENERATOR"),
        ("np.random.normal", "RANDOM_GENERATOR"),
        ("np.random.choice", "RANDOM_GENERATOR"),
        ("random.random(", "RANDOM_GENERATOR"),
        ("random.randint(", "RANDOM_GENERATOR"),
        ("from unittest.mock import", "MOCK_IMPORT"),
        ("from unittest import mock", "MOCK_IMPORT"),
        ("import mock", "MOCK_IMPORT"),
        ("MagicMock", "MOCK_USAGE"),
        ("@patch(", "MOCK_USAGE"),
        ("@mock.patch", "MOCK_USAGE"),
    ]

    # Directories to exclude from the audit
    excluded_dirs = {"testing", "tests", "test", "__pycache__", ".git"}

    findings = []

    for root, dirs, files in os.walk(base_dir):
        # Skip excluded directories
        dirs[:] = [d for d in dirs if d not in excluded_dirs]

        for fname in files:
            if not fname.endswith(".py"):
                continue

            fpath = os.path.join(root, fname)
            try:
                with open(fpath, "r", encoding="utf-8", errors="replace") as f:
                    for line_num, line in enumerate(f, 1):
                        for pattern, ptype in suspicious_patterns:
                            if pattern in line:
                                findings.append((fpath, line_num, line.rstrip(), ptype))
            except (IOError, OSError):
                pass

    return findings


def main():
    """Main verification entry point."""
    parser = argparse.ArgumentParser(
        description="WiFi-DensePose Trust Kill Switch -- Pipeline Proof Replay"
    )
    parser.add_argument(
        "--generate-hash",
        action="store_true",
        help="Generate and print the expected hash (do not verify)",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Show detailed feature statistics and Doppler spectrum",
    )
    parser.add_argument(
        "--audit",
        action="store_true",
        help="Scan production codebase for mock/random patterns",
    )
    args = parser.parse_args()

    print_banner()

    # Locate data file
    data_path = os.path.join(SCRIPT_DIR, "sample_csi_data.json")
    hash_path = os.path.join(SCRIPT_DIR, "expected_features.sha256")

    # ---------------------------------------------------------------
    # Step 0: Print source provenance
    # ---------------------------------------------------------------
    print("[0/4] SOURCE PROVENANCE")
    print_source_provenance()

    # ---------------------------------------------------------------
    # Step 1: Load and describe reference signal
    # ---------------------------------------------------------------
    print("[1/4] LOADING REFERENCE SIGNAL")
    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)
    print(f"    Path: {data_path}")
    print(f"    Size: {os.path.getsize(data_path):,} bytes")
    print()

    # ---------------------------------------------------------------
    # Step 2: Process through the real pipeline
    # ---------------------------------------------------------------
    print("[2/4] PROCESSING THROUGH PRODUCTION PIPELINE")
    print("    This runs the SAME CSIProcessor.preprocess_csi_data() and")
    print("    CSIProcessor.extract_features() used in production.")
    print()
    computed_hash, stats = compute_pipeline_hash(data_path, verbose=args.verbose)

    # ---------------------------------------------------------------
    # Step 3: Hash comparison
    # ---------------------------------------------------------------
    print("[3/4] SHA-256 HASH COMPARISON")
    print(f"    Computed: {computed_hash}")

    if args.generate_hash:
        with open(hash_path, "w") as f:
            f.write(computed_hash + "\n")
        print(f"    Wrote expected hash to {hash_path}")
        print()
        print("  HASH GENERATED -- run without --generate-hash to verify.")
        print("=" * 72)
        return

    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("=" * 72)
        sys.exit(2)

    with open(hash_path, "r") as f:
        expected_hash = f.read().strip()

    print(f"    Expected: {expected_hash}")

    if computed_hash == expected_hash:
        match_status = "MATCH"
    else:
        match_status = "MISMATCH"
    print(f"    Status:   {match_status}")
    print()

    # ---------------------------------------------------------------
    # Step 4: Audit (if requested or always in full mode)
    # ---------------------------------------------------------------
    if args.audit:
        print("[4/4] CODEBASE AUDIT -- scanning for mock/random patterns")
        findings = audit_codebase()
        if findings:
            print(f"    Found {len(findings)} suspicious pattern(s) in production code:")
            for fpath, line_num, line, ptype in findings:
                relpath = os.path.relpath(fpath, V1_DIR)
                print(f"      [{ptype}] {relpath}:{line_num}: {line.strip()}")
        else:
            print("    CLEAN -- no mock/random patterns found in production code.")
        print()
    else:
        print("[4/4] CODEBASE AUDIT (skipped -- use --audit to enable)")
        print()

    # ---------------------------------------------------------------
    # Final verdict
    # ---------------------------------------------------------------
    print("=" * 72)
    if computed_hash == expected_hash:
        print("  VERDICT: PASS")
        print()
        print("  The pipeline produced a SHA-256 hash that matches the published")
        print("  expected hash. This proves:")
        print("    1. The SAME signal processing code ran on the reference signal")
        print("    2. The output is DETERMINISTIC (same input -> same output)")
        print("    3. No randomness was introduced (hash would differ)")
        print("    4. The code path includes: noise removal, Hamming windowing,")
        print("       amplitude normalization, FFT-based Doppler extraction,")
        print("       and power spectral density computation")
        print()
        print(f"  Pipeline hash: {computed_hash}")
        print("=" * 72)
        sys.exit(0)
    else:
        print("  VERDICT: FAIL")
        print()
        print("  The pipeline output does NOT match the expected hash.")
        print()
        print("  Possible causes:")
        print("    - Numpy/scipy version mismatch (check requirements)")
        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("=" * 72)
        sys.exit(1)


if __name__ == "__main__":
    main()