How the CSI (Channel State Information) Processing Pipeline Works in WiFi DensePose

The CSI processing pipeline in WiFi DensePose extracts raw Channel State Information from Wi-Fi hardware, cleans and normalizes the data, extracts statistical and spectral features, and applies motion analysis to detect human presence with confidence scoring.

The WiFi DensePose system transforms raw Wi-Fi channel measurements into high-level human presence and pose information using a multi-stage CSI (Channel State Information) processing pipeline. Implemented in the ruvnet/wifi-densepose repository, this pipeline bridges low-level radio-frequency hardware interfacing with machine learning inference through three logical stages: extraction, pre-processing, and detection.

CSI Extraction from Hardware (csi_extractor.py)

The extraction stage interfaces with Wi-Fi hardware and parses raw packets into structured data. Located in v1/src/hardware/csi_extractor.py, the CSIExtractor class orchestrates this process through hardware abstraction and validation.

Configuration and Parser Selection

Upon initialization, the extractor validates required configuration keys including hardware_type, sampling_rate, buffer_size, and timeout through the _validate_config method. Based on the hardware_type value, it instantiates either an ESP32CSIParser or RouterCSIParser to handle hardware-specific byte formats.

# v1/src/hardware/csi_extractor.py#L73-L79

if config["hardware_type"] == "esp32":
    self.parser = ESP32CSIParser()
elif config["hardware_type"] == "router":
    self.parser = RouterCSIParser()

Data Acquisition and Parsing

The connect method establishes hardware connections through the async _establish_hardware_connection stub, while _read_raw_data retrieves raw CSI byte strings. The extract_csi method passes these bytes to the selected parser, which returns a CSIData dataclass containing timestamp, amplitude, phase, and metadata. Optional validation through validate_csi_data checks for non-empty arrays, sensible frequency ranges, bandwidth limits, antenna counts, and SNR thresholds.

CSI Pre-processing and Feature Extraction (csi_processor.py)

The processing stage cleans raw measurements and derives features for human detection. Implemented in v1/src/core/csi_processor.py, the CSIProcessor class provides configurable signal processing and statistical analysis.

Signal Cleaning and Normalization

The preprocess_csi_data method chains three operations to prepare raw CSI for analysis. First, _remove_noise zeroes amplitudes below a configurable decibel threshold (default -80 dB). Second, _apply_windowing multiplies sub-carriers by a Hamming window to reduce spectral leakage. Finally, _normalize_amplitude scales the signal to unit variance.

# v1/src/core/csi_processor.py#L26-L40

def preprocess_csi_data(self, csi_data: CSIData) -> CSIData:
    cleaned = self._remove_noise(csi_data)
    windowed = self._apply_windowing(cleaned)
    normalized = self._normalize_amplitude(windowed)
    return normalized

Feature Computation

The extract_features method computes statistical and spectral characteristics including mean and variance of amplitudes, phase differences across antennas, antenna correlation matrices, Doppler-shift placeholders, and power spectral density estimates. These features populate the CSIFeatures dataclass for downstream consumption.

Human Presence Detection

The detect_human_presence method analyzes motion patterns through _analyze_motion_patterns, calculates detection confidence by combining amplitude, phase, and motion indicators via _calculate_detection_confidence, and applies exponential temporal smoothing through _apply_temporal_smoothing. The result is a HumanDetectionResult containing a boolean detection flag, confidence score, motion score, timestamp, and the complete feature set.

End-to-End Pipeline Orchestration

The process_csi_data coroutine in CSIProcessor stitches the stages together into a unified async workflow:

# v1/src/core/csi_processor.py#L124-L139

async def process_csi_data(self, csi_data: CSIData) -> HumanDetectionResult:
    self._total_processed += 1
    preprocessed = self.preprocess_csi_data(csi_data)
    features = self.extract_features(preprocessed)
    detection = self.detect_human_presence(features)
    self.add_to_history(csi_data)
    return detection

This method increments the processing counter, applies pre-processing, extracts features, performs human detection, maintains a rolling history deque of raw samples for temporal context, and returns the final HumanDetectionResult.

Practical Implementation Example

The following runnable example demonstrates the complete pipeline from hardware extraction to human detection:

import asyncio
import logging
from src.hardware.csi_extractor import CSIExtractor
from src.core.csi_processor import CSIProcessor

# 1️⃣ Configure the extractor (ESP32 example)

extractor_cfg = {
    "hardware_type": "esp32",
    "sampling_rate": 10,
    "buffer_size": 1024,
    "timeout": 2,
    "validation_enabled": True,
    "retry_attempts": 3,
}
extractor = CSIExtractor(config=extractor_cfg, logger=logging.getLogger("extractor"))

# 2️⃣ Configure the processor

processor_cfg = {
    "sampling_rate": 10,
    "window_size": 256,
    "overlap": 0.5,
    "noise_threshold": -80,          # dB

    "human_detection_threshold": 0.7,
    "smoothing_factor": 0.85,
    "max_history_size": 200,
}
processor = CSIProcessor(config=processor_cfg, logger=logging.getLogger("processor"))

async def run_once():
    await extractor.connect()
    raw_csi = await extractor.extract_csi()      # → CSIData

    result = await processor.process_csi_data(raw_csi)
    print(f"Human detected: {result.human_detected}, confidence={result.confidence:.2f}")

asyncio.run(run_once())

This implementation pulls a single CSI sample from an ESP32 device, processes it through the full pipeline, and outputs the human detection result with confidence scoring.

Key Files and Components

File	Role
v1/src/hardware/csi_extractor.py	Core extractor, hardware parsers, and validation logic
v1/src/core/csi_processor.py	Pre-processing, feature extraction, and detection pipeline
v1/tests/unit/test_csi_extractor.py	Unit tests for parsing, validation, and error handling
v1/tests/unit/test_csi_processor.py	Tests covering each processing stage and async workflow
references/wifi_densepose_pytorch.py	Reference implementation for PyTorch pose-estimation integration

Summary

• CSIExtractor (v1/src/hardware/csi_extractor.py) handles hardware abstraction, parsing raw Wi-Fi packets into validated CSIData objects using hardware-specific parsers for ESP32 and router platforms.
• CSIProcessor (v1/src/core/csi_processor.py) implements the signal processing chain: noise removal, Hamming windowing, normalization, statistical feature extraction, and motion-based human detection with exponential smoothing.
• process_csi_data orchestrates the complete async pipeline, maintaining temporal history for context-aware detection and returning structured HumanDetectionResult objects.
• The pipeline supports multiple hardware backends through pluggable parsers and configurable thresholds for noise filtering and detection sensitivity, enabling real-time human presence detection from Wi-Fi channel measurements.

Frequently Asked Questions

How does the CSI extraction stage handle different Wi-Fi hardware types?

The CSIExtractor class uses a factory pattern to select hardware-specific parsers. Based on the hardware_type configuration key in v1/src/hardware/csi_extractor.py, it instantiates either an ESP32CSIParser or RouterCSIParser. Each parser handles the unique byte format and metadata structure of its respective hardware platform, ensuring consistent CSIData output regardless of whether the source is an ESP32 microcontroller or a commercial router.

What pre-processing steps are applied to raw CSI data before feature extraction?

The CSIProcessor applies a three-stage cleaning pipeline defined in v1/src/core/csi_processor.py. First, _remove_noise zeroes amplitudes below a configurable decibel threshold (default -80 dB). Second, _apply_windowing multiplies sub-carriers by a Hamming window to reduce spectral leakage. Finally, _normalize_amplitude scales the signal to unit variance, preparing clean data for statistical feature extraction and motion analysis.

How does the pipeline determine if a human is present in the detection area?

Human detection occurs in the detect_human_presence method, which combines multiple signal indicators into a confidence score. The pipeline analyzes motion patterns through _analyze_motion_patterns, calculates detection confidence by correlating amplitude, phase, and motion metrics via _calculate_detection_confidence, and applies exponential temporal smoothing through _apply_temporal_smoothing. A human is detected when the smoothed confidence exceeds the human_detection_threshold configuration parameter, typically set to 0.7.

Can the CSI processing pipeline run asynchronously for real-time applications?

Yes, the entire pipeline is designed for asynchronous operation. The CSIExtractor uses async methods like connect and extract_csi to handle hardware I/O without blocking. The CSIProcessor.process_csi_data method is declared as async def and can process samples concurrently. This architecture supports real-time streaming applications where CSI samples arrive continuously from Wi-Fi hardware, enabling non-blocking data acquisition and processing suitable for pose estimation inference.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}