Common utilities (`vital_sqi.common`)

Shared helpers used across the SQI, pipeline, and calibration layers: band-pass filtering, R-peak detection, template generation, power spectrum estimation, and general utilities.

Band-pass filter 

Filtering of raw signals using bandpass filters.

class vital_sqi.common.band_filter.BandpassFilter(band_type='butter', fs=100)[source]

Bases: object

A class for bandpass filtering of signals using different filter types.

Parameters:

band_type (str, optional) –
The type of bandpass filter to use, default is “butter”. Options:
- ”butter”: Butterworth filter
- ”cheby1”: Chebyshev Type I filter
- ”cheby2”: Chebyshev Type II filter
- ”ellip”: Elliptic (Cauer) filter
- ”bessel”: Bessel filter
fs (int, optional) – Sampling frequency of the signal (default is 100 Hz).

Examples

>>> filter = BandpassFilter(band_type="butter", fs=100)
>>> filtered_signal = filter.signal_lowpass_filter(data, cutoff=10, order=3)

signal_highpass_filter(data, cutoff, order=5, a_pass=3, rp=4, rs=40)[source]

Applies a high-pass filter to the input signal.

Parameters:

data (array_like) – The input signal to be filtered.
cutoff (float) – The cutoff frequency of the filter.
order (int, optional) – The order of the filter (default is 5).
a_pass (float, optional) – Passband maximum loss (only for Chebyshev Type I, default is 3).
rp (float, optional) – Maximum ripple in the passband (only for elliptic filters, default is 4 dB).
rs (float, optional) – Minimum stopband attenuation (only for elliptic filters, default is 40 dB).

Returns:

The filtered signal.

Return type:

array_like

signal_lowpass_filter(data, cutoff, order=3, a_pass=3, rp=4, rs=40)[source]

Applies a low-pass filter to the input signal.

Parameters:

data (array_like) – The input signal to be filtered.
cutoff (float) – The cutoff frequency of the filter.
order (int, optional) – The order of the filter (default is 3).
a_pass (float, optional) – Passband maximum loss (only for Chebyshev Type I, default is 3).
rp (float, optional) – Maximum ripple in the passband (only for elliptic filters, default is 4 dB).
rs (float, optional) – Minimum stopband attenuation (only for elliptic filters, default is 40 dB).

Returns:

The filtered signal.

Return type:

array_like

R-peak detection 

R peak detection approaches for PPG and ECG.

class vital_sqi.common.rpeak_detection.PeakDetector(wave_type='PPG', fs=100)[source]

Bases: object

Detects peaks in PPG and ECG signals using various algorithms.

Parameters:

wave_type (str, optional) – The type of waveform to detect peaks from, either ‘PPG’ or ‘ECG’ (default is ‘PPG’).
fs (int, optional) – Sampling frequency of the signal (default is 100).

Examples

>>> detector = PeakDetector(wave_type="PPG", fs=100)
>>> signal = np.random.randn(1000)
>>> peaks, troughs = detector.ppg_detector(signal)  # DEFAULT (vitalDSP)
>>> peaks, troughs = detector.ppg_detector(signal, detector_type=ADAPTIVE_THRESHOLD)
>>> r, q, s, p, t = PeakDetector(wave_type="ECG", fs=256).ecg_detector(signal)

detect_peak_trough_DEFAULT(s)[source]

Detects peaks and troughs using SciPy’s find_peaks function.

Parameters:: s (array_like) – Input signal.
Returns:: Detected peaks and troughs.
Return type:: tuple

detect_peak_trough_adaptive_threshold(s, adaptive_size=0.75, overlap=0, sliding=1)[source]

Detect peaks and troughs in a signal using an adaptive threshold approach.

Parameters:

s (array_like) – Input signal.
adaptive_size (float, optional) – Window size for adaptive thresholding as a fraction of the sampling rate (default is 0.75).
overlap (float, optional) – Overlapping ratio for the sliding window (default is 0).
sliding (int, optional) – Step size for sliding window (default is 1).

Returns:

Detected peaks and troughs.

Return type:

tuple

detect_peak_trough_ampd(s)[source]

Automatic Multiscale Peak Detection (AMPD) for PPG signals.

AMPD finds peaks without requiring pre-set parameters by computing a Local Maxima Scalogram (LMS) across all window scales and identifying the scale with the fewest maxima (optimal scale), then intersecting maxima across all scales up to that point.

