Skip to content

Latest commit

 

History

History
1047 lines (716 loc) · 26.8 KB

File metadata and controls

1047 lines (716 loc) · 26.8 KB

WSDP API Reference

Complete API documentation for WSDP (Wi-Fi Sensing Data Processing).

Table of Contents


Core Pipeline

wsdp.pipeline()

Main training pipeline function.

from wsdp import pipeline

pipeline(
    input_path: str,
    output_folder: str,
    dataset: str,
    model_path: Optional[str] = None,
    learning_rate: Optional[float] = None,
    num_epochs: Optional[int] = None,
    batch_size: Optional[int] = None,
    config_file: Optional[str] = None,
)

Parameters:

Parameter Type Default Description
input_path str required Path to input data directory
output_folder str required Path to output directory
dataset str required Dataset name (widar, gait, xrf55, elderAL, zte)
model_path str None Path to custom model file
learning_rate float None Learning rate (overrides config)
num_epochs int None Number of epochs (overrides config)
batch_size int None Batch size (overrides config)
config_file str None Path to YAML config file

Data Structures

CSIData

Container for raw CSI data from a file.

from wsdp.structure import CSIData

data = CSIData(file_path="/path/to/file.dat")

CSIFrame

Standardized frame structure for processed CSI.

from wsdp.structure import CSIFrame

frame = CSIFrame(
    csi_matrix: np.ndarray,  # Shape: (T, F, A)
    timestamp: float,
    metadata: dict,
)

Readers

Reader Dataset Format
BfeeReader Widar / Gait .dat (bfee)
XRF55Reader XRF55 .npy
ElderALReader ElderAL .csv
ZTEReader ZTE .csv

Algorithms

Denoising

wavelet_denoise_csi()

Wavelet-based denoising using VisuShrink thresholding.

from wsdp.algorithms import wavelet_denoise_csi

denoised = wavelet_denoise_csi(csi_tensor)

Parameters:

Parameter Type Default Description
csi_tensor np.ndarray required CSI array of shape (T, F, A), complex

Returns: np.ndarray — Denoised CSI with same shape and dtype.

Reference: Donoho DL, Johnstone IM. "Ideal spatial adaptation by wavelet shrinkage." Biometrika, 1994.


butterworth_denoise()

Butterworth low-pass filter for CSI denoising.

from wsdp.algorithms import butterworth_denoise

denoised = butterworth_denoise(csi, order=5, cutoff=0.3)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F), complex or real
order int 5 Filter order
cutoff float 0.3 Normalized cutoff frequency in (0, 1]

Returns: np.ndarray — Denoised CSI with same shape and dtype.

Reference: Butterworth S. "On the theory of filter amplifiers." Wireless Engineer, vol. 7, 1930.


savgol_denoise()

Savitzky-Golay polynomial smoothing filter.

from wsdp.algorithms import savgol_denoise

denoised = savgol_denoise(csi, window_length=11, polyorder=3)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F), complex or real
window_length int 11 Window length (must be odd, >= 3)
polyorder int 3 Polynomial order (must be < window_length)

Returns: np.ndarray — Smoothed CSI with same shape and dtype.

Reference: Savitzky A, Golay MJE. "Smoothing and differentiation of data by simplified least squares procedures." Analytical Chemistry, 1964.


Phase Calibration

phase_calibration() (Linear)

Standard linear phase calibration using polynomial fitting.

from wsdp.algorithms import phase_calibration

calibrated = phase_calibration(csi_data)

Parameters:

Parameter Type Default Description
csi_data np.ndarray required 3D CSI array (T, F, A), complex

Returns: np.ndarray — Phase-calibrated CSI with same shape.

Reference: Halperin D, Hu W, Sheth A, Wetherall D. "Predictable 802.11 packet delivery from wireless channel measurements." ACM SIGCOMM, 2010.


polynomial_calibration()

Polynomial phase calibration across subcarriers.

from wsdp.algorithms import polynomial_calibration

calibrated = polynomial_calibration(csi, degree=3)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A), complex
degree int 3 Polynomial degree (1=linear, 2=quadratic, 3=cubic)

Returns: np.ndarray — Phase-calibrated CSI with same shape.

Reference: Generalization of linear calibration from Halperin et al., ACM SIGCOMM, 2010.


stc_calibration()

Sanitize-then-Calibrate (STC) phase error removal.

from wsdp.algorithms import stc_calibration

calibrated = stc_calibration(csi)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A), complex

Returns: np.ndarray — Phase-calibrated CSI with same shape.

Reference: Xie Y, Li Z, Li M. "Precise Power Delay Profiling with Commodity WiFi." IEEE Transactions on Wireless Communications (TWC), 2019.


robust_phase_sanitization()

Robust phase sanitization using median-based approach.

from wsdp.algorithms import robust_phase_sanitization

calibrated = robust_phase_sanitization(csi)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A), complex

Returns: np.ndarray — Phase-sanitized CSI with same shape.

Reference: Wang G, Zou Y, Zhou Z, Wu K, Ni LM. "FIMD: Fine-grained Device-free Motion Detection." IEEE ICPADS, 2012.


Amplitude Processing

normalize_amplitude()

Normalize CSI amplitude along the time axis.

from wsdp.algorithms import normalize_amplitude

normalized = normalize_amplitude(csi, method='z-score')

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F)
method str 'z-score' 'z-score' or 'min-max'

Returns: np.ndarray — Normalized CSI with same shape.

Reference: Ma Y, et al. "PhaseFi: Phase Fingerprinting for Indoor Localization with a Deep Learning Approach." IEEE GLOBECOM, 2015.


remove_outliers()

Remove or clip outlier amplitudes in CSI data.

from wsdp.algorithms import remove_outliers

cleaned = remove_outliers(csi, method='iqr', factor=1.5)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F)
method str 'iqr' 'iqr' (IQR-based) or 'z-score' (std-dev based)
factor float 1.5 Detection threshold multiplier

Returns: np.ndarray — CSI with outliers clipped, same shape.


Interpolation

interpolate_grid()

Interpolate CSI data to a target number of subcarriers.

from wsdp.algorithms import interpolate_grid

interpolated = interpolate_grid(csi, target_K=30, method='cubic')

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A)
target_K int 30 Target number of subcarriers
method str 'cubic' 'linear', 'cubic', or 'nearest'

Returns: np.ndarray — Interpolated CSI with shape (T, target_K, A).

Reference: Bianchi V, et al. "Indoor Localization by Interpolation of Radio Maps." Sensors, 2020.


Feature Extraction

doppler_spectrum()

Compute Doppler spectrum from CSI time series using STFT.

from wsdp.algorithms import doppler_spectrum

spectrum = doppler_spectrum(csi, n_fft=64, hop_length=32)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F), complex
n_fft int 64 FFT size for STFT
hop_length int 32 Hop size between STFT windows

Returns: np.ndarray — Doppler spectrogram of shape (n_freq, n_time, F[, A]).

Reference: Ali K, et al. "Keystroke Recognition Using WiFi Signals." ACM MobiCom, 2015.


entropy_features()

Compute information entropy features from CSI amplitude distribution.

from wsdp.algorithms import entropy_features

entropy = entropy_features(csi, bins=50)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F)
bins int 50 Number of histogram bins for entropy estimation

Returns: np.ndarray — Entropy values of shape (F,) or (F, A).

Reference: Shannon CE. "A Mathematical Theory of Communication." Bell System Technical Journal, 1948.


csi_ratio()

Compute CSI ratio between antenna pairs.

from wsdp.algorithms import csi_ratio

ratio = csi_ratio(csi, antenna_pairs=[(0, 1), (1, 2)])

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A), A >= 2
antenna_pairs list None List of (ant1, ant2) tuples. Default: consecutive pairs

Returns: np.ndarray — CSI ratios with shape (T, F, n_pairs).

Reference: Halperin D, et al. "Tool release: Gathering 802.11n traces with channel state information." ACM SIGCOMM CCR, 2011.


tensor_decomposition()

Decompose CSI tensor using CP or Tucker decomposition.

from wsdp.algorithms import tensor_decomposition

decomposed = tensor_decomposition(csi, rank=10, method='cp')

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A)
rank int 10 Rank of decomposition
method str 'cp' 'cp' (Canonical Polyadic) or 'tucker'

Returns: np.ndarray — Reconstructed low-rank CSI tensor, same shape.

Reference: Kolda TG, Bader BW. "Tensor Decompositions and Applications." SIAM Review, 2009.


Activity Detection

detect_activity()

Detect activity using sliding window variance analysis.

from wsdp.algorithms import detect_activity

activity = detect_activity(csi, window=32, threshold=0.1)

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F)
window int 32 Sliding window size in samples
threshold float 0.1 Detection threshold for normalized variance

Returns: np.ndarray — Boolean array of shape (T,) indicating activity.

Reference: Zhou Z, et al. "Device-Free Passive Localization for Human Activity Recognition." IEEE Communications Magazine, 2013.


change_point_detection()

Detect change points in CSI time series.

from wsdp.algorithms import change_point_detection

change_points = change_point_detection(csi, method='mean_shift_ratio')

Parameters:

Parameter Type Default Description
csi np.ndarray required CSI array of shape (T, F, A) or (T, F)
method str 'mean_shift_ratio' 'mean_shift_ratio', 'cusum', or 'variance'

Returns: np.ndarray — Array of time indices where change points detected.

Reference: Page ES. "Continuous Inspection Schemes." Biometrika, 1954. (CUSUM)


Visualization

plot_csi_heatmap()

Plot CSI amplitude as a time-frequency heatmap.

from wsdp.algorithms.visualization import plot_csi_heatmap

fig = plot_csi_heatmap(csi_data, antenna_idx=0, save_path='heatmap.png')

plot_denoising_comparison()

Plot before/after denoising comparison.

from wsdp.algorithms.visualization import plot_denoising_comparison

fig = plot_denoising_comparison(original, denoised, antenna_idx=0)

plot_phase_calibration()

Plot phase before and after calibration.

from wsdp.algorithms.visualization import plot_phase_calibration

fig = plot_phase_calibration(original, calibrated, antenna_idx=0)

Unified API

The unified API provides a consistent interface for all algorithm categories.

denoise()

from wsdp.algorithms import denoise

denoised = denoise(csi, method='wavelet', **kwargs)
Method Source Function Key Parameters
'wavelet' wavelet_denoise_csi()
'butterworth' butterworth_denoise() order, cutoff
'savgol' savgol_denoise() window_length, polyorder

calibrate()

from wsdp.algorithms import calibrate

calibrated = calibrate(csi, method='linear', **kwargs)
Method Source Function Key Parameters
'linear' phase_calibration()
'polynomial' polynomial_calibration() degree
'stc' stc_calibration()
'robust' robust_phase_sanitization()

normalize()

from wsdp.algorithms import normalize

normalized = normalize(csi, method='z-score')
Method Source Function Key Parameters
'z-score' normalize_amplitude()
'min-max' normalize_amplitude()

interpolate()

from wsdp.algorithms import interpolate

result = interpolate(csi, target_K=30, method='cubic')
Method Source Function Key Parameters
'linear' interpolate_grid() target_K
'cubic' interpolate_grid() target_K
'nearest' interpolate_grid() target_K

extract_features()

from wsdp.algorithms import extract_features

features = extract_features(csi, features=['doppler', 'entropy'])
Feature Source Function Key Parameters
'doppler' doppler_spectrum() n_fft, hop_length
'entropy' entropy_features() bins
'ratio' csi_ratio() antenna_pairs
'decomposition' tensor_decomposition() rank, method
'conjugate_multiply' conjugate_multiply() antenna_pairs
'pca_fusion' pca_fusion() n_components

Pluggable Architecture

WSDP provides a pluggable algorithm architecture built on the Registry Pattern, allowing users to switch algorithms, add custom implementations, configure via files, and use preset pipelines.

Algorithm Registry

register_algorithm()

Register a custom algorithm.

from wsdp.algorithms import register_algorithm

def my_denoise(csi, strength=1.0, **kwargs):
    return csi * strength

register_algorithm('denoise', 'my_method', my_denoise)

Parameters:

Parameter Type Description
category str Category: 'denoise', 'calibrate', 'normalize', 'interpolate', 'extract_features', 'detect', 'outliers'
name str Algorithm name (used as method= parameter)
func Callable Algorithm implementation

get_algorithm()

Get an algorithm function by category and name.

from wsdp.algorithms import get_algorithm

func = get_algorithm('denoise', 'butterworth')
result = func(csi, order=5)

list_algorithms()

List available algorithms.

from wsdp.algorithms import list_algorithms

# All categories
>>> list_algorithms()
{'denoise': ['wavelet', 'butterworth', 'savgol'], 'calibrate': [...]}

# Specific category
>>> list_algorithms('denoise')
{'wavelet': 'wsdp.algorithms.denoising:wavelet_denoise_csi', ...}

is_registered()

Check if an algorithm is registered.

>>> is_registered('denoise', 'wavelet')
True

algorithm_info()

Get detailed information about an algorithm.

>>> algorithm_info('denoise', 'butterworth')
{'name': 'butterworth', 'category': 'denoise', 'module': '...', ...}

unregister_algorithm()

Remove a custom algorithm (built-in algorithms cannot be unregistered).

unregister_algorithm('denoise', 'my_method')

Configuration Files

