Respiratory Analysis

This notebook demonstrates respiratory rate estimation using multiple methods and sleep apnea event detection from ECG and PPG signals using vitalDSP.

Setup

import numpy as np
import plotly.io as pio
pio.renderers.default = "sphinx_gallery"
from plotly import graph_objects as go
from vitalDSP.notebooks import load_sample_ecg_small, load_sample_ppg, plot_trace

fs = 128
signal_col, date_col = load_sample_ecg_small()
signal_col = np.array(signal_col)

ppg_col, ppg_date = load_sample_ppg()
ppg_col = np.array(ppg_col)

print(f"ECG signal: {len(signal_col)} samples at {fs} Hz")

ECG signal: 82176 samples at 128 Hz

FFT-Based Respiratory Rate Estimation

from vitalDSP.respiratory_analysis.estimate_rr import fft_based_rr, peak_detection_rr, frequency_domain_rr

# FFT method — finds dominant frequency in the respiratory band (0.1-0.5 Hz)
rr_fft = fft_based_rr(signal_col, sampling_rate=fs)
print(f"Respiratory Rate (FFT)        : {rr_fft:.2f} breaths/min")

Respiratory Rate (FFT)        : 14.30 breaths/min

Peak Detection Respiratory Rate

rr_peaks = peak_detection_rr(signal_col, sampling_rate=fs, min_peak_distance=1.5)
print(f"Respiratory Rate (Peak Detect): {rr_peaks:.2f} breaths/min")

Respiratory Rate (Peak Detect): 37.83 breaths/min

Frequency Domain Respiratory Rate

rr_freq = frequency_domain_rr(signal_col, sampling_rate=fs)
print(f"Respiratory Rate (Freq Domain): {rr_freq:.2f} breaths/min")

WARNING:vitalDSP.respiratory_analysis.estimate_rr.frequency_domain_rr:Low SNR: 1.14

Respiratory Rate (Freq Domain): 15.00 breaths/min

Ensemble Respiratory Rate Estimation

from vitalDSP.respiratory_analysis.respiratory_analysis import RespiratoryAnalysis

ra = RespiratoryAnalysis(signal_col, fs=fs)

# Single method
rr_single = ra.compute_respiratory_rate(method='fft')
print(f"Single Method (FFT): {rr_single:.2f} breaths/min")

# Ensemble: weighted combination of multiple methods
ensemble_result = ra.compute_respiratory_rate_ensemble()

print("\nEnsemble Respiratory Rate Estimation:")
print(f"  Rate       : {ensemble_result['respiratory_rate']:.2f} breaths/min")
print(f"  Confidence : {ensemble_result['confidence']:.3f}")
print(f"  Quality    : {ensemble_result['quality']}")

WARNING:vitalDSP.respiratory_analysis.estimate_rr.frequency_domain_rr:Low SNR: 1.14

Single Method (FFT): 14.30 breaths/min

Ensemble Respiratory Rate Estimation:
  Rate       : 21.17 breaths/min
  Confidence : 0.047
  Quality    : medium

Visualizing the Respiratory Band

from vitalDSP.filtering.signal_filtering import SignalFiltering

# Isolate the respiratory frequency band (0.1-0.5 Hz)
sf = SignalFiltering(signal_col)
resp_band = sf.bandpass(0.1, 0.5, fs, order=4)

t = np.arange(len(signal_col)) / fs
fig = go.Figure()
fig.add_trace(go.Scatter(x=t, y=signal_col, mode='lines', name='ECG Signal', opacity=0.4))
fig.add_trace(go.Scatter(x=t, y=resp_band, mode='lines', name='Respiratory Band (0.1-0.5 Hz)'))
fig.update_layout(
    title='Respiratory Modulation in ECG Signal',
    xaxis_title='Time (s)', yaxis_title='Amplitude'
)
fig.show()

PPG Respiratory Rate

ra_ppg = RespiratoryAnalysis(ppg_col, fs=fs)
rr_ppg = ra_ppg.compute_respiratory_rate(method='fft')
rr_ppg_ensemble = ra_ppg.compute_respiratory_rate_ensemble()

print(f"PPG Respiratory Rate (FFT)      : {rr_ppg:.2f} breaths/min")
print(f"PPG Respiratory Rate (Ensemble) : {rr_ppg_ensemble['respiratory_rate']:.2f} breaths/min")
print(f"PPG Ensemble Confidence         : {rr_ppg_ensemble['confidence']:.3f}")

WARNING:vitalDSP.respiratory_analysis.estimate_rr.peak_detection_rr:Invalid intervals: [223.2578125 215.0390625 102.9140625]

WARNING:vitalDSP.respiratory_analysis.estimate_rr.frequency_domain_rr:Low SNR: 1.91

PPG Respiratory Rate (FFT)      : 7.29 breaths/min
PPG Respiratory Rate (Ensemble) : 9.64 breaths/min
PPG Ensemble Confidence         : 0.000

Sleep Apnea Detection

from vitalDSP.respiratory_analysis.sleep_apnea_detection.pause_detection import detect_apnea_pauses
from vitalDSP.respiratory_analysis.sleep_apnea_detection.amplitude_threshold import detect_apnea_amplitude

# Detect apnea pauses (cessation of respiratory activity)
pause_events = detect_apnea_pauses(signal_col, sampling_rate=fs, min_pause_duration=10)
print(f"Apnea pauses detected     : {len(pause_events)}")

# Detect apnea by amplitude threshold (signal amplitude drops below threshold)
threshold = np.std(signal_col) * 0.3
amplitude_events = detect_apnea_amplitude(signal_col, sampling_rate=fs, threshold=threshold, min_duration=10)
print(f"Amplitude-based events detected: {len(amplitude_events)}")

# Visualize detected events
fig = go.Figure()
t = np.arange(len(signal_col)) / fs
fig.add_trace(go.Scatter(x=t, y=signal_col, mode='lines', name='Signal', opacity=0.6))
for start, end in pause_events[:5]:  # show first 5 events
    fig.add_vrect(x0=start/fs, x1=end/fs, fillcolor='red', opacity=0.2, layer='below', line_width=0)
fig.update_layout(
    title='Sleep Apnea Pause Detection (red = apnea event)',
    xaxis_title='Time (s)', yaxis_title='Amplitude'
)
fig.show()

Apnea pauses detected     : 0
Amplitude-based events detected: 0