Reference: Scholkmann F et al. “An Efficient Algorithm for Automatic Peak Detection in Noisy Periodic and Quasi-Periodic Signals.” Algorithms 2012;5(4):588-603.

Parameters:: s (array_like) – Input PPG signal.
Returns:: (peaks, troughs) as numpy arrays of indices.
Return type:: tuple

detect_peak_trough_billauer(s, delta=0.8)[source]

Billauer’s method for peak and trough detection, translated from MATLAB.

Parameters:

s (array_like) – Input signal.
delta (float, optional) – Minimum difference required to consider a peak (default is 0.8).

Returns:

Detected peaks and troughs.

Return type:

tuple

detect_peak_trough_clusterer(s)[source]

Detects peaks and troughs in a signal using a clustering technique.

Parameters:: s (array_like) – Input signal.
Returns:: Detected peaks and troughs.
Return type:: tuple

detect_peak_trough_count_orig(s)[source]

Detect peaks and troughs in a signal using local extrema and thresholding.

Parameters:: s (array_like) – Input signal.
Returns:: Detected peaks and troughs as numpy arrays.
Return type:: tuple

detect_peak_trough_local_max_ibi(s)[source]

Local-maximum PPG detector with inter-beat-interval (IBI) tracking.

Finds all local maxima above a dynamic threshold, then prunes false positives by enforcing a physiologically plausible IBI window derived from the running median IBI (40–200 BPM, i.e. 300–1500 ms).

This is particularly robust to low-amplitude diastolic humps that confuse simpler detectors.

Parameters:: s (array_like) – Input PPG signal.
Returns:: (peaks, troughs) as numpy arrays of indices.
Return type:: tuple

detect_peak_trough_moving_average_threshold(s)[source]

Detect peaks using a two-moving-average threshold (Elgendi et al. 2013). Reference: Elgendi M. et al., PLoS ONE 8(10):e76585, 2013.

A short MA (w1 ≈ 120 ms) rises above a long MA (w2 ≈ 670 ms) baseline to form “blocks of interest”; the raw-signal argmax inside each block is the peak.

Parameters:: s (array_like) – Input signal.
Returns:: Detected peaks and troughs.
Return type:: tuple

detect_peak_trough_slope_sum(s)[source]

Detect peaks and troughs in a signal using the slope sum method. Reference: Zong et al. (2003) Computers in Cardiology 30:259-262.

Parameters:: s (array_like) – Input signal.
Returns:: Detected peaks and troughs as lists of indices.
Return type:: tuple

detect_r_peaks_engzee(s)[source]

Engzee-Zeelenberg ECG R-peak detector (1979), as described in: Engzee R, Zeelenberg C. “A single scan algorithm for QRS-detection and feature extraction.” Computers in Cardiology 1979;6:37-42.

Uses the absolute first difference of a high-pass filtered signal with a dynamic threshold that adapts every detected beat.

Parameters:: s (array_like) – Input ECG signal.
Returns:: Indices of detected R-peaks.
Return type:: np.ndarray

detect_r_peaks_hamilton(s)[source]

Hamilton-Tompkins simplified ECG R-peak detector (2002).

Uses a single-pass derivative-square-integrate pipeline with fixed refractory period and mean-based threshold, suitable for real-time use.

Reference: Hamilton PS. “Open source ECG analysis.” Computers in Cardiology 2002;29:101-104.

Parameters:: s (array_like) – Input ECG signal.
Returns:: Indices of detected R-peaks.
Return type:: np.ndarray

detect_r_peaks_pan_tompkins(s)[source]

Pan-Tompkins (1985) R-peak detector.

Pipeline: bandpass → derivative → square → moving-window integration → adaptive threshold with two learning periods.

Reference: Pan J, Tompkins WJ. IEEE Trans Biomed Eng. 1985;32(3):230-6.

Parameters:: s (array_like) – Input ECG signal.
Returns:: Indices of detected R-peaks in the original signal.
Return type:: np.ndarray

ecg_detector(s, detector_type=10, get_session=False)[source]

Detects R-peaks and critical points in ECG signals.

The default path (ECG_DEFAULT) uses vitalDSP WaveformMorphology, which provides high-quality derivative-based R-peak detection plus full critical point extraction (Q/S/P/T). Alternative R-peak algorithms (PAN_TOMPKINS, HAMILTON, ENGZEE) also use WaveformMorphology for the critical-point step so that Q/S/P/T quality is never downgraded.

Parameters:

s (array_like) – Input ECG signal.
detector_type (int, optional) –
R-peak detection algorithm. Available constants:
- ECG_DEFAULT (10) — vitalDSP WaveformMorphology (recommended; also provides Q/S/P/T morphology points)
- PAN_TOMPKINS (11) — classic Pan-Tompkins 1985 with adaptive dual-threshold; bandpass 5–15 Hz
- HAMILTON (12) — Hamilton-Tompkins simplified 2002; single-pass derivative-square-integrate; bandpass 8–16 Hz
- ENGZEE (13) — Engzee-Zeelenberg 1979; dynamic threshold on high-pass filtered derivative
All alternatives use vitalDSP WaveformMorphology for Q/S/P/T extraction, anchored to the chosen R-peak indices. Default is ECG_DEFAULT.
get_session (bool, optional) – If True return (r_peaks, ecg_session) instead of the full tuple.

Returns:

(r_peaks, q_valleys, s_valleys, p_peaks, t_peaks) normally, or (r_peaks, ecg_session) when get_session=True.

Return type:

tuple

get_ROI(s, adaptive_threshold, margin=0.1)[source]

Identify regions of interest (ROIs) in the signal where peaks or troughs are likely to occur.

Parameters:

s (array_like) – Input signal.
adaptive_threshold (array_like) – Adaptive threshold values for the signal.
margin (float, optional) – Margin (fraction of the signal range) to include before and after the ROI (default is 0.1).

Returns:

Two lists: start_ROIs and end_ROIs, which contain the start and end indices of the ROIs.

Return type:

tuple

Notes

ROIs are defined as contiguous regions where the signal exceeds the adaptive threshold.
Margins can be added to widen the ROIs for more inclusive peak/trough detection.

Example

>>> s = [0, 1, 3, 7, 5, 2, 0, 6, 9, 8, 4, 1, 0]
>>> adaptive_threshold = [4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4]
>>> peak_detector = PeakDetection(fs=100)
>>> start_ROIs, end_ROIs = peak_detector.get_ROI(s, adaptive_threshold)

get_moving_average(q, w)[source]

Calculates the moving average of a sequence.

Parameters:

q (array_like) – Input sequence.
w (int) – Window size for the moving average.

Returns:

Moving average of the sequence.

Return type:

array_like

ppg_detector(s, detector_type=6, get_session=False, preprocess=False, cubing=False)[source]

Detects peaks in PPG signals using specified detector type.

Parameters:

s (array_like) – Input PPG signal.
detector_type (int, optional) –
Method for peak detection (default is DEFAULT = 6, which uses vitalDSP WaveformMorphology systolic-peak detection). Available constants:
- ADAPTIVE_THRESHOLD (1) — threshold adapts to local signal amplitude
- COUNT_ORIG_METHOD (2) — count-based local maxima
- CLUSTERER_METHOD (3) — KMeans clustering
- SLOPE_SUM_METHOD (4) — Zong 2003 slope-sum onset
- MOVING_AVERAGE_METHOD (5) — Elgendi two-moving-average
- DEFAULT (6) — vitalDSP WaveformMorphology (recommended)
- BILLAUER_METHOD (7) — Billauer peak/trough tracker
- AMPD_METHOD (8) — Automatic Multiscale Peak Detection
- LOCAL_MAX_IBI (9) — local-max with IBI tracking
preprocess (bool, optional) – Whether to apply filtering to the signal (default is False).
cubing (bool, optional) – Whether to cube the signal for enhanced peak detection (default is False).

Returns:

Detected peaks and troughs.

Return type:

tuple

search_for_onset(slope_sum, n, local_max)[source]: Walk back from detection point n to find the pulse onset. Following Zong et al. 2003: onset is where slope_sum drops below 1% of local_max, up to 200 ms before the detection point.

Template generation 

Generating templates of ECG and PPG complexes

vital_sqi.common.generate_template.ecg_dynamic_template(width, sfecg=256, N=256, Anoise=0, hrmean=60, hrstd=1, lfhfratio=0.5, sfint=512, ti=array([-70, -15, 0, 15, 100]), ai=array([1.2, -5., 30., -7.5, 0.75]), bi=array([0.25, 0.1, 0.1, 0.1, 0.4]))[source]

Generate a synthetic ECG signal template.

Parameters:

width (int) – Desired length of the output template.
sfecg (int or float, optional) – Parameters for the ECG generation model.
N (int or float, optional) – Parameters for the ECG generation model.
Anoise (int or float, optional) – Parameters for the ECG generation model.
hrmean (int or float, optional) – Parameters for the ECG generation model.
hrstd (int or float, optional) – Parameters for the ECG generation model.
lfhfratio (int or float, optional) – Parameters for the ECG generation model.
sfint (int or float, optional) – Parameters for the ECG generation model.
ti (numpy.ndarray, optional) – Arrays of model parameters.
ai (numpy.ndarray, optional) – Arrays of model parameters.
bi (numpy.ndarray, optional) – Arrays of model parameters.

Returns:

Generated ECG template.

Return type:

numpy.ndarray

vital_sqi.common.generate_template.interp(ys, mul)[source]

Perform cubic interpolation and extrapolation on a sequence.

Parameters:

ys (array_like) – Input sequence.
mul (int) – Multiplication factor for interpolation.

Returns:

Interpolated sequence.

Return type:

numpy.ndarray

vital_sqi.common.generate_template.ordinary_differential_equation(t, x_equations, rr, sfint, ti, ai, bi)[source]

Solve the ordinary differential equation for synthetic ECG generation.

Parameters:

t (float) – Time variable.
x_equations (array_like) – Initial values for x, y, and z variables.
rr (numpy.ndarray) – Resampled RR interval sequence.
sfint (int) – Sampling frequency for interpolation.
ti (numpy.ndarray) – Model parameters for ECG generation.
ai (numpy.ndarray) – Model parameters for ECG generation.
bi (numpy.ndarray) – Model parameters for ECG generation.

Returns:

Derivatives of x, y, and z.

Return type:

list

vital_sqi.common.generate_template.ppg_absolute_dual_skewness_template(width, e_1=1, w_1=2.5, e_2=3, w_2=3, a=4)[source]

Generate a PPG template using two skewness distributions.

Parameters:

width (int) – Sample size of the generated waveform.
e_1 (float, optional) – Parameters for the first skew distribution (default e_1=1, w_1=2.5).
w_1 (float, optional) – Parameters for the first skew distribution (default e_1=1, w_1=2.5).
e_2 (float, optional) – Parameters for the second skew distribution (default e_2=3, w_2=3).
w_2 (float, optional) – Parameters for the second skew distribution (default e_2=3, w_2=3).
a (float, optional) – Shape parameter for both distributions (default is 4).

Returns:

A 1-D array representing the PPG waveform.

Return type:

numpy.ndarray

vital_sqi.common.generate_template.ppg_dual_double_frequency_template(width)[source]

Generate a PPG template using two sine waveforms with different frequencies.

Parameters:: width (int) – The sample size of the generated waveform.
Returns:: A 1-D array representing the PPG waveform with a diastolic peak at a low position.
Return type:: numpy.ndarray

vital_sqi.common.generate_template.ppg_nonlinear_dynamic_system_template(width)[source]

Generate a PPG template based on a nonlinear dynamic system.

Parameters:: width (int) – Desired length of the template.
Returns:: A rescaled signal template for PPG.
Return type:: numpy.ndarray

vital_sqi.common.generate_template.rr_process(flo, fhi, flostd, fhistd, lfhfratio, hrmean, hrstd, sfrr, n, seed=0)[source]

Generate an RR interval time series with bimodal power spectrum.

Parameters:

flo (float) – Low and high frequencies of the Gaussian distributions.
fhi (float) – Low and high frequencies of the Gaussian distributions.
flostd (float) – Standard deviations of low and high frequencies.
fhistd (float) – Standard deviations of low and high frequencies.
lfhfratio (float) – Ratio of low-frequency to high-frequency power.
hrmean (float) – Mean and standard deviation of heart rate.
hrstd (float) – Mean and standard deviation of heart rate.
sfrr (int) – Sampling frequency for RR intervals.
n (int) – Length of the generated time series.
seed (int or None, optional) – Seed for the random phase generator (default 0). Pass None to draw fresh randomness from the global numpy state. A fixed default makes ecg_dynamic_template deterministic, so the DTW reference template is reproducible across runs.

Returns:

Generated RR interval time series.

Return type:

numpy.ndarray

vital_sqi.common.generate_template.skew_func(x, e=0, w=1, a=0)[source]

Generate a skewness distribution.

Parameters:

x (array_like) – Input sequence of time points.
e (float, optional) – Location parameter (default is 0).
w (float, optional) – Scale parameter (default is 1).
a (float, optional) – Shape parameter (default is 0).

Returns:

A 1-D array of a skewness distribution.

Return type:

numpy.ndarray

vital_sqi.common.generate_template.squeeze_template(s, width)[source]

Compress a signal template by averaging values within a given width.

Parameters:

s (array_like) – Input signal array.
width (int) – Desired compressed length of the output signal.

Returns:

Compressed signal.

Return type:

numpy.ndarray

Power spectrum 

vital_sqi.common.power_spectrum.calculate_band_power(freq, power, fmin, fmax)[source]

Calculate the absolute power within a specified frequency band.

Parameters:

freq (array-like) – Array of frequencies.
power (array-like) – Array of power values corresponding to each frequency.
fmin (float) – Lower bound of the frequency band.
fmax (float) – Upper bound of the frequency band.

Returns:

Absolute power within the specified frequency band.

Return type:

float

vital_sqi.common.power_spectrum.calculate_psd(rr_intervals, method='welch', hr_sampling_frequency=4, power_type='density', max_lag=3)[source]

Calculate the power spectral density (PSD) from RR intervals.

Parameters:

rr_intervals (array-like) – List of RR intervals (in ms).
method (str, optional) – Method to calculate PSD (‘welch’, ‘lomb’, or ‘ar’), by default “welch”.
hr_sampling_frequency (int, optional) – Sampling frequency for heart rate, by default 4 Hz.
power_type (str, optional) – Power type for AR method (‘density’ or ‘spectrogram’), by default “density”.
max_lag (int, optional) – Maximum lag for autoregressive (AR) method, by default 3.

Returns:

Frequencies and power spectral density values.

Return type:

tuple

vital_sqi.common.power_spectrum.calculate_spectrogram(rr_intervals, hr_sampling_frequency=4)[source]

Compute the spectrogram for the given RR intervals.

Parameters:

rr_intervals (array-like) – List of RR intervals (in ms).
hr_sampling_frequency (int, optional) – Sampling frequency for heart rate, by default 4 Hz.

Returns:

Frequencies, power spectral density values, and time segments.

Return type:

tuple

vital_sqi.common.power_spectrum.calculate_wavelet_power(rr_intervals, heart_rate=4, mother_wave='morlet')[source]

Calculate spectral power using continuous wavelet transform (CWT).

Parameters:

rr_intervals (array-like) – List of RR intervals (in ms).
heart_rate (int, optional) – Sampling rate for heart rate data, by default 4 Hz.
mother_wave (str, optional) – Mother wavelet type (‘gaussian’, ‘paul’, ‘mexican_hat’, or ‘morlet’), by default “morlet”.

Returns:

Frequencies and wavelet power values.

Return type:

tuple

vital_sqi.common.power_spectrum.compute_time_and_bpm(rr_intervals)[source]

Generate timestamps and convert RR intervals to bpm.

Parameters:: rr_intervals (array-like) – List of RR intervals (in ms).
Returns:: Timestamps for each heartbeat (in seconds) and corresponding bpm values.
Return type:: tuple

vital_sqi.common.power_spectrum.interpolate_rr_intervals(ts_rr, bpm_list, sampling_frequency, method='linear')[source]

Interpolate RR intervals to resample the heart rate data.

Parameters:

ts_rr (array-like) – Timestamps for RR intervals (in ms).
bpm_list (array-like) – Heart rate values in bpm.
sampling_frequency (int) – Sampling frequency for interpolation.
method (str, optional) – Interpolation method, by default “linear”.

Returns:

Interpolated heart rate in bpm.

Return type:

numpy.ndarray

Utilities 

class vital_sqi.common.utils.HiddenPrints[source]

Bases: object

Context manager to suppress console output temporarily.

Usage:

with HiddenPrints():: # Code that prints to console

vital_sqi.common.utils.calculate_sampling_rate(timestamps)[source]

Calculates the sampling rate from an array of timestamps.

Parameters:: timestamps (array_like) – Array of timestamps (float, pd.Timestamp, or np.datetime64).
Returns:: The calculated sampling rate in Hz, or None if calculation fails.
Return type:: float or None

vital_sqi.common.utils.check_conflict(decision_lt, decision_gt)[source]

Checks for conflicts between two decisions.

Parameters:

decision_lt (pd.DataFrame) – Decision with less-than operand.
decision_gt (pd.DataFrame) – Decision with greater-than operand.

Returns:

Label if no conflict, otherwise raises ValueError.

Return type:

str

vital_sqi.common.utils.check_signal_format(s)[source]

Validate input and return a DataFrame with a 'timestamps' column.

Coerces lists/ndarrays into a single-column 'signal' DataFrame, then ensures a leading 'timestamps' column of dtype datetime64.

Parameters:: s (pd.DataFrame, list, or np.ndarray) – Input signal data.
Returns:: DataFrame with a 'timestamps' column (datetime64) followed by at least one numeric signal column.
Return type:: pd.DataFrame
Raises:: TypeError – If s is not a DataFrame, list, or numpy array, or if the signal column is not numeric.

vital_sqi.common.utils.check_unique_pair(pair)[source]

Checks that there are no duplicate decisions.

Parameters:: pair (pd.DataFrame) – DataFrame containing a pair of values.
Returns:: True if the pair is unique.
Return type:: bool

vital_sqi.common.utils.check_valid_signal(signal)[source]

Validates that the input is a numeric array-like signal.

Parameters:: signal (array_like) – The signal to validate.
Returns:: True if the signal is valid.
Return type:: bool
Raises:: ValueError – If the signal is invalid.

vital_sqi.common.utils.create_rule_def(sqi_name, lower_bound=0, upper_bound=1)[source]

Create a default rule definition for an SQI.

The rule accepts values in the half-open interval (lower_bound, upper_bound) and rejects everything else.

Warning

The default parameter values upper_bound=0, lower_bound=1 produce a degenerate rule where lower_bound > upper_bound, meaning nothing is ever accepted. Always supply explicit, physiologically meaningful bounds when calling this function, e.g.:

create_rule_def("kurtosis_sqi", lower_bound=0.5, upper_bound=5.0)

Parameters:

sqi_name (str) – Name of the SQI; used as both the dict key and the "name" field in the rule definition.
upper_bound (float, optional) – Exclusive upper bound for the accept region (default 0).
lower_bound (float, optional) – Exclusive lower bound for the accept region (default 1).

Returns:

Nested dict with the structure expected by the Rule engine:

{
  sqi_name: {
    "name": sqi_name,
    "def": [...],
    "desc": "",
    "ref": ""
  }
}

Return type:

dict

vital_sqi.common.utils.cut_segment(df, milestones, yield_mode=False)[source]

Splits a DataFrame into segments based on the start and end indices provided in the milestones DataFrame.

Parameters:

df (pd.DataFrame) – Signal DataFrame containing the full data to be segmented.
milestones (pd.DataFrame) – DataFrame containing ‘start’ and ‘end’ columns, representing the start and end indices for each segment.
yield_mode (bool, optional) – When True, return a generator instead of a list. Useful for large recordings where materialising all segments at once would be memory-intensive. Default False.

Returns:

DataFrame segments based on the specified start and end indices.

Return type:

list of pd.DataFrame or generator of pd.DataFrame

Raises:

AssertionError – If milestones is not a DataFrame or if ‘start’ and ‘end’ columns are missing. If start and end indices are out of bounds.

vital_sqi.common.utils.decompose_operand(rule_dict)[source]

Decomposes operands into distinct components for comparison.

Parameters:: rule_dict (dict) – Dictionary of rule definitions.
Returns:: DataFrame of all operands after decomposition.
Return type:: pd.DataFrame

vital_sqi.common.utils.format_milestone(start_milestone, end_milestone)[source]

Formats start and end milestones as a DataFrame, trimming to the minimum length if necessary.

Parameters:

start_milestone (array-like) – Start indices of segments.
end_milestone (array-like) – End indices of segments.

Returns:

DataFrame with ‘start’ and ‘end’ columns, truncated to minimum length if input lengths differ.

Return type:

pd.DataFrame

vital_sqi.common.utils.generate_labels(df, boundaries)[source]

Efficiently generates interval and value labels for the boundaries.

Parameters:

df (pd.DataFrame) – DataFrame of decomposed rule definitions.
boundaries (np.ndarray) – Unique sorted boundary values.

Returns:

Interval and value labels for the boundaries.

Return type:

tuple of np.ndarray

vital_sqi.common.utils.generate_timestamp(start_datetime, sampling_rate, signal_length)[source]

Generates a sequence of timestamps for a signal.

Parameters:

start_datetime (datetime.datetime or pd.Timestamp or None) – The starting timestamp. If None, uses current time.
sampling_rate (float) – The sampling rate in Hz.
signal_length (int) – The number of timestamps to generate.

Returns:

Array of pd.Timestamp objects.

Return type:

np.ndarray

Raises:

ValueError – If sampling_rate is not a real number.

vital_sqi.common.utils.get_decision(df, boundaries, idx)[source]

Recursively fetches the decision for a given boundary index.

Parameters:

df (pd.DataFrame) – DataFrame of rule definitions.
boundaries (np.ndarray) – Array of boundary values.
idx (int) – Current index of the boundary.

Returns:

Decision label or None if not found.

Return type:

str

vital_sqi.common.utils.get_inteveral_label_list(df, boundaries)[source]

vital_sqi.common.utils.get_nn(signal, wave_type='PPG', sample_rate=100, rpeak_method=6, remove_ectopic_beat=False)[source]

Calculate NN intervals from a PPG or ECG signal.

Parameters:

signal (array_like) – Input signal data.
wave_type (str, optional) –
Type of waveform (‘PPG’ or ‘ECG’), by default ‘PPG’.

Warning

Callers must pass the correct wave_type. Passing ECG data through the PPG default produces no peaks because the underlying vitalDSP.transforms.beats_transformation.RRTransformation runs morphology-specific preprocessing that destroys the wrong signal type.
sample_rate (int or float, optional) –
Sampling frequency in Hz, by default 100.

Warning

The default 100 exists for backwards compatibility but is almost never right for ECG data (commonly 128/250/256/500 Hz). Pass the actual sampling rate or peak detection will silently fail and return an empty array.
rpeak_method (int, optional) – Method identifier for R-peak detection, by default 6.
remove_ectopic_beat (bool, optional) – If True, removes ectopic beats, by default False.

Returns:

Array of NN intervals in milliseconds. Empty array on failure; the failure reason is logged at DEBUG level on the module’s named logger.

Return type:

np.ndarray

vital_sqi.common.utils.get_value_label_list(df, boundaries, interval_label_list)[source]

vital_sqi.common.utils.parse_datetime(string, type='datetime')[source]

A simple dateparser that detects common datetime formats

Parameters:: string (str) – a date string in format as denoted below.
Returns:: datetime object of a time.
Return type:: datetime.datetime

vital_sqi.common.utils.parse_rule(name, source)[source]

Parses rule definitions from a JSON file or dictionary.

Parameters:

name (str) – Name of the SQI to retrieve.
source (str or dict) – Path to the JSON file or dictionary containing rule definitions.

Returns:

A tuple containing rule definitions, boundaries, and label lists.

Return type:

tuple

vital_sqi.common.utils.sanitize_sqi(values)[source]

Replace inf/-inf with NaN, then fill NaN with the column median.

Parameters:: values (array-like) – SQI values that may contain inf, -inf, or NaN.
Returns:: Cleaned float array of the same length with no inf or NaN values.
Return type:: np.ndarray

vital_sqi.common.utils.sort_rule(rule_def)[source]

Sorts rule definitions by value and operand order.

Parameters:: rule_def (list) – List of rule definitions.
Returns:: Sorted DataFrame of rules.
Return type:: pd.DataFrame

vital_sqi.common.utils.update_rule(rule_def, threshold_list=None, is_update=True)[source]

Updates rule definitions with new thresholds.

Parameters:

rule_def (list) – Existing rule definitions.
threshold_list (list, optional) – List of new thresholds to add (default is empty).
is_update (bool, optional) – Indicates whether to update or create new rules (default is True).

Returns:

Updated rule definitions, boundaries, and label lists.

Return type:

tuple

Common utilities (vital_sqi.common)

Band-pass filter

R-peak detection

Template generation

Power spectrum

Utilities

Usage:

Common utilities (`vital_sqi.common`)

Band-pass filter 

R-peak detection 

Template generation 

Power spectrum 

Utilities 