WSDP supports YAML and JSON configuration files for algorithm selection.

YAML Format

# examples/configs/algorithms_config.yaml
denoise:
  method: butterworth
  params:
    order: 5
    cutoff: 0.3

calibrate:
  method: polynomial
  params:
    degree: 3

normalize:
  method: z-score

JSON Format

{
  "denoise": {
    "method": "butterworth",
    "params": {"order": 5, "cutoff": 0.3}
  },
  "calibrate": {
    "method": "polynomial",
    "params": {"degree": 3}
  }
}

Preset Reference

preset: high_quality

load_config()

Load algorithm configuration from a file.

from wsdp.algorithms import load_config

config = load_config('examples/configs/algorithms_config.yaml')

save_config()

Save algorithm configuration to a file.

from wsdp.algorithms import save_config, apply_preset

steps = apply_preset('high_quality')
save_config(steps, 'my_config.yaml', format='yaml')

Pipeline Presets

Built-in Presets

Preset Denoise Calibrate Normalize Use Case
high_quality butterworth (order=5) stc z-score Maximum accuracy
fast savgol (w=7) linear min-max Speed-optimized
robust wavelet robust z-score Noisy environments
gesture_recognition butterworth (order=4) stc z-score Gesture tasks
activity_detection savgol (w=11) polynomial (deg=2) z-score HAR tasks
localization wavelet robust z-score Localization tasks

apply_preset()

Get a preset pipeline configuration.

from wsdp.algorithms import apply_preset

steps = apply_preset('high_quality')
# {'denoise': {'method': 'butterworth', 'order': 5, ...}, ...}

execute_pipeline()

Execute a processing pipeline on CSI data.

from wsdp.algorithms import apply_preset, execute_pipeline

steps = apply_preset('high_quality')
processed = execute_pipeline(csi, steps)

register_preset()

Register a custom preset.

from wsdp.algorithms import register_preset

register_preset('my_preset', {
    'denoise': {'method': 'butterworth', 'order': 3},
    'calibrate': {'method': 'linear'},
})

list_presets()

List all available presets.

>>> list_presets()
{'high_quality': ['denoise', 'calibrate', 'normalize'], 'fast': [...], ...}

Custom Algorithms

Users can register their own algorithm implementations:

from wsdp.algorithms import register_algorithm, denoise, calibrate

# 1. Define your algorithm
def my_denoise(csi, strength=1.0, **kwargs):
    """Custom denoising: simple smoothing."""
    from scipy.ndimage import uniform_filter1d
    result = np.empty_like(csi)
    for f in range(csi.shape[1]):
        for a in range(csi.shape[2]):
            real_part = uniform_filter1d(np.real(csi[:, f, a]), size=int(strength * 5))
            imag_part = uniform_filter1d(np.imag(csi[:, f, a]), size=int(strength * 5))
            result[:, f, a] = real_part + 1j * imag_part
    return result

# 2. Register it
register_algorithm('denoise', 'my_smooth', my_denoise)

# 3. Use it via unified API
denoised = denoise(csi, method='my_smooth', strength=2.0)

# 4. Or use it in a pipeline
from wsdp.algorithms import execute_pipeline
steps = {
    'denoise': {'method': 'my_smooth', 'strength': 1.5},
    'calibrate': {'method': 'stc'},
}
result = execute_pipeline(csi, steps)

Models

Unified Model API

from wsdp.models import create_model, list_models, get_model

# Create any model
model = create_model(name, num_classes, input_shape, **kwargs)

# List models
all_models = list_models()                          # All models
baselines = list_models("baseline")                 # By category

# Get model class
model = get_model("ResNet1D", num_classes=10, input_shape=(20, 30, 3))

Parameters:

  • name: Model name (case-insensitive)
  • num_classes: Number of output classes
  • input_shape: Tuple of (T, F, A) — time steps, frequency bins, antennas
  • **kwargs: Model-specific hyperparameters

Model Registry

All models are stored in MODEL_REGISTRY and can be accessed by category:

Category Models
baseline MLPModel, CNN1DModel, CNN2DModel, LSTMModel
mainstream ResNet1D, ResNet2D, BiLSTMAttention, EfficientNetCSI
sota VisionTransformerCSI, MambaCSI, GraphNeuralCSI, CSIModel

Baseline Models

MLPModel

Fully-connected baseline, flattens input through MLP.

model = create_model("MLPModel", num_classes=10, input_shape=(20, 30, 3),
                      hidden_dims=[512, 256], dropout=0.3)

CNN1DModel

1D convolution extracting temporal features.

model = create_model("CNN1DModel", num_classes=10, input_shape=(20, 30, 3),
                      base_channels=64, num_layers=4)

CNN2DModel

2D convolution processing spectral representations per time step.

model = create_model("CNN2DModel", num_classes=10, input_shape=(20, 30, 3),
                      base_channels=32, num_layers=3)

LSTMModel

LSTM for temporal sequence modeling.

model = create_model("LSTMModel", num_classes=10, input_shape=(20, 30, 3),
                      hidden_size=128, num_layers=2, dropout=0.3)

Mainstream Models

ResNet1D

1D residual network with 3 residual blocks.

model = create_model("ResNet1D", num_classes=10, input_shape=(20, 30, 3),
                      base_channels=64)

ResNet2D

2D residual network for spatial feature extraction.

model = create_model("ResNet2D", num_classes=10, input_shape=(20, 30, 3),
                      base_channels=32)

BiLSTMAttention

Bidirectional LSTM with multi-head self-attention.

model = create_model("BiLSTMAttention", num_classes=10, input_shape=(20, 30, 3),
                      hidden_size=128, num_layers=2, num_heads=4, dropout=0.3)

EfficientNetCSI

Efficient CNN with configurable width and depth multipliers.

model = create_model("EfficientNetCSI", num_classes=10, input_shape=(20, 30, 3),
                      width_mult=1.0, depth_mult=1.0, base_channels=16)

SOTA Models

VisionTransformerCSI

Vision Transformer treating F×A spatial patches across time steps.

model = create_model("VisionTransformerCSI", num_classes=10, input_shape=(20, 30, 3),
                      embed_dim=128, num_heads=4, num_layers=4,
                      patch_size_f=4, patch_size_a=4, dropout=0.1)

MambaCSI

State space model (Mamba) for efficient long-range temporal modeling.

model = create_model("MambaCSI", num_classes=10, input_shape=(20, 30, 3),
                      d_model=128, d_state=16, num_layers=4)

GraphNeuralCSI

Graph neural network modeling antenna/subcarrier topology.

model = create_model("GraphNeuralCSI", num_classes=10, input_shape=(20, 30, 3),
                      hidden_dim=64, num_gcn_layers=3, num_heads=4)

CSIModel

Original CNN + Transformer model (backward compatible).

model = create_model("CSIModel", num_classes=10, input_shape=(20, 30, 3),
                      base_channels=32, latent_dim=128)

Custom Model Registration

from wsdp.models import register_model
import torch.nn as nn

class MyModel(nn.Module):
    def __init__(self, num_classes, input_shape, **kwargs):
        super().__init__()
        # Your architecture
        
    def forward(self, x):
        # x: (B, T, F, A) — complex or real
        # return: (B, num_classes)
        ...

register_model("custom", "MyModel", MyModel)

Inference

wsdp.predict()

Run inference on CSI data.

from wsdp import predict

predictions = predict(csi_data, model_path, num_classes=6)

CLI

wsdp run

wsdp run ./data/elderAL ./output elderAL --lr 0.001 --epochs 50

wsdp download

wsdp download elderAL ./data --email you@example.com --password yourpassword

wsdp list

wsdp list -V

Algorithm Library Overview

Category Algorithm Function Key Reference
Denoising Wavelet wavelet_denoise_csi() Donoho & Johnstone, 1994
Butterworth butterworth_denoise() Butterworth, 1930
Savitzky-Golay savgol_denoise() Savitzky & Golay, 1964
Phase Calibration Linear phase_calibration() Halperin et al., 2010
Polynomial polynomial_calibration() Extension of linear
STC stc_calibration() Xie et al., IEEE TWC 2019
Robust robust_phase_sanitization() Wang et al., IEEE ICPADS 2012
Normalization Z-Score normalize_amplitude() Standard statistical
Min-Max normalize_amplitude() Standard statistical
Interpolation Linear/Cubic/Nearest interpolate_grid() de Boor, 1978
Features Doppler doppler_spectrum() Ali et al., MobiCom 2015
Entropy entropy_features() Shannon, 1948
CSI Ratio csi_ratio() Halperin et al., 2011
Tensor Decomposition tensor_decomposition() Kolda & Bader, SIAM 2009
Conjugate Multiply conjugate_multiply() Xie et al., IEEE TWC 2019
PCA Fusion pca_fusion() Ma et al., IEEE GLOBECOM 2015
Detection Activity detect_activity() Zhou et al., 2013
Change Point change_point_detection() Adams & MacKay, 2007

For more examples, see the examples/ directory.