Physiological Features 

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

Comprehensive heart rate variability analysis including time-domain, frequency-domain, and nonlinear metrics.

HRV Features

Comprehensive heart rate variability analysis including time-domain, frequency-domain, and nonlinear metrics.

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 HrvAnalysis
>>> signal = np.random.randn(1000)
>>> processor = HrvAnalysis(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

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)

Nonlinear Analysis

Nonlinear dynamics and complexity measures for physiological signals.

Nonlinear Features

Nonlinear dynamics and complexity measures for physiological signal 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 - Performance optimization

Examples:

Basic usage:

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

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}")

Advanced Nonlinear Features

State-of-the-art nonlinear dynamics and information-theoretic methods for advanced physiological signal analysis.

Multi-Scale Entropy Analysis

Multi-scale entropy (MSE) analysis quantifies signal complexity across multiple temporal scales, providing insights into physiological regulation and system health.

Key Features:

Standard MSE, Composite MSE (CMSE), and Refined Composite MSE (RCMSE)
KD-tree accelerated sample entropy calculation (O(N log N))
Complexity Index for single-value assessment
Fuzzy entropy option for noisy signals

Clinical Applications:

Cardiac arrhythmia detection and classification
Aging assessment and cardiovascular health evaluation
Autonomic nervous system function assessment
Disease progression monitoring (heart failure, diabetes)

Usage Example:

from vitalDSP.physiological_features.advanced_entropy import MultiScaleEntropy

# Analyze RR interval complexity
mse = MultiScaleEntropy(rr_intervals, max_scale=20, m=2, r=0.15)
entropy_curve = mse.compute_rcmse()
complexity_index = mse.get_complexity_index(entropy_curve, scale_range=(1, 15))

# Interpret results
if complexity_index > 30:
    print("Healthy complexity profile")
elif complexity_index > 15:
    print("Reduced complexity - monitoring recommended")
else:
    print("Severely reduced complexity - clinical attention needed")

Advanced Entropy Analysis Module

This module provides advanced entropy-based complexity measures for physiological signal analysis, including:

Multi-Scale Entropy (MSE) - Costa et al. (2002)
Composite Multi-Scale Entropy (CMSE) - Wu et al. (2013)
Refined Composite Multi-Scale Entropy (RCMSE) - Wu et al. (2014)
Multi-Scale Sample Entropy (MSSE)
Multi-Scale Fuzzy Entropy (MFE)

These methods analyze signal complexity across multiple time scales, providing insights into the multi-scale structure of physiological signals.

Clinical Applications:

Cardiac arrhythmia detection and classification
Aging assessment and cardiovascular health
Autonomic nervous system function evaluation
Disease progression monitoring (heart failure, diabetes)
Sleep stage classification
Seizure prediction and epilepsy monitoring

Mathematical Background:

Multi-scale entropy extends traditional entropy measures by analyzing the signal at multiple temporal scales through a coarse-graining procedure. This reveals complexity at different time scales, which is crucial for understanding physiological regulation mechanisms.

References:

Costa, M., Goldberger, A. L., & Peng, C. K. (2002). Multiscale entropy analysis of complex physiologic time series. Physical review letters, 89(6), 068102.
Wu, S. D., Wu, C. W., Lin, S. G., Wang, C. C., & Lee, K. Y. (2013). Time series analysis using composite multiscale entropy. Entropy, 15(3), 1069-1084.
Wu, S. D., Wu, C. W., Lin, S. G., Lee, K. Y., & Peng, C. K. (2014). Analysis of complex time series using refined composite multiscale entropy. Physics Letters A, 378(20), 1369-1374.
Ahmed, M. U., & Mandic, D. P. (2011). Multivariate multiscale entropy: A tool for complexity analysis of multichannel data. Physical Review E, 84(6), 061918.

Date: October 10, 2025 Version: 1.0

class vitalDSP.physiological_features.advanced_entropy.MultiScaleEntropy(signal: ndarray, max_scale: int = 20, m: int = 2, r: float = 0.15, fuzzy: bool = False)[source]

Bases: object

Multi-Scale Entropy (MSE) analysis for physiological signals.

MSE quantifies the complexity of a signal across multiple temporal scales through coarse-graining followed by entropy calculation at each scale.

The method reveals how signal complexity changes with scale, providing insights into the multi-scale regulatory mechanisms of physiological systems.

Parameters:

signal (numpy.ndarray) – Input time series signal (1D array)
max_scale (int, optional) – Maximum scale factor for coarse-graining (default: 20) Recommended: 20 for HRV analysis, 10-15 for shorter signals
m (int, optional) – Embedding dimension (pattern length) for entropy calculation (default: 2) Typically m=2 for physiological signals
r (float, optional) – Tolerance for pattern matching (default: 0.15) Expressed as fraction of signal standard deviation Recommended: 0.15-0.25 for physiological signals
fuzzy (bool, optional) – Use fuzzy membership functions instead of binary matching (default: False) Fuzzy entropy is more stable for short signals

signal

Original input signal

Type:: numpy.ndarray

max_scale

Maximum scale for analysis

Type:: int

m

Embedding dimension

Type:: int

r

Tolerance (absolute value)

Type:: float

fuzzy

Whether to use fuzzy entropy

Type:: bool

compute_mse()[source]: Compute Multi-Scale Entropy across all scales

compute_cmse()[source]: Compute Composite Multi-Scale Entropy (improved stability)

compute_rcmse()[source]: Compute Refined Composite Multi-Scale Entropy (best stability)

get_complexity_index()[source]: Calculate complexity index (area under MSE curve)

Examples

>>> # Analyze heart rate variability
>>> import numpy as np
>>> from vitalDSP.physiological_features.advanced_entropy import MultiScaleEntropy
>>>
>>> # Generate synthetic HRV signal (RR intervals in seconds)
>>> np.random.seed(42)
>>> rr_intervals = 1.0 + 0.05 * np.random.randn(1000)  # 60 BPM baseline
>>>
>>> # Compute MSE
>>> mse = MultiScaleEntropy(rr_intervals, max_scale=20, m=2, r=0.15)
>>> entropy_values = mse.compute_mse()
>>>
>>> # Get complexity index
>>> ci = mse.get_complexity_index(entropy_values)
>>> print(f"Complexity Index: {ci:.4f}")
>>>
>>> # Compare young vs elderly (example)
>>> # Young: Higher complexity at multiple scales
>>> # Elderly: Reduced complexity, flatter MSE curve

Notes

Interpretation Guidelines:

Healthy/Young: MSE values remain high or increase at larger scales indicating rich multi-scale complexity
Disease/Aging: MSE values decrease more rapidly with scale, indicating loss of complexity and adaptive capacity
Scale-Specific Information:
- Scales 1-4: Short-term dynamics (seconds to minutes)
- Scales 5-10: Mid-term dynamics (minutes to tens of minutes)
- Scales 10-20: Long-term dynamics (tens of minutes to hours)

Signal Length Requirements: - Minimum: 100 * scale samples for reliable estimation - Recommended: 500-1000+ samples for max_scale=20 - Shorter signals: Use smaller max_scale or CMSE/RCMSE variants

Parameter Selection: - m=2: Standard for most physiological signals - m=3: For signals requiring more detailed patterns - r=0.15: Conservative choice (good specificity) - r=0.20-0.25: More lenient (better for noisy signals)

compute_cmse() → ndarray[source]

Compute Composite Multi-Scale Entropy (CMSE).

CMSE improves upon standard MSE by averaging entropy values across multiple coarse-grained series with different starting points. This reduces variance and provides more stable estimates, especially for shorter signals.

Returns:

cmse_values (numpy.ndarray) – Array of composite entropy values for each scale
Algorithm
———
For each scale τ = 1, 2, …, max_scale –
1. Create τ different coarse-grained series starting at indices 0, 1, …, τ-1
2. Compute entropy for each coarse-grained series
3. Average the τ entropy values
Advantages over Standard MSE
—————————–
1. **Reduced Variance (*** Averaging reduces statistical fluctuations*)
2. **Better Stability (*** More reliable for short signals*)
3. **Improved Discrimination (*** Better separates different signal classes*)
4. **Consistent Results (*** Less sensitive to signal length*)
Time Complexity
—————
O(max_scale² * N log N)
Note (~τ times slower than MSE due to multiple coarse-grainings)
Examples
——–
>>> mse = MultiScaleEntropy(signal, max_scale=15)
>>> cmse_values = mse.compute_cmse()
>>>
>>> # Compare with standard MSE
>>> mse_values = mse.compute_mse()
>>>
>>> import matplotlib.pyplot as plt
>>> scales = np.arange(1, 16)
>>> plt.plot(scales, mse_values, ‘o-’, label=’MSE’)
>>> plt.plot(scales, cmse_values, ‘s-’, label=’CMSE’)
>>> plt.xlabel(‘Scale’)
>>> plt.ylabel(‘Entropy’)
>>> plt.legend()
>>> plt.grid(True)
References
———-
Wu, S. D., Wu, C. W., Lin, S. G., Wang, C. C., & Lee, K. Y. (2013).
Time series analysis using composite multiscale entropy. Entropy,
15(3), 1069-1084.
Notes
—–
CMSE is particularly recommended when
- Signal length < 1000 samples
- max_scale > 10
- Comparing signals of different lengths
- High precision is required

compute_mse() → ndarray[source]

Compute Multi-Scale Entropy (MSE) across all scales.

This is the standard MSE algorithm that computes entropy at each coarse-grained scale from 1 to max_scale.

Returns:

mse_values (numpy.ndarray) – Array of entropy values for each scale (length: max_scale) Index i corresponds to scale i+1
Algorithm
———
For each scale τ = 1, 2, …, max_scale –
1. Coarse-grain signal at scale τ
2. Compute Sample Entropy (or Fuzzy Entropy) of coarse-grained signal
3. Store entropy value for scale τ
Time Complexity
—————
O(max_scale * N log N) where N is signal length
Examples
——–
>>> mse = MultiScaleEntropy(signal, max_scale=20)
>>> entropy_values = mse.compute_mse()
>>>
>>> # Plot MSE curve
>>> import matplotlib.pyplot as plt
>>> scales = np.arange(1, 21)
>>> plt.plot(scales, entropy_values, ‘o-‘)
>>> plt.xlabel(‘Scale Factor’)
>>> plt.ylabel(‘Sample Entropy’)
>>> plt.title(‘Multi-Scale Entropy’)
>>> plt.grid(True)
>>> plt.show()
Clinical Interpretation
———————–
- **Healthy/Young (*** MSE stays elevated or increases at larger scales*)
- **Disease/Aging (*** MSE decreases rapidly with scale*)
- **Heart Failure (*** Marked decrease in entropy at all scales*)
- **Atrial Fibrillation (*** Very high entropy at small scales, rapid decrease*)

compute_rcmse() → ndarray[source]

Compute Refined Composite Multi-Scale Entropy (RCMSE).

RCMSE further refines CMSE by using a modified coarse-graining procedure that preserves more information from the original signal.

Returns:

rcmse_values (numpy.ndarray) – Array of refined composite entropy values
Refined Coarse-Graining
———————–
Instead of non-overlapping windows, RCMSE uses overlapping windows
y^(τ)_j = (1/τ) * Σ(i=j to j+τ-1) x_i
This preserves more temporal structure and reduces information loss.
Advantages over CMSE
——————–
1. **Better Information Preservation (*** Overlapping windows retain more details*)
2. **Smoother Curves (*** Less jagged MSE curves*)
3. **Improved Sensitivity (*** Better detects subtle changes*)
4. **Best Stability (*** Superior performance on short signals*)
When to Use RCMSE
—————–
- Short signals (< 500 samples)
- Need maximum stability
- Require smooth, interpretable curves
- Comparing very different conditions
References
———-
Wu, S. D., Wu, C. W., Lin, S. G., Lee, K. Y., & Peng, C. K. (2014).
Analysis of complex time series using refined composite multiscale
entropy. Physics Letters A, 378(20), 1369-1374.

get_complexity_index(entropy_values: ndarray, scale_range: Tuple[int, int] | None = None) → float[source]

Calculate Complexity Index (CI) as area under the MSE curve.

The complexity index summarizes the overall complexity across scales into a single scalar value. Higher CI indicates more complex, healthy physiological regulation.

Parameters:

entropy_values (numpy.ndarray) – MSE/CMSE/RCMSE values
scale_range (tuple of int, optional) – (start_scale, end_scale) for integration (default: all scales) Useful for focusing on specific temporal scales

Returns:

complexity_index (float) – Area under the entropy curve (using trapezoidal integration)
Formula
——-
CI = Σ(i=1 to max_scale-1) [(Entropy_i + Entropy_(i+1)) / 2]
Clinical Interpretation
———————–
- **High CI (*** Complex, adaptive physiological regulation (healthy)*)
- **Low CI (*** Simple, less adaptive regulation (disease, aging)*)
- **Very Low CI (*** Pathological simplification (severe disease)*)
Examples
——–
>>> mse = MultiScaleEntropy(signal)
>>> entropy = mse.compute_mse()
>>>
>>> # Overall complexity
>>> ci_total = mse.get_complexity_index(entropy)
>>>
>>> # Short-term complexity (scales 1-5)
>>> ci_short = mse.get_complexity_index(entropy, scale_range=(1, 5))
>>>
>>> # Long-term complexity (scales 10-20)
>>> ci_long = mse.get_complexity_index(entropy, scale_range=(10, 20))
Notes
—–
Different scale ranges provide insights into different regulatory mechanisms
- Scales 1-5 (Intrinsic cardiac dynamics)
- Scales 5-10 (Sympathovagal balance)
- Scales 10-20 (Long-term regulatory mechanisms)

Symbolic Dynamics Analysis

Symbolic dynamics transforms continuous physiological signals into discrete symbol sequences for pattern analysis and complexity quantification.

Key Features:

Multiple symbolization methods: 0V (HRV-specific), Quantile, SAX, Threshold
Shannon entropy, Renyi entropy, Permutation entropy
Word distribution and forbidden words analysis
Transition matrix computation for Markov analysis

Clinical Applications:

Heart rate variability pattern classification
Autonomic regulation assessment
Arrhythmia detection (atrial fibrillation screening)
Sleep stage classification from EEG

Usage Example:

from vitalDSP.physiological_features.symbolic_dynamics import SymbolicDynamics

# HRV symbolic analysis with 0V method
sd = SymbolicDynamics(rr_intervals, n_symbols=4, word_length=3, method='0V')

# Compute multiple metrics
shannon = sd.compute_shannon_entropy()
forbidden = sd.detect_forbidden_words()
perm_ent = sd.compute_permutation_entropy(order=3)

print(f"Shannon Entropy: {shannon['normalized_entropy']:.3f}")
print(f"Forbidden Words: {forbidden['forbidden_percentage']:.1f}%")
print(f"Interpretation: {forbidden['interpretation']}")

Symbolic Dynamics Analysis Module

This module provides symbolic dynamics methods for analyzing physiological signals by transforming continuous signals into discrete symbol sequences for pattern analysis.

Implemented Methods:

Symbolic Transformation (0V, 1V, 2LV, 2UV patterns)
Shannon Entropy of Symbol Distribution
Word Distribution Analysis
Forbidden Words Detection
Pattern Transition Analysis
Renyi Entropy
Permutation Entropy

Clinical Applications:

Cardiac autonomic function assessment
Arrhythmia detection and classification
Sleep stage classification
Fetal heart rate monitoring
Blood pressure variability analysis
Seizure prediction

Mathematical Background:

Symbolic dynamics transforms a continuous-valued time series into a sequence of discrete symbols based on pattern recognition. This approach reduces noise sensitivity and reveals underlying regulatory patterns.

The transformation captures important dynamical features while being robust to: - Measurement noise - Non-stationarity - Missing data - Computational complexity

References:

