How to Set Up Basic CSI Processing in WiFi DensePose: A Complete Implementation Guide

WiFi DensePose processes raw Channel State Information (CSI) through two core classes—CSIExtractor and CSIProcessor—to parse Wi-Fi hardware data and detect human presence using configurable signal processing pipelines.

WiFi DensePose (ruvnet/wifi-densepose) enables human pose estimation by analyzing Wi-Fi signal variations captured as Channel State Information. Setting up basic CSI processing requires configuring the hardware abstraction layer to ingest raw bytes and the signal processing module to filter noise and extract detection features. This implementation relies on the source code in v1/src/hardware/csi_extractor.py for data acquisition and v1/src/core/csi_processor.py for feature extraction and human detection.

CSI Processing Architecture Overview

The WiFi DensePose CSI pipeline consists of two primary components that transform raw radio signals into structured detection results.

CSIExtractor Hardware Abstraction

The CSIExtractor class in v1/src/hardware/csi_extractor.py manages the interface between your application and Wi-Fi hardware. It supports both ESP32 devices and commercial routers through dedicated parsers (ESP32CSIParser and RouterCSIParser). The extractor validates configuration parameters, establishes hardware connections, and converts raw byte streams into structured CSIData objects.

CSIProcessor Signal Pipeline

The CSIProcessor class in v1/src/core/csi_processor.py executes a four-stage pipeline on each CSI sample: preprocessing (noise removal, windowing, normalization), feature extraction (amplitude, phase, correlation, Doppler), temporal smoothing using exponential moving averages, and binary human detection. The processor maintains internal state through a fixed-size deque (csi_history) to enable temporal analysis across multiple frames.

Configuring the CSI Extractor

Before processing signals, you must configure the extractor with hardware-specific parameters and sampling settings.

Hardware Parameters

Define a configuration dictionary specifying the device type and acquisition settings:

extractor_cfg = {
    "hardware_type": "esp32",       # Alternative: "router"

    "sampling_rate": 30,           # Hz

    "buffer_size": 1024,           # Frames per read operation

    "timeout": 5,                  # Seconds

    "validation_enabled": True,    # Enable payload validation

    "retry_attempts": 3,           # Automatic retries on failure

}

The hardware_type key determines which parser class the extractor instantiates internally via the _validate_config method.

Connection Management

Instantiate the extractor and establish the hardware link asynchronously:

from wifi_densepose.hardware.csi_extractor import CSIExtractor

extractor = CSIExtractor(config=extractor_cfg)
await extractor.connect()  # Returns True on successful initialization

The connect() method initializes the appropriate parser and prepares the data acquisition interface.

Configuring the CSI Processor

Signal processing parameters control how raw CSI data transforms into detection features.

Windowing and Noise Settings

Configure the processor with parameters matching your extractor's sampling rate:

processor_cfg = {
    "sampling_rate": 30,               # Must match extractor configuration

    "window_size": 64,                 # Sub-carriers per FFT window

    "overlap": 0.5,                    # 50% overlap between consecutive windows

    "noise_threshold": -80,            # dB; values below are discarded

    "human_detection_threshold": 0.8,  # Minimum confidence for positive detection

    "smoothing_factor": 0.9,           # Exponential moving average weight

}

The window_size and overlap parameters control the spectral analysis resolution, while noise_threshold filters low-power artifacts before feature extraction.

Detection Thresholds

The human_detection_threshold (0.0–1.0) determines the sensitivity of the binary classification stage. Lower values increase recall but may introduce false positives, while higher values improve precision at the cost of missing subtle motion.

Implementing the Processing Pipeline

Once configured, execute the pipeline using either discrete sampling or continuous streaming modes.

Single-Sample Processing

For one-shot detection, pull a single CSI frame and process it immediately:

from wifi_densepose.core.csi_processor import CSIProcessor

# Initialize processor with configuration

processor = CSIProcessor(config=processor_cfg)

# Acquire and process one sample

raw_csi = await extractor.extract_csi()
detection = await processor.process_csi_data(raw_csi)

if detection and detection.human_detected:
    print(f"Human detected! Confidence: {detection.confidence:.2f}")
else:
    print("No human present.")

The process_csi_data() method executes the full preprocessing chain (_remove_noise, _apply_windowing, _normalize_amplitude), extracts statistical features, applies temporal smoothing via _apply_temporal_smoothing, and returns a HumanDetectionResult object containing human_detected, confidence, motion_score, and timestamp attributes.

Continuous Streaming Mode

For real-time applications, register an asynchronous callback to process each incoming frame:

async def on_csi_sample(sample):
    """Process each CSI frame as it arrives."""
    result = await processor.process_csi_data(sample)
    if result and result.human_detected:
        print(f"[{result.timestamp.isoformat()}] Human detected: {result.confidence:.2f}")

# Start the acquisition loop

await extractor.start_streaming(callback=on_csi_sample)

The streaming loop continues until extractor.stop_streaming() is invoked or an exception terminates the acquisition thread. This mode is essential for real-time pose estimation applications requiring low latency between signal capture and inference.

Complete Implementation Examples

One-Shot Detection Script

import asyncio
from wifi_densepose.hardware.csi_extractor import CSIExtractor
from wifi_densepose.core.csi_processor import CSIProcessor

async def main():
    # 1️⃣ Configure and connect extractor

    extractor_cfg = {
        "hardware_type": "esp32",
        "sampling_rate": 30,
        "buffer_size": 1024,
        "timeout": 5,
    }
    extractor = CSIExtractor(config=extractor_cfg)
    await extractor.connect()

    # 2️⃣ Configure processor

    processor_cfg = {
        "sampling_rate": 30,
        "window_size": 64,
        "overlap": 0.5,
        "noise_threshold": -80,
        "human_detection_threshold": 0.8,
    }
    processor = CSIProcessor(config=processor_cfg)

    # 3️⃣ Execute pipeline

    csi_sample = await extractor.extract_csi()
    detection = await processor.process_csi_data(csi_sample)

    status = "✅ Human detected" if detection and detection.human_detected else "❌ No human detected"
    print(f"{status} – confidence {detection.confidence:.2f}" if detection else status)

    await extractor.disconnect()

asyncio.run(main())

Continuous Monitoring Script

import asyncio
from wifi_densepose.hardware.csi_extractor import CSIExtractor
from wifi_densepose.core.csi_processor import CSIProcessor

async def handle_csi(sample):
    """Callback for streaming mode."""
    result = await processor.process_csi_data(sample)
    if result and result.human_detected:
        print(f"[{result.timestamp.isoformat()}] Motion: {result.confidence:.2f}")

async def main():
    # Shared configuration for ESP32 or router

    cfg = {
        "hardware_type": "router",
        "sampling_rate": 20,
        "buffer_size": 2048,
        "timeout": 3,
    }
    extractor = CSIExtractor(config=cfg)
    await extractor.connect()

    proc_cfg = {
        "sampling_rate": 20,
        "window_size": 56,
        "overlap": 0.4,
        "noise_threshold": -75,
        "human_detection_threshold": 0.75,
        "smoothing_factor": 0.85,
    }
    global processor
    processor = CSIProcessor(config=proc_cfg)

    # Runs until interrupted

    await extractor.start_streaming(callback=handle_csi)

asyncio.run(main())

Both examples assume compatible hardware drivers are available; the stub methods in CSIExtractor must be replaced with actual ESP32 or router driver implementations for production deployments.

Summary

• CSIExtractor (v1/src/hardware/csi_extractor.py) handles hardware abstraction, supporting ESP32 and router platforms through configurable parsers.
• CSIProcessor (v1/src/core/csi_processor.py) implements the signal processing chain: noise filtering, windowing, feature extraction, and temporal smoothing.
• Configuration requires matching sampling_rate values between extractor and processor to maintain temporal alignment.
• The pipeline produces HumanDetectionResult objects containing boolean detection flags, confidence scores, and motion metrics.
• Both one-shot (extract_csi) and continuous (start_streaming) acquisition modes support async/await patterns for non-blocking I/O.

Frequently Asked Questions

What Wi-Fi hardware is compatible with WiFi DensePose CSI extraction?

WiFi DensePose supports ESP32 microcontrollers and commercial Wi-Fi routers through the ESP32CSIParser and RouterCSIParser classes in v1/src/hardware/csi_extractor.py. The hardware_type configuration key selects the appropriate parser for your specific device firmware.

How do I calibrate the human detection sensitivity?

Adjust the human_detection_threshold parameter in the processor configuration (default 0.8). Values closer to 1.0 require stronger motion signatures to trigger detection, reducing false positives in noisy environments. Values below 0.5 increase sensitivity for detecting subtle movements but may require stricter noise_threshold settings to compensate.

What is the difference between one-shot and streaming processing modes?

One-shot mode uses extractor.extract_csi() to pull single frames synchronously, suitable for triggered sampling or low-power applications. Streaming mode uses start_streaming(callback) to continuously buffer and process CSI data in real-time, invoking your callback function for each new sample without blocking the main thread.

How does the noise filtering stage work?

The processor applies a three-stage preprocessing sequence: _remove_noise eliminates samples below the configured noise_threshold (in dB), _apply_windowing applies a Hamming window to reduce spectral leakage, and _normalize_amplitude scales the signal to unit variance before feature extraction begins.

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