12042684047

Committed 27 Nov 2024 02:35AM UTC coverage: 91.55% (-0.04%) from 91.587%

Build # 12042684047

Build Type

push

github

Committed by

Koaha

Commit Message

minor changes

Run Details

1556 of 1908 branches covered (81.55%)

Branch coverage included in aggregate %.

6 of 7 new or added lines in 2 files covered. (85.71%)

1 existing line in 1 file now uncovered.

6071 of 6423 relevant lines covered (94.52%)

0.95 hits per line

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

85.94

/src/vitalDSP/transforms/beats_transformation.py

import numpy as np

# from scipy.signal import find_peaks
# from vitalDSP.utils.common import find_peaks
from vitalDSP.physiological_features.waveform import WaveformMorphology
from vitalDSP.transforms.vital_transformation import VitalTransformation
from vitalDSP.utils.interpolations import (
    linear_interpolation,
    spline_interpolation,
    mean_imputation,
    median_imputation,
    forward_fill,
    backward_fill,
    rolling_mean_imputation,
)


class RRTransformation:
    """
    A class to perform RR interval computation and enhancement from ECG/PPG signals,
    incorporating peak detection, invalid interval removal, imputation, and advanced filtering techniques.

    This class leverages the `VitalTransformation` for initial signal processing and then applies RR interval computation
    and enhancement steps.

    Parameters
    ----------
    signal : np.array
        The input ECG or PPG signal.
    fs : int
        Sampling frequency of the signal.
    signal_type : str, optional
        The type of the signal ('ecg' or 'ppg'). Default is 'ecg'.
    options : dict, optional
        Options to customize the signal processing and RR interval computations.

    Methods
    -------
    compute_rr_intervals(options=None)
        Compute the RR intervals after applying transformations and peak detection.
    remove_invalid_rr_intervals(rr_intervals, min_rr=0.3, max_rr=2.0)
        Remove invalid RR intervals based on physiological constraints and trends.
    impute_rr_intervals(rr_intervals, method='adaptive', **kwargs)
        Impute invalid RR intervals using various methods.
    enhance_rr_intervals(rr_intervals, options=None)
        Apply enhancements to the RR intervals, such as peak enhancement.
    """

    def __init__(self, signal, fs, signal_type="ECG", options=None):
        """
        Initialize the RRTransformation class.

        Parameters
        ----------
        signal : np.array
            The input ECG or PPG signal.
        fs : int
            Sampling frequency of the signal.
        signal_type : str, optional
            The type of the signal ('ecg' or 'ppg'). Default is 'ecg'.
        options : dict, optional
            Options to customize the signal processing and RR interval computations.
        """
        self.signal = signal
        self.fs = fs
        self.signal_type = signal_type
        self.options = options

        # Apply the initial signal processing using VitalTransformation
        self.transformer = VitalTransformation(signal, fs=fs, signal_type=signal_type)
        self.transformed_signal = self.transformer.apply_transformations(self.options)

    def compute_rr_intervals(self, options=None):
        """
        Compute the RR intervals after applying transformations and peak detection.

        Parameters
        ----------
        options : dict, optional
            Options to customize peak detection. Default uses scipy's `find_peaks`.

        Returns
        -------
        rr_intervals : np.array
            The RR intervals computed from the detected peaks.
        """
        # if options is None:
        #     options = {
        #         "lowcut": 0.5,
        #         "highcut": 30,
        #         "filter_order": 4,
        #         "filter_type": "butter",
        #     }

        # # Detect peaks in the transformed signal
        # height = options.get("height", None)
        # distance = options.get(
        #     "distance", int(0.6 * self.fs)
        # )  # Minimum distance between peaks
        # peaks = find_peaks(self.transformed_signal, height=height, distance=distance)
        waveform = WaveformMorphology(
            waveform=self.signal, fs=self.fs, signal_type=self.signal_type
        )
        if waveform.signal_type == "ECG":
            peaks = waveform.r_peaks
        else:
            peaks = waveform.systolic_peaks
        if len(peaks) == 0:
            raise ValueError(
                "No peaks detected in the signal. RR interval computation failed."
            )

        # Compute RR intervals from the detected peaks
        rr_intervals = np.diff(peaks) / self.fs  # Time between peaks (in seconds)

        if len(rr_intervals) == 0:
            raise ValueError(
                "RR interval transformation failed to extract valid RR intervals."
            )

        return rr_intervals

    def remove_invalid_rr_intervals(
        self,
        rr_intervals,
        min_rr=0.3,
        max_rr=2.0,
        std_dev_factor=2.0,
        sudden_change_threshold=0.2,
    ):
        """
        Removes invalid RR intervals based on physiological limits, standard deviation filtering,
        and sudden change detection. Gradual trends are reconsidered as valid.

        Parameters
        ----------
        rr_intervals : np.array
            The array of RR intervals (in seconds).
        min_rr : float, optional
            The minimum allowable RR interval (default is 0.3 seconds).
        max_rr : float, optional
            The maximum allowable RR interval (default is 2.0 seconds).
        std_dev_factor : float, optional
            The factor used for standard deviation filtering (default is 2.0).
        sudden_change_threshold : float, optional
            The relative threshold for detecting sudden changes in RR intervals. Expressed as a percentage of the previous interval.

        Returns
        -------
        np.array
            The array of RR intervals with invalid intervals removed (or marked as NaN).

        Example
        -------
        >>> valid_rr_intervals = remove_invalid_rr_intervals(rr_intervals)
        """
        rr_intervals_filtered = np.copy(rr_intervals)

        # Step 1: Remove values that fall outside physiological limits
        rr_intervals_filtered = np.where(
            (rr_intervals_filtered >= min_rr) & (rr_intervals_filtered <= max_rr),
            rr_intervals_filtered,
            np.nan,
        )

        # Step 2: Compute mean and standard deviation of valid RR intervals
        valid_rr_intervals = rr_intervals_filtered[~np.isnan(rr_intervals_filtered)]
        mean_rr = np.mean(valid_rr_intervals)
        std_rr = np.std(valid_rr_intervals)

        # Step 3: Mark intervals outside the range of mean ± std_dev_factor * std_rr as invalid
        rr_intervals_filtered = np.where(
            (rr_intervals_filtered >= mean_rr - std_dev_factor * std_rr)
            & (rr_intervals_filtered <= mean_rr + std_dev_factor * std_rr),
            rr_intervals_filtered,
            np.nan,
        )

        # Step 4: Detect sudden changes
        for i in range(1, len(rr_intervals_filtered)):
            if not np.isnan(rr_intervals_filtered[i]) and not np.isnan(
                rr_intervals_filtered[i - 1]
            ):
                change = (
                    np.abs(rr_intervals_filtered[i] - rr_intervals_filtered[i - 1])
                    / rr_intervals_filtered[i - 1]
                )
                # If a sudden change is detected and it isn't part of a trend, mark it invalid
                if change > sudden_change_threshold:
                    rr_intervals_filtered[i] = np.nan

        # Step 5: Reconsider trends using a running average
        rr_intervals_filtered = self._reconsider_trends(rr_intervals_filtered)

        return rr_intervals_filtered

    def _reconsider_trends(self, rr_intervals, window_size=5):
        """
        Reconsiders RR intervals that show a gradual trend (increase or decrease) and avoids marking them as invalid.

        Parameters
        ----------
        rr_intervals : np.array
            The array of RR intervals with invalid intervals marked as NaN.
        window_size : int, optional
            The size of the window used to detect trends. Defaults to 5.

        Returns
        -------
        np.array
            The array of RR intervals with trends reconsidered.
        """
        rr_intervals_adaptive = np.copy(rr_intervals)
        running_mean = np.full(len(rr_intervals), np.nan)

        # Calculate a running mean of valid RR intervals
        for i in range(window_size, len(rr_intervals)):
            if not np.isnan(rr_intervals[i - window_size : i]).any():
                running_mean[i] = np.mean(rr_intervals[i - window_size : i])

        # If an RR interval is invalid, but close to the running mean, it could be a part of a trend
        for i in range(window_size, len(rr_intervals)):
            if np.isnan(rr_intervals[i]) and not np.isnan(running_mean[i]):
                if np.abs(rr_intervals[i] - running_mean[i]) < 0.1 * running_mean[i]:
                    # Reconsider this interval as valid
                    rr_intervals_adaptive[i] = running_mean[i]

        return rr_intervals_adaptive

    def impute_rr_intervals(self, intervals, method="adaptive", **kwargs):
        """
        Impute invalid RR intervals (NaN values) using various methods, with adaptive selection based on data patterns.

        Parameters
        ----------
        intervals : np.array
            The array of RR intervals with NaN values for invalid intervals.
        method : str, optional
            The imputation method to use. If 'adaptive', the function selects the best method based on data.
            Other options are 'linear', 'spline', 'mean', 'median', 'forward_fill', 'backward_fill', 'rolling_mean'.
            Defaults to 'adaptive'.
        **kwargs
            Additional keyword arguments for specific imputation methods, such as `order` for spline or `window` for rolling mean.

        Returns
        -------
        np.array
            The array of RR intervals with invalid intervals imputed.

        Examples
        --------
        >>> intervals = np.array([0.8, np.nan, 0.82, np.nan, np.nan, 0.85, 0.83])
        >>> imputed_intervals = impute_intervals(intervals, method='spline', order=3)
        >>> print(imputed_intervals)
        """
        # Handle adaptive imputation selection
        if method == "adaptive":
            nan_ratio = np.isnan(intervals).sum() / len(intervals)
            if nan_ratio < 0.05:
                return linear_interpolation(intervals)
            elif nan_ratio < 0.2:
                order = kwargs.get("order", 3)
                return spline_interpolation(intervals, order)
            else:
                window = kwargs.get("window", 5)
                return rolling_mean_imputation(intervals, window)

        # Manually specified method
        elif method == "linear":
            return linear_interpolation(intervals)
        elif method == "spline":
            order = kwargs.get("order", 3)
            return spline_interpolation(intervals, order)
        elif method == "mean":
            return mean_imputation(intervals)
        elif method == "median":
            return median_imputation(intervals)
        elif method == "forward_fill":
            return forward_fill(intervals)
        elif method == "backward_fill":
            return backward_fill(intervals)
        elif method == "rolling_mean":
            window = kwargs.get("window", 5)
            return rolling_mean_imputation(intervals, window)
        else:
            raise ValueError(f"Unsupported imputation method: {method}")

    def process_rr_intervals(self, remove_invalid=True, impute_invalid=True):
        """
        Full RR interval computation process:
        1. Compute RR intervals from the transformed signal.
        2. Remove invalid intervals.
        3. Impute missing values.
        4. Apply enhancement (if needed).

        Parameters
        ----------
        remove_invalid : bool, optional
            Whether to remove invalid RR intervals. Default is True.
        impute_invalid : bool, optional
            Whether to impute missing RR intervals. Default is True.
        Returns
        -------
        final_rr_intervals : np.array
            The fully processed RR intervals.
        """
        # Step 1: Compute RR intervals
        rr_intervals = self.compute_rr_intervals()

        if remove_invalid:
            # Step 2: Remove invalid RR intervals
            rr_intervals = self.remove_invalid_rr_intervals(rr_intervals)

        if impute_invalid:
            # Step 3: Impute missing RR intervals
            rr_intervals = self.impute_rr_intervals(rr_intervals)

        return rr_intervals

1	import numpy as np	1✔
2
3	# from scipy.signal import find_peaks
4	# from vitalDSP.utils.common import find_peaks
5	from vitalDSP.physiological_features.waveform import WaveformMorphology	1✔
6	from vitalDSP.transforms.vital_transformation import VitalTransformation	1✔
7	from vitalDSP.utils.interpolations import (	1✔
8	linear_interpolation,
9	spline_interpolation,
10	mean_imputation,
11	median_imputation,
12	forward_fill,
13	backward_fill,
14	rolling_mean_imputation,
15	)
16
17
18	class RRTransformation:	1✔
19	"""
20	A class to perform RR interval computation and enhancement from ECG/PPG signals,
21	incorporating peak detection, invalid interval removal, imputation, and advanced filtering techniques.
22
23	This class leverages the `VitalTransformation` for initial signal processing and then applies RR interval computation
24	and enhancement steps.
25
26	Parameters
27	----------
28	signal : np.array
29	The input ECG or PPG signal.
30	fs : int
31	Sampling frequency of the signal.
32	signal_type : str, optional
33	The type of the signal ('ecg' or 'ppg'). Default is 'ecg'.
34	options : dict, optional
35	Options to customize the signal processing and RR interval computations.
36
37	Methods
38	-------
39	compute_rr_intervals(options=None)
40	Compute the RR intervals after applying transformations and peak detection.
41	remove_invalid_rr_intervals(rr_intervals, min_rr=0.3, max_rr=2.0)
42	Remove invalid RR intervals based on physiological constraints and trends.
43	impute_rr_intervals(rr_intervals, method='adaptive', **kwargs)
44	Impute invalid RR intervals using various methods.
45	enhance_rr_intervals(rr_intervals, options=None)
46	Apply enhancements to the RR intervals, such as peak enhancement.
47	"""
48
49	def __init__(self, signal, fs, signal_type="ECG", options=None):	1✔
50	"""
51	Initialize the RRTransformation class.
52
53	Parameters
54	----------
55	signal : np.array
56	The input ECG or PPG signal.
57	fs : int
58	Sampling frequency of the signal.
59	signal_type : str, optional
60	The type of the signal ('ecg' or 'ppg'). Default is 'ecg'.
61	options : dict, optional
62	Options to customize the signal processing and RR interval computations.
63	"""
64	self.signal = signal	1✔
65	self.fs = fs	1✔
66	self.signal_type = signal_type	1✔
67	self.options = options	1✔
68
69	# Apply the initial signal processing using VitalTransformation
70	self.transformer = VitalTransformation(signal, fs=fs, signal_type=signal_type)	1✔
71	self.transformed_signal = self.transformer.apply_transformations(self.options)	1✔
72
73	def compute_rr_intervals(self, options=None):	1✔
74	"""
75	Compute the RR intervals after applying transformations and peak detection.
76
77	Parameters
78	----------
79	options : dict, optional
80	Options to customize peak detection. Default uses scipy's `find_peaks`.
81
82	Returns
83	-------
84	rr_intervals : np.array
85	The RR intervals computed from the detected peaks.
86	"""
87	# if options is None:
88	# options = {
89	# "lowcut": 0.5,
90	# "highcut": 30,
91	# "filter_order": 4,
92	# "filter_type": "butter",
93	# }
94
95	# # Detect peaks in the transformed signal
96	# height = options.get("height", None)
97	# distance = options.get(
98	# "distance", int(0.6 * self.fs)
99	# ) # Minimum distance between peaks
100	# peaks = find_peaks(self.transformed_signal, height=height, distance=distance)
101	waveform = WaveformMorphology(	1✔
102	waveform=self.signal, fs=self.fs, signal_type=self.signal_type
103	)
104	if waveform.signal_type == "ECG":	1!
105	peaks = waveform.r_peaks	1✔
106	else:
NEW 107	peaks = waveform.systolic_peaks	×
108	if len(peaks) == 0:	1✔
109	raise ValueError(	1✔
110	"No peaks detected in the signal. RR interval computation failed."
111	)
112
113	# Compute RR intervals from the detected peaks
114	rr_intervals = np.diff(peaks) / self.fs # Time between peaks (in seconds)	1✔
115
116	if len(rr_intervals) == 0:	1!
117	raise ValueError(	×
118	"RR interval transformation failed to extract valid RR intervals."
119	)
120
121	return rr_intervals	1✔
122
123	def remove_invalid_rr_intervals(	1✔
124	self,
125	rr_intervals,
126	min_rr=0.3,
127	max_rr=2.0,
128	std_dev_factor=2.0,
129	sudden_change_threshold=0.2,
130	):
131	"""
132	Removes invalid RR intervals based on physiological limits, standard deviation filtering,
133	and sudden change detection. Gradual trends are reconsidered as valid.
134
135	Parameters
136	----------
137	rr_intervals : np.array
138	The array of RR intervals (in seconds).
139	min_rr : float, optional
140	The minimum allowable RR interval (default is 0.3 seconds).
141	max_rr : float, optional
142	The maximum allowable RR interval (default is 2.0 seconds).
143	std_dev_factor : float, optional
144	The factor used for standard deviation filtering (default is 2.0).
145	sudden_change_threshold : float, optional
146	The relative threshold for detecting sudden changes in RR intervals. Expressed as a percentage of the previous interval.
147
148	Returns
149	-------
150	np.array
151	The array of RR intervals with invalid intervals removed (or marked as NaN).
152
153	Example
154	-------
155	>>> valid_rr_intervals = remove_invalid_rr_intervals(rr_intervals)
156	"""
157	rr_intervals_filtered = np.copy(rr_intervals)	1✔
158
159	# Step 1: Remove values that fall outside physiological limits
160	rr_intervals_filtered = np.where(	1✔
161	(rr_intervals_filtered >= min_rr) & (rr_intervals_filtered <= max_rr),
162	rr_intervals_filtered,
163	np.nan,
164	)
165
166	# Step 2: Compute mean and standard deviation of valid RR intervals
167	valid_rr_intervals = rr_intervals_filtered[~np.isnan(rr_intervals_filtered)]	1✔
168	mean_rr = np.mean(valid_rr_intervals)	1✔
169	std_rr = np.std(valid_rr_intervals)	1✔
170
171	# Step 3: Mark intervals outside the range of mean ± std_dev_factor * std_rr as invalid
172	rr_intervals_filtered = np.where(	1✔
173	(rr_intervals_filtered >= mean_rr - std_dev_factor * std_rr)
174	& (rr_intervals_filtered <= mean_rr + std_dev_factor * std_rr),
175	rr_intervals_filtered,
176	np.nan,
177	)
178
179	# Step 4: Detect sudden changes
180	for i in range(1, len(rr_intervals_filtered)):	1✔
181	if not np.isnan(rr_intervals_filtered[i]) and not np.isnan(	1✔
182	rr_intervals_filtered[i - 1]
183	):
184	change = (	1✔
185	np.abs(rr_intervals_filtered[i] - rr_intervals_filtered[i - 1])
186	/ rr_intervals_filtered[i - 1]
187	)
188	# If a sudden change is detected and it isn't part of a trend, mark it invalid
189	if change > sudden_change_threshold:	1✔
190	rr_intervals_filtered[i] = np.nan	1✔
191
192	# Step 5: Reconsider trends using a running average
193	rr_intervals_filtered = self._reconsider_trends(rr_intervals_filtered)	1✔
194
195	return rr_intervals_filtered	1✔
196
197	def _reconsider_trends(self, rr_intervals, window_size=5):	1✔
198	"""
199	Reconsiders RR intervals that show a gradual trend (increase or decrease) and avoids marking them as invalid.
200
201	Parameters
202	----------
203	rr_intervals : np.array
204	The array of RR intervals with invalid intervals marked as NaN.
205	window_size : int, optional
206	The size of the window used to detect trends. Defaults to 5.
207
208	Returns
209	-------
210	np.array
211	The array of RR intervals with trends reconsidered.
212	"""
213	rr_intervals_adaptive = np.copy(rr_intervals)	1✔
214	running_mean = np.full(len(rr_intervals), np.nan)	1✔
215
216	# Calculate a running mean of valid RR intervals
217	for i in range(window_size, len(rr_intervals)):	1✔
218	if not np.isnan(rr_intervals[i - window_size : i]).any():	1!
219	running_mean[i] = np.mean(rr_intervals[i - window_size : i])	×
220
221	# If an RR interval is invalid, but close to the running mean, it could be a part of a trend
222	for i in range(window_size, len(rr_intervals)):	1✔
223	if np.isnan(rr_intervals[i]) and not np.isnan(running_mean[i]):	1!
224	if np.abs(rr_intervals[i] - running_mean[i]) < 0.1 * running_mean[i]:	×
225	# Reconsider this interval as valid
226	rr_intervals_adaptive[i] = running_mean[i]	×
227
228	return rr_intervals_adaptive	1✔
229
230	def impute_rr_intervals(self, intervals, method="adaptive", **kwargs):	1✔
231	"""
232	Impute invalid RR intervals (NaN values) using various methods, with adaptive selection based on data patterns.
233
234	Parameters
235	----------
236	intervals : np.array
237	The array of RR intervals with NaN values for invalid intervals.
238	method : str, optional
239	The imputation method to use. If 'adaptive', the function selects the best method based on data.
240	Other options are 'linear', 'spline', 'mean', 'median', 'forward_fill', 'backward_fill', 'rolling_mean'.
241	Defaults to 'adaptive'.
242	**kwargs
243	Additional keyword arguments for specific imputation methods, such as `order` for spline or `window` for rolling mean.
244
245	Returns
246	-------
247	np.array
248	The array of RR intervals with invalid intervals imputed.
249
250	Examples
251	--------
252	>>> intervals = np.array([0.8, np.nan, 0.82, np.nan, np.nan, 0.85, 0.83])
253	>>> imputed_intervals = impute_intervals(intervals, method='spline', order=3)
254	>>> print(imputed_intervals)
255	"""
256	# Handle adaptive imputation selection
257	if method == "adaptive":	1✔
258	nan_ratio = np.isnan(intervals).sum() / len(intervals)	1✔
259	if nan_ratio < 0.05:	1!
UNCOV 260	return linear_interpolation(intervals)	×
261	elif nan_ratio < 0.2:	1!
262	order = kwargs.get("order", 3)	×
263	return spline_interpolation(intervals, order)	×
264	else:
265	window = kwargs.get("window", 5)	1✔
266	return rolling_mean_imputation(intervals, window)	1✔
267
268	# Manually specified method
269	elif method == "linear":	1✔
270	return linear_interpolation(intervals)	1✔
271	elif method == "spline":	1✔
272	order = kwargs.get("order", 3)	1✔
273	return spline_interpolation(intervals, order)	1✔
274	elif method == "mean":	1✔
275	return mean_imputation(intervals)	1✔
276	elif method == "median":	1✔
277	return median_imputation(intervals)	1✔
278	elif method == "forward_fill":	1✔
279	return forward_fill(intervals)	1✔
280	elif method == "backward_fill":	1✔
281	return backward_fill(intervals)	1✔
282	elif method == "rolling_mean":	1✔
283	window = kwargs.get("window", 5)	1✔
284	return rolling_mean_imputation(intervals, window)	1✔
285	else:
286	raise ValueError(f"Unsupported imputation method: {method}")	1✔
287
288	def process_rr_intervals(self, remove_invalid=True, impute_invalid=True):	1✔
289	"""
290	Full RR interval computation process:
291	1. Compute RR intervals from the transformed signal.
292	2. Remove invalid intervals.
293	3. Impute missing values.
294	4. Apply enhancement (if needed).
295
296	Parameters
297	----------
298	remove_invalid : bool, optional
299	Whether to remove invalid RR intervals. Default is True.
300	impute_invalid : bool, optional
301	Whether to impute missing RR intervals. Default is True.
302	Returns
303	-------
304	final_rr_intervals : np.array
305	The fully processed RR intervals.
306	"""
307	# Step 1: Compute RR intervals
308	rr_intervals = self.compute_rr_intervals()	1✔
309
310	if remove_invalid:	1!
311	# Step 2: Remove invalid RR intervals
312	rr_intervals = self.remove_invalid_rr_intervals(rr_intervals)	1✔
313
314	if impute_invalid:	1!
315	# Step 3: Impute missing RR intervals
316	rr_intervals = self.impute_rr_intervals(rr_intervals)	1✔
317
318	return rr_intervals	1✔

Oucru-Innovations / vital-DSP / 12042684047

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