Voss, A., Schulz, S., Schroeder, R., Baumert, M., & Caminal, P. (2009). Methods derived from nonlinear dynamics for analysing heart rate variability. Philosophical Transactions of the Royal Society A, 367(1887), 277-296.
Porta, A., Guzzetti, S., Montano, N., Furlan, R., Pagani, M., Malliani, A., & Cerutti, S. (2001). Entropy, entropy rate, and pattern classification as tools to typify complexity in short heart period variability series. IEEE Transactions on Biomedical Engineering, 48(11), 1282-1291.
Bandt, C., & Pompe, B. (2002). Permutation entropy: a natural complexity measure for time series. Physical review letters, 88(17), 174102.

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.symbolic_dynamics import SymbolicDynamics
>>> signal = np.random.randn(1000)
>>> processor = SymbolicDynamics(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.physiological_features.symbolic_dynamics.SymbolicDynamics(signal: ndarray, n_symbols: int = 4, word_length: int = 3, method: str = '0V')[source]

Bases: object

Symbolic Dynamics Analysis for physiological signals.

Transforms continuous time series into symbolic sequences and analyzes the distribution and patterns of symbols.

Parameters:

signal (numpy.ndarray) – Input time series signal (1D array)
n_symbols (int, optional) – Number of symbols to use (default: 4) Common choices: 3, 4, 6
word_length (int, optional) – Length of words to analyze (default: 3) Typical range: 2-5
method (str, optional) – Symbolization method (default: ‘0V’) Options: ‘0V’ (variations), ‘quantile’, ‘SAX’, ‘threshold’

signal

Original signal

Type:: numpy.ndarray

n_symbols

Number of symbols

Type:: int

word_length

Word length for pattern analysis

Type:: int

method

Symbolization method

Type:: str

symbols

Symbolic sequence

Type:: numpy.ndarray

symbolize()[source]: Transform signal to symbol sequence

compute_shannon_entropy()[source]: Shannon entropy of symbol distribution

compute_word_distribution()[source]: Distribution of words

detect_forbidden_words()[source]: Find patterns that never occur

compute_transition_matrix()[source]: Symbol transition probabilities

compute_renyi_entropy(alpha)[source]: Generalized Renyi entropy

compute_permutation_entropy()[source]: Permutation entropy

Examples

>>> # Analyze heart rate variability
>>> from vitalDSP.physiological_features.symbolic_dynamics import SymbolicDynamics
>>> import numpy as np
>>>
>>> # RR intervals (seconds)
>>> rr = np.array([1.0, 0.95, 1.02, 0.98, 1.01, 0.96, ...])
>>>
>>> # Create symbolic representation
>>> sd = SymbolicDynamics(rr, n_symbols=4, word_length=3)
>>> symbols = sd.symbolize()
>>>
>>> # Compute Shannon entropy
>>> h = sd.compute_shannon_entropy()
>>> print(f"Shannon Entropy: {h:.4f}")
>>>
>>> # Analyze word distribution
>>> word_dist = sd.compute_word_distribution()
>>>
>>> # Find forbidden words (never occurring patterns)
>>> forbidden = sd.detect_forbidden_words()
>>> print(f"Forbidden words: {len(forbidden)}")

Notes

Symbol Interpretation (0V method):

0V (no variation): Three consecutive values are approximately equal Represents stable regulation
1V (one variation): Two values equal, one different Represents small perturbations
2LV (two variations, low first): Low-High-Low or similar Represents oscillatory pattern with deceleration
2UV (two variations, high first): High-Low-High or similar Represents oscillatory pattern with acceleration

Clinical Interpretation:

Healthy: Balanced distribution of symbols, few forbidden words
Disease: Skewed distribution, many forbidden words
Atrial Fibrillation: Very high entropy, nearly uniform distribution
Heart Failure: Low entropy, many forbidden words

Parameter Recommendations:

n_symbols: 4-6 for HRV analysis
word_length: 3 for balance of detail and statistics
method: ‘0V’ for HRV, ‘quantile’ for general signals

compute_permutation_entropy(order: int = 3) → float[source]

Compute Permutation Entropy.

Permutation entropy analyzes the order relationships between consecutive values, making it robust to noise and monotonic transformations.

Parameters:

order (int) – Order of permutation patterns (default: 3) Typical range: 3-7

Returns:

perm_entropy (float) – Permutation entropy value
Algorithm
———
1. Extract overlapping windows of length ‘order’
2. Determine ranking permutation for each window
3. Count frequency of each permutation pattern
4. Calculate Shannon entropy of permutation distribution
Advantages
———-
- Robust to noise
- Fast computation
- Conceptually simple
- Good for nonlinear signals
References
———-
Bandt, C., & Pompe, B. (2002). Permutation entropy (a natural complexity)
measure for time series. Physical review letters, 88(17), 174102.
Examples
——–
>>> sd = SymbolicDynamics(signal)
>>> pe = sd.compute_permutation_entropy(order=3)
>>> print(f”Permutation Entropy ({pe:.4f}”))

compute_renyi_entropy(alpha: float = 2.0) → float[source]

Compute Renyi entropy (generalized entropy measure).

Parameters:

alpha (float) – Order parameter - alpha=0: Hartley entropy (log of number of distinct symbols) - alpha=1: Shannon entropy (limit as alpha→1) - alpha=2: Collision entropy - alpha=∞: Min-entropy

Returns:

renyi_entropy (float) – Renyi entropy value
Formula
——-
H_α = (1/(1-α)) * log2(Σ p_i^α)
where p_i are symbol probabilities.
Clinical Use
————
Different alpha values emphasize different aspects
- α < 1 (Emphasizes rare events)
- α > 1 (Emphasizes common events)
- α = 2 (Good balance, computationally efficient)

compute_shannon_entropy() → float[source]

Compute Shannon entropy of symbol distribution.

Shannon entropy quantifies the average information content or unpredictability of the symbol sequence.

Returns:

entropy (float) – Shannon entropy in bits (log base 2)
Formula
——-
H = -Σ p(i) * log2(p(i))
where p(i) is the probability of symbol i.
Interpretation
————–
- **0 (*** Completely predictable (only one symbol appears)*)
- **log2(n_symbols) (*** Maximum entropy (uniform distribution)*)
- **Between (*** Degree of predictability/complexity*)
Clinical Significance
———————
- **Low H (*** Regular, predictable rhythm (may indicate reduced adaptability)*)
- **High H (*** Variable, unpredictable rhythm (healthy variability)*)
- **Very High H (*** Chaotic, random (e.g., atrial fibrillation)*)
Examples
——–
>>> sd = SymbolicDynamics(signal)
>>> sd.symbolize()
>>> h = sd.compute_shannon_entropy()
>>>
>>> # Normalize by maximum possible entropy
>>> h_max = np.log2(sd.n_symbols)
>>> h_norm = h / h_max
>>> print(f”Normalized entropy ({h_norm:.4f}”))

compute_symbolic_features() → Dict[str, float][source]

Convenience method that computes all symbolic dynamics features.

Returns:

Dictionary containing all symbolic dynamics metrics:

’shannon_entropy’: Shannon entropy of symbol distribution
’renyi_entropy’: Renyi entropy (alpha=2)
’permutation_entropy’: Permutation entropy (order=3)
’num_words’: Total number of words in symbol sequence
’num_forbidden_words’: Number of forbidden word patterns

Return type:

Example

>>> nn_intervals = [800, 810, 790, 805, 795, 820, 780, 815]
>>> sd = SymbolicDynamics(nn_intervals)
>>> features = sd.compute_symbolic_features()
>>> print(f"Shannon Entropy: {features['shannon_entropy']:.3f}")

compute_transition_matrix() → ndarray[source]

Compute symbol transition probability matrix.

Returns:

transition_matrix (numpy.ndarray) – Matrix of transition probabilities (n_symbols x n_symbols) Element [i,j] = P(next symbol is j | current symbol is i)
Examples
——–
>>> sd = SymbolicDynamics(signal, n_symbols=4)
>>> sd.symbolize()
>>> trans = sd.compute_transition_matrix()
>>>
>>> # Visualize transition matrix
>>> import matplotlib.pyplot as plt
>>> plt.imshow(trans, cmap=’hot’, interpolation=’nearest’)
>>> plt.colorbar(label=’Transition Probability’)
>>> plt.xlabel(‘Next Symbol’)
>>> plt.ylabel(‘Current Symbol’)
>>> plt.title(‘Symbol Transition Matrix’)

compute_word_distribution() → Dict[str, float][source]

Compute distribution of words (symbol patterns).

Returns:

word_dist (dict) – Dictionary mapping words to their probabilities Keys: words (strings of symbols) Values: probabilities (0-1)
Examples
——–
>>> sd = SymbolicDynamics(signal, word_length=3)
>>> sd.symbolize()
>>> word_dist = sd.compute_word_distribution()
>>>
>>> # Most common words
>>> sorted_words = sorted(word_dist.items(), key=lambda x (x[1], reverse=True))
>>> print(“Top 5 most common words (“))
>>> for word, prob in sorted_words[ (5]:)
… print(f”{word} ({prob:.4f}”))

detect_forbidden_words() → List[str][source]

Detect forbidden words (patterns that never occur).

Returns:

forbidden_words (list of str) – List of words that never appear in the sequence
Significance
————
Forbidden words indicate deterministic constraints or regulatory
mechanisms that prevent certain patterns from occurring.
- **Many forbidden words (*** Strong regulatory constraints (often pathological)*)
- **Few forbidden words (*** Flexible regulation (typically healthy)*)
- **No forbidden words (*** Complete randomness (e.g., atrial fibrillation)*)
Examples
——–
>>> sd = SymbolicDynamics(signal, n_symbols=4, word_length=3)
>>> sd.symbolize()
>>> forbidden = sd.detect_forbidden_words()
>>>
>>> total_possible = sd.n_symbols * sd.word_length*
>>> forbidden_ratio = len(forbidden) / total_possible
>>> print(f”Forbidden word ratio ({forbidden_ratio:.2%}”))

symbolize() → ndarray[source]

Transform continuous signal to symbolic sequence.

Returns:

symbols (numpy.ndarray) – Array of symbol indices (integers 0 to n_symbols-1)
Methods
——-
**1. 0V Method (Variations) (****)
Classifies triplets based on pattern variations
- 0V (all approximately equal (|a-b|<δ, |b-c|<δ, |a-c|<δ))
- 1V (two equal, one different)
- 2LV (two variations with low-high-low pattern)
- 2UV (two variations with high-low-high pattern)
**2. Quantile Method (****)
Divides signal into n_symbols quantiles
**3. SAX (Symbolic Aggregate approXimation) (****)
Uses Gaussian quantiles for symbolization
**4. Threshold Method (****)
Simple thresholding based on percentiles
Examples
——–
>>> sd = SymbolicDynamics(signal, n_symbols=4, method=’0V’)
>>> symbols = sd.symbolize()
>>>
>>> # Convert to letter representation
>>> letters = ‘’.join([chr(65+s) for s in symbols]) # A, B, C, D…
>>> print(f”Symbolic sequence ({letters[:50]}…”))

Transfer Entropy Analysis

Transfer entropy quantifies directional information flow between coupled physiological signals, revealing causal relationships and coupling dynamics.

Key Features:

KNN-based transfer entropy estimation (Kraskov-Stögbauer-Grassberger method)
Bidirectional coupling analysis
Time-delayed TE for temporal dynamics assessment
Statistical significance testing with surrogate data

Clinical Applications:

Cardio-respiratory coupling analysis
Brain-heart interaction assessment
Autonomic nervous system evaluation
Multi-organ system dynamics monitoring

Usage Example:

from vitalDSP.physiological_features.transfer_entropy import TransferEntropy

# Analyze respiration → heart rate coupling
te = TransferEntropy(respiration, heart_rate, k=2, l=2, delay=1, k_neighbors=3)

# Bidirectional analysis
bidirectional = te.compute_bidirectional_te()
print(f"Respiration → HR: {bidirectional['te_forward']:.3f}")
print(f"HR → Respiration: {bidirectional['te_backward']:.3f}")
print(f"Coupling type: {bidirectional['interpretation']}")

# Statistical significance
significance = te.test_significance(n_surrogates=1000)
print(f"p-value: {significance['p_value']:.4f} {significance['significance']}")

# Find optimal coupling delay
delayed = te.compute_time_delayed_te(max_delay=10)
print(f"Optimal delay: {delayed['optimal_delay']} seconds")

Transfer Entropy Module for Coupling Analysis

This module provides transfer entropy methods for analyzing directional information flow and coupling between physiological signals.

Implemented Methods:

Transfer Entropy (TE)
Conditional Transfer Entropy
Time-Delayed Transfer Entropy
Normalized Transfer Entropy
Effective Transfer Entropy

Clinical Applications:

Cardio-respiratory coupling analysis
Brain-heart interaction
Autonomic nervous system assessment
Multi-organ system dynamics
Baroreflex sensitivity
Neurovascular coupling

Mathematical Background:

Transfer entropy quantifies the directed (causal) information flow from one time series to another. Unlike correlation, it captures nonlinear relationships and distinguishes the direction of influence.

TE from X to Y measures how much uncertainty about the future of Y is reduced by knowing the past of X, given the past of Y.

References:

Schreiber, T. (2000). Measuring information transfer. Physical review letters, 85(2), 461.
Faes, L., Nollo, G., & Porta, A. (2011). Information-based detection of nonlinear Granger causality in multivariate processes via a nonuniform embedding technique. Physical Review E, 83(5), 051112.
Barnett, L., Barrett, A. B., & Seth, A. K. (2009). Granger causality and transfer entropy are equivalent for Gaussian variables. Physical review letters, 103(23), 238701.
Vakorin, V. A., Krakovska, O. A., & McIntosh, A. R. (2009). Confounding effects of indirect connections on causality estimation. Journal of neuroscience methods, 184(1), 152-160.

Date: October 10, 2025 Version: 1.0

class vitalDSP.physiological_features.transfer_entropy.TransferEntropy(source: ndarray, target: ndarray, k_coef: int = 1, l_coef: int = 1, delay: int = 1, n_bins: int | None = None, k_neighbors: int = 3)[source]

Bases: object

Transfer Entropy analysis for directional coupling between signals.

Transfer Entropy (TE) quantifies the directional information flow from a source signal to a target signal, revealing causal relationships.

Parameters:

source (numpy.ndarray) – Source time series (potential driver)
target (numpy.ndarray) – Target time series (potentially driven)
k (int, optional) – History length (embedding dimension) for target (default: 1)
l (int, optional) – History length for source (default: 1)
delay (int, optional) – Time delay for embedding (default: 1)
n_bins (int, optional) – Number of bins for histogram estimation (default: None, uses KNN)
k_neighbors (int, optional) – Number of nearest neighbors for KNN estimation (default: 3)

source

Source signal

Type:: numpy.ndarray

target

Target signal

Type:: numpy.ndarray

k

Target history length

Type:: int

l

Source history length

Type:: int

delay

Embedding delay

Type:: int

compute_transfer_entropy()[source]: Compute TE from source to target

compute_bidirectional_te()[source]: Compute TE in both directions

compute_time_delayed_te(max_delay)[source]: TE across multiple time delays

compute_effective_te()[source]: Normalized effective TE

test_significance(n_surrogates)[source]: Statistical significance testing

Examples

>>> # Analyze cardio-respiratory coupling
>>> from vitalDSP.physiological_features.transfer_entropy import TransferEntropy
>>> import numpy as np
>>>
>>> # Heart rate (BPM) and respiration rate
>>> heart_rate = np.array([...])  # Time series of HR
>>> resp_rate = np.array([...])   # Time series of respiration
>>>
>>> # Compute transfer entropy
>>> te = TransferEntropy(resp_rate, heart_rate, k=1, l=1)
>>>
>>> # Respiratory influence on heart rate
>>> te_resp_to_hr = te.compute_transfer_entropy()
>>> print(f"TE(Resp → HR): {te_resp_to_hr:.4f}")
>>>
>>> # Bidirectional coupling
>>> te_forward, te_backward = te.compute_bidirectional_te()
>>> print(f"TE(Resp → HR): {te_forward:.4f}")
>>> print(f"TE(HR → Resp): {te_backward:.4f}")
>>>
>>> # Net directional influence
>>> net_te = te_forward - te_backward
>>> if net_te > 0:
...     print("Respiration drives heart rate")
>>> else:
...     print("Heart rate drives respiration")

Notes

Interpretation:

TE > 0: Information flows from source to target
TE ≈ 0: No directional coupling detected
TE < 0: Should not occur (implementation error)

Comparison with Bidirectional TE:

If TE(X→Y) > TE(Y→X): X predominantly drives Y
If TE(X→Y) ≈ TE(Y→X): Bidirectional coupling or common drive
Significance testing required to confirm non-zero values

Parameter Guidelines:

k, l: Start with 1, increase if signals have memory
delay: Typically 1 for high sampling rate, larger for slower dynamics
k_neighbors: 3-5 for most applications

Computational Considerations:

Uses KNN estimation (Kraskov method) for continuous signals
Time complexity: O(N log N) with KD-trees
Requires signals of same length
Stationary signals recommended

compute_bidirectional_te() → Tuple[float, float][source]

Compute transfer entropy in both directions.

Returns:

te_forward (float) – TE from source to target
te_backward (float) – TE from target to source
Examples
——–
>>> te = TransferEntropy(resp, hr)
>>> te_resp_hr, te_hr_resp = te.compute_bidirectional_te()
>>>
>>> # Net directional coupling
>>> net_coupling = te_resp_hr - te_hr_resp
>>> dominant_direction = “Resp → HR” if net_coupling > 0 else “HR → Resp”
>>> print(f”Dominant direction ({dominant_direction}”))
>>> print(f”Coupling asymmetry ({abs(net_coupling):.4f}”))
Interpretation
————–
Comparing bidirectional TE reveals
1. **Dominant Direction (****) –
- TE(X→Y) >> TE(Y→X): X drives Y
- TE(X→Y) << TE(Y→X): Y drives X
- TE(X→Y) ≈ TE(Y→X): Bidirectional or common drive
2. **Coupling Strength (****) –
- Sum = TE(X→Y) + TE(Y→X): Total coupling
- Difference = |TE(X→Y) - TE(Y→X)|: Directional asymmetry

compute_effective_te() → float[source]

Compute normalized effective transfer entropy.

Returns:

effective_te (float) – Normalized TE in range [0, 1]
Formula
——-
Effective TE = TE / H(target_future | target_past)
Normalization provides
- Scale-independent measure
- Interpretability as fraction of uncertainty reduced
- Easier comparison across different signal pairs
Examples
——–
>>> te_analyzer = TransferEntropy(x, y)
>>> eff_te = te_analyzer.compute_effective_te()
>>> print(f”Effective TE ({eff_te:.2%}”))

compute_time_delayed_te(max_delay: int = 10) → ndarray[source]

Compute transfer entropy across multiple time delays.

Parameters:

max_delay (int) – Maximum time delay to test

Returns:

te_values (numpy.ndarray) – TE values for each delay (length: max_delay)
Purpose
——-
Different physiological processes operate at different time scales.
Time-delayed TE reveals the temporal dynamics of coupling.
Examples
——–
>>> te = TransferEntropy(source, target)
>>> te_delays = te.compute_time_delayed_te(max_delay=20)
>>>
>>> # Find optimal delay
>>> optimal_delay = np.argmax(te_delays) + 1
>>> print(f”Peak coupling at delay ({optimal_delay}”))
>>>
>>> # Plot delay profile
>>> import matplotlib.pyplot as plt
>>> delays = np.arange(1, 21)
>>> plt.plot(delays, te_delays, ‘o-‘)
>>> plt.xlabel(‘Time Delay’)
>>> plt.ylabel(‘Transfer Entropy’)
>>> plt.title(‘TE vs Time Delay’)
>>> plt.grid(True)
Clinical Significance
———————
- **Short delays (1-3) (*** Immediate physiological responses*)
- **Medium delays (5-10) (*** Regulatory mechanisms*)
- **Long delays (>10) (*** Slow adaptive processes*)

compute_transfer_entropy() → float[source]

Compute transfer entropy from source to target.

Returns:

te (float) – Transfer entropy value in nats
Formula
——-
TE(X→Y) = I(Y_future; X_past | Y_past)
More formally
TE(X→Y) = H(Y_t | Y_past) - H(Y_t | Y_past, X_past)
where
- Y_t = target at time t
- Y_past = k past values of target
- X_past = l past values of source
Algorithm Steps
—————
1. Create embeddings for target history (k values)
2. Create embeddings for source history (l values)
3. Extract future target values
4. Compute conditional mutual information
5. Return TE estimate
Examples
——–
>>> te_analyzer = TransferEntropy(x, y, k=1, l=1)
>>> te_value = te_analyzer.compute_transfer_entropy()
>>>
>>> # Convert nats to bits
>>> te_bits = te_value / np.log(2)
>>> print(f”TE ({te_bits:.4f} bits”))
Clinical Interpretation
———————–
- **Cardio-respiratory (****) –
- Healthy: Moderate bidirectional coupling
- Sleep apnea: Reduced respiratory → cardiac TE
- Heart failure: Altered coupling patterns
- **Brain-heart (****) –
- Mental stress: Increased brain → heart TE
- Relaxation: Reduced directional coupling
Notes
—–
- Returns value in nats (natural logarithm base)
- Convert to bits by dividing by ln(2)
- Significance should be tested with surrogate data

test_significance(n_surrogates: int = 100, method: str = 'shuffle') → Tuple[float, float][source]

Test statistical significance of transfer entropy.

Parameters:

n_surrogates (int) – Number of surrogate datasets
method (str) – Surrogate generation method - ‘shuffle’: Random permutation (destroys temporal structure) - ‘phase’: Phase randomization (preserves power spectrum)

Returns:

p_value (float) – Statistical significance (0-1)
te_original (float) – Original TE value
Algorithm
———
1. Compute TE for original data
2. Generate n_surrogates by shuffling source signal
3. Compute TE for each surrogate
4. p-value = fraction of surrogates with TE >= original TE
Examples
——–
>>> te = TransferEntropy(x, y)
>>> p_value, te_value = te.test_significance(n_surrogates=1000)
>>>
>>> if p_value < 0.05
… print(f”Significant coupling (p={p_value (.4f})”))
>>> else
… print(f”No significant coupling (p={p_value (.4f})”))
Notes
—–
- p < 0.05 (Significant coupling)
- p < 0.01 (Highly significant)
- More surrogates = more reliable p-value
- Computationally expensive for large n_surrogates

Advanced Features Guide

For comprehensive documentation including detailed mathematical formulations, code explanations, clinical validation results, and performance optimization guidelines, see the Advanced Features Guide.

Morphological Analysis

Waveform morphology and shape analysis for physiological signals.

Waveform Morphology

Waveform morphology and shape analysis for physiological signals.

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 Waveform
>>> signal = np.random.randn(1000)
>>> processor = Waveform(signal)
>>> result = processor.process()
>>> print(f'Processing result: {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 WaveformMorphology
>>>
>>> # 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

Cross-Signal Analysis

Multi-signal analysis and relationship detection.

Cross Correlation Analysis

Techniques for analyzing the cross correlation between two physiological signals.

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

Examples:

Basic usage:

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

class vitalDSP.physiological_features.cross_correlation.CrossCorrelationFeatures(signal1, signal2, fs=1000)[source]

Bases: object

A class for computing cross-correlation features between physiological signals (ECG, PPG, EEG).

signal1

The first physiological signal (e.g., ECG).

Type:: np.array

signal2

The second physiological signal (e.g., PPG).

Type:: np.array

fs

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

Type:: int

compute_cross_correlation(mode='full')[source]

Computes the cross-correlation between two physiological signals (e.g., ECG and PPG).

Parameters:: mode (str) – Specifies the mode for cross-correlation. Can be ‘full’, ‘valid’, or ‘same’. Default is ‘full’.
Returns:: The cross-correlation values and the corresponding lag.
Return type:: tuple

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> ppg_signal = [...]  # Sample PPG signal
>>> ccf = CrossCorrelationFeatures(ecg_signal, ppg_signal)
>>> cross_corr, lag = ccf.compute_cross_correlation()
>>> print(f"Cross-Correlation: {cross_corr}, Lag: {lag}")

compute_lag(r_peaks)[source]

Computes the time lag between the ECG signal (R-peaks) and the corresponding PPG signal (foot of the waveform) based on cross-correlation.

Parameters:: r_peaks (np.array) – Indices of R-peaks detected in the ECG signal.
Returns:: The average time lag between the signals in milliseconds.
Return type:: float

Example

>>> r_peaks = [50, 150, 250]  # Detected R-peaks
>>> lag = ccf.compute_lag(r_peaks)
>>> print(f"Average Lag: {lag} ms")

compute_normalized_cross_correlation()[source]

Computes the normalized cross-correlation between two physiological signals to account for differences in amplitude.

Returns:: The normalized cross-correlation values and the corresponding lag.
Return type:: tuple

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> ppg_signal = [...]  # Sample PPG signal
>>> ccf = CrossCorrelationFeatures(ecg_signal, ppg_signal)
>>> norm_corr, lag = ccf.compute_normalized_cross_correlation()
>>> print(f"Normalized Cross-Correlation: {norm_corr}, Lag: {lag}")

compute_pulse_transit_time(r_peaks)[source]

Computes Pulse Transit Time (PTT) as the time difference between the ECG signal (R-peaks) and the corresponding PPG signal (foot of the waveform). PTT is used to estimate blood pressure and vascular health.

Parameters:: r_peaks (np.array) – Indices of R-peaks detected in the ECG signal.
Returns:: The average Pulse Transit Time (PTT) in milliseconds.
Return type:: float

Example

>>> r_peaks = [50, 150, 250]  # Detected R-peaks
>>> ptt = ccf.compute_pulse_transit_time(r_peaks)
>>> print(f"Pulse Transit Time: {ptt} ms")

Cross-Signal Analysis

Techniques for analyzing the relationships between multiple signals over time, useful for studying interactions and dependencies.

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 - Configurable parameters and settings - Comprehensive signal analysis

Examples:

Basic usage:

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

class vitalDSP.physiological_features.cross_signal_analysis.CrossSignalAnalysis(signal1, signal2, fs=1.0)[source]

Bases: object

A comprehensive class to analyze interactions between two physiological signals, such as ECG and respiration.

The class provides several methods to compute various metrics that describe the relationships between the two signals, such as correlation, coherence, phase synchronization, mutual information, and Granger causality.

compute_correlation : function: Computes the Pearson correlation coefficient between two signals.

compute_cross_correlation : function: Computes the cross-correlation between two signals.

compute_coherence : function: Computes the coherence between two signals.

compute_phase_synchronization : function: Computes the phase synchronization index between two signals.

compute_mutual_information : function: Computes the mutual information between two signals.

compute_granger_causality : function: Computes Granger causality between two signals.

compute_coherence(nperseg=256)[source]

Compute the coherence between the two signals.

Coherence is a measure of the degree of synchronization between two signals in the frequency domain. It is often used to determine how much one signal corresponds to another at different frequencies.

Parameters:

nperseg (int, optional) – Length of each segment for coherence computation. Default is 256.

Returns:

freqs (numpy.ndarray) – Frequency array.
coherence (numpy.ndarray) – Coherence between the two signals. Values range from 0 (no coherence) to 1 (complete coherence).

Examples

>>> analysis = CrossSignalAnalysis(signal1, signal2)
>>> freqs, coh = analysis.compute_coherence(nperseg=256)
>>> print(freqs, coh)

compute_correlation()[source]

Compute the Pearson correlation coefficient between the two signals.

The Pearson correlation coefficient is a measure of the linear relationship between two signals.

Returns:: correlation_coefficient – The Pearson correlation coefficient between the two signals. A value close to 1 indicates a strong positive correlation, while a value close to -1 indicates a strong negative correlation.
Return type:: float

Examples

>>> analysis = CrossSignalAnalysis(signal1, signal2)
>>> corr = analysis.compute_correlation()
>>> print(corr)

compute_cross_correlation(max_lag=None)[source]

Compute the cross-correlation between the two signals.

Cross-correlation measures the similarity between two signals as a function of the time-lag applied to one of them. It is useful for identifying time delays between the signals.

Parameters:

max_lag (int or None, optional) – Maximum lag to compute cross-correlation. If None, the full range is used. Default is None.

Returns:

lags (numpy.ndarray) – Array of lags at which the cross-correlation was computed.
cross_corr (numpy.ndarray) – The cross-correlation between the two signals.

Examples

>>> analysis = CrossSignalAnalysis(signal1, signal2)
>>> lags, cross_corr = analysis.compute_cross_correlation(max_lag=50)
>>> print(lags, cross_corr)

compute_granger_causality(max_lag=10)[source]

Compute the Granger causality between the two signals.

Granger causality is a statistical hypothesis test to determine whether one time series can predict another. It does not imply true causality but can indicate predictive relationships between signals.

Parameters:: max_lag (int, optional) – Maximum lag to consider for causality. Default is 10.
Returns:: gc_result – Granger causality results with keys ‘signal1->signal2’ and ‘signal2->signal1’. Each key contains the test statistic and p-value indicating whether one signal Granger-causes the other.
Return type:: dict

Examples

>>> analysis = CrossSignalAnalysis(signal1, signal2)
>>> gc_result = analysis.compute_granger_causality(max_lag=10)
>>> print(gc_result)

compute_mutual_information(bins=10)[source]

Compute the mutual information between the two signals.

Mutual information is a measure of the amount of information obtained about one signal through the other. It captures both linear and non-linear dependencies between the signals.

Parameters:: bins (int, optional) – Number of bins for histogram estimation. Default is 10.
Returns:: mutual_info – The mutual information between the two signals. Higher values indicate greater dependency between the signals.
Return type:: float

Examples

>>> analysis = CrossSignalAnalysis(signal1, signal2)
>>> mi = analysis.compute_mutual_information(bins=10)
>>> print(mi)

compute_phase_synchronization()[source]

Compute the phase synchronization index between the two signals.

Phase synchronization measures the degree to which the phases of the two signals are synchronized over time, regardless of their amplitude. It is useful for analyzing coupling between oscillatory signals.

Returns:: psi – The phase synchronization index (PSI) between the two signals. Values close to 1 indicate strong synchronization.
Return type:: float

Examples

>>> analysis = CrossSignalAnalysis(signal1, signal2)
>>> psi = analysis.compute_phase_synchronization()
>>> print(psi)

Coherence Analysis

Techniques for analyzing the coherence between two physiological signals.

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.coherence_analysis import CoherenceAnalysis
>>> signal = np.random.randn(1000)
>>> processor = CoherenceAnalysis(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.physiological_features.coherence_analysis.CoherenceAnalysis(signal1, signal2, fs=1000)[source]

Bases: object

A class for performing coherence analysis between two physiological signals (e.g., ECG, PPG, respiratory signals), with built-in preprocessing to handle noise, delay, and signal lag.

signal1

The first signal to analyze (e.g., ECG).

Type:: numpy.ndarray

signal2

The second signal to analyze (e.g., PPG).

Type:: numpy.ndarray

fs

The sampling frequency of the signals in Hz.

Type:: int

align_signals(signal1, signal2)[source]

Aligns the two signals by compensating for delay or lag using cross-correlation.

Parameters:

signal1 (numpy.ndarray) – The first signal (preprocessed).
signal2 (numpy.ndarray) – The second signal (preprocessed).

Returns:

aligned_signal1 (numpy.ndarray) – The aligned first signal.
aligned_signal2 (numpy.ndarray) – The aligned second signal.

Examples

>>> ca = CoherenceAnalysis(signal1, signal2, fs=1000)
>>> aligned_signal1, aligned_signal2 = ca.align_signals(preprocessed_signal1, preprocessed_signal2)

compute_coherence(preprocess_config1=None, preprocess_config2=None, nperseg=256)[source]

Computes the coherence between two signals after preprocessing and alignment.

Parameters:

preprocess_config1 (PreprocessConfig, optional) – Preprocessing configuration for the first signal.
preprocess_config2 (PreprocessConfig, optional) – Preprocessing configuration for the second signal.
nperseg (int, optional) – Length of each segment for computing the coherence. Default is 256.

Returns:

f (numpy.ndarray) – Array of sample frequencies.
Cxy (numpy.ndarray) – Coherence between the two signals, ranging from 0 to 1.

Examples

>>> signal1 = np.sin(np.linspace(0, 10, 1000)) + np.random.normal(0, 0.2, 1000)
>>> signal2 = np.sin(np.linspace(0, 10, 1000) + 0.1) + np.random.normal(0, 0.2, 1000)
>>> ca = CoherenceAnalysis(signal1, signal2, fs=1000)
>>> f, Cxy = ca.compute_coherence(nperseg=256)
>>> print(f"Frequencies: {f}")
>>> print(f"Coherence: {Cxy}")

plot_coherence(f, Cxy)[source]

Plots the coherence between two signals.

Parameters:

f (numpy.ndarray) – Array of sample frequencies.
Cxy (numpy.ndarray) – Coherence between the two signals.

Examples

>>> signal1 = np.sin(np.linspace(0, 10, 1000)) + np.random.normal(0, 0.2, 1000)
>>> signal2 = np.sin(np.linspace(0, 10, 1000) + 0.1) + np.random.normal(0, 0.2, 1000)
>>> ca = CoherenceAnalysis(signal1, signal2, fs=1000)
>>> f, Cxy = ca.compute_coherence(nperseg=256)
>>> ca.plot_coherence(f, Cxy)

preprocess_signals(preprocess_config1=None, preprocess_config2=None)[source]

Preprocesses the input signals using filtering, noise reduction, and delay compensation.

Parameters:

preprocess_config1 (PreprocessConfig, optional) – Configuration for preprocessing the first signal (e.g., ECG).
preprocess_config2 (PreprocessConfig, optional) – Configuration for preprocessing the second signal (e.g., PPG).

Returns:

preprocessed_signal1 (numpy.ndarray) – The preprocessed first signal.
preprocessed_signal2 (numpy.ndarray) – The preprocessed second signal.

Examples

>>> ca = CoherenceAnalysis(signal1, signal2, fs=1000)
>>> preprocessed_signal1, preprocessed_signal2 = ca.preprocess_signals(preprocess_config1, preprocess_config2)

Signal Processing Features

Signal segmentation, power analysis, and energy features.

Signal Segmentation

Techniques for segmenting physiological signals into meaningful intervals.

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

Examples:

Basic usage:

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

class vitalDSP.physiological_features.signal_segmentation.SignalSegmentation(signal)[source]

Bases: object

A comprehensive class for segmenting physiological signals.

fixed_size_segmentation : function: Segments the signal into fixed-size segments.

adaptive_segmentation : function: Segments the signal based on adaptive criteria.

threshold_based_segmentation : function: Segments the signal based on a threshold.

variance_based_segmentation : function: Segments the signal based on local variance.

peak_based_segmentation : function: Segments the signal based on detected peaks.

ml_based_segmentation : function: Segments the signal using a machine learning-based approach with multiple default models.

custom_segmentation : function: Allows custom segmentation based on a user-defined function.

adaptive_segmentation(adaptive_fn)[source]

Segment the signal based on adaptive criteria.

Parameters:: adaptive_fn (callable) – A function that defines the adaptive criteria for segmentation. This function should take a segment of the signal as input and return the length of the segment.
Returns:: segments – A list of adaptively segmented parts of the signal.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> def adaptive_fn(segment):
...     return len(segment) // 2  # Example criterion
>>> seg = SignalSegmentation(signal)
>>> adaptive_segments = seg.adaptive_segmentation(adaptive_fn)
>>> print(adaptive_segments)
[array([1, 2, 3, 4]), array([5, 6, 7, 8, 9])]

custom_segmentation(custom_fn)[source]

Segment the signal based on a custom function.

Parameters:: custom_fn (callable) – A user-defined function that returns segment boundaries. The function should take the entire signal as input and return an array of indices where segments start and end.
Returns:: segments – A list of segments based on custom criteria.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> def custom_fn(signal):
...     return np.array([0, 3, 6, 9])
>>> seg = SignalSegmentation(signal)
>>> custom_segments = seg.custom_segmentation(custom_fn)
>>> print(custom_segments)
[array([1, 2, 3]), array([4, 5, 6]), array([7, 8, 9])]

fixed_size_segmentation(segment_size)[source]

Segment the signal into fixed-size segments.

Parameters:: segment_size (int) – The size of each segment.
Returns:: segments – A list of fixed-size segments.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> seg = SignalSegmentation(signal)
>>> fixed_segments = seg.fixed_size_segmentation(3)
>>> print(fixed_segments)
[array([1, 2, 3]), array([4, 5, 6]), array([7, 8, 9])]

ml_based_segmentation(model='change_detection')[source]

Segment the signal using a machine learning-based approach.

Parameters:: model (str, optional) – The name of the default model to use. Options include “change_detection”, “kmeans”, “gmm”, “decision_tree”, “dtw”, “spectral”, “autoencoder”. Default is “change_detection”.
Returns:: segments – A list of segments predicted by the model.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 2, 2, 5, 6, 1, 1, 8, 1])
>>> seg = SignalSegmentation(signal)
>>> ml_segments = seg.ml_based_segmentation(model="kmeans")
>>> print(ml_segments)

peak_based_segmentation(min_distance=50, height=None)[source]

Segment the signal based on detected peaks.

Parameters:

min_distance (int, optional) – Minimum distance between consecutive peaks.
height (float or None, optional) – Minimum height of peaks to be considered.

Returns:

segments – A list of segments around detected peaks.

Return type:

Examples

>>> signal = np.array([1, 2, 1, 2, 1, 2, 1, 8, 1])
>>> seg = SignalSegmentation(signal)
>>> peak_segments = seg.peak_based_segmentation(min_distance=2, height=5)
>>> print(peak_segments)
[array([1, 8, 1])]

threshold_based_segmentation(threshold)[source]

Segment the signal based on a threshold.

Parameters:: threshold (float) – The threshold value for segmentation. The signal is segmented whenever it exceeds this threshold.
Returns:: segments – A list of segments where the signal exceeds the threshold.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 5, 2, 8, 1])
>>> seg = SignalSegmentation(signal)
>>> threshold_segments = seg.threshold_based_segmentation(4)
>>> print(threshold_segments)
[array([5]), array([8])]

variance_based_segmentation(window_size, variance_threshold)[source]

Segment the signal based on local variance.

Parameters:

window_size (int) – The size of the window for variance calculation.
variance_threshold (float) – The threshold of variance to determine segment boundaries.

Returns:

segments – A list of segments based on local variance.

Return type:

Examples

>>> signal = np.array([1, 2, 2, 2, 5, 6, 1, 1, 1, 8])
>>> seg = SignalSegmentation(signal)
>>> variance_segments = seg.variance_based_segmentation(3, 2.0)
>>> print(variance_segments)
[array([1, 2, 2, 2, 5, 6]), array([1, 1, 1, 8])]

Signal Power Analysis

Power analysis and energy features for physiological signals.

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.signal_power_analysis import SignalPowerAnalysis
>>> signal = np.random.randn(1000)
>>> processor = SignalPowerAnalysis(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.physiological_features.signal_power_analysis.SignalPowerAnalysis(signal)[source]

Bases: object

A comprehensive class for measuring the power of physiological signals.

This class provides methods to compute various power-related metrics, including RMSE, mean square value, total power, peak power, SNR, PSD, band power, and signal energy. These metrics are essential for analyzing the characteristics of physiological signals in applications such as ECG, EEG, and PPG analysis.

compute_rmse :: Computes the Root Mean Square Error (RMSE) of the signal.

compute_mean_square :: Computes the mean square value of the signal.

compute_total_power :: Computes the total power of the signal.

compute_peak_power :: Computes the peak power of the signal.

compute_snr :: Computes the Signal-to-Noise Ratio (SNR) of the signal.

compute_psd :: Computes the Power Spectral Density (PSD) of the signal.

compute_band_power :: Computes the power of the signal within a specific frequency band.

compute_energy :: Computes the total energy of the signal.

compute_band_power(band, fs=1.0, nperseg=None)[source]

Compute the power of the signal within a specific frequency band.

Band power is useful in applications like EEG analysis, where power in specific frequency bands (e.g., alpha, beta, delta) is of interest.

Parameters:

band (tuple) – Frequency band as a tuple (low_freq, high_freq).
fs (float, optional) – The sampling frequency of the signal. Default is 1.0.

Returns:

The power within the specified band.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
>>> spa = SignalPowerAnalysis(signal)
>>> band_power = spa.compute_band_power((1, 3), fs=10.0)
>>> print(band_power)

compute_energy()[source]

Compute the total energy of the signal.

The energy of a signal is the sum of the squared amplitudes and is related to the power by the length of the signal.

Returns:: The total energy of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> energy = spa.compute_energy()
>>> print(energy)
55.0

compute_mean_square()[source]

Compute the mean square value of the signal.

The mean square value is a measure of the average power of the signal over time.

Returns:: The mean square value of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> mean_square = spa.compute_mean_square()
>>> print(mean_square)
11.0

compute_peak_power()[source]

Compute the peak power of the signal.

Peak power refers to the maximum instantaneous power, which is the highest value of the squared signal.

Returns:: The peak power of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> peak_power = spa.compute_peak_power()
>>> print(peak_power)
25.0

compute_psd(fs=1.0, nperseg=256)[source]

Compute the Power Spectral Density (PSD) of the signal.

PSD represents the power distribution of a signal as a function of frequency, providing insights into the frequency content of the signal.

Parameters:

fs (float, optional) – The sampling frequency of the signal. Default is 1.0.
nperseg (int, optional) – Length of each segment for PSD computation. Default is 256.

Returns:

freqs (numpy.ndarray) – Array of frequency bins.
psd (numpy.ndarray) – Power Spectral Density values.

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
>>> spa = SignalPowerAnalysis(signal)
>>> freqs, psd = spa.compute_psd(fs=10.0, nperseg=5)
>>> print(freqs)
>>> print(psd)

compute_rmse()[source]

Compute the Root Mean Square Error (RMSE) of the signal.

RMSE is a measure of the magnitude of the signal and is commonly used to assess the variability or “loudness” of a signal.

Returns:: The RMSE of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> rmse = spa.compute_rmse()
>>> print(rmse)
3.3166247903554

compute_snr(noise_signal)[source]

Compute the Signal-to-Noise Ratio (SNR) of the signal.

SNR is a measure of how much the signal power exceeds the noise power, expressed in decibels (dB). A higher SNR indicates a cleaner signal with less noise.

Parameters:: noise_signal (numpy.ndarray) – The noise component of the signal.
Returns:: The SNR of the signal in decibels (dB).
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> noise_signal = np.array([0.1, 0.1, 0.1, 0.1, 0.1])
>>> spa = SignalPowerAnalysis(signal)
>>> snr = spa.compute_snr(noise_signal)
>>> print(snr)
20.0

compute_total_power()[source]

Compute the total power of the signal.

Total power represents the sum of the squared amplitudes normalized by the length of the signal, giving an indication of the overall power content.

Returns:: The total power of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> total_power = spa.compute_total_power()
>>> print(total_power)
11.0

Energy Analysis

Energy analysis and power spectral features for physiological signals.

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 - Comprehensive signal analysis

Examples:

Basic usage:

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

class vitalDSP.physiological_features.energy_analysis.EnergyAnalysis(signal, fs=1000)[source]

Bases: object

A class for computing energy-related features from physiological signals (ECG, PPG, EEG).

signal

The physiological signal (ECG, PPG, EEG).

Type:: np.array

fs

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

Type:: int

compute_band_energy(low_freq, high_freq)[source]

Computes the energy in a specific frequency band for EEG signals.

Parameters:

low_freq (float) – The lower bound of the frequency band (in Hz).
high_freq (float) – The upper bound of the frequency band (in Hz).

Returns:

The energy in the specified frequency band.

Return type:

Example

>>> eeg_signal = [...]  # Sample EEG signal
>>> ea = EnergyAnalysis(eeg_signal)
>>> band_energy = ea.compute_band_energy(8, 12)  # Alpha band (8-12 Hz)
>>> print(f"Band Energy (8-12 Hz): {band_energy}")

compute_qrs_energy(r_peaks)[source]

Computes the energy of the QRS complex in an ECG signal based on detected R-peaks.

Parameters:: r_peaks (np.array) – Indices of R-peaks detected in the ECG signal.
Returns:: The total energy of the QRS complex.
Return type:: float

Example

>>> r_peaks = [50, 150, 250]  # Detected R-peaks
>>> qrs_energy = ea.compute_qrs_energy(r_peaks)
>>> print(f"QRS Energy: {qrs_energy}")

compute_segment_energy(start_idx, end_idx)[source]

Computes the energy of a specific segment of the signal.

Parameters:

start_idx (int) – The starting index of the segment.
end_idx (int) – The ending index of the segment.

Returns:

The energy of the segment.

Return type:

Example

>>> ppg_signal = [...]  # Sample PPG signal
>>> ea = EnergyAnalysis(ppg_signal)
>>> segment_energy = ea.compute_segment_energy(100, 200)
>>> print(f"Segment Energy: {segment_energy}")

compute_spectral_energy()[source]

Computes the total spectral energy of the signal using Welch’s method for power spectral density (PSD).

Returns:: The total spectral energy of the signal.
Return type:: float

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> ea = EnergyAnalysis(ecg_signal)
>>> spectral_energy = ea.compute_spectral_energy()
>>> print(f"Spectral Energy: {spectral_energy}")

compute_systolic_diastolic_energy(systolic_peaks, diastolic_notches)[source]

Computes the energy of systolic and diastolic phases in a PPG signal.

Parameters:

systolic_peaks (np.array) – Indices of systolic peaks.
diastolic_notches (np.array) – Indices of diastolic notches.

Returns:

Systolic and diastolic energy values.

Return type:

tuple

Example

>>> systolic_peaks = [100, 300, 500]
>>> diastolic_notches = [150, 350, 550]
>>> systolic_energy, diastolic_energy = ea.compute_systolic_diastolic_energy(systolic_peaks, diastolic_notches)
>>> print(f"Systolic Energy: {systolic_energy}, Diastolic Energy: {diastolic_energy}")

compute_total_energy()[source]

Computes the total energy of the signal using the sum of squared signal values.

Returns:: The total energy of the signal.
Return type:: float

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> ea = EnergyAnalysis(ecg_signal)
>>> total_energy = ea.compute_total_energy()
>>> print(f"Total Energy: {total_energy}")

Envelope Detection

Envelope detection and amplitude modulation 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 - Pattern and anomaly detection

Examples:

Basic usage:

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

class vitalDSP.physiological_features.envelope_detection.EnvelopeDetection(signal)[source]

Bases: object

A comprehensive class for detecting the envelope of physiological signals.

This class offers several methods to compute the envelope of a given signal using different techniques, ranging from classical methods like the Hilbert transform to more advanced approaches like wavelet transform and machine learning-inspired methods.

hilbert_envelope : function: Computes the envelope using the Hilbert transform.

moving_average_envelope : function: Computes the envelope using a moving average filter.

absolute_value_envelope : function: Computes the envelope by taking the absolute value and applying exponential smoothing.

peak_envelope : function: Computes the envelope by identifying and connecting peaks in the signal.

wavelet_envelope : function: Computes the envelope using wavelet decomposition.

adaptive_filter_envelope : function: Computes the envelope using an adaptive filtering technique.

ml_based_envelope : function: Computes the envelope using a machine learning-inspired approach.

absolute_value_envelope(smoothing_factor=0.01)[source]

Compute the envelope by taking the absolute value and applying exponential smoothing.

This method first computes the absolute value of the signal and then applies an exponential smoothing function to estimate the envelope.

Parameters:: smoothing_factor (float, optional) – The smoothing factor for exponential smoothing, with a typical range between 0 and 1. Smaller values result in more smoothing.
Returns:: envelope – The smoothed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.absolute_value_envelope(0.1)
>>> print(envelope)

adaptive_filter_envelope(step_size=0.01, filter_order=10)[source]

Compute the envelope using adaptive filtering.

Adaptive filtering adjusts its parameters dynamically to track the signal characteristics, making it suitable for real-time envelope detection.

Parameters:

step_size (float, optional) – The step size for the adaptive filter, which controls the speed of adaptation.
filter_order (int, optional) – The order of the adaptive filter, which determines the filter’s complexity.

Returns:

envelope – The computed envelope of the signal.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.adaptive_filter_envelope(0.01, 10)
>>> print(envelope)

hilbert_envelope()[source]

Compute the envelope using the Hilbert transform.

The Hilbert transform is a mathematical operation that produces the analytic signal of a real-valued signal, allowing for the computation of the amplitude envelope.

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

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.hilbert_envelope()
>>> print(envelope)
[1. 2. 3. 4. 5.]

ml_based_envelope(model=None)[source]

Compute the envelope using a machine learning-inspired method.

This method allows the use of a custom machine learning model or function to predict the envelope of the signal.

Parameters:: model (callable or None, optional) – A custom model or function for predicting the envelope. If None, a simple moving average model is used.
Returns:: envelope – The computed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.ml_based_envelope()
>>> print(envelope)

moving_average_envelope(window_size)[source]

Compute the envelope using a moving average filter.

The moving average envelope is computed by taking the absolute value of the signal and then applying a moving average filter, which smooths the signal and provides the envelope.

Parameters:: window_size (int) – The size of the moving average window. A larger window results in a smoother envelope.
Returns:: envelope – The computed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.moving_average_envelope(3)
>>> print(envelope)
[2. 3. 4.]

peak_envelope(interpolation_method='linear')[source]

Compute the envelope by identifying and connecting peaks in the signal.

This method finds local maxima (peaks) in the signal and connects them using interpolation to form the envelope.

Parameters:: interpolation_method (str, optional) – The method to use for interpolation, which can be ‘linear’, ‘quadratic’, or ‘cubic’.
Returns:: envelope – The computed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.peak_envelope('linear')
>>> print(envelope)

wavelet_envelope(wavelet_name='db', level=1)[source]

Compute the envelope using wavelet transform.

This method decomposes the signal using wavelet transform, typically capturing the low-frequency components which represent the envelope.

Parameters:

wavelet_name (str, optional (default="db")) – The type of wavelet to use, such as ‘db’ (Daubechies), ‘haar’, ‘sym’, etc.
level (int, optional (default=1)) – The level of decomposition, which determines the resolution of the envelope.

Returns:

envelope – The computed envelope of the signal.

Return type:

Trend Analysis

Trend analysis and long-term variability measures.

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.trend_analysis import TrendAnalysis
>>> signal = np.random.randn(1000)
>>> processor = TrendAnalysis(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.physiological_features.trend_analysis.TrendAnalysis(signal)[source]

Bases: object

A comprehensive class to track and analyze long-term trends in physiological signals.

This class provides various methods to compute trends, moving averages, momentum, and decompose the signal into its components for in-depth trend analysis.

compute_moving_average : function: Computes the moving average of the signal.

compute_weighted_moving_average : function: Computes the weighted moving average of the signal.

compute_exponential_smoothing : function: Computes exponential smoothing of the signal.

compute_linear_trend : function: Computes the linear trend using least squares regression.

compute_polynomial_trend : function: Computes a polynomial trend.

compute_momentum : function: Computes the momentum of the signal.

compute_seasonal_decomposition : function: Decomposes the signal into trend, seasonal, and residual components.

compute_trend_strength : function: Calculates the strength of the trend.

detect_trend_reversal : function: Detects trend reversal points in the signal.

compute_exponential_smoothing(alpha)[source]

Compute exponential smoothing of the signal.

Exponential smoothing applies a decay factor to smooth the signal, giving more weight to recent observations while maintaining a smooth trend.

Parameters:: alpha (float) – Smoothing factor (0 < alpha <= 1).
Returns:: smoothed_signal – The exponentially smoothed signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_exponential_smoothing(0.3)
array([1., 1.3, 1.81, 2.467, 3.227, 4.059, 4.941])

compute_linear_trend()[source]

Compute the linear trend of the signal using least squares regression.

This method fits a straight line to the signal data, representing the overall direction of the signal over time.

Returns:: linear_trend – The linear trend of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_linear_trend()
array([ 0.6       ,  2.45714286,  4.31428571,  6.17142857,  8.02857143,
        9.88571429, 11.74285714])

compute_momentum(window_size)[source]

Compute the momentum of the signal.

Momentum measures the rate of change of the signal over a specified window size, indicating the strength and direction of the trend.

Parameters:: window_size (int) – The size of the window to calculate momentum.
Returns:: momentum – The momentum of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_momentum(window_size=2)
array([ 0,  0,  1,  2,  3,  5,  8])

compute_moving_average(window_size)[source]

Compute the moving average of the signal.

This method smooths the signal by averaging data points within a moving window.

Parameters:: window_size (int) – The size of the moving average window.
Returns:: moving_avg – The moving average of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_moving_average(3)
array([2., 3., 4., 5., 6.])

compute_polynomial_trend(degree)[source]

Compute a polynomial trend of the signal.

This method fits a polynomial of the specified degree to the signal data, providing a more flexible trend line that can account for curvature in the data.

Parameters:: degree (int) – The degree of the polynomial to fit.
Returns:: polynomial_trend – The polynomial trend of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_polynomial_trend(degree=2)
array([ 1.05714286,  1.84285714,  3.05714286,  4.7       ,  6.77142857,
        9.27142857, 12.2       ])

compute_seasonal_decomposition(period)[source]

Decompose the signal into trend, seasonal, and residual components using seasonal decomposition.

This method separates the signal into its underlying trend, seasonal variations, and random noise components, providing a clearer view of each aspect of the signal.

Parameters:: period (int) – The period of the seasonal component.
Returns:: decomposition – A dictionary containing ‘trend’, ‘seasonal’, and ‘residual’ components.
Return type:: dict

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7])
>>> ta = TrendAnalysis(signal)
>>> decomposition = ta.compute_seasonal_decomposition(period=2)
>>> decomposition['trend']
array([1.5, 2.5, 3.5, 4.5, 5.5, 6.5])
>>> decomposition['seasonal']
array([ 0. ,  0. , -0.5,  0.5, -0.5,  0.5])
>>> decomposition['residual']
array([-0.5,  0.5,  0. ,  0. ,  0. ,  0. ])

compute_trend_strength()[source]

Calculate the strength of the trend in the signal.

Trend strength is determined by comparing the variance of the detrended signal to the variance of the original signal, with a value close to 1 indicating a strong trend.

Returns:: trend_strength – The strength of the trend (between 0 and 1).
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_trend_strength()
0.9571428571428573

compute_weighted_moving_average(window_size, weights=None)[source]

Compute the weighted moving average of the signal.

This method smooths the signal by averaging data points within a moving window, with each point weighted by the given weights.

Parameters:

window_size (int) – The size of the moving average window.
weights (numpy.ndarray or None, optional) – Weights for the moving average. If None, uniform weights are used.

Returns:

weighted_avg – The weighted moving average of the signal.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_weighted_moving_average(3, weights=[0.1, 0.3, 0.6])
array([2.2, 3.2, 4.2, 5.2, 6.2])

detect_trend_reversal(window_size)[source]

Detect trend reversal points in the signal.

This method identifies points in the signal where the trend changes direction, which can be critical in understanding shifts in physiological signals.

Parameters:: window_size (int) – The window size to evaluate trend reversals.
Returns:: reversals – Indices of the detected trend reversal points.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 2, 1, 2, 3])
>>> ta = TrendAnalysis(signal)
>>> ta.detect_trend_reversal(window_size=2)
array([2, 4])

Signal Change Detection

Change point detection and signal transition 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 - Pattern and anomaly detection

Examples:

Basic usage:

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

class vitalDSP.physiological_features.signal_change_detection.SignalChangeDetection(signal)[source]

Bases: object

A comprehensive class for detecting changes in physiological signals.

This class provides multiple methods to analyze physiological signals and detect significant changes based on various criteria such as zero crossings, variance, energy levels, adaptive thresholds, and machine learning-inspired techniques.

zero_crossing_rate : function: Computes the Zero Crossing Rate (ZCR) of the signal.

absolute_difference : function: Computes the absolute difference between consecutive samples.

variance_based_detection : function: Detects signal changes based on local variance.

energy_based_detection : function: Detects signal changes based on local energy.

adaptive_threshold_detection : function: Detects signal changes using an adaptive threshold.

ml_based_change_detection : function: Detects signal changes using a machine learning-inspired method.

absolute_difference()[source]

Compute the absolute difference between consecutive samples.

This method highlights changes between successive signal values, which can be useful for detecting sudden changes or anomalies. It is a simple yet effective technique to measure the magnitude of change in the signal.

Returns:: abs_diff – The absolute differences of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 4, 7, 11])
>>> scd = SignalChangeDetection(signal)
>>> abs_diff = scd.absolute_difference()
>>> print(abs_diff)
[1 2 3 4]

adaptive_threshold_detection(threshold_factor=1.5, window_size=10)[source]

Detect signal changes using an adaptive threshold based on local statistics.

This method calculates local statistics (mean and standard deviation) over a sliding window and identifies signal changes where the deviation from the local mean exceeds an adaptive threshold. This approach is effective for detecting outliers or anomalies in signals with varying baseline levels.

Parameters:

threshold_factor (float, optional) – Factor to multiply the local standard deviation to determine the threshold. Default is 1.5.
window_size (int, optional) – The size of the window to compute local statistics. Default is 10.

Returns:

signal_changes – Indices of detected signal changes.

Return type:

Examples

>>> signal = np.array([1, 2, 1, 2, 1, 2, 10, 2, 1])
>>> scd = SignalChangeDetection(signal)
>>> changes = scd.adaptive_threshold_detection(threshold_factor=2.0, window_size=3)
>>> print(changes)
[6]

energy_based_detection(window_size)[source]

Detect signal changes based on local energy.

This method calculates the energy within a sliding window over the signal. Energy is calculated as the sum of the squares of the signal values within the window. High energy may indicate periods of significant activity or events in the signal, making this method useful in detecting bursts of activity.

Parameters:: window_size (int) – The size of the window to compute energy.
Returns:: energies – The local energy of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> scd = SignalChangeDetection(signal)
>>> energies = scd.energy_based_detection(window_size=3)
>>> print(energies)
[14 29 50]

ml_based_change_detection(model=None)[source]

Detect signal changes using a machine learning-inspired method.

This method allows the use of a custom model or function to detect change points in the signal. If no model is provided, a simple thresholding method based on the absolute difference between consecutive samples is used by default. This approach is flexible and can be extended with sophisticated models like decision trees, neural networks, or clustering algorithms for more complex change detection tasks.

Parameters:: model (callable or None, optional) – A custom model or function for predicting signal changes. The model should take the signal as input and return an array of change points. If None, a default thresholding method is used. Default is None.
Returns:: change_points – Predicted change points in the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 1, 2, 5, 10, 1, 2])
>>> scd = SignalChangeDetection(signal)
>>> changes = scd.ml_based_change_detection()
>>> print(changes)
[4 5]

variance_based_detection(window_size)[source]

Detect signal changes based on local variance.

This method computes the variance within a sliding window over the signal. High variance may indicate areas of the signal with significant changes or noise, which is useful in detecting regions with high activity or instability.

Parameters:: window_size (int) – The size of the window to compute variance.
Returns:: variances – The local variance of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 2, 3, 5, 8, 13, 21])
>>> scd = SignalChangeDetection(signal)
>>> variances = scd.variance_based_detection(window_size=3)
>>> print(variances)
[0.33333333 0.33333333 2.33333333 6.33333333 13.33333333]

zero_crossing_rate()[source]

Compute the Zero Crossing Rate (ZCR) of the signal.

The Zero Crossing Rate is the rate at which the signal changes sign, which can indicate changes in the signal’s behavior or underlying processes. This metric is often used in speech and audio signal processing to detect changes in frequency or amplitude.

Returns:: zcr – The Zero Crossing Rate of the signal.
Return type:: float

Examples

>>> signal = np.array([1, -1, 1, -1, 1])
>>> scd = SignalChangeDetection(signal)
>>> zcr = scd.zero_crossing_rate()
>>> print(zcr)
0.8

Ensemble-Based Feature Extraction

Extract features from signals using ensemble methods, which aggregate multiple models or analyses to improve robustness and accuracy.

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 - Feature extraction capabilities

Examples:

Basic usage:

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

class vitalDSP.physiological_features.ensemble_based_feature_extraction.EnsembleBasedFeatureExtraction(n_estimators=100, max_depth=None, min_samples_split=2)[source]

Bases: object

A comprehensive class for feature extraction using ensemble methods such as Random Forest, Bagging, Boosting, and Stacking.

This class implements various ensemble techniques to extract features from data, enhancing the representational capacity of the models and improving prediction accuracy.

random_forest_features : function: Extracts features using a custom Random Forest.

bagging_features : function: Extracts features using a Bagging ensemble.

boosting_features : function: Extracts features using a Boosting ensemble.

stacking_features : function: Extracts features using a Stacking ensemble.

bagging_features(X, y)[source]

Extract features using a Bagging ensemble.

Bagging (Bootstrap Aggregating) builds multiple trees on different subsets of the data, each generated by random sampling with replacement, and aggregates their predictions.

Parameters:

X (numpy.ndarray) – The input features with shape (n_samples, n_features).
y (numpy.ndarray) – The target labels with shape (n_samples,).

Returns:

aggregated_predictions – The aggregated predictions from the Bagging ensemble, with shape (n_samples,). This represents the averaged output of all trees.

Return type:

Examples

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> y = np.array([0, 1, 0])
>>> extractor = EnsembleBasedFeatureExtraction()
>>> features = extractor.bagging_features(X, y)
>>> print(features)

boosting_features(X, y, learning_rate=0.1)[source]

Extract features using a Boosting ensemble.

Boosting builds trees sequentially, each one trying to correct the errors of the previous one. The final prediction is a weighted sum of the predictions from all trees.

Parameters:

X (numpy.ndarray) – The input features with shape (n_samples, n_features).
y (numpy.ndarray) – The target labels with shape (n_samples,).
learning_rate (float, optional) – The learning rate for boosting, controlling the contribution of each tree. Default is 0.1.

Returns:

predictions – The extracted features from the Boosting ensemble, with shape (n_samples,). This represents the cumulative prediction after all boosting iterations.

Return type:

Examples

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> y = np.array([0, 1, 0])
>>> extractor = EnsembleBasedFeatureExtraction()
>>> features = extractor.boosting_features(X, y, learning_rate=0.1)
>>> print(features)

random_forest_features(X, y)[source]

Extract features using a custom Random Forest.

This method builds multiple decision trees and aggregates their predictions to form a feature representation for each sample.

Parameters:

X (numpy.ndarray) – The input features with shape (n_samples, n_features).
y (numpy.ndarray) – The target labels with shape (n_samples,).

Returns:

features – The extracted features from the Random Forest, with shape (n_samples, n_estimators). Each feature corresponds to the prediction of a single tree in the forest.

Return type:

Examples

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> y = np.array([0, 1, 0])
>>> extractor = EnsembleBasedFeatureExtraction()
>>> features = extractor.random_forest_features(X, y)
>>> print(features)

stacking_features(X, y, meta_model=None)[source]

Extract features using a Stacking ensemble.

Stacking combines the predictions of multiple base models using a meta-model. The base models are first trained independently, and their predictions are used as inputs to the meta-model.

Parameters:

X (numpy.ndarray) – The input features with shape (n_samples, n_features).
y (numpy.ndarray) – The target labels with shape (n_samples,).
meta_model (function or None, optional) – The meta-model used to aggregate the base models’ predictions. If None, a simple average is used. Default is None.

Returns:

stacked_features – The extracted features from the Stacking ensemble, with shape (n_samples,). These features are the output of the meta-model applied to the base models’ predictions.

Return type:

Examples

>>> X = np.array([[1, 2], [3, 4], [5, 6]])
>>> y = np.array([0, 1, 0])
>>> extractor = EnsembleBasedFeatureExtraction()
>>> features = extractor.stacking_features(X, y)
>>> print(features)

Energy Analysis

Techniques for analyzing the energy-related features from physiological signals (ECG, PPG, EEG).

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 - Comprehensive signal analysis

Examples:

Basic usage:

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

class vitalDSP.physiological_features.energy_analysis.EnergyAnalysis(signal, fs=1000)[source]

Bases: object

A class for computing energy-related features from physiological signals (ECG, PPG, EEG).

signal

The physiological signal (ECG, PPG, EEG).

Type:: np.array

fs

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

Type:: int

compute_band_energy(low_freq, high_freq)[source]

Computes the energy in a specific frequency band for EEG signals.

Parameters:

low_freq (float) – The lower bound of the frequency band (in Hz).
high_freq (float) – The upper bound of the frequency band (in Hz).

Returns:

The energy in the specified frequency band.

Return type:

Example

>>> eeg_signal = [...]  # Sample EEG signal
>>> ea = EnergyAnalysis(eeg_signal)
>>> band_energy = ea.compute_band_energy(8, 12)  # Alpha band (8-12 Hz)
>>> print(f"Band Energy (8-12 Hz): {band_energy}")

compute_qrs_energy(r_peaks)[source]

Computes the energy of the QRS complex in an ECG signal based on detected R-peaks.

Parameters:: r_peaks (np.array) – Indices of R-peaks detected in the ECG signal.
Returns:: The total energy of the QRS complex.
Return type:: float

Example

>>> r_peaks = [50, 150, 250]  # Detected R-peaks
>>> qrs_energy = ea.compute_qrs_energy(r_peaks)
>>> print(f"QRS Energy: {qrs_energy}")

compute_segment_energy(start_idx, end_idx)[source]

Computes the energy of a specific segment of the signal.

Parameters:

start_idx (int) – The starting index of the segment.
end_idx (int) – The ending index of the segment.

Returns:

The energy of the segment.

Return type:

Example

>>> ppg_signal = [...]  # Sample PPG signal
>>> ea = EnergyAnalysis(ppg_signal)
>>> segment_energy = ea.compute_segment_energy(100, 200)
>>> print(f"Segment Energy: {segment_energy}")

compute_spectral_energy()[source]

Computes the total spectral energy of the signal using Welch’s method for power spectral density (PSD).

Returns:: The total spectral energy of the signal.
Return type:: float

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> ea = EnergyAnalysis(ecg_signal)
>>> spectral_energy = ea.compute_spectral_energy()
>>> print(f"Spectral Energy: {spectral_energy}")

compute_systolic_diastolic_energy(systolic_peaks, diastolic_notches)[source]

Computes the energy of systolic and diastolic phases in a PPG signal.

Parameters:

systolic_peaks (np.array) – Indices of systolic peaks.
diastolic_notches (np.array) – Indices of diastolic notches.

Returns:

Systolic and diastolic energy values.

Return type:

tuple

Example

>>> systolic_peaks = [100, 300, 500]
>>> diastolic_notches = [150, 350, 550]
>>> systolic_energy, diastolic_energy = ea.compute_systolic_diastolic_energy(systolic_peaks, diastolic_notches)
>>> print(f"Systolic Energy: {systolic_energy}, Diastolic Energy: {diastolic_energy}")

compute_total_energy()[source]

Computes the total energy of the signal using the sum of squared signal values.

Returns:: The total energy of the signal.
Return type:: float

Example

>>> ecg_signal = [...]  # Sample ECG signal
>>> ea = EnergyAnalysis(ecg_signal)
>>> total_energy = ea.compute_total_energy()
>>> print(f"Total Energy: {total_energy}")

Envelope Detection

Methods for detecting the envelope of a signal, which captures the signal’s amplitude variations over time, often used in audio and biomedical signal processing.

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 - Pattern and anomaly detection

Examples:

Basic usage:

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

class vitalDSP.physiological_features.envelope_detection.EnvelopeDetection(signal)[source]

Bases: object

A comprehensive class for detecting the envelope of physiological signals.

This class offers several methods to compute the envelope of a given signal using different techniques, ranging from classical methods like the Hilbert transform to more advanced approaches like wavelet transform and machine learning-inspired methods.

hilbert_envelope : function: Computes the envelope using the Hilbert transform.

moving_average_envelope : function: Computes the envelope using a moving average filter.

absolute_value_envelope : function: Computes the envelope by taking the absolute value and applying exponential smoothing.

peak_envelope : function: Computes the envelope by identifying and connecting peaks in the signal.

wavelet_envelope : function: Computes the envelope using wavelet decomposition.

adaptive_filter_envelope : function: Computes the envelope using an adaptive filtering technique.

ml_based_envelope : function: Computes the envelope using a machine learning-inspired approach.

absolute_value_envelope(smoothing_factor=0.01)[source]

Compute the envelope by taking the absolute value and applying exponential smoothing.

This method first computes the absolute value of the signal and then applies an exponential smoothing function to estimate the envelope.

Parameters:: smoothing_factor (float, optional) – The smoothing factor for exponential smoothing, with a typical range between 0 and 1. Smaller values result in more smoothing.
Returns:: envelope – The smoothed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.absolute_value_envelope(0.1)
>>> print(envelope)

adaptive_filter_envelope(step_size=0.01, filter_order=10)[source]

Compute the envelope using adaptive filtering.

Adaptive filtering adjusts its parameters dynamically to track the signal characteristics, making it suitable for real-time envelope detection.

Parameters:

step_size (float, optional) – The step size for the adaptive filter, which controls the speed of adaptation.
filter_order (int, optional) – The order of the adaptive filter, which determines the filter’s complexity.

Returns:

envelope – The computed envelope of the signal.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.adaptive_filter_envelope(0.01, 10)
>>> print(envelope)

hilbert_envelope()[source]

Compute the envelope using the Hilbert transform.

The Hilbert transform is a mathematical operation that produces the analytic signal of a real-valued signal, allowing for the computation of the amplitude envelope.

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

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.hilbert_envelope()
>>> print(envelope)
[1. 2. 3. 4. 5.]

ml_based_envelope(model=None)[source]

Compute the envelope using a machine learning-inspired method.

This method allows the use of a custom machine learning model or function to predict the envelope of the signal.

Parameters:: model (callable or None, optional) – A custom model or function for predicting the envelope. If None, a simple moving average model is used.
Returns:: envelope – The computed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.ml_based_envelope()
>>> print(envelope)

moving_average_envelope(window_size)[source]

Compute the envelope using a moving average filter.

The moving average envelope is computed by taking the absolute value of the signal and then applying a moving average filter, which smooths the signal and provides the envelope.

Parameters:: window_size (int) – The size of the moving average window. A larger window results in a smoother envelope.
Returns:: envelope – The computed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.moving_average_envelope(3)
>>> print(envelope)
[2. 3. 4.]

peak_envelope(interpolation_method='linear')[source]

Compute the envelope by identifying and connecting peaks in the signal.

This method finds local maxima (peaks) in the signal and connects them using interpolation to form the envelope.

Parameters:: interpolation_method (str, optional) – The method to use for interpolation, which can be ‘linear’, ‘quadratic’, or ‘cubic’.
Returns:: envelope – The computed envelope of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> ed = EnvelopeDetection(signal)
>>> envelope = ed.peak_envelope('linear')
>>> print(envelope)

wavelet_envelope(wavelet_name='db', level=1)[source]

Compute the envelope using wavelet transform.

This method decomposes the signal using wavelet transform, typically capturing the low-frequency components which represent the envelope.

Parameters:

wavelet_name (str, optional (default="db")) – The type of wavelet to use, such as ‘db’ (Daubechies), ‘haar’, ‘sym’, etc.
level (int, optional (default=1)) – The level of decomposition, which determines the resolution of the envelope.

Returns:

envelope – The computed envelope of the signal.

Return type:

Frequency Domain

Techniques for analyzing the frequency domain features from physiological signals (ECG, PPG, EEG).

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}")

Nonlinear Analysis

Techniques for analyzing the nonlinear features from physiological signals (ECG, PPG, EEG).

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 Nonlinear
>>> signal = np.random.randn(1000)
>>> processor = Nonlinear(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

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}")

Signal Change Detection

Detect abrupt changes in signals, which may signify transitions between different physiological states or responses.

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 - Pattern and anomaly detection

Examples:

Basic usage:

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

class vitalDSP.physiological_features.signal_change_detection.SignalChangeDetection(signal)[source]

Bases: object

A comprehensive class for detecting changes in physiological signals.

This class provides multiple methods to analyze physiological signals and detect significant changes based on various criteria such as zero crossings, variance, energy levels, adaptive thresholds, and machine learning-inspired techniques.

zero_crossing_rate : function: Computes the Zero Crossing Rate (ZCR) of the signal.

absolute_difference : function: Computes the absolute difference between consecutive samples.

variance_based_detection : function: Detects signal changes based on local variance.

energy_based_detection : function: Detects signal changes based on local energy.

adaptive_threshold_detection : function: Detects signal changes using an adaptive threshold.

ml_based_change_detection : function: Detects signal changes using a machine learning-inspired method.

absolute_difference()[source]

Compute the absolute difference between consecutive samples.

This method highlights changes between successive signal values, which can be useful for detecting sudden changes or anomalies. It is a simple yet effective technique to measure the magnitude of change in the signal.

Returns:: abs_diff – The absolute differences of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 4, 7, 11])
>>> scd = SignalChangeDetection(signal)
>>> abs_diff = scd.absolute_difference()
>>> print(abs_diff)
[1 2 3 4]

adaptive_threshold_detection(threshold_factor=1.5, window_size=10)[source]

Detect signal changes using an adaptive threshold based on local statistics.

This method calculates local statistics (mean and standard deviation) over a sliding window and identifies signal changes where the deviation from the local mean exceeds an adaptive threshold. This approach is effective for detecting outliers or anomalies in signals with varying baseline levels.

Parameters:

threshold_factor (float, optional) – Factor to multiply the local standard deviation to determine the threshold. Default is 1.5.
window_size (int, optional) – The size of the window to compute local statistics. Default is 10.

Returns:

signal_changes – Indices of detected signal changes.

Return type:

Examples

>>> signal = np.array([1, 2, 1, 2, 1, 2, 10, 2, 1])
>>> scd = SignalChangeDetection(signal)
>>> changes = scd.adaptive_threshold_detection(threshold_factor=2.0, window_size=3)
>>> print(changes)
[6]

energy_based_detection(window_size)[source]

Detect signal changes based on local energy.

This method calculates the energy within a sliding window over the signal. Energy is calculated as the sum of the squares of the signal values within the window. High energy may indicate periods of significant activity or events in the signal, making this method useful in detecting bursts of activity.

Parameters:: window_size (int) – The size of the window to compute energy.
Returns:: energies – The local energy of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> scd = SignalChangeDetection(signal)
>>> energies = scd.energy_based_detection(window_size=3)
>>> print(energies)
[14 29 50]

ml_based_change_detection(model=None)[source]

Detect signal changes using a machine learning-inspired method.

This method allows the use of a custom model or function to detect change points in the signal. If no model is provided, a simple thresholding method based on the absolute difference between consecutive samples is used by default. This approach is flexible and can be extended with sophisticated models like decision trees, neural networks, or clustering algorithms for more complex change detection tasks.

Parameters:: model (callable or None, optional) – A custom model or function for predicting signal changes. The model should take the signal as input and return an array of change points. If None, a default thresholding method is used. Default is None.
Returns:: change_points – Predicted change points in the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 1, 2, 5, 10, 1, 2])
>>> scd = SignalChangeDetection(signal)
>>> changes = scd.ml_based_change_detection()
>>> print(changes)
[4 5]

variance_based_detection(window_size)[source]

Detect signal changes based on local variance.

This method computes the variance within a sliding window over the signal. High variance may indicate areas of the signal with significant changes or noise, which is useful in detecting regions with high activity or instability.

Parameters:: window_size (int) – The size of the window to compute variance.
Returns:: variances – The local variance of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 2, 3, 5, 8, 13, 21])
>>> scd = SignalChangeDetection(signal)
>>> variances = scd.variance_based_detection(window_size=3)
>>> print(variances)
[0.33333333 0.33333333 2.33333333 6.33333333 13.33333333]

zero_crossing_rate()[source]

Compute the Zero Crossing Rate (ZCR) of the signal.

The Zero Crossing Rate is the rate at which the signal changes sign, which can indicate changes in the signal’s behavior or underlying processes. This metric is often used in speech and audio signal processing to detect changes in frequency or amplitude.

Returns:: zcr – The Zero Crossing Rate of the signal.
Return type:: float

Examples

>>> signal = np.array([1, -1, 1, -1, 1])
>>> scd = SignalChangeDetection(signal)
>>> zcr = scd.zero_crossing_rate()
>>> print(zcr)
0.8

Signal Power Analysis

Analyze the power of a signal over time, which can provide insights into the energy distribution and overall activity within the signal.

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.signal_power_analysis import SignalPowerAnalysis
>>> signal = np.random.randn(1000)
>>> processor = SignalPowerAnalysis(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.physiological_features.signal_power_analysis.SignalPowerAnalysis(signal)[source]

Bases: object

A comprehensive class for measuring the power of physiological signals.

This class provides methods to compute various power-related metrics, including RMSE, mean square value, total power, peak power, SNR, PSD, band power, and signal energy. These metrics are essential for analyzing the characteristics of physiological signals in applications such as ECG, EEG, and PPG analysis.

compute_rmse :: Computes the Root Mean Square Error (RMSE) of the signal.

compute_mean_square :: Computes the mean square value of the signal.

compute_total_power :: Computes the total power of the signal.

compute_peak_power :: Computes the peak power of the signal.

compute_snr :: Computes the Signal-to-Noise Ratio (SNR) of the signal.

compute_psd :: Computes the Power Spectral Density (PSD) of the signal.

compute_band_power :: Computes the power of the signal within a specific frequency band.

compute_energy :: Computes the total energy of the signal.

compute_band_power(band, fs=1.0, nperseg=None)[source]

Compute the power of the signal within a specific frequency band.

Band power is useful in applications like EEG analysis, where power in specific frequency bands (e.g., alpha, beta, delta) is of interest.

Parameters:

band (tuple) – Frequency band as a tuple (low_freq, high_freq).
fs (float, optional) – The sampling frequency of the signal. Default is 1.0.

Returns:

The power within the specified band.

Return type:

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
>>> spa = SignalPowerAnalysis(signal)
>>> band_power = spa.compute_band_power((1, 3), fs=10.0)
>>> print(band_power)

compute_energy()[source]

Compute the total energy of the signal.

The energy of a signal is the sum of the squared amplitudes and is related to the power by the length of the signal.

Returns:: The total energy of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> energy = spa.compute_energy()
>>> print(energy)
55.0

compute_mean_square()[source]

Compute the mean square value of the signal.

The mean square value is a measure of the average power of the signal over time.

Returns:: The mean square value of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> mean_square = spa.compute_mean_square()
>>> print(mean_square)
11.0

compute_peak_power()[source]

Compute the peak power of the signal.

Peak power refers to the maximum instantaneous power, which is the highest value of the squared signal.

Returns:: The peak power of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> peak_power = spa.compute_peak_power()
>>> print(peak_power)
25.0

compute_psd(fs=1.0, nperseg=256)[source]

Compute the Power Spectral Density (PSD) of the signal.

PSD represents the power distribution of a signal as a function of frequency, providing insights into the frequency content of the signal.

Parameters:

fs (float, optional) – The sampling frequency of the signal. Default is 1.0.
nperseg (int, optional) – Length of each segment for PSD computation. Default is 256.

Returns:

freqs (numpy.ndarray) – Array of frequency bins.
psd (numpy.ndarray) – Power Spectral Density values.

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
>>> spa = SignalPowerAnalysis(signal)
>>> freqs, psd = spa.compute_psd(fs=10.0, nperseg=5)
>>> print(freqs)
>>> print(psd)

compute_rmse()[source]

Compute the Root Mean Square Error (RMSE) of the signal.

RMSE is a measure of the magnitude of the signal and is commonly used to assess the variability or “loudness” of a signal.

Returns:: The RMSE of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> rmse = spa.compute_rmse()
>>> print(rmse)
3.3166247903554

compute_snr(noise_signal)[source]

Compute the Signal-to-Noise Ratio (SNR) of the signal.

SNR is a measure of how much the signal power exceeds the noise power, expressed in decibels (dB). A higher SNR indicates a cleaner signal with less noise.

Parameters:: noise_signal (numpy.ndarray) – The noise component of the signal.
Returns:: The SNR of the signal in decibels (dB).
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> noise_signal = np.array([0.1, 0.1, 0.1, 0.1, 0.1])
>>> spa = SignalPowerAnalysis(signal)
>>> snr = spa.compute_snr(noise_signal)
>>> print(snr)
20.0

compute_total_power()[source]

Compute the total power of the signal.

Total power represents the sum of the squared amplitudes normalized by the length of the signal, giving an indication of the overall power content.

Returns:: The total power of the signal.
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 4, 5])
>>> spa = SignalPowerAnalysis(signal)
>>> total_power = spa.compute_total_power()
>>> print(total_power)
11.0

Signal Segmentation

Segment signals into meaningful parts for further analysis, a crucial step in processing long-duration biomedical signals.

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

Examples:

Basic usage:

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

class vitalDSP.physiological_features.signal_segmentation.SignalSegmentation(signal)[source]

Bases: object

A comprehensive class for segmenting physiological signals.

fixed_size_segmentation : function: Segments the signal into fixed-size segments.

adaptive_segmentation : function: Segments the signal based on adaptive criteria.

threshold_based_segmentation : function: Segments the signal based on a threshold.

variance_based_segmentation : function: Segments the signal based on local variance.

peak_based_segmentation : function: Segments the signal based on detected peaks.

ml_based_segmentation : function: Segments the signal using a machine learning-based approach with multiple default models.

custom_segmentation : function: Allows custom segmentation based on a user-defined function.

adaptive_segmentation(adaptive_fn)[source]

Segment the signal based on adaptive criteria.

Parameters:: adaptive_fn (callable) – A function that defines the adaptive criteria for segmentation. This function should take a segment of the signal as input and return the length of the segment.
Returns:: segments – A list of adaptively segmented parts of the signal.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> def adaptive_fn(segment):
...     return len(segment) // 2  # Example criterion
>>> seg = SignalSegmentation(signal)
>>> adaptive_segments = seg.adaptive_segmentation(adaptive_fn)
>>> print(adaptive_segments)
[array([1, 2, 3, 4]), array([5, 6, 7, 8, 9])]

custom_segmentation(custom_fn)[source]

Segment the signal based on a custom function.

Parameters:: custom_fn (callable) – A user-defined function that returns segment boundaries. The function should take the entire signal as input and return an array of indices where segments start and end.
Returns:: segments – A list of segments based on custom criteria.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> def custom_fn(signal):
...     return np.array([0, 3, 6, 9])
>>> seg = SignalSegmentation(signal)
>>> custom_segments = seg.custom_segmentation(custom_fn)
>>> print(custom_segments)
[array([1, 2, 3]), array([4, 5, 6]), array([7, 8, 9])]

fixed_size_segmentation(segment_size)[source]

Segment the signal into fixed-size segments.

Parameters:: segment_size (int) – The size of each segment.
Returns:: segments – A list of fixed-size segments.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
>>> seg = SignalSegmentation(signal)
>>> fixed_segments = seg.fixed_size_segmentation(3)
>>> print(fixed_segments)
[array([1, 2, 3]), array([4, 5, 6]), array([7, 8, 9])]

ml_based_segmentation(model='change_detection')[source]

Segment the signal using a machine learning-based approach.

Parameters:: model (str, optional) – The name of the default model to use. Options include “change_detection”, “kmeans”, “gmm”, “decision_tree”, “dtw”, “spectral”, “autoencoder”. Default is “change_detection”.
Returns:: segments – A list of segments predicted by the model.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 2, 2, 5, 6, 1, 1, 8, 1])
>>> seg = SignalSegmentation(signal)
>>> ml_segments = seg.ml_based_segmentation(model="kmeans")
>>> print(ml_segments)

peak_based_segmentation(min_distance=50, height=None)[source]

Segment the signal based on detected peaks.

Parameters:

min_distance (int, optional) – Minimum distance between consecutive peaks.
height (float or None, optional) – Minimum height of peaks to be considered.

Returns:

segments – A list of segments around detected peaks.

Return type:

Examples

>>> signal = np.array([1, 2, 1, 2, 1, 2, 1, 8, 1])
>>> seg = SignalSegmentation(signal)
>>> peak_segments = seg.peak_based_segmentation(min_distance=2, height=5)
>>> print(peak_segments)
[array([1, 8, 1])]

threshold_based_segmentation(threshold)[source]

Segment the signal based on a threshold.

Parameters:: threshold (float) – The threshold value for segmentation. The signal is segmented whenever it exceeds this threshold.
Returns:: segments – A list of segments where the signal exceeds the threshold.
Return type:: list of numpy.ndarray

Examples

>>> signal = np.array([1, 2, 5, 2, 8, 1])
>>> seg = SignalSegmentation(signal)
>>> threshold_segments = seg.threshold_based_segmentation(4)
>>> print(threshold_segments)
[array([5]), array([8])]

variance_based_segmentation(window_size, variance_threshold)[source]

Segment the signal based on local variance.

Parameters:

window_size (int) – The size of the window for variance calculation.
variance_threshold (float) – The threshold of variance to determine segment boundaries.

Returns:

segments – A list of segments based on local variance.

Return type:

Examples

>>> signal = np.array([1, 2, 2, 2, 5, 6, 1, 1, 1, 8])
>>> seg = SignalSegmentation(signal)
>>> variance_segments = seg.variance_based_segmentation(3, 2.0)
>>> print(variance_segments)
[array([1, 2, 2, 2, 5, 6]), array([1, 1, 1, 8])]

Time Domain Analysis

Techniques for analyzing the time domain features from physiological signals (ECG, PPG, EEG).

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

Trend Analysis

Analyze trends within signals over time, identifying underlying patterns and long-term behaviors that may be of clinical significance.

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.trend_analysis import TrendAnalysis
>>> signal = np.random.randn(1000)
>>> processor = TrendAnalysis(signal)
>>> result = processor.process()
>>> print(f'Processing result: {result}')

class vitalDSP.physiological_features.trend_analysis.TrendAnalysis(signal)[source]

Bases: object

A comprehensive class to track and analyze long-term trends in physiological signals.

This class provides various methods to compute trends, moving averages, momentum, and decompose the signal into its components for in-depth trend analysis.

compute_moving_average : function: Computes the moving average of the signal.

compute_weighted_moving_average : function: Computes the weighted moving average of the signal.

compute_exponential_smoothing : function: Computes exponential smoothing of the signal.

compute_linear_trend : function: Computes the linear trend using least squares regression.

compute_polynomial_trend : function: Computes a polynomial trend.

compute_momentum : function: Computes the momentum of the signal.

compute_seasonal_decomposition : function: Decomposes the signal into trend, seasonal, and residual components.

compute_trend_strength : function: Calculates the strength of the trend.

detect_trend_reversal : function: Detects trend reversal points in the signal.

compute_exponential_smoothing(alpha)[source]

Compute exponential smoothing of the signal.

Exponential smoothing applies a decay factor to smooth the signal, giving more weight to recent observations while maintaining a smooth trend.

Parameters:: alpha (float) – Smoothing factor (0 < alpha <= 1).
Returns:: smoothed_signal – The exponentially smoothed signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_exponential_smoothing(0.3)
array([1., 1.3, 1.81, 2.467, 3.227, 4.059, 4.941])

compute_linear_trend()[source]

Compute the linear trend of the signal using least squares regression.

This method fits a straight line to the signal data, representing the overall direction of the signal over time.

Returns:: linear_trend – The linear trend of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_linear_trend()
array([ 0.6       ,  2.45714286,  4.31428571,  6.17142857,  8.02857143,
        9.88571429, 11.74285714])

compute_momentum(window_size)[source]

Compute the momentum of the signal.

Momentum measures the rate of change of the signal over a specified window size, indicating the strength and direction of the trend.

Parameters:: window_size (int) – The size of the window to calculate momentum.
Returns:: momentum – The momentum of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_momentum(window_size=2)
array([ 0,  0,  1,  2,  3,  5,  8])

compute_moving_average(window_size)[source]

Compute the moving average of the signal.

This method smooths the signal by averaging data points within a moving window.

Parameters:: window_size (int) – The size of the moving average window.
Returns:: moving_avg – The moving average of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_moving_average(3)
array([2., 3., 4., 5., 6.])

compute_polynomial_trend(degree)[source]

Compute a polynomial trend of the signal.

This method fits a polynomial of the specified degree to the signal data, providing a more flexible trend line that can account for curvature in the data.

Parameters:: degree (int) – The degree of the polynomial to fit.
Returns:: polynomial_trend – The polynomial trend of the signal.
Return type:: numpy.ndarray

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_polynomial_trend(degree=2)
array([ 1.05714286,  1.84285714,  3.05714286,  4.7       ,  6.77142857,
        9.27142857, 12.2       ])

compute_seasonal_decomposition(period)[source]

Decompose the signal into trend, seasonal, and residual components using seasonal decomposition.

This method separates the signal into its underlying trend, seasonal variations, and random noise components, providing a clearer view of each aspect of the signal.

Parameters:: period (int) – The period of the seasonal component.
Returns:: decomposition – A dictionary containing ‘trend’, ‘seasonal’, and ‘residual’ components.
Return type:: dict

Examples

>>> signal = np.array([1, 2, 3, 4, 5, 6, 7])
>>> ta = TrendAnalysis(signal)
>>> decomposition = ta.compute_seasonal_decomposition(period=2)
>>> decomposition['trend']
array([1.5, 2.5, 3.5, 4.5, 5.5, 6.5])
>>> decomposition['seasonal']
array([ 0. ,  0. , -0.5,  0.5, -0.5,  0.5])
>>> decomposition['residual']
array([-0.5,  0.5,  0. ,  0. ,  0. ,  0. ])

compute_trend_strength()[source]

Calculate the strength of the trend in the signal.

Trend strength is determined by comparing the variance of the detrended signal to the variance of the original signal, with a value close to 1 indicating a strong trend.

Returns:: trend_strength – The strength of the trend (between 0 and 1).
Return type:: float

Examples

>>> signal = np.array([1, 2, 3, 5, 8, 13, 21])
>>> ta = TrendAnalysis(signal)
>>> ta.compute_trend_strength()
0.9571428571428573

compute_weighted_moving_average(window_size, weights=None)[source]

Compute the weighted moving average of the signal.

This method smooths the signal by averaging data points within a moving window, with each point weighted by the given weights.

Parameters:

window_size (int) – The size of the moving average window.
weights (numpy.ndarray or None, optional) – Weights for the moving average. If None, uniform weights are used.

Returns:

weighted_avg – The weighted moving average of the signal.

Return type: