WSDP API Reference
Complete API documentation for WSDP (Wi-Fi Sensing Data Processing).
Table of Contents
- Core Pipeline
- Data Structures
- Readers
- Algorithms
- Denoising
- Phase Calibration
- Amplitude Processing
- Interpolation
- Feature Extraction
- Activity Detection
- Visualization
- Unified API
- Pluggable Architecture
- Algorithm Registry
- Configuration Files
- Pipeline Presets
- Custom Algorithms
- Models
- Inference
- CLI
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.
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 |
|---|---|---|
WidarReader |
Widar | .dat (bfee) |
GaitReader |
Gait | .dat (bfee) |
XRF55Reader |
XRF55 | .npy |
ElderALReader |
ElderAL | .csv |
ZTEReader |
ZTE | .csv |
Algorithms
Denoising
wavelet_denoise_csi()
Wavelet-based denoising using VisuShrink thresholding.
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.
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.
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.
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.
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.
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='bayesian')
Parameters:
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| csi | np.ndarray | required | CSI array of shape (T, F, A) or (T, F) |
| method | str | 'bayesian' | 'bayesian', 'cusum', or 'variance' |
Returns: np.ndarray — Array of time indices where change points detected.
Reference: Adams RP, MacKay DJC. "Bayesian Online Changepoint Detection." arXiv:0710.3742, 2007.
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()
| Method | Source Function | Key Parameters |
|---|---|---|
'wavelet' |
wavelet_denoise_csi() |
— |
'butterworth' |
butterworth_denoise() |
order, cutoff |
'savgol' |
savgol_denoise() |
window_length, polyorder |
calibrate()
| Method | Source Function | Key Parameters |
|---|---|---|
'linear' |
phase_calibration() |
— |
'polynomial' |
polynomial_calibration() |
degree |
'stc' |
stc_calibration() |
— |
'robust' |
robust_phase_sanitization() |
— |
normalize()
| Method | Source Function | Key Parameters |
|---|---|---|
'z-score' |
normalize_amplitude() |
— |
'min-max' |
normalize_amplitude() |
— |
interpolate()
| 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 |
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.
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).
Configuration Files
WSDP supports YAML and JSON configuration files for algorithm selection.
YAML Format
# 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
load_config()
Load algorithm configuration from a file.
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.
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.
ResNet2D
2D residual network for spatial feature extraction.
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.
CLI
wsdp run
wsdp download
wsdp list
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 | |
| Detection | Activity | detect_activity() |
Zhou et al., 2013 |
| Change Point | change_point_detection() |
Adams & MacKay, 2007 |
For more examples, see the examples/ directory.