med.signal - Biosignal Processing
Peak detection, heart rate analysis, HRV metrics, rhythm classification, SpO2 calculation, respiratory rate estimation, and waveform analysis for medical device integration.
Heart Rate & Rhythm
heart_rate_from_rr(rr_intervals_ms) → dict
Calculate heart rate from R-R intervals (milliseconds). HR = 60000 / mean_RR.
Returns classification: BRADYCARDIA, NORMAL_SINUS, TACHYCARDIA, or SEVERE_TACHYCARDIA.
lib.signal.heart_rate_from_rr([800, 810, 790, 800, 820])
# {'mean_hr_bpm': 74.5, 'classification': 'NORMAL_SINUS'}
classify_rhythm(rr_intervals_ms) → dict
Basic cardiac rhythm classification from R-R interval regularity:
| Rhythm | Criteria |
|---|---|
| Normal Sinus | HR 60-100, regular |
| Sinus Bradycardia | HR under 60, regular |
| Sinus Tachycardia | HR 100-150, regular |
| AF Suspected | Irregular with CV over 25% |
| SVT Possible | HR over 150 |
lib.signal.classify_rhythm([800, 810, 790, 800, 820])
# {'rhythm': 'NORMAL_SINUS_RHYTHM', 'heart_rate': 74.5, 'regularity': 'REGULAR'}
hrv_metrics(rr_intervals_ms) → dict
Heart Rate Variability time-domain analysis:
| Metric | Description | Clinical Significance |
|---|---|---|
| SDNN | SD of NN intervals | Overall HRV; under 50ms = high cardiac risk |
| RMSSD | Root mean square of successive diffs | Parasympathetic activity |
| pNN50 | % of successive diffs over 50ms | Vagal tone marker |
lib.signal.hrv_metrics([800, 810, 790, 800, 820, 830, 780])
# {'sdnn_ms': 17.1, 'rmssd_ms': 17.3, 'pnn50_pct': 0.0,
# 'autonomic_status': 'LOW_HRV'}
Pulse Oximetry
spo2_from_ratio(red_ac, red_dc, ir_ac, ir_dc) → dict
Calculate SpO2 from pulse oximeter raw data using Beer-Lambert approximation:
R = (Red_AC / Red_DC) / (IR_AC / IR_DC)
SpO2 = 110 - 25 * R
Returns status: NORMAL (≥95%), MILD_HYPOXEMIA, MODERATE_HYPOXEMIA, SEVERE_HYPOXEMIA.
perfusion_index(ac_component, dc_component) → dict
Perfusion Index: PI = (AC / DC) * 100. Normal over 1.0%, low perfusion under 0.5%.
Waveform Analysis
detect_peaks(waveform, threshold=0.5) → dict
Threshold-based peak detection with refractory period. Works for ECG QRS complexes, pulse waveforms, etc.
lib.signal.detect_peaks([0, 0.2, 0.8, 1.5, 0.9, 0.1, 0, 0.3, 0.7, 1.4, 0.8, 0.2])
# {'peaks_detected': 2, 'peaks': [{'index': 3, 'amplitude': 1.5}, ...]}
moving_average(data, window=5) → dict
Simple moving average filter for signal smoothing and noise removal.
detect_anomaly(data, baseline_mean=None, threshold_sd=2.0) → dict
Statistical anomaly detection. Flags data points exceeding threshold_sd standard deviations from baseline.
Respiratory
respiratory_rate(waveform, sampling_rate_hz=25.0) → dict
Estimate respiratory rate from impedance or chest waveform using zero-crossing method.
Returns status: BRADYPNEA (under 12), NORMAL (12-20), TACHYPNEA (20-30), SEVERE_TACHYPNEA (over 30).
Integration with med.io
med.signal pairs naturally with med.io for real-time device monitoring:
protocol WaveformMonitor {
input: Patient p;
// Capture ECG waveform from bedside monitor
let ecg = med.io.read_waveform("Monitor_01", "ECG_II", 5.0, 250.0);
// Analyze
let peaks = med.signal.detect_peaks(ecg.data);
let rhythm = med.signal.classify_rhythm(peaks.inter_peak_intervals);
if rhythm.rhythm == "ATRIAL_FIBRILLATION_SUSPECTED" {
alert "Possible AF detected" severity: critical;
}
}
See Also
- med.io — device interfaces for real-time waveform capture
- med.scores — clinical scoring from vital sign parameters
- med.lab — correlate biosignal findings with lab results