11532286550

Committed 26 Oct 2024 01:41PM UTC coverage: 91.103% (-1.6%) from 92.744%

Build # 11532286550

Build Type

push

github

Committed by

Koaha

Commit Message

enhance critical points detection

Run Details

1256 of 1578 branches covered (79.59%)

Branch coverage included in aggregate %.

20 of 96 new or added lines in 1 file covered. (20.83%)

10 existing lines in 2 files now uncovered.

5584 of 5930 relevant lines covered (94.17%)

0.94 hits per line

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

70.83

/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 not p_waves:
            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 not pr_intervals:
            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.
        """
        qrs_duration = (
            self.morphology.compute_qrs_duration()
        )  # No arguments passed here
        return qrs_duration

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

        Returns:
            float: Mean ST segment deviation from baseline.
        """
        return self.morphology.compute_qrs_duration()

    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 not p_waves:	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 not pr_intervals:	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	"""
UNCOV 100	if r_peaks is None:	×
UNCOV 101	r_peaks = self.detect_r_peaks() # Detect R-peaks first	×
UNCOV 102	qrs_durations = self.morphology.detect_qrs_session(r_peaks)	×
UNCOV 103	if not qrs_durations:	×
104	return 0.0	×
UNCOV 105	durations = [(end - start) / self.fs for start, end in qrs_durations]	×
UNCOV 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	qrs_duration = (	1✔
127	self.morphology.compute_qrs_duration()
128	) # No arguments passed here
129	return qrs_duration	1✔
130
131	def compute_st_segment(self):	1✔
132	"""
133	Computes the ST segment elevation or depression.
134
135	Returns:
136	float: Mean ST segment deviation from baseline.
137	"""
138	return self.morphology.compute_qrs_duration()	1✔
139
140	def detect_arrhythmias(self, r_peaks=None):	1✔
141	"""
142	Detects basic arrhythmias such as:
143	- Atrial Fibrillation (AFib)
144	- Ventricular Tachycardia (VTach)
145	- Bradycardia (slow heart rate)
146
147	Returns:
148	dict: Dictionary containing the detected arrhythmias.
149	"""
150	if r_peaks is None:	1!
151	r_peaks = self.detect_r_peaks()	1✔
152	rr_intervals = np.diff(r_peaks) / self.fs	1✔
153	mean_rr = np.mean(rr_intervals)	1✔
154
155	arrhythmias = {"AFib": False, "VTach": False, "Bradycardia": False}	1✔
156
157	# Detect Atrial Fibrillation (AFib): Irregular RR intervals
158	if np.std(rr_intervals) > 0.2 * mean_rr:	1!
159	arrhythmias["AFib"] = True	×
160
161	# Detect Ventricular Tachycardia (VTach): Sustained high heart rate (> 100 bpm)
162	if np.mean(rr_intervals) < 0.6: # Corresponds to HR > 100 bpm	1!
163	arrhythmias["VTach"] = True	1✔
164
165	# Detect Bradycardia: Slow heart rate (< 60 bpm)
166	if np.mean(rr_intervals) > 1.0: # Corresponds to HR < 60 bpm	1!
167	arrhythmias["Bradycardia"] = True	×
168
169	return arrhythmias	1✔

Oucru-Innovations / vital-DSP / 11532286550

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