hrv

HRV

estimate_hrv(nni, fps, freq=False)

Estimates HRV (Heart Rate Variability) parameters including RMSSD, SDNN, PNN50, lf power, hf power, and ratio.

nni (array): NNI (Normal-to-Normal Interval) time series data. fps (int): Sampling frequency of the data in frames per second. freq (bool, optional): Boolean flag to indicate whether to include lf, hf, and ratio values. Defaults to False.

list: List containing HRV estimates [rmssd, sdnn, pnn50, lf, hf, ratio].

Notes: The current threshold settings (31/10/2022) are: - sdnn: 0.75 - rmssd: 0.5 - pnn50: 0.5 - lf power, hf power, ratio: 0.75

Source code in redesign_pipeline/vital_calculations/hrv.py

def estimate_hrv(nni, fps, freq=False):
    '''
    Estimates HRV (Heart Rate Variability) parameters including RMSSD, SDNN, PNN50, lf power, hf power, and ratio.

    Parameters:
    nni (array): NNI (Normal-to-Normal Interval) time series data.
    fps (int): Sampling frequency of the data in frames per second.
    freq (bool, optional): Boolean flag to indicate whether to include lf, hf, and ratio values. Defaults to False.

    Returns:
    list: List containing HRV estimates [rmssd, sdnn, pnn50, lf, hf, ratio].

    Notes:
    The current threshold settings (31/10/2022) are:
    - sdnn: 0.75
    - rmssd: 0.5
    - pnn50: 0.5
    - lf power, hf power, ratio: 0.75
    '''
    nni_50 = nni_threshold(nni, 0.5) / fps
    nni_75 = nni_threshold(nni, 0.75) / fps
    sdnn = pyhrv.time_domain.sdnn(nni=nni_75)['sdnn'] if len(nni_75)>1 else 0
    rmssd = pyhrv.time_domain.rmssd(nni=nni_50)['rmssd'] if len(nni_50)>1 else 0
    pnn50 = pyhrv.time_domain.nn50(nni=nni_50)['pnn50'] if len(nni_50)>1 else 0
    if freq:
        if len(nni_75)>3: 
            vlf, lf, hf = pyhrv.frequency_domain.welch_psd(nni=nni_75, show=False, mode='dev')[0]['fft_abs'] 
        else:
            vlf, lf, hf = (0,1,1)
        ratio = lf / hf
    else:
        lf = 0
        hf = 0
        ratio = 0

    return [rmssd, sdnn, pnn50, lf, hf, ratio]

get_hrv_estimates(nni_methods, fps, freq=False)

Returns HRV estimates such as SDNN, RMSSD, PNN50, lf power, hf power and ratio.

Parameters

array

NNI time series data.

int

Sampling frequency of the data in fps.

bool, optional

Boolean flag to indicate whether to include the lf, hf and ratio values, by default False.

Returns

tuple HRV estimates such as SDNN, RMSSD, PNN50, lf power, hf power and ratio.

Notes

The current threshold settings (31/10/2022) are: - sdnn: 0.75 - rmssd: 0.5 - pnn50: 0.5 - lf power, hf power, ratio: 0.75

Source code in redesign_pipeline/vital_calculations/hrv.py

def get_hrv_estimates(nni_methods, fps, freq=False):
    '''
    Returns HRV estimates such as SDNN, RMSSD, PNN50, lf power, hf power and ratio.

    Parameters
    ----------
    nni : array
        NNI time series data.
    fps : int
        Sampling frequency of the data in fps.
    freq : bool, optional
        Boolean flag to indicate whether to include the lf, hf and ratio values, by default False.

    Returns
    -------
    tuple
        HRV estimates such as SDNN, RMSSD, PNN50, lf power, hf power and ratio.

    Notes
    -----
    The current threshold settings (31/10/2022) are:
    - sdnn: 0.75
    - rmssd: 0.5
    - pnn50: 0.5
    - lf power, hf power, ratio: 0.75
    '''
    hrv_per_method = {}
    for method,nni_rois in nni_methods.items():
        hrv_per_lm = []
        for nni in nni_rois:
            hrv_per_lm.append(estimate_hrv(nni, fps, freq))
        hrv_per_method[method] = hrv_per_lm

    return hrv_per_method

multiple_peaks_finding(bvp_signal, trim_last=False)

Finds multiple peaks in the BVP (Blood Volume Pulse) signal.

bvp_signal (dict): Dictionary containing BVP signal data for different methods. trim_last (bool, optional): Flag to indicate whether to trim the last peak. Defaults to False.

dict: Dictionary containing peak indices for each method.

Source code in redesign_pipeline/vital_calculations/hrv.py

def multiple_peaks_finding(bvp_signal,trim_last=False):
    '''
    Finds multiple peaks in the BVP (Blood Volume Pulse) signal.

    Parameters:
    bvp_signal (dict): Dictionary containing BVP signal data for different methods.
    trim_last (bool, optional): Flag to indicate whether to trim the last peak. Defaults to False.

    Returns:
    dict: Dictionary containing peak indices for each method.

    '''

    #https://www.baeldung.com/cs/signal-peak-detection

    peaks_for_method = {}

    for k,src_signals_roi in bvp_signal.items():
        peaks_for_rois = []
        for src_signal in src_signals_roi:
            peaks_for_rois.append(peak_finder(src_signal,trim_last))

        peaks_for_method[k] = peaks_for_rois
    return peaks_for_method

multiple_peaks_finding_scipy(bvp_signal, trim_last=False)

Finds multiple peaks in the BVP (Blood Volume Pulse) signal using the scipy library.

bvp_signal (dict): Dictionary containing BVP signal data for different methods. trim_last (bool, optional): Flag to indicate whether to trim the last peak. Defaults to False.

dict: Dictionary containing peak indices for each method.

Source code in redesign_pipeline/vital_calculations/hrv.py

def multiple_peaks_finding_scipy(bvp_signal,trim_last=False):
    '''
    Finds multiple peaks in the BVP (Blood Volume Pulse) signal using the scipy library.

    Parameters:
    bvp_signal (dict): Dictionary containing BVP signal data for different methods.
    trim_last (bool, optional): Flag to indicate whether to trim the last peak. Defaults to False.

    Returns:
    dict: Dictionary containing peak indices for each method.

    '''

    peaks_for_method = {}

    for k,src_signals_roi in bvp_signal.items():
        peaks_for_rois = []
        for src_signal in src_signals_roi:

            peaks_for_rois.append(find_peaks(src_signal,distance=10,height=0.8)[0][1:-1] if trim_last else find_peaks(src_signal,distance=10,height=0.8)[0])

        peaks_for_method[k] = peaks_for_rois
    return peaks_for_method

nni_from_peaks_indices(peaks_indices)

Calculates NNI (Normal-to-Normal Interval) from peak indices for multiple methods.

peaks_indices (dict): Dictionary containing peak indices for different methods.

dict: Dictionary containing NNI values for each method.

Source code in redesign_pipeline/vital_calculations/hrv.py

def nni_from_peaks_indices(peaks_indices):
    '''
    Calculates NNI (Normal-to-Normal Interval) from peak indices for multiple methods.

    Parameters:
    peaks_indices (dict): Dictionary containing peak indices for different methods.

    Returns:
    dict: Dictionary containing NNI values for each method.

    '''
    nni_per_method = {}
    for method,peaks in peaks_indices.items():
        nni_per_lm = []
        for peaks_per_lm in peaks:
            nni_per_lm.append(nni_per_signal(peaks_per_lm))

        nni_per_method[method] = nni_per_lm

    return nni_per_method

nni_per_signal(peaks)

Calculates NNI (Normal-to-Normal Interval) from peak indices.

peaks (array): Array containing peak indices.

array: Array containing NNI values.

Source code in redesign_pipeline/vital_calculations/hrv.py

def nni_per_signal(peaks):
    '''
    Calculates NNI (Normal-to-Normal Interval) from peak indices.

    Parameters:
    peaks (array): Array containing peak indices.

    Returns:
    array: Array containing NNI values.

    '''
    nni = []
    peaks = np.array(peaks)
    for n in range(peaks.shape[0]):
        if n == 0:
            continue
        else:
            nni.append(peaks[n] - peaks[n-1])
    return np.array(nni)

nni_threshold(nni, threshold)

Applies thresholding to NNI (Normal-to-Normal Interval) data.

nni (array): NNI data. threshold (float): Threshold value.

array: NNI data after thresholding.

Source code in redesign_pipeline/vital_calculations/hrv.py

def nni_threshold(nni, threshold):
    '''
    Applies thresholding to NNI (Normal-to-Normal Interval) data.

    Parameters:
    nni (array): NNI data.
    threshold (float): Threshold value.

    Returns:
    array: NNI data after thresholding.

    '''
    nni_copy = np.array(nni)
    #set to 3std, can change in the future
    mean = np.nanmean(nni_copy)
    std = np.std(nni_copy)
    lower = mean - threshold*std
    upper = mean + threshold*std

    for n in range(nni_copy.shape[0]):
        if n < 5:
            if (nni_copy[n] < lower):
                nni_copy[n] = mean
            if (nni_copy[n] > upper):
                if n == 1:
                    continue
                nni_copy[n] = np.mean(nni_copy[0:n-1])

            continue
        if (nni_copy[n] < lower): #false positive
            nni_copy[n] = mean
        if (nni_copy[n] > upper): #false negative
            prev_5 = nni_copy[n-5:n]
            nni_copy[n] = np.mean(prev_5)

    return nni_copy

