11740053247

Committed 08 Nov 2024 10:04AM UTC coverage: 88.136% (-0.4%) from 88.524%

Build # 11740053247

Build Type

push

github

Committed by

Koaha

Commit Message

enhanced morphology features with peak detection

Run Details

1346 of 1792 branches covered (75.11%)

Branch coverage included in aggregate %.

156 of 252 new or added lines in 4 files covered. (61.9%)

73 existing lines in 2 files now uncovered.

5704 of 6207 relevant lines covered (91.9%)

0.92 hits per line

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

74.42

/src/vitalDSP/feature_engineering/ecg_autonomic_features.py

import numpy as np
from vitalDSP.utils.peak_detection import PeakDetection
from vitalDSP.physiological_features.waveform import WaveformMorphology


class ECGExtractor:
    """
    A class to extract ECG features including:
    - P-wave analysis (duration, amplitude)
    - PR Interval (P-wave to QRS onset)
    - QRS Complex (width, amplitude)
    - ST Segment (elevation, depression)
    - QT Interval (QRS onset to T-wave end)
    - Detection of Arrhythmias (AFib, VTach, Bradycardia)

    Example usage:
    ```
    ecg_signal = np.random.rand(1000)  # Replace with actual ECG signal
    fs = 250  # Sampling frequency in Hz

    extractor = ECGExtractor(ecg_signal, fs)

    p_wave_duration = extractor.compute_p_wave_duration()
    pr_interval = extractor.compute_pr_interval()
    qrs_width = extractor.compute_qrs_width()
    qt_interval = extractor.compute_qt_interval()
    st_segment = extractor.compute_st_segment()
    arrhythmias = extractor.detect_arrhythmias()

    print(f"P-wave Duration: {p_wave_duration}, PR Interval: {pr_interval}, QRS Width: {qrs_width}")
    ```
    """

    def __init__(self, ecg_signal, sampling_frequency):
        if not isinstance(ecg_signal, np.ndarray):
            raise TypeError("Input signal must be a numpy array")
        if len(ecg_signal) < 2:
            raise ValueError("ECG signal is too short to compute features")
        if np.isnan(ecg_signal).any() or np.isinf(ecg_signal).any():
            raise ValueError("ECG signal contains invalid values")

        self.ecg_signal = ecg_signal
        self.fs = sampling_frequency

        # Initialize the WaveformMorphology for Q, R, S, T detection
        self.morphology = WaveformMorphology(
            ecg_signal, fs=sampling_frequency, signal_type="ECG"
        )

    def detect_r_peaks(self):
        """
        Detects R-peaks from the ECG signal using WaveformMorphology.

        Returns:
            np.array: Array of indices where R-peaks are detected.
        """
        detector = PeakDetection(self.ecg_signal, method="ecg_r_peak")
        r_peaks = detector.detect_peaks()
        if len(r_peaks) == 0:
            raise ValueError("No R-peaks detected in ECG signal")
        return r_peaks

    def compute_p_wave_duration(self, r_peaks=None):
        """
        Computes the P-wave duration based on R-peak detection.

        Returns:
            float: Duration of P-wave in seconds.
        """
        if r_peaks is None:
            r_peaks = self.detect_r_peaks()  # Detect R-peaks first
        p_waves = self.morphology.detect_q_session(r_peaks)
        if p_waves.size == 0:
            return 0.0
        durations = [(end - start) / self.fs for start, end in p_waves]
        return np.mean(durations)

    def compute_pr_interval(self, r_peaks=None):
        """
        Computes the PR interval (P-wave to QRS onset).

        Returns:
            float: PR interval in seconds.
        """
        if r_peaks is None:
            r_peaks = self.detect_r_peaks()  # Detect R-peaks first
        pr_intervals = self.morphology.detect_q_session(r_peaks)
        if pr_intervals.size == 0:
            return 0.0
        durations = [(end - start) / self.fs for start, end in pr_intervals]
        return np.mean(durations)

    def compute_qrs_duration(self, r_peaks=None):
        """
        Computes the QRS duration using WaveformMorphology.

        Returns:
            float: The mean duration of QRS complexes in seconds.
        """
        if r_peaks is None:
            r_peaks = self.detect_r_peaks()  # Detect R-peaks first
        qrs_durations = self.morphology.detect_qrs_session(r_peaks)
        if not qrs_durations:
            return 0.0
        durations = [(end - start) / self.fs for start, end in qrs_durations]
        return np.mean(durations)

    def compute_s_wave(self, r_peaks=None):
        """
        Detects the S-wave based on the R-peaks using WaveformMorphology.

        Returns:
            np.array: Indices of detected S-wave points.
        """
        if r_peaks is None:
            r_peaks = self.detect_r_peaks()
        return self.morphology.detect_s_session(r_peaks)

    def compute_qt_interval(self):
        """
        Computes the QT interval (from QRS onset to T-wave end).

        Returns:
            float: QT interval in seconds.
        """
        q_valleys = self.morphology.detect_q_valley()
        t_peaks = self.morphology.detect_t_peak()
        if q_valleys[0] > t_peaks[0]:
            t_peaks = t_peaks[1:]  # skip the first peak
        sessions = []
        for i in range(min(len(q_valleys), len(t_peaks))):
            q_start = q_valleys[i]
            t_end = t_peaks[i]
            # Ensure there is a valid interval for the R session
            if q_start < t_end:
                sessions.append((q_start, t_end))
        sessions = np.array(sessions)

        return self.morphology.compute_duration(sessions=sessions, mode="Custom")

    def compute_st_interval(self):
        """
        Computes the ST segment elevation or depression.

        Returns:
            float: Mean ST segment deviation from baseline.
        """
        s_valleys = self.morphology.detect_s_valley()
        t_peaks = self.morphology.detect_t_peak()
        if s_valleys[0] > t_peaks[0]:
            t_peaks = t_peaks[1:]  # skip the first peak
        sessions = []
        for i in range(min(len(s_valleys), len(t_peaks))):
            s_start = s_valleys[i]
            t_end = t_peaks[i]
            # Ensure there is a valid interval for the R session
            if s_start < t_end:
                sessions.append((s_start, t_end))
        sessions = np.array(sessions)

        return self.morphology.compute_duration(sessions=sessions, mode="Custom")

    def detect_arrhythmias(self, r_peaks=None):
        """
        Detects basic arrhythmias such as:
        - Atrial Fibrillation (AFib)
        - Ventricular Tachycardia (VTach)
        - Bradycardia (slow heart rate)

        Returns:
            dict: Dictionary containing the detected arrhythmias.
        """
        if r_peaks is None:
            r_peaks = self.detect_r_peaks()
        rr_intervals = np.diff(r_peaks) / self.fs
        mean_rr = np.mean(rr_intervals)

        arrhythmias = {"AFib": False, "VTach": False, "Bradycardia": False}

        # Detect Atrial Fibrillation (AFib): Irregular RR intervals
        if np.std(rr_intervals) > 0.2 * mean_rr:
            arrhythmias["AFib"] = True

        # Detect Ventricular Tachycardia (VTach): Sustained high heart rate (> 100 bpm)
        if np.mean(rr_intervals) < 0.6:  # Corresponds to HR > 100 bpm
            arrhythmias["VTach"] = True

        # Detect Bradycardia: Slow heart rate (< 60 bpm)
        if np.mean(rr_intervals) > 1.0:  # Corresponds to HR < 60 bpm
            arrhythmias["Bradycardia"] = True

        return arrhythmias

