streamsim.src.features.heart_rate module

R-Peak Detection and Heart Rate Calculation for ECG Signals

This module provides a streaming feature deriver that detects R-peaks in electrocardiogram (ECG) signals and calculates instantaneous heart rate. It implements a complete signal processing pipeline inspired by the Pan-Tompkins algorithm, adapted for real-time streaming applications.

Signal Processing Pipeline:

Raw ECG → Bandpass Filter → Derivative → Squaring → Moving Average → Thresholding → Peak Detection

Heart Rate Calculation:

  • Measures RR intervals between consecutive R-peaks

  • Computes instantaneous heart rate: HR = 60 / RR_interval (BPM)

  • Applies moving average smoothing over recent intervals for stability

  • Validates intervals against physiological bounds (30-220 BPM)

Features:

  • Adaptive thresholding for robust peak detection in varying signal conditions

  • Refractory period enforcement to prevent double-counting peaks

  • Physiological validation of RR intervals

  • Configurable smoothing window for heart rate stability

  • Comprehensive state reset capability for multi-segment analysis

Important Dependencies:

  • collections.deque: Efficient circular buffers for streaming data

  • streamsim.src.core.interfaces.StreamingFeatureDeriver: Base interface

Author: F.Feenstra with inspiration from Pan-Tompkins algorithm

Example

>>> from streamsim.src.features.heart_rate import HRFeatureDeriver
>>> deriver = HRFeatureDeriver(fs=360.0, threshold_factor=0.7)
class streamsim.src.features.heart_rate.HRFeatureDeriver(fs: float = 360.0, min_rr_sec: float = 0.2, threshold_factor: float = 0.7, rr_window_size: int = 5, min_hr: float = 30.0, max_hr: float = 220.0)[source]

Bases: StreamingFeatureDeriver

Feature deriver that detects R-peaks in ECG signals and calculates heart rate.

Signal Processing Pipeline:

Raw ECG → Bandpass → Derivative → Square → Moving Average → Threshold

Heart Rate Calculation:

  • Measures RR intervals between consecutive R-peaks

  • Computes instantaneous heart rate: HR = 60 / RR_interval (BPM)

  • Optionally smooths using a moving average of recent intervals

add_sample(sample: float, timestamp: float = None) → None[source]

Process a new ECG sample through the detection pipeline.

Executes the full signal processing chain: bandpass filtering, differentiation, squaring, moving average integration, and adaptive thresholding. If an R-peak is detected, updates the heart rate calculation.

Parameters:

  • sample (float) – The raw ECG signal sample value.

  • timestamp (float, optional) – Timestamp of the sample in seconds. If None, calculates based on sample count and fs. Default: None.

Note

The first ~200ms of data (depending on fs) will not produce valid detections as the buffers fill up. The get_heart_rate() method returns None until sufficient RR intervals are collected.

get_feature() → float | None[source]

Get the latest heart rate

get_features() → Tuple[float | None, float | None][source]

Get both the peak timestamp and heart rate.

Convenience method for retrieving all derived features at once.

Returns:

(peak_timestamp, heart_rate)

Return type:

Tuple[Optional[float], Optional[float]]

get_heart_rate() → float | None[source]

Get the current heart rate in beats per minute (BPM).

The heart rate is calculated as a moving average of the last N RR intervals, where N is determined by rr_window_size.

Returns:

Heart rate in BPM, or None if insufficient data.

Return type:

Optional[float]

get_rr_interval() → float | None[source]

Get the time difference between the last two detected R-peaks. Useful for analyzing beat-to-beat variability.

Returns:

Last RR interval in seconds, or None if no interval recorded.

Return type:

Optional[float]

reset() → None[source]

Reset all internal state and buffers.