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

> Understand the CSI processing pipeline in WiFi DensePose. Learn how raw Wi-Fi data is cleaned, normalized, and analyzed for accurate human presence detection.

- Repository: [rUv/wifi-densepose](https://github.com/ruvnet/wifi-densepose)
- Tags: internals
- Published: 2026-02-16

---

**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`](https://github.com/ruvnet/wifi-densepose/blob/main/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`](https://github.com/ruvnet/wifi-densepose/blob/main/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.

```python

# 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`](https://github.com/ruvnet/wifi-densepose/blob/main/csi_processor.py))

The processing stage cleans raw measurements and derives features for human detection. Implemented in [`v1/src/core/csi_processor.py`](https://github.com/ruvnet/wifi-densepose/blob/main/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.

```python

# 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:

```python

# 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:

```python
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`](https://github.com/ruvnet/wifi-densepose/blob/main/v1/src/hardware/csi_extractor.py) | Core extractor, hardware parsers, and validation logic |
| [`v1/src/core/csi_processor.py`](https://github.com/ruvnet/wifi-densepose/blob/main/v1/src/core/csi_processor.py) | Pre-processing, feature extraction, and detection pipeline |
| [`v1/tests/unit/test_csi_extractor.py`](https://github.com/ruvnet/wifi-densepose/blob/main/v1/tests/unit/test_csi_extractor.py) | Unit tests for parsing, validation, and error handling |
| [`v1/tests/unit/test_csi_processor.py`](https://github.com/ruvnet/wifi-densepose/blob/main/v1/tests/unit/test_csi_processor.py) | Tests covering each processing stage and async workflow |
| [`references/wifi_densepose_pytorch.py`](https://github.com/ruvnet/wifi-densepose/blob/main/references/wifi_densepose_pytorch.py) | Reference implementation for PyTorch pose-estimation integration |

## Summary

- **CSIExtractor** ([`v1/src/hardware/csi_extractor.py`](https://github.com/ruvnet/wifi-densepose/blob/main/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`](https://github.com/ruvnet/wifi-densepose/blob/main/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`](https://github.com/ruvnet/wifi-densepose/blob/main/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`](https://github.com/ruvnet/wifi-densepose/blob/main/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.