peak_finder(src_signal, trim_last=True)

Finds peak indices in a source signal.

src_signal (array): Source signal. trim_last (bool, optional): Flag to indicate whether to trim the last peak. Defaults to True.

array: Array containing peak indices.

Source code in redesign_pipeline/vital_calculations/hrv.py

def peak_finder(src_signal,trim_last=True):

    '''
    Finds peak indices in a source signal.

    Parameters:
    src_signal (array): Source signal.
    trim_last (bool, optional): Flag to indicate whether to trim the last peak. Defaults to True.

    Returns:
    array: Array containing peak indices.

    '''

    peak_indices = []
    peak_index = -1
    peak_value = -1
    baseline = np.nanmean(src_signal)

    #peak must be at least > min_peak_height
    # min_peak_height = np.nanmean(src_signal) #+ 1*abs(np.std(src_signal))

    #ensure np array
    src_signal = np.array(src_signal)

    for index, val in enumerate(src_signal):
        if val > baseline:
            if peak_value == -1 or val > peak_value:
                peak_index = index
                peak_value = val
        elif val < baseline and peak_index != -1:
            #if peak_value > min_peak_height:
            peak_indices.append(peak_index)
            peak_index = -1
            peak_value = -1

    if peak_index != -1:
        peak_indices.append(peak_index)

    if trim_last:
        peak_indices = peak_indices[1:-1]

    return np.array(peak_indices)

reject_outliers(data, m=5.0)

Removes outliers from the data using the modified Z-score method.

data (array): Data array. m (float, optional): Z-score threshold. Defaults to 5.0.

array: Data array with outliers removed.

Source code in redesign_pipeline/vital_calculations/hrv.py

def reject_outliers(data, m = 5.0):
    '''
    Removes outliers from the data using the modified Z-score method.

    Parameters:
    data (array): Data array.
    m (float, optional): Z-score threshold. Defaults to 5.0.

    Returns:
    array: Data array with outliers removed.

    '''

    data = np.array(data)
    d = np.abs(data - np.median(data))
    mdev = np.median(d)
    s = d/mdev if mdev else np.zeros(len(d))
    return data[s<m]

reject_outliers_fill(data, m=5.0)

Replaces outliers in the data with the median value using the modified Z-score method.

data (array): Data array. m (float, optional): Z-score threshold. Defaults to 5.0.

array: Data array with outliers replaced by the median value.

Source code in redesign_pipeline/vital_calculations/hrv.py

def reject_outliers_fill(data, m = 5.0):
    '''
    Replaces outliers in the data with the median value using the modified Z-score method.

    Parameters:
    data (array): Data array.
    m (float, optional): Z-score threshold. Defaults to 5.0.

    Returns:
    array: Data array with outliers replaced by the median value.

    '''

    data = np.array(data)
    median_value = np.median(data)
    d = np.abs(data - median_value)
    mdev = np.median(d)
    s = d/mdev if mdev else np.zeros(len(d))
    for i in range(len(d)):
        if s[i]>=m:
            data[i] = median_value
    return data

reject_outliers_multimodel(data, m=5.0, replace_median=False)

Removes outliers from data for multiple models using the modified Z-score method.

data (dict): Dictionary containing data for different methods. m (float, optional): Z-score threshold. Defaults to 5.0. replace_median (bool, optional): Flag to indicate whether to replace outliers with the median value. Defaults to False.

dict: Dictionary containing data with outliers removed.

Source code in redesign_pipeline/vital_calculations/hrv.py

def reject_outliers_multimodel(data, m = 5.0, replace_median=False):
    '''
    Removes outliers from data for multiple models using the modified Z-score method.

    Parameters:
    data (dict): Dictionary containing data for different methods.
    m (float, optional): Z-score threshold. Defaults to 5.0.
    replace_median (bool, optional): Flag to indicate whether to replace outliers with the median value. Defaults to False.

    Returns:
    dict: Dictionary containing data with outliers removed.

    '''
    data_for_method = {}

    for k,data_rois in data.items():
        data_for_rois = []
        for data in data_rois:
            data_for_rois.append(reject_outliers_fill(data,m) if replace_median else reject_outliers(data,m))

        data_for_method[k] = data_for_rois
    return data_for_method