1	import numpy as np	1✔
2	from vitalDSP.utils.peak_detection import PeakDetection	1✔
3	from vitalDSP.physiological_features.waveform import WaveformMorphology	1✔
4
5
6	class ECGExtractor:	1✔
7	"""
8	A class to extract ECG features including:
9	- P-wave analysis (duration, amplitude)
10	- PR Interval (P-wave to QRS onset)
11	- QRS Complex (width, amplitude)
12	- ST Segment (elevation, depression)
13	- QT Interval (QRS onset to T-wave end)
14	- Detection of Arrhythmias (AFib, VTach, Bradycardia)
15
16	Example usage:
17	```
18	ecg_signal = np.random.rand(1000) # Replace with actual ECG signal
19	fs = 250 # Sampling frequency in Hz
20
21	extractor = ECGExtractor(ecg_signal, fs)
22
23	p_wave_duration = extractor.compute_p_wave_duration()
24	pr_interval = extractor.compute_pr_interval()
25	qrs_width = extractor.compute_qrs_width()
26	qt_interval = extractor.compute_qt_interval()
27	st_segment = extractor.compute_st_segment()
28	arrhythmias = extractor.detect_arrhythmias()
29
30	print(f"P-wave Duration: {p_wave_duration}, PR Interval: {pr_interval}, QRS Width: {qrs_width}")
31	```
32	"""
33
34	def __init__(self, ecg_signal, sampling_frequency):	1✔
35	if not isinstance(ecg_signal, np.ndarray):	1✔
36	raise TypeError("Input signal must be a numpy array")	1✔
37	if len(ecg_signal) < 2:	1✔
38	raise ValueError("ECG signal is too short to compute features")	1✔
39	if np.isnan(ecg_signal).any() or np.isinf(ecg_signal).any():	1✔
40	raise ValueError("ECG signal contains invalid values")	1✔
41
42	self.ecg_signal = ecg_signal	1✔
43	self.fs = sampling_frequency	1✔
44
45	# Initialize the WaveformMorphology for Q, R, S, T detection
46	self.morphology = WaveformMorphology(	1✔
47	ecg_signal, fs=sampling_frequency, signal_type="ECG"
48	)
49
50	def detect_r_peaks(self):	1✔
51	"""
52	Detects R-peaks from the ECG signal using WaveformMorphology.
53
54	Returns:
55	np.array: Array of indices where R-peaks are detected.
56	"""
57	detector = PeakDetection(self.ecg_signal, method="ecg_r_peak")	1✔
58	r_peaks = detector.detect_peaks()	1✔
59	if len(r_peaks) == 0:	1✔
60	raise ValueError("No R-peaks detected in ECG signal")	1✔
61	return r_peaks	1✔
62
63	def compute_p_wave_duration(self, r_peaks=None):	1✔
64	"""
65	Computes the P-wave duration based on R-peak detection.
66
67	Returns:
68	float: Duration of P-wave in seconds.
69	"""
70	if r_peaks is None:	1!
71	r_peaks = self.detect_r_peaks() # Detect R-peaks first	1✔
72	p_waves = self.morphology.detect_q_session(r_peaks)	1✔
73	if p_waves.size == 0:	1!
74	return 0.0	×
75	durations = [(end - start) / self.fs for start, end in p_waves]	1✔
76	return np.mean(durations)	1✔
77
78	def compute_pr_interval(self, r_peaks=None):	1✔
79	"""
80	Computes the PR interval (P-wave to QRS onset).
81
82	Returns:
83	float: PR interval in seconds.
84	"""
85	if r_peaks is None:	1!
86	r_peaks = self.detect_r_peaks() # Detect R-peaks first	1✔
87	pr_intervals = self.morphology.detect_q_session(r_peaks)	1✔
88	if pr_intervals.size == 0:	1!
89	return 0.0	×
90	durations = [(end - start) / self.fs for start, end in pr_intervals]	1✔
91	return np.mean(durations)	1✔
92
93	def compute_qrs_duration(self, r_peaks=None):	1✔
94	"""
95	Computes the QRS duration using WaveformMorphology.
96
97	Returns:
98	float: The mean duration of QRS complexes in seconds.
99	"""
100	if r_peaks is None:	×
101	r_peaks = self.detect_r_peaks() # Detect R-peaks first	×
102	qrs_durations = self.morphology.detect_qrs_session(r_peaks)	×
103	if not qrs_durations:	×
104	return 0.0	×
105	durations = [(end - start) / self.fs for start, end in qrs_durations]	×
106	return np.mean(durations)	×
107
108	def compute_s_wave(self, r_peaks=None):	1✔
109	"""
110	Detects the S-wave based on the R-peaks using WaveformMorphology.
111
112	Returns:
113	np.array: Indices of detected S-wave points.
114	"""
115	if r_peaks is None:	×
116	r_peaks = self.detect_r_peaks()	×
117	return self.morphology.detect_s_session(r_peaks)	×
118
119	def compute_qt_interval(self):	1✔
120	"""
121	Computes the QT interval (from QRS onset to T-wave end).
122
123	Returns:
124	float: QT interval in seconds.
125	"""
126	q_valleys = self.morphology.detect_q_valley()	1✔
127	t_peaks = self.morphology.detect_t_peak()	1✔
128	if q_valleys[0] > t_peaks[0]:	1!
NEW 129	t_peaks = t_peaks[1:] # skip the first peak	×
130	sessions = []	1✔
131	for i in range(min(len(q_valleys), len(t_peaks))):	1✔
132	q_start = q_valleys[i]	1✔
133	t_end = t_peaks[i]	1✔
134	# Ensure there is a valid interval for the R session
135	if q_start < t_end:	1!
136	sessions.append((q_start, t_end))	1✔
137	sessions = np.array(sessions)	1✔
138
139	return self.morphology.compute_duration(sessions=sessions, mode="Custom")	1✔
140
141	def compute_st_interval(self):	1✔
142	"""
143	Computes the ST segment elevation or depression.
144
145	Returns:
146	float: Mean ST segment deviation from baseline.
147	"""
148	s_valleys = self.morphology.detect_s_valley()	1✔
149	t_peaks = self.morphology.detect_t_peak()	1✔
150	if s_valleys[0] > t_peaks[0]:	1!
NEW 151	t_peaks = t_peaks[1:] # skip the first peak	×
152	sessions = []	1✔
153	for i in range(min(len(s_valleys), len(t_peaks))):	1✔
154	s_start = s_valleys[i]	1✔
155	t_end = t_peaks[i]	1✔
156	# Ensure there is a valid interval for the R session
157	if s_start < t_end:	1✔
158	sessions.append((s_start, t_end))	1✔
159	sessions = np.array(sessions)	1✔
160
161	return self.morphology.compute_duration(sessions=sessions, mode="Custom")	1✔
162
163	def detect_arrhythmias(self, r_peaks=None):	1✔
164	"""
165	Detects basic arrhythmias such as:
166	- Atrial Fibrillation (AFib)
167	- Ventricular Tachycardia (VTach)
168	- Bradycardia (slow heart rate)
169
170	Returns:
171	dict: Dictionary containing the detected arrhythmias.
172	"""
173	if r_peaks is None:	1!
174	r_peaks = self.detect_r_peaks()	1✔
175	rr_intervals = np.diff(r_peaks) / self.fs	1✔
176	mean_rr = np.mean(rr_intervals)	1✔
177
178	arrhythmias = {"AFib": False, "VTach": False, "Bradycardia": False}	1✔
179
180	# Detect Atrial Fibrillation (AFib): Irregular RR intervals
181	if np.std(rr_intervals) > 0.2 * mean_rr:	1!
182	arrhythmias["AFib"] = True	×
183
184	# Detect Ventricular Tachycardia (VTach): Sustained high heart rate (> 100 bpm)
185	if np.mean(rr_intervals) < 0.6: # Corresponds to HR > 100 bpm	1!
186	arrhythmias["VTach"] = True	1✔
187
188	# Detect Bradycardia: Slow heart rate (< 60 bpm)
189	if np.mean(rr_intervals) > 1.0: # Corresponds to HR < 60 bpm	1!
190	arrhythmias["Bradycardia"] = True	×
191
192	return arrhythmias	1✔

Oucru-Innovations / vital-DSP / 11740053247

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