11545366242

Committed 28 Oct 2024 12:54AM UTC coverage: 90.288% (-0.8%) from 91.103%

Build # 11545366242

Build Type

push

github

Committed by

Koaha

Commit Message

fix broken workflow

Run Details

1263 of 1606 branches covered (78.64%)

Branch coverage included in aggregate %.

5 of 12 new or added lines in 2 files covered. (41.67%)

143 existing lines in 3 files now uncovered.

5607 of 6003 relevant lines covered (93.4%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.98

/src/vitalDSP/utils/peak_detection.py

import numpy as np
from vitalDSP.utils.common import find_peaks, filtfilt, argrelextrema
from vitalDSP.utils.scaler import StandardScaler
from vitalDSP.filtering.signal_filtering import SignalFiltering


class PeakDetection:
    """
    A class to detect peaks in various physiological signals like ECG, PPG, EEG, respiratory, and arterial blood pressure (ABP) signals.

    Methods
    -------
    detect_peaks : function
        Detects the peaks in the signal using the selected method.
    """

    def __init__(self, signal, method="threshold", **kwargs):
        """
        Initialize the PeakDetection class with the signal and the selected peak detection method.

        Parameters
        ----------
        signal : numpy.ndarray
            The input signal to be analyzed for peak detection.
        method : str, optional
            The method to use for peak detection. Default is "threshold".
            Available methods are:
            - 'threshold'
            - 'savgol'
            - 'gaussian'
            - 'rel_extrema'
            - 'scaler_threshold'
            - 'ecg_r_peak'
            - 'ecg_derivative'
            - 'ppg_first_derivative'
            - 'ppg_second_derivative'
            - 'eeg_wavelet'
            - 'eeg_bandpass'
            - 'resp_autocorrelation'
            - 'resp_zero_crossing'
            - 'abp_systolic'
            - 'abp_diastolic'
        kwargs : dict
            Additional parameters specific to the selected method.
        """
        self.signal = signal
        self.method = method
        # Extract relevant parameters from kwargs and set as instance attributes
        self.height = kwargs.get("height", None)
        self.distance = kwargs.get("distance", None)
        self.threshold = kwargs.get("threshold", None)
        self.prominence = kwargs.get("prominence", None)
        self.width = kwargs.get("width", None)
        self.search_window = kwargs.get("search_window", 4)
        self.window_length = kwargs.get("window_length", 11)
        self.polyorder = kwargs.get("polyorder", 2)
        self.sigma = kwargs.get("sigma", 1)
        self.order = kwargs.get("order", 1)
        self.lowcut = kwargs.get("lowcut", 0.5)
        self.highcut = kwargs.get("highcut", 50)
        self.fs = kwargs.get("fs", 100.0)
        self.window_size = kwargs.get("window_size", 7)
        self.threshold_factor = kwargs.get("threshold_factor", 1.2)
        self.kwargs = kwargs

    def detect_peaks(self):
        """
        Detect peaks in the signal based on the selected method.

        Returns
        -------
        peaks : numpy.ndarray
            The indices of the detected peaks.

        Raises
        ------
        ValueError
            If an invalid method is selected.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="threshold", threshold=4)
        >>> peaks = detector.detect_peaks()
        >>> print(peaks)
        """
        method_dict = {
            "threshold": self._threshold_based_detection,
            "savgol": self._savgol_based_detection,
            "gaussian": self._gaussian_based_detection,
            "rel_extrema": self._relative_extrema_detection,
            "scaler_threshold": self._scaler_threshold_based_detection,
            "ecg_r_peak": self._ecg_r_peak_detection,
            "ecg_derivative": self._ecg_derivative_detection,
            "ppg_systolic_peaks": self._ppg_systolic_peak_detection,
            "ppg_first_derivative": self._ppg_first_derivative_detection,
            "ppg_second_derivative": self._ppg_second_derivative_detection,
            "eeg_wavelet": self._eeg_wavelet_detection,
            "eeg_bandpass": self._eeg_bandpass_detection,
            "resp_autocorrelation": self._resp_autocorrelation_detection,
            "resp_zero_crossing": self._resp_zero_crossing_detection,
            "abp_systolic": self._abp_systolic_peak_detection,
            "abp_diastolic": self._abp_diastolic_peak_detection,
        }
        if self.method not in method_dict:
            raise ValueError(f"Invalid method '{self.method}' selected.")
        return method_dict[self.method]()

    def _ppg_systolic_peak_detection(self):
        """
        Detect systolic peaks in PPG signals.

        This method enhances systolic peaks by taking the first derivative of the PPG signal, squaring it to emphasize the peaks,
        and applying a moving window integrator.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of systolic peaks detected in the PPG signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 3, 2, 1, 2, 3, 4, 3])
        >>> detector = PeakDetection(signal, method="ppg_systolic_peak")
        >>> peaks = detector.detect_peaks()
        >>> print(peaks)
        """
        # First derivative of the signal
        # Step 1: Set adaptive height and prominence thresholds to isolate systolic peaks.
        mean_signal = np.mean(self.signal)
        std_signal = np.std(self.signal)
        height_threshold = mean_signal + (self.threshold_factor * std_signal)
        prominence_threshold = self.prominence
        # Step 3: Minimum distance constraint (e.g., 30% of the sampling rate)
        min_distance = max(self.distance, int(0.3 * self.fs))
        # Step 4: Detect peaks based on height and prominence thresholds.
        peaks = find_peaks(
            self.signal,
            height=height_threshold,
            distance=min_distance,
            prominence=prominence_threshold,
            width=self.width,
        )

        # Step 2: Return refined peaks to ensure they align with systolic peaks.
        return self._refine_peaks(peaks)

    def _refine_peaks(self, peaks):
        """
        Refine detected peaks by searching for the local maximum within a window.

        Parameters
        ----------
        peaks : numpy.ndarray
            Initial peak indices detected.
        search_window : int, optional
            Number of samples around each peak to search for the true maximum.

        Returns
        -------
        refined_peaks : numpy.ndarray
            Refined peak indices.
        """
        search_window = self.kwargs.get(
            "search_window", 4
        )  # Default search window size is 4
        refined_peaks = []

        for peak in peaks:
            start = max(0, peak - search_window)
            end = min(len(self.signal), peak + search_window)
            true_peak = np.argmax(self.signal[start:end]) + start
            refined_peaks.append(true_peak)

        return np.array(refined_peaks)

    def _threshold_based_detection(self):
        """
        Detect peaks using a simple thresholding approach.

        This method identifies peaks in the signal that exceed a specified threshold and are separated by a minimum distance.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected based on the threshold.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="threshold", threshold=4, distance=50)
        >>> peaks = detector._threshold_based_detection()
        >>> print(peaks)
        """
        peaks = find_peaks(
            self.signal,
            height=self.height,
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _savgol_based_detection(self):
        """
        Detect peaks using the Savitzky-Golay filter to smooth the signal.

        The Savitzky-Golay filter is applied to the signal to reduce noise and highlight peaks, which are then detected.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected after smoothing with the Savitzky-Golay filter.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="savgol", window_length=11, polyorder=2)
        >>> peaks = detector._savgol_based_detection()
        >>> print(peaks)
        """
        smoothed_signal = SignalFiltering.savgol_filter(
            self.signal, window_length=self.window_length, polyorder=self.polyorder
        )
        peaks = find_peaks(
            smoothed_signal,
            height=self.height,
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _gaussian_based_detection(self):
        """
        Detect peaks by smoothing the signal with a Gaussian filter.

        Gaussian filtering is applied to smooth the signal, making it easier to identify peaks.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected after Gaussian filtering.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="gaussian", sigma=1)
        >>> peaks = detector._gaussian_based_detection()
        >>> print(peaks)
        """
        smoothed_signal = SignalFiltering.gaussian_filter1d(
            self.signal, sigma=self.sigma
        )
        peaks = find_peaks(
            smoothed_signal,
            height=self.height,
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _relative_extrema_detection(self):
        """
        Detect relative extrema (peaks) in the signal using the specified order.

        This method identifies peaks by finding points that are higher than their neighboring values within a defined window.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected as relative extrema.

        Examples
        --------
        >>> signal = np.array([1, 3, 2, 4, 3, 5, 4])
        >>> detector = PeakDetection(signal, method="rel_extrema", order=1)
        >>> peaks = detector._relative_extrema_detection()
        >>> print(peaks)
        """
        peaks = argrelextrema(self.signal, np.greater, order=self.order)
        return self._refine_peaks(peaks)

    def _scaler_threshold_based_detection(self):
        """
        Detect peaks after scaling the signal and applying a threshold.

        The signal is standardized using a scaler, and peaks are detected based on the standardized values exceeding a threshold.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected after scaling and thresholding.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="scaler_threshold", threshold=1, distance=50)
        >>> peaks = detector._scaler_threshold_based_detection()
        >>> print(peaks)
        """
        scaler = StandardScaler()
        standardized_signal = scaler.fit_transform(self.signal)
        peaks = find_peaks(
            standardized_signal,
            height=self.height,
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _ecg_r_peak_detection(self):
        """
        Detect R-peaks in ECG signals using the derivative and squared signal method.

        This method enhances the R-peaks by taking the derivative of the ECG signal, squaring it, and applying a moving window integrator.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of R-peaks detected in the ECG signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 1, 2, 1, 2])
        >>> detector = PeakDetection(signal, method="ecg_r_peak")
        >>> peaks = detector.find_peaks()
        >>> print(peaks)
        """
        # First derivative of the signal
        diff_signal = np.diff(self.signal)

        # Cube the derivative to enhance peaks
        squared_signal = diff_signal**2

        # Apply a moving window integrator to smooth the squared signal
        integrator = np.convolve(
            squared_signal, np.ones(self.window_size) / self.window_size, mode="same"
        )

        # Set the threshold dynamically based on mean and a factor to reduce noise
        threshold = np.mean(integrator) * self.threshold_factor

        # Detect R-peaks in the processed signal
        peaks = find_peaks(
            integrator,
            height=threshold,
            distance=self.distance,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _ecg_derivative_detection(self):
        """
        Detect peaks in the ECG signal based on the absolute derivative.

        Peaks are identified by calculating the absolute value of the signal's derivative and detecting significant increases.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected in the ECG signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 1, 2, 1, 2])
        >>> detector = PeakDetection(signal, method="ecg_derivative")
        >>> peaks = detector._ecg_derivative_detection()
        >>> print(peaks)
        """
        diff_signal = np.diff(self.signal)
        peaks = find_peaks(
            np.abs(diff_signal),
            height=np.mean(np.abs(diff_signal)),
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _ppg_first_derivative_detection(self):
        """
        Detect peaks in PPG signals using the first derivative.

        This method identifies peaks by calculating the first derivative of the PPG signal and detecting points of significant positive change.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected in the first derivative of the PPG signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="ppg_first_derivative")
        >>> peaks = detector._ppg_first_derivative_detection()
        >>> print(peaks)
        """
        diff_signal = np.diff(self.signal)
        peaks = find_peaks(
            diff_signal,
            height=np.mean(diff_signal),
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _ppg_second_derivative_detection(self):
        """
        Detect peaks in PPG signals using the second derivative.

        This method enhances the detection of peaks by calculating the second derivative of the PPG signal, which highlights inflection points.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected in the second derivative of the PPG signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="ppg_second_derivative")
        >>> peaks = detector._ppg_second_derivative_detection()
        >>> print(peaks)
        """
        diff_signal = np.diff(self.signal, n=2)
        peaks = find_peaks(
            diff_signal,
            height=np.mean(diff_signal),
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _eeg_wavelet_detection(self):
        """
        Detect peaks in EEG signals using wavelet transformation.

        The signal is transformed using wavelets, and peaks are detected based on the transformed coefficients.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected in the wavelet-transformed EEG signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 1, 2, 1, 2])
        >>> detector = PeakDetection(signal, method="eeg_wavelet")
        >>> peaks = detector._eeg_wavelet_detection()
        >>> print(peaks)
        """
        wavelet_coeffs = np.abs(np.convolve(self.signal, np.ones(10), "same"))
        peaks = find_peaks(
            wavelet_coeffs,
            height=np.mean(wavelet_coeffs),
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _eeg_bandpass_detection(self):
        """
        Detect peaks in EEG signals after applying a bandpass filter.

        A bandpass filter is applied to the signal to isolate specific frequency ranges before peak detection.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected in the bandpass-filtered EEG signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 1, 2, 1, 2])
        >>> detector = PeakDetection(signal, method="eeg_bandpass", lowcut=0.5, highcut=50, fs=100)
        >>> peaks = detector._eeg_bandpass_detection()
        >>> print(peaks)
        """
        b, a = SignalFiltering.butter(
            5, [self.lowcut, self.highcut], btype="band", fs=self.fs
        )
        filtered_signal = filtfilt(b, a, self.signal)
        peaks = find_peaks(
            filtered_signal,
            height=np.mean(filtered_signal),
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _resp_autocorrelation_detection(self):
        """
        Detect peaks in respiratory signals using autocorrelation.

        Autocorrelation is used to find repeating patterns in the signal, with peaks representing periodic components.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected in the autocorrelated respiratory signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 1, 2, 1, 2])
        >>> detector = PeakDetection(signal, method="resp_autocorrelation")
        >>> peaks = detector._resp_autocorrelation_detection()
        >>> print(peaks)
        """
        autocorr = np.correlate(self.signal, self.signal, mode="full")
        peaks = find_peaks(
            autocorr,
            distance=len(self.signal) // 2,
            height=self.height,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _resp_zero_crossing_detection(self):
        """
        Detect peaks in respiratory signals using zero-crossing method.

        Peaks are detected by finding zero crossings in the signal, which correspond to points where the signal changes direction.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of peaks detected by finding zero crossings.

        Examples
        --------
        >>> signal = np.array([1, -1, 1, -1, 1, -1])
        >>> detector = PeakDetection(signal, method="resp_zero_crossing")
        >>> peaks = detector._resp_zero_crossing_detection()
        >>> print(peaks)
        """
        # Detect zero-crossings (sign changes)
        zero_crossings = np.where(np.diff(np.sign(self.signal)))[0]

        # Check if zero_crossings is non-empty and contains enough points
        if zero_crossings.size == 0:
            raise ValueError("No zero-crossings detected in the signal.")

        # Handle cases where there may be an odd number of zero-crossings
        if len(zero_crossings) % 2 != 0:
            zero_crossings = zero_crossings[
                :-1
            ]  # Ensure an even number for peak detection

        # Return every second zero-crossing as a peak
        peaks = zero_crossings[::2]

        return self._refine_peaks(peaks)

    def _abp_systolic_peak_detection(self):
        """
        Detect systolic peaks in ABP signals using Savitzky-Golay smoothing.

        Systolic peaks are identified in arterial blood pressure (ABP) signals by smoothing the signal and detecting peaks.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of systolic peaks detected in the ABP signal.

        Examples
        --------
        >>> signal = np.array([1, 2, 3, 4, 3, 2, 1])
        >>> detector = PeakDetection(signal, method="abp_systolic")
        >>> peaks = detector._abp_systolic_peak_detection()
        >>> print(peaks)
        """
        smoothed_signal = SignalFiltering.savgol_filter(
            self.signal, window_length=self.window_length, polyorder=self.polyorder
        )
        peaks = find_peaks(
            smoothed_signal,
            height=self.height,
            distance=self.distance,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

    def _abp_diastolic_peak_detection(self):
        """
        Detect diastolic peaks in ABP signals by inverting and smoothing the signal.

        Diastolic peaks are identified by inverting the arterial blood pressure (ABP) signal, smoothing it, and detecting peaks.

        Returns
        -------
        peaks : numpy.ndarray
            Indices of diastolic peaks detected in the ABP signal.

        Examples
        --------
        >>> signal = np.array([3, 2, 1, 2, 3, 4, 3])
        >>> detector = PeakDetection(signal, method="abp_diastolic")
        >>> peaks = detector._abp_diastolic_peak_detection()
        >>> print(peaks)
        """
        inverted_signal = -self.signal
        smoothed_signal = SignalFiltering.savgol_filter(
            inverted_signal, window_length=self.window_length, polyorder=self.polyorder
        )
        peaks = find_peaks(
            smoothed_signal,
            distance=self.distance,
            height=self.height,
            threshold=self.threshold,
            prominence=self.prominence,
            width=self.width,
        )
        return self._refine_peaks(peaks)

1	import numpy as np	1✔
2	from vitalDSP.utils.common import find_peaks, filtfilt, argrelextrema	1✔
3	from vitalDSP.utils.scaler import StandardScaler	1✔
4	from vitalDSP.filtering.signal_filtering import SignalFiltering	1✔
5
6
7	class PeakDetection:	1✔
8	"""
9	A class to detect peaks in various physiological signals like ECG, PPG, EEG, respiratory, and arterial blood pressure (ABP) signals.
10
11	Methods
12	-------
13	detect_peaks : function
14	Detects the peaks in the signal using the selected method.
15	"""
16
17	def __init__(self, signal, method="threshold", **kwargs):	1✔
18	"""
19	Initialize the PeakDetection class with the signal and the selected peak detection method.
20
21	Parameters
22	----------
23	signal : numpy.ndarray
24	The input signal to be analyzed for peak detection.
25	method : str, optional
26	The method to use for peak detection. Default is "threshold".
27	Available methods are:
28	- 'threshold'
29	- 'savgol'
30	- 'gaussian'
31	- 'rel_extrema'
32	- 'scaler_threshold'
33	- 'ecg_r_peak'
34	- 'ecg_derivative'
35	- 'ppg_first_derivative'
36	- 'ppg_second_derivative'
37	- 'eeg_wavelet'
38	- 'eeg_bandpass'
39	- 'resp_autocorrelation'
40	- 'resp_zero_crossing'
41	- 'abp_systolic'
42	- 'abp_diastolic'
43	kwargs : dict
44	Additional parameters specific to the selected method.
45	"""
46	self.signal = signal	1✔
47	self.method = method	1✔
48	# Extract relevant parameters from kwargs and set as instance attributes
49	self.height = kwargs.get("height", None)	1✔
50	self.distance = kwargs.get("distance", None)	1✔
51	self.threshold = kwargs.get("threshold", None)	1✔
52	self.prominence = kwargs.get("prominence", None)	1✔
53	self.width = kwargs.get("width", None)	1✔
54	self.search_window = kwargs.get("search_window", 4)	1✔
55	self.window_length = kwargs.get("window_length", 11)	1✔
56	self.polyorder = kwargs.get("polyorder", 2)	1✔
57	self.sigma = kwargs.get("sigma", 1)	1✔
58	self.order = kwargs.get("order", 1)	1✔
59	self.lowcut = kwargs.get("lowcut", 0.5)	1✔
60	self.highcut = kwargs.get("highcut", 50)	1✔
61	self.fs = kwargs.get("fs", 100.0)	1✔
62	self.window_size = kwargs.get("window_size", 7)	1✔
63	self.threshold_factor = kwargs.get("threshold_factor", 1.2)	1✔
64	self.kwargs = kwargs	1✔
65
66	def detect_peaks(self):	1✔
67	"""
68	Detect peaks in the signal based on the selected method.
69
70	Returns
71	-------
72	peaks : numpy.ndarray
73	The indices of the detected peaks.
74
75	Raises
76	------
77	ValueError
78	If an invalid method is selected.
79
80	Examples
81	--------
82	>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
83	>>> detector = PeakDetection(signal, method="threshold", threshold=4)
84	>>> peaks = detector.detect_peaks()
85	>>> print(peaks)
86	"""
87	method_dict = {	1✔
88	"threshold": self._threshold_based_detection,
89	"savgol": self._savgol_based_detection,
90	"gaussian": self._gaussian_based_detection,
91	"rel_extrema": self._relative_extrema_detection,
92	"scaler_threshold": self._scaler_threshold_based_detection,
93	"ecg_r_peak": self._ecg_r_peak_detection,
94	"ecg_derivative": self._ecg_derivative_detection,
95	"ppg_systolic_peaks": self._ppg_systolic_peak_detection,
96	"ppg_first_derivative": self._ppg_first_derivative_detection,
97	"ppg_second_derivative": self._ppg_second_derivative_detection,
98	"eeg_wavelet": self._eeg_wavelet_detection,
99	"eeg_bandpass": self._eeg_bandpass_detection,
100	"resp_autocorrelation": self._resp_autocorrelation_detection,
101	"resp_zero_crossing": self._resp_zero_crossing_detection,
102	"abp_systolic": self._abp_systolic_peak_detection,
103	"abp_diastolic": self._abp_diastolic_peak_detection,
104	}
105	if self.method not in method_dict:	1✔
106	raise ValueError(f"Invalid method '{self.method}' selected.")	1✔
107	return method_dict[self.method]()	1✔
108
109	def _ppg_systolic_peak_detection(self):	1✔
110	"""
111	Detect systolic peaks in PPG signals.
112
113	This method enhances systolic peaks by taking the first derivative of the PPG signal, squaring it to emphasize the peaks,
114	and applying a moving window integrator.
115
116	Returns
117	-------
118	peaks : numpy.ndarray
119	Indices of systolic peaks detected in the PPG signal.
120
121	Examples
122	--------
123	>>> signal = np.array([1, 2, 3, 4, 3, 2, 1, 2, 3, 4, 3])
124	>>> detector = PeakDetection(signal, method="ppg_systolic_peak")
125	>>> peaks = detector.detect_peaks()
126	>>> print(peaks)
127	"""
128	# First derivative of the signal
129	# Step 1: Set adaptive height and prominence thresholds to isolate systolic peaks.
130	mean_signal = np.mean(self.signal)	1✔
131	std_signal = np.std(self.signal)	1✔
132	height_threshold = mean_signal + (self.threshold_factor * std_signal)	1✔
133	prominence_threshold = self.prominence	1✔
134	# Step 3: Minimum distance constraint (e.g., 30% of the sampling rate)
135	min_distance = max(self.distance, int(0.3 * self.fs))	1✔
136	# Step 4: Detect peaks based on height and prominence thresholds.
137	peaks = find_peaks(	1✔
138	self.signal,
139	height=height_threshold,
140	distance=min_distance,
141	prominence=prominence_threshold,
142	width=self.width,
143	)
144
145	# Step 2: Return refined peaks to ensure they align with systolic peaks.
146	return self._refine_peaks(peaks)	1✔
147
148	def _refine_peaks(self, peaks):	1✔
149	"""
150	Refine detected peaks by searching for the local maximum within a window.
151
152	Parameters
153	----------
154	peaks : numpy.ndarray
155	Initial peak indices detected.
156	search_window : int, optional
157	Number of samples around each peak to search for the true maximum.
158
159	Returns
160	-------
161	refined_peaks : numpy.ndarray
162	Refined peak indices.
163	"""
164	search_window = self.kwargs.get(	1✔
165	"search_window", 4
166	) # Default search window size is 4
167	refined_peaks = []	1✔
168
169	for peak in peaks:	1✔
170	start = max(0, peak - search_window)	1✔
171	end = min(len(self.signal), peak + search_window)	1✔
172	true_peak = np.argmax(self.signal[start:end]) + start	1✔
173	refined_peaks.append(true_peak)	1✔
174
175	return np.array(refined_peaks)	1✔
176
177	def _threshold_based_detection(self):	1✔
178	"""
179	Detect peaks using a simple thresholding approach.
180
181	This method identifies peaks in the signal that exceed a specified threshold and are separated by a minimum distance.
182
183	Returns
184	-------
185	peaks : numpy.ndarray
186	Indices of peaks detected based on the threshold.
187
188	Examples
189	--------
190	>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
191	>>> detector = PeakDetection(signal, method="threshold", threshold=4, distance=50)
192	>>> peaks = detector._threshold_based_detection()
193	>>> print(peaks)
194	"""
195	peaks = find_peaks(	1✔
196	self.signal,
197	height=self.height,
198	distance=self.distance,
199	threshold=self.threshold,
200	prominence=self.prominence,
201	width=self.width,
202	)
203	return self._refine_peaks(peaks)	1✔
204
205	def _savgol_based_detection(self):	1✔
206	"""
207	Detect peaks using the Savitzky-Golay filter to smooth the signal.
208
209	The Savitzky-Golay filter is applied to the signal to reduce noise and highlight peaks, which are then detected.
210
211	Returns
212	-------
213	peaks : numpy.ndarray
214	Indices of peaks detected after smoothing with the Savitzky-Golay filter.
215
216	Examples
217	--------
218	>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
219	>>> detector = PeakDetection(signal, method="savgol", window_length=11, polyorder=2)
220	>>> peaks = detector._savgol_based_detection()
221	>>> print(peaks)
222	"""
223	smoothed_signal = SignalFiltering.savgol_filter(	1✔
224	self.signal, window_length=self.window_length, polyorder=self.polyorder
225	)
226	peaks = find_peaks(	1✔
227	smoothed_signal,
228	height=self.height,
229	distance=self.distance,
230	threshold=self.threshold,
231	prominence=self.prominence,
232	width=self.width,
233	)
234	return self._refine_peaks(peaks)	1✔
235
236	def _gaussian_based_detection(self):	1✔
237	"""
238	Detect peaks by smoothing the signal with a Gaussian filter.
239
240	Gaussian filtering is applied to smooth the signal, making it easier to identify peaks.
241
242	Returns
243	-------
244	peaks : numpy.ndarray
245	Indices of peaks detected after Gaussian filtering.
246
247	Examples
248	--------
249	>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
250	>>> detector = PeakDetection(signal, method="gaussian", sigma=1)
251	>>> peaks = detector._gaussian_based_detection()
252	>>> print(peaks)
253	"""
254	smoothed_signal = SignalFiltering.gaussian_filter1d(	1✔
255	self.signal, sigma=self.sigma
256	)
257	peaks = find_peaks(	1✔
258	smoothed_signal,
259	height=self.height,
260	distance=self.distance,
261	threshold=self.threshold,
262	prominence=self.prominence,
263	width=self.width,
264	)
265	return self._refine_peaks(peaks)	1✔
266
267	def _relative_extrema_detection(self):	1✔
268	"""
269	Detect relative extrema (peaks) in the signal using the specified order.
270
271	This method identifies peaks by finding points that are higher than their neighboring values within a defined window.
272
273	Returns
274	-------
275	peaks : numpy.ndarray
276	Indices of peaks detected as relative extrema.
277
278	Examples
279	--------
280	>>> signal = np.array([1, 3, 2, 4, 3, 5, 4])
281	>>> detector = PeakDetection(signal, method="rel_extrema", order=1)
282	>>> peaks = detector._relative_extrema_detection()
283	>>> print(peaks)
284	"""
285	peaks = argrelextrema(self.signal, np.greater, order=self.order)	1✔
286	return self._refine_peaks(peaks)	1✔
287
288	def _scaler_threshold_based_detection(self):	1✔
289	"""
290	Detect peaks after scaling the signal and applying a threshold.
291
292	The signal is standardized using a scaler, and peaks are detected based on the standardized values exceeding a threshold.
293
294	Returns
295	-------
296	peaks : numpy.ndarray
297	Indices of peaks detected after scaling and thresholding.
298
299	Examples
300	--------
301	>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
302	>>> detector = PeakDetection(signal, method="scaler_threshold", threshold=1, distance=50)
303	>>> peaks = detector._scaler_threshold_based_detection()
304	>>> print(peaks)
305	"""
306	scaler = StandardScaler()	1✔
307	standardized_signal = scaler.fit_transform(self.signal)	1✔
308	peaks = find_peaks(	1✔
309	standardized_signal,
310	height=self.height,
311	distance=self.distance,
312	threshold=self.threshold,
313	prominence=self.prominence,
314	width=self.width,
315	)
316	return self._refine_peaks(peaks)	1✔
317
318	def _ecg_r_peak_detection(self):	1✔
319	"""
320	Detect R-peaks in ECG signals using the derivative and squared signal method.
321
322	This method enhances the R-peaks by taking the derivative of the ECG signal, squaring it, and applying a moving window integrator.
323
324	Returns
325	-------
326	peaks : numpy.ndarray
327	Indices of R-peaks detected in the ECG signal.
328
329	Examples
330	--------
331	>>> signal = np.array([1, 2, 1, 2, 1, 2])
332	>>> detector = PeakDetection(signal, method="ecg_r_peak")
333	>>> peaks = detector.find_peaks()
334	>>> print(peaks)
335	"""
336	# First derivative of the signal
337	diff_signal = np.diff(self.signal)	1✔
338
339	# Cube the derivative to enhance peaks
340	squared_signal = diff_signal**2	1✔
341
342	# Apply a moving window integrator to smooth the squared signal
343	integrator = np.convolve(	1✔
344	squared_signal, np.ones(self.window_size) / self.window_size, mode="same"
345	)
346
347	# Set the threshold dynamically based on mean and a factor to reduce noise
348	threshold = np.mean(integrator) * self.threshold_factor	1✔
349
350	# Detect R-peaks in the processed signal
351	peaks = find_peaks(	1✔
352	integrator,
353	height=threshold,
354	distance=self.distance,
355	prominence=self.prominence,
356	width=self.width,
357	)
358	return self._refine_peaks(peaks)	1✔
359
360	def _ecg_derivative_detection(self):	1✔
361	"""
362	Detect peaks in the ECG signal based on the absolute derivative.
363
364	Peaks are identified by calculating the absolute value of the signal's derivative and detecting significant increases.
365
366	Returns
367	-------
368	peaks : numpy.ndarray
369	Indices of peaks detected in the ECG signal.
370
371	Examples
372	--------
373	>>> signal = np.array([1, 2, 1, 2, 1, 2])
374	>>> detector = PeakDetection(signal, method="ecg_derivative")
375	>>> peaks = detector._ecg_derivative_detection()
376	>>> print(peaks)
377	"""
378	diff_signal = np.diff(self.signal)	1✔
379	peaks = find_peaks(	1✔
380	np.abs(diff_signal),
381	height=np.mean(np.abs(diff_signal)),
382	distance=self.distance,
383	threshold=self.threshold,
384	prominence=self.prominence,
385	width=self.width,
386	)
387	return self._refine_peaks(peaks)	1✔
388
389	def _ppg_first_derivative_detection(self):	1✔
390	"""
391	Detect peaks in PPG signals using the first derivative.
392
393	This method identifies peaks by calculating the first derivative of the PPG signal and detecting points of significant positive change.
394
395	Returns
396	-------
397	peaks : numpy.ndarray
398	Indices of peaks detected in the first derivative of the PPG signal.
399
400	Examples
401	--------
402	>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
403	>>> detector = PeakDetection(signal, method="ppg_first_derivative")
404	>>> peaks = detector._ppg_first_derivative_detection()
405	>>> print(peaks)
406	"""
407	diff_signal = np.diff(self.signal)	1✔
408	peaks = find_peaks(	1✔
409	diff_signal,
410	height=np.mean(diff_signal),
411	distance=self.distance,
412	threshold=self.threshold,
413	prominence=self.prominence,
414	width=self.width,
415	)
416	return self._refine_peaks(peaks)	1✔
417
418	def _ppg_second_derivative_detection(self):	1✔
419	"""
420	Detect peaks in PPG signals using the second derivative.
421
422	This method enhances the detection of peaks by calculating the second derivative of the PPG signal, which highlights inflection points.
423
424	Returns
425	-------
426	peaks : numpy.ndarray
427	Indices of peaks detected in the second derivative of the PPG signal.
428
429	Examples
430	--------
431	>>> signal = np.array([1, 2, 3, 4, 5, 4, 3, 2, 1])
432	>>> detector = PeakDetection(signal, method="ppg_second_derivative")
433	>>> peaks = detector._ppg_second_derivative_detection()
434	>>> print(peaks)
435	"""
436	diff_signal = np.diff(self.signal, n=2)	1✔
437	peaks = find_peaks(	1✔
438	diff_signal,
439	height=np.mean(diff_signal),
440	distance=self.distance,
441	threshold=self.threshold,
442	prominence=self.prominence,
443	width=self.width,
444	)
445	return self._refine_peaks(peaks)	1✔
446
447	def _eeg_wavelet_detection(self):	1✔
448	"""
449	Detect peaks in EEG signals using wavelet transformation.
450
451	The signal is transformed using wavelets, and peaks are detected based on the transformed coefficients.
452
453	Returns
454	-------
455	peaks : numpy.ndarray
456	Indices of peaks detected in the wavelet-transformed EEG signal.
457
458	Examples
459	--------
460	>>> signal = np.array([1, 2, 1, 2, 1, 2])
461	>>> detector = PeakDetection(signal, method="eeg_wavelet")
462	>>> peaks = detector._eeg_wavelet_detection()
463	>>> print(peaks)
464	"""
465	wavelet_coeffs = np.abs(np.convolve(self.signal, np.ones(10), "same"))	1✔
466	peaks = find_peaks(	1✔
467	wavelet_coeffs,
468	height=np.mean(wavelet_coeffs),
469	distance=self.distance,
470	threshold=self.threshold,
471	prominence=self.prominence,
472	width=self.width,
473	)
474	return self._refine_peaks(peaks)	1✔
475
476	def _eeg_bandpass_detection(self):	1✔
477	"""
478	Detect peaks in EEG signals after applying a bandpass filter.
479
480	A bandpass filter is applied to the signal to isolate specific frequency ranges before peak detection.
481
482	Returns
483	-------
484	peaks : numpy.ndarray
485	Indices of peaks detected in the bandpass-filtered EEG signal.
486
487	Examples
488	--------
489	>>> signal = np.array([1, 2, 1, 2, 1, 2])
490	>>> detector = PeakDetection(signal, method="eeg_bandpass", lowcut=0.5, highcut=50, fs=100)
491	>>> peaks = detector._eeg_bandpass_detection()
492	>>> print(peaks)
493	"""
494	b, a = SignalFiltering.butter(	1✔
495	5, [self.lowcut, self.highcut], btype="band", fs=self.fs
496	)
497	filtered_signal = filtfilt(b, a, self.signal)	1✔
498	peaks = find_peaks(	1✔
499	filtered_signal,
500	height=np.mean(filtered_signal),
501	distance=self.distance,
502	threshold=self.threshold,
503	prominence=self.prominence,
504	width=self.width,
505	)
506	return self._refine_peaks(peaks)	1✔
507
508	def _resp_autocorrelation_detection(self):	1✔
509	"""
510	Detect peaks in respiratory signals using autocorrelation.
511
512	Autocorrelation is used to find repeating patterns in the signal, with peaks representing periodic components.
513
514	Returns
515	-------
516	peaks : numpy.ndarray
517	Indices of peaks detected in the autocorrelated respiratory signal.
518
519	Examples
520	--------
521	>>> signal = np.array([1, 2, 1, 2, 1, 2])
522	>>> detector = PeakDetection(signal, method="resp_autocorrelation")
523	>>> peaks = detector._resp_autocorrelation_detection()
524	>>> print(peaks)
525	"""
526	autocorr = np.correlate(self.signal, self.signal, mode="full")	1✔
527	peaks = find_peaks(	1✔
528	autocorr,
529	distance=len(self.signal) // 2,
530	height=self.height,
531	threshold=self.threshold,
532	prominence=self.prominence,
533	width=self.width,
534	)
535	return self._refine_peaks(peaks)	1✔
536
537	def _resp_zero_crossing_detection(self):	1✔
538	"""
539	Detect peaks in respiratory signals using zero-crossing method.
540
541	Peaks are detected by finding zero crossings in the signal, which correspond to points where the signal changes direction.
542
543	Returns
544	-------
545	peaks : numpy.ndarray
546	Indices of peaks detected by finding zero crossings.
547
548	Examples
549	--------
550	>>> signal = np.array([1, -1, 1, -1, 1, -1])
551	>>> detector = PeakDetection(signal, method="resp_zero_crossing")
552	>>> peaks = detector._resp_zero_crossing_detection()
553	>>> print(peaks)
554	"""
555	# Detect zero-crossings (sign changes)
UNCOV 556	zero_crossings = np.where(np.diff(np.sign(self.signal)))[0]	×
557
558	# Check if zero_crossings is non-empty and contains enough points
UNCOV 559	if zero_crossings.size == 0:	×
UNCOV 560	raise ValueError("No zero-crossings detected in the signal.")	×
561
562	# Handle cases where there may be an odd number of zero-crossings
UNCOV 563	if len(zero_crossings) % 2 != 0:	×
UNCOV 564	zero_crossings = zero_crossings[	×
565	:-1
566	] # Ensure an even number for peak detection
567
568	# Return every second zero-crossing as a peak
UNCOV 569	peaks = zero_crossings[::2]	×
570
UNCOV 571	return self._refine_peaks(peaks)	×
572
573	def _abp_systolic_peak_detection(self):	1✔
574	"""
575	Detect systolic peaks in ABP signals using Savitzky-Golay smoothing.
576
577	Systolic peaks are identified in arterial blood pressure (ABP) signals by smoothing the signal and detecting peaks.
578
579	Returns
580	-------
581	peaks : numpy.ndarray
582	Indices of systolic peaks detected in the ABP signal.
583
584	Examples
585	--------
586	>>> signal = np.array([1, 2, 3, 4, 3, 2, 1])
587	>>> detector = PeakDetection(signal, method="abp_systolic")
588	>>> peaks = detector._abp_systolic_peak_detection()
589	>>> print(peaks)
590	"""
591	smoothed_signal = SignalFiltering.savgol_filter(	1✔
592	self.signal, window_length=self.window_length, polyorder=self.polyorder
593	)
594	peaks = find_peaks(	1✔
595	smoothed_signal,
596	height=self.height,
597	distance=self.distance,
598	threshold=self.threshold,
599	prominence=self.prominence,
600	width=self.width,
601	)
602	return self._refine_peaks(peaks)	1✔
603
604	def _abp_diastolic_peak_detection(self):	1✔
605	"""
606	Detect diastolic peaks in ABP signals by inverting and smoothing the signal.
607
608	Diastolic peaks are identified by inverting the arterial blood pressure (ABP) signal, smoothing it, and detecting peaks.
609
610	Returns
611	-------
612	peaks : numpy.ndarray
613	Indices of diastolic peaks detected in the ABP signal.
614
615	Examples
616	--------
617	>>> signal = np.array([3, 2, 1, 2, 3, 4, 3])
618	>>> detector = PeakDetection(signal, method="abp_diastolic")
619	>>> peaks = detector._abp_diastolic_peak_detection()
620	>>> print(peaks)
621	"""
622	inverted_signal = -self.signal	1✔
623	smoothed_signal = SignalFiltering.savgol_filter(	1✔
624	inverted_signal, window_length=self.window_length, polyorder=self.polyorder
625	)
626	peaks = find_peaks(	1✔
627	smoothed_signal,
628	distance=self.distance,
629	height=self.height,
630	threshold=self.threshold,
631	prominence=self.prominence,
632	width=self.width,
633	)
634	return self._refine_peaks(peaks)	1✔

Oucru-Innovations / vital-DSP / 11545366242

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous