cibrrig.preprocess.physiology ============================= .. py:module:: cibrrig.preprocess.physiology .. autoapi-nested-parse:: Code to process physiological signals. This module is general in that it does no I/O, only processing. Arbitrarily sourced data that has been shaped into numpy arrays can be processed Functions --------- .. autoapisummary:: cibrrig.preprocess.physiology.burst_stats_dia cibrrig.preprocess.physiology.remove_EKG cibrrig.preprocess.physiology._remove_EKG_explicit cibrrig.preprocess.physiology._remove_EKG_inferred cibrrig.preprocess.physiology.get_hr_from_dia cibrrig.preprocess.physiology.extract_hr_from_ekg cibrrig.preprocess.physiology.compute_avg_hr cibrrig.preprocess.physiology.compute_sighs cibrrig.preprocess.physiology.compute_obstructive_apneas cibrrig.preprocess.physiology.compute_dia_phase Module Contents --------------- .. py:function:: burst_stats_dia(integrated, sr, dia_thresh=1, rel_height=0.9, min_distance=0.1, min_width=0.025) Calculate diaphragm burst features. :param integrated: Integrated diaphragm trace. :type integrated: np.ndarray :param sr: Sample rate of the integrated diaphragm trace. :type sr: float :param dia_thresh: Prominence threshold (z-score). Defaults to 1. :type dia_thresh: int, optional :param rel_height: Relative height to use when finding onset times. Defaults to 0.9. :type rel_height: float, optional :param min_distance: Minimum time between breaths in seconds. Effectively sets the maximum breathing rate. Defaults to 0.1. :type min_distance: float, optional :param min_width: Minimum time duration to consider a breath in seconds. Defaults to 0.025. :type min_width: float, optional :returns: DataFrame containing burst statistics. :rtype: pd.DataFrame .. py:function:: remove_EKG(x, sr, thresh=2, heartbeats=None) Remove EKG artifacts from the diaphragm signal. Can either infer the heartbeat times from the signal to clean (e.g., diaphragm), or take an explicit set of heartbeat times Remove the EKG from an ephys trace using a BGM classifier :param signal: Input diaphragm signal. :type signal: np.ndarray :param sr: Sampling rate of the input signal. :type sr: float :param thresh: Threshold for EKG detection. Defaults to 2. :type thresh: float, optional :param heartbeats: Precomputed heartbeats. Defaults to None. :type heartbeats: np.ndarray, optional :returns: A tuple containing: - np.ndarray: Diaphragm signal with EKG artifacts removed. - np.ndarray: Detected heartbeats. :rtype: tuple .. py:function:: _remove_EKG_explicit(x, sr, heartbeats) Remove the EKG signal from an ephys trace using a Bayesian Gaussian Mixture (BGM) classifier. This function removes EKG artifacts from an electrophysiological signal by using pre-detected heartbeat times. It extracts segments around each heartbeat. It then classifies them as during a breath or not using a BGM classifier, It then subtracts the EKG artifact from the original signal. :param x: Signal to remove EKG from. :type x: np.ndarray :param sr: Sampling rate of the signal in samples per second. :type sr: float :param heartbeats: Times in seconds of the detected heartbeats. :type heartbeats: np.ndarray :returns: Signal with EKG artifacts removed. :rtype: np.ndarray .. py:function:: _remove_EKG_inferred(x, sr, thresh) Remove EKG artifacts from an ephys trace by inferring heartbeat times. This function applies a bandpass filter [5,500] to isolate the EKG signal, detects heartbeats, and then removes the EKG artifacts from the original signal. :param x: Ephys trace to clean (e.g., diaphragm or other EMG). :type x: np.ndarray :param sr: Sampling rate in samples per second. :type sr: float :param thresh: Threshold in standard deviations to detect a heartbeat. Defaults to 2. :type thresh: int, optional :returns: A tuple containing: - np.ndarray: Ephys trace with the EKG filtered out. Has the same timestamps as the input trace. - np.ndarray: Indices of the detected heartbeats. :rtype: tuple .. py:function:: get_hr_from_dia(pks, dia_df, sr) Compute heart rate from diaphragm signal. Not as good as getting it from a dedicated channel :param pulse: Pulse signal. :type pulse: np.ndarray :param dia_df: DataFrame containing diaphragm burst statistics. :type dia_df: pd.DataFrame :param sr: Sampling rate of the input signal. :type sr: float :returns: A tuple containing: - np.ndarray: Heart rate values. - np.ndarray: Detected heartbeats. :rtype: tuple .. py:function:: extract_hr_from_ekg(x, sr, thresh=5, min_distance=0.05, low=100, high=1000, in_samples=False, filter=False) Finds heartbeats in an EKG trace using a bandpass filter and peak detection. :param x: EKG trace. :type x: np.ndarray :param sr: Sampling rate in samples per second. :type sr: float :param thresh: Prominence threshold in standard deviations to identify peaks. Defaults to 5. :type thresh: int, optional :param min_distance: Minimum time between detected heartbeats in seconds. Effectively sets a maximum heart rate. Defaults to 0.05. :type min_distance: float, optional :param low: Low cut frequency in the bandpass filter (Hz). Defaults to 100. :type low: int, optional :param high: High cut frequency in the bandpass filter (Hz). Defaults to 1000. :type high: int, optional :param in_samples: If True, return the peaks as sample indices. If False, return the peaks as time in seconds. Defaults to False. :type in_samples: bool, optional :param filter: If True, apply a bandpass filter to the EKG signal. Defaults to False. :type filter: bool, optional :returns: Detected heartbeats. If `in_samples` is True, returns the sample indices of the heartbeats. Otherwise, returns the time in seconds of the heartbeats. :rtype: np.ndarray .. py:function:: compute_avg_hr(heartbeats, smoothing_window='10s', dt=0.1, t_target=None) Compute the mean heart rate from detected hearteats using a sliding average window. First performs a median filter to remove large instantaneous outliers Then performs a mean filter to smooth out the average. Can pass t_target to interpolate to a new time basis. Otherwise defualts to a 100ms window If both t_target and dt are None - outputs the smoothed heart rate only with length (# heartbeats) Created with the help of ChatGPT3.5 :param heartbeats: Timestamps of detected heartbeats. :type heartbeats: np.ndarray :param smoothing_window: Smoothing window for the heart rate computation. Defaults to "10s". :type smoothing_window: str, optional :param dt: Time step for the heart rate computation. Defaults to 0.1. :type dt: float, optional :param t_target: Target time vector for the heart rate computation. Defaults to None. :type t_target: np.ndarray, optional :returns: A tuple containing: - np.ndarray: Time vector for the average heart rate. - np.ndarray: Smoothed heart rate values. :rtype: tuple .. py:function:: compute_sighs(breath_times, auc, thresh=7, win='20s') Use a rolling Median Absolute Deviation (MAD) to identify sighs based on the AUC of the diaphragm. This function calculates the rolling MAD of the diaphragm's area under the curve (AUC) and identifies breaths that exceed a specified threshold as sighs. :param breath_times: Times of each breath in seconds. :type breath_times: np.ndarray :param auc: Area under the curve of the diaphragm for each breath. :type auc: np.ndarray :param thresh: Multiplier for the MAD to set the threshold for sigh detection. Defaults to 7. :type thresh: int, optional :param win: Window size for the rolling MAD calculation. Defaults to '20s'. :type win: str, optional :returns: Boolean array indicating which breaths are identified as sighs. :rtype: np.ndarray .. py:function:: compute_obstructive_apneas(breath_times, inhale_onsets) Identify obstructive apneas based on the absence of detected inhale onsets. This function determines periods of obstructive apnea by checking for missing inhale onsets while breath onsets are present. :param breath_times: Times of breath onsets (not used, but required to ensure diaphragm data is provided). :type breath_times: np.ndarray :param inhale_onsets: Times of inhale onsets. :type inhale_onsets: np.ndarray :returns: Boolean array indicating where obstructive apneas are detected. :rtype: np.ndarray .. py:function:: compute_dia_phase(ons, offs=None, t_start=0, t_stop=None, dt=1 / 1000, transform=True) Computes breathing phase based on the diaphragm signal Phase is [0,1] where 0 is diaphragm onset, 0.5 is diaphragm offset, and 1 is diaphragm onset again, - NB: Technically can generalize to any on/off signal, but standard usage should be diaphragm - By default, the phase is transformed to the range [-pi, pi] If no offset is given, the phase is linearly spaced between onsets If no stop time is given, the phase is computed until the last offset Defaults to a signal sampled at 1kHz :param ons: Onset times of the diaphragm bursts. :type ons: np.ndarray :param offs: Offset times of the diaphragm bursts. Defaults to None. :type offs: np.ndarray, optional :param t_start: Start time for the phase computation. Defaults to 0. :type t_start: float, optional :param t_stop: Stop time for the phase computation. Defaults to None. :type t_stop: float, optional :param dt: Time step for the output phase signal. Defaults to 1/1000. :type dt: float, optional :param transform: If True, transform the phase to the range [-pi, pi]. Defaults to True. :type transform: bool, optional :returns: A tuple containing: - np.ndarray: Time vector for the phase. - np.ndarray: Phase values. :rtype: tuple