API Reference 

Examples

>>> bp_filter = BandpassFilter("butter", fs=100)
>>> b, a = bp_filter.signal_bypass(cutoff=0.3, order=4, a_pass=3, rp=4, rs=40, btype='low')
>>> print(b, a)

signal_highpass_filter(data, cutoff, order=5, a_pass=3, rp=4, rs=40)[source]

Apply a high-pass filter to the data.

Parameters:

data (numpy.ndarray) – The input signal.
cutoff (float) – Cutoff frequency of the filter.
order (int, optional) – Order of the filter. Default is 5.
a_pass (float, optional) – Passband ripple for Chebyshev and Elliptic filters. Default is 3.
rp (float, optional) – Passband ripple for Elliptic filters. Default is 4.
rs (float, optional) – Stopband attenuation for Elliptic filters. Default is 40.

Returns:

y – The filtered signal.

Return type:

Raises:

ValueError – If the length of the input signal is too short for the specified filter.

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> bp_filter = BandpassFilter("butter", fs=100)
>>> filtered_signal = bp_filter.signal_highpass_filter(signal, cutoff=0.3, order=4)
>>> print(filtered_signal)

signal_lowpass_filter(data, cutoff, order=3, a_pass=3, rp=4, rs=40)[source]

Apply a low-pass filter to the data.

Parameters:

data (numpy.ndarray) – The input signal.
cutoff (float) – Cutoff frequency of the filter.
order (int, optional) – Order of the filter. Default is 3.
a_pass (float, optional) – Passband ripple for Chebyshev and Elliptic filters. Default is 3.
rp (float, optional) – Passband ripple for Elliptic filters. Default is 4.
rs (float, optional) – Stopband attenuation for Elliptic filters. Default is 40.

Returns:

y – The filtered signal.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> bp_filter = BandpassFilter("butter", fs=100)
>>> filtered_signal = bp_filter.signal_lowpass_filter(signal, cutoff=0.3, order=4)
>>> print(filtered_signal)

class vitalDSP.filtering.signal_filtering.SignalFiltering(signal)[source]

Bases: object

A class for applying various filtering techniques to signals.

This class provides methods for common signal filtering tasks, such as applying moving averages, Gaussian filters, Butterworth filters, and median filters. These techniques are essential for preprocessing signals in various fields, including biomedical signal processing (e.g., ECG, EEG).

savgol_filter : static method: Applies a Savitzky-Golay filter.

moving_average : function: Applies a moving average filter.

gaussian : function: Applies a Gaussian filter.

butterworth : function: Applies a Butterworth filter.

chebyshev : function: Applies a Chebyshev filter.

elliptic : function: Applies an Elliptic filter.

bandpass : function: Applies a bandpass filter using a selected filter type.

median : function: Applies a median filter.

_apply_iir_filter : function: Internal method to apply IIR filters.

Examples

>>> import numpy as np
>>> from vitalDSP.filtering.signal_filtering import SignalFiltering
>>>
>>> # Example 1: Basic signal filtering
>>> signal = np.random.randn(1000)  # Simulated signal
>>> sf = SignalFiltering(signal)
>>> filtered_signal = sf.bandpass(lowcut=0.5, highcut=30, fs=256, order=4)
>>> print(f"Filtered signal shape: {filtered_signal.shape}")
>>>
>>> # Example 2: Moving average filtering
>>> ma_filtered = sf.moving_average(window_size=5)
>>> print(f"Moving average filtered: {ma_filtered.shape}")
>>>
>>> # Example 3: Gaussian filtering
>>> gaussian_filtered = sf.gaussian(sigma=1.0)
>>> print(f"Gaussian filtered: {gaussian_filtered.shape}")
>>>
>>> # Example 4: Savitzky-Golay filtering
>>> sg_filtered = SignalFiltering.savgol_filter(signal, window_length=5, polyorder=2)
>>> print(f"Savitzky-Golay filtered: {sg_filtered.shape}")

bandpass(lowcut, highcut, fs, order=4, filter_type='butter', iterations=1)[source]

Apply a bandpass filter using the selected filter type.

Parameters:

lowcut (float) – The lower cutoff frequency.
highcut (float) – The upper cutoff frequency.
fs (float) – The sampling frequency.
order (int, optional) – The order of the filter. Default is 4.
filter_type (str, optional) – Type of filter to apply (‘butter’, ‘cheby’, ‘elliptic’). Default is ‘butter’.

Returns:

filtered_signal – The filtered signal.

Return type:

Examples

>>> import numpy as np
>>> fs = 1000  # Sampling frequency
>>> lowcut = 0.5  # Lower cutoff frequency
>>> highcut = 50  # Upper cutoff frequency
>>> signal = np.sin(2 * np.pi * 10 * np.arange(0, 10, 1/fs)) + np.random.randn(10 * fs) * 0.1
>>> sf = SignalFiltering(signal)
>>> filtered_signal = sf.bandpass(lowcut, highcut, fs, order=4, filter_type='butter')
>>> print(filtered_signal)

bessel(cutoff, fs, order=4, btype='low', iterations=1)[source]

Bessel (Thomson) filter implementation.

The Bessel filter has a maximally flat group delay, which means it preserves the waveform shape of filtered signals in the passband. This makes it ideal for applications where maintaining pulse shape is critical, such as ECG and PPG signal processing.

Parameters:

cutoff (float or list) – Cutoff frequency of the filter. For bandpass/bandstop, provide [low, high].
fs (float) – Sampling frequency of the signal.
order (int, optional) – Order of the Bessel filter. Default is 4.
btype (str, optional) – Type of filter - ‘low’, ‘high’, ‘band’, or ‘bandstop’. Default is ‘low’.
iterations (int, optional) – The number of times to apply the filter. Default is 1.

Returns:

filtered_signal – The filtered signal.

Return type:

Notes

The Bessel filter is particularly useful for: - ECG signal processing (preserves QRS complex shape) - PPG signal processing (preserves pulse waveform) - Applications requiring linear phase response - Situations where overshoot and ringing must be minimized

Examples

>>> import numpy as np
>>> from vitalDSP.filtering.signal_filtering import SignalFiltering
>>> # ECG-like signal with sharp peaks
>>> t = np.linspace(0, 1, 1000)
>>> ecg_signal = np.sin(2 * np.pi * 1.2 * t) + 0.5 * np.sin(2 * np.pi * 60 * t)
>>> sf = SignalFiltering(ecg_signal)
>>> # Apply Bessel lowpass to remove high-frequency noise while preserving peak shape
>>> filtered = sf.bessel(cutoff=50, fs=1000, order=4, btype='low')
>>> print(f"Filtered signal shape: {filtered.shape}")

butter(order, cutoff, btype='low', fs=1.0)[source]

Custom implementation of the Butterworth filter design using bilinear transformation.

Parameters:

order (int) – The order of the filter.
cutoff (float or list of float) – The critical frequency or frequencies.
btype (str) – The type of filter (‘low’, ‘high’, ‘band’).
fs (float) – The sampling frequency.

Returns:

b, a – Numerator (b) and denominator (a) polynomials of the IIR filter.

Return type:

Examples

>>> import numpy as np
>>> b, a = SignalFiltering().butter(4, 0.3, btype='low', fs=1.0)
>>> print(b, a)

butterworth(cutoff, fs, order=4, btype='low', iterations=1, adaptive=True)[source]

Apply a Butterworth filter to the signal.

Parameters:

cutoff (float) – Cutoff frequency of the filter.
fs (float) – Sampling frequency of the signal.
order (int, optional) – Order of the Butterworth filter. Default is 4.
btype (str, optional) – Type of filter - ‘low’ or ‘high’. Default is ‘low’.
iterations (int, optional) – The number of times to apply the Butterworth filter for additional filtering. Default is 1.
adaptive (bool, optional) – Whether to use adaptive parameter adjustment. Default is True.

Returns:

filtered_signal – The filtered signal.

Return type:

Examples

>>> import numpy as np
>>> fs = 1000  # Sampling frequency
>>> cutoff = 0.5  # Cutoff frequency
>>> signal = np.sin(2 * np.pi * 0.1 * np.arange(0, 10, 1/fs)) + np.random.randn(10 * fs) * 0.1
>>> sf = SignalFiltering(signal)
>>> filtered_signal = sf.butterworth(cutoff, fs, order=4)
>>> print(filtered_signal)

chebyshev(cutoff, fs, order=4, btype='low', ripple=0.05, iterations=1)[source]

Custom implementation of the Chebyshev Type I filter.

Parameters:

cutoff (float) – Cutoff frequency of the filter.
fs (float) – Sampling frequency of the signal.
order (int, optional) – Order of the Chebyshev filter. Default is 4.
btype (str, optional) – Type of filter - ‘low’ or ‘high’. Default is ‘low’.
ripple (float, optional) – The maximum ripple allowed in the passband. Default is 0.05.
iterations (int, optional) – The number of times to apply the Chebyshev filter for additional filtering. Default is 1.

Returns:

filtered_signal – The filtered signal.

Return type:

chebyshev2(cutoff, fs, order=4, btype='low', stopband_attenuation=40, iterations=1)[source]

Chebyshev Type II filter implementation.

The Chebyshev Type II filter has a flat passband and equiripple stopband attenuation. Unlike Type I, it has zeros in the stopband which provides sharper roll-off characteristics. This filter is useful when you need better stopband attenuation with acceptable passband characteristics.

Parameters:

cutoff (float or list) – Cutoff frequency of the filter. For bandpass/bandstop, provide [low, high].
fs (float) – Sampling frequency of the signal.
order (int, optional) – Order of the Chebyshev Type II filter. Default is 4.
btype (str, optional) – Type of filter - ‘low’, ‘high’, ‘band’, or ‘bandstop’. Default is ‘low’.
stopband_attenuation (float, optional) – Minimum attenuation required in the stopband in dB. Default is 40 dB.
iterations (int, optional) – The number of times to apply the filter. Default is 1.

Returns:

filtered_signal – The filtered signal.

Return type:

Examples

>>> import numpy as np
>>> from vitalDSP.filtering.signal_filtering import SignalFiltering
>>> signal = np.random.randn(1000)
>>> sf = SignalFiltering(signal)
>>> filtered = sf.chebyshev2(cutoff=50, fs=250, order=4, btype='low', stopband_attenuation=40)
>>> print(f"Filtered signal shape: {filtered.shape}")

elliptic(cutoff, fs, order=4, btype='low', ripple=0.05, stopband_attenuation=40, iterations=1)[source]

Custom implementation of the Elliptic filter.

Parameters:

cutoff (float) – Cutoff frequency of the filter.
fs (float) – Sampling frequency of the signal.
order (int, optional) – Order of the Elliptic filter. Default is 4.
btype (str, optional) – Type of filter - ‘low’ or ‘high’. Default is ‘low’.
ripple (float, optional) – The maximum ripple allowed in the passband. Default is 0.05.
stopband_attenuation (float, optional) – Minimum attenuation in the stopband. Default is 40 dB.
iterations (int, optional) – The number of times to apply the Elliptic filter for additional filtering. Default is 1.

Returns:

filtered_signal – The filtered signal.

Return type:

gaussian(sigma=1.0, iterations=1)[source]

Applies a Gaussian filter to the signal.

Parameters:

sigma (float) – The standard deviation of the Gaussian kernel.
iterations (int, optional) – The number of times to apply the Gaussian filter for additional smoothing. Default is 1.

Returns:

filtered_signal – The filtered signal.

Return type:

Examples

>>> import numpy as np
>>> signal = np.array([1, 2, 3, 4, 5])
>>> sf = SignalFiltering(signal)
>>> filtered_signal = sf.gaussian(1.5)
>>> print(filtered_signal)

static gaussian_filter1d(signal, sigma)[source]

Custom implementation of a 1D Gaussian filter.

Parameters:

signal (numpy.ndarray) – The input signal.
sigma (float) – The standard deviation of the Gaussian kernel.

Returns:

smoothed_signal – The smoothed signal.

Return type:

Examples

>>> import numpy as np
>>> signal = np.array([1, 2, 3, 4, 5])
>>> smoothed_signal = SignalFiltering.gaussian_filter1d(signal, 1.0)
>>> print(smoothed_signal)
[1.14285714 2.14285714 3. 4. 5.]

static gaussian_kernel(size, sigma)[source]

Generate a Gaussian kernel.

Parameters:

size (int) – The size of the kernel (must be odd).
sigma (float) – Standard deviation of the Gaussian distribution.

Returns:

kernel – The Gaussian kernel.

Return type:

Examples

>>> kernel = SignalFiltering.gaussian_kernel(5, sigma=1.0)
>>> print(kernel)
[0.05448868 0.24420134 0.40261995 0.24420134 0.05448868]

median(kernel_size=3, iterations=1, method='edge')[source]

Apply a median filter to the signal with optional repeated filtering.

Parameters:

kernel_size (int, optional) – Size of the median filter kernel. Default is 3.
iterations (int, optional) – The number of times to apply the median filter for additional filtering. Default is 1.
method (str, optional) – Padding method: ‘edge’ (default), ‘reflect’, or ‘constant’.

Returns:

filtered_signal – The filtered signal.

Return type:

Examples

>>> import numpy as np
>>> signal = np.array([1, 2, 3, 100, 5, 6, 7, 8, 9, 10])  # Note the spike at value 100
>>> sf = SignalFiltering(signal)
>>> filtered_signal = sf.median(kernel_size=3)
>>> print(filtered_signal)
[1 2 3 5 5 6 7 8 9 9]

moving_average(window_size, iterations=1, method='edge')[source]

Applies a moving average filter to the signal with optional repeated scanning.

This method smooths the signal by averaging neighboring data points within a defined window size. Optionally, the smoothing can be repeated multiple times for enhanced effect. This technique is commonly used to reduce random noise and reveal trends in signals like EEG, ECG, and PPG.

Parameters:

window_size (int) – The size of the moving window.
iterations (int, optional) – The number of times to apply the moving average for additional smoothing. Default is 1.
method (str, optional) – Padding method: ‘edge’ (default), ‘reflect’, or ‘constant’. Different methods may yield better results for specific types of signals (e.g., vital signals like EEG, ECG, PPG).

Returns:

filtered_signal – The filtered signal.

Return type:

Examples

>>> import numpy as np
>>> signal = np.array([1, 2, 3, 4, 5])
>>> sf = SignalFiltering(signal)
>>> filtered_signal = sf.moving_average(3, iterations=2, method="reflect")
>>> print(filtered_signal)

static savgol_filter(signal, window_length, polyorder)[source]

Apply a Savitzky-Golay filter to smooth the signal.

Parameters:

signal (numpy.ndarray) – The input signal.
window_length (int) – The length of the filter window (must be odd).
polyorder (int) – The order of the polynomial to fit.

Returns:

smoothed_signal – The smoothed signal.

Return type:

Examples

>>> import numpy as np
>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> filter = SignalFiltering(signal)
>>> smoothed_signal = filter.savgol_filter(signal, 3, 2)
>>> print(smoothed_signal)
[1. 2. 3. 4. 5. 6. 7. 8. 9.]

Artifact Removal

Signal Filtering Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.filtering.artifact_removal import ArtifactRemoval
>>> signal = np.random.randn(1000)
>>> ar = ArtifactRemoval(signal, fs=256)
>>> cleaned = ar.baseline_correction()
>>> print(f'Processing result: {result}')

class vitalDSP.filtering.artifact_removal.ArtifactRemoval(signal)[source]

Bases: object

A class for removing various types of artifacts from signals.

mean_subtraction : function: Removes artifacts by subtracting the mean of the signal.

baseline_correction : function: Corrects baseline drift by applying a high-pass filter.

median_filter_removal : function: Removes spike artifacts using a median filter.

wavelet_denoising : function: Removes noise using wavelet-based denoising with various mother wavelets.

adaptive_filtering : function: Uses an adaptive filter to remove artifacts correlated with reference signals.

notch_filter : function: Removes powerline interference using a notch filter.

pca_artifact_removal : function: Uses Principal Component Analysis (PCA) to remove artifacts.

ica_artifact_removal : function: Uses Independent Component Analysis (ICA) to remove artifacts using NumPy.

adaptive_filtering(reference_signal=None, learning_rate=0.01, num_iterations=100)[source]

Use an adaptive filter to remove artifacts correlated with a reference signal.

This method uses Least Mean Squares (LMS) adaptive filtering to iteratively adjust the signal to minimize the error between the filtered signal and a reference signal. It is particularly useful for removing artifacts that are correlated with another signal, such as EOG artifacts in EEG recordings, motion artifacts in PPG, or respiratory artifacts.

If no reference signal is provided, the filter adapts towards zero (artifact removal/denoising).

Parameters:

reference_signal (numpy.ndarray, optional) – The reference signal correlated with the artifact. If None, adapts towards zero (removes DC offset and baseline drift). Must have the same length as the input signal.
learning_rate (float, default=0.01) – The learning rate (step size) for the adaptive filter. Controls convergence speed. Typical range: 0.001 - 0.5 - Lower values (0.001-0.01): Slower convergence, more stable - Higher values (0.1-0.5): Faster convergence, may oscillate
num_iterations (int, default=100) – The number of iterations for adaptation. More iterations = better convergence.

Returns:

clean_signal – The artifact-removed signal with the same length as the input.

Return type:

Raises:

ValueError – If reference_signal length doesn’t match signal length.

Notes

Algorithm: Least Mean Squares (LMS) adaptive filtering - The filter iteratively adjusts the signal based on the error - Error = filtered_signal - reference_signal - Update rule: filtered_signal -= learning_rate * error

Use Cases: 1. With reference signal: Remove correlated artifacts (EOG from EEG, motion from PPG) 2. Without reference signal: General denoising and DC offset removal

Convergence: - Monitor the error reduction to ensure proper convergence - If the filter diverges (error increases), reduce the learning rate - Typical convergence: 50-200 iterations with learning_rate=0.01

Advanced Signal Filtering

Advanced Signal Filtering Module for Physiological Signal Processing

This module implements advanced filtering techniques for physiological signals including Kalman filtering, optimization-based filtering, gradient descent filtering, ensemble filtering, convolution-based filtering, and attention-based filtering. These methods are designed for complex scenarios such as real-time filtering, adaptive filtering, and filtering in noisy environments.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Kalman filtering for real-time signal estimation - Optimization-based filtering with custom loss functions - Gradient descent adaptive filtering - Ensemble filtering combining multiple methods - Convolution-based filtering with custom kernels - Attention-based dynamic filtering

Examples:

Basic Kalman filtering:

>>> import numpy as np
>>> from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
>>> signal = np.random.randn(1000) + np.sin(np.linspace(0, 10, 1000))
>>> af = AdvancedSignalFiltering(signal)
>>> filtered = af.kalman_filter(R=0.1, Q=0.01)

Optimization-based filtering:

>>> target = np.sin(np.linspace(0, 10, 1000))
>>> filtered = af.optimization_based_filtering(target, loss_type="mse", iterations=50)

Ensemble filtering:

>>> filters = ['kalman', 'optimization', 'gradient_descent']
>>> ensemble_result = af.ensemble_filtering(filters, method="mean", num_iterations=5)

class vitalDSP.filtering.advanced_signal_filtering.AdvancedSignalFiltering(signal)[source]

Bases: object

A class for applying advanced filtering techniques to signals.

This class implements a variety of advanced filtering methods, including Kalman filters, optimization-based filtering, gradient descent filtering, ensemble filtering, convolution-based filtering, and attention-based filtering. These methods are designed to enhance signal processing in complex scenarios such as real-time filtering, adaptive filtering, and filtering in noisy environments.

kalman_filter(R=1, Q=1)[source]: Applies a Kalman filter for real-time signal estimation and noise reduction.

optimization_based_filtering(target, loss_type='mse', custom_loss_func=None, initial_guess=0, learning_rate=0.01, iterations=100)[source]: Applies filtering based on a custom or predefined loss function using optimization techniques.

gradient_descent_filter(target, learning_rate=0.01, iterations=100)[source]: Uses gradient descent for adaptive filtering to reduce the difference between the signal and a target.

ensemble_filtering(filters, method='mean', weights=None, num_iterations=10, learning_rate=0.01)[source]: Combines multiple filters using ensemble methods such as mean, weighted mean, bagging, or boosting.

convolution_based_filter(kernel_type='smoothing', custom_kernel=None, kernel_size=3)[source]: Applies convolutional filtering using predefined or custom kernels.

attention_based_filter(attention_type="uniform", custom_weights=None, size=5, \*\*kwargs)[source]: Uses attention mechanisms for dynamic filtering based on various attention schemes.

adaptive_filtering(desired_signal, mu=0.01, filter_order=4)[source]

Apply an adaptive filter to the signal using the Least Mean Squares (LMS) algorithm.

The LMS adaptive filter is useful for real-time signal processing tasks where the characteristics of the signal may change over time. The filter coefficients are updated iteratively to minimize the error between the desired signal and the filter output.

Parameters:

desired_signal (numpy.ndarray) – The desired signal that the filter output should approximate.
mu (float, optional) – The step size or learning rate for the LMS algorithm. Default is 0.01.
filter_order (int, optional) – The order of the adaptive filter. Default is 4.

Returns:

filtered_signal – The filtered signal after applying the LMS adaptive filter.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6])
>>> desired_signal = np.array([1.1, 1.9, 3.1, 4.1, 5.0, 5.9])
>>> af = AdvancedSignalFiltering(signal)
>>> filtered_signal = af.adaptive_filtering(desired_signal, mu=0.01, filter_order=4)
>>> print(filtered_signal)

attention_based_filter(attention_type='uniform', custom_weights=None, size=5, **kwargs)[source]

Apply an attention-based filter to the signal using predefined or custom attention weights.

Attention-based filtering uses an attention mechanism to dynamically adjust the influence of different parts of the signal. This method supports various types of attention mechanisms, including uniform, linear, gaussian, and exponential, as well as custom-defined attention weights.

Parameters:

attention_type (str) – The type of attention weights to use (‘uniform’, ‘linear’, ‘gaussian’, ‘exponential’, ‘custom’).
custom_weights (numpy.ndarray, optional) – The custom attention weights to use if ‘custom’ is selected.
size (int, optional) – The size of the attention window for predefined attention weights. Default is 5.
**kwargs (dict, optional) – Additional parameters for specific attention types (e.g., ‘ascending’ for linear, ‘sigma’ for gaussian, ‘base’ for exponential).

Returns:

filtered_signal – The filtered signal after applying the attention-based filter.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6])
>>> af = AdvancedSignalFiltering(signal)
>>> filtered_signal = af.attention_based_filter(attention_type='gaussian', size=5, sigma=1.0)
>>> print(filtered_signal)

convolution_based_filter(kernel_type='smoothing', custom_kernel=None, kernel_size=3)[source]

Apply a convolution-based filter to the signal using predefined or custom kernels.

Convolution-based filtering is a powerful technique that applies a kernel to the signal to achieve various effects such as smoothing, sharpening, and edge detection. This method allows for the use of standard convolutional kernels or custom kernels provided by the user.

Parameters:

kernel_type (str) – The type of kernel to use (‘smoothing’, ‘sharpening’, ‘edge_detection’, ‘custom’).
custom_kernel (numpy.ndarray, optional) – The custom kernel to use if ‘custom’ is selected. This must be provided by the user.
kernel_size (int, optional) – The size of the kernel for predefined kernels. Default is 3.

Returns:

filtered_signal – The filtered signal after convolution.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6])
>>> af = AdvancedSignalFiltering(signal)
>>> filtered_signal = af.convolution_based_filter(kernel_type='sharpening')
>>> print(filtered_signal)

ensemble_filtering(filters, method='mean', weights=None, num_iterations=10, learning_rate=0.01)[source]

Apply ensemble filtering by combining the results of multiple filters using various ensemble techniques.

This method aggregates the results of different filtering techniques to achieve a more robust and reliable filtered signal. Ensemble methods like mean, weighted mean, bagging, and boosting are implemented to leverage the strengths of multiple filters.

Parameters:

filters (list of callable) – A list of filtering functions to apply. Each function should return a filtered signal.
method (str) – The ensemble method to use (‘mean’, ‘weighted_mean’, ‘bagging’, ‘boosting’).
weights (list of float, optional) – Weights for weighted mean if ‘weighted_mean’ is chosen. Should match the number of filters.
num_iterations (int, optional) – Number of iterations for bagging/boosting. Default is 10.
learning_rate (float, optional) – Learning rate for boosting. Default is 0.01.

Returns:

filtered_signal – The ensemble filtered signal.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6])
>>> af = AdvancedSignalFiltering(signal)
>>> filters = [af.kalman_filter, af.kalman_filter]  # Example with the same filter twice
>>> filtered_signal = af.ensemble_filtering(filters, method='mean')
>>> print(filtered_signal)

gradient_descent_filter(target=None, learning_rate=0.01, iterations=100)[source]

Apply a gradient descent adaptive filter to the signal.

This method uses gradient descent to iteratively minimize the difference between the signal and a target signal. It is useful in scenarios where the signal needs to be progressively adjusted to match a desired outcome. If no target is provided, the filter will adapt towards zero (denoising).

Adaptive Filtering: This is a true adaptive filter that adjusts the signal iteratively based on the gradient of the error between the filtered signal and the target. The learning rate controls how quickly the filter adapts, and the number of iterations determines convergence.

Parameters:

target (numpy.ndarray, optional) – The target signal to adapt to. If None, adapts towards zero (removes DC offset and trends). Must have the same length as the input signal.
learning_rate (float, default=0.01) – Learning rate for the gradient descent, controlling how much the signal is adjusted at each step. Typical range: 0.001 - 0.5 - Lower values (0.001-0.01): Slower convergence, more stable - Higher values (0.1-0.5): Faster convergence, may oscillate
iterations (int, default=100) – Number of iterations for the gradient descent process. More iterations = better convergence to target, but slower processing

Returns:

filtered_signal – The filtered signal after applying gradient descent adaptive filtering.

Return type:

Raises:

ValueError – If target is provided but has different length than signal

Examples

Example 1: Adaptive filtering towards a target (basic)

>>> import numpy as np
>>> from vitalDSP.filtering.advanced_signal_filtering import AdvancedSignalFiltering
>>>
>>> # Create noisy ECG-like signal
>>> t = np.linspace(0, 2, 200)
>>> clean_signal = np.sin(2 * np.pi * 1.2 * t)  # Simulated ECG
>>> noisy_signal = clean_signal + 0.3 * np.random.randn(len(t))  # Add noise
>>>
>>> # Apply adaptive filtering
>>> af = AdvancedSignalFiltering(noisy_signal)
>>> filtered = af.gradient_descent_filter(target=clean_signal, learning_rate=0.1, iterations=100)
>>>
>>> print(f"Original signal std: {np.std(noisy_signal):.3f}")
>>> print(f"Filtered signal std: {np.std(filtered):.3f}")
>>> print(f"MSE reduction: {np.mean((noisy_signal - clean_signal)**2) / np.mean((filtered - clean_signal)**2):.2f}x")

Example 2: Denoising without target (remove DC and trends)

>>> # Signal with DC offset and noise
>>> signal = np.sin(2 * np.pi * 0.5 * np.linspace(0, 4, 200)) + 2.0  # DC offset = 2.0
>>> signal += 0.2 * np.random.randn(len(signal))  # Add noise
>>>
>>> af = AdvancedSignalFiltering(signal)
>>> denoised = af.gradient_descent_filter(learning_rate=0.05, iterations=200)
>>>
>>> print(f"Original mean (DC): {np.mean(signal):.3f}")
>>> print(f"Filtered mean (DC removed): {np.mean(denoised):.3f}")

Example 3: Real-time PPG signal processing

>>> # Simulated PPG signal with baseline wander
>>> t = np.linspace(0, 10, 1000)
>>> ppg_clean = np.sin(2 * np.pi * 1.0 * t)  # Heart rate ~60 BPM
>>> baseline_wander = 0.3 * np.sin(2 * np.pi * 0.1 * t)  # Slow drift
>>> ppg_noisy = ppg_clean + baseline_wander + 0.1 * np.random.randn(len(t))
>>>
>>> # Remove baseline using adaptive filter
>>> af = AdvancedSignalFiltering(ppg_noisy)
>>> ppg_filtered = af.gradient_descent_filter(
...     target=ppg_clean,
...     learning_rate=0.08,
...     iterations=150
... )
>>>
>>> # Calculate improvement
>>> snr_before = 10 * np.log10(np.var(ppg_clean) / np.var(ppg_noisy - ppg_clean))
>>> snr_after = 10 * np.log10(np.var(ppg_clean) / np.var(ppg_filtered - ppg_clean))
>>> print(f"SNR improvement: {snr_after - snr_before:.2f} dB")

Example 4: Respiratory signal adaptive filtering

>>> # Respiratory signal from PPG/ECG amplitude modulation
>>> t = np.linspace(0, 30, 3000)  # 30 seconds at 100 Hz
>>> respiratory = 0.2 * np.sin(2 * np.pi * 0.25 * t)  # 15 breaths/min
>>> noise = 0.1 * np.random.randn(len(t))
>>> signal = respiratory + noise
>>>
>>> # Adaptive filtering to extract respiratory pattern
>>> af = AdvancedSignalFiltering(signal)
>>> filtered = af.gradient_descent_filter(
...     target=respiratory,
...     learning_rate=0.05,
...     iterations=100
... )
>>>
>>> print(f"Respiratory rate estimation error: {abs(np.mean(filtered) - np.mean(respiratory)):.4f}")

Notes

The gradient descent filter is a simple adaptive filter that works well for gradual changes
For rapid signal changes, use lower learning rates and more iterations
This implementation uses the sign of the gradient (sign function) which makes it robust to outliers
For better results with complex signals, consider using Kalman filtering or ensemble methods
The filter converges to the target signal as iterations increase
If target is None, the filter will remove trends and DC offset (adapt towards zero)

Physiological Features Module

Time Domain Features

Time Domain Features Module for Physiological Signal Processing

This module provides comprehensive time-domain feature extraction capabilities for physiological signals such as ECG and PPG. It implements standard Heart Rate Variability (HRV) metrics and additional time-domain analysis methods for characterizing signal variability and patterns.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Standard HRV metrics (SDNN, RMSSD, NN50, pNN50) - Advanced time-domain statistics (median, IQR, CVNN) - HRV Triangular Index and TINN computation - Successive difference analysis (SDSD) - Comprehensive NN interval analysis

Examples:

Basic HRV analysis:

>>> import numpy as np
>>> from vitalDSP.physiological_features.time_domain import TimeDomainFeatures
>>> nn_intervals = [800, 810, 790, 805, 795, 820, 780, 815, 800, 810]
>>> tdf = TimeDomainFeatures(nn_intervals)
>>> sdnn = tdf.compute_sdnn()
>>> rmssd = tdf.compute_rmssd()
>>> print(f"SDNN: {sdnn:.2f} ms, RMSSD: {rmssd:.2f} ms")

Advanced metrics:

>>> pnn50 = tdf.compute_pnn50()
>>> cvnn = tdf.compute_cvnn()
>>> hrv_triangular = tdf.compute_hrv_triangular_index()
>>> print(f"pNN50: {pnn50:.2f}%, CVNN: {cvnn:.2f}")

Statistical analysis:

>>> median_nn = tdf.compute_median_nn()
>>> iqr_nn = tdf.compute_iqr_nn()
>>> print(f"Median NN: {median_nn:.2f} ms, IQR: {iqr_nn:.2f} ms")

class vitalDSP.physiological_features.time_domain.TimeDomainFeatures(nn_intervals)[source]

Bases: object

A class for computing time-domain features from physiological signals (ECG, PPG).

nn_intervals

The NN intervals (in milliseconds) between heartbeats.

Type:: list or np.array

compute_sdnn()[source]: Computes the standard deviation of NN intervals (SDNN).

compute_rmssd()[source]: Computes the root mean square of successive differences (RMSSD).

compute_nn50()[source]: Computes the number of successive NN intervals differing by >50 ms.

compute_pnn50()[source]: Computes the percentage of NN50 over the total NN intervals.

compute_median_nn()[source]: Computes the median NN interval.

compute_iqr_nn()[source]: Computes the interquartile range (IQR) of NN intervals.

compute_mean_nn()[source]: Computes the mean of NN intervals.

compute_std_nn()[source]: Computes the standard deviation of NN intervals (alias for SDNN).

compute_pnn20()[source]: Computes the percentage of NN intervals differing by more than 20 ms.

compute_nn20(): Computes the count of NN intervals differing by more than 20 ms.

compute_cvnn()[source]: Computes the coefficient of variation of NN intervals (CVNN).

compute_hrv_triangular_index()[source]: Computes the HRV Triangular Index based on the histogram of NN intervals.

compute_tinn()[source]: Computes TINN (Triangular Interpolation of NN Interval Histogram).

compute_sdsd()[source]: Computes the standard deviation of successive differences (SDSD).

compute_cvnn()[source]

Computes the coefficient of variation of NN intervals (CVNN), which is the ratio of the standard deviation to the mean NN interval.

Returns:: The CVNN value.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_cvnn()
0.009354143466934854

compute_hrv_triangular_index()[source]

Computes the HRV Triangular Index, which is the total number of NN intervals divided by the height of the histogram of all NN intervals.

Returns:: The HRV Triangular Index.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_hrv_triangular_index()
5.0

compute_iqr_nn()[source]

Computes the interquartile range (IQR) of the NN intervals.

Returns:: The IQR of NN intervals.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_iqr_nn()
10.0

compute_mean_nn()[source]

Computes the mean of NN intervals.

Returns:: The mean NN interval.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_mean_nn()
800.0

compute_median_nn()[source]

Computes the median of the NN intervals.

Returns:: The median NN interval.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_median_nn()
800.0

compute_nn50()[source]

Computes the number of successive NN intervals differing by more than 50 ms (NN50).

Returns:: The NN50 value.
Return type:: int

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_nn50()
1

compute_pnn20()[source]

Computes the percentage of successive NN intervals differing by more than 20 ms (pNN20).

Returns:: The pNN20 value as a percentage.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_pnn20()
40.0

compute_pnn50()[source]

Computes the percentage of NN50 over the total number of NN intervals (pNN50).

Returns:: The pNN50 value as a percentage.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_pnn50()
20.0

compute_rmssd()[source]

Computes the root mean square of successive differences (RMSSD).

Returns:: The RMSSD value.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_rmssd()
10.0

compute_sdnn()[source]

Computes the standard deviation of NN intervals (SDNN).

Returns:: The SDNN value.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_sdnn()
7.483314773547883

compute_sdsd()[source]

Computes the standard deviation of successive differences (SDSD) between NN intervals.

Returns:: The SDSD value.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_sdsd()
10.0

compute_std_nn()[source]

Computes the standard deviation of NN intervals (alias for SDNN).

Returns:: The standard deviation of NN intervals (SDNN).
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_std_nn()
7.483314773547883

compute_tinn()[source]

Computes the Triangular Interpolation of NN Interval Histogram (TINN).

TINN is the baseline width of the NN interval histogram triangle, computed as the difference between the points N and M on the x-axis where the triangular interpolation reaches the baseline.

Returns:: The TINN value in the same units as nn_intervals.
Return type:: float

Example

>>> tdf = TimeDomainFeatures([800, 810, 790, 805, 795])
>>> tdf.compute_tinn()
20.0

Frequency Domain Features

Frequency Domain Features Module for Physiological Signal Processing

This module provides comprehensive frequency-domain feature extraction capabilities for physiological signals including ECG, PPG, and other vital signs. It implements Heart Rate Variability (HRV) analysis in the frequency domain with power spectral density computation and autonomic nervous system assessment.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Power Spectral Density (PSD) computation - Low-Frequency (LF) and High-Frequency (HF) power analysis - LF/HF ratio for autonomic balance assessment - Ultra-Low-Frequency (ULF) and Very-Low-Frequency (VLF) analysis - Normalized frequency domain metrics (LFnu, HFnu) - Total power computation across frequency bands - Comprehensive HRV frequency domain analysis

Examples:

Basic frequency domain analysis:

>>> import numpy as np
>>> from vitalDSP.physiological_features.frequency_domain import FrequencyDomainFeatures
>>> nn_intervals = [800, 810, 790, 805, 795, 820, 780, 815, 800, 810]
>>> fdf = FrequencyDomainFeatures(nn_intervals, fs=4)
>>> psd_result = fdf.compute_psd()
>>> print(f"LF: {psd_result['lf']:.2f}, HF: {psd_result['hf']:.2f}")

Autonomic balance assessment:

>>> lf_hf_ratio = fdf.compute_lf_hf_ratio()
>>> lfnu = fdf.compute_lfnu()
>>> hfnu = fdf.compute_hfnu()
>>> print(f"LF/HF ratio: {lf_hf_ratio:.2f}, LFnu: {lfnu:.2f}, HFnu: {hfnu:.2f}")

Comprehensive frequency analysis:

>>> total_power = fdf.compute_total_power()
>>> ulf = fdf.compute_ulf()
>>> vlf = fdf.compute_vlf()
>>> print(f"Total power: {total_power:.2f}, ULF: {ulf:.2f}, VLF: {vlf:.2f}")

class vitalDSP.physiological_features.frequency_domain.FrequencyDomainFeatures(nn_intervals, fs=4)[source]

Bases: object

A class for computing frequency-domain features from physiological signals (ECG, PPG).

nn_intervals

The NN intervals (in milliseconds) between heartbeats.

Type:: list or np.array

fs

The sampling frequency (Hz). Defaults to 4 Hz, typically used for HRV analysis.

Type:: int

compute_psd()[source]: Computes the power spectral density (PSD) for HRV and returns LF, HF, and LF/HF ratio.

compute_lf()[source]: Computes the Low-Frequency (LF) power from the PSD.

compute_hf()[source]: Computes the High-Frequency (HF) power from the PSD.

compute_lf_hf_ratio()[source]: Computes the ratio of LF to HF power (LF/HF ratio).

compute_ulf()[source]: Computes ULF power (0.0033 to 0.04 Hz).

compute_vlf()[source]: Computes VLF power (0.0033 to 0.04 Hz).

compute_total_power()[source]: Computes the total power of ULF, VLF, LF, and HF bands.

compute_lfnu()[source]: Computes normalized LF power (LFnu).

compute_hfnu()[source]: Computes normalized HF power (HFnu).

compute_hf()[source]

Computes the High-Frequency (HF) power from the PSD.

Returns:: The HF power.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> hf = fdf.compute_hf()
>>> print(f"HF: {hf}")

compute_hfnu()[source]

Computes the normalized High-Frequency (HFnu) power as a percentage of total power (LF + HF).

Returns:: The HFnu value.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> hfnu = fdf.compute_hfnu()
>>> print(f"HFnu: {hfnu}")

compute_lf()[source]

Computes the Low-Frequency (LF) power from the PSD.

Returns:: The LF power.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> lf = fdf.compute_lf()
>>> print(f"LF: {lf}")

compute_lf_hf_ratio()[source]

Computes the ratio of LF to HF power (LF/HF ratio).

Returns:: The LF/HF ratio.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> lf_hf_ratio = fdf.compute_lf_hf_ratio()
>>> print(f"LF/HF Ratio: {lf_hf_ratio}")

compute_lfnu()[source]

Computes the normalized Low-Frequency (LFnu) power as a percentage of total power (LF + HF).

Returns:: The LFnu value.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> lfnu = fdf.compute_lfnu()
>>> print(f"LFnu: {lfnu}")

compute_psd()[source]

Computes the Power Spectral Density (PSD) using Welch’s method to estimate the power in ULF, VLF, LF, and HF bands.

Returns:

Dictionary containing frequency domain HRV metrics:

’ulf_power’: Ultra-Low Frequency power (ms²)
’vlf_power’: Very-Low Frequency power (ms²)
’lf_power’: Low Frequency power (ms²)
’hf_power’: High Frequency power (ms²)
’lf_hf_ratio’: LF/HF ratio
’total_power’: Total spectral power (ms²)
’frequencies’: Frequency array
’psd’: Power spectral density array

Return type:

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> psd_result = fdf.compute_psd()
>>> print(f"LF: {psd_result['lf_power']}, HF: {psd_result['hf_power']}")

compute_total_power()[source]

Computes the total power, which is the sum of power in ULF, VLF, LF, and HF bands.

Returns:: The total power.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> total_power = fdf.compute_total_power()
>>> print(f"Total Power: {total_power}")

compute_ulf()[source]

Computes the Ultra-Low Frequency (ULF) power from the PSD.

Returns:: The ULF power.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> ulf = fdf.compute_ulf()
>>> print(f"ULF: {ulf}")

compute_vlf()[source]

Computes the Very Low Frequency (VLF) power from the PSD.

Returns:: The VLF power.
Return type:: float

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> fdf = FrequencyDomainFeatures(nn_intervals)
>>> vlf = fdf.compute_vlf()
>>> print(f"VLF: {vlf}")

HRV Analysis

Physiological Features Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - Comprehensive signal analysis

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.physiological_features.hrv_analysis import HRVFeatures
>>> ecg_signal = np.random.randn(1000)
>>> hrv = HRVFeatures(ecg_signal, fs=256, signal_type='ECG')
>>> features = hrv.compute_all_features()
>>> print(f"SDNN: {features.get('sdnn', 'N/A')}")

class vitalDSP.physiological_features.hrv_analysis.HRVFeatures(signals, nn_intervals=None, fs=100, signal_type='PPG', options=None)[source]

Bases: object

A class to compute HRV features from physiological signals such as ECG and PPG.

It combines time-domain, frequency-domain, and nonlinear features into one comprehensive feature extraction module.

nn_intervals

The NN intervals (in milliseconds) between heartbeats.

Type:: np.array

signal

The physiological signal (ECG, PPG, etc.).

Type:: np.array

fs

The sampling frequency in Hz. Default is 1000 Hz.

Type:: int

compute_all_features()[source]: Computes all HRV features and returns them in a dictionary format.

Examples

>>> import numpy as np
>>> from vitalDSP.physiological_features.hrv_analysis import HRVFeatures
>>>
>>> # Example 1: Using ECG signal
>>> ecg_signal = np.random.randn(1000)  # Simulated ECG signal
>>> hrv_ecg = HRVFeatures(ecg_signal, fs=256, signal_type="ECG")
>>> hrv_features = hrv_ecg.compute_all_features()
>>> print(f"SDNN: {hrv_features.get('sdnn', 'N/A')}")
>>>
>>> # Example 2: Using PPG signal
>>> ppg_signal = np.random.randn(2000)  # Simulated PPG signal
>>> hrv_ppg = HRVFeatures(ppg_signal, fs=128, signal_type="PPG")
>>> hrv_features = hrv_ppg.compute_all_features()
>>> print(f"RMSSD: {hrv_features.get('rmssd', 'N/A')}")
>>>
>>> # Example 3: Using pre-computed NN intervals
>>> nn_intervals = np.array([800, 850, 820, 900, 780])  # RR intervals in ms
>>> hrv_precomputed = HRVFeatures(None, nn_intervals=nn_intervals, fs=100)
>>> hrv_features = hrv_precomputed.compute_all_features()
>>> print(f"pNN50: {hrv_features.get('pnn50', 'N/A')}")

compute_all_features(include_complex_methods=None, **kwargs)[source]

Computes all nonlinear features of the signal, with an option to skip time-consuming methods.

Parameters:

include_complex_methods (bool, optional) – Whether to compute the time-consuming methods: compute_sample_entropy, compute_approximate_entropy, and compute_recurrence_features. If None (default), automatically enables for signals with ≥50 NN intervals.
**kwargs – Additional parameters for specific feature computations.

Returns:

A dictionary containing all the computed features.

Return type:

Example usage

>>> nn_intervals = [800, 810, 790, 805, 795]  # NN intervals in ms
>>> ecg_signal = np.random.randn(1000)  # Example ECG signal
>>> fs = 1000  # Sampling frequency in Hz
>>> hrv = HRVFeatures(nn_intervals, ecg_signal, fs)
>>> all_features = hrv.compute_all_features()
>>> print(all_features)

Beat-to-Beat Analysis

Physiological Features Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Configurable parameters and settings

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.physiological_features.beat_to_beat import BeatToBeatAnalysis
>>> ecg_signal = np.random.randn(5000)
>>> bba = BeatToBeatAnalysis(ecg_signal, fs=500)
>>> rr = bba.compute_rr_intervals()
>>> sdnn = bba.compute_sdnn()
>>> print(f'SDNN: {sdnn:.2f} ms')

class vitalDSP.physiological_features.beat_to_beat.BeatToBeatAnalysis(signal, r_peaks=None, fs=1000, signal_type='ECG')[source]

Bases: object

A class for advanced beat-to-beat and heart rate variability (HRV) analysis in ECG and PPG signals.

signal

The ECG or PPG signal.

Type:: np.array

r_peaks

The indices of R-peaks (ECG) or systolic peaks (PPG).

Type:: np.array

fs

The sampling frequency of the signal in Hz. Default is 1000 Hz.

Type:: int

signal_type

The type of signal (‘ECG’ or ‘PPG’).

Type:: str

compute_hr()[source]

Computes the heart rate (HR) from the R-R or P-P intervals.

Returns:: The heart rate (in beats per minute) for each R-R interval.
Return type:: np.array

compute_mean_rr()[source]

Computes the mean of the R-R intervals.

Returns:: The mean R-R interval in milliseconds.
Return type:: float

compute_pnn50()[source]

Computes the percentage of successive R-R intervals that differ by more than 50 milliseconds (pNN50), a commonly used HRV metric.

Returns:: The percentage of R-R intervals differing by more than 50 ms (pNN50).
Return type:: float

compute_rmssd()[source]

Computes the Root Mean Square of Successive Differences (RMSSD), a measure of short-term heart rate variability (HRV).

Returns:: The RMSSD value in milliseconds.
Return type:: float

compute_rr_intervals(correction_method=None, threshold=150)[source]

Computes the R-R intervals (for ECG) or P-P intervals (for PPG) from the detected peaks. Optionally applies a correction method to fix false beat detections.

Parameters:

correction_method (str) – Method to correct false detections (‘interpolation’, ‘resampling’, or ‘adaptive_threshold’). Default is None (no correction).
threshold (int) – Threshold for adaptive correction methods in milliseconds. Default is 150 ms.

Returns:

The corrected R-R or P-P intervals in milliseconds.

Return type:

np.array

Example

>>> r_peaks = [100, 200, 310, 420]  # Detected R-peaks or PPG systolic peaks
>>> btb = BeatToBeatAnalysis(ecg_signal, r_peaks)
>>> rr_intervals = btb.compute_rr_intervals(correction_method='interpolation')
>>> print(f"R-R Intervals: {rr_intervals}")

compute_sdnn()[source]

Computes the standard deviation of the R-R intervals (SDNN), a common HRV metric.

Returns:: The standard deviation of R-R intervals (SDNN).
Return type:: float

detect_arrhythmias(threshold=150)[source]

Detects arrhythmias by identifying irregular R-R intervals based on a threshold for variability.

Parameters:: threshold (int) – The threshold for detecting abnormal R-R intervals in milliseconds. Default is 150 ms.
Returns:: Indices of arrhythmic beats.
Return type:: list

Nonlinear Features

Physiological Features Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Performance optimization

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.physiological_features.nonlinear import NonlinearFeatures
>>> signal = np.random.randn(1000)
>>> nf = NonlinearFeatures(signal)
>>> sample_ent = nf.compute_sample_entropy()
>>> dfa_alpha = nf.compute_dfa()
>>> poincare = nf.compute_poincare_features()
>>> print(f'Sample Entropy: {sample_ent}, DFA: {dfa_alpha}')

class vitalDSP.physiological_features.nonlinear.NonlinearFeatures(signal, fs=1000)[source]

Bases: object

A class for computing nonlinear (geometric) features from physiological signals (ECG, PPG, EEG).

signal

The physiological signal (e.g., ECG, PPG, EEG).

Type:: np.array

fs

The sampling frequency of the signal in Hz. Default is 1000 Hz.

Type:: int

compute_sample_entropy(m=2, r=0.2)[source]: Computes the sample entropy of the signal, measuring its complexity.

compute_approximate_entropy(m=2, r=0.2)[source]: Computes the approximate entropy of the signal, quantifying its unpredictability.

compute_fractal_dimension(kmax=10)[source]: Computes the fractal dimension of the signal using Higuchi’s method.

compute_lyapunov_exponent()[source]: Computes the largest Lyapunov exponent, indicating the presence of chaos in the signal.

compute_dfa(order=1)[source]: Computes the detrended fluctuation analysis (DFA) for assessing fractal scaling.

compute_poincare_features()[source]: Computes Poincaré plot features (SD1 and SD2) to assess short- and long-term HRV variability.

compute_recurrence_features(threshold=0.2)[source]: Computes features from the recurrence plot, including recurrence rate, determinism, and laminarity.

compute_approximate_entropy(m=2, r=0.2)[source]

Computes the approximate entropy of the signal. Approximate entropy quantifies the unpredictability and regularity of signal patterns.

Parameters:

m (int) – Embedding dimension (default is 2).
r (float) – Tolerance (default is 0.2).

Returns:

The computed approximate entropy of the signal.

Return type:

Example

>>> ppg_signal = [...]  # Sample PPG signal
>>> nf = NonlinearFeatures(ppg_signal)
>>> approx_entropy = nf.compute_approximate_entropy()
>>> print(f"Approximate Entropy: {approx_entropy}")

compute_dfa(order=1)[source]

Computes the Detrended Fluctuation Analysis (DFA) of the signal. DFA is used to assess the fractal scaling properties of time-series data, especially in physiological signals.

Parameters:: order (int) – The order of the polynomial fit for detrending. Default is 1 (linear detrending).
Returns:: The DFA scaling exponent (α).
Return type:: float

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> nf = NonlinearFeatures(ecg_signal)
>>> dfa = nf.compute_dfa(order=1)
>>> print(f"DFA Scaling Exponent: {dfa}")

compute_fractal_dimension(kmax=10)[source]

Computes the fractal dimension of the signal using Higuchi’s method. Fractal dimension is a measure of complexity, reflecting how the signal fills space as its scale changes.

Returns:: The fractal dimension of the signal.
Return type:: float

Example

>>> eeg_signal = [...]  # Sample EEG signal
>>> nf = NonlinearFeatures(eeg_signal)
>>> fractal_dimension = nf.compute_fractal_dimension()
>>> print(f"Fractal Dimension: {fractal_dimension}")

compute_lyapunov_exponent()[source]

Computes the largest Lyapunov exponent (LLE) of the signal. LLE measures the rate at which nearby trajectories in phase space diverge, indicating chaotic behavior in the signal.

Returns:: The largest Lyapunov exponent of the signal.
Return type:: float

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> nf = NonlinearFeatures(ecg_signal)
>>> lyapunov_exponent = nf.compute_lyapunov_exponent()
>>> print(f"Largest Lyapunov Exponent: {lyapunov_exponent}")

compute_poincare_features()[source]

Computes the SD1 and SD2 features from the Poincaré plot of the NN intervals. SD1 reflects short-term HRV, while SD2 reflects long-term HRV.

Returns:

Dictionary containing Poincaré plot features:

’sd1’: Short-term HRV variability
’sd2’: Long-term HRV variability
’sd_ratio’: SD1/SD2 ratio

Return type:

Example

>>> nn_intervals = [800, 810, 790, 805, 795]
>>> nf = NonlinearFeatures(nn_intervals)
>>> poincare = nf.compute_poincare_features()
>>> print(f"SD1: {poincare['sd1']}, SD2: {poincare['sd2']}")

compute_recurrence_features(threshold=0.2, sample_size=10000)[source]

Computes approximate recurrence features by sampling point pairs. This approach significantly reduces computation time for large datasets by avoiding the full pairwise distance calculations.

Parameters:

threshold (float) – The threshold to define recurrences. Default is 0.2.
sample_size (int) – The number of point pairs to sample. Default is 10,000.

Returns:

A dictionary containing approximate recurrence rate, determinism, and laminarity.

Return type:

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> nf = NonlinearFeatures(ecg_signal)
>>> rqa_features = nf.compute_recurrence_features(threshold=0.2, sample_size=10000)
>>> print(rqa_features)

compute_sample_entropy(m=2, r=0.2)[source]

Computes the sample entropy of the signal. Sample entropy is a measure of signal complexity, specifically used for detecting the regularity and unpredictability of fluctuations in a signal.

Parameters:

m (int) – Embedding dimension (default is 2).
r (float) – Tolerance (default is 0.2).

Returns:

The computed sample entropy of the signal.

Return type:

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> nf = NonlinearFeatures(ecg_signal)
>>> sample_entropy = nf.compute_sample_entropy()
>>> print(f"Sample Entropy: {sample_entropy}")

Waveform Morphology

Physiological Features Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Interactive visualization capabilities

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.physiological_features.waveform import WaveformMorphology
>>> signal = np.random.ra    >>> wm = WaveformMorphology(signal, fs=256)
>>> troughs = wm.detect_troughs(){result}')

class vitalDSP.physiological_features.waveform.WaveformMorphology(waveform, fs=256, qrs_ratio=0.05, signal_type='ECG', peak_config=None, simple_mode=True, options=None)[source]

Bases: object

A class for computing morphological features from physiological waveforms (ECG, PPG, EEG).

waveform

The waveform signal (ECG, PPG, EEG).

Type:: np.array

fs

The sampling frequency of the signal in Hz. Default is 1000 Hz.

Type:: int

signal_type

The type of signal (‘ECG’, ‘PPG’, ‘EEG’).

Type:: str

simple_mode

If True, uses simplified diastolic peak detection (midpoint-based). Default is True.

Type:: bool, optional

Examples

>>> import numpy as np
>>> from vitalDSP.physiological_features.waveform import WaveformMorphologyMorphology
>>>
>>> # Example 1: ECG waveform analysis
>>> ecg_signal = np.random.randn(1000)  # Simulated ECG signal
>>> wm_ecg = WaveformMorphology(ecg_signal, fs=256, signal_type="ECG")
>>> r_peaks = wm_ecg.r_peaks
>>> q_valleys = wm_ecg.detect_q_valley()
>>> print(f"Detected {len(r_peaks)} R-peaks")
>>>
>>> # Example 2: PPG waveform analysis
>>> ppg_signal = np.random.randn(2000)  # Simulated PPG signal
>>> wm_ppg = WaveformMorphology(ppg_signal, fs=128, signal_type="PPG")
>>> systolic_peaks = wm_ppg.detect_systolic_peaks()
>>> diastolic_peaks = wm_ppg.detect_diastolic_peaks()
>>> print(f"Detected {len(systolic_peaks)} systolic peaks")
>>>
>>> # Example 3: Custom peak detection configuration
>>> custom_config = {
...     "distance": 30,
...     "window_size": 5,
...     "threshold_factor": 1.2,
...     "search_window": 4
... }
>>> wm_custom = WaveformMorphology(ecg_signal, fs=256, signal_type="ECG", peak_config=custom_config)
>>> r_peaks_custom = wm_custom.r_peaks
>>> print(f"Custom detection found {len(r_peaks_custom)} R-peaks")

compute_amplitude(interval_type='R-to-S', baseline_method='moving_average', compare_to_baseline=False, signal_type='ECG')[source]

Computes the amplitude (max-min) of the waveform for specified intervals.

Parameters:

interval_type (str, optional) – The interval type to calculate the amplitude for: - For ECG: - “R-to-S”: Between R peak and S valley. - “R-to-Q”: Between Q valley and R peak. - “P-to-Q”: Between P peak and Q valley. - “T-to-S”: Between S valley and T peak. - “T-to-Baseline”: Between T peak and baseline. - “R-to-Baseline”: Between R peak and baseline. - “S-to-Baseline”: Between S valley and baseline. - For PPG: - “Sys-to-Notch”: Between systolic peak and dicrotic notch. - “Notch-to-Dia”: Between dicrotic notch and diastolic peak. - “Sys-to-Dia”: Between systolic and diastolic peaks. - “Sys-to-Baseline”: Between systolic peak and baseline. - “Notch-to-Baseline”: Between dicrotic notch and baseline. - “Dia-to-Baseline”: Between diastolic peak and baseline.
signal_type (str, optional) – The type of signal: “ECG” or “PPG”. Default is “ECG”.
compare_to_baseline (bool, optional) – If True, compute amplitudes relative to a baseline.

Returns:

List[float]

Return type:

A list of amplitude values for each interval within a single complex.

Examples

>>> amplitude_values = wm.compute_amplitude(interval_type="R-to-S", signal_type="ECG")
>>> print(f"Amplitude values for each complex in R-to-S interval: {amplitude_values}")

compute_curvature(points=None, option=None, window=3)[source]

Compute the curvature of the waveform at specified points or critical points.

Parameters:

points (list or np.array, optional) – Indices of the points where the curvature is to be calculated. Ignored if option is specified.
option (str, optional) – Specifies the critical points to use for curvature calculation. Options for ECG are “p_peaks”, “q_valleys”, “r_peaks”, “s_valleys”, “t_peaks”. Options for PPG are “troughs”, “systolic_peaks”, “diastolic_peaks”.
window (int, optional) – The number of points before and after each point to consider for curvature calculation.

Returns:

curvatures – Curvature values at the specified points or critical points.

Return type:

np.array

Example

>>> waveform = np.sin(np.linspace(0, 2 * np.pi, 100))
>>> wm = WaveformMorphology(waveform, signal_type="PPG")
>>> curvatures = wm.compute_curvature(option="systolic_peaks")
>>> print(f"Curvatures at systolic peaks: {curvatures}")

compute_duration(sessions=None, mode='Custom')[source]

Computes the duration of the QRS complex in an ECG waveform.

Returns:: The QRS duration in milliseconds.
Return type:: float

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> wm = WaveformMorphology(ecg_signal, signal_type="ECG")
>>> qrs_duration = wm.compute_qrs_duration()
>>> print(f"QRS Duration: {qrs_duration} ms")

compute_eeg_wavelet_features()[source]

Computes EEG wavelet features by applying a wavelet transform and extracting relevant frequency bands.

Returns:: A dictionary of extracted EEG wavelet features (e.g., delta, theta, alpha).
Return type:: dict

Example

>>> eeg_signal = [...]  # Sample EEG signal
>>> wm = WaveformMorphology(eeg_signal, signal_type="EEG")
>>> wavelet_features = wm.compute_eeg_wavelet_features()
>>> print(wavelet_features)

compute_ppg_dicrotic_notch()[source]

Detects the dicrotic notch in a PPG waveform and computes its timing.

Returns:: The timing of the dicrotic notch in milliseconds.
Return type:: float

Example

>>> ppg_signal = [...]  # Sample PPG signal
>>> wm = WaveformMorphology(ppg_signal, signal_type="PPG")
>>> notch_timing = wm.compute_ppg_dicrotic_notch()
>>> print(f"Dicrotic Notch Timing: {notch_timing} ms")

compute_skewness(signal_type='ECG')[source]

Compute the skewness of each complex in the signal.

Parameters:: signal_type (str, optional) – The type of signal, either “ECG” or “PPG”. Default is “ECG”.
Returns:: List[float]
Return type:: A list of skewness values, one for each complex.

Examples

>>> signal = np.random.randn(1000)
>>> extractor = PhysiologicalFeatureExtractor(signal)
>>> skewness_values = extractor.compute_skewness()
>>> print(f"Skewness values for each complex: {skewness_values}")

compute_slope(points=None, option=None, window=3, slope_unit='radians')[source]

Compute the slope of the waveform at specified points or critical points.

Parameters:

points (list or np.array, optional) – Indices of the points where the slope is to be calculated. Ignored if option is specified.
option (str, optional) – Specifies the critical points to use for slope calculation. Options for ECG are “p_peaks”, “q_valleys”, “r_peaks”, “s_valleys”, “t_peaks”. Options for PPG are “troughs”, “systolic_peaks”, “diastolic_peaks”.
window (int, optional) – The number of points before and after each point to consider for slope calculation.

Returns:

slopes – Slope values at the specified points or critical points.

Return type:

np.array

Example

>>> waveform = np.sin(np.linspace(0, 2 * np.pi, 100))
>>> wm = WaveformMorphology(waveform, signal_type="ECG")
>>> slopes = wm.compute_slope(option="r_peaks")
>>> print(f"Slopes at R peaks: {slopes}")

compute_volume(interval_type='P-to-T', signal_type='ECG', mode='peak')[source]

Compute the area under the curve between two sets of peaks and valleys for specified intervals.

Parameters:

interval_type (str, optional) –
The interval type to calculate the volume for: - For ECG: - “P-to-T”: Entire complex from P peak to T peak. - “R-to-S”: Between R peak and S valley. - “R-to-Q”: Between Q valley and R peak. - “P-to-Q”: Between P peak and Q valley. - “T-to-S”: Between S valley and T peak. - For PPG: - “Sys-to-Notch”: Between systolic peak and dicrotic notch. - “Notch-to-Dia”: Between dicrotic notch and diastolic peak. - “Sys-to-Dia”: Between systolic and diastolic peaks. - “Sys-to-Sys”: Between consecutive systolic peaks (full PPG complex).

Default is “P-to-T” for ECG and “Sys-to-Notch” for PPG.
signal_type (str, optional) – The type of signal: “ECG” or “PPG”. Default is “ECG”.
mode (str, optional) – The area computation method (“peak” or “trough”). Default is “peak”. - “peak”: Computes the area under the curve. - “trough”: Computes the area bounded by troughs.

Returns:

List[float] – within a single complex.

Return type:

A list of volume values, each representing the area for the specified interval

Examples

>>> volume_values = wm.compute_volume(interval_type="R-to-S", signal_type="ECG")
>>> print(f"Volume values for each complex in R-to-S interval: {volume_values}")

detect_diastolic_peak(notches=None, diastolic_troughs=None)[source]

Detect diastolic peaks in PPG signals based on notches and diastolic troughs.

Parameters:

notches (list of int) – Indices of detected notches in the PPG signal.
diastolic_troughs (list of int) – Indices of diastolic troughs in the PPG signal.

Returns:

diastolic_peaks – Indices of diastolic peaks detected in the PPG signal.

Return type:

list of int

detect_dicrotic_notches(systolic_peaks=None, diastolic_troughs=None)[source]

Detects the dicrotic notches in a PPG waveform using second derivative.

Parameters:

systolic_peaks (np.array) – Indices of detected systolic peaks in the PPG waveform.
diastolic_troughs (np.array) – Indices of the detected troughs between the systolic peaks.

Returns:

notches – Indices of detected dicrotic notches in the PPG waveform.

Return type:

np.array

detect_ecg_session(p_peaks=None, t_peaks=None)[source]

Detects the ECG session (start and end) based on flat lines before the P peak and after the T peak.

Parameters:

p_peaks (np.array, optional) – Indices of detected P peaks in the ECG waveform.
t_peaks (np.array, optional) – Indices of detected T peaks in the ECG waveform.

Returns:

ecg_sessions – Each tuple contains the start and end index of an ECG session.

Return type:

list of tuples

detect_p_peak(r_peaks=None, q_valleys=None)[source]

Detects the P peak (local maximum) in the ECG waveform just before each Q valley.

Parameters:: q_valleys (list of int) – Indices of detected Q valleys in the ECG waveform.
Returns:: p_peaks – Indices of the P peaks (local maximum) for each Q valley.
Return type:: list of int

detect_ppg_session(troughs=None)[source]

Detects PPG sessions between consecutive troughs.

Parameters:: troughs (np.array, optional) – Indices of detected troughs in the PPG waveform.
Returns:: Array of tuples containing start and end indices of PPG sessions.
Return type:: np.ndarray

detect_q_session(p_peaks=None, q_valleys=None, r_peaks=None)[source]

Detects the Q sessions (start and end) in the ECG waveform based on R peaks.

Parameters:: r_peaks (np.array) – Indices of detected R peaks in the ECG waveform.
Returns:: q_sessions – Each tuple contains the start and end index of a Q session.
Return type:: list of tuples

Example

>>> waveform = np.sin(np.linspace(0, 2*np.pi, 100))  # Simulated ECG signal
>>> wm = WaveformMorphology(waveform, signal_type="ECG")
>>> r_peaks = PeakDetection(waveform).detect_peaks()
>>> q_sessions = wm.detect_q_session(r_peaks)
>>> print(f"Q Sessions: {q_sessions}")

detect_q_valley(r_peaks=None)[source]

Detects the Q valley (local minimum) in the ECG waveform just before each R peak.

Parameters:: r_peaks (np.array) – Indices of detected R peaks in the ECG waveform.
Returns:: q_valleys – Indices of the Q valley (local minimum) for each R peak.
Return type:: list of int

detect_qrs_session(rpeaks=None, q_session=None, s_session=None)[source]

Detects the QRS complex sessions (start and end) in the ECG waveform.

Parameters:: r_peaks (np.array) – Indices of detected R peaks in the ECG waveform.
Returns:: qrs_sessions – Each tuple contains the start and end index of a QRS session.
Return type:: np.ndarray

Example

>>> waveform = np.sin(np.linspace(0, 2*np.pi, 100))  # Simulated ECG signal
>>> wm = WaveformMorphology(waveform, signal_type="ECG")
>>> r_peaks = PeakDetection(waveform).detect_peaks()
>>> qrs_sessions = wm.detect_qrs_session(r_peaks)
>>> print(f"QRS Sessions: {qrs_sessions}")

detect_r_session(rpeaks=None, q_sessions=None, s_sessions=None)[source]

Detects the R session (start and end) in the ECG waveform based on Q and S sessions.

Parameters:

q_session (tuple) – The start and end index of the Q session.
s_session (tuple) – The start and end index of the S session.

Returns:

r_sessions – Each tuple contains the start and end index of an R session.

Return type:

list of tuples

detect_s_session(t_peaks=None, s_valleys=None, r_peaks=None)[source]

Detects the S sessions (start and end) in the ECG waveform based on R peaks.

Parameters:: r_peaks (np.array) – Indices of detected R peaks in the ECG waveform.
Returns:: s_sessions – Each tuple contains the start and end index of an S session.
Return type:: list of tuples

Example

>>> waveform = np.sin(np.linspace(0, 2*np.pi, 100))  # Simulated ECG signal
>>> wm = WaveformMorphology(waveform, signal_type="ECG")
>>> r_peaks = PeakDetection(waveform).detect_peaks()
>>> s_sessions = wm.detect_s_session(r_peaks)
>>> print(f"S Sessions: {s_sessions}")

detect_s_valley(r_peaks=None)[source]

Detects the S valleys (local minima after each R peak).

Parameters:: r_peaks (list of int) – Indices of detected R peaks in the ECG waveform.
Returns:: s_valleys – Indices of the S valleys for each R peak.
Return type:: list of int

detect_t_peak(r_peaks=None, s_valleys=None)[source]

Detect T peaks (local maxima) between the S valley and the midpoint to the next R peak.

Parameters:

r_peaks (list of int) – Indices of detected R peaks in the ECG waveform.
s_valleys (list of int) – Indices of detected S valleys in the ECG waveform.

Returns:

t_peaks – Indices of the T peaks for each S valley.

Return type:

list of int

detect_troughs(systolic_peaks=None)[source]

Detects the troughs (valleys) in the PPG waveform between systolic peaks. In simple_mode, uses the minimum value between adjacent peaks; otherwise, uses flat segment detection.

Parameters:: systolic_peaks (np.array) – Indices of detected systolic peaks in the PPG waveform.
Returns:: troughs – Indices of the detected troughs between the systolic peaks.
Return type:: np.array

Example

>>> waveform = np.sin(np.linspace(0, 2*np.pi, 100))  # Simulated PPG signal
>>> wm = WaveformMorphology(waveform, signal_type="PPG")
>>> peaks = PeakDetection(waveform).detect_peaks()
>>> troughs = wm.detect_troughs(peaks)
>>> print(f"Troughs: {troughs}")

get_amplitude_variability(interval_type='Sys-to-Baseline', baseline_method='moving_average', signal_type='PPG', method='std_dev')[source]

Calculates the variability in amplitude over the specified interval or baseline comparison.

Parameters:

interval_type (str) – The interval type for amplitude calculation, with default “Sys-to-Baseline”.
signal_type (str) – The type of signal: “PPG” or “ECG”.
method (str, optional) – The method to calculate variability: “std” for standard deviation or “cv” for coefficient of variation.

Returns:

float

Return type:

Variability of the amplitudes.

Examples

>>> variability = wm.get_amplitude_variability(interval_type="Sys-to-Baseline", signal_type="PPG")
>>> print(f"Amplitude variability (Sys-to-Baseline): {variability}")

get_area(interval_type, signal_type='PPG', summary_type='mean')[source]

Computes the area of the specified interval in the signal with the chosen summary type.

Parameters:

interval_type (str) – Type of interval to compute the area for. Options include: - For PPG: “Sys-to-Notch”, “Notch-to-Dia”, “Sys-to-Dia” - For ECG: “R-to-Q”, “R-to-S”, “QRS”, “T-to-S”
signal_type (str, optional) – The type of signal: “PPG” or “ECG”. Default is “PPG”.
summary_type (str, optional) – Specifies the summary statistic to return: - ‘mean’: Mean of all areas - ‘median’: Median of all areas - ‘2nd_quartile’: 25th percentile (1st quartile) - ‘3rd_quartile’: 75th percentile (3rd quartile) - ‘full’: Returns all areas as a list Default is ‘mean’.

Returns:

The summary statistic for the area, or the list of all areas if summary_type is ‘full’. Returns NaN if an error occurs.

Return type:

float or list

Notes

Logs an error and returns NaN if the area computation fails or if an invalid summary type is provided.

get_duration(start_points=None, end_points=None, session_type='systolic', summary_type='mean')[source]

Computes the duration based on detected start and end points for a specified session type.

Parameters:

start_points (list, optional) – List of detected start indices in the signal. If None, defaults based on session_type.
end_points (list, optional) – List of detected end indices in the signal. If None, defaults based on session_type.
session_type (str, optional) – Specifies the type of session: “systolic”, “diastolic”, “qrs”, or “Custom”. - For “systolic” and “diastolic”, appropriate start and end points will be detected if not provided. - For “qrs”, QRS sessions are automatically detected and do not require start or end points. - For “Custom”, start and end points must be provided as arguments.
summary_type (str, optional) – Type of summary to apply to the computed durations. Options are ‘mean’, ‘median’, ‘2nd_quartile’, ‘3rd_quartile’, or ‘full’ (returns all durations).

Returns:

The summarized duration based on summary_type, or a list of durations if summary_type is ‘full’.

Return type:

float or list

Notes

Logs an error and returns NaN if no valid sessions are detected or if an invalid summary type is provided.

get_heart_rate(summary_type='mean')[source]

Computes the heart rate based on R-R intervals.

Parameters:: summary_type (str, optional) – Specifies the summary statistic to return: - ‘mean’: Mean heart rate - ‘median’: Median heart rate - ‘2nd_quartile’: 25th percentile (1st quartile) - ‘3rd_quartile’: 75th percentile (3rd quartile) - ‘full’: Returns all computed heart rates as a list Default is ‘mean’.
Returns:: The summary statistic for heart rates, or a list of all heart rates if summary_type is ‘full’. Returns NaN if an error occurs.
Return type:: float or list

Notes

Logs an error and returns NaN if an invalid summary type is provided or if heart rate computation fails.

get_peak_trend_slope(peaks=None, method='linear_regression', window_size=5)[source]

Calculate the trend slope of peak values using specified method.

Parameters:

peaks (list or np.ndarray) – The y-values of detected peaks (peak amplitudes).
method (str, optional) – The method to calculate trend slope. Options are ‘linear_regression’, ‘moving_average’, and ‘rate_of_change’. Default is ‘linear_regression’.
window_size (int, optional) – The window size for moving average calculation. Used only when method=’moving_average’. Default is 5.

Returns:

The calculated trend slope. If ‘moving_average’ method is selected, returns an array of slopes.

Return type:

float or np.ndarray

get_qrs_amplitude(summary_type='mean')[source]

Calculates the QRS amplitude by finding the maximum amplitude between the R-to-S and R-to-Q intervals.

Parameters:: summary_type (str, optional) – Specifies the summary statistic to return: - ‘mean’: Mean of all QRS amplitudes - ‘median’: Median of all QRS amplitudes - ‘2nd_quartile’: 25th percentile (1st quartile) - ‘3rd_quartile’: 75th percentile (3rd quartile) - ‘full’: Returns all QRS amplitudes as a list Default is ‘mean’.
Returns:: The summary statistic for QRS amplitude, or the list of amplitudes if summary_type is ‘full’. Returns NaN if an error occurs.
Return type:: float or list

Notes

Logs an error and returns NaN if an invalid summary type is provided or if amplitude computation fails.

get_signal_skewness(signal_type='PPG', summary_type='mean')[source]

Computes the skewness of the signal based on the specified signal type and summary type.

Parameters:

signal_type (str, optional) – Type of signal to compute skewness for (e.g., ‘PPG’ or ‘ECG’). Default is ‘PPG’.
summary_type (str, optional) – Type of summary to apply to the computed skewness values. Options are ‘mean’, ‘median’, ‘2nd_quartile’, ‘3rd_quartile’, or ‘full’ (returns all skewness values).

Returns:

The summarized skewness value, or a list of skewness values if summary_type is ‘full’.

Return type:

float or list

get_slope(slope_type='systolic', window=5, summary_type='mean', slope_unit='radians')[source]

Computes the slope of the specified type (systolic, diastolic, QRS) using the chosen summary type.

Parameters:

slope_type (str, optional) – The type of slope to calculate. Options are: - “systolic”: Slope from systolic peaks. - “diastolic”: Slope from diastolic peaks. - “qrs”: Slope from QRS R peaks. Default is “systolic”.
window (int, optional) – The window size for slope calculation. Default is 5.
summary_type (str, optional) – Specifies the summary statistic to return: - ‘mean’: Mean of all slopes - ‘median’: Median of all slopes - ‘2nd_quartile’: 25th percentile (1st quartile) - ‘3rd_quartile’: 75th percentile (3rd quartile) - ‘full’: Returns all slopes as a list Default is ‘mean’.

Returns:

The summary statistic for the slopes, or the list of slopes if summary_type is ‘full’. Returns NaN if an error occurs.

Return type:

float or list

Respiratory Analysis Module

Respiratory Analysis

Respiratory Analysis Module for Physiological Signal Processing

This module provides comprehensive respiratory analysis capabilities for physiological signals including PPG, ECG, and other vital signs. It implements multiple methods for respiratory rate estimation including time-domain counting, frequency-domain analysis, and advanced signal processing techniques.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple respiratory rate estimation methods - Time-domain peak counting and interval analysis - Frequency-domain FFT-based analysis - Advanced preprocessing and filtering options - Noise reduction and artifact handling - Comprehensive respiratory pattern analysis

Examples:

Basic respiratory rate estimation:

>>> import numpy as np
>>> from vitalDSP.respiratory_analysis.respiratory_analysis import RespiratoryAnalysis
>>> ppg_signal = np.random.randn(2000)  # Simulated PPG signal
>>> resp_analysis = RespiratoryAnalysis(ppg_signal, fs=128)
>>> rr_result = resp_analysis.compute_respiratory_rate(method="counting")
>>> print(f"Respiratory rate: {rr_result['respiratory_rate']:.2f} breaths/min")

FFT-based analysis:

>>> rr_fft = resp_analysis.compute_respiratory_rate(method="fft_based")
>>> print(f"FFT-based RR: {rr_fft['respiratory_rate']:.2f} breaths/min")

With preprocessing:

>>> from vitalDSP.preprocess.preprocess_operations import PreprocessConfig
>>> config = PreprocessConfig(filter_type="bandpass", lowcut=0.1, highcut=2.0)
>>> rr_preprocessed = resp_analysis.compute_respiratory_rate(method="counting", preprocess_config=config)
>>> print(f"Preprocessed RR: {rr_preprocessed['respiratory_rate']:.2f} breaths/min")

class vitalDSP.respiratory_analysis.respiratory_analysis.RespiratoryAnalysis(signal, fs=256)[source]

Bases: object

A class for analyzing respiratory patterns in physiological signals (e.g., PPG, ECG), with built-in preprocessing, filtering, and noise reduction options.

signal

The raw PPG or ECG signal to analyze.

Type:: numpy.ndarray

fs

The sampling frequency of the signal in Hz.

Type:: int

Examples

>>> import numpy as np
>>> from vitalDSP.respiratory_analysis.respiratory_analysis import RespiratoryAnalysis
>>>
>>> # Example 1: Basic respiratory rate estimation
>>> ppg_signal = np.random.randn(2000)  # Simulated PPG signal
>>> resp_analysis = RespiratoryAnalysis(ppg_signal, fs=128)
>>> rr_result = resp_analysis.compute_respiratory_rate(method="counting")
>>> print(f"Respiratory rate: {rr_result['respiratory_rate']:.2f} breaths/min")
>>>
>>> # Example 2: FFT-based respiratory rate estimation
>>> rr_fft = resp_analysis.compute_respiratory_rate(method="fft_based")
>>> print(f"FFT-based RR: {rr_fft['respiratory_rate']:.2f} breaths/min")
>>>
>>> # Example 3: Respiratory rate with preprocessing
>>> from vitalDSP.preprocess.preprocess_operations import PreprocessConfig
>>> config = PreprocessConfig(
...     filter_type="bandpass",
...     lowcut=0.1,
...     highcut=2.0,
...     noise_reduction_method="wavelet"
... )
>>> rr_preprocessed = resp_analysis.compute_respiratory_rate(
...     method="peaks",
...     preprocess_config=config
... )
>>> print(f"Preprocessed RR: {rr_preprocessed['respiratory_rate']:.2f} breaths/min")

compute_respiratory_rate(method='counting', correction_method=None, min_breath_duration=0.5, max_breath_duration=6, preprocess_config=None)[source]

Computes the respiratory rate from the signal after applying preprocessing, using peak detection or zero-crossing methods.

Parameters:

method (str, optional) – Method used for breath detection. Options: ‘peaks’, ‘zero_crossing’, ‘time_domain’, ‘frequency_domain’, ‘fft_based’, ‘counting’. Default is ‘counting’.
correction_method (str, optional) – Method for correcting false detections. Options: ‘interpolation’, ‘adaptive_threshold’. Default is None.
min_breath_duration (float, optional) – Minimum breath duration in seconds. Default is 0.5s (30 breaths/min).
max_breath_duration (float, optional) – Maximum breath duration in seconds. Default is 6s (10 breaths/min).
preprocess_config (PreprocessConfig, optional) – Configuration for signal preprocessing (filtering and noise reduction). Default is None, and a default PreprocessConfig will be used.

Returns:

The estimated respiratory rate in breaths per minute.

Return type:

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.2, 100)
>>> ra = RespiratoryAnalysis(signal, fs=1000)
>>> preprocess_config = PreprocessConfig(filter_type='bandpass', noise_reduction_method='wavelet')
>>> respiratory_rate = ra.compute_respiratory_rate(method='peaks', preprocess_config=preprocess_config)
>>> print(f"Respiratory Rate: {respiratory_rate} breaths per minute")

compute_respiratory_rate_ensemble(preprocess_config=None, methods=None)[source]

Compute respiratory rate using multiple methods and return consensus estimate.

This method runs multiple RR estimation algorithms in parallel and combines their results using median consensus, providing a more robust estimate than any single method. It also returns quality metrics including standard deviation and confidence scores.

Parameters:

preprocess_config (PreprocessConfig, optional) – Configuration for signal preprocessing. If None, uses default configuration.
methods (list of str, optional) – List of methods to use for ensemble estimation. Default: [‘counting’, ‘fft_based’, ‘frequency_domain’, ‘time_domain’] Available methods: - ‘counting’: Peak detection with interval analysis (peak_detection_rr) - ‘fft_based’: Fast Fourier Transform based estimation - ‘frequency_domain’: Welch PSD based estimation - ‘time_domain’: Autocorrelation based estimation - ‘peaks’: Peak detection (old method, uses _detect_breaths_by_peaks) - ‘zero_crossing’: Zero-crossing detection

Returns:

Dictionary containing: - ‘respiratory_rate’: float, consensus RR estimate (median of valid methods) - ‘individual_estimates’: dict, per-method estimates (method_name: rr_value) - ‘std’: float, standard deviation across methods (measure of agreement) - ‘confidence’: float, confidence score (0-1 scale, higher = more agreement) - ‘n_methods’: int, number of methods that returned valid estimates - ‘quality’: str, quality rating (‘high’, ‘medium’, ‘low’)

Return type:

Examples

>>> import numpy as np
>>> from vitalDSP.respiratory_analysis.respiratory_analysis import RespiratoryAnalysis
>>> signal = np.sin(2 * np.pi * 0.25 * np.arange(0, 60, 1/128))  # 15 BPM signal
>>> resp = RespiratoryAnalysis(signal, fs=128)
>>> result = resp.compute_respiratory_rate_ensemble()
>>> print(f"RR: {result['respiratory_rate']:.1f} BPM")
>>> print(f"Confidence: {result['confidence']:.2f}")
>>> print(f"Individual estimates: {result['individual_estimates']}")

>>> # Use specific methods only
>>> result = resp.compute_respiratory_rate_ensemble(methods=['fft_based', 'frequency_domain'])

Notes

The ensemble approach provides several advantages: 1. More robust to method-specific failures 2. Automatic outlier rejection (methods with implausible results) 3. Quality/confidence metrics for result interpretation 4. Better performance on noisy or challenging signals

Confidence scoring: - confidence > 0.8: High agreement (std < 1 BPM) - Very reliable - confidence 0.5-0.8: Medium agreement (std 1-3 BPM) - Reliable - confidence < 0.5: Low agreement (std > 3 BPM) - Use with caution

If fewer than 2 methods return valid estimates, confidence is automatically set to 0.3 (low) as consensus requires multiple opinions.

References

FFT-Based RR Estimation

Respiratory Analysis Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple processing methods and functions - NumPy integration for numerical computations

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.estimate_rr.fft_based_rr import FftBasedRr
>>> signal = np.random.randn(1000)
>>> processor = FftBasedRr(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

vitalDSP.respiratory_analysis.estimate_rr.fft_based_rr.fft_based_rr(signal, sampling_rate, preprocess=None, freq_min=0.1, freq_max=0.5, **preprocess_kwargs)[source]

Estimate respiratory rate using the FFT (Fast Fourier Transform) method with respiratory band filtering.

This method computes the FFT power spectrum and identifies the dominant frequency within the physiological respiratory range (0.1-0.5 Hz or 6-30 BPM). The respiratory band filtering prevents false detection of cardiac frequencies or high-frequency noise artifacts.

Parameters:

signal (numpy.ndarray) – The input respiratory signal.
sampling_rate (float) – The sampling rate of the signal in Hz.
preprocess (str, optional) – The preprocessing method to apply before estimation (e.g., “bandpass”, “wavelet”).
freq_min (float, optional (default=0.1)) – Minimum respiratory frequency in Hz (6 BPM). Can be adjusted for different populations (e.g., 0.05 Hz for very slow breathing).
freq_max (float, optional (default=0.5)) – Maximum respiratory frequency in Hz (30 BPM). Can be increased to 0.67 Hz (40 BPM) for tachypnea or exercise conditions.
preprocess_kwargs (dict, optional) – Additional arguments for the preprocessing function.

Returns:

rr – Estimated respiratory rate in breaths per minute.

Return type:

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> rr = fft_based_rr(signal, sampling_rate=100, preprocess='bandpass', lowcut=0.1, highcut=0.5)
>>> print(rr)

>>> # For exercise/tachypnea (up to 40 BPM)
>>> rr = fft_based_rr(signal, sampling_rate=100, freq_max=0.67)

Notes

This implementation fixes the critical bug in the original version that searched the entire frequency spectrum, often picking cardiac frequencies (1-2 Hz / 60-120 BPM) instead of respiratory frequencies. The corrected version restricts the search to the physiological respiratory band.

The method includes SNR validation to warn if the detected peak has low prominence, which may indicate poor signal quality or unreliable estimation.

References

[charlton2018]

Charlton, P.H., et al. (2018). Breathing rate estimation from the electrocardiogram and photoplethysmogram: A review. IEEE Reviews in Biomedical Engineering, 11, 2-20.

Peak Detection RR Estimation

Respiratory Analysis Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Configurable parameters and settings - Pattern and anomaly detection

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.estimate_rr.peak_detection_rr import PeakDetectionRr
>>> signal = np.random.randn(1000)
>>> processor = PeakDetectionRr(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

vitalDSP.respiratory_analysis.estimate_rr.peak_detection_rr.peak_detection_rr(signal, sampling_rate, preprocess=None, min_peak_distance=1.5, height=None, threshold=None, prominence=None, width=None, **preprocess_kwargs)[source]

Estimate respiratory rate using peak detection with interval-based analysis.

This method detects breathing peaks in the respiratory signal and calculates the rate based on the median interval between peaks, which is more robust than simple peak counting. Peaks are validated against physiological constraints (1.5-10 seconds between breaths).

Parameters:

signal (numpy.ndarray) – The input respiratory signal.
sampling_rate (float) – The sampling rate of the signal in Hz.
preprocess (str, optional) – The preprocessing method to apply before estimation (e.g., “bandpass”, “wavelet”).
min_peak_distance (float, optional (default=1.5)) – Minimum distance between peaks in seconds. Default 1.5s corresponds to 40 BPM max. Previous default of 0.5s was too permissive (allowed 120 BPM).
height (float or None, optional) – Minimum height required for a peak. If None, uses adaptive threshold.
threshold (float or None, optional) – Minimum difference between a peak and its neighboring points.
prominence (float or None, optional) – Minimum prominence of peaks. If None, uses adaptive prominence (0.3 * signal std).
width (int or None, optional) – Minimum width required for a peak, measured as the number of samples.
preprocess_kwargs (dict, optional) – Additional arguments for the preprocessing function.

Returns:

rr – Estimated respiratory rate in breaths per minute.

Return type:

Examples

>>> import numpy as np
>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> rr = peak_detection_rr(signal, sampling_rate=100, preprocess='bandpass', lowcut=0.1, highcut=0.5)
>>> print(rr)

>>> # For faster breathing (exercise), reduce min_peak_distance
>>> rr = peak_detection_rr(signal, sampling_rate=100, min_peak_distance=1.0)

Notes

This implementation fixes critical bugs in the original version: 1. Changed from simple peak counting (num_peaks / duration) to interval-based analysis 2. Increased min_peak_distance default from 0.5s to 1.5s (40 BPM max instead of 120 BPM) 3. Added adaptive prominence threshold if not specified 4. Validates intervals against physiological range (1.5-10 seconds) 5. Uses median interval for robustness against outliers 6. Adds quality metrics (coefficient of variation) for breathing regularity

The interval-based approach is more accurate because: - Handles irregular breathing patterns - Robust to missed or false peaks - Provides quality metrics for result confidence

References

[pan1985]

Pan, J., & Tompkins, W. J. (1985). A real-time QRS detection algorithm. IEEE transactions on biomedical engineering, (3), 230-236.

Sleep Apnea Detection

Respiratory Analysis Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple processing methods and functions - NumPy integration for numerical computations

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.sleep_apnea_detection.amplitude_threshold import AmplitudeThreshold
>>> signal = np.random.randn(1000)
>>> processor = AmplitudeThreshold(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

vitalDSP.respiratory_analysis.sleep_apnea_detection.amplitude_threshold.detect_apnea_amplitude(signal, sampling_rate, threshold, min_duration=10, preprocess=None, **preprocess_kwargs)[source]

Detect sleep apnea events based on amplitude thresholding.

Parameters:

signal (numpy.ndarray) – The input respiratory signal.
sampling_rate (float) – The sampling rate of the signal in Hz.
threshold (float) – Amplitude threshold below which an apnea event is detected.
min_duration (int, optional (default=10)) – Minimum duration (in seconds) for which the signal must be below the threshold to be considered an apnea event.
preprocess (str, optional) – The preprocessing method to apply before detection (e.g., “bandpass”, “wavelet”).
preprocess_kwargs (dict, optional) – Additional arguments for the preprocessing function.

Returns:

apnea_events – List of apnea events, each represented as a tuple (start_time, end_time) in seconds.

Return type:

list of tuple

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 60, 0.01)) * 0.2
>>> apnea_events = detect_apnea_amplitude(signal, sampling_rate=100, threshold=0.1)
>>> print(apnea_events)

Transforms Module

Fourier Transform

Fourier Transform Module for Physiological Signal Processing

This module provides comprehensive Fourier Transform capabilities for analyzing the frequency content of physiological signals such as ECG, EEG, and PPG. It implements both Discrete Fourier Transform (DFT) and Inverse Discrete Fourier Transform (IDFT) with signal validation and error handling.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Discrete Fourier Transform (DFT) computation - Inverse Discrete Fourier Transform (IDFT) reconstruction - Signal validation and error handling - Frequency domain analysis capabilities - Signal reconstruction from frequency components

Examples:

Basic frequency analysis:

>>> import numpy as np
>>> from vitalDSP.transforms.fourier_transform import FourierTransform
>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> ft = FourierTransform(signal)
>>> frequency_spectrum = ft.compute_dft()
>>> print(f"Spectrum shape: {frequency_spectrum.shape}")

Signal reconstruction:

>>> reconstructed_signal = ft.compute_idft(frequency_spectrum)
>>> print(f"Reconstruction error: {np.mean((signal - reconstructed_signal)**2):.6f}")

Frequency domain filtering:

>>> # Zero out high frequencies
>>> filtered_spectrum = frequency_spectrum.copy()
>>> filtered_spectrum[20:] = 0
>>> filtered_signal = ft.compute_idft(filtered_spectrum)

class vitalDSP.transforms.fourier_transform.FourierTransform(signal)[source]

Bases: object

A class to perform Fourier Transform for analyzing the frequency content in signals such as ECG/EEG.

The Fourier Transform is a mathematical technique that transforms a time-domain signal into its constituent frequencies, providing insights into the signal’s frequency content. This class allows for both the computation of the Discrete Fourier Transform (DFT) and the Inverse Discrete Fourier Transform (IDFT), making it possible to analyze and reconstruct signals.

compute_dft : method: Computes the Discrete Fourier Transform (DFT) of the signal.

compute_idft : method: Computes the Inverse Discrete Fourier Transform (IDFT) to reconstruct the signal.

compute_dft()[source]

Compute the Discrete Fourier Transform (DFT) of the input signal.

The DFT converts the time-domain signal into the frequency domain, allowing for the analysis of its frequency components. This is particularly useful in identifying periodicities, filtering, and spectral analysis of biomedical signals like ECG and EEG.

Returns:: The frequency domain representation of the signal, where each element corresponds to a specific frequency component.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> ft = FourierTransform(signal)
>>> frequency_content = ft.compute_dft()
>>> print(frequency_content)

compute_idft(frequency_content)[source]

Compute the Inverse Discrete Fourier Transform (IDFT) to reconstruct the time-domain signal.

The IDFT converts the frequency-domain data back into the time domain, reconstructing the original signal from its frequency components. This is useful for understanding how different frequency components contribute to the overall signal and for signal reconstruction after processing in the frequency domain.

Parameters:: frequency_content (numpy.ndarray) – The frequency domain representation of the signal, as obtained from the DFT.
Returns:: The time-domain signal reconstructed from its frequency components.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> ft = FourierTransform(signal)
>>> frequency_content = ft.compute_dft()
>>> reconstructed_signal = ft.compute_idft(frequency_content)
>>> print(reconstructed_signal)

filter_frequencies(low_cutoff=None, high_cutoff=None, fs=1.0)[source]

Apply a bandpass filter in the frequency domain.

Parameters:

low_cutoff (float, optional) – The low cutoff frequency. Frequencies below this will be zeroed out.
high_cutoff (float, optional) – The high cutoff frequency. Frequencies above this will be zeroed out.
fs (float, optional) – The sampling frequency of the signal. Default is 1.0.

Returns:

The filtered signal in the time domain.

Return type:

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> ft = FourierTransform(signal)
>>> filtered_signal = ft.filter_frequencies(low_cutoff=0.5, high_cutoff=2.0, fs=10)
>>> print(filtered_signal)

Wavelet Transform

Wavelet Transform Module for Physiological Signal Processing

This module provides comprehensive wavelet transform capabilities for physiological signals including ECG, PPG, EEG, and other vital signs. It implements Discrete Wavelet Transform (DWT) with multiple mother wavelets and inverse transform capabilities for signal analysis and reconstruction.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Discrete Wavelet Transform (DWT) implementation - Multiple mother wavelets (Haar, Daubechies, Coiflets, etc.) - Inverse Wavelet Transform for signal reconstruction - Multi-level decomposition capabilities - Signal length preservation options - Integration with mother wavelet utilities

Examples:

Basic wavelet transform:

>>> import numpy as np
>>> from vitalDSP.transforms.wavelet_transform import WaveletTransform
>>> signal = np.sin(np.linspace(0, 10, 1000)) + np.random.normal(0, 0.1, 1000)
>>> wt = WaveletTransform(signal, wavelet_name="haar")
>>> coefficients = wt.perform_wavelet_transform()
>>> print(f"Coefficients shape: {len(coefficients)}")

Signal reconstruction:

>>> reconstructed = wt.perform_inverse_wavelet_transform(coefficients)
>>> print(f"Reconstruction error: {np.mean((signal - reconstructed)**2):.6f}")

Different wavelets:

>>> wt_db4 = WaveletTransform(signal, wavelet_name="db4")
>>> wt_coif2 = WaveletTransform(signal, wavelet_name="coif2")
>>> db4_coeffs = wt_db4.perform_wavelet_transform()
>>> coif2_coeffs = wt_coif2.perform_wavelet_transform()

class vitalDSP.transforms.wavelet_transform.WaveletTransform(signal, wavelet_name='haar', same_length=True)[source]

Bases: object

A class to perform Discrete Wavelet Transform (DWT) on signals using different mother wavelets.

perform_wavelet_transform : method: Computes the DWT of the signal.

perform_inverse_wavelet_transform : method: Reconstructs the signal using the inverse DWT.

perform_inverse_wavelet_transform(coeffs)[source]

Perform the Inverse Discrete Wavelet Transform (IDWT) to reconstruct the signal.

Parameters:: coeffs (list) – Wavelet coefficients from the wavelet transform.
Returns:: Reconstructed signal from the wavelet coefficients.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> wavelet_transform = WaveletTransform(signal, wavelet_name='db')
>>> coeffs = wavelet_transform.perform_wavelet_transform(level=3)
>>> reconstructed_signal = wavelet_transform.perform_inverse_wavelet_transform(coeffs)
>>> print(reconstructed_signal)

perform_wavelet_transform(level=1)[source]

Perform the Discrete Wavelet Transform (DWT) on the signal.

Parameters:: level (int, optional) – The number of decomposition levels (default is 1).
Returns:: Wavelet coefficients as a list of arrays. coeffs[0] through coeffs[-2] are detail coefficient arrays (one per level, in order from finest to coarsest). coeffs[-1] is the final approximation array.
Return type:: list

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> wavelet_transform = WaveletTransform(signal, wavelet_name='db')
>>> coeffs = wavelet_transform.perform_wavelet_transform(level=3)
>>> print(coeffs)

Discrete Cosine Transform

Signal Transforms Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Signal transformation methods

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.transforms.discrete_cosine_transform import DiscreteCosineTransform
>>> signal = np.random.randn(1000)
>>> processor = DiscreteCosineTransform(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.transforms.discrete_cosine_transform.DiscreteCosineTransform(signal)[source]

Bases: object

A class to perform Discrete Cosine Transform (DCT) and its inverse (IDCT).

The DCT is widely used in signal processing, particularly for data compression (such as JPEG image compression) and feature extraction. It transforms a signal from the time domain into the frequency domain, where most of the signal information is compacted into a few coefficients. The inverse DCT (IDCT) reconstructs the signal from its DCT coefficients.

compute_dct : method: Computes the DCT of the signal.

compute_idct : method: Computes the Inverse DCT to reconstruct the signal.

compress_signal(threshold=0.1)[source]

Compress the signal by zeroing out small DCT coefficients.

This method removes high-frequency components that are below the threshold, which often represent noise, improving the signal’s reconstruction quality and reducing its size.

Parameters:: threshold (float, optional) – The threshold below which DCT coefficients are set to zero. Default is 0.1.
Returns:: The compressed DCT coefficients.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> dct = DiscreteCosineTransform(signal)
>>> compressed_coefficients = dct.compress_signal(threshold=0.05)
>>> print(compressed_coefficients)

compute_dct(norm='ortho')[source]

Compute the Discrete Cosine Transform (DCT) of the signal.

The DCT converts the input signal from the time domain to the frequency domain, emphasizing the low-frequency components of the signal. It is particularly effective for compressing signals because it tends to concentrate most of the signal’s energy into a few low-frequency components.

Parameters:: norm (str, optional) – Normalization type. Default is ‘ortho’ for orthogonal DCT, which is usually recommended.
Returns:: The DCT coefficients of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> dct = DiscreteCosineTransform(signal)
>>> dct_coefficients = dct.compute_dct()
>>> print(dct_coefficients)

compute_idct(dct_coefficients, norm='ortho')[source]

Compute the Inverse Discrete Cosine Transform (IDCT) to reconstruct the signal.

The IDCT converts the DCT coefficients back into the time domain, reconstructing the original signal. This process is essential in applications like image compression, where the signal is compressed using DCT and then decompressed back to its original form using IDCT.

Parameters:

dct_coefficients (numpy.ndarray) – The DCT coefficients of the signal.
norm (str, optional) – Normalization type. Default is ‘ortho’ for orthogonal DCT.

Returns:

The time-domain signal reconstructed from its DCT coefficients.

Return type:

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> dct = DiscreteCosineTransform(signal)
>>> dct_coefficients = dct.compute_dct()
>>> reconstructed_signal = dct.compute_idct(dct_coefficients)
>>> print(reconstructed_signal)

Hilbert Transform

Signal Transforms Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - Signal transformation methods

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.transforms.hilbert_transform import HilbertTransform
>>> signal = np.random.randn(1000)
>>> processor = HilbertTransform(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.transforms.hilbert_transform.HilbertTransform(signal)[source]

Bases: object

A class to perform the Hilbert Transform, which is used to generate analytic signals.

The Hilbert Transform is a key tool in signal processing, particularly for generating the analytic signal from a real-valued signal. The analytic signal is complex, with the original signal as the real part and the Hilbert transform as the imaginary part. This is particularly useful in applications like QRS detection in ECG signals, where phase and amplitude information are crucial.

compute_hilbert : method: Computes the Hilbert Transform of the signal to obtain the analytic signal.

compute_hilbert()[source]

Compute the Hilbert Transform to obtain the analytic signal.

The Hilbert Transform is applied in the frequency domain by first taking the Fourier transform of the input signal, modifying the Fourier coefficients to zero out the negative frequencies, and then applying the inverse Fourier transform. This process effectively shifts the signal in such a way that the imaginary part represents the phase information, while the real part remains the original signal.

Returns:: The analytic signal with both real and imaginary components, where the real part is the original signal and the imaginary part is the Hilbert transform.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> ht = HilbertTransform(signal)
>>> analytic_signal = ht.compute_hilbert()
>>> print(analytic_signal)

Notes

The analytic signal is often used in applications where the instantaneous amplitude and phase of the signal are required, such as in the detection of QRS complexes in ECG signals, modulation, and demodulation in communications, and envelope detection in various signal processing tasks.

envelope()[source]

Compute the envelope of the signal using the Hilbert Transform.

The envelope is the magnitude of the analytic signal and represents the instantaneous amplitude of the signal. This is particularly useful in applications such as PPG signal analysis, where the envelope can be used to assess pulse amplitude variations.

Returns:: The envelope of the input signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> ht = HilbertTransform(signal)
>>> envelope = ht.envelope()
>>> print(envelope)

instantaneous_phase()[source]

Compute the instantaneous phase of the signal using the Hilbert Transform.

The instantaneous phase is the phase angle of the analytic signal and is useful in applications such as ECG analysis, where phase information can help in detecting the QRS complex or other waveform characteristics.

Returns:: The instantaneous phase of the input signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> ht = HilbertTransform(signal)
>>> phase = ht.instantaneous_phase()
>>> print(phase)

Advanced Computation Module

Anomaly Detection

Anomaly Detection Module for Physiological Signal Processing

This module provides comprehensive anomaly detection capabilities for physiological signals including ECG, PPG, EEG, and other vital signs. It implements multiple detection methods including statistical approaches, machine learning techniques, and frequency-domain analysis for identifying unusual patterns and artifacts.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Statistical anomaly detection (Z-score, IQR) - Moving average and rolling window methods - Local Outlier Factor (LOF) for density-based detection - Fourier-based frequency domain analysis - Real-time streaming anomaly detection - Performance monitoring and optimization

Examples:

Basic Z-score anomaly detection:

>>> import numpy as np
>>> from vitalDSP.advanced_computation.anomaly_detection import AnomalyDetection
>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> anomaly_detector = AnomalyDetection(signal)
>>> anomalies = anomaly_detector.detect_anomalies(method="z_score", threshold=2.0)
>>> print(f"Detected {len(anomalies)} anomalies")

Moving average detection:

>>> anomalies_ma = anomaly_detector.detect_anomalies(method="moving_average", window_size=5, threshold=0.5)
>>> print(f"Moving average anomalies: {len(anomalies_ma)}")

LOF-based detection:

>>> anomalies_lof = anomaly_detector.detect_anomalies(method="lof", n_neighbors=20)
>>> print(f"LOF anomalies: {len(anomalies_lof)}")

FFT-based detection:

>>> anomalies_fft = anomaly_detector.detect_anomalies(method="fft", threshold=0.1)
>>> print(f"FFT anomalies: {len(anomalies_fft)}")

class vitalDSP.advanced_computation.anomaly_detection.AnomalyDetection(signal)[source]

Bases: object

Comprehensive Anomaly Detection for detecting anomalies in real-time from streaming data.

This class offers multiple methods to detect anomalies in a given signal, including statistical methods, moving averages, Local Outlier Factor (LOF), and Fourier-based methods.

detect_anomalies : function: Detects anomalies using various methods including z-score, moving average, custom LOF, and more.

Examples

>>> import numpy as np
>>> from vitalDSP.advanced_computation.anomaly_detection import AnomalyDetection
>>>
>>> # Example 1: Z-score anomaly detection
>>> signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)
>>> anomaly_detector = AnomalyDetection(signal)
>>> anomalies_z_score = anomaly_detector.detect_anomalies(method="z_score", threshold=2.0)
>>> print(f"Z-score anomalies: {len(anomalies_z_score)}")
>>>
>>> # Example 2: Moving average anomaly detection
>>> anomalies_moving_avg = anomaly_detector.detect_anomalies(method="moving_average", window_size=5, threshold=0.5)
>>> print(f"Moving average anomalies: {len(anomalies_moving_avg)}")
>>>
>>> # Example 3: LOF anomaly detection
>>> anomalies_lof = anomaly_detector.detect_anomalies(method="lof", n_neighbors=20)
>>> print(f"LOF anomalies: {len(anomalies_lof)}")
>>>
>>> # Example 4: FFT-based anomaly detection
>>> anomalies_fft = anomaly_detector.detect_anomalies(method="fft", threshold=0.1)
>>> print(f"FFT anomalies: {len(anomalies_fft)}")

detect_anomalies(method='z_score', **kwargs)[source]

Detect anomalies in the signal using the specified method.

Parameters:

method (str, optional) – The method to use for detecting anomalies. Options: ‘z_score’, ‘moving_average’, ‘lof’, ‘fft’, ‘threshold’. Default is ‘z_score’.
**kwargs (additional arguments) – Additional parameters depending on the chosen method.

Returns:

Indices of the detected anomalies.

Return type:

Raises:

ValueError – If the specified method is unknown.

Examples

>>> anomalies_z_score = anomaly_detector.detect_anomalies(method="z_score", threshold=2.0)
>>> anomalies_moving_avg = anomaly_detector.detect_anomalies(method="moving_average", window_size=5, threshold=0.5)

Bayesian Analysis

Advanced Computation Module for Physiological Signal Processing

This module provides Bayesian inference and optimization capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.1

Key Features: - Gaussian Process modeling for signal prediction - Bayesian Optimization for hyperparameter tuning - Expected Improvement acquisition function - Numerical stability improvements - Graceful error handling

Classes:

GaussianProcess: Implements Gaussian Process regression for probabilistic modeling.
BayesianOptimization: Implements Bayesian Optimization using Gaussian Processes.

Examples:

Example 1: Gaussian Process Prediction

>>> import numpy as np
>>> from vitalDSP.advanced_computation.bayesian_analysis import GaussianProcess
>>>
>>> # Create GP model
>>> gp = GaussianProcess(length_scale=1.0, noise=1e-5)
>>>
>>> # Training data
>>> X_train = np.array([[0.1], [0.4], [0.7]])
>>> y_train = np.sin(3 * X_train.flatten())
>>> gp.update(X_train, y_train)
>>>
>>> # Predict at new points
>>> X_new = np.array([[0.2], [0.5]])
>>> mean, variance = gp.predict(X_new)
>>> print(f"Predicted mean: {mean}")
>>> print(f"Predicted variance: {variance}")

Example 2: Bayesian Optimization for Hyperparameter Tuning

>>> from vitalDSP.advanced_computation.bayesian_analysis import BayesianOptimization
>>>
>>> # Define objective function to maximize
>>> def filter_performance(cutoff_freq):
...     # Simulate filter performance metric
...     cutoff = np.atleast_1d(cutoff_freq)[0]
...     return -(cutoff - 0.5)**2 + 1.0  # Peak at 0.5
>>>
>>> # Optimize cutoff frequency in range [0, 1]
>>> optimizer = BayesianOptimization(
...     func=filter_performance,
...     bounds=(0, 1),
...     length_scale=0.1,
...     noise=1e-5
... )
>>>
>>> # Run optimization
>>> best_cutoff, best_performance = optimizer.optimize(n_iter=20, random_seed=42)
>>> print(f"Best cutoff frequency: {best_cutoff[0]:.4f}")
>>> print(f"Best performance: {best_performance:.4f}")

Example 3: ECG Filter Parameter Optimization

>>> from vitalDSP.filtering.signal_filtering import SignalFiltering
>>>
>>> # Simulated ECG signal
>>> ecg_signal = np.sin(2*np.pi*1.2*np.linspace(0, 10, 1000)) + 0.1*np.random.randn(1000)
>>>
>>> # Define objective: minimize noise while preserving signal
>>> def filter_quality(params):
...     cutoff = np.atleast_1d(params)[0]
...     sf = SignalFiltering(ecg_signal)
...     filtered = sf.butterworth_lowpass(cutoff_freq=cutoff, fs=100, order=4)
...     # Quality metric: SNR-like measure
...     signal_power = np.var(filtered)
...     noise_power = np.var(ecg_signal - filtered)
...     return signal_power / (noise_power + 1e-10)
>>>
>>> # Optimize filter cutoff
>>> optimizer = BayesianOptimization(filter_quality, bounds=(0.1, 10))
>>> best_cutoff, best_snr = optimizer.optimize(n_iter=15, random_seed=42)
>>> print(f"Optimal cutoff: {best_cutoff[0]:.2f} Hz, SNR: {best_snr:.2f}")

Notes:

Numerical Stability: The default noise parameter (1e-5) provides good numerical stability. Avoid using values smaller than 1e-7 as they can cause singular matrix errors.
Initialization: BayesianOptimization starts with 3 random samples to bootstrap the Gaussian Process before beginning optimization.
Fallback: If numerical issues occur during optimization, the algorithm falls back to random sampling to ensure it always returns a result.

class vitalDSP.advanced_computation.bayesian_analysis.BayesianOptimization(func, bounds, length_scale=1.0, noise=1e-05)[source]

Bases: object

A class implementing Bayesian Optimization for parameter tuning.

Bayesian Optimization is an efficient method for finding the minimum or maximum of an objective function, especially when the function is expensive to evaluate. It uses a Gaussian Process model to predict the function’s behavior and an acquisition function to determine the next point to sample.

optimize(n_iter=10, random_seed=None)[source]: Perform Bayesian optimization to find the best parameters.

acquisition(X, xi=0.01)[source]: Compute the Expected Improvement (EI) at new points.

propose_location(n_restarts=10)[source]: Propose the next sampling location based on the acquisition function.

Example Usage

-------------

>>> def objective_function(x):

>>> return -np.sin(3 * x) - x ** 2 + 0.7 * x

>>>

>>> bayesian_optimizer = BayesianOptimization(objective_function, bounds=(0, 2))

>>> best_x, best_y = bayesian_optimizer.optimize(n_iter=10)

>>> print("Best X:", best_x)

>>> print("Best Y:", best_y)

acquisition(X, xi=0.01)[source]

Expected Improvement (EI) acquisition function.

The EI acquisition function is used to balance exploration and exploitation in Bayesian Optimization. It selects points that have the potential to improve upon the best observed value.

Parameters:

X (numpy.ndarray) – The points at which to evaluate the acquisition function.
xi (float, optional) – Exploration-exploitation trade-off parameter (default is 0.01).

Returns:

The EI values at the provided points.

Return type:

optimize(n_iter=10, random_seed=None)[source]

Perform Bayesian optimization to find the best parameters.

Parameters:

n_iter (int, optional) – Number of iterations for the optimization (default is 10).
random_seed (int or None, optional) – Random seed for reproducibility (default is None).

Returns:

The best parameters and the corresponding function value.

Return type:

Notes

The optimization process starts with a random sample to initialize the Gaussian Process, then iteratively selects new points using the acquisition function to balance exploration and exploitation.

Examples

>>> def objective(x):
...     return -np.sin(3 * x) - x ** 2 + 0.7 * x
>>> optimizer = BayesianOptimization(objective, bounds=(0, 2))
>>> best_x, best_y = optimizer.optimize(n_iter=10, random_seed=42)
>>> print(f"Best X: {best_x}, Best Y: {best_y}")

propose_location(n_restarts=10)[source]

Propose the next sampling location by optimizing the acquisition function.

Parameters:: n_restarts (int, optional) – Number of restarts for the optimization to avoid local optima (default is 10).
Returns:: The proposed next sampling point.
Return type:: numpy.ndarray

class vitalDSP.advanced_computation.bayesian_analysis.GaussianProcess(length_scale=1.0, noise=1e-10)[source]

Bases: object

A class implementing Gaussian Process (GP) for Bayesian Optimization.

Gaussian Process models are used to predict the mean and variance of an unknown function based on observed data. This is particularly useful in Bayesian Optimization, where the GP helps in selecting the next sample point to evaluate.

predict(X)[source]: Predict the mean and variance of the objective function at new points X.

update(X_train, y_train)[source]: Update the GP model with new observations.

Example Usage

-------------

>>> gp = GaussianProcess(length_scale=1.0, noise=1e-10)

>>> X_train = np.array([[0.1], [0.4], [0.7]])

>>> y_train = np.sin(3 * X_train) - X_train ** 2 + 0.7 * X_train

>>> gp.update(X_train, y_train)

>>> X_new = np.array([[0.2], [0.5]])

>>> mean, variance = gp.predict(X_new)

>>> print("Predicted Mean:", mean)

>>> print("Predicted Variance:", variance)

predict(X)[source]

Predict the mean and variance of the objective function at new points.

Parameters:: X (numpy.ndarray) – The points at which to predict the mean and variance.
Returns:: The predicted mean and variance at the input points X.
Return type:: tuple
Raises:: ValueError – If the GP model has not been updated with any training data.

update(X_train, y_train)[source]

Update the GP model with new observations.

Parameters:

X_train (numpy.ndarray) – The observed input points.
y_train (numpy.ndarray) – The observed output values corresponding to X_train.

Neural Network Filtering

Advanced Computation Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - Configurable parameters and settings - Advanced filtering algorithms

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.advanced_computation.neural_network_filtering import NeuralNetworkFiltering
>>> signal = np.random.randn(1000)
>>> nn = NeuralNetworkFiltering(input_size=100)
>>> filtered = nn.apply_filter(signal)

class vitalDSP.advanced_computation.neural_network_filtering.ConvolutionalNetwork(dropout_rate, batch_norm)[source]

Bases: object

predict(X)[source]

train(X, y, learning_rate, epochs, batch_size)[source]

class vitalDSP.advanced_computation.neural_network_filtering.FeedforwardNetwork(hidden_layers, dropout_rate, batch_norm)[source]

Bases: object

predict(X)[source]

train(X, y, learning_rate, epochs, batch_size)[source]

class vitalDSP.advanced_computation.neural_network_filtering.NeuralNetworkFiltering(signal, network_type='feedforward', hidden_layers=[64, 64], learning_rate=0.001, epochs=100, batch_size=32, dropout_rate=0.5, batch_norm=True, recurrent_type='lstm')[source]

Bases: object

A comprehensive neural network-based filtering approach for adaptive signal processing.

This class supports various neural network architectures, including feedforward, convolutional, and recurrent networks. It includes advanced features such as dropout for regularization, batch normalization for faster convergence, and customizable training options.

train : method: Trains the neural network on the given signal.

apply_filter : method: Applies the trained neural network to filter the signal.

evaluate : method: Evaluates the performance of the neural network on a test signal.

Example Usage

-------------

signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)

nn_filter = NeuralNetworkFiltering(signal, network_type='feedforward', hidden_layers=[64, 64], epochs=100)

# Train the neural network filter

nn_filter.train()

# Apply the trained filter

filtered_signal = nn_filter.apply_filter()

print("Filtered Signal:", filtered_signal)

# Evaluate the filter on a test signal

test_signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)

mse = nn_filter.evaluate(test_signal)

print("Mean Squared Error on Test Signal:", mse)

apply_filter()[source]

Apply the trained neural network to filter the signal.

The trained neural network is used to predict and filter the input signal.

Returns:: The filtered signal after applying the neural network.
Return type:: numpy.ndarray

evaluate(test_signal)[source]

Evaluate the performance of the neural network on a test signal.

The method calculates the Mean Squared Error (MSE) between the predicted and actual values of the test signal.

Parameters:: test_signal (numpy.ndarray) – The test signal to evaluate the neural network’s performance.
Returns:: The mean squared error (MSE) of the neural network on the test signal.
Return type:: float

train()[source]

Train the neural network on the given signal.

The signal is divided into input-output pairs for supervised training. This method prepares the data, then trains the network using the specified learning rate, number of epochs, and batch size.

class vitalDSP.advanced_computation.neural_network_filtering.RecurrentNetwork(recurrent_type='lstm', dropout_rate=0.5, batch_norm=True)[source]

Bases: object

predict(X)[source]

train(X, y, learning_rate, epochs, batch_size)[source]

Reinforcement Learning Filter

Advanced Computation Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - Performance optimization - Advanced filtering algorithms

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.advanced_computation.reinforcement_learning_filter import ReinforcementLearningFilter
>>> signal = np.random.randn(1000)
>>> rlf = ReinforcementLearningFilter(signal)
>>> filtered = rlf.apply_filter(signal)

class vitalDSP.advanced_computation.reinforcement_learning_filter.ReinforcementLearningFilter(signal, action_space, state_space=None)[source]

Bases: object

A comprehensive Reinforcement Learning Filter for adaptive signal processing. This class provides methods to train filters using reinforcement learning algorithms such as Q-learning, Deep Q-Networks (DQN), and Proximal Policy Optimization (PPO), and to apply these trained filters to signals.

train_q_learning(episodes=1000, alpha=0.1, gamma=0.99, epsilon=0.1)[source]: Trains a filter using Q-learning.

train_dqn(episodes=1000, batch_size=32, gamma=0.99, epsilon=0.1, target_update=10)[source]: Trains a filter using Deep Q-Networks (DQN).

train_ppo(epochs=1000, gamma=0.99, lambda_=0.95, clip_ratio=0.2)[source]: Trains a filter using Proximal Policy Optimization (PPO).

apply_filter()[source]: Applies the trained filter to the signal.

Example Usage

-------------

signal = np.sin(np.linspace(0, 10, 100)) + np.random.normal(0, 0.1, 100)

rl_filter = ReinforcementLearningFilter(signal, action_space=[-1, 0, 1])

# Train Q-learning based filter

rl_filter.train_q_learning(episodes=500, alpha=0.1, gamma=0.99, epsilon=0.1)

# Apply the trained filter

filtered_signal = rl_filter.apply_filter()

print("Filtered Signal (Q-Learning):", filtered_signal)

# Train DQN-based filter

rl_filter.train_dqn(episodes=500, batch_size=32, gamma=0.99, epsilon=0.1, target_update=10)

# Apply the trained filter

filtered_signal_dqn = rl_filter.apply_filter()

print("Filtered Signal (DQN):", filtered_signal_dqn)

# Train PPO-based filter

rl_filter.train_ppo(epochs=500, gamma=0.99, lambda_=0.95, clip_ratio=0.2)

# Apply the trained filter

filtered_signal_ppo = rl_filter.apply_filter()

print("Filtered Signal (PPO):", filtered_signal_ppo)

apply_filter()[source]

Apply the trained filter to the signal using the learned policy or Q-values.

Returns:: The filtered signal.
Return type:: numpy.ndarray

train_dqn(episodes=1000, batch_size=32, gamma=0.99, epsilon=0.1, target_update=10)[source]

Train the filter using Deep Q-Networks (DQN).

DQN approximates the Q-function using a neural network and uses experience replay to stabilize learning.

Parameters:

episodes (int, optional) – The number of training episodes (default is 1000).
batch_size (int, optional) – The batch size for experience replay (default is 32).
gamma (float, optional) – The discount factor (default is 0.99).
epsilon (float, optional) – The exploration rate for epsilon-greedy policy (default is 0.1).
target_update (int, optional) – The frequency of updating the target network (default is 10 episodes).

train_ppo(epochs=1000, gamma=0.99, lambda_=0.95, clip_ratio=0.2)[source]

Train the filter using Proximal Policy Optimization (PPO).

PPO is a policy gradient method that seeks to optimize policies while ensuring the updates don’t deviate too much from the previous policy.

Parameters:

epochs (int, optional) – The number of training epochs (default is 1000).
gamma (float, optional) – The discount factor (default is 0.99).
lambda (float, optional) – The GAE-lambda parameter (default is 0.95).
clip_ratio (float, optional) – The clipping parameter for PPO (default is 0.2).

train_q_learning(episodes=1000, alpha=0.1, gamma=0.99, epsilon=0.1)[source]

Train the filter using Q-learning.

Q-learning updates a Q-table where each state-action pair has a value. The algorithm iterates over episodes and updates the Q-values based on the reward received for each action taken.

Parameters:

episodes (int, optional) – The number of training episodes (default is 1000).
alpha (float, optional) – The learning rate (default is 0.1).
gamma (float, optional) – The discount factor (default is 0.99).
epsilon (float, optional) – The exploration rate for epsilon-greedy policy (default is 0.1).

class vitalDSP.advanced_computation.reinforcement_learning_filter.SimpleNeuralNetwork(input_size, output_size)[source]

Bases: object

get_weights()[source]

predict(state)[source]

set_weights(weights)[source]

train(state, action, target)[source]

class vitalDSP.advanced_computation.reinforcement_learning_filter.SimplePolicyNetwork(input_size, output_size)[source]

Bases: object

predict(state)[source]

sample_action(state)[source]

train(state, action, advantage)[source]

update(state, action, objective)[source]

class vitalDSP.advanced_computation.reinforcement_learning_filter.SimpleValueNetwork(input_size)[source]

Bases: object

predict(state)[source]

Predict the value of a given state.

Parameters:: state (numpy.ndarray) – The input state vector.
Returns:: The predicted value of the state.
Return type:: float

train(state, reward)[source]

Train the value network by updating weights based on the reward.

Parameters:

state (numpy.ndarray) – The input state vector.
reward (float) – The reward value used for training.

EMD (Empirical Mode Decomposition)

Empirical Mode Decomposition (EMD) Module for Physiological Signal Processing

This module provides comprehensive Empirical Mode Decomposition capabilities for physiological signals including ECG, PPG, EEG, and other vital signs. EMD is particularly effective for analyzing non-linear and non-stationary signals by decomposing them into Intrinsic Mode Functions (IMFs).

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Empirical Mode Decomposition (EMD) implementation - Intrinsic Mode Functions (IMFs) extraction - Non-linear and non-stationary signal analysis - Customizable stop criteria and IMF limits - Signal decomposition and reconstruction - Advanced signal analysis capabilities

Examples:

Basic EMD decomposition:

>>> import numpy as np
>>> from vitalDSP.advanced_computation.emd import EMD
>>> signal = np.sin(np.linspace(0, 10, 100)) + 0.5 * np.random.normal(size=100)
>>> emd = EMD(signal)
>>> imfs = emd.emd()
>>> print(f"Number of IMFs: {len(imfs)}")

Limited IMF decomposition:

>>> emd_limited = EMD(signal)
>>> imfs_limited = emd_limited.emd(max_imfs=3)
>>> print(f"Limited IMFs: {len(imfs_limited)}")

Custom stop criterion:

>>> emd_custom = EMD(signal)
>>> imfs_custom = emd_custom.emd(stop_criterion=0.01)
>>> print(f"Custom stop criterion IMFs: {len(imfs_custom)}")

Signal reconstruction:

>>> reconstructed = np.sum(imfs, axis=0)
>>> print(f"Reconstruction error: {np.mean((signal - reconstructed)**2):.6f}")

class vitalDSP.advanced_computation.emd.EMD(signal)[source]

Bases: object

Empirical Mode Decomposition (EMD) for decomposing non-linear and non-stationary signals into Intrinsic Mode Functions (IMFs).

EMD is particularly useful for analyzing signals that are non-linear and non-stationary, where traditional methods like Fourier Transform may not be effective.

emd : method: Performs the EMD on the input signal and returns the IMFs.

Examples

>>> import numpy as np
>>> from vitalDSP.advanced_computation.emd import EMD
>>>
>>> # Example 1: Basic EMD decomposition
>>> signal = np.sin(np.linspace(0, 10, 100)) + 0.5 * np.random.normal(size=100)
>>> emd = EMD(signal)
>>> imfs = emd.emd()
>>> print(f"Number of IMFs: {len(imfs)}")
>>>
>>> # Example 2: EMD with limited number of IMFs
>>> emd_limited = EMD(signal)
>>> imfs_limited = emd_limited.emd(max_imfs=3)
>>> print(f"Limited IMFs: {len(imfs_limited)}")
>>>
>>> # Example 3: EMD with custom stop criterion
>>> emd_custom = EMD(signal)
>>> imfs_custom = emd_custom.emd(stop_criterion=0.01)
>>> print(f"Custom stop criterion IMFs: {len(imfs_custom)}")

emd(max_imfs=None, stop_criterion=0.05, max_sifting_iterations=20, max_decomposition_iterations=10)[source]

Perform Empirical Mode Decomposition (EMD) on the input signal.

Parameters:

max_imfs (int or None, optional) – Maximum number of IMFs to extract. If None, extract all possible IMFs (default is None).
stop_criterion (float, optional) – The stopping criterion for the sifting process, which controls how close the IMF is to an ideal IMF (default is 0.05).
max_sifting_iterations (int, optional) – Maximum number of sifting iterations per IMF to prevent infinite loops (default is 20).
max_decomposition_iterations (int, optional) – Maximum number of decomposition iterations to prevent excessive computation (default is 10).

Returns:

imfs – A list of IMFs (Intrinsic Mode Functions) extracted from the signal.

Return type:

list of numpy.ndarray

Notes

Each IMF represents a simple oscillatory mode embedded in the signal. The sum of all IMFs plus the final residual will reconstruct the original signal.

OPTIMIZATION: Added convergence limits to prevent infinite loops and improve reliability.

Examples

>>> signal = np.sin(np.linspace(0, 10, 100)) + 0.5 * np.random.normal(size=100)
>>> emd = EMD(signal)
>>> imfs = emd.emd()
>>> print("IMFs:", imfs)

Machine Learning Module

Deep Learning Models

Deep Learning Models Module for Physiological Signal Processing

This module provides state-of-the-art deep learning architectures for physiological signal analysis including ECG, PPG, EEG, and other vital signs. It implements comprehensive neural network models with support for both TensorFlow and PyTorch frameworks, enabling advanced signal classification, anomaly detection, and sequence modeling.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - 1D CNN for signal classification and feature extraction - LSTM networks for sequence modeling and temporal dependencies - Transformer architectures for long-range dependencies - Autoencoder models for anomaly detection and denoising - Pre-trained models and transfer learning capabilities - Model training utilities and callbacks - Cross-framework compatibility (TensorFlow/PyTorch)

Examples:

Basic CNN model for ECG classification:

>>> import numpy as np
>>> from vitalDSP.ml_models.deep_models import CNN1D
>>> model = CNN1D(input_shape=(1000, 1), n_classes=5)
>>> model.build_model()
>>> # Train with your data: model.train(X_train, y_train, epochs=10)

LSTM for sequence modeling:

>>> from vitalDSP.ml_models.deep_models import LSTMModel
>>> lstm_model = LSTMModel(input_shape=(100, 1), n_classes=3)
>>> lstm_model.build_model()
>>> # lstm_model.train(X_train, y_train, epochs=10)

class vitalDSP.ml_models.deep_models.BaseDeepModel(input_shape: Tuple[int, ...], n_classes: int, backend: str = 'tensorflow', **kwargs)[source]

Bases: ABC

Base class for deep learning models.

Provides common interface for all deep models with support for both TensorFlow/Keras and PyTorch backends.

abstract build_model()[source]: Build the model architecture.

load(filepath: str)[source]: Load model from disk.

abstract predict(X)[source]: Make predictions.

save(filepath: str)[source]: Save model to disk.

abstract train(X_train, y_train, X_val=None, y_val=None, **kwargs)[source]: Train the model.

class vitalDSP.ml_models.deep_models.CNN1D(input_shape: Tuple[int, ...], n_classes: int, n_filters: List[int] | None = None, kernel_sizes: List[int] | None = None, pool_sizes: List[int] | None = None, dropout_rate: float = 0.5, use_residual: bool = False, backend: str = 'tensorflow', **kwargs)[source]

Bases: BaseDeepModel

1D Convolutional Neural Network for signal classification.

Designed for physiological signal classification tasks such as: - ECG arrhythmia detection - PPG quality assessment - EEG sleep stage classification - Respiratory event detection

Architecture: - Multiple 1D convolutional layers with batch normalization - MaxPooling for downsampling - Dropout for regularization - Dense layers for classification - Residual connections (optional)

Parameters:

input_shape (tuple) – Shape of input signals (sequence_length, n_channels)
n_classes (int) – Number of output classes
n_filters (list of int, default=[32, 64, 128]) – Number of filters in each conv layer
kernel_sizes (list of int, default=[7, 5, 3]) – Kernel sizes for each conv layer
pool_sizes (list of int, default=[2, 2, 2]) – Pool sizes for each pooling layer
dropout_rate (float, default=0.5) – Dropout rate for regularization
use_residual (bool, default=False) – Whether to use residual connections
backend (str, default='tensorflow') – Deep learning backend

model

The underlying model

Type:: keras.Model or torch.nn.Module

history

Training history

Type:: dict

Examples

>>> from vitalDSP.ml_models import CNN1D
>>> model = CNN1D(input_shape=(1000, 1), n_classes=5)
>>> model.build_model()
>>> model.train(X_train, y_train, epochs=50, batch_size=32)
>>> predictions = model.predict(X_test)

build_model()[source]: Build 1D CNN architecture.

predict(X: ndarray) → ndarray[source]

Make predictions.

Parameters:: X (ndarray of shape (n_samples, sequence_length, n_channels)) – Input data
Returns:: predictions – Predicted class probabilities or labels
Return type:: ndarray

train(X_train: ndarray, y_train: ndarray, X_val: ndarray | None = None, y_val: ndarray | None = None, epochs: int = 50, batch_size: int = 32, learning_rate: float = 0.001, **kwargs)[source]

Train the CNN model.

Parameters:

X_train (ndarray of shape (n_samples, sequence_length, n_channels)) – Training data
y_train (ndarray of shape (n_samples,) or (n_samples, n_classes)) – Training labels
X_val (ndarray, optional) – Validation data
y_val (ndarray, optional) – Validation labels
epochs (int, default=50) – Number of training epochs
batch_size (int, default=32) – Batch size for training
learning_rate (float, default=0.001) – Learning rate
**kwargs (dict) – Additional training parameters

Returns:

history – Training history

Return type:

class vitalDSP.ml_models.deep_models.LSTMModel(input_shape: Tuple[int, ...], n_classes: int, lstm_units: List[int] | None = None, dropout_rate: float = 0.3, bidirectional: bool = True, task: str = 'classification', backend: str = 'tensorflow', **kwargs)[source]

Bases: BaseDeepModel

LSTM (Long Short-Term Memory) for sequence modeling.

Ideal for: - Time series forecasting - Sequential pattern recognition - Long-term dependency modeling - Real-time signal prediction

Parameters:

input_shape (tuple) – Shape of input sequences
n_classes (int) – Number of output classes (use 1 for regression)
lstm_units (list of int, default=[128, 64]) – Number of units in each LSTM layer
dropout_rate (float, default=0.3) – Dropout rate
bidirectional (bool, default=True) – Whether to use bidirectional LSTM
task (str, default='classification') – Task type (‘classification’ or ‘regression’)

Examples

>>> model = LSTMModel(input_shape=(100, 12), n_classes=4)
>>> model.build_model()
>>> model.train(X_train, y_train, epochs=100)

build_model()[source]: Build LSTM architecture.

predict(X: ndarray) → ndarray[source]: Make predictions.

train(X_train: ndarray, y_train: ndarray, X_val: ndarray | None = None, y_val: ndarray | None = None, epochs: int = 100, batch_size: int = 32, learning_rate: float = 0.001, **kwargs)[source]: Train the LSTM model.

Autoencoder Models

Autoencoder Models Module for Physiological Signal Processing

This module provides comprehensive autoencoder architectures for physiological signal analysis including ECG, PPG, EEG, and other vital signs. It implements various autoencoder types for unsupervised anomaly detection, signal denoising, dimensionality reduction, and feature learning.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Standard Autoencoder for basic reconstruction - Variational Autoencoder (VAE) for probabilistic modeling - Denoising Autoencoder for noise reduction - Convolutional Autoencoder for spatial feature learning - LSTM Autoencoder for temporal sequence modeling - Comprehensive training and evaluation utilities - Model saving and loading capabilities

Examples:

Basic autoencoder for anomaly detection:

>>> import numpy as np
>>> from vitalDSP.ml_models.autoencoder import StandardAutoencoder
>>> signal_data = np.random.randn(1000, 100)  # 1000 samples, 100 features
>>> autoencoder = StandardAutoencoder(input_shape=(100,), latent_dim=32)
>>> autoencoder.fit(signal_data, epochs=10)
>>> anomalies, scores, threshold = autoencoder.detect_anomalies(signal_data)

Variational autoencoder:

>>> from vitalDSP.ml_models.autoencoder import VariationalAutoencoder
>>> vae = VariationalAutoencoder(input_shape=(100,), latent_dim=16)
>>> vae.fit(signal_data, epochs=10)

Denoising autoencoder:

>>> from vitalDSP.ml_models.autoencoder import DenoisingAutoencoder
>>> noisy_data = signal_data + np.random.normal(0, 0.1, signal_data.shape)
>>> dae = DenoisingAutoencoder(input_shape=(100,), latent_dim=32)
>>> dae.fit(noisy_data, epochs=10)

class vitalDSP.ml_models.autoencoder.BaseAutoencoder(input_shape: Tuple[int, ...], latent_dim: int = 32, backend: str = 'tensorflow', random_state: int | None = None)[source]

Bases: object

Base class for all autoencoder models.

Provides common functionality for encoding, decoding, and anomaly detection.

compute_reconstruction_error(X: ndarray, metric: str = 'mse') → ndarray[source]

Compute reconstruction error for anomaly detection.

Parameters:

X (np.ndarray) – Input signals
metric (str, default='mse') – Error metric (‘mse’, ‘mae’, ‘rmse’)

Returns:

Reconstruction errors of shape (n_samples,)

Return type:

np.ndarray

decode(latent: ndarray) → ndarray[source]

Decode latent representations to signals.

Parameters:: latent (np.ndarray) – Latent representations of shape (n_samples, latent_dim)
Returns:: Reconstructed signals
Return type:: np.ndarray

detect_anomalies(X: ndarray, threshold: float | None = None, contamination: float = 0.1, metric: str = 'mse') → Tuple[ndarray, ndarray, float][source]

Detect anomalies using reconstruction error.

Parameters:

X (np.ndarray) – Input signals
threshold (float, optional) – Anomaly threshold. If None, computed from contamination
contamination (float, default=0.1) – Expected proportion of anomalies (used if threshold is None)
metric (str, default='mse') – Error metric

Returns:

anomalies (np.ndarray) – Boolean array indicating anomalies
scores (np.ndarray) – Anomaly scores (reconstruction errors)
threshold (float) – Threshold used for detection

encode(X: ndarray) → ndarray[source]

Encode signals to latent space.

Parameters:: X (np.ndarray) – Input signals of shape (n_samples, length) or (n_samples, length, n_channels)
Returns:: Latent representations of shape (n_samples, latent_dim)
Return type:: np.ndarray

load(filepath: str)[source]: Load model from file.

predict(X: ndarray) → ndarray[source]

Predict (reconstruct) signals.

Parameters:: X (np.ndarray) – Input signals of shape (n_samples, length) or (n_samples, length, n_channels)
Returns:: Reconstructed signals
Return type:: np.ndarray

save(filepath: str)[source]: Save model to file.

class vitalDSP.ml_models.autoencoder.ConvolutionalAutoencoder(input_shape: Tuple[int, ...], latent_dim: int = 32, n_filters: List[int] = [32, 64, 128], kernel_sizes: int | List[int] = 7, pool_sizes: int | List[int] = 2, activation: str = 'relu', use_batch_norm: bool = True, dropout_rate: float = 0.2, backend: str = 'tensorflow', random_state: int | None = None)[source]

Convolutional autoencoder for 1D signals.

Architecture: - Encoder: Conv1D layers with pooling - Decoder: Transposed Conv1D layers (upsampling)

Best for signals with spatial structure and local patterns.

Examples

>>> from vitalDSP.ml_models.autoencoder import ConvolutionalAutoencoder
>>> import numpy as np
>>>
>>> # Generate sample signals
>>> X_train = np.random.randn(1000, 500, 1)  # 1000 signals, 500 samples, 1 channel
>>>
>>> # Create and train
>>> cae = ConvolutionalAutoencoder(
...     input_shape=(500, 1),
...     latent_dim=32,
...     n_filters=[32, 64, 128],
...     kernel_sizes=[7, 5, 3],
...     pool_sizes=[2, 2, 2]
... )
>>> cae.fit(X_train, epochs=50)
>>>
>>> # Encode signals
>>> latent_features = cae.encode(X_train[:10])

fit(X: ndarray, **kwargs)[source]: Train the autoencoder. See StandardAutoencoder.fit() for parameters.

class vitalDSP.ml_models.autoencoder.DenoisingAutoencoder(input_shape: Tuple[int, ...], latent_dim: int = 32, hidden_dims: List[int] = [256, 128, 64], noise_type: str = 'gaussian', noise_level: float = 0.1, **kwargs)[source]

Bases: StandardAutoencoder

Denoising Autoencoder (DAE) for signal cleaning.

Trained to reconstruct clean signals from noisy inputs.

Architecture: Same as StandardAutoencoder Training: Add noise to inputs, train to reconstruct clean signals

Examples

>>> from vitalDSP.ml_models.autoencoder import DenoisingAutoencoder
>>> import numpy as np
>>>
>>> # Clean ECG signals
>>> X_clean = np.random.randn(1000, 500)
>>>
>>> # Create and train denoising autoencoder
>>> dae = DenoisingAutoencoder(
...     input_shape=(500,),
...     latent_dim=32,
...     noise_type='gaussian',
...     noise_level=0.1
... )
>>> dae.fit(X_clean, epochs=100)
>>>
>>> # Denoise signals
>>> X_noisy = X_clean + 0.1 * np.random.randn(*X_clean.shape)
>>> X_denoised = dae.predict(X_noisy)

denoise(X: ndarray) → ndarray[source]

Denoise signals.

Parameters:: X (np.ndarray) – Noisy signals
Returns:: Denoised signals
Return type:: np.ndarray

fit(X: ndarray, **kwargs)[source]

Train the denoising autoencoder.

Parameters:

X (np.ndarray) – Clean training signals
**kwargs – Additional arguments passed to StandardAutoencoder.fit()

Return type:

self

class vitalDSP.ml_models.autoencoder.LSTMAutoencoder(input_shape: Tuple[int, ...], latent_dim: int = 32, lstm_units: List[int] = [64, 32], use_bidirectional: bool = False, dropout_rate: float = 0.2, backend: str = 'tensorflow', random_state: int | None = None)[source]

LSTM-based autoencoder for sequential signals.

Architecture: - Encoder: LSTM layers -> Latent representation - Decoder: RepeatVector -> LSTM layers -> Output sequence

Best for signals with temporal dependencies.

Examples

>>> from vitalDSP.ml_models.autoencoder import LSTMAutoencoder
>>> import numpy as np
>>>
>>> # Generate time series data
>>> X_train = np.random.randn(1000, 100, 1)  # 1000 sequences, 100 timesteps, 1 feature
>>>
>>> # Create and train
>>> lstm_ae = LSTMAutoencoder(
...     input_shape=(100, 1),
...     latent_dim=32,
...     lstm_units=[64, 32]
... )
>>> lstm_ae.fit(X_train, epochs=50)
>>>
>>> # Detect anomalies in heartbeat sequences
>>> anomalies, scores, threshold = lstm_ae.detect_anomalies(X_test)

fit(X: ndarray, **kwargs)[source]: Train the autoencoder. See StandardAutoencoder.fit() for parameters.

class vitalDSP.ml_models.autoencoder.StandardAutoencoder(input_shape: Tuple[int, ...], latent_dim: int = 32, hidden_dims: List[int] = [256, 128, 64], activation: str = 'relu', output_activation: str = 'linear', use_batch_norm: bool = True, dropout_rate: float = 0.2, backend: str = 'tensorflow', random_state: int | None = None)[source]

Standard feedforward autoencoder.

Architecture: - Encoder: Input -> Dense layers -> Latent space - Decoder: Latent space -> Dense layers -> Output

Use cases: - Dimensionality reduction - Feature learning - Anomaly detection

Examples

>>> from vitalDSP.ml_models.autoencoder import StandardAutoencoder
>>> import numpy as np
>>>
>>> # Generate sample ECG signals
>>> X_train = np.random.randn(1000, 500)  # 1000 signals, 500 samples each
>>> X_test = np.random.randn(100, 500)
>>>
>>> # Create and train autoencoder
>>> ae = StandardAutoencoder(
...     input_shape=(500,),
...     latent_dim=32,
...     hidden_dims=[256, 128, 64],
...     activation='relu'
... )
>>> ae.fit(X_train, epochs=50, batch_size=32, validation_split=0.2)
>>>
>>> # Detect anomalies
>>> anomalies, scores, threshold = ae.detect_anomalies(X_test, contamination=0.1)
>>> print(f"Detected {anomalies.sum()} anomalies")

fit(X: ndarray, y: ndarray | None = None, epochs: int = 100, batch_size: int = 32, validation_split: float = 0.2, validation_data: Tuple[ndarray, ndarray] | None = None, callbacks: List | None = None, verbose: int = 1)[source]

Train the autoencoder.

Parameters:

X (np.ndarray) – Training signals
y (np.ndarray, optional) – Ignored (for sklearn compatibility)
epochs (int, default=100) – Number of training epochs
batch_size (int, default=32) – Batch size
validation_split (float, default=0.2) – Fraction of data to use for validation
validation_data (tuple, optional) – Validation data (X_val, X_val)
callbacks (list, optional) – Training callbacks
verbose (int, default=1) – Verbosity level

Return type:

self

class vitalDSP.ml_models.autoencoder.VariationalAutoencoder(input_shape: Tuple[int, ...], latent_dim: int = 32, hidden_dims: List[int] = [256, 128, 64], activation: str = 'relu', beta: float = 1.0, use_batch_norm: bool = True, dropout_rate: float = 0.2, backend: str = 'tensorflow', random_state: int | None = None, learning_rate: float = 0.001)[source]

Variational Autoencoder (VAE) for probabilistic signal generation.

Architecture: - Encoder: Input -> mu and log_var (latent distribution parameters) - Sampling: Reparameterization trick - Decoder: Latent sample -> Reconstructed output

Loss: Reconstruction loss + KL divergence

Use cases: - Generative modeling - Signal synthesis - Anomaly detection with probability

Examples

>>> from vitalDSP.ml_models.autoencoder import VariationalAutoencoder
>>> import numpy as np
>>>
>>> # Generate sample data
>>> X_train = np.random.randn(1000, 500)
>>>
>>> # Create and train VAE
>>> vae = VariationalAutoencoder(
...     input_shape=(500,),
...     latent_dim=32,
...     beta=1.0  # KL divergence weight
... )
>>> vae.fit(X_train, epochs=100)
>>>
>>> # Generate new signals
>>> z_samples = np.random.randn(10, 32)
>>> generated_signals = vae.decode(z_samples)

encode(X: ndarray, return_distribution: bool = False) → ndarray[source]

Encode signals to latent space.

Parameters:

X (np.ndarray) – Input signals
return_distribution (bool, default=False) – If True, return (z_mean, z_log_var, z). If False, return only z.

Returns:

Latent representations

Return type:

np.ndarray or tuple

fit(X: ndarray, **kwargs)[source]: Train the VAE. See StandardAutoencoder.fit() for parameters.

sample(n_samples: int = 1) → ndarray[source]

Generate new samples from the learned latent distribution.

Parameters:: n_samples (int, default=1) – Number of samples to generate
Returns:: Generated samples of shape (n_samples, *input_shape)
Return type:: np.ndarray

vitalDSP.ml_models.autoencoder.denoise_signal(X_noisy: ndarray, X_clean: ndarray | None = None, noise_type: str = 'gaussian', noise_level: float = 0.1, **autoencoder_kwargs) → ndarray[source]

Denoise signals using denoising autoencoder.

Parameters:

X_noisy (np.ndarray) – Noisy signals to denoise
X_clean (np.ndarray, optional) – Clean signals for training. If None, use X_noisy as both input and target.
noise_type (str, default='gaussian') – Type of noise
noise_level (float, default=0.1) – Noise intensity
**autoencoder_kwargs – Arguments passed to DenoisingAutoencoder

Returns:

Denoised signals

Return type:

np.ndarray

Examples

>>> from vitalDSP.ml_models.autoencoder import denoise_signal
>>> import numpy as np
>>>
>>> # Generate clean signals
>>> X_clean = np.random.randn(1000, 500)
>>>
>>> # Add noise
>>> X_noisy = X_clean + 0.2 * np.random.randn(*X_clean.shape)
>>>
>>> # Denoise
>>> X_denoised = denoise_signal(
...     X_noisy,
...     X_clean=X_clean,
...     noise_type='gaussian',
...     noise_level=0.1,
...     latent_dim=32
... )

vitalDSP.ml_models.autoencoder.detect_anomalies(X: ndarray, autoencoder_type: str = 'standard', contamination: float = 0.1, **autoencoder_kwargs) → Tuple[ndarray, ndarray, float][source]

Quick anomaly detection using autoencoders.

Parameters:

X (np.ndarray) – Input signals
autoencoder_type (str, default='standard') – Type of autoencoder (‘standard’, ‘conv’, ‘lstm’, ‘vae’)
contamination (float, default=0.1) – Expected proportion of anomalies
**autoencoder_kwargs – Arguments passed to autoencoder constructor

Returns:

anomalies (np.ndarray) – Boolean array indicating anomalies
scores (np.ndarray) – Anomaly scores
threshold (float) – Threshold used for detection

Examples

>>> from vitalDSP.ml_models.autoencoder import detect_anomalies
>>> import numpy as np
>>>
>>> # Generate signals with anomalies
>>> X_normal = np.random.randn(900, 500)
>>> X_anomaly = np.random.randn(100, 500) * 3  # Larger variance
>>> X = np.vstack([X_normal, X_anomaly])
>>>
>>> # Detect anomalies
>>> anomalies, scores, threshold = detect_anomalies(
...     X,
...     autoencoder_type='standard',
...     contamination=0.1,
...     latent_dim=32
... )
>>> print(f"Detected {anomalies.sum()} anomalies")

Transformer Models

vitalDSP Transformer Model for Physiological Signals

State-of-the-art Transformer architecture for long-range dependency modeling in physiological signal analysis.

Features: - Multi-head self-attention mechanism - Positional encoding - Layer normalization - Feed-forward networks with residual connections - Encoder-only (BERT-style) and Encoder-Decoder architectures - Optimized for 1D time series data

Applications: - Long ECG signal classification - Multi-lead ECG interpretation - EEG temporal pattern recognition - Long-term signal forecasting - Sequence-to-sequence tasks

Author: vitalDSP Team Date: 2025

class vitalDSP.ml_models.transformer_model.MultiHeadSelfAttention(d_model, n_heads, dropout_rate=0.1, **kwargs)[source]

Bases: object

Multi-Head Self-Attention mechanism (TensorFlow).

Parameters:

d_model (int) – Dimension of model
n_heads (int) – Number of attention heads
dropout_rate (float) – Dropout rate

call(x, mask=None, training=False)[source]

split_heads(x, batch_size)[source]: Split the last dimension into (n_heads, depth).

class vitalDSP.ml_models.transformer_model.PositionalEncoding(d_model, max_len=5000, **kwargs)[source]

Bases: object

Positional Encoding for Transformer (TensorFlow).

Adds positional information to input embeddings using sinusoidal functions.

build(input_shape)[source]

call(x)[source]

class vitalDSP.ml_models.transformer_model.TransformerEncoderLayer(d_model, n_heads, d_ff, dropout_rate=0.1, **kwargs)[source]

Bases: object

Single Transformer Encoder Layer (TensorFlow).

Consists of: - Multi-head self-attention - Feed-forward network - Layer normalization - Residual connections

call(x, mask=None, training=False)[source]

class vitalDSP.ml_models.transformer_model.TransformerModel(input_shape: Tuple[int, ...], n_classes: int, d_model: int = 128, n_heads: int = 8, n_layers: int = 4, d_ff: int = 512, dropout_rate: float = 0.1, max_len: int = 5000, task: str = 'classification', backend: str = 'tensorflow', **kwargs)[source]

Bases: object

Transformer model for physiological signal analysis.

Features: - Multi-head self-attention for capturing long-range dependencies - Positional encoding for sequence information - Stacked encoder layers - Classification or regression head - Optimized for 1D time series

Parameters:

input_shape (tuple) – Shape of input sequences (sequence_length, n_features)
n_classes (int) – Number of output classes (use 1 for regression)
d_model (int, default=128) – Dimension of model (embedding dimension)
n_heads (int, default=8) – Number of attention heads
n_layers (int, default=4) – Number of transformer encoder layers
d_ff (int, default=512) – Dimension of feed-forward network
dropout_rate (float, default=0.1) – Dropout rate
max_len (int, default=5000) – Maximum sequence length for positional encoding
task (str, default='classification') – Task type (‘classification’ or ‘regression’)
backend (str, default='tensorflow') – Backend framework (‘tensorflow’ or ‘pytorch’)

model

The transformer model

Type:: keras.Model or torch.nn.Module

history

Training history

Type:: dict

Examples

>>> from vitalDSP.ml_models import TransformerModel
>>>
>>> # Long ECG classification
>>> model = TransformerModel(
...     input_shape=(5000, 1),
...     n_classes=5,
...     d_model=128,
...     n_heads=8,
...     n_layers=4
... )
>>>
>>> model.build_model()
>>> history = model.train(X_train, y_train, epochs=100)
>>> predictions = model.predict(X_test)

build_model()[source]: Build the Transformer model architecture.

get_attention_weights(X: ndarray, layer_idx: int = 0)[source]

Extract attention weights for visualization.

Parameters:

X (ndarray) – Input data
layer_idx (int) – Index of transformer layer to extract attention from

Returns:

attention_weights – Attention weight matrices

Return type:

ndarray

load(filepath: str)[source]: Load model from disk.

predict(X: ndarray) → ndarray[source]

Make predictions.

Parameters:: X (ndarray of shape (n_samples, sequence_length, n_features)) – Input data
Returns:: predictions – Model predictions
Return type:: ndarray

save(filepath: str)[source]: Save model to disk.

train(X_train: ndarray, y_train: ndarray, X_val: ndarray | None = None, y_val: ndarray | None = None, epochs: int = 100, batch_size: int = 32, learning_rate: float = 0.0001, warmup_epochs: int = 10, **kwargs)[source]

Train the Transformer model.

Parameters:

X_train (ndarray of shape (n_samples, sequence_length, n_features)) – Training data
y_train (ndarray) – Training labels
X_val (ndarray, optional) – Validation data
y_val (ndarray, optional) – Validation labels
epochs (int, default=100) – Number of training epochs
batch_size (int, default=32) – Batch size
learning_rate (float, default=0.0001) – Initial learning rate
warmup_epochs (int, default=10) – Number of warmup epochs with linear LR increase

Returns:

history – Training history

Return type:

Feature Extractor

vitalDSP Machine Learning Feature Extraction Module

Comprehensive feature extraction pipeline for machine learning applications with physiological signals (ECG, PPG, EEG, respiratory signals).

This module provides: - Automated feature extraction from multiple domains - Feature engineering and selection - Dimensionality reduction - Pipeline integration with scikit-learn - Feature importance analysis

Author: vitalDSP Team Date: 2025

class vitalDSP.ml_models.feature_extractor.FeatureEngineering(interaction_terms: bool = False, polynomial_degree: int = 1, **kwargs)[source]

Bases: object

Advanced feature engineering for physiological signals.

Provides methods for: - Feature construction - Feature interaction - Polynomial features - Domain-specific transformations

Parameters:

interaction_terms (bool, default=False) – Whether to create interaction terms
polynomial_degree (int, default=1) – Degree of polynomial features (1 = no polynomial features)

Examples

>>> from vitalDSP.ml_models import FeatureEngineering
>>> engineer = FeatureEngineering(interaction_terms=True)
>>> X_engineered = engineer.fit_transform(X_features)

fit(X: ndarray, y: ndarray | None = None)[source]: Fit the feature engineer.

fit_transform(X: ndarray, y: ndarray | None = None) → ndarray[source]: Fit and transform features.

transform(X: ndarray) → ndarray[source]: Transform features.

class vitalDSP.ml_models.feature_extractor.FeatureExtractor(signal_type: str = 'ecg', sampling_rate: float = 250.0, domains: List[str] | None = None, normalize: bool = True, feature_selection: str | None = None, n_features: int | None = None, **kwargs)[source]

Bases: BaseEstimator, TransformerMixin

Comprehensive feature extraction for physiological signals.

Extracts features from multiple domains: - Time domain (statistical features) - Frequency domain (spectral features) - Time-frequency domain (wavelet, STFT) - Nonlinear dynamics (entropy, fractal dimension) - Morphological features (peaks, waveform shape)

Compatible with scikit-learn pipelines.

Parameters:

signal_type (str, default='ecg') – Type of physiological signal (‘ecg’, ‘ppg’, ‘eeg’, ‘resp’)
sampling_rate (float) – Sampling rate of the signal in Hz
domains (list of str, default=['time', 'frequency', 'nonlinear']) – Feature domains to extract from
normalize (bool, default=True) – Whether to normalize features
feature_selection (str, optional) – Feature selection method (‘kbest’, ‘mutual_info’, ‘pca’, None)
n_features (int, optional) – Number of features to select (if feature_selection is not None)

feature_names_

Names of extracted features

Type:: list

n_features_in_

Number of input samples

Type:: int

feature_importances_

Feature importance scores (if available)

Type:: dict

Examples

>>> from vitalDSP.ml_models import FeatureExtractor
>>> extractor = FeatureExtractor(signal_type='ecg', sampling_rate=250)
>>> features = extractor.fit_transform(signals)
>>> print(f"Extracted {features.shape[1]} features")

>>> # Use in scikit-learn pipeline
>>> from sklearn.pipeline import Pipeline
>>> from sklearn.ensemble import RandomForestClassifier
>>> pipeline = Pipeline([
...     ('features', FeatureExtractor(signal_type='ecg', sampling_rate=250)),
...     ('classifier', RandomForestClassifier())
... ])
>>> pipeline.fit(X_train, y_train)

fit(X: ndarray | List[ndarray], y: ndarray | None = None)[source]

Fit the feature extractor.

Parameters:

X (array-like of shape (n_samples,) or (n_samples, n_timesteps)) – Training signals
y (array-like of shape (n_samples,), optional) – Target values for supervised feature selection

Returns:

self – Fitted transformer

Return type:

object

get_feature_importances() → Dict[str, float][source]

Get feature importance scores (if available).

Returns:: importances – Dictionary mapping feature names to importance scores
Return type:: dict

get_feature_names() → List[str][source]

Get names of extracted features.

Returns:: feature_names – Names of all extracted features
Return type:: list of str

transform(X: ndarray | List[ndarray]) → ndarray[source]

Transform signals to feature matrix.

Parameters:: X (array-like of shape (n_samples,) or (n_samples, n_timesteps)) – Input signals
Returns:: X_transformed – Extracted features
Return type:: ndarray of shape (n_samples, n_features)

vitalDSP.ml_models.feature_extractor.extract_features(signals: ndarray | List[ndarray], signal_type: str = 'ecg', sampling_rate: float = 250.0, domains: List[str] | None = None, return_dataframe: bool = False) → ndarray | DataFrame[source]

Quick feature extraction from physiological signals.

Parameters:

signals (array-like) – Input signals
signal_type (str, default='ecg') – Type of signal
sampling_rate (float, default=250.0) – Sampling rate in Hz
domains (list of str, optional) – Feature domains to extract
return_dataframe (bool, default=False) – If True, return pandas DataFrame with feature names

Returns:

features – Extracted features

Return type:

ndarray or DataFrame

Examples

>>> features = extract_features(ecg_signals, signal_type='ecg', sampling_rate=250)
>>> print(f"Shape: {features.shape}")

Transfer Learning

Transfer Learning for Physiological Signal Analysis

This module provides utilities for transfer learning, enabling efficient model adaptation to new tasks and domains with limited labeled data.

Features: - Feature extraction (frozen base model) - Fine-tuning strategies - Domain adaptation techniques - Few-shot learning support - Progressive unfreezing - Learning rate scheduling

Use Cases: - Adapt ECG classifier to different populations - Transfer from large to small datasets - Cross-domain transfer (e.g., ECG to PPG) - Personalized model adaptation

Author: vitalDSP License: MIT

class vitalDSP.ml_models.transfer_learning.DomainAdapter(base_model: Any, method: str = 'mmd', backend: str = 'tensorflow')[source]

Bases: object

Domain adaptation for transfer learning.

Adapts model from source domain to target domain when distributions differ.

Techniques: - Feature alignment - Adversarial training - Maximum Mean Discrepancy (MMD)

Use cases: - Transfer from one patient population to another - Adapt across different sensors/devices - Cross-institution transfer

Examples

>>> from vitalDSP.ml_models.transfer_learning import DomainAdapter
>>> import numpy as np
>>>
>>> # Source and target data from different domains
>>> X_source = np.random.randn(500, 187, 1)
>>> y_source = np.random.randint(0, 3, 500)
>>> X_target = np.random.randn(100, 187, 1) * 1.5  # Different distribution
>>>
>>> # Create domain adapter
>>> adapter = DomainAdapter(base_model, method='mmd')
>>> adapter.fit(X_source, y_source, X_target, epochs=50)

fit(X_source: ndarray, y_source: ndarray, X_target: ndarray, y_target: ndarray | None = None, n_classes: int | None = None, epochs: int = 50, batch_size: int = 32, learning_rate: float = 0.0001, lambda_da: float = 0.1, verbose: int = 1) → Dict[str, List[float]][source]

Fit domain adaptation model.

Parameters:

X_source (np.ndarray) – Source domain signals (labeled)
y_source (np.ndarray) – Source domain labels
X_target (np.ndarray) – Target domain signals (can be unlabeled)
y_target (np.ndarray, optional) – Target domain labels (if available)
n_classes (int, optional) – Number of classes
epochs (int, default=50) – Training epochs
batch_size (int, default=32) – Batch size
learning_rate (float, default=1e-4) – Learning rate
lambda_da (float, default=0.1) – Weight for domain adaptation loss
verbose (int, default=1) – Verbosity level

Returns:

Training history

Return type:

predict(X: ndarray, batch_size: int = 32) → ndarray[source]: Make predictions.

class vitalDSP.ml_models.transfer_learning.FineTuner(base_model: Any, strategy: str = 'all_at_once', backend: str = 'tensorflow')[source]

Bases: TransferLearningStrategy

Fine-tuning transfer learning.

Unfreezes part or all of the base model and trains with a small learning rate.

Best when: - Target dataset is moderately sized - Target task differs somewhat from source task - More accuracy is needed than feature extraction provides

Strategies: 1. All-at-once: Unfreeze all layers, train with small LR 2. Progressive: Gradually unfreeze layers from top to bottom 3. Discriminative: Use different learning rates for different layers

Examples

>>> from vitalDSP.ml_models.transfer_learning import FineTuner
>>> import numpy as np
>>>
>>> # Assuming we have a base model
>>> finetuner = FineTuner(base_model, strategy='progressive')
>>>
>>> # Fine-tune with progressive unfreezing
>>> X_train = np.random.randn(500, 187, 1)
>>> y_train = np.random.randint(0, 4, 500)
>>> finetuner.fit(X_train, y_train, n_classes=4, epochs=30)

fit(X_train: ndarray, y_train: ndarray, n_classes: int | None = None, task: str = 'classification', X_val: ndarray | None = None, y_val: ndarray | None = None, epochs: int = 50, batch_size: int = 32, base_learning_rate: float = 1e-05, head_learning_rate: float = 0.001, progressive_epochs: List[int] | None = None, verbose: int = 1) → Dict[str, List[float]][source]

Fine-tune model on new task.

Parameters:

X_train (np.ndarray) – Training signals
y_train (np.ndarray) – Training labels
n_classes (int, optional) – Number of classes
task (str, default='classification') – Task type
X_val (np.ndarray, optional) – Validation signals
y_val (np.ndarray, optional) – Validation labels
epochs (int, default=50) – Total training epochs
batch_size (int, default=32) – Batch size
base_learning_rate (float, default=1e-5) – Learning rate for base model (smaller)
head_learning_rate (float, default=1e-3) – Learning rate for new head (larger)
progressive_epochs (list of int, optional) – Epochs at which to unfreeze layers (for progressive strategy)
verbose (int, default=1) – Verbosity level

Returns:

Training history

Return type:

freeze_layers(n_layers: int | None = None)[source]: Freeze layers.

predict(X: ndarray, batch_size: int = 32) → ndarray[source]: Make predictions.

unfreeze_layers(n_layers: int | None = None)[source]: Unfreeze layers.

class vitalDSP.ml_models.transfer_learning.TransferFeatureExtractor(base_model: Any, freeze_base: bool = True, backend: str = 'tensorflow')[source]

Bases: TransferLearningStrategy

Feature extraction transfer learning.

Uses pre-trained model as fixed feature extractor, only training new classification/regression head.

This is the fastest approach and works well when: - Source and target tasks are similar - Target dataset is small - Computational resources are limited

Examples

>>> from vitalDSP.ml_models.transfer_learning import TransferFeatureExtractor
>>> import numpy as np
>>> import tensorflow as tf
>>>
>>> # Create base model
>>> base = tf.keras.Sequential([
...     tf.keras.layers.Conv1D(32, 7, input_shape=(187, 1)),
...     tf.keras.layers.MaxPooling1D(2),
...     tf.keras.layers.GlobalAveragePooling1D()
... ])
>>>
>>> # Create feature extractor
>>> extractor = TransferFeatureExtractor(base, freeze_base=True)
>>>
>>> # Fit to new task
>>> X = np.random.randn(100, 187, 1)
>>> y = np.random.randint(0, 3, 100)
>>> extractor.fit(X, y, n_classes=3, epochs=10)

fit(X_train: ndarray, y_train: ndarray, n_classes: int | None = None, n_outputs: int = 1, task: str = 'classification', X_val: ndarray | None = None, y_val: ndarray | None = None, epochs: int = 50, batch_size: int = 32, learning_rate: float = 0.001, verbose: int = 1) → Dict[str, List[float]][source]

Fit feature extractor to new task.

Parameters:

X_train (np.ndarray) – Training signals
y_train (np.ndarray) – Training labels
n_classes (int, optional) – Number of classes (for classification)
n_outputs (int, default=1) – Number of outputs (for regression)
task (str, default='classification') – Task type (‘classification’ or ‘regression’)
X_val (np.ndarray, optional) – Validation signals
y_val (np.ndarray, optional) – Validation labels
epochs (int, default=50) – Number of training epochs
batch_size (int, default=32) – Batch size
learning_rate (float, default=1e-3) – Learning rate
verbose (int, default=1) – Verbosity level

Returns:

Training history

Return type:

freeze_layers(n_layers: int | None = None)[source]: Freeze base model layers.

predict(X: ndarray, batch_size: int = 32) → ndarray[source]: Make predictions.

unfreeze_layers(n_layers: int | None = None)[source]: Unfreeze base model layers.

class vitalDSP.ml_models.transfer_learning.TransferLearningStrategy(base_model: Any, backend: str = 'tensorflow')[source]

Bases: object

Base class for transfer learning strategies.

Implements common functionality for adapting pre-trained models to new tasks.

Examples

>>> from vitalDSP.ml_models.transfer_learning import TransferFeatureExtractor
>>> from vitalDSP.ml_models.pretrained_models import load_pretrained_model
>>> import numpy as np
>>>
>>> # Load pre-trained model
>>> base_model = load_pretrained_model('ecg_classifier_mitbih')
>>>
>>> # Create feature extractor
>>> extractor = TransferFeatureExtractor(base_model.model, freeze_base=True)
>>>
>>> # Train on new task
>>> X_train = np.random.randn(100, 187, 1)
>>> y_train = np.random.randint(0, 3, 100)  # New 3-class task
>>> extractor.fit(X_train, y_train, n_classes=3, epochs=20)

freeze_layers(n_layers: int | None = None)[source]

Freeze layers in base model.

Parameters:: n_layers (int, optional) – Number of layers to freeze. If None, freeze all.

unfreeze_layers(n_layers: int | None = None)[source]

Unfreeze layers for fine-tuning.

Parameters:: n_layers (int, optional) – Number of layers to unfreeze from the end. If None, unfreeze all.

vitalDSP.ml_models.transfer_learning.quick_transfer(base_model: Any, X_train: ndarray, y_train: ndarray, strategy: str = 'feature_extraction', n_classes: int | None = None, epochs: int = 30, **kwargs) → TransferFeatureExtractor | FineTuner[source]

Quick transfer learning with sensible defaults.

Parameters:

base_model (object) – Pre-trained base model
X_train (np.ndarray) – Training signals
y_train (np.ndarray) – Training labels
strategy (str, default='feature_extraction') – Transfer strategy (‘feature_extraction’, ‘fine_tuning’)
n_classes (int, optional) – Number of classes
epochs (int, default=30) – Training epochs
**kwargs – Additional arguments

Returns:

Trained transfer learning model

Return type:

TransferFeatureExtractor or FineTuner

Examples

>>> from vitalDSP.ml_models.transfer_learning import quick_transfer
>>> from vitalDSP.ml_models.pretrained_models import load_pretrained_model
>>> import numpy as np
>>>
>>> # Load pre-trained model
>>> base = load_pretrained_model('ecg_classifier_mitbih')
>>>
>>> # Quick transfer learning
>>> X = np.random.randn(200, 187, 1)
>>> y = np.random.randint(0, 3, 200)
>>> model = quick_transfer(base.model, X, y, strategy='feature_extraction', epochs=20)

Pre-trained Models

Pre-trained Models for Physiological Signal Analysis

This module provides a repository of pre-trained models for common physiological signal analysis tasks, enabling transfer learning and quick deployment.

Features: - Automatic model download and caching - Pre-trained models for ECG, PPG, EEG - Model versioning and metadata - Easy fine-tuning interface

Available Models: 1. ECG Classification (MIT-BIH, PTB-XL) 2. PPG Quality Assessment 3. EEG Sleep Stage Classification 4. Arrhythmia Detection 5. Heart Rate Estimation

Author: vitalDSP License: MIT

class vitalDSP.ml_models.pretrained_models.ModelHub(cache_dir: str | None = None)[source]

Bases: object

Central hub for managing pre-trained models.

Provides utilities for: - Listing available models - Downloading models - Managing model cache - Model comparison

Examples

>>> from vitalDSP.ml_models.pretrained_models import ModelHub
>>>
>>> # List all available models
>>> hub = ModelHub()
>>> models = hub.list_models()
>>> print(models)
>>>
>>> # Filter models by signal type
>>> ecg_models = hub.list_models(signal_type='ecg', task='classification')
>>> print(ecg_models)
>>>
>>> # Download specific model
>>> model = hub.get_model('ecg_classifier_mitbih')

clear_cache(model_name: str | None = None)[source]

Clear model cache.

Parameters:: model_name (str, optional) – Specific model to clear. If None, clear all.

compare_models(model_names: List[str], metric: str = 'accuracy') → Dict[str, float][source]

Compare models by a specific metric.

Parameters:

model_names (list of str) – Names of models to compare
metric (str, default='accuracy') – Metric to compare

Returns:

Model names and their metric values

Return type:

get_cache_size() → float[source]

Get total size of cached models in MB.

Returns:: Total cache size in MB
Return type:: float

get_model(model_name: str, force_download: bool = False) → PretrainedModel[source]

Get pre-trained model.

Parameters:

model_name (str) – Name of model
force_download (bool, default=False) – Force re-download

Returns:

Loaded model

Return type:

PretrainedModel

list_models(signal_type: str | None = None, task: str | None = None, architecture: str | None = None) → List[Dict[str, Any]][source]

List available models with optional filtering.

Parameters:

signal_type (str, optional) – Filter by signal type (‘ecg’, ‘ppg’, ‘eeg’)
task (str, optional) – Filter by task (‘classification’, ‘regression’, ‘autoencoder’)
architecture (str, optional) – Filter by architecture (‘cnn1d’, ‘lstm’, ‘transformer’)

Returns:

List of model metadata

Return type:

list of dict

class vitalDSP.ml_models.pretrained_models.PretrainedModel(model: Any, metadata: Dict[str, Any], backend: str = 'tensorflow')[source]

Bases: object

Wrapper for pre-trained physiological signal analysis models.

Provides a unified interface for loading, using, and fine-tuning pre-trained models regardless of architecture or backend.

Examples

>>> from vitalDSP.ml_models.pretrained_models import PretrainedModel
>>> import numpy as np
>>>
>>> # Load pre-trained ECG classifier
>>> model = PretrainedModel.from_registry('ecg_classifier_mitbih')
>>> print(model.info())
>>>
>>> # Make predictions
>>> ecg_signals = np.random.randn(10, 187, 1)
>>> predictions = model.predict(ecg_signals)
>>> print(f"Predicted classes: {predictions}")
>>>
>>> # Fine-tune on your data
>>> X_train = np.random.randn(100, 187, 1)
>>> y_train = np.random.randint(0, 5, 100)
>>> model.fine_tune(X_train, y_train, epochs=10)

evaluate(X_test: ndarray, y_test: ndarray, batch_size: int = 32) → Dict[str, float][source]

Evaluate model on test data.

Parameters:

X_test (np.ndarray) – Test signals
y_test (np.ndarray) – Test labels
batch_size (int, default=32) – Batch size

Returns:

Evaluation metrics

Return type:

fine_tune(X_train: ndarray, y_train: ndarray, X_val: ndarray | None = None, y_val: ndarray | None = None, epochs: int = 10, batch_size: int = 32, learning_rate: float = 0.0001, freeze_layers: int | None = None, verbose: int = 1) → Dict[str, List[float]][source]

Fine-tune model on new data.

Parameters:

X_train (np.ndarray) – Training signals
y_train (np.ndarray) – Training labels
X_val (np.ndarray, optional) – Validation signals
y_val (np.ndarray, optional) – Validation labels
epochs (int, default=10) – Number of training epochs
batch_size (int, default=32) – Batch size
learning_rate (float, default=1e-4) – Learning rate (smaller than training from scratch)
freeze_layers (int, optional) – Number of initial layers to freeze
verbose (int, default=1) – Verbosity level

Returns:

Training history

Return type:

classmethod from_registry(model_name: str, cache_dir: str | None = None, force_download: bool = False) → PretrainedModel[source]

Load model from registry.

Parameters:

model_name (str) – Name of the model in the registry
cache_dir (str, optional) – Directory to cache downloaded models. If None, use default.
force_download (bool, default=False) – Force re-download even if cached

Returns:

Loaded pre-trained model

Return type:

PretrainedModel

Raises:

ValueError – If model not found in registry

get_features(X: ndarray, layer_name: str | None = None, batch_size: int = 32) → ndarray[source]

Extract features from intermediate layer.

Parameters:

X (np.ndarray) – Input signals
layer_name (str, optional) – Name of layer to extract features from. If None, use second-to-last layer.
batch_size (int, default=32) – Batch size

Returns:

Extracted features

Return type:

np.ndarray

get_layer_names() → List[str][source]: Get list of layer names in the model.

info() → str[source]

Get model information.

Returns:: Formatted model information
Return type:: str

classmethod load(filepath: str) → PretrainedModel[source]

Load model from file.

Parameters:: filepath (str) – Path to saved model
Returns:: Loaded model
Return type:: PretrainedModel

predict(X: ndarray, batch_size: int = 32, return_proba: bool = False) → ndarray[source]

Make predictions on new data.

Parameters:

X (np.ndarray) – Input signals
batch_size (int, default=32) – Batch size for prediction
return_proba (bool, default=False) – Return class probabilities (classification only)

Returns:

Predictions

Return type:

np.ndarray

save(filepath: str)[source]

Save model to file.

Parameters:: filepath (str) – Path to save model

vitalDSP.ml_models.pretrained_models.load_pretrained_model(model_name: str, cache_dir: str | None = None, force_download: bool = False) → PretrainedModel[source]

Quick function to load a pre-trained model.

Parameters:

model_name (str) – Name of model from registry
cache_dir (str, optional) – Cache directory
force_download (bool, default=False) – Force re-download

Returns:

Loaded model

Return type:

PretrainedModel

Examples

>>> from vitalDSP.ml_models.pretrained_models import load_pretrained_model
>>> import numpy as np
>>>
>>> # Load ECG classifier
>>> model = load_pretrained_model('ecg_classifier_mitbih')
>>>
>>> # Make predictions
>>> ecg_signal = np.random.randn(1, 187, 1)
>>> prediction = model.predict(ecg_signal)
>>> print(f"Predicted class: {prediction[0]}")

Model Explainability

Explainable AI (XAI) for Physiological Signal Analysis

This module provides interpretability and explainability tools for machine learning and deep learning models applied to physiological signals.

Supported Methods: 1. SHAP (SHapley Additive exPlanations) 2. LIME (Local Interpretable Model-agnostic Explanations) 3. GradCAM for 1D signals 4. Attention Visualization 5. Feature Importance Analysis

Author: vitalDSP License: MIT

class vitalDSP.ml_models.explainability.AttentionVisualizer[source]

Bases: object

Visualize attention weights from transformer models.

Examples

>>> from vitalDSP.ml_models.explainability import AttentionVisualizer
>>> import numpy as np
>>>
>>> # Assume we have attention weights from a transformer
>>> attention_weights = np.random.rand(8, 100, 100)  # (n_heads, seq_len, seq_len)
>>>
>>> # Create visualizer
>>> viz = AttentionVisualizer()
>>>
>>> # Plot attention patterns
>>> viz.plot_attention_map(attention_weights, head_idx=0)
>>> viz.plot_attention_rollout(attention_weights)

plot_attention_map(attention_weights: ndarray, head_idx: int = 0, figsize: Tuple[int, int] = (10, 10), title: str | None = None)[source]

Plot attention map for a specific head.

Parameters:

attention_weights (np.ndarray) – Attention weights of shape (n_heads, seq_len, seq_len)
head_idx (int, default=0) – Index of attention head to visualize
figsize (tuple, default=(10, 10)) – Figure size
title (str, optional) – Plot title

plot_attention_rollout(attention_weights: ndarray, figsize: Tuple[int, int] = (12, 6))[source]

Plot attention rollout (average across heads and layers).

Parameters:

attention_weights (np.ndarray) – Attention weights of shape (n_heads, seq_len, seq_len)
figsize (tuple, default=(12, 6)) – Figure size

plot_head_comparison(attention_weights: ndarray, query_idx: int = 0, figsize: Tuple[int, int] = (15, 4))[source]

Compare attention patterns across different heads.

Parameters:

attention_weights (np.ndarray) – Attention weights of shape (n_heads, seq_len, seq_len)
query_idx (int, default=0) – Query position to visualize
figsize (tuple, default=(15, 4)) – Figure size

class vitalDSP.ml_models.explainability.BaseExplainer(model: Any, feature_names: List[str] | None = None, class_names: List[str] | None = None)[source]

Bases: object

Base class for all explainability methods.

Provides common functionality for model interpretation and visualization.

explain(X: ndarray, **kwargs) → Dict[str, Any][source]

Generate explanations for predictions.

Parameters:

X (np.ndarray) – Input samples to explain
**kwargs – Additional arguments

Returns:

Explanation results

Return type:

plot(explanation: Dict[str, Any], **kwargs)[source]

Visualize explanations.

Parameters:

explanation (dict) – Explanation to visualize
**kwargs – Additional plotting arguments

class vitalDSP.ml_models.explainability.GradCAM1D(model: Any, layer_name: str | None = None, backend: str = 'tensorflow')[source]

Bases: object

Gradient-weighted Class Activation Mapping (GradCAM) for 1D signals.

Visualizes which parts of the signal are important for the prediction by computing gradients of the output with respect to feature maps.

Examples

>>> from vitalDSP.ml_models.explainability import GradCAM1D
>>> import tensorflow as tf
>>> import numpy as np
>>>
>>> # Assume we have a trained 1D CNN model
>>> model = tf.keras.models.load_model('my_cnn_model.h5')
>>>
>>> # Create GradCAM explainer
>>> gradcam = GradCAM1D(model, layer_name='conv1d_3')
>>>
>>> # Generate heatmap for a signal
>>> signal = np.random.randn(1, 1000, 1)
>>> heatmap = gradcam.compute_heatmap(signal, class_idx=1)
>>>
>>> # Visualize
>>> gradcam.plot_overlay(signal[0], heatmap)

compute_heatmap(signal: ndarray, class_idx: int | None = None) → ndarray[source]

Compute GradCAM heatmap.

Parameters:

signal (np.ndarray) – Input signal of shape (batch, length, channels) or (length, channels)
class_idx (int, optional) – Target class index. If None, use predicted class.

Returns:

Heatmap of shape (length,)

Return type:

np.ndarray

plot_overlay(signal: ndarray, heatmap: ndarray, alpha: float = 0.4, colormap: str = 'jet', figsize: Tuple[int, int] = (15, 5))[source]

Plot signal with GradCAM heatmap overlay.

Parameters:

signal (np.ndarray) – Original signal (1D or 2D with channels)
heatmap (np.ndarray) – GradCAM heatmap
alpha (float, default=0.4) – Transparency of heatmap overlay
colormap (str, default='jet') – Colormap for heatmap
figsize (tuple, default=(15, 5)) – Figure size

class vitalDSP.ml_models.explainability.LIMEExplainer(model: Any, training_data: ndarray, mode: str = 'classification', feature_names: List[str] | None = None, class_names: List[str] | None = None, discretize_continuous: bool = False)[source]

Bases: BaseExplainer

LIME (Local Interpretable Model-agnostic Explanations).

LIME explains individual predictions by approximating the model locally with an interpretable model.

Examples

>>> from vitalDSP.ml_models.explainability import LIMEExplainer
>>> from sklearn.ensemble import RandomForestClassifier
>>> import numpy as np
>>>
>>> # Train a model
>>> X_train = np.random.randn(1000, 50)
>>> y_train = np.random.randint(0, 2, 1000)
>>> model = RandomForestClassifier()
>>> model.fit(X_train, y_train)
>>>
>>> # Create explainer
>>> explainer = LIMEExplainer(
...     model,
...     training_data=X_train,
...     feature_names=[f'feature_{i}' for i in range(50)],
...     class_names=['Normal', 'Abnormal']
... )
>>>
>>> # Explain a prediction
>>> X_test = np.random.randn(1, 50)
>>> explanation = explainer.explain(X_test[0])
>>> explainer.plot(explanation)

explain(instance: ndarray, num_features: int = 10, num_samples: int = 5000, labels: Tuple[int, ...] | None = None) → Any[source]

Explain a single prediction.

Parameters:

instance (np.ndarray) – Instance to explain (1D array)
num_features (int, default=10) – Number of features to include in explanation
num_samples (int, default=5000) – Number of samples for local approximation
labels (tuple of int, optional) – Labels to explain (for classification)

Returns:

LIME explanation object

Return type:

lime.explanation.Explanation

plot(explanation: Any, label: int | None = None, figsize: Tuple[int, int] = (10, 6))[source]

Plot LIME explanation.

Parameters:

explanation (lime.explanation.Explanation) – Explanation to plot
label (int, optional) – Label to visualize (for classification)
figsize (tuple, default=(10, 6)) – Figure size

class vitalDSP.ml_models.explainability.SHAPExplainer(model: Any, explainer_type: str = 'kernel', feature_names: List[str] | None = None, class_names: List[str] | None = None)[source]

Bases: BaseExplainer

SHAP (SHapley Additive exPlanations) for model interpretation.

SHAP values represent the contribution of each feature to the prediction, based on game-theoretic Shapley values.

Supports: - TreeExplainer (for tree-based models) - DeepExplainer (for deep learning models) - KernelExplainer (model-agnostic)

Examples

>>> from vitalDSP.ml_models.explainability import SHAPExplainer
>>> from sklearn.ensemble import RandomForestClassifier
>>> import numpy as np
>>>
>>> # Train a model
>>> X_train = np.random.randn(1000, 50)
>>> y_train = np.random.randint(0, 2, 1000)
>>> model = RandomForestClassifier()
>>> model.fit(X_train, y_train)
>>>
>>> # Create explainer
>>> explainer = SHAPExplainer(
...     model,
...     explainer_type='tree',
...     feature_names=[f'feature_{i}' for i in range(50)]
... )
>>>
>>> # Explain predictions
>>> X_test = np.random.randn(10, 50)
>>> explanations = explainer.explain(X_test, background_data=X_train)
>>>
>>> # Visualize
>>> explainer.plot_summary(explanations)
>>> explainer.plot_waterfall(explanations, instance_idx=0)

explain(X: ndarray, background_data: ndarray | None = None, nsamples: int = 100) → Dict[str, Any][source]

Generate SHAP explanations.

Parameters:

X (np.ndarray) – Samples to explain
background_data (np.ndarray, optional) – Background dataset for kernel/deep explainers
nsamples (int, default=100) – Number of samples for kernel explainer

Returns:

Dictionary containing: - ‘shap_values’: SHAP values for each sample - ‘base_values’: Base values (expected model output) - ‘data’: Original input data

Return type:

plot_dependence(feature_idx: int | str, explanation: Dict[str, Any] | None = None, interaction_idx: int | str | None = 'auto', show: bool = True)[source]

Create SHAP dependence plot showing feature interaction.

Parameters:

feature_idx (int or str) – Feature to plot
explanation (dict, optional) – Explanation to plot
interaction_idx (int, str, or None, default='auto') – Feature to show interaction with
show (bool, default=True) – Whether to show the plot

plot_force(explanation: Dict[str, Any] | None = None, instance_idx: int = 0, matplotlib: bool = True)[source]

Create SHAP force plot.

Parameters:

explanation (dict, optional) – Explanation to plot
instance_idx (int, default=0) – Index of instance to explain
matplotlib (bool, default=True) – Whether to use matplotlib backend

plot_summary(explanation: Dict[str, Any] | None = None, plot_type: str = 'dot', max_display: int = 20, show: bool = True)[source]

Create SHAP summary plot.

Parameters:

explanation (dict, optional) – Explanation to plot. If None, use last explanation.
plot_type (str, default='dot') – Type of plot (‘dot’, ‘bar’, ‘violin’)
max_display (int, default=20) – Maximum number of features to display
show (bool, default=True) – Whether to show the plot

plot_waterfall(explanation: Dict[str, Any] | None = None, instance_idx: int = 0, max_display: int = 20, show: bool = True)[source]

Create SHAP waterfall plot for a single prediction.

Parameters:

explanation (dict, optional) – Explanation to plot
instance_idx (int, default=0) – Index of instance to explain
max_display (int, default=20) – Maximum number of features to display
show (bool, default=True) – Whether to show the plot

vitalDSP.ml_models.explainability.explain_prediction(model: Any, X: ndarray, method: str = 'shap', background_data: ndarray | None = None, feature_names: List[str] | None = None, class_names: List[str] | None = None, **kwargs) → Dict[str, Any][source]

Quick explanation of model predictions.

Parameters:

model (object) – Trained model
X (np.ndarray) – Samples to explain
method (str, default='shap') – Explanation method (‘shap’, ‘lime’)
background_data (np.ndarray, optional) – Background data (required for some methods)
feature_names (list of str, optional) – Feature names
class_names (list of str, optional) – Class names
**kwargs – Additional arguments for the explainer

Returns:

Explanation results

Return type:

Examples

>>> from vitalDSP.ml_models.explainability import explain_prediction
>>> from sklearn.ensemble import RandomForestClassifier
>>> import numpy as np
>>>
>>> # Train model
>>> X_train = np.random.randn(1000, 50)
>>> y_train = np.random.randint(0, 2, 1000)
>>> model = RandomForestClassifier()
>>> model.fit(X_train, y_train)
>>>
>>> # Explain test predictions
>>> X_test = np.random.randn(10, 50)
>>> explanation = explain_prediction(
...     model, X_test,
...     method='shap',
...     explainer_type='tree'
... )

Feature Engineering Module

ECG Autonomic Features

ECG Autonomic Features Module for Physiological Signal Processing

This module provides comprehensive ECG feature extraction capabilities focusing on autonomic nervous system analysis. It implements advanced algorithms for detecting ECG waveform components, computing intervals, and identifying arrhythmias for cardiovascular health assessment.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - P-wave analysis (duration, amplitude) - PR Interval computation (P-wave to QRS onset) - QRS Complex analysis (width, amplitude) - ST Segment analysis (elevation, depression) - QT Interval computation (QRS onset to T-wave end) - Arrhythmia detection (AFib, VTach, Bradycardia) - Waveform morphology analysis - Comprehensive ECG feature extraction

Examples:

Basic ECG feature extraction:

>>> import numpy as np
>>> from vitalDSP.feature_engineering.ecg_autonomic_features import ECGExtractor
>>> ecg_signal = np.random.rand(1000)  # Replace with actual ECG signal
>>> fs = 250  # Sampling frequency in Hz
>>> extractor = ECGExtractor(ecg_signal, fs)
>>> p_wave_duration = extractor.compute_p_wave_duration()
>>> pr_interval = extractor.compute_pr_interval()
>>> qrs_width = extractor.compute_qrs_width()
>>> print(f"P-wave Duration: {p_wave_duration}, PR Interval: {pr_interval}")

Advanced ECG analysis:

>>> qt_interval = extractor.compute_qt_interval()
>>> st_segment = extractor.compute_st_segment()
>>> arrhythmias = extractor.detect_arrhythmias()
>>> print(f"QT Interval: {qt_interval}, ST Segment: {st_segment}")
>>> print(f"Arrhythmias detected: {arrhythmias}")

Comprehensive feature extraction:

>>> all_features = extractor.extract_all_features()
>>> print(f"Extracted {len(all_features)} ECG features")

class vitalDSP.feature_engineering.ecg_autonomic_features.ECGExtractor(ecg_signal, sampling_frequency)[source]

Bases: object

A class to extract ECG features including: - P-wave analysis (duration, amplitude) - PR Interval (P-wave to QRS onset) - QRS Complex (width, amplitude) - ST Segment (elevation, depression) - QT Interval (QRS onset to T-wave end) - Detection of Arrhythmias (AFib, VTach, Bradycardia)

Example usage:

ecg_signal = np.random.rand(1000)  # Replace with actual ECG signal
fs = 250  # Sampling frequency in Hz

extractor = ECGExtractor(ecg_signal, fs)

p_wave_duration = extractor.compute_p_wave_duration()
pr_interval = extractor.compute_pr_interval()
qrs_width = extractor.compute_qrs_width()
qt_interval = extractor.compute_qt_interval()
st_segment = extractor.compute_st_segment()
arrhythmias = extractor.detect_arrhythmias()

print(f"P-wave Duration: {p_wave_duration}, PR Interval: {pr_interval}, QRS Width: {qrs_width}")

compute_p_wave_duration(r_peaks=None)[source]

Computes the P-wave duration by finding the onset and offset around each detected P-peak.

Returns:: Mean duration of P-waves in seconds.
Return type:: float

compute_pr_interval(r_peaks=None)[source]

Computes the PR interval from P-wave onset to QRS onset (Q-valley).

Returns:: Mean PR interval in seconds.
Return type:: float

compute_qrs_duration(r_peaks=None)[source]

Computes the QRS duration using WaveformMorphology.

Returns:: The mean duration of QRS complexes in seconds.
Return type:: float

compute_qt_interval()[source]

Computes the QT interval (from QRS onset to T-wave end).

Returns:: QT interval in seconds.
Return type:: float

compute_s_wave(r_peaks=None)[source]

Detects the S-wave based on the R-peaks using WaveformMorphology.

Returns:: Indices of detected S-wave points.
Return type:: np.array

compute_st_interval()[source]

Computes the ST segment duration (from S-wave to T-wave peak).

Returns:: Mean ST segment duration in seconds.
Return type:: float

detect_arrhythmias(r_peaks=None)[source]

Detects basic arrhythmias such as: - Atrial Fibrillation (AFib) - Ventricular Tachycardia (VTach) - Bradycardia (slow heart rate)

Returns:: Dictionary containing the detected arrhythmias.
Return type:: dict

detect_r_peaks()[source]

Detects R-peaks from the ECG signal using WaveformMorphology.

Returns:: Array of indices where R-peaks are detected.
Return type:: np.array

PPG Autonomic Features

Feature Engineering Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Feature extraction capabilities

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.feature_engineering.ppg_autonomic_features import PpgAutonomicFeatures
>>> signal = np.random.randn(1000)
>>> processor = PpgAutonomicFeatures(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.feature_engineering.ppg_autonomic_features.PPGAutonomicFeatures(ppg_signal, sampling_frequency)[source]

Bases: object

A class to compute respiratory and autonomic features from PPG signals.

Features included: - Respiratory Rate Variability (RRV) - Respiratory Sinus Arrhythmia (RSA) - Autonomic Nervous System Balance (Fractal Dimension, DFA)

Example usage:

import numpy as np
from vitalDSP.feature_engineering.ppg_autonomic_features import PPGAutonomicFeatures

# Simulated PPG signal data
ppg_signal = np.random.rand(1000)  # Replace with actual PPG signal
fs = 100  # Sampling frequency in Hz

features = PPGAutonomicFeatures(ppg_signal, fs)

rrv = features.compute_rrv()
rsa = features.compute_rsa()
fractal = features.compute_fractal_dimension()
dfa_value = features.compute_dfa()

print(f"RRV: {rrv}, RSA: {rsa}, Fractal Dimension: {fractal}, DFA: {dfa_value}")

compute_dfa(min_scale=4, max_scale=None, num_scales=20)[source]

Computes the Detrended Fluctuation Analysis (DFA) of the PPG signal using proper multi-scale analysis.

DFA measures the fractal scaling properties by computing the fluctuation function F(n) at multiple window sizes n, then fitting log(F) vs log(n).

Parameters:

min_scale (int) – Minimum window size (default 4).
max_scale (int) – Maximum window size (default N//4).
num_scales (int) – Number of scales to evaluate (default 20).

Returns:

DFA scaling exponent (alpha).

Return type:

compute_fractal_dimension(k_max=10)[source]

Computes the fractal dimension of the PPG signal using the Higuchi method.

Parameters:: k_max (int) – The maximum number of intervals to calculate (default is 10).
Returns:: Fractal dimension of the signal.
Return type:: float

compute_rrv()[source]

Computes Respiratory Rate Variability (RRV) from the PPG signal.

Returns:: Respiratory rate variability value.
Return type:: float

compute_rsa()[source]

Computes Respiratory Sinus Arrhythmia (RSA) from the PPG signal by bandpass filtering pulse-to-pulse intervals in the respiratory frequency band (0.15-0.4 Hz).

Returns:

RSA amplitude (std of respiratory-band pulse interval variability): in seconds.

Return type:

Morphology Features

Feature Engineering Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Configurable parameters and settings

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.feature_engineering.morphology_features import MorphologyFeatures
>>> signal = np.random.randn(1000)
>>> processor = MorphologyFeatures(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.feature_engineering.morphology_features.PhysiologicalFeatureExtractor(signal, fs=1000)[source]

Bases: object

A class to extract various physiological features from ECG and PPG signals, such as durations, areas, amplitude variability, slope ratios, and dicrotic notch locations.

Features for PPG: - Systolic/diastolic duration, area, amplitude variability - Signal skewness, slope, peak trends, and dicrotic notch locations

Features for ECG: - QRS duration, area, T-wave area - Amplitude variability, QRS-T ratios, and QRS slope - Signal skewness, peak trends

preprocess_signal(preprocess_config): Preprocess the signal by applying bandpass filtering and noise reduction.

extract_features(signal_type='ECG', preprocess_config=None)[source]: Extract all features (morphology, volume, amplitude variability, dicrotic notch) for ECG or PPG signals.

Examples

>>> import numpy as np
>>> from vitalDSP.feature_engineering.morphology_features import PhysiologicalFeatureExtractor
>>> from vitalDSP.preprocess.preprocess_operations import PreprocessConfig
>>>
>>> # Example 1: ECG feature extraction
>>> ecg_signal = np.random.randn(1000)  # Simulated ECG signal
>>> extractor = PhysiologicalFeatureExtractor(ecg_signal, fs=256)
>>> ecg_features = extractor.extract_features(signal_type="ECG")
>>> print(f"ECG features extracted: {len(ecg_features)}")
>>> print(f"QRS duration: {ecg_features.get('qrs_duration', 'N/A')}")
>>>
>>> # Example 2: PPG feature extraction with preprocessing
>>> ppg_signal = np.random.randn(2000)  # Simulated PPG signal
>>> extractor_ppg = PhysiologicalFeatureExtractor(ppg_signal, fs=128)
>>> config = PreprocessConfig(
...     filter_type="bandpass",
...     lowcut=0.5,
...     highcut=8.0,
...     noise_reduction_method="wavelet"
... )
>>> ppg_features = extractor_ppg.extract_features(signal_type="PPG", preprocess_config=config)
>>> print(f"PPG features extracted: {len(ppg_features)}")
>>> print(f"Systolic duration: {ppg_features.get('systolic_duration', 'N/A')}")
>>>
>>> # Example 3: EEG feature extraction
>>> eeg_signal = np.random.randn(1500)  # Simulated EEG signal
>>> extractor_eeg = PhysiologicalFeatureExtractor(eeg_signal, fs=512)
>>> eeg_features = extractor_eeg.extract_features(signal_type="EEG")
>>> print(f"EEG features extracted: {len(eeg_features)}")

compute_amplitude_variability(peaks)[source]

Compute the variability of the amplitudes at the given peak locations.

Parameters:: peaks (numpy.ndarray) – The set of peaks.
Returns:: variability – The amplitude variability (standard deviation of the peak amplitudes).
Return type:: float

Examples

>>> signal = np.random.randn(1000)
>>> peaks = np.array([100, 200, 300])
>>> extractor = PhysiologicalFeatureExtractor(signal)
>>> variability = extractor.compute_amplitude_variability(peaks)
>>> print(variability)

compute_peak_trend(peaks)[source]

Compute the trend slope of peak amplitudes over time.

Parameters:: peaks (numpy.ndarray) – The set of peaks.
Returns:: trend_slope – The slope of the peak amplitude trend over time.
Return type:: float

Examples

>>> signal = np.random.randn(1000)
>>> peaks = np.array([100, 200, 300])
>>> extractor = PhysiologicalFeatureExtractor(signal)
>>> trend_slope = extractor.compute_peak_trend(peaks)
>>> print(trend_slope)

detect_troughs(peaks)[source]

Detect troughs (valleys) in the signal based on the given peaks.

Parameters:: peaks (numpy.ndarray) – The indices of detected peaks.
Returns:: troughs – The indices of detected troughs.
Return type:: numpy.ndarray

Examples

>>> peaks = np.array([100, 200, 300])
>>> troughs = extractor.detect_troughs(peaks)
>>> print(troughs)

extract_features(signal_type='ECG', preprocess_config=None, peak_config=None, options=None)[source]

Extract all physiological features from the signal for either ECG or PPG.

Parameters:

signal_type (str, optional) – The type of signal. Options: ‘ECG’, ‘PPG’, ‘EEG’. Default is “ECG”.
preprocess_config (PreprocessConfig, optional) – The configuration object for signal preprocessing. If None, default settings are used.
peak_config (dict, optional) – Configuration for peak detection parameters. If None, default settings are used.

Returns:

features – A dictionary containing the extracted features, such as durations, areas, amplitude variability, slopes, skewness, peak trends, and dicrotic notch locations.

Return type:

Examples

>>> signal = np.random.randn(1000)
>>> preprocess_config = PreprocessConfig()
>>> extractor = PhysiologicalFeatureExtractor(signal, fs=1000)
>>> features = extractor.extract_features(signal_type="ECG", preprocess_config=preprocess_config)
>>> print(features)

get_preprocess_signal(preprocess_config)[source]

Preprocess the signal by applying bandpass filtering and noise reduction.

Parameters:: preprocess_config (PreprocessConfig) – Configuration for both signal filtering and artifact removal.
Returns:: clean_signal – The preprocessed signal, cleaned of noise and artifacts.
Return type:: numpy.ndarray

Examples

>>> signal = np.sin(np.linspace(0, 10, 1000)) + np.random.normal(0, 0.1, 1000)
>>> preprocess_config = PreprocessConfig()
>>> extractor = PhysiologicalFeatureExtractor(signal, fs=1000)
>>> preprocessed_signal = extractor.preprocess_signal(preprocess_config)
>>> print(preprocessed_signal)

Signal Quality Assessment Module

Signal Quality

Signal Quality Assessment Module for Physiological Signal Processing

This module provides comprehensive signal quality assessment capabilities for physiological signals including ECG, PPG, and other vital signs. It implements various quality metrics such as Signal-to-Noise Ratio (SNR), Peak Signal-to-Noise Ratio (PSNR), and Mean Square Error (MSE) to evaluate signal quality and the impact of noise or processing artifacts.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Signal-to-Noise Ratio (SNR) computation - Peak Signal-to-Noise Ratio (PSNR) calculation - Mean Square Error (MSE) assessment - Noise-based SNR estimation - Comprehensive quality metrics for signal evaluation

Examples:

Basic signal quality assessment:

>>> import numpy as np
>>> from vitalDSP.signal_quality_assessment.signal_quality import SignalQuality
>>> original_signal = np.sin(np.linspace(0, 10, 1000))
>>> noise = np.random.normal(0, 0.1, 1000)
>>> noisy_signal = original_signal + noise
>>> sq = SignalQuality(original_signal, noisy_signal)
>>> print(f"SNR: {sq.snr():.2f} dB")
>>> print(f"PSNR: {sq.psnr():.2f} dB")
>>> print(f"MSE: {sq.mse():.6f}")

Quality assessment with different noise levels:

>>> high_noise = np.random.normal(0, 0.5, 1000)
>>> very_noisy_signal = original_signal + high_noise
>>> sq_high_noise = SignalQuality(original_signal, very_noisy_signal)
>>> print(f"High noise SNR: {sq_high_noise.snr():.2f} dB")

Using noise signal directly:

>>> sq_noise = SignalQuality(original_signal)
>>> snr_from_noise = sq_noise.snr_of_noise(noise)
>>> print(f"SNR from noise: {snr_from_noise:.2f} dB")

class vitalDSP.signal_quality_assessment.signal_quality.SignalQuality(original_signal, processed_signal=None)[source]

Bases: object

A class to assess the quality of signals using various metrics.

This class provides methods to compute commonly used signal quality metrics such as Signal-to-Noise Ratio (SNR), Peak Signal-to-Noise Ratio (PSNR), and Mean Square Error (MSE). It can be used to evaluate the impact of noise or other processing on the original signal.

snr : function: Computes the Signal-to-Noise Ratio.

psnr : function: Computes the Peak Signal-to-Noise Ratio.

mse : function: Computes the Mean Square Error between the original and processed signals.

snr_of_noise : function: Computes the Signal-to-Noise Ratio given a noise signal.

Examples

>>> import numpy as np
>>> from vitalDSP.signal_quality_assessment.signal_quality import SignalQuality
>>>
>>> # Example 1: Basic signal quality assessment
>>> original_signal = np.sin(np.linspace(0, 10, 1000))
>>> noise = np.random.normal(0, 0.1, 1000)
>>> noisy_signal = original_signal + noise
>>> sq = SignalQuality(original_signal, noisy_signal)
>>> print(f"SNR: {sq.snr():.2f} dB")
>>> print(f"PSNR: {sq.psnr():.2f} dB")
>>> print(f"MSE: {sq.mse():.6f}")
>>>
>>> # Example 2: Quality assessment with different noise levels
>>> high_noise = np.random.normal(0, 0.5, 1000)
>>> very_noisy_signal = original_signal + high_noise
>>> sq_high_noise = SignalQuality(original_signal, very_noisy_signal)
>>> print(f"High noise SNR: {sq_high_noise.snr():.2f} dB")
>>>
>>> # Example 3: Using noise signal directly
>>> sq_noise = SignalQuality(original_signal)
>>> snr_from_noise = sq_noise.snr_of_noise(noise)
>>> print(f"SNR from noise: {snr_from_noise:.2f} dB")

mse()[source]

Compute the Mean Square Error (MSE) between the original and processed signals.

MSE measures the average squared difference between the original and processed signals. It is commonly used to quantify the error introduced by noise or signal processing.

Returns:: mse_value – The MSE value.
Return type:: float
Raises:: ValueError – If processed_signal is not provided during initialization.

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> noise = np.array([0.1, 0.1, 0.1, 0.1, 0.1])
>>> noisy_signal = signal + noise
>>> sq = SignalQuality(signal, noisy_signal)
>>> print(sq.mse())
0.010000000000000002

psnr()[source]

Compute the Peak Signal-to-Noise Ratio (PSNR) of the signal.

PSNR compares the maximum possible signal power to the noise power. It is commonly used in image and signal processing to assess the quality of signal reconstruction or compression techniques.

Returns:: psnr_value – The PSNR value in decibels (dB).
Return type:: float
Raises:: ValueError – If processed_signal is not provided during initialization.

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> noise = np.array([0.1, 0.1, 0.1, 0.1, 0.1])
>>> noisy_signal = signal + noise
>>> sq = SignalQuality(signal, noisy_signal)
>>> print(sq.psnr())
26.020599913279625

snr()[source]

Compute the Signal-to-Noise Ratio (SNR) of the signal.

SNR is a measure of signal quality that compares the level of the desired signal to the level of background noise. A higher SNR indicates a cleaner signal with less noise.

Returns:: snr_value – The SNR value in decibels (dB).
Return type:: float
Raises:: ValueError – If processed_signal is not provided during initialization.

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> noise = np.array([0.1, 0.1, 0.1, 0.1, 0.1])
>>> noisy_signal = signal + noise
>>> sq = SignalQuality(signal, noisy_signal)
>>> print(sq.snr())
14.154543666201898

snr_of_noise(noise_signal)[source]

Compute the Signal-to-Noise Ratio (SNR) given a noise signal.

This method calculates the SNR by comparing the power of the original signal to the power of the provided noise signal, without needing a processed signal.

Parameters:: noise_signal (numpy.ndarray) – The noise signal.
Returns:: snr_value – The SNR value in decibels (dB).
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> noise = np.array([0.1, 0.1, 0.1, 0.1, 0.1])
>>> sq = SignalQuality(signal)
>>> print(sq.snr_of_noise(noise))
20.0

Signal Quality Index

Signal Quality Assessment Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Signal validation and error handling

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.signal_quality_assessment.signal_quality_index import SignalQualityIndex
>>> signal = np.random.randn(1000)
>>> processor = SignalQualityIndex(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.signal_quality_assessment.signal_quality_index.SignalQualityIndex(signal)[source]

Bases: object

A class to compute various Signal Quality Index (SQI) metrics for assessing the quality of vital signals.

This class includes methods to evaluate signal quality based on different characteristics such as amplitude variability, baseline wander, zero-crossing consistency, waveform similarity, entropy, and more. These metrics are useful in ensuring that physiological signals like ECG, PPG, EEG, and respiratory signals are of high quality and reliable for further analysis.

amplitude_variability_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the amplitude variability SQI over segments of the signal.

Parameters:

window_size (int) – The size of each segment to compute the SQI.
step_size (int) – The step size to move the window for each segment.
threshold (float or tuple, optional) – The threshold to determine normal and abnormal segments.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.
aggregate (bool, optional) – If True (default), returns the mean SQI value as a single float. If False, returns detailed tuple (sqi_values, normal_segments, abnormal_segments).

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments).

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 1, 2, 3, 4, 5])
>>> sqi = SignalQualityIndex(signal)
>>> # Get single aggregated score (default)
>>> quality_score = sqi.amplitude_variability_sqi(window_size=5, step_size=2)
>>> # Get detailed results
>>> sqi_values, normal_segments, abnormal_segments = sqi.amplitude_variability_sqi(
...     window_size=5, step_size=2, threshold=0.8, aggregate=False)

baseline_wander_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', moving_avg_window=50, aggregate=True)[source]

Compute the baseline wander SQI.

This metric evaluates the amount of baseline wander in the signal, which is unwanted low-frequency noise that can distort the true signal. Baseline wander is particularly important in ECG and PPG signals.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.
moving_avg_window (int, optional) – Size of the window for calculating the moving average. Default is 50.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 1.1, 1.2, 1.3, 1.4, 1.5, 1.6])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.baseline_wander_sqi(window_size=3, step_size=1, threshold=0.8)

eeg_band_power_sqi(band_power, window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the EEG band power SQI.

This metric assesses the consistency of EEG band power, which is important for ensuring signal quality. Stable band power indicates a high-quality EEG signal.

Parameters:

band_power (numpy.ndarray) – Array of band power values for EEG.
window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> band_power = np.array([10, 12, 11, 13, 12])
>>> sqi = SignalQualityIndex(band_power)
>>> sqi_values, normal_segments, abnormal_segments = sqi.eeg_band_power_sqi(band_power, window_size=3, step_size=1, threshold=0.8)

energy_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the energy SQI.

This metric computes the energy of the signal over a segment, which can be an indicator of the signal’s strength or activity.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 2, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.energy_sqi(window_size=3, step_size=1, threshold=0.8)

heart_rate_variability_sqi(rr_intervals, window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the heart rate variability (HRV) SQI.

This metric assesses the variability in RR intervals of ECG, which should be within a certain range for a healthy signal. Abnormal HRV can indicate issues with heart health or signal quality.

Parameters:

rr_intervals (numpy.ndarray) – Array of RR intervals.
window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> rr_intervals = np.array([0.8, 0.9, 1.0, 0.9, 0.8])
>>> sqi = SignalQualityIndex(rr_intervals)
>>> sqi_values, normal_segments, abnormal_segments = sqi.heart_rate_variability_sqi(rr_intervals, window_size=3, step_size=1, threshold=0.8)

kurtosis_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the kurtosis SQI.

This metric measures the “tailedness” of the signal distribution. High kurtosis values indicate the presence of outliers or sharp peaks in the signal.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 2, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.kurtosis_sqi(window_size=3, step_size=1, threshold=0.8)

peak_to_peak_amplitude_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the peak-to-peak amplitude SQI.

This metric assesses the peak-to-peak amplitude of the signal, which is the difference between the maximum and minimum values within a segment.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 2, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.peak_to_peak_amplitude_sqi(window_size=3, step_size=1, threshold=0.8)

ppg_signal_quality_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the PPG signal quality SQI.

This metric evaluates the overall quality of PPG signals based on amplitude variability and baseline wander. High-quality PPG signals should have minimal noise and stable amplitude.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 1.1, 0.9, 1, 1.2, 0.8, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.ppg_signal_quality_sqi(window_size=3, step_size=1, threshold=0.8)

respiratory_signal_quality_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the respiratory signal quality SQI.

This metric evaluates the quality of respiratory signals, considering factors like consistency in breathing cycles and stability of the amplitude.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> resp_signal = np.array([1, 1.1, 1.0, 1.1, 1.0])
>>> sqi = SignalQualityIndex(resp_signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.respiratory_signal_quality_sqi(window_size=3, step_size=1, threshold=0.8)

signal_entropy_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the signal entropy SQI.

This metric measures the entropy of the signal, which indicates the complexity or predictability of the signal. Lower entropy generally indicates a more regular and stable signal, while higher entropy suggests more randomness.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 1, 1, 1, 1, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.signal_entropy_sqi(window_size=3, step_size=1, threshold=0.8)

skewness_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the skewness SQI.

This metric assesses the asymmetry of the signal distribution, where a high skewness value indicates that the signal is not symmetrically distributed.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 2, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.skewness_sqi(window_size=3, step_size=1, threshold=0.8)

snr_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the Signal-to-Noise Ratio (SNR) SQI.

This metric measures the ratio of the signal power to the noise power, which is an important metric for evaluating signal quality.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 2, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.snr_sqi(window_size=3, step_size=1, threshold=0.8)

waveform_similarity_sqi(window_size, step_size, reference_waveform, threshold=None, threshold_type='below', scale='zscore', similarity_method='correlation', aggregate=True)[source]

Compute the waveform similarity SQI using various similarity metrics.

This metric compares the similarity between consecutive waveforms in the signal. High similarity indicates that the signal is stable and free from significant artifacts.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
reference_waveform (numpy.ndarray) – A reference waveform to compare against.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.
similarity_method (str, optional) – The method to use for similarity calculation (‘correlation’, ‘custom’). Default is ‘correlation’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 2, 1])
>>> reference = np.array([1, 2, 3, 2, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.waveform_similarity_sqi(
...     window_size=3, step_size=1, reference_waveform=reference, threshold=0.8)

zero_crossing_sqi(window_size, step_size, threshold=None, threshold_type='below', scale='zscore', aggregate=True)[source]

Compute the zero-crossing SQI.

This metric assesses the number of zero crossings in the signal, which should be consistent in high-quality signals. Irregular zero crossings can indicate noise or instability in the signal.

Parameters:

window_size (int) – The size of each segment.
step_size (int) – The step size to slide the window.
threshold (float or tuple, optional) – Threshold to classify a segment as normal or abnormal.
threshold_type (str, optional) – The type of thresholding (‘below’, ‘above’, ‘range’). Default is ‘below’.
scale (str, optional) – The scaling method to apply to SQI values (‘zscore’, ‘iqr’, ‘minmax’). Default is ‘zscore’.

Returns:

If aggregate=True: Mean SQI value as float. If aggregate=False: Tuple of (sqi_values, normal_segments, abnormal_segments) where sqi_values is array of SQI values for each segment, normal_segments and abnormal_segments are lists of (start, end) indices.

Return type:

Examples

>>> signal = np.array([-1, 1, -1, 1, -1, 1])
>>> sqi = SignalQualityIndex(signal)
>>> sqi_values, normal_segments, abnormal_segments = sqi.zero_crossing_sqi(window_size=3, step_size=1, threshold=0.8)

SNR Computation

Signal Quality Assessment Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple processing methods and functions - NumPy integration for numerical computations

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.signal_quality_assessment.snr_computation import SnrComputation
>>> signal = np.random.randn(1000)
>>> processor = SnrComputation(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

vitalDSP.signal_quality_assessment.snr_computation.crest_factor(signal)[source]

Calculate the crest factor of the signal, which is the ratio of the peak amplitude to the RMS value.

Parameters:: signal (numpy.ndarray) – The input signal.
Returns:: crest_factor – The crest factor of the signal.
Return type:: float

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> cf = crest_factor(signal)
>>> print(cf)

vitalDSP.signal_quality_assessment.snr_computation.harmonic_distortion(signal, fundamental_freq, sampling_rate, harmonics=5)[source]

Calculate the total harmonic distortion (THD) of the signal.

Parameters:

signal (numpy.ndarray) – The input signal.
fundamental_freq (float) – The fundamental frequency of the signal.
sampling_rate (float) – The sampling rate of the signal.
harmonics (int, optional (default=5)) – The number of harmonics to include in the THD calculation.

Returns:

thd – The total harmonic distortion as a percentage.

Return type:

Examples

>>> signal = np.sin(2 * np.pi * 50 * np.arange(0, 1, 1/1000))
>>> thd = harmonic_distortion(signal, fundamental_freq=50, sampling_rate=1000)
>>> print(thd)

vitalDSP.signal_quality_assessment.snr_computation.signal_to_noise_and_distortion_ratio(signal, noise)[source]

Calculate the Signal-to-Noise and Distortion Ratio (SINAD).

Parameters:

signal (numpy.ndarray) – The original clean signal.
noise (numpy.ndarray) – The noise and distortion components of the signal.

Returns:

sinad – The SINAD in dB.

Return type:

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> noise = 0.1 * np.random.normal(size=signal.shape)
>>> sinad = signal_to_noise_and_distortion_ratio(signal, noise)
>>> print(sinad)

vitalDSP.signal_quality_assessment.snr_computation.signal_to_noise_and_interference_ratio(signal, interference)[source]

Calculate the Signal-to-Noise and Interference Ratio (SNIR).

Parameters:

signal (numpy.ndarray) – The original clean signal.
interference (numpy.ndarray) – The interference in the signal.

Returns:

snir – The SNIR in dB.

Return type:

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> interference = 0.1 * np.random.normal(size=signal.shape)
>>> snir = signal_to_noise_and_interference_ratio(signal, interference)
>>> print(snir)

vitalDSP.signal_quality_assessment.snr_computation.snr_mean_square(signal, noise)[source]

Calculate the SNR using the mean square of the signal and noise.

Parameters:

signal (numpy.ndarray) – The original clean signal.
noise (numpy.ndarray) – The noise signal.

Returns:

snr – The SNR in dB.

Return type:

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> noise = 0.1 * np.random.normal(size=signal.shape)
>>> snr = snr_mean_square(signal, noise)
>>> print(snr)

vitalDSP.signal_quality_assessment.snr_computation.snr_peak_to_peak(signal, noise)[source]

Calculate the SNR using the peak-to-peak amplitude method.

Parameters:

signal (numpy.ndarray) – The original clean signal.
noise (numpy.ndarray) – The noise signal.

Returns:

snr – The SNR in dB.

Return type:

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> noise = 0.1 * np.random.normal(size=signal.shape)
>>> snr = snr_peak_to_peak(signal, noise)
>>> print(snr)

vitalDSP.signal_quality_assessment.snr_computation.snr_power_ratio(signal, noise)[source]

Calculate the SNR using the power ratio method.

Parameters:

signal (numpy.ndarray) – The original clean signal.
noise (numpy.ndarray) – The noise signal.

Returns:

snr – The SNR in dB.

Return type:

Examples

>>> signal = np.sin(2 * np.pi * 0.2 * np.arange(0, 10, 0.01))
>>> noise = 0.1 * np.random.normal(size=signal.shape)
>>> snr = snr_power_ratio(signal, noise)
>>> print(snr)

Utils Module

Peak Detection

Peak Detection Module for Physiological Signal Processing

This module provides comprehensive peak detection capabilities for various physiological signals including ECG, PPG, EEG, respiratory, and arterial blood pressure (ABP) signals. It implements multiple detection algorithms optimized for different signal types and characteristics.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple peak detection algorithms - Signal-specific optimizations (ECG R-peaks, PPG systolic peaks) - Advanced filtering and preprocessing integration - Scalable threshold-based detection - Derivative-based detection methods - Wavelet and bandpass filtering for EEG

Examples:

Basic peak detection:

>>> import numpy as np
>>> from vitalDSP.utils.signal_processing.peak_detection import PeakDetection
>>> signal = np.sin(np.linspace(0, 10, 1000)) + np.random.normal(0, 0.1, 1000)
>>> detector = PeakDetection(signal, method="threshold", height=0.5)
>>> peaks = detector.detect_peaks()
>>> print(f"Detected {len(peaks)} peaks")

ECG R-peak detection:

>>> ecg_detector = PeakDetection(ecg_signal, method="ecg_r_peak")
>>> r_peaks = ecg_detector.detect_peaks()
>>> print(f"R-peaks: {len(r_peaks)}")

PPG systolic peak detection:

>>> ppg_detector = PeakDetection(ppg_signal, method="ppg_systolic_peaks")
>>> systolic_peaks = ppg_detector.detect_peaks()
>>> print(f"Systolic peaks: {len(systolic_peaks)}")

class vitalDSP.utils.signal_processing.peak_detection.PeakDetection(signal, method='threshold', **kwargs)[source]

Bases: object

A class to detect peaks in various physiological signals like ECG, PPG, EEG, respiratory, and arterial blood pressure (ABP) signals.

detect_peaks : function: Detects the peaks in the signal using the selected method.

detect_peaks()[source]

Detect peaks in the signal based on the selected method.

Returns:: peaks – The indices of the detected peaks.
Return type:: numpy.ndarray
Raises:: ValueError – If an invalid method is selected.

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
>>> detector = PeakDetection(signal, method="threshold", threshold=4)
>>> peaks = detector.detect_peaks()
>>> print(peaks)

Data Synthesis

Utility Functions Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Interactive visualization capabilities

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.data_processing.synthesize_data import SynthesizeData
>>> signal = np.random.randn(1000)
>>> processor = SynthesizeData(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.utils.data_processing.synthesize_data.SynthesizeData[source]

Bases: object

A class that provides access to all signal synthesis functions. This class wraps the individual functions for easier access and organization.

static generate_ecg_signal(duration=10, sampling_rate=1000, heart_rate=60, noise_level=0.01, qrs_width=0.08, t_wave_amplitude=0.3, p_wave_amplitude=0.25, randomize=False, display=False)[source]: Generate a synthetic ECG signal.

static generate_noisy_signal(base_signal, noise_level=0.1)[source]: Add Gaussian noise to a base signal.

static generate_ppg_data(duration=50, fs=1000)[source]: Generate synthetic PPG data for testing.

static generate_resp_signal(sampling_rate, duration, frequency=0.2, amplitude=0.5)[source]: Generate a synthetic respiratory signal.

static generate_sinusoidal(frequency, sampling_rate, duration, amplitude=1.0, phase=0.0)[source]: Generate a sinusoidal wave.

static generate_square_wave(frequency, sampling_rate, duration, amplitude=1.0, duty_cycle=0.5)[source]: Generate a square wave.

static generate_synthetic_ppg(duration=10, sampling_rate=1000, heart_rate=60, noise_level=0.01, diastolic_amplitude=0.8, diastolic_width=0.15, dicrotic_notch_depth=0.6, dicrotic_notch_delay=0.2, randomize=False, display=False)[source]: Generate a synthetic PPG signal.

static generate_synthetic_ppg_reversed(duration=10, sampling_rate=1000, heart_rate=60, noise_level=0.01, diastolic_amplitude=0.8, diastolic_width=0.15, dicrotic_notch_depth=0.6, dicrotic_notch_delay=0.2, randomize=False, display=False)[source]: Generate a synthetic PPG signal with reversed parameters.

vitalDSP.utils.data_processing.synthesize_data.generate_ecg_signal(sfecg=256, N=None, duration=30, Anoise=0.0, hrmean=60, hrstd=1.0, lfhfratio=0.5, sfint=512, ti=None, ai=None, bi=None)[source]

Generate a synthetic ECG signal using the dynamical model from McSharry et al.

Parameters:

sfecg (int, optional) – ECG sampling frequency in Hz (default is 256).
N (int, optional) – Approximate number of heartbeats (default is 60).
duration (int, optional) – Approximate number of seconds (default is 30).
Anoise (float, optional) – Amplitude of additive uniformly distributed measurement noise (default is 0).
hrmean (float, optional) – Mean heart rate in beats per minute (default is 60).
hrstd (float, optional) – Standard deviation of heart rate in beats per minute (default is 1).
lfhfratio (float, optional) – Low frequency to high frequency ratio (default is 0.5).
sfint (int, optional) – Internal sampling frequency in Hz (default is 512).
ti (list, optional) – Angles of extrema in degrees for PQRST (default is [-70, -15, 0, 15, 100]).
ai (list, optional) – Z-position of extrema for PQRST (default is [1.2, -5, 30, -7.5, 0.75]).
bi (list, optional) – Gaussian width of peaks for PQRST (default is [0.25, 0.1, 0.1, 0.1, 0.4]).

Returns:

s (ndarray) – Generated ECG signal.
ipeaks (ndarray) – Labels for PQRST peaks.

Examples

>>> signal = generate_ecg_signal(sfecg=256, N=256, Anoise=0.01, hrmean=70, sfint=512)
>>> print(signal)

vitalDSP.utils.data_processing.synthesize_data.generate_noisy_signal(base_signal, noise_level=0.1)[source]

Add Gaussian noise to a base signal.

Parameters:

base_signal (numpy.ndarray) – The input signal to which noise will be added.
noise_level (float, optional (default=0.1)) – The standard deviation of the Gaussian noise.

Returns:

noisy_signal – The signal with added Gaussian noise.

Return type:

Examples

>>> base_signal = generate_sinusoidal(frequency=1.0, sampling_rate=100.0, duration=5.0)
>>> noisy_signal = generate_noisy_signal(base_signal, noise_level=0.2)
>>> print(noisy_signal)

vitalDSP.utils.data_processing.synthesize_data.generate_resp_signal(sampling_rate, duration, frequency=0.2, amplitude=0.5)[source]

Generate a synthetic respiratory signal.

Parameters:

sampling_rate (float) – Sampling rate in Hz.
duration (float) – Duration of the signal in seconds.
frequency (float, optional (default=0.2)) – Frequency of the respiratory signal in Hz.
amplitude (float, optional (default=0.5)) – Amplitude of the respiratory signal.

Returns:

resp_signal – The generated synthetic respiratory signal.

Return type:

Examples

>>> resp_signal = generate_resp_signal(sampling_rate=1000.0, duration=10.0)
>>> print(resp_signal)

vitalDSP.utils.data_processing.synthesize_data.generate_sinusoidal(frequency, sampling_rate, duration, amplitude=1.0, phase=0.0)[source]

Generate a sinusoidal wave.

Parameters:

frequency (float) – Frequency of the sinusoid in Hz.
sampling_rate (float) – Sampling rate in Hz.
duration (float) – Duration of the signal in seconds.
amplitude (float, optional (default=1.0)) – Amplitude of the sinusoid.
phase (float, optional (default=0.0)) – Phase shift of the sinusoid in radians.

Returns:

signal – The generated sinusoidal signal.

Return type:

Examples

>>> signal = generate_sinusoidal(frequency=1.0, sampling_rate=100.0, duration=5.0)
>>> print(signal)

vitalDSP.utils.data_processing.synthesize_data.generate_square_wave(frequency, sampling_rate, duration, amplitude=1.0, duty_cycle=0.5)[source]

Generate a square wave.

Parameters:

frequency (float) – Frequency of the square wave in Hz.
sampling_rate (float) – Sampling rate in Hz.
duration (float) – Duration of the signal in seconds.
amplitude (float, optional (default=1.0)) – Amplitude of the square wave.
duty_cycle (float, optional (default=0.5)) – Duty cycle of the square wave (fraction of one period in which the signal is high).

Returns:

signal – The generated square wave signal.

Return type:

Examples

>>> signal = generate_square_wave(frequency=1.0, sampling_rate=100.0, duration=5.0)
>>> print(signal)

vitalDSP.utils.data_processing.synthesize_data.generate_synthetic_ppg(duration=10, sampling_rate=1000, heart_rate=60, noise_level=0.01, diastolic_amplitude=0.8, diastolic_width=0.15, dicrotic_notch_depth=0.6, dicrotic_notch_delay=0.2, randomize=False, display=False)[source]

Generate a synthetic PPG signal with adjustable parameters.

Parameters:

duration (float, optional) – Duration of the signal in seconds (default is 10).
sampling_rate (int, optional) – Sampling rate in Hz (default is 1000).
heart_rate (float, optional) – Heart rate in beats per minute (default is 60).
noise_level (float, optional) – Standard deviation of the Gaussian noise to be added (default is 0.01).
diastolic_amplitude (float, optional) – Amplitude of the diastolic peak relative to the systolic peak (default is 0.8).
diastolic_width (float, optional) – Width of the diastolic peak (default is 0.15).
dicrotic_notch_depth (float, optional) – Depth of the dicrotic notch (default is 0.6).
dicrotic_notch_delay (float, optional) – Time delay for the diastolic peak (default is 0.2).
randomize (bool, optional) – If True, introduces randomness in the diastolic amplitude, width, and dicrotic notch depth (default is False).
display (bool, optional) – If True, displays the generated PPG signal (default is False).

Returns:

time (numpy.ndarray) – Array of time values.
ppg_signal (numpy.ndarray) – The generated synthetic PPG signal.

Example

>>> time, ppg_signal = generate_synthetic_ppg(duration=10, heart_rate=60, randomize=True, display=True)

vitalDSP.utils.data_processing.synthesize_data.generate_synthetic_ppg_reversed(duration=10, sampling_rate=1000, heart_rate=60, noise_level=0.01, display=False)[source]

Generate a synthetic PPG signal.

Parameters:

duration (float, optional) – Duration of the signal in seconds (default is 10).
sampling_rate (int, optional) – Sampling rate in Hz (default is 1000).
heart_rate (float, optional) – Heart rate in beats per minute (default is 60).
noise_level (float, optional) – Standard deviation of the Gaussian noise to be added (default is 0.01).
display (bool, optional) – If True, displays the generated PPG signal (default is False).

Returns:

time (numpy.ndarray) – Array of time values.
ppg_signal (numpy.ndarray) – The generated synthetic PPG signal.

Example

>>> time, ppg_signal = generate_synthetic_ppg(duration=10, heart_rate=60, display=True)

vitalDSP.utils.data_processing.synthesize_data.ordinary_differential_equation(t, x_equations, rr=None, sfint=None, ti=None, ai=None, bi=None)[source]

vitalDSP.utils.data_processing.synthesize_data.rrprocess(flo, fhi, flostd, fhistd, lfhfratio, hrmean, hrstd, sfrr, n)[source]

Standard Scaler

Utility Functions Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.signal_processing.scaler import Scaler
>>> signal = np.random.randn(1000)
>>> processor = Scaler(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.utils.signal_processing.scaler.StandardScaler[source]

Bases: object

A custom implementation of the Standard Scaler, which standardizes signals by removing the mean and scaling to unit variance.

This class can be used to normalize signals before applying transformations like the Discrete Wavelet Transform (DWT).

mean_

The mean of the signal after fitting.

Type:: float or None

std_

The standard deviation of the signal after fitting.

Type:: float or None

fit(signal)[source]: Calculate mean and standard deviation for scaling.

transform(signal)[source]: Scale the signal using the mean and standard deviation.

fit_transform(signal)[source]: Fit to the data and then transform it.

fit(signal)[source]

Fit the scaler to the signal by calculating the mean and standard deviation.

Parameters:: signal (numpy.ndarray) – The input signal to be standardized.
Return type:: None

fit_transform(signal)[source]

Fit to the signal, then transform it.

This is a convenience method that combines fit and transform into a single step.

Parameters:: signal (numpy.ndarray) – The input signal to be fitted and transformed.
Returns:: scaled_signal – The standardized signal with zero mean and unit variance.
Return type:: numpy.ndarray

Examples

>>> scaler = StandardScaler()
>>> scaled_signal = scaler.fit_transform(np.array([1, 2, 3, 4, 5]))
>>> print(scaled_signal)

transform(signal)[source]

Transform the signal using the fitted scaler by removing the mean and scaling to unit variance.

Parameters:: signal (numpy.ndarray) – The input signal to be transformed.
Returns:: scaled_signal – The standardized signal with zero mean and unit variance.
Return type:: numpy.ndarray
Raises:: ValueError – If the scaler has not been fitted yet.

Examples

>>> scaler = StandardScaler()
>>> scaler.fit(np.array([1, 2, 3, 4, 5]))
>>> scaled_signal = scaler.transform(np.array([1, 2, 3, 4, 5]))
>>> print(scaled_signal)

Normalization

Utility Functions Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple processing methods and functions - NumPy integration for numerical computations

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.signal_processing.normalization import Normalization
>>> signal = np.random.randn(1000)
>>> processor = Normalization(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

vitalDSP.utils.signal_processing.normalization.min_max_normalization(signal, min_value=0, max_value=1)[source]

Normalize the signal using Min-Max scaling.

Parameters:

signal (numpy.ndarray) – The input signal to be normalized.
min_value (float, optional (default=0)) – The minimum value of the desired range.
max_value (float, optional (default=1)) – The maximum value of the desired range.

Returns:

normalized_signal – The Min-Max normalized signal.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> normalized_signal = min_max_normalization(signal, min_value=-1, max_value=1)
>>> print(normalized_signal)
[-1.  -0.5  0.   0.5  1. ]

vitalDSP.utils.signal_processing.normalization.z_score_normalization(signal)[source]

Normalize the signal using Z-score normalization.

Parameters:: signal (numpy.ndarray) – The input signal to be normalized.
Returns:: normalized_signal – The Z-score normalized signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> normalized_signal = z_score_normalization(signal)
>>> print(normalized_signal)
[-1.26491106 -0.63245553  0.          0.63245553  1.26491106]

Interpolations

Utility Functions Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.signal_processing.interpolations import Interpolations
>>> signal = np.random.randn(1000)
>>> processor = Interpolations(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

vitalDSP.utils.signal_processing.interpolations.backward_fill(intervals)[source]

Impute missing values by carrying backward the next valid RR interval.

Parameters:: intervals (np.array) – The array of RR intervals with NaN values.
Returns:: The array with missing values imputed using backward fill.
Return type:: np.array

Example

>>> backward_fill(np.array([0.8, np.nan, 0.82, np.nan, 0.85]))

vitalDSP.utils.signal_processing.interpolations.forward_fill(intervals)[source]

Impute missing values by carrying forward the last valid RR interval.

Parameters:: intervals (np.array) – The array of RR intervals with NaN values.
Returns:: The array with missing values imputed using forward fill.
Return type:: np.array

Example

>>> forward_fill(np.array([0.8, np.nan, 0.82, np.nan, 0.85]))

vitalDSP.utils.signal_processing.interpolations.linear_interpolation(intervals)[source]

Impute missing values using linear interpolation.

Parameters:: intervals (np.array) – The array of RR intervals with NaN values.
Returns:: The array with missing values imputed using linear interpolation.
Return type:: np.array

Example

>>> linear_interpolation(np.array([0.8, np.nan, 0.82, np.nan, 0.85]))

vitalDSP.utils.signal_processing.interpolations.mean_imputation(intervals)[source]

Impute missing values by replacing them with the mean of the valid RR intervals.

Parameters:: intervals (np.array) – The array of RR intervals with NaN values.
Returns:: The array with missing values imputed using the mean of valid intervals.
Return type:: np.array

Example

>>> mean_imputation(np.array([0.8, np.nan, 0.82, np.nan, 0.85]))

vitalDSP.utils.signal_processing.interpolations.median_imputation(intervals)[source]

Impute missing values by replacing them with the median of the valid RR intervals.

Parameters:: intervals (np.array) – The array of RR intervals with NaN values.
Returns:: The array with missing values imputed using the median of valid intervals.
Return type:: np.array

Example

>>> median_imputation(np.array([0.8, np.nan, 0.82, np.nan, 0.85]))

vitalDSP.utils.signal_processing.interpolations.rolling_mean_imputation(intervals, window=5)[source]

Impute missing values using the rolling mean of valid RR intervals.

Parameters:

intervals (np.array) – The array of RR intervals with NaN values.
window (int, optional) – The window size for the rolling mean. Defaults to 5.

Returns:

The array with missing values imputed using the rolling mean.

Return type:

np.array

Example

>>> rolling_mean_imputation(np.array([0.8, np.nan, 0.82, np.nan, 0.85]), window=3)

vitalDSP.utils.signal_processing.interpolations.spline_interpolation(intervals, order=3)[source]

Impute missing values using spline interpolation.

Parameters:

intervals (np.array) – The array of RR intervals with NaN values.
order (int, optional) – The order of the spline, defaults to 3 (cubic spline).

Returns:

The array with missing values imputed using spline interpolation.

Return type:

np.array

Example

>>> spline_interpolation(np.array([0.8, np.nan, 0.82, np.nan, 0.85]), order=3)

Health Analysis Module

Health Report Generator

Health Report Generator Module for Physiological Signal Analysis

This module provides comprehensive health report generation capabilities for physiological signal analysis including ECG, PPG, EEG, and other vital signs. It implements automated report creation with visualization, interpretation, and multi-threaded processing for efficient large-scale analysis.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Automated health report generation - Multi-threaded visualization processing - Comprehensive feature analysis and interpretation - HTML report templates and rendering - Performance monitoring and optimization - Parallel processing for large datasets

Examples:

Basic health report generation:

>>> from vitalDSP.health_analysis.health_report_generator import HealthReportGenerator
>>> feature_data = {'sdnn': [45.2, 48.1, 42.3], 'rmssd': [32.1, 35.4, 30.8]}
>>> generator = HealthReportGenerator(feature_data=feature_data, segment_duration='1_min')
>>> report_html = generator.generate(output_dir='./reports')

Multi-threaded processing:

>>> generator = HealthReportGenerator(feature_data=feature_data, segment_duration='1_min', max_workers=4)
>>> generator.set_concurrency(max_workers=4)
>>> report_html = generator.generate()

class vitalDSP.health_analysis.health_report_generator.HealthReportGenerator(feature_data, segment_duration='1_min', feature_config_path=None, max_workers=None)[source]

Bases: object

A class to generate a health report based on feature data, including interpretations, visualizations, and contradictions/correlations.

This class handles the process of interpreting feature data (e.g., heart rate variability, ECG/PPG data), generating visualizations for features, and rendering the final HTML report. It uses fully concurrent processing: - Feature interpretation: Concurrent processing using ThreadPoolExecutor (CPU-bound, thread-safe) - Visualization generation: Concurrent processing using ProcessPoolExecutor (matplotlib process-safe)

feature_data

Dictionary containing feature names as keys and their corresponding values.

Type:: dict

segment_duration

Duration of the segment, either “1_min” or “5_min”. Default is “1_min”.

Type:: str

interpreter

Instance of InterpretationEngine to interpret feature data.

Type:: InterpretationEngine

visualizer

Instance of HealthReportVisualizer to create visualizations.

Type:: HealthReportVisualizer

max_workers

Maximum number of concurrent workers for processing. Default is CPU count.

Type:: int

Examples

>>> from vitalDSP.health_analysis.health_report_generator import HealthReportGenerator
>>>
>>> # Example 1: Basic health report generation
>>> feature_data = {"nn50": 45, "rmssd": 70, "sdnn": 120}
>>> generator = HealthReportGenerator(feature_data, segment_duration="5_min")
>>> report_html = generator.generate()
>>> print(f"Report generated: {len(report_html)} characters")
>>>
>>> # Example 2: Health report with custom configuration
>>> generator_custom = HealthReportGenerator(
...     feature_data,
...     segment_duration="1_min",
...     feature_config_path="path/to/custom_config.yml"
... )
>>> report_custom = generator_custom.generate()
>>>
>>> # Example 3: Health report with filtering
>>> report_filtered = generator.generate(filter_status="above_range")
>>> print("Filtered report for above-range parameters only")

static batch_visualization(visualizer, feature_data, output_dir, processes=4)[source]

Generate visualizations in parallel using ProcessPoolExecutor to avoid matplotlib threading issues. :param visualizer: The visualizer instance. :type visualizer: HealthReportVisualizer :param feature_data: Dictionary of features and their values. :type feature_data: dict :param output_dir: Directory to save the visualizations. :type output_dir: str :param processes: Number of processes for parallel visualization. :type processes: int

Returns:: Paths to the generated visualizations.
Return type:: dict

static downsample(values, factor=10)[source]

Downsample large datasets by selecting every nth value. :param values: The dataset to downsample. :type values: list or np.array :param factor: The downsampling factor. :type factor: int

Returns:: The downsampled dataset.
Return type:: np.array

generate(filter_status='all', output_dir=None)[source]

Generates the complete health report by interpreting the features, generating visualizations, and rendering an HTML report.

The report will include the description of each feature, its interpretation (in-range, above range, or below range), any detected contradictions, correlations, and visualizations for each feature.

Returns:: HTML content of the generated health report.
Return type:: str

Example Usage:

>>> feature_data = {"nn50": 45, "rmssd": 70, "sdnn": 120}
>>> generator = HealthReportGenerator(feature_data, segment_duration="5_min")
>>> report_html = generator.generate()
>>> with open('health_report.html', 'w') as file:
>>>     file.write(report_html)

get_performance_info()[source]

Get information about system performance and recommended settings.

Returns:: Performance information including CPU count, recommended workers, etc.
Return type:: dict

set_concurrency(max_workers=None)[source]

Configure concurrency settings for report generation.

Parameters:: max_workers (int, optional) – Maximum number of concurrent workers. If None, uses CPU count.

Example

>>> generator.set_concurrency(max_workers=4)
>>> generator.set_concurrency()  # Reset to CPU count

vitalDSP.health_analysis.health_report_generator.process_single_feature_visualization(feature_item, config, segment_duration, output_dir)[source]: Process a single feature for visualization in a separate process. This function must be at module level for multiprocessing to work.

Health Report Visualization

Health Analysis Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - SciPy integration for advanced signal processing - Interactive visualization capabilities

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.health_analysis.health_report_visualization import HealthReportVisualization
>>> signal = np.random.randn(1000)
>>> hrv = HealthReportVisualizer(config={'sdnn': {'normal_range': {'1_min': [30,70]}}})
>>> result = hrv.create_visualizations({'sdnn': [35,40,45]}, output_dir='./out')
>>> print(f'Processing result: {result}')

class vitalDSP.health_analysis.health_report_visualization.HealthReportVisualizer(config, segment_duration='1_min')[source]

Bases: object

A class responsible for creating visualizations of health feature data, including line plots and heatmaps.

The class takes feature data and creates visualizations, such as normal distributions for ranges and heatmaps, and stores them as images.

create_summary_chart(segment_values, output_dir='visualizations')[source]

Creates a single horizontal bar chart showing ALL features color-coded by range status. Called once per report, not per feature.

Parameters:

segment_values (dict) – {feature: list_of_values_per_segment}
output_dir (str) – Directory where the chart will be saved.

Returns:

Path to the saved summary chart image.

Return type:

str

create_visualizations(feature_data, output_dir='visualizations')[source]

Creates visualizations for the provided feature data and saves them as image files.

Parameters:

feature_data (dict) – Dictionary containing feature values.
output_dir (str) – Directory where the visualizations will be saved.

Returns:

Dictionary with feature names as keys and paths to the saved visualizations as values.

Return type:

Example Usage:

>>> visualizations = visualization.create_visualizations(feature_data)
>>> print(visualizations)

Interpretation Engine

Health Analysis Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Object-oriented design with comprehensive classes - Multiple processing methods and functions - NumPy integration for numerical computations - Configurable parameters and settings

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.health_analysis.interpretation_engine import InterpretationEngine
>>> signal = np.random.randn(1000)
>>> ie = InterpretationEngine()
>>> status = ie.get_range_status('sdnn', 45.0)
>>> interp = ie.interpret_feature('sdnn', 45.0)
>>> print(f'Processing result: {result}')

class vitalDSP.health_analysis.interpretation_engine.InterpretationEngine(yaml_file=None)[source]

Bases: object

A class responsible for interpreting feature values based on a YAML configuration file.

This class loads a YAML file that contains the normal ranges, interpretation, contradictions, and correlations for different features. It then provides methods to interpret feature values, check for contradictions, and identify correlations.

config

The loaded YAML configuration data for feature interpretation.

Type:: dict

Examples

>>> from vitalDSP.health_analysis.interpretation_engine import InterpretationEngine
>>>
>>> # Example 1: Using default configuration
>>> engine = InterpretationEngine()
>>> result = engine.interpret_feature("sdnn", 45.0, "1_min")
>>> print(f"SDNN interpretation: {result['interpretation']}")
>>>
>>> # Example 2: Using custom configuration file
>>> engine_custom = InterpretationEngine("path/to/custom_config.yml")
>>> result_custom = engine_custom.interpret_feature("rmssd", 25.0, "5_min")
>>> print(f"RMSSD range status: {engine_custom.get_range_status('rmssd', 25.0, '5_min')}")
>>>
>>> # Example 3: Multiple feature interpretation
>>> feature_data = {"sdnn": 45.0, "rmssd": 25.0, "nn50": 15}
>>> multi_result = engine.interpret_multiple_features(feature_data, "1_min")
>>> print(f"Overall assessment: {multi_result['overall_assessment']}")

get_range_status(feature_name, value, segment_duration='1 min')[source]

Determines whether the given feature value is within, above, or below the normal range.

Parameters:

feature_name (str) – The name of the feature to check.
value (float) – The value of the feature to check.
segment_duration (str) – Segment duration, either “1 min” or “5 min”.

Returns:

‘in_range’, ‘above_range’, or ‘below_range’ based on the value’s relation to the normal range.

Return type:

str

interpret_feature(feature_name, value, segment_duration='1 min')[source]

Interprets the value of a given feature based on its configuration in the YAML file.

This method compares the feature value with its normal range and provides interpretation, which could indicate whether the value is in-range, below-range, or above-range.

Parameters:

feature_name (str) – The name of the feature to interpret (e.g., “NN50”, “RMSSD”).
value (float) – The value of the feature to be interpreted.
segment_duration (str) – The duration of the segment, either ‘1 min’ or ‘5 min’. Default is ‘1 min’.

Returns:

A dictionary containing:

”description”: The description of the feature.
”normal_range”: The normal range of the feature for the given segment duration.
”interpretation”: The interpretation of the feature value (whether in-range, below-range, or above-range).
”contradiction”: Contradiction explanation with other related features.
”correlation”: Correlation explanation with other related features.

Return type:

Example Usage:

>>> engine = InterpretationEngine()
>>> result = engine.interpret_feature("NN50", 45, "5 min")
>>> print(result)
{
    "description": "Number of significant changes in heart rate. High NN50 suggests healthy parasympathetic activity.",
    "normal_range": [40, 150],
    "interpretation": "Normal parasympathetic activity. No immediate concern.",
    "contradiction": "Low NN50 contradicts high RMSSD, as both should indicate parasympathetic activity.",
    "correlation": "Positively correlated with RMSSD, as both represent short-term heart rate variability."
}

interpret_multiple_features(feature_data, segment_duration='1 min')[source]

Interprets multiple features together to provide context-aware analysis.

Parameters:

feature_data (dict) – Dictionary containing feature names and their values.
segment_duration (str) – The duration of the segment, either ‘1 min’ or ‘5 min’.

Returns:

Dictionary containing cross-feature analysis and patterns.

Return type:

load_feature_config()[source]

Loads the default feature_config.yml file from the package resources.

Returns:: Parsed YAML content from the feature_config.yml file.
Return type:: dict
Raises:: FileNotFoundError – If the feature_config.yml cannot be loaded.

Example Usage:

>>> config = engine.load_feature_config()

Web Application API

Data Service

Data service for vitalDSP webapp.

This module provides data management and processing services.

class vitalDSP_webapp.services.data.data_service.DataService[source]

Bases: object

Service for managing data operations.

clear_all_data()[source]: Clear all stored data.

clear_data(data_id: str)[source]: Clear data by ID.

clear_filtered_data(data_id: str) → bool[source]: Clear filtered data for a specific data ID.

get_all_data() → Dict[str, Any][source]: Get all stored data.

get_column_mapping(data_id: str) → Dict[str, str][source]: Get column mapping for a specific data ID.

get_config() → Dict[str, Any][source]: Get current data configuration.

get_current_config() → Dict[str, Any][source]: Get current data configuration.

get_current_data() → DataFrame | None[source]: Get current data.

get_data(data_id: str) → DataFrame | None[source]: Get data by ID.

get_data_info(data_id: str) → Dict[str, Any] | None[source]: Get data info by ID.

get_data_summary() → Dict[str, Any] | None[source]: Get summary of current data.

get_filter_info(data_id: str) → Dict[str, Any] | None[source]: Get filter information for a specific data ID.

get_filtered_data(data_id: str) → ndarray | None[source]: Retrieve filtered signal data.

has_filtered_data(data_id: str) → bool[source]: Check if filtered data is available.

load_data(file_path: str) → DataFrame | None[source]: Load data from file path.

process_data(df: DataFrame, sampling_freq: float, time_unit: str = 'seconds') → Dict[str, Any][source]: Process uploaded data and return metadata.

set_column_mapping(data_id: str, mapping: Dict[str, str])[source]: Set column mapping for a specific data ID.

store_data(df: DataFrame, info: Dict[str, Any]) → str[source]: Store data with a unique ID and return the ID.

store_filtered_data(data_id: str, filtered_signal: ndarray, filter_info: Dict[str, Any]) → bool[source]: Store filtered signal data from filtering screen.

update_column_mapping(data_id: str, mapping: Dict[str, str])[source]: Update column mapping for a specific data ID.

update_config(config: Dict[str, Any])[source]: Update data configuration.

vitalDSP_webapp.services.data.data_service.get_data_service() → DataService[source]: Get the global data service instance.

Settings Service

Settings service for vitalDSP webapp.

This module provides comprehensive settings management including persistence, validation, and integration with the application configuration.

class vitalDSP_webapp.services.settings_service.AnalysisSettings(default_sampling_freq: int = 1000, default_fft_points: int = 1024, default_window_type: str = 'hann', peak_threshold: float = 0.5, quality_threshold: float = 0.7, analysis_options: List[str] | None = None)[source]

Bases: object

Analysis and signal processing settings.

analysis_options: List[str] = None

default_fft_points: int = 1024

default_sampling_freq: int = 1000

default_window_type: str = 'hann'

peak_threshold: float = 0.5

quality_threshold: float = 0.7

class vitalDSP_webapp.services.settings_service.ApplicationSettings(general: GeneralSettings | None = None, analysis: AnalysisSettings | None = None, data: DataSettings | None = None, system: SystemSettings | None = None, last_updated: str | None = None, version: str = '1.0.0')[source]

Bases: object

Complete application settings container.

analysis: AnalysisSettings = None

data: DataSettings = None

general: GeneralSettings = None

last_updated: str = None

system: SystemSettings = None

version: str = '1.0.0'

class vitalDSP_webapp.services.settings_service.DataSettings(max_file_size: int = 100, auto_save_interval: int = 5, data_retention_days: int = 30, export_format: str = 'csv', image_format: str = 'png', export_options: List[str] | None = None)[source]

Bases: object

Data management and export settings.

auto_save_interval: int = 5

data_retention_days: int = 30

export_format: str = 'csv'

export_options: List[str] = None

image_format: str = 'png'

max_file_size: int = 100

class vitalDSP_webapp.services.settings_service.GeneralSettings(theme: str = 'light', timezone: str = 'UTC', page_size: int = 25, auto_refresh_interval: int = 30, display_options: List[str] | None = None)[source]

Bases: object

General application settings.

auto_refresh_interval: int = 30

display_options: List[str] = None

page_size: int = 25

theme: str = 'light'

timezone: str = 'UTC'

class vitalDSP_webapp.services.settings_service.SettingsService(settings_file: str = 'settings.json')[source]

Bases: object

Service for managing application settings.

export_settings(format_type: str = 'json') → str | None[source]: Export settings to specified format.

get_all_settings() → ApplicationSettings[source]: Get all current settings.

get_analysis_settings() → AnalysisSettings[source]: Get analysis settings.

get_data_settings() → DataSettings[source]: Get data settings.

get_general_settings() → GeneralSettings[source]: Get general settings.

get_settings_summary() → Dict[str, Any][source]: Get a summary of current settings for display.

get_system_settings() → SystemSettings[source]: Get system settings.

import_settings(settings_data: str, format_type: str = 'json') → bool[source]: Import settings from specified format.

reset_to_defaults() → bool[source]: Reset all settings to default values.

update_analysis_settings(**kwargs) → bool[source]: Update analysis settings.

update_data_settings(**kwargs) → bool[source]: Update data settings.

update_general_settings(**kwargs) → bool[source]: Update general settings.

update_system_settings(**kwargs) → bool[source]: Update system settings.

validate_settings() → Dict[str, List[str]][source]: Validate current settings and return any errors.

class vitalDSP_webapp.services.settings_service.SystemSettings(max_cpu_usage: int = 80, memory_limit_gb: int = 4, parallel_threads: int = 4, session_timeout_minutes: int = 60, security_options: List[str] | None = None)[source]

Bases: object

System performance and security settings.

max_cpu_usage: int = 80

memory_limit_gb: int = 4

parallel_threads: int = 4

security_options: List[str] = None

session_timeout_minutes: int = 60

vitalDSP_webapp.services.settings_service.get_current_settings() → ApplicationSettings[source]: Get current application settings.

vitalDSP_webapp.services.settings_service.get_default_settings() → Dict[str, Any][source]: Get default settings (compatibility function).

vitalDSP_webapp.services.settings_service.get_settings_service() → SettingsService[source]: Get the global settings service instance.

vitalDSP_webapp.services.settings_service.load_settings() → Dict[str, Any][source]: Load settings from file (compatibility function).

vitalDSP_webapp.services.settings_service.merge_settings(base: Dict[str, Any], override: Dict[str, Any]) → Dict[str, Any][source]: Merge settings (compatibility function).

vitalDSP_webapp.services.settings_service.save_settings(settings: Dict[str, Any]) → bool[source]: Save settings to file (compatibility function).

vitalDSP_webapp.services.settings_service.update_settings(category: str, **kwargs) → bool[source]: Update settings for a specific category.

vitalDSP_webapp.services.settings_service.validate_settings(settings: Dict[str, Any]) → bool[source]: Validate settings (compatibility function).

API Endpoints

Web Application Callbacks

Core Callbacks

Core application callbacks for vitalDSP webapp.

This module handles basic application functionality like sidebar toggling.

vitalDSP_webapp.callbacks.core.app_callbacks.register_sidebar_callbacks(app)[source]: Registers the callback to toggle the sidebar between expanded and collapsed states. Also toggles the icon between the hamburger icon and a right-pointing arrow.

Upload Callbacks

Core upload callbacks for vitalDSP webapp.

This module handles file uploads, data validation, and data processing.

vitalDSP_webapp.callbacks.core.upload_callbacks.create_data_preview(df: DataFrame, data_info: dict) → Div[source]

Render a modern Data Preview block: stat chips + styled DataTable.

Stat chips replace the four bulleted html.P lines (rows, fs, duration, memory) so the most-used numbers are scannable at a glance. The table picks up the page-scoped styles defined in upload_page.css via the data-preview-table-wrap class on the outer wrapper.

vitalDSP_webapp.callbacks.core.upload_callbacks.load_data_headers_only(file_path, data_format, sampling_freq=None, data_type='auto')[source]

Load only the headers/columns from a data file without parsing the actual data. This is used to show available columns to the user for selection.

Parameters:

file_path – Path to the data file
data_format – Format of the data file
sampling_freq – Sampling frequency (optional)
data_type – Type of data (auto, ppg, ecg, etc.)

Returns:

(available_columns, metadata_dict)

Return type:

vitalDSP_webapp.callbacks.core.upload_callbacks.load_data_with_format(file_path, data_format, sampling_freq=None, signal_type=None, signal_column=None, time_column=None, oucru_sampling_rate_column=None, oucru_interpolate_time=None)[source]

Load data using DataLoader based on the specified format.

Parameters:

file_path – Path to the data file
data_format – Format type (‘auto’, ‘csv’, ‘oucru_csv’, ‘excel’, etc.)
sampling_freq – Sampling frequency (optional)
signal_type – Signal type for OUCRU format (ppg/ecg)
signal_column – User-selected signal column name
time_column – User-selected time column name
oucru_sampling_rate_column – Column name for sampling rates in OUCRU format
oucru_interpolate_time – Whether to interpolate timestamps in OUCRU format

Returns:

(DataFrame, metadata dict)

Return type:

vitalDSP_webapp.callbacks.core.upload_callbacks.register_upload_callbacks(app)[source]

Register the slim upload-page callback set.

The page exposes one file dropzone, a signal-type select, a sampling rate field, a single signal-column dropdown (populated post-upload), and a Process Data button. Everything else (OUCRU vs flat CSV detection, signal-column auto-pick, sampling-rate inference) happens behind the scenes in load_data_with_format().

vitalDSP_webapp.callbacks.core.upload_callbacks.rehydrate_payload(payload: dict) → DataFrame[source]

Inverse of _df_to_compact_payload().

Detects __ns__-suffixed columns and converts them back to datetime64[ns] with their original names. Other columns pass through unchanged. Returns an empty DataFrame on an empty payload.

Page Routing Callbacks

Core page routing callbacks for vitalDSP webapp.

This module handles page navigation and routing based on URL paths.

vitalDSP_webapp.callbacks.core.page_routing_callbacks.display_page(pathname: str) → Div[source]

Function to dynamically update the content of the main page based on the current URL path.

Parameters:: pathname (str) – The current URL path to determine which page to display.
Returns:: The HTML content for the selected page.
Return type:: html.Div

vitalDSP_webapp.callbacks.core.page_routing_callbacks.register_page_routing_callbacks(app)[source]

Registers the callback for page routing based on the URL path.

Parameters:: app (Dash) – The Dash app object where the callback is being registered.

Analysis Callbacks

Analysis callbacks for vitalDSP core functionality.

This module handles core vitalDSP analysis callbacks including frequency domain, and signal processing operations.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.apply_filter(signal_data, sampling_freq, filter_family, filter_response, low_freq, high_freq, filter_order)[source]

Apply filter to the signal using vitalDSP SignalFiltering class.

This function now uses the validated vitalDSP SignalFiltering implementation for improved error handling, parameter validation, and consistency.

Parameters:

signal_data – Signal array
sampling_freq – Sampling frequency in Hz
filter_family – Filter type (butter, cheby1, cheby2, ellip, bessel)
filter_response – Filter response (bandpass, lowpass, highpass, bandstop)
low_freq – Low cutoff frequency in Hz
high_freq – High cutoff frequency in Hz
filter_order – Filter order

Returns:

Filtered signal

Return type:

np.ndarray

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_enhanced_psd_plot(signal_data, sampling_freq, window_sec, overlap, freq_max, log_scale, normalize, channel, column_mapping)[source]: Create enhanced PSD plot using vitalDSP FrequencyDomainFeatures.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_enhanced_spectrogram_plot(signal_data, sampling_freq, window_size, overlap, freq_max, colormap, scaling, window_type)[source]: Create enhanced spectrogram plot.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_fft_plot(signal_data, sampling_freq, window_type, n_points, freq_min, freq_max)[source]: Create FFT plot using vitalDSP FourierTransform.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_frequency_band_power_table(signal_data, sampling_freq, analysis_type, analysis_options)[source]: Create frequency band power analysis table.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_frequency_harmonics_table(signal_data, sampling_freq, analysis_type, analysis_options)[source]: Create frequency harmonics analysis table.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_frequency_peak_analysis_table(signal_data, sampling_freq, analysis_type, analysis_options, signal_type='unknown')[source]: Create frequency peak analysis table.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_frequency_stability_table(signal_data, sampling_freq, analysis_type, analysis_options)[source]: Create frequency stability analysis table.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_stft_plot(signal_data, sampling_freq, window_size, hop_size, freq_min, freq_max)[source]: Create STFT plot using vitalDSP STFT.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.create_wavelet_plot(signal_data, sampling_freq, wavelet_type, levels, freq_min, freq_max)[source]: Create wavelet plot using vitalDSP WaveletTransform.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.generate_frequency_analysis_results(signal_data, sampling_freq, analysis_type, analysis_options)[source]: Generate frequency analysis results summary.

vitalDSP_webapp.callbacks.analysis.vitaldsp_callbacks.register_vitaldsp_callbacks(app)[source]

Register vitalDSP core callbacks.

Note: Time domain callbacks have been migrated to time_domain_callbacks.py. This file now only contains helper functions for frequency domain analysis. No callbacks are registered here anymore.

Signal Filtering Callbacks

Signal filtering callbacks for vitalDSP webapp. Handles traditional filtering, advanced filtering, artifact removal, neural filtering, and ensemble filtering. Uses actual vitalDSP functions for all computations.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_additional_traditional_filters(signal_data, savgol_window, savgol_polyorder, moving_avg_window, moving_avg_iterations, gaussian_sigma, gaussian_iterations)[source]: Apply additional traditional filters like Savitzky-Golay, moving average, and Gaussian.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_advanced_filter(signal_data, advanced_method, kalman_r=None, kalman_q=None, optimization_loss_type=None, optimization_initial_guess=None, optimization_learning_rate=None, optimization_iterations=None, gradient_learning_rate=None, gradient_iterations=None, convolution_kernel_type=None, convolution_kernel_size=None, attention_type=None, attention_size=None, attention_sigma=None, attention_ascending=None, attention_base=None, adaptive_mu=None, adaptive_order=None)[source]: Apply advanced filtering using vitalDSP functions with method-specific parameters.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_enhanced_artifact_removal(signal_data, sampling_freq, artifact_type, artifact_strength, wavelet_type, wavelet_level, threshold_type, threshold_value, powerline_freq, notch_q_factor, pca_components, ica_components)[source]: Apply enhanced artifact removal using vitalDSP functions with configurable parameters.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_enhanced_ensemble_filter(signal_data, ensemble_method, ensemble_n_filters, ensemble_learning_rate, ensemble_iterations, adaptive_filter_order, adaptive_step_size, forgetting_factor, regularization_param)[source]: Apply enhanced ensemble filtering with real-time capabilities.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_ensemble_filter(signal_data, ensemble_method, ensemble_n_filters)[source]: Apply ensemble filtering using vitalDSP functions.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_filter_chain(signal, sampling_freq, signal_type, chain, logger=None)[source]

Apply a full filter chain (list of stages) to signal.

Used by the filtering page on Apply AND by downstream pages (time-domain, frequency, features) when they re-apply the user’s saved preprocessing chain to a new window. chain=None or an empty list returns the signal unchanged.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_multi_modal_filtering(signal_data, reference_signal, fusion_method, update_rate, performance_window)[source]: Apply multi-modal filtering using vitalDSP functions.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_neural_filter(signal_data, neural_type, neural_complexity)[source]: Apply neural network filtering using vitalDSP functions.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_smoothing_filter(signal_data, method, savgol_window, savgol_polyorder, moving_avg_window, gaussian_sigma)[source]

Apply a single smoothing method to signal_data.

method is one of "savgol", "moving_avg", "gaussian". Returns the smoothed signal; on failure returns the input unchanged.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.apply_traditional_filter(signal_data, sampling_freq, filter_family, filter_response, low_freq, high_freq, filter_order)[source]: Apply traditional filtering using vitalDSP functions.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.calculate_mse(original_signal, filtered_signal)[source]: Calculate Mean Square Error between original and filtered signals.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.configure_plot_with_pan_zoom(fig, title='', height=400)[source]

Configure plotly figure with pan/zoom tools and consistent styling.

Parameters:

fig – Plotly figure object
title – Plot title
height – Plot height in pixels

Returns:

Configured plotly figure

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.create_empty_figure()[source]: Create an empty figure for error cases.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.create_filter_comparison_plot(time_axis, original_signal, filtered_signal, sampling_freq, signal_type, peaks=None, notches=None)[source]

Overlay the original signal and the user-filtered signal.

Optional peaks (systolic peak indices) and notches (dicrotic notch indices) are rendered as markers on the FILTERED trace so the user can see what survived the filter. Pre-computed by the caller from a SINGLE shared WaveformMorphology pass so we don’t repeat the heavy peak detection here.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.create_filter_quality_plots(original_signal, filtered_signal, sampling_freq, quality_options, signal_type='ECG')[source]

Two focused panels: amplitude overlay + frequency response.

Replaces the old 6-panel mega-plot (which mixed signal comparison, frequency response, statistical distribution, temporal features, error analysis, performance metrics, plus PPG-hardcoded critical- point overlays). Most of that information is better presented in the text Quality Metrics panel below; here we keep only the two plots a filter user actually reads.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.create_filtered_signal_plot(time_axis, filtered_data, sampling_freq, signal_type)[source]: Create plot for filtered signal with critical points detection.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.create_original_signal_plot(time_axis, signal_data, sampling_freq, signal_type)[source]: Create plot for original signal with critical points detection.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.generate_filter_quality_metrics(original_signal, filtered_signal, sampling_freq, quality_options, signal_type='ECG')[source]

Render the headline filter-quality metrics as a compact table.

Uses vitalDSP.signal_quality_assessment.FilteringQualityAssessment.assess_quality() as the single source of truth. The previous version called the same library helper internally but then layered ~1800 lines of per-domain breakdowns (statistical, temporal, morphological, frequency, …) that duplicated calculate_* helpers also defined here. Most of that breakdown was unread by users and made the page heavy. The table below shows the five metrics that actually drive the overall quality verdict, plus the recommendation string.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.register_signal_filtering_callbacks(app)[source]: Register all signal filtering callbacks.

vitalDSP_webapp.callbacks.analysis.signal_filtering_callbacks.safe_log_range(logger, data, name='data', precision=4)[source]

Safely log min/max range of data, handling empty arrays.

Parameters:

logger – Logger instance
data – Array-like data
name – Name for logging (default: “data”)
precision – Decimal precision for formatting (default: 4)

Respiratory Analysis Callbacks

Respiratory rate analysis callbacks for vitalDSP webapp.

Runs all 6 RR extraction methods via RespiratoryAnalysis and produces per-method insight plots plus an ensemble summary.

vitalDSP_webapp.callbacks.analysis.respiratory_callbacks.register_respiratory_callbacks(app)[source]

Features Callbacks

Physiological Callbacks

Respiratory Callbacks

Utility Functions

Common Utilities

Utility Functions Module for Physiological Signal Processing

This module provides comprehensive capabilities for physiological signal processing including ECG, PPG, EEG, and other vital signs.

Author: vitalDSP Team Date: 2025-01-27 Version: 1.0.0

Key Features: - Multiple processing methods and functions - NumPy integration for numerical computations

Examples:

Basic usage:

>>> import numpy as np
>>> from vitalDSP.config_utilities.common import Common
>>> signal = np.random.randn(1000)
>>> processor = Common(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

vitalDSP.utils.config_utilities.common.argrelextrema(signal, comparator=<ufunc 'greater'>, order=1)[source]

Find the relative extrema (maxima or minima) in a 1D signal.

This function identifies local maxima or minima in a signal by comparing each point with its neighbors.

Parameters:

signal (numpy.ndarray) – The input signal in which to find the extrema.
comparator (function, optional) – The comparison function to use (e.g., np.greater for maxima, np.less for minima). Defaults to np.greater.
order (int, optional) – The number of points on each side to use for comparison. Must be a positive integer. Defaults to 1.

Returns:

extrema – Indices of the relative extrema in the signal.

Return type:

Raises:

ValueError – If the order is less than 1 or if the signal is too short to find extrema with the given order.

Examples

>>> signal = np.array([1, 3, 2, 4, 3, 5, 4])
>>> maxima = argrelextrema(signal, comparator=np.greater, order=1)
>>> print(maxima)
[1, 3, 5]

vitalDSP.utils.config_utilities.common.coherence(x, y, fs=1.0, nperseg=256)[source]

Compute the coherence between two signals.

Coherence measures the degree of correlation between two signals in the frequency domain.

Parameters:

x (numpy.ndarray) – First input signal.
y (numpy.ndarray) – Second input signal.
fs (float, optional) – Sampling frequency of the signals (default is 1.0).
nperseg (int, optional) – Length of each segment for coherence computation (default is 256).

Returns:

numpy.ndarray – Frequency array.
numpy.ndarray – Coherence values.

Examples

>>> x = np.sin(2 * np.pi * np.linspace(0, 1, 500))
>>> y = np.sin(2 * np.pi * np.linspace(0, 1, 500) + np.pi / 4)
>>> freqs, coh = coherence(x, y, fs=500)
>>> print(freqs, coh)

vitalDSP.utils.config_utilities.common.deprecated(reason)[source]

vitalDSP.utils.config_utilities.common.dtw_distance_windowed(x, y, window=None)[source]

Compute the Dynamic Time Warping (DTW) distance between two sequences using a sliding window.

Parameters:

x (numpy.ndarray) – The first time series.
y (numpy.ndarray) – The second time series.
window (int, optional) – The size of the window for the DTW computation. If None, it uses the full sequence.

Returns:

The DTW distance between the two sequences.

Return type:

vitalDSP.utils.config_utilities.common.filtfilt(b, a, signal)[source]

Apply a forward-backward IIR/FIR filter to a signal for zero-phase filtering.

Parameters:

b (numpy.ndarray) – Numerator coefficients of the filter (the feedforward part).
a (numpy.ndarray) – Denominator coefficients of the filter (the feedback part).
signal (numpy.ndarray) – The input signal to be filtered.

Returns:

y – The filtered signal with zero phase distortion.

Return type:

Examples

>>> b = np.array([0.0675, 0.1349, 0.0675])
>>> a = np.array([1.0000, -1.1430, 0.4128])
>>> signal = np.array([0.0, 0.5, 1.0, 0.5, 0.0])
>>> filtered_signal = filtfilt(b, a, signal)
>>> print(filtered_signal)

vitalDSP.utils.config_utilities.common.find_peaks(signal, height=None, distance=None, threshold=None, prominence=None, width=None)[source]

Identify peaks in a 1D signal.

This function finds local maxima in a signal that meet specific criteria such as minimum height, distance, and prominence.

Parameters:

signal (numpy.ndarray) – The input signal in which to find peaks.
height (float or None, optional) – Minimum height required for a peak. Peaks below this value are ignored.
distance (int or None, optional) – Minimum number of samples required between neighboring peaks.
threshold (float or None, optional) – Minimum difference between a peak and its neighboring points.
prominence (float or None, optional) – Minimum prominence of peaks, which measures how much a peak stands out relative to its surroundings.
width (int or None, optional) – Minimum width required for a peak, measured as the number of samples.

Returns:

peaks – Indices of the peaks in the signal that meet the specified criteria.

Return type:

Examples

>>> signal = np.array([0, 1, 0, 2, 0, 3, 0])
>>> peaks = find_peaks(signal, height=1)
>>> print(peaks)
[1, 3, 5]

vitalDSP.utils.config_utilities.common.grangercausalitytests(data, max_lag, verbose=False)[source]

Perform Granger causality tests to determine if one time series can predict another.

The Granger causality test evaluates whether the past values of one time series can provide statistically significant information about the future values of another time series.

Parameters:

data (numpy.ndarray) – The input data array with shape (n_samples, 2), where the first column is the dependent variable.
max_lag (int) – Maximum lag to consider for causality.
verbose (bool, optional) – If True, prints out detailed test statistics (default is False).

Returns:

Granger causality test results, including F-test statistics for each lag.

Return type:

Examples

>>> data = np.random.rand(100, 2)
>>> results = grangercausalitytests(data, max_lag=4, verbose=True)
>>> print(results)

vitalDSP.utils.config_utilities.common.pearsonr(x, y)[source]

Compute the Pearson correlation coefficient between two signals.

The Pearson correlation coefficient is a measure of linear correlation between two signals, with a value between -1 and 1.

Parameters:

x (numpy.ndarray) – First input signal.
y (numpy.ndarray) – Second input signal.

Returns:

Pearson correlation coefficient.

Return type:

Raises:

ValueError – If the input arrays do not have the same length.

Examples

>>> x = np.array([1, 2, 3, 4])
>>> y = np.array([1, 2, 3, 5])
>>> corr = pearsonr(x, y)
>>> print(corr)
0.98

Error Handling

Error handling utilities for vitalDSP webapp. Provides consistent error handling and user feedback.

exception vitalDSP_webapp.utils.error_handler.AnalysisError(message: str, error_code: str | None = None, details: Dict[str, Any] | None = None)[source]

Bases: WebappError

Exception raised for analysis errors.

exception vitalDSP_webapp.utils.error_handler.DataProcessingError(message: str, error_code: str | None = None, details: Dict[str, Any] | None = None)[source]

Bases: WebappError

Exception raised for data processing errors.

exception vitalDSP_webapp.utils.error_handler.FileUploadError(message: str, error_code: str | None = None, details: Dict[str, Any] | None = None)[source]

Bases: WebappError

Exception raised for file upload errors.

exception vitalDSP_webapp.utils.error_handler.ValidationError(message: str, error_code: str | None = None, details: Dict[str, Any] | None = None)[source]

Bases: WebappError

Exception raised for validation errors.

exception vitalDSP_webapp.utils.error_handler.WebappError(message: str, error_code: str | None = None, details: Dict[str, Any] | None = None)[source]

Bases: Exception

Base exception class for webapp errors.

vitalDSP_webapp.utils.error_handler.create_error_alert(error: Exception | str, title: str = 'Error', color: str = 'danger', show_details: bool = False, dismissible: bool = False) → Alert[source]

Create a standardized error alert component.

Parameters:

error (Union[Exception, str]) – Error object or error message string
title (str) – Title for the error alert
color (str) – Bootstrap color for the alert
show_details (bool) – Whether to show error details

Returns:

Error alert component

Return type:

dbc.Alert

vitalDSP_webapp.utils.error_handler.create_info_alert(message: str, title: str = 'Information') → Alert[source]

Create a standardized info alert component.

Parameters:

message (str) – Information message
title (str) – Title for the info alert

Returns:

Info alert component

Return type:

dbc.Alert

vitalDSP_webapp.utils.error_handler.create_success_alert(message: str, title: str = 'Success') → Alert[source]

Create a standardized success alert component.

Parameters:

message (str) – Success message
title (str) – Title for the success alert

Returns:

Success alert component

Return type:

dbc.Alert

vitalDSP_webapp.utils.error_handler.create_user_friendly_error_message(error: Exception) → str[source]

Create a user-friendly error message from an exception.

Parameters:: error (Exception) – The error to convert to a user-friendly message
Returns:: User-friendly error message
Return type:: str

vitalDSP_webapp.utils.error_handler.create_warning_alert(message: str, title: str = 'Warning') → Alert[source]

Create a standardized warning alert component.

Parameters:

message (str) – Warning message
title (str) – Title for the warning alert

Returns:

Warning alert component

Return type:

dbc.Alert

vitalDSP_webapp.utils.error_handler.format_error_for_display(error: Exception, include_details: bool = False) → dict[source]

Format an error for display in the UI.

Parameters:

error (Exception) – The error to format
include_details (bool) – Whether to include detailed error information

Returns:

Formatted error information for display

Return type:

vitalDSP_webapp.utils.error_handler.format_error_message(error: Exception, include_traceback: bool = False) → str[source]

Format an error message for display.

Parameters:

error (Exception) – The error to format
include_traceback (bool) – Whether to include traceback information

Returns:

Formatted error message

Return type:

str

vitalDSP_webapp.utils.error_handler.get_analysis_error_suggestions(error: Exception) → list[source]: Get user-friendly suggestions for analysis errors.

vitalDSP_webapp.utils.error_handler.get_processing_error_suggestions(error: Exception) → list[source]: Get user-friendly suggestions for processing errors.

vitalDSP_webapp.utils.error_handler.get_upload_error_suggestions(error: Exception) → list[source]: Get user-friendly suggestions for upload errors.

vitalDSP_webapp.utils.error_handler.handle_analysis_error(error: Exception, context: dict | None = None) → dict[source]

Handle analysis errors specifically.

Parameters:

error (Exception) – The analysis error that occurred
context (dict, optional) – Additional context about the analysis

Returns:

Standardized error response for analysis errors

Return type:

vitalDSP_webapp.utils.error_handler.handle_error(error: Exception, context: str = 'Unknown') → Dict[str, Any][source]

Handle an error and return standardized error information.

Parameters:

error (Exception) – The error that occurred
context (str) – Context where the error occurred

Returns:

Standardized error information

Return type:

Dict[str, Any]

vitalDSP_webapp.utils.error_handler.handle_processing_error(error: Exception, context: dict | None = None) → dict[source]

Handle data processing errors specifically.

Parameters:

error (Exception) – The processing error that occurred
context (dict, optional) – Additional context about the processing

Returns:

Standardized error response for processing errors

Return type:

vitalDSP_webapp.utils.error_handler.handle_upload_error(error: Exception, context: dict | None = None) → dict[source]

Handle file upload errors specifically.

Parameters:

error (Exception) – The upload error that occurred
context (dict, optional) – Additional context about the upload

Returns:

Standardized error response for upload errors

Return type: