Signal

Signal processing utilities that make complete sense

---

source

iqr_normalization

def iqr_normalization(
    waveform_array, is_spo2:bool=False
):

Call self as a function.

---

source

iir_filter

def iir_filter(
    waveform_array, freq_range, btype, order:int=16, fs:int=128
):

as described in https://www.researchsquare.com/article/rs-6307069/v1

---

source

resample_waveform

def resample_waveform(
    waveform_array, fs_in, fs_out, is_spo2:bool=False
):

Call self as a function.

---

source

baseline_filter

def baseline_filter(
    data
):

Call self as a function.

---

source

high_frequency_noise_filter

def high_frequency_noise_filter(
    data
):

Call self as a function.

---

source

butterworth

def butterworth(
    waveform_array, freq_range, btype, fs:int=128, order:int=4, # Recommend playing around with the order as well
):

Call self as a function.

---

source

stft

def stft(
    signal_array, n_fft:int=256, # number of ffts to perform
    win_length:int=250, # window of ffts
    pad_mode:str='reflect',
    pad_win_length_to_nfft:bool=True, # indicator to pad the end of the sequence with 0s for nfft-win_legnth to ensuure correct size output
    center:bool=False, hop_length:int=125, normalized:bool=True, decibel_scale:bool=False, return_complex:bool=True,
    onesided:bool=True, channel_stft_means:NoneType=None, # precalculated stft channel means
    channel_stft_stds:NoneType=None, # precalculated stft channel stds
):

in: [bs x n_vars x max_seq_len] out: [bs x n_vars x n_fft // 2 + 1 x stft_len]

---

source

preprocess_abp_signal

def preprocess_abp_signal(
    abp_signal, fs
):

ABP preprocessing with all features and quality assessment.

returns averaged sbp, dbp, map, hr for given abp_signal

---

source

jSQI_complete

def jSQI_complete(
    features, onset, abp, fs:int=125
):

ABP waveform signal quality index calculation. From physionet: https://physionet.org/content/cardiac-output/1.0.0/code/2analyze/jSQI.m

Args: features: Features extracted from ABP (nx12 array from abpfeature function) onset: Onset times of ABP beats abp: Arterial blood pressure waveform fs: Sampling frequency (default 125 Hz)

Returns: BeatQ: SQI of each beat (nx10 array): 0=good, 1=bad Col 0: logical OR of cols 1 thru 9 Col 1: P not physiologic (<20 or >300 mmHg) Col 2: MAP not physiologic (<30 or >200 mmHg) Col 3: HR not physiologic (<20 or >200 bpm) Col 4: PP not physiologic (<20 mmHg) Col 5: abnormal Psys (beat-to-beat change > 20 mmHg) Col 6: abnormal Pdias (beat-to-beat change > 20 mmHg) Col 7: abnormal period (beat-to-beat change > 1/2 sec) Col 8: abnormal P(onset) (beat-to-beat change > 20 mmHg) Col 9: noisy beat (mean of negative dP < -3) r: fraction of good beats in ABP

---

source

localfun_area

def localfun_area(
    abp, onset, end_sys, P_dias, fs:int=125
):

Helper function to calculate systolic area. From physionet: https://physionet.org/content/cardiac-output/1.0.0/code/2analyze/abpfeature.m

Args: abp: ABP signal onset: Onset times end_sys: End of systole times P_dias: Diastolic pressures fs: Sampling frequency

Returns: sys_area: Systolic area [mmHg*sec]

---

source

abp_features

def abp_features(
    abp, onset_times, fs:int=125
):

ABP waveform feature extractor. From physionet: https://physionet.org/content/cardiac-output/1.0.0/code/2analyze/abpfeature.m

Args: abp: ABP waveform (sampled at fs Hz) onset_times: Array of onset times in samples fs: Sampling frequency (default 125 Hz)

Returns: features: Array with beat-to-beat ABP features (12 columns) Col 0: Time of systole [samples] Col 1: Systolic BP [mmHg] Col 2: Time of diastole [samples] Col 3: Diastolic BP [mmHg] Col 4: Pulse pressure [mmHg] Col 5: Mean pressure [mmHg] Col 6: Beat Period [samples] Col 7: mean_dyneg Col 8: End of systole time 0.3sqrt(RR) method Col 9: Area under systole 0.3sqrt(RR) method Col 10: End of systole time 1st min-slope method Col 11: Area under systole 1st min-slope method Col 12: HR

---

source

wabp_onset_detector

def wabp_onset_detector(
    abp_signal, fs:int=125
):

ABP onset detector adapted from physionet at: https://physionet.org/content/cardiac-output/1.0.0/code/2analyze/wabp.m Detects onset of each beat in ABP waveform.

This was written for a 125 Hz ABP signal. Specific params could likely be adjusted for different frequencies.

Args: abp_signal: ABP waveform in mmHg fs: sampling frequency (default 125 Hz)

Returns: onset_indices: array of onset sample indices

---

source

check_ecg_signal_quality

def check_ecg_signal_quality(
    ecg_signal, fs, filter_ecg:bool=True
):

Call self as a function.

---

source

extract_heartbeats

def extract_heartbeats(
    signal, # Input ECG signal.
    rpeaks, # R-peak location indices.
    before:int=200, # Number of samples to include before the R peak.
    after:int=400, # Number of samples to include after the R peak.
): # List of (start_index, end_index) tuples for each heartbeat.

Extract heartbeat indices. Adapted from biosppy

---

source

ecg_clean_beats

def ecg_clean_beats(
    signal, # Raw ECG signal.
    fs, # Sampling frequency (Hz).
    new_fs:NoneType=None, # Desired sampling frequency for output beat indices. If None, will use original fs.
    filter_ecg:bool=False, # Whether to apply ECG filtering before beat detection. Default is False.
): # List of (start_index, end_index) tuples for each heartbeat.

Process a raw ECG signal and extract relevant signal features using default parameters. Adapted from biosppy.ecg.ecg

---

source

clean_ecg_signal

def clean_ecg_signal(
    signal, fs
):

Process a raw ECG signal and extract relevant signal features using biosppy.

# import numpy as np
# from sleepecg import detect_heartbeats, get_toy_ecg

# ecg, fs = get_toy_ecg()  # 5 min of ECG data at 360 Hz
# beats = detect_heartbeats(ecg, fs)

# temp, beat_indices = ecg_clean_beats(ecg, fs, filter_ecg=True)

# import matplotlib.pyplot as plt
# plt.figure(figsize=(12, 4))
# plt.plot(temp[beat_indices[0]:beat_indices[1]], label='Raw ECG')

# ecg_length = len(temp) / fs  # in seconds
# n_beats = len(beat_indices) / 2
# bpm = n_beats / (ecg_length / 60)
# if bpm < 20 or bpm > 250:
#     raise ValueError(f"ECG quality check failed, bpm out of range: {bpm}")

---

source

create_patch_b2b

def create_patch_b2b(
    xb, beat_indices, resample_length:int=256, fill_missing_beats:bool=False
):

xb: [n_vars x seq_len] out: [num_patch x n_vars x patch_len]

# import zarr, torch
# import neurokit2 as nk

# rt = zarr.open('/Users/benfox/Downloads/MICU_BED09-1662509845.zarr')

# frequency = 125
# channel_frequency = rt['II'].attrs.get('sample_rate')
# start_index = 4102500 / frequency * channel_frequency
# end_index = 4140000 / frequency * channel_frequency
# ecg_signal = rt['II'][int(start_index):int(end_index)]
# ecg_signal = nk.ecg.ecg_simulate(duration=50, length=None, sampling_rate=channel_frequency)
# print(check_ecg_signal_quality(ecg_signal, fs=channel_frequency, filter_ecg=True))
# xb = torch.from_numpy(ecg_signal).unsqueeze(0)
# l = create_patch_b2b(xb.numpy(), ecg_index=0, frequency=channel_frequency, resample_length=256, fill_missing_beats=False, segmenter='hamilton')
# print(len(l), l.shape)

True
56 (56, 1, 256)