11792991704

Committed 12 Nov 2024 07:58AM UTC coverage: 91.587% (+0.2%) from 91.402%

Build # 11792991704

Build Type

push

github

Committed by

Koaha

Commit Message

linting src

Run Details

1557 of 1908 branches covered (81.6%)

Branch coverage included in aggregate %.

3 of 7 new or added lines in 2 files covered. (42.86%)

35 existing lines in 3 files now uncovered.

6074 of 6424 relevant lines covered (94.55%)

0.95 hits per line

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

78.79

/src/vitalDSP/feature_engineering/morphology_features.py

import numpy as np
from scipy.stats import linregress

# from vitalDSP.utils.peak_detection import PeakDetection
from vitalDSP.preprocess.preprocess_operations import (
    PreprocessConfig,
    preprocess_signal,
)
from vitalDSP.physiological_features.waveform import WaveformMorphology
import warnings
import logging as logger


class PhysiologicalFeatureExtractor:
    """
    A class to extract various physiological features from ECG and PPG signals, such as durations,
    areas, amplitude variability, slope ratios, and dicrotic notch locations.

    Features for PPG:
    - Systolic/diastolic duration, area, amplitude variability
    - Signal skewness, slope, peak trends, and dicrotic notch locations

    Features for ECG:
    - QRS duration, area, T-wave area
    - Amplitude variability, QRS-T ratios, and QRS slope
    - Signal skewness, peak trends

    Methods
    -------
    preprocess_signal(preprocess_config)
        Preprocess the signal by applying bandpass filtering and noise reduction.
    extract_features(signal_type='ECG', preprocess_config=None)
        Extract all features (morphology, volume, amplitude variability, dicrotic notch) for ECG or PPG signals.
    """

    def __init__(self, signal, fs=1000):
        """
        Initialize the PhysiologicalFeatureExtractor class.

        Parameters
        ----------
        signal : numpy.ndarray
            The input physiological signal (ECG or PPG).
        fs : int, optional
            Sampling frequency in Hz. Default is 1000.
        """
        self.signal = np.asarray(signal, dtype=np.float64)
        self.fs = fs

    def detect_troughs(self, peaks):
        """
        Detect troughs (valleys) in the signal based on the given peaks.

        Parameters
        ----------
        peaks : numpy.ndarray
            The indices of detected peaks.

        Returns
        -------
        troughs : numpy.ndarray
            The indices of detected troughs.

        Examples
        --------
        >>> peaks = np.array([100, 200, 300])
        >>> troughs = extractor.detect_troughs(peaks)
        >>> print(troughs)
        """
        warnings.warn(
            "Deprecated. Please use vitalDSP.physiological_features.waveform.WaveformMorphology instead.",
            DeprecationWarning,
        )
        troughs = []
        for i in range(len(peaks) - 1):
            # Find the local minimum between two consecutive peaks
            segment = self.signal[peaks[i] : peaks[i + 1]]
            trough = np.argmin(segment) + peaks[i]
            troughs.append(trough)
        return np.array(troughs)

    def compute_peak_trend(self, peaks):
        """
        Compute the trend slope of peak amplitudes over time.

        Parameters
        ----------
        peaks : numpy.ndarray
            The set of peaks.

        Returns
        -------
        trend_slope : float
            The slope of the peak amplitude trend over time.

        Examples
        --------
        >>> signal = np.random.randn(1000)
        >>> peaks = np.array([100, 200, 300])
        >>> extractor = PhysiologicalFeatureExtractor(signal)
        >>> trend_slope = extractor.compute_peak_trend(peaks)
        >>> print(trend_slope)
        """
        amplitudes = self.signal[peaks]
        if len(peaks) > 1:
            slope, _, _, _, _ = linregress(peaks, amplitudes)
            return slope
        return 0.0

    def compute_amplitude_variability(self, peaks):
        """
        Compute the variability of the amplitudes at the given peak locations.

        Parameters
        ----------
        peaks : numpy.ndarray
            The set of peaks.

        Returns
        -------
        variability : float
            The amplitude variability (standard deviation of the peak amplitudes).

        Examples
        --------
        >>> signal = np.random.randn(1000)
        >>> peaks = np.array([100, 200, 300])
        >>> extractor = PhysiologicalFeatureExtractor(signal)
        >>> variability = extractor.compute_amplitude_variability(peaks)
        >>> print(variability)
        """
        amplitudes = self.signal[peaks]
        return np.std(amplitudes) if len(amplitudes) > 1 else 0.0

    def get_preprocess_signal(self, preprocess_config):
        """
        Preprocess the signal by applying bandpass filtering and noise reduction.

        Parameters
        ----------
        preprocess_config : PreprocessConfig
            Configuration for both signal filtering and artifact removal.

        Returns
        -------
        clean_signal : numpy.ndarray
            The preprocessed signal, cleaned of noise and artifacts.

        Examples
        --------
        >>> signal = np.sin(np.linspace(0, 10, 1000)) + np.random.normal(0, 0.1, 1000)
        >>> preprocess_config = PreprocessConfig()
        >>> extractor = PhysiologicalFeatureExtractor(signal, fs=1000)
        >>> preprocessed_signal = extractor.preprocess_signal(preprocess_config)
        >>> print(preprocessed_signal)
        """
        return preprocess_signal(
            signal=self.signal,
            sampling_rate=self.fs,
            filter_type=preprocess_config.filter_type,
            lowcut=preprocess_config.lowcut,
            highcut=preprocess_config.highcut,
            order=preprocess_config.order,
            noise_reduction_method=preprocess_config.noise_reduction_method,
            wavelet_name=preprocess_config.wavelet_name,
            level=preprocess_config.level,
            window_length=preprocess_config.window_length,
            polyorder=preprocess_config.polyorder,
            kernel_size=preprocess_config.kernel_size,
            sigma=preprocess_config.sigma,
            respiratory_mode=preprocess_config.respiratory_mode,
        )

    def extract_features(self, signal_type="ECG", preprocess_config=None):
        """
        Extract all physiological features from the signal for either ECG or PPG.

        Parameters
        ----------
        signal_type : str, optional
            The type of signal ("ECG" or "PPG"). Default is "ECG".
        preprocess_config : PreprocessConfig, optional
            The configuration object for signal preprocessing. If None, default settings are used.

        Returns
        -------
        features : dict
            A dictionary containing the extracted features, such as durations, areas, amplitude variability,
            slopes, skewness, peak trends, and dicrotic notch locations.

        Examples
        --------
        >>> signal = np.random.randn(1000)
        >>> preprocess_config = PreprocessConfig()
        >>> extractor = PhysiologicalFeatureExtractor(signal, fs=1000)
        >>> features = extractor.extract_features(signal_type="ECG", preprocess_config=preprocess_config)
        >>> print(features)
        """
        if preprocess_config is None:
            preprocess_config = PreprocessConfig()

        # Initialize features as an empty dictionary before the try block
        features = {}

        # Preprocess the signal
        try:
            clean_signal = self.get_preprocess_signal(preprocess_config)
        except Exception as e:
            logger.error(f"Error during preprocessing: {e}")
            return {key: np.nan for key in features}  # Set all features to NaN

        # Baseline correction
        clean_signal = clean_signal - np.min(clean_signal)

        # Initialize the morphology class
        try:
            morphology = WaveformMorphology(
                clean_signal, fs=self.fs, signal_type=signal_type
            )
        except Exception as e:
            logger.error(f"Error initializing morphology: {e}")
            return {key: np.nan for key in features}  # Set all features to NaN

        try:
            if signal_type == "PPG":
                features = {
                    "systolic_duration": np.nan,
                    "diastolic_duration": np.nan,
                    "systolic_area": np.nan,
                    "diastolic_area": np.nan,
                    "systolic_slope": np.nan,
                    "diastolic_slope": np.nan,
                    "signal_skewness": np.nan,
                    "peak_trend_slope": np.nan,
                    "heart_rate": np.nan,
                    "systolic_amplitude_variability": np.nan,
                    "diastolic_amplitude_variability": np.nan,
                }

                features = {
                    "systolic_duration": morphology.get_duration(
                        session_type="systolic"
                    ),
                    "diastolic_duration": morphology.get_duration(
                        session_type="diastolic"
                    ),
                    "systolic_area": morphology.get_area(
                        interval_type="Sys-to-Notch", signal_type="PPG"
                    ),
                    "diastolic_area": morphology.get_area(
                        interval_type="Notch-to-Dia", signal_type="PPG"
                    ),
                    "systolic_slope": morphology.get_slope(
                        slope_type="systolic", window=5
                    ),
                    "diastolic_slope": morphology.get_slope(
                        slope_type="systolic", window=5
                    ),
                    "signal_skewness": morphology.get_signal_skewness(),
                    "peak_trend_slope": morphology.get_peak_trend_slope(),
                    "heart_rate": morphology.get_heart_rate(),
                    "systolic_amplitude_variability": morphology.get_amplitude_variability(
                        interval_type="Sys-to-Baseline", signal_type="PPG"
                    ),
                    "diastolic_amplitude_variability": morphology.get_amplitude_variability(
                        interval_type="Dia-to-Baseline", signal_type="PPG"
                    ),
                }

            elif signal_type == "ECG":
                features = {
                    "qrs_duration": np.nan,
                    "qrs_area": np.nan,
                    "qrs_amplitude": np.nan,
                    "qrs_slope": np.nan,
                    "t_wave_area": np.nan,
                    "heart_rate": np.nan,
                    "r_peak_amplitude_variability": np.nan,
                    "signal_skewness": np.nan,
                    "peak_trend_slope": np.nan,
                }

                features = {
                    "qrs_duration": morphology.get_duration(session_type="qrs"),
                    "qrs_area": morphology.get_area(
                        interval_type="QRS", signal_type="ECG"
                    ),
                    "qrs_amplitude": morphology.get_qrs_amplitude(),
                    "qrs_slope": morphology.get_slope(slope_type="qrs", window=5),
                    "t_wave_area": morphology.get_area(
                        interval_type="T-to-S", signal_type="ECG"
                    ),
                    "heart_rate": morphology.get_heart_rate(),
                    "r_peak_amplitude_variability": morphology.get_amplitude_variability(
                        interval_type="R-to-Baseline", signal_type="ECG"
                    ),
                    "signal_skewness": morphology.get_signal_skewness(
                        signal_type="ECG"
                    ),
                    "peak_trend_slope": morphology.get_peak_trend_slope(),
                }

            else:
                raise ValueError(f"Unsupported signal type: {signal_type}")

        except Exception as e:
            # print(f"Error during feature extraction: {e}")
            # features = {
            #     key: np.nan for key in features
            # }  # Set all features to np.nan in case of error
            logger.error(f"Error during feature extraction: {e}")
            features = {
                key: np.nan for key in features
            }  # Set all features to NaN in case of error

        return features

