# augsig
**augsig** is a lightweight Python package for augmenting 1D physiological signals such as PPG and ECG. It supports training-time and test-time augmentation using additive noise, smooth geometric warping (Bezier and PCHIP), and artifact-style distortions (burst, powerline, low-frequency noise, linear drift).
---
## Installation
**Option 1 — From PyPI:**
```bash
pip install augsig
```
**Option 2 — From source:**
```bash
git clone https://github.com/davood-fattahi/augsig.git
cd augsig
pip install .
```
**Option 3 — Without installation:** download or copy the `augsig/` folder into your project directory and import directly:
```python
from augsig import augment, Augment
```
Only `numpy` (>= 1.21) and `scipy` (>= 1.7) must be importable.
`matplotlib` is required only for the demo script (`tests/test.py`).
---
## Function Reference
These lower-level functions can be called directly for fine-grained control. They are also composed automatically by the augmentation pipeline described in the next section.
> **Input shape:** every function takes a 1D signal of shape `(N,)`. Shapes with singleton axes, such as `(N, 1)` or `(1, N)`, are squeezed automatically. Any other shape raises `ValueError`.
### Noise & Artifacts
#### `noisify(signal, snr_db, ...)`
Adds SNR-controlled colored noise to a signal. Returns `(noisy_signal, noise)`.
| Parameter | Type | Default | Description |
|---|---|---|---|
| `signal` | np.ndarray | — | 1D input signal |
| `snr_db` | float | — | Target SNR in dB. `float('inf')` → no noise (returns zeros). `float('-inf')` → unit-power noise (power = 1.0); both outputs are the noise itself — useful when you want to scale by amplitude instead of relative power (see example below) |
| `color` | str | `'white'` | Noise color: `'white'`, `'pink'`, `'brown'`, `'blue'`, `'violet'` |
| `bpass_params` | list | `[0, 1]` | `[fl_norm, fh_norm]` or `[fl_norm, fh_norm, order]`; Nyquist-normalized (0–1) |
| `dist` | str | `'gauss'` | Sample distribution: `'gauss'`, `'uniform'`, `'laplace'`, `'resample'` |
| `resample_pool` | array or `'self'` | `None` | Pool for empirical resampling; required when `dist='resample'` |
| `zero_mean` | bool | `True` | Subtract mean from noise before power scaling |
| `rng` | np.random.Generator | `None` | Seeded RNG for reproducibility |
**Example: SNR-controlled noise**
```python
from augsig.noisifier import noisify
noisy, noise = noisify(signal, snr_db=20, color='pink', bpass_params=[0.01, 0.9], dist='gauss')
```
**Example: Amplitude-controlled noise** — get unit-power noise, then scale manually
```python
_, noise = noisify(signal, snr_db=float('-inf'), color='pink')
noisy = signal + 0.05 * noise
```
**Example: Empirical resampling** — noise samples drawn from the signal itself or an external pool
```python
# Resample from the signal itself (captures its amplitude distribution)
noisy, noise = noisify(signal, snr_db=15, dist='resample', resample_pool='self')
# Resample from an external pool (e.g. a library of real noise recordings)
noise_pool = np.load('real_noise_library.npy') # 1D array
noisy, noise = noisify(signal, snr_db=15, dist='resample', resample_pool=noise_pool)
```
---
#### `burstify(signal, snr_db, ...)`
Generates localized burst artifacts by masking colored noise into short, randomly placed windows. Returns `(noisy_signal, burst_noise)`.
| Parameter | Type | Default | Description |
|---|---|---|---|
| `signal` | np.ndarray | — | 1D input signal |
| `snr_db` | float | — | Target SNR in dB, applied to the burst noise **after** masking. `float('inf')` → no noise. `float('-inf')` → unit-power burst noise — use this for amplitude-controlled scaling (see example below) |
| `color` | str | `'white'` | Noise color (see `noisify`) |
| `bpass_params` | list | `[0, 0.1]` | Bandpass cutoffs for the base noise |
| `dist` | str | `'laplace'` | Sample distribution (see `noisify`) |
| `resample_pool` | array or `'self'` | `None` | Pool for empirical resampling |
| `n_bursts` | int | `10` | Number of bursts. `0` adds no burst noise; negative values raise `ValueError` |
| `burst_width` | int or [min, max] | `[1, 10]` | Burst width in samples; a 2-element list samples each width uniformly |
| `burst_base` | float or [min, max] | `0.0` | DC offset added inside each burst |
| `zero_mean` | bool | `False` | Subtract mean from the final burst signal |
| `rng` | np.random.Generator | `None` | Seeded RNG |
**Example: SNR-controlled bursts**
```python
from augsig.noisifier import burstify
noisy, bursts = burstify(signal, snr_db=20, n_bursts=5, burst_width=[5, 20])
```
**Example: Amplitude-controlled bursts** — get unit-power burst noise, then scale manually
```python
_, bursts = burstify(signal, snr_db=float('-inf'), n_bursts=5,
burst_width=[5, 20], burst_base=[-0.2, 0.2])
noisy = signal + 0.1 * bursts
```
---
### Warping & Drift
All warping functions share the parameters below.
| Parameter | Type | Default | Description |
|---|---|---|---|
| `signal` | np.ndarray | — | 1D input signal |
| `k` | int | `4` | Total number of control points / knots, endpoints included. Scale with signal length for consistent warp density. Time-warp and PCHIP functions require `k >= 3`; `adrift_bezier` and `amod_bezier` require `k >= 1` |
| `variance` | float | `0.05` | Perturbation strength. Capped internally at `1 / (2k − 4)` to ensure monotonicity |
| `rng` | np.random.Generator | `None` | Seeded RNG |
---
#### `twarp_bezier(signal, k, variance, rng)` / `twarp_pchip(signal, k, variance, rng)`
Nonlinear time-axis warping — the signal is resampled along a smooth monotone mapping. Bezier produces globally smooth distortions; PCHIP produces more localized, segment-wise ones. Returns a warped signal of the same length.
**Example:**
```python
from augsig.warper import twarp_bezier, twarp_pchip
warped = twarp_bezier(signal, k=6, variance=0.02)
warped = twarp_pchip(signal, k=6, variance=0.02)
```
---
#### `adrift_bezier(signal, k, variance, rng)` / `adrift_pchip(signal, k, variance, rng)`
Additive amplitude drift. A smooth curve centered near zero is superimposed on the signal, simulating baseline wander. Returns `signal + drift`.
**Example:**
```python
from augsig.warper import adrift_bezier, adrift_pchip
drifted = adrift_bezier(signal, k=4, variance=0.1)
drifted = adrift_pchip(signal, k=4, variance=0.1)
```
---
#### `amod_bezier(signal, k, variance, rng)` / `amod_pchip(signal, k, variance, rng)`
Multiplicative amplitude modulation. A smooth envelope fluctuating around unity is multiplied into the signal, simulating respiration-induced or pressure-induced amplitude variation. Returns `signal × envelope`.
**Example:**
```python
from augsig.warper import amod_bezier, amod_pchip
modulated = amod_bezier(signal, k=4, variance=0.1)
modulated = amod_pchip(signal, k=4, variance=0.1)
```
---
#### `drift_linear(signal, a, b, rng)`
Adds a linear baseline drift `y = a·t + b`. Slope and intercept can each be a fixed scalar or a `[min, max]` range sampled uniformly.
| Parameter | Type | Default | Description |
|---|---|---|---|
| `signal` | np.ndarray | — | 1D input signal |
| `a` | float or [min, max] | — | Slope or slope range |
| `b` | float or [min, max] | — | Intercept or intercept range |
| `rng` | np.random.Generator | `None` | Seeded RNG |
**Example: random slope and intercept**
```python
from augsig.warper import drift_linear
drifted = drift_linear(signal, a=[-0.3, 0.3], b=[-0.1, 0.1])
```
---
## Augmentation Pipeline
The `augment()` function and `Augment` class in `augmenter.py` compose all primitives above through a single configuration dictionary. This is the recommended interface for most use cases.
### Config Dict
Augmentation behavior is specified through a nested dictionary. Each top-level key names one augmentation recipe; recipes are independent and each produces one or more output columns.
**Example: config dict structure**
```python
config = {
"recipe_name": {
"num_copies": 2, # stochastic variants to generate from this recipe (default: 1)
"Add_noise": True, # enable noise
"SNRdb": 20,
# ... other keys
},
"another_recipe": { ... },
}
```
Within each recipe, transforms are applied in a fixed order: additive noise → flip → invert → time warp → amplitude drift → amplitude modulation → low-frequency noise → powerline → burst → linear drift → normalization.
**Full config key reference:**
| Key | Type | Description |
|---|---|---|
| `num_copies` | int | Number of variants from this recipe (default: 1) |
| **Noise** | | |
| `Add_noise` | bool | Enable SNR-controlled additive noise |
| `SNRdb` | float | Target SNR in dB |
| `noise_color` | str | `white`, `pink`, `brown`, `violet`, or `blue` |
| `bpass_params` | list | `[fl_norm, fh_norm]` or `[fl_norm, fh_norm, order]`; Nyquist-normalized (0–1) |
| `dist` | str | `gauss`, `uniform`, `laplace`, or `resample` |
| `resample_pool` | array or `"self"` | Required when `dist="resample"` |
| **Geometric** | | |
| `Flip` | bool | Reverse the signal in time |
| `Invert` | bool | Apply `max(signal) − signal` (vertical flip) |
| **Time warping** | | |
| `Bezier_time_warp` | bool | Nonlinear time warp via Bezier curve |
| `Bezier_time_warp_k` | int | Number of control points (default: 4) |
| `Bezier_time_warp_var` | float | Warp strength (default: 0.01) |
| `PCHIP_time_warp` | bool | Nonlinear time warp via PCHIP interpolation |
| `PCHIP_time_warp_k` | int | Number of knots (default: 4) |
| `PCHIP_time_warp_var` | float | Warp strength (default: 0.01) |
| **Amplitude drift** | | |
| `Bezier_amp_drift` | bool | Additive baseline drift via Bezier curve |
| `Bezier_amp_drift_k` | int | Number of control points (default: 4) |
| `Bezier_amp_drift_var` | float | Drift strength (default: 0.05) |
| `PCHIP_amp_drift` | bool | Additive baseline drift via PCHIP |
| `PCHIP_amp_drift_k` | int | Number of knots (default: 4) |
| `PCHIP_amp_drift_var` | float | Drift strength (default: 0.05) |
| **Amplitude modulation** | | |
| `Bezier_amp_mod` | bool | Multiplicative amplitude envelope via Bezier curve |
| `Bezier_amp_mod_k` | int | Number of control points (default: 4) |
| `Bezier_amp_mod_var` | float | Modulation depth (default: 0.05) |
| `PCHIP_amp_mod` | bool | Multiplicative amplitude envelope via PCHIP |
| `PCHIP_amp_mod_k` | int | Number of knots (default: 4) |
| `PCHIP_amp_mod_var` | float | Modulation depth (default: 0.05) |
| **Artifact noise** | | |
| `lf_noise` | bool | Low-frequency additive noise |
| `lf_noise_amplitude` | float | Amplitude scale (default: 0.1) |
| `lf_noise_frequency` | float | Upper cutoff normalized to Nyquist (default: 0.1) |
| `powerline` | bool | Sinusoidal powerline-like interference |
| `powerline_amplitude` | float | Amplitude (default: 0.1) |
| `powerline_frequency` | float | Frequency normalized to Nyquist (default: 0.1) |
| `burst` | bool | Burst artifact noise |
| `burst_amplitude` | float | Burst amplitude scale (default: 0.1) |
| `burst_number` | int | Number of bursts (default: 3) |
| `burst_width` | int or [min, max] | Burst width in samples (default: [1, 10]) |
| `burst_base` | float or [min, max] | DC offset inside each burst (default: 0.0) |
| `burst_frequency` | float | Upper cutoff for burst noise, normalized to Nyquist (default: 0.1) |
| `drift` | bool | Linear baseline drift |
| `drift_a` | float or [min, max] | Slope or slope range (default: [−0.5, 0.5]) |
| `drift_b` | float or [min, max] | Intercept or intercept range (default: [−0.5, 0.5]) |
> **Note:** every generated variant is min-max normalized to [0, 1] by default. Pass `normalize_output=False` to `augment()` or `Augment()` to preserve the original amplitude scale.
---
### Function Style
`augment(data, aug_config, seed=None, normalize_output=True)` returns an `(N, K)` array in one call. The first column is always the unmodified original; each subsequent column is one augmented variant.
**Example:**
```python
import numpy as np
from augsig import augment
signal = np.load("data/sample_ppg.npy")
config = {
"noise": {"Add_noise": True, "SNRdb": 20, "noise_color": "white",
"bpass_params": [0, 1], "dist": "gauss"},
"timewarp": {"Bezier_time_warp": True, "Bezier_time_warp_k": 4,
"Bezier_time_warp_var": 0.01},
"drift": {"drift": True, "drift_a": [-0.3, 0.3], "drift_b": [-0.1, 0.1]},
}
augmented = augment(signal, config, seed=42) # normalized to [0, 1]
augmented = augment(signal, config, normalize_output=False) # preserve amplitude
print(augmented.shape) # (N, 4): original + 3 variants
```
---
### Object Style
`Augment(config, seed=None, normalize_output=True)` binds the configuration at construction time, producing a reusable callable — convenient inside training loops or dataset classes.
**Example:**
```python
from augsig import Augment
augmentor = Augment(config, seed=42)
augmented = augmentor(signal) # equivalent to augment(signal, config, seed=42)
```
---
### Training-Time Augmentation
Omit the seed so each call produces a freshly sampled variant, giving the model a different augmented view across epochs.
**Example:**
```python
import torch
from augsig import Augment
aug_config = {
"noise": {"Add_noise": True, "SNRdb": 20, "noise_color": "pink",
"bpass_params": [0, 1], "dist": "gauss"},
"warp": {"Bezier_time_warp": True, "Bezier_time_warp_var": 0.01,
"Bezier_amp_drift": True, "Bezier_amp_drift_var": 0.05},
}
augmentor = Augment(aug_config) # no seed → different view each epoch
class PPGDataset(torch.utils.data.Dataset):
def __getitem__(self, idx):
x = self.data[idx] # (N,)
variants = augmentor(x) # (N, K)
return variants[:, 1] # one augmented variant per step
```
---
### Test-Time Augmentation (TTA)
Fix the seed so the augmentation bundle is identical across evaluation runs, then average model predictions over all variants to reduce prediction variance.
**Example:**
```python
import numpy as np
from augsig import augment
tta_config = {
"flip": {"Flip": True},
"noise": {"Add_noise": True, "SNRdb": 20, "noise_color": "white",
"bpass_params": [0, 1], "dist": "gauss", "num_copies": 3},
"warp": {"Bezier_time_warp": True, "Bezier_time_warp_var": 0.01},
}
variants = augment(signal, tta_config, seed=42) # (N, K), deterministic
preds = np.stack([model(variants[:, k]) for k in range(variants.shape[1])])
ensemble_pred = preds.mean(axis=0)
```
---
## Package Structure
```
augsig/
├── augsig/
│ ├── __init__.py # Re-exports Augment and augment
│ ├── augmenter.py # Augment class and augment() entry point
│ ├── noisifier.py # Additive noise, burst artifacts (noisify, burstify)
│ ├── warper.py # Bezier and PCHIP warping, linear drift
│ └── utils.py # Normalization and Butterworth filtering
├── tests/
│ ├── test.py # Informal demo and visualization script
│ └── test_augsig.py # pytest unit tests
├── data/
│ └── sample_ppg.npy # 30-second PPG segment from BIDMC (PhysioNet, ODC-BY), record bidmc01 PLETH channel, resampled to 40 Hz
├── requirements.txt
├── pyproject.toml # Packaging config (setuptools, Python >= 3.8)
├── LICENSE
└── AUTHORS.txt
```
---
## License
BSD 3-Clause License
---
## Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.