Autonomic and ECG/PPG Feature Engineering

This notebook demonstrates ECG morphology analysis, PPG autonomic nervous system features, SpO2 and perfusion index from dual-wavelength PPG, and ECG-PPG synchronization metrics.

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: {len(signal_col)} samples at {fs} Hz")
print(f"PPG: {len(ppg_col)} samples")

ECG Morphology Features

from vitalDSP.feature_engineering.ecg_autonomic_features import ECGExtractor

ecg_extractor = ECGExtractor(signal_col, sampling_frequency=fs)

# Detect R peaks
r_peaks = ecg_extractor.detect_r_peaks()
print(f"R peaks detected: {len(r_peaks)}")

# Compute ECG morphology intervals
qrs_duration = ecg_extractor.compute_qrs_duration()
pr_interval  = ecg_extractor.compute_pr_interval()
qt_interval  = ecg_extractor.compute_qt_interval()
st_interval  = ecg_extractor.compute_st_interval()

print("ECG Morphology Features:")
print(f"  QRS Duration : {qrs_duration:.4f} s")
print(f"  PR Interval  : {pr_interval:.4f} s")
print(f"  QT Interval  : {qt_interval:.4f} s")
print(f"  ST Interval  : {st_interval:.4f} s")

# Visualize R peaks on ECG
fig = go.Figure()
t = np.arange(len(signal_col)) / fs
fig.add_trace(go.Scatter(x=t[:512], y=signal_col[:512], mode='lines', name='ECG'))
r_in_range = [r for r in r_peaks if r < 512]
if r_in_range:
    fig.add_trace(go.Scatter(
        x=np.array(r_in_range) / fs,
        y=signal_col[r_in_range],
        mode='markers', marker=dict(color='red', size=8, symbol='x'),
        name='R Peaks'
    ))
fig.update_layout(title='ECG with R Peak Detection', xaxis_title='Time (s)', yaxis_title='Amplitude')
fig.show()

P-Wave and S-Wave Analysis

p_wave_dur = ecg_extractor.compute_p_wave_duration()
s_wave     = ecg_extractor.compute_s_wave()

print(f"P-Wave Duration : {p_wave_dur:.4f} s")
print(f"S-Wave Feature  : {s_wave}")

Arrhythmia Detection

arrhythmias = ecg_extractor.detect_arrhythmias()
print("Arrhythmia Detection:")
print(arrhythmias)

PPG Autonomic Nervous System Features

from vitalDSP.feature_engineering.ppg_autonomic_features import PPGAutonomicFeatures

ppg_auto = PPGAutonomicFeatures(ppg_col, sampling_frequency=fs)

# Respiratory Sinus Arrhythmia — autonomic balance indicator
rsa = ppg_auto.compute_rsa()
print(f"RSA (Respiratory Sinus Arrhythmia): {rsa:.4f}")

# Respiratory Rate Variability
rrv = ppg_auto.compute_rrv()
print(f"RRV (Resp. Rate Variability)       : {rrv:.4f}")

# Fractal dimension of PPG
fd = ppg_auto.compute_fractal_dimension()
print(f"Fractal Dimension                  : {fd:.4f}")

# DFA scaling exponent
dfa = ppg_auto.compute_dfa()
print(f"DFA Scaling Exponent               : {dfa:.4f}")

PPG Light Features (SpO2, Perfusion Index)

from vitalDSP.feature_engineering.ppg_light_features import PPGLightFeatureExtractor

# Simulate IR and Red PPG signals (in real use, load dual-wavelength sensor data)
ir_signal  = ppg_col
red_signal = ppg_col * 0.95 + np.random.normal(0, 0.001, len(ppg_col))

ppg_light = PPGLightFeatureExtractor(ir_signal=ir_signal, red_signal=red_signal, sampling_freq=fs)

spo2_values, spo2_indices = ppg_light.calculate_spo2()
pi   = ppg_light.calculate_perfusion_index()
rr   = ppg_light.calculate_respiratory_rate()
ppr  = ppg_light.calculate_ppr()

print("PPG Light-Based Features:")
print(f"  SpO2 (mean)        : {float(np.mean(spo2_values)):.1f} %")
print(f"  Perfusion Index    : {pi:.4f} %")
print(f"  Respiratory Rate   : {rr:.2f} breaths/min")
print(f"  PPR (Pulse/Pleth)  : {ppr:.4f}")

ECG-PPG Synchronization

from vitalDSP.feature_engineering.ecg_ppg_synchronization_features import ECGPPGSynchronization

sync = ECGPPGSynchronization(
    ecg_signal=signal_col,
    ppg_signal=ppg_col,
    ecg_fs=fs,
    ppg_fs=fs
)

# Pulse Transit Time — correlates with blood pressure
ptt = sync.compute_ptt()
print(f"Pulse Transit Time (PTT)  : {ptt:.4f} s")

# Pulse Arrival Time
pat = sync.compute_pat()
print(f"Pulse Arrival Time (PAT)  : {pat:.4f} s")

# Heart rate / pulse rate synchrony
hr_pr = sync.compute_hr_pr_sync()
print(f"HR-PR Synchrony           : {hr_pr:.4f}")

# PPG rise time
rise_time = sync.compute_ppg_rise_time()
print(f"PPG Rise Time             : {rise_time:.4f} s")

ECG-PPG Overlay Visualization

# Show ECG and PPG overlaid, aligned in time
n = min(len(signal_col), len(ppg_col), 512)
t = np.arange(n) / fs

ecg_norm = (signal_col[:n] - signal_col[:n].mean()) / (signal_col[:n].std() + 1e-8)
ppg_norm = (ppg_col[:n] - ppg_col[:n].mean()) / (ppg_col[:n].std() + 1e-8)

fig = go.Figure()
fig.add_trace(go.Scatter(x=t, y=ecg_norm, mode='lines', name='ECG (normalized)'))
fig.add_trace(go.Scatter(x=t, y=ppg_norm, mode='lines', name='PPG (normalized)'))
fig.update_layout(
    title='ECG-PPG Synchronization',
    xaxis_title='Time (s)', yaxis_title='Normalized Amplitude'
)
fig.show()