1	import numpy as np	1✔
2	from scipy.stats import linregress	1✔
3
4	# from vitalDSP.utils.peak_detection import PeakDetection
5	from vitalDSP.preprocess.preprocess_operations import (	1✔
6	PreprocessConfig,
7	preprocess_signal,
8	)
9	from vitalDSP.physiological_features.waveform import WaveformMorphology	1✔
10	import warnings	1✔
11	import logging as logger	1✔
12
13
14	class PhysiologicalFeatureExtractor:	1✔
15	"""
16	A class to extract various physiological features from ECG and PPG signals, such as durations,
17	areas, amplitude variability, slope ratios, and dicrotic notch locations.
18
19	Features for PPG:
20	- Systolic/diastolic duration, area, amplitude variability
21	- Signal skewness, slope, peak trends, and dicrotic notch locations
22
23	Features for ECG:
24	- QRS duration, area, T-wave area
25	- Amplitude variability, QRS-T ratios, and QRS slope
26	- Signal skewness, peak trends
27
28	Methods
29	-------
30	preprocess_signal(preprocess_config)
31	Preprocess the signal by applying bandpass filtering and noise reduction.
32	extract_features(signal_type='ECG', preprocess_config=None)
33	Extract all features (morphology, volume, amplitude variability, dicrotic notch) for ECG or PPG signals.
34	"""
35
36	def __init__(self, signal, fs=1000):	1✔
37	"""
38	Initialize the PhysiologicalFeatureExtractor class.
39
40	Parameters
41	----------
42	signal : numpy.ndarray
43	The input physiological signal (ECG or PPG).
44	fs : int, optional
45	Sampling frequency in Hz. Default is 1000.
46	"""
47	self.signal = np.asarray(signal, dtype=np.float64)	1✔
48	self.fs = fs	1✔
49
50	def detect_troughs(self, peaks):	1✔
51	"""
52	Detect troughs (valleys) in the signal based on the given peaks.
53
54	Parameters
55	----------
56	peaks : numpy.ndarray
57	The indices of detected peaks.
58
59	Returns
60	-------
61	troughs : numpy.ndarray
62	The indices of detected troughs.
63
64	Examples
65	--------
66	>>> peaks = np.array([100, 200, 300])
67	>>> troughs = extractor.detect_troughs(peaks)
68	>>> print(troughs)
69	"""
70	warnings.warn(	×
71	"Deprecated. Please use vitalDSP.physiological_features.waveform.WaveformMorphology instead.",
72	DeprecationWarning,
73	)
74	troughs = []	×
75	for i in range(len(peaks) - 1):	×
76	# Find the local minimum between two consecutive peaks
77	segment = self.signal[peaks[i] : peaks[i + 1]]	×
78	trough = np.argmin(segment) + peaks[i]	×
79	troughs.append(trough)	×
80	return np.array(troughs)	×
81
82	def compute_peak_trend(self, peaks):	1✔
83	"""
84	Compute the trend slope of peak amplitudes over time.
85
86	Parameters
87	----------
88	peaks : numpy.ndarray
89	The set of peaks.
90
91	Returns
92	-------
93	trend_slope : float
94	The slope of the peak amplitude trend over time.
95
96	Examples
97	--------
98	>>> signal = np.random.randn(1000)
99	>>> peaks = np.array([100, 200, 300])
100	>>> extractor = PhysiologicalFeatureExtractor(signal)
101	>>> trend_slope = extractor.compute_peak_trend(peaks)
102	>>> print(trend_slope)
103	"""
104	amplitudes = self.signal[peaks]	1✔
105	if len(peaks) > 1:	1✔
106	slope, _, _, _, _ = linregress(peaks, amplitudes)	1✔
107	return slope	1✔
108	return 0.0	1✔
109
110	def compute_amplitude_variability(self, peaks):	1✔
111	"""
112	Compute the variability of the amplitudes at the given peak locations.
113
114	Parameters
115	----------
116	peaks : numpy.ndarray
117	The set of peaks.
118
119	Returns
120	-------
121	variability : float
122	The amplitude variability (standard deviation of the peak amplitudes).
123
124	Examples
125	--------
126	>>> signal = np.random.randn(1000)
127	>>> peaks = np.array([100, 200, 300])
128	>>> extractor = PhysiologicalFeatureExtractor(signal)
129	>>> variability = extractor.compute_amplitude_variability(peaks)
130	>>> print(variability)
131	"""
132	amplitudes = self.signal[peaks]	1✔
133	return np.std(amplitudes) if len(amplitudes) > 1 else 0.0	1✔
134
135	def get_preprocess_signal(self, preprocess_config):	1✔
136	"""
137	Preprocess the signal by applying bandpass filtering and noise reduction.
138
139	Parameters
140	----------
141	preprocess_config : PreprocessConfig
142	Configuration for both signal filtering and artifact removal.
143
144	Returns
145	-------
146	clean_signal : numpy.ndarray
147	The preprocessed signal, cleaned of noise and artifacts.
148
149	Examples
150	--------
151	>>> signal = np.sin(np.linspace(0, 10, 1000)) + np.random.normal(0, 0.1, 1000)
152	>>> preprocess_config = PreprocessConfig()
153	>>> extractor = PhysiologicalFeatureExtractor(signal, fs=1000)
154	>>> preprocessed_signal = extractor.preprocess_signal(preprocess_config)
155	>>> print(preprocessed_signal)
156	"""
157	return preprocess_signal(	1✔
158	signal=self.signal,
159	sampling_rate=self.fs,
160	filter_type=preprocess_config.filter_type,
161	lowcut=preprocess_config.lowcut,
162	highcut=preprocess_config.highcut,
163	order=preprocess_config.order,
164	noise_reduction_method=preprocess_config.noise_reduction_method,
165	wavelet_name=preprocess_config.wavelet_name,
166	level=preprocess_config.level,
167	window_length=preprocess_config.window_length,
168	polyorder=preprocess_config.polyorder,
169	kernel_size=preprocess_config.kernel_size,
170	sigma=preprocess_config.sigma,
171	respiratory_mode=preprocess_config.respiratory_mode,
172	)
173
174	def extract_features(self, signal_type="ECG", preprocess_config=None):	1✔
175	"""
176	Extract all physiological features from the signal for either ECG or PPG.
177
178	Parameters
179	----------
180	signal_type : str, optional
181	The type of signal ("ECG" or "PPG"). Default is "ECG".
182	preprocess_config : PreprocessConfig, optional
183	The configuration object for signal preprocessing. If None, default settings are used.
184
185	Returns
186	-------
187	features : dict
188	A dictionary containing the extracted features, such as durations, areas, amplitude variability,
189	slopes, skewness, peak trends, and dicrotic notch locations.
190
191	Examples
192	--------
193	>>> signal = np.random.randn(1000)
194	>>> preprocess_config = PreprocessConfig()
195	>>> extractor = PhysiologicalFeatureExtractor(signal, fs=1000)
196	>>> features = extractor.extract_features(signal_type="ECG", preprocess_config=preprocess_config)
197	>>> print(features)
198	"""
199	if preprocess_config is None:	1✔
200	preprocess_config = PreprocessConfig()	1✔
201
202	# Initialize features as an empty dictionary before the try block
203	features = {}	1✔
204
205	# Preprocess the signal
206	try:	1✔
207	clean_signal = self.get_preprocess_signal(preprocess_config)	1✔
208	except Exception as e:	1✔
209	logger.error(f"Error during preprocessing: {e}")	1✔
210	return {key: np.nan for key in features} # Set all features to NaN	1✔
211
212	# Baseline correction
213	clean_signal = clean_signal - np.min(clean_signal)	1✔
214
215	# Initialize the morphology class
216	try:	1✔
217	morphology = WaveformMorphology(	1✔
218	clean_signal, fs=self.fs, signal_type=signal_type
219	)
220	except Exception as e:	1✔
221	logger.error(f"Error initializing morphology: {e}")	1✔
222	return {key: np.nan for key in features} # Set all features to NaN	1✔
223
224	try:	1✔
225	if signal_type == "PPG":	1✔
226	features = {	1✔
227	"systolic_duration": np.nan,
228	"diastolic_duration": np.nan,
229	"systolic_area": np.nan,
230	"diastolic_area": np.nan,
231	"systolic_slope": np.nan,
232	"diastolic_slope": np.nan,
233	"signal_skewness": np.nan,
234	"peak_trend_slope": np.nan,
235	"heart_rate": np.nan,
236	"systolic_amplitude_variability": np.nan,
237	"diastolic_amplitude_variability": np.nan,
238	}
239
240	features = {	1✔
241	"systolic_duration": morphology.get_duration(
242	session_type="systolic"
243	),
244	"diastolic_duration": morphology.get_duration(
245	session_type="diastolic"
246	),
247	"systolic_area": morphology.get_area(
248	interval_type="Sys-to-Notch", signal_type="PPG"
249	),
250	"diastolic_area": morphology.get_area(
251	interval_type="Notch-to-Dia", signal_type="PPG"
252	),
253	"systolic_slope": morphology.get_slope(
254	slope_type="systolic", window=5
255	),
256	"diastolic_slope": morphology.get_slope(
257	slope_type="systolic", window=5
258	),
259	"signal_skewness": morphology.get_signal_skewness(),
260	"peak_trend_slope": morphology.get_peak_trend_slope(),
261	"heart_rate": morphology.get_heart_rate(),
262	"systolic_amplitude_variability": morphology.get_amplitude_variability(
263	interval_type="Sys-to-Baseline", signal_type="PPG"
264	),
265	"diastolic_amplitude_variability": morphology.get_amplitude_variability(
266	interval_type="Dia-to-Baseline", signal_type="PPG"
267	),
268	}
269
270	elif signal_type == "ECG":	1!
271	features = {	1✔
272	"qrs_duration": np.nan,
273	"qrs_area": np.nan,
274	"qrs_amplitude": np.nan,
275	"qrs_slope": np.nan,
276	"t_wave_area": np.nan,
277	"heart_rate": np.nan,
278	"r_peak_amplitude_variability": np.nan,
279	"signal_skewness": np.nan,
280	"peak_trend_slope": np.nan,
281	}
282
283	features = {	1✔
284	"qrs_duration": morphology.get_duration(session_type="qrs"),
285	"qrs_area": morphology.get_area(
286	interval_type="QRS", signal_type="ECG"
287	),
288	"qrs_amplitude": morphology.get_qrs_amplitude(),
289	"qrs_slope": morphology.get_slope(slope_type="qrs", window=5),
290	"t_wave_area": morphology.get_area(
291	interval_type="T-to-S", signal_type="ECG"
292	),
293	"heart_rate": morphology.get_heart_rate(),
294	"r_peak_amplitude_variability": morphology.get_amplitude_variability(
295	interval_type="R-to-Baseline", signal_type="ECG"
296	),
297	"signal_skewness": morphology.get_signal_skewness(
298	signal_type="ECG"
299	),
300	"peak_trend_slope": morphology.get_peak_trend_slope(),
301	}
302
303	else:
304	raise ValueError(f"Unsupported signal type: {signal_type}")	×
305
UNCOV 306	except Exception as e:	×
307	# print(f"Error during feature extraction: {e}")
308	# features = {
309	# key: np.nan for key in features
310	# } # Set all features to np.nan in case of error
UNCOV 311	logger.error(f"Error during feature extraction: {e}")	×
UNCOV 312	features = {	×
313	key: np.nan for key in features
314	} # Set all features to NaN in case of error
315
316	return features	1✔

Oucru-Innovations / vital-DSP / 11792991704

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