10909444855

Committed 17 Sep 2024 07:07PM UTC coverage: 91.331% (+2.2%) from 89.161%

Build # 10909444855

Build Type

push

github

Committed by

Koaha

Commit Message

release rc16

Run Details

1120 of 1363 branches covered (82.17%)

Branch coverage included in aggregate %.

14 of 15 new or added lines in 4 files covered. (93.33%)

1 existing line in 1 file now uncovered.

4527 of 4820 relevant lines covered (93.92%)

0.94 hits per line

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

65.63

/src/vitalDSP/physiological_features/hrv_analysis.py

import numpy as np
from vitalDSP.physiological_features.time_domain import TimeDomainFeatures
from vitalDSP.physiological_features.frequency_domain import FrequencyDomainFeatures
from vitalDSP.physiological_features.nonlinear import NonlinearFeatures
from vitalDSP.transforms.beats_transformation import RRTransformation


class HRVFeatures:
    """
    A class to compute HRV features from physiological signals such as ECG and PPG.

    It combines time-domain, frequency-domain, and nonlinear features into one comprehensive feature extraction module.

    Attributes
    ----------
    nn_intervals : np.array
        The NN intervals (in milliseconds) between heartbeats.
    signal : np.array
        The physiological signal (ECG, PPG, etc.).
    fs : int
        The sampling frequency in Hz. Default is 1000 Hz.

    Methods
    -------
    compute_all_features()
        Computes all HRV features and returns them in a dictionary format.
    """

    def __init__(
        self, signals, nn_intervals=None, fs=100, signal_type="ppg", options=None
    ):
        """
        Initializes the HRVFeatures object.

        Parameters
        ----------
        nn_intervals : np.array
            The NN intervals (in milliseconds) between heartbeats.
        signal : np.array
            The physiological signal (e.g., ECG, PPG).
        fs : int, optional
            The sampling frequency in Hz. Default is 1000 Hz.
        """
        if nn_intervals is None:
            rr_transformation = RRTransformation(
                signals, fs=fs, signal_type=signal_type, options=options
            )
            rr_intervals = rr_transformation.process_rr_intervals()

            # Ensure valid RR intervals are returned
            if rr_intervals is None or len(rr_intervals) == 0:
                raise ValueError(
                    "RR interval transformation failed to extract valid RR intervals."
                )

            self.nn_intervals = np.array(rr_intervals)
        else:
            self.nn_intervals = np.array(nn_intervals)

        # Check for invalid NN intervals
        if len(self.nn_intervals) == 0:
            raise ValueError("NN intervals cannot be empty.")
        if np.all(self.nn_intervals == 0):
            raise ValueError("NN intervals cannot contain all zeros.")

        self.signal = np.array(signals)
        self.fs = fs

    def compute_all_features(self, include_complex_methods=False, **kwargs):
        """
        Computes all nonlinear features of the signal, with an option to skip
        time-consuming methods.

        Args:
            include_complex_methods (bool): Whether to compute the time-consuming
                methods: compute_sample_entropy, compute_approximate_entropy, and
                compute_recurrence_features.
            **kwargs: Additional parameters for specific feature computations.

        Returns:
            dict: A dictionary containing all the computed features.

        Example usage
        -------
        >>> nn_intervals = [800, 810, 790, 805, 795]  # NN intervals in ms
        >>> ecg_signal = np.random.randn(1000)  # Example ECG signal
        >>> fs = 1000  # Sampling frequency in Hz
        >>> hrv = HRVFeatures(nn_intervals, ecg_signal, fs)
        >>> all_features = hrv.compute_all_features()
        >>> print(all_features)
        """
        features = {}

        # Time-domain features
        time_features = TimeDomainFeatures(self.nn_intervals)
        for feature, method in [
            ("sdnn", time_features.compute_sdnn),
            ("rmssd", time_features.compute_rmssd),
            ("nn50", time_features.compute_nn50),
            ("pnn50", time_features.compute_pnn50),
            ("mean_nn", time_features.compute_mean_nn),
            ("median_nn", time_features.compute_median_nn),
            ("iqr_nn", time_features.compute_iqr_nn),
            ("std_nn", time_features.compute_std_nn),
            ("pnn_20", time_features.compute_pnn20),
            ("cvnn", time_features.compute_cvnn),
            ("hrv_triangular_index", time_features.compute_hrv_triangular_index),
            ("tinn", time_features.compute_tinn),
            ("sdsd", time_features.compute_sdsd),
        ]:
            try:
                features[feature] = method()
            except Exception as e:
                features[feature] = np.nan
                print(f"Error computing {feature}: {e}")

        # Frequency-domain features
        freq_features = FrequencyDomainFeatures(self.nn_intervals, fs=4)
        for feature, method in [
            # ("psd", freq_features.compute_psd),
            ("lf_power", freq_features.compute_lf),
            ("hf_power", freq_features.compute_hf),
            ("lf_hf_ratio", freq_features.compute_lf_hf_ratio),
            ("ulf_power", freq_features.compute_ulf),
            ("vlf_power", freq_features.compute_vlf),
            ("total_power", freq_features.compute_total_power),
            ("lfnu_power", freq_features.compute_lfnu),
            ("hfnu_power", freq_features.compute_hfnu),
        ]:
            try:
                features[feature] = method()
            except Exception as e:
                features[feature] = np.nan
                print(f"Error computing {feature}: {e}")

        # Nonlinear features
        nonlinear_features = NonlinearFeatures(self.signal, self.fs)
        for feature, method in [
            # ("sample_entropy", nonlinear_features.compute_sample_entropy),
            # ("approx_entropy", nonlinear_features.compute_approximate_entropy),
            ("fractal_dimension", nonlinear_features.compute_fractal_dimension),
            ("lyapunov_exponent", nonlinear_features.compute_lyapunov_exponent),
            ("dfa", nonlinear_features.compute_dfa),
            (
                "poincare",
                lambda: nonlinear_features.compute_poincare_features(self.nn_intervals),
            ),
            # ("recurrence", nonlinear_features.compute_recurrence_features),
        ]:
            try:
                if feature == "recurrence":
                    recurrence_dict = method()
                    features["recurrence_rate"] = recurrence_dict["recurrence_rate"]
                    features["determinism"] = recurrence_dict["determinism"]
                    features["laminarity"] = recurrence_dict["laminarity"]
                elif feature == "poincare":
                    poincare_sd1, poincare_sd2 = method()
                    features["poincare_sd1"] = poincare_sd1
                    features["poincare_sd2"] = poincare_sd2
                else:
                    features[feature] = method()
            except Exception as e:
                features[feature] = np.nan
                print(f"Error computing {feature}: {e}")

        try:
            if include_complex_methods:
                features["sample_entropy"] = nonlinear_features.compute_sample_entropy(
                    **kwargs
                )
                features["approximate_entropy"] = (
                    nonlinear_features.compute_approximate_entropy(**kwargs)
                )
                recurrence_dict = nonlinear_features.compute_recurrence_features(
                    **kwargs
                )
                features["recurrence_rate"] = recurrence_dict["recurrence_rate"]
                features["determinism"] = recurrence_dict["determinism"]
                features["laminarity"] = recurrence_dict["laminarity"]
            else:
                features["sample_entropy"] = None
                features["approximate_entropy"] = None
                features["recurrence_rate"] = None
                features["determinism"] = None
                features["laminarity"] = None
        except Exception as e:
            print(f"Error computing complex features: {e}")
            features["sample_entropy"] = None
            features["approximate_entropy"] = None
            features["recurrence_rate"] = None
            features["determinism"] = None
            features["laminarity"] = None

        return features

1	import numpy as np	1✔
2	from vitalDSP.physiological_features.time_domain import TimeDomainFeatures	1✔
3	from vitalDSP.physiological_features.frequency_domain import FrequencyDomainFeatures	1✔
4	from vitalDSP.physiological_features.nonlinear import NonlinearFeatures	1✔
5	from vitalDSP.transforms.beats_transformation import RRTransformation	1✔
6
7
8	class HRVFeatures:	1✔
9	"""
10	A class to compute HRV features from physiological signals such as ECG and PPG.
11
12	It combines time-domain, frequency-domain, and nonlinear features into one comprehensive feature extraction module.
13
14	Attributes
15	----------
16	nn_intervals : np.array
17	The NN intervals (in milliseconds) between heartbeats.
18	signal : np.array
19	The physiological signal (ECG, PPG, etc.).
20	fs : int
21	The sampling frequency in Hz. Default is 1000 Hz.
22
23	Methods
24	-------
25	compute_all_features()
26	Computes all HRV features and returns them in a dictionary format.
27	"""
28
29	def __init__(	1✔
30	self, signals, nn_intervals=None, fs=100, signal_type="ppg", options=None
31	):
32	"""
33	Initializes the HRVFeatures object.
34
35	Parameters
36	----------
37	nn_intervals : np.array
38	The NN intervals (in milliseconds) between heartbeats.
39	signal : np.array
40	The physiological signal (e.g., ECG, PPG).
41	fs : int, optional
42	The sampling frequency in Hz. Default is 1000 Hz.
43	"""
44	if nn_intervals is None:	1✔
45	rr_transformation = RRTransformation(	1✔
46	signals, fs=fs, signal_type=signal_type, options=options
47	)
48	rr_intervals = rr_transformation.process_rr_intervals()	1✔
49
50	# Ensure valid RR intervals are returned
51	if rr_intervals is None or len(rr_intervals) == 0:	×
52	raise ValueError(	×
53	"RR interval transformation failed to extract valid RR intervals."
54	)
55
56	self.nn_intervals = np.array(rr_intervals)	×
57	else:
58	self.nn_intervals = np.array(nn_intervals)	1✔
59
60	# Check for invalid NN intervals
61	if len(self.nn_intervals) == 0:	1✔
62	raise ValueError("NN intervals cannot be empty.")	1✔
63	if np.all(self.nn_intervals == 0):	1✔
64	raise ValueError("NN intervals cannot contain all zeros.")	1✔
65
66	self.signal = np.array(signals)	1✔
67	self.fs = fs	1✔
68
69	def compute_all_features(self, include_complex_methods=False, **kwargs):	1✔
70	"""
71	Computes all nonlinear features of the signal, with an option to skip
72	time-consuming methods.
73
74	Args:
75	include_complex_methods (bool): Whether to compute the time-consuming
76	methods: compute_sample_entropy, compute_approximate_entropy, and
77	compute_recurrence_features.
78	**kwargs: Additional parameters for specific feature computations.
79
80	Returns:
81	dict: A dictionary containing all the computed features.
82
83	Example usage
84	-------
85	>>> nn_intervals = [800, 810, 790, 805, 795] # NN intervals in ms
86	>>> ecg_signal = np.random.randn(1000) # Example ECG signal
87	>>> fs = 1000 # Sampling frequency in Hz
88	>>> hrv = HRVFeatures(nn_intervals, ecg_signal, fs)
89	>>> all_features = hrv.compute_all_features()
90	>>> print(all_features)
91	"""
92	features = {}	1✔
93
94	# Time-domain features
95	time_features = TimeDomainFeatures(self.nn_intervals)	1✔
96	for feature, method in [	1✔
97	("sdnn", time_features.compute_sdnn),
98	("rmssd", time_features.compute_rmssd),
99	("nn50", time_features.compute_nn50),
100	("pnn50", time_features.compute_pnn50),
101	("mean_nn", time_features.compute_mean_nn),
102	("median_nn", time_features.compute_median_nn),
103	("iqr_nn", time_features.compute_iqr_nn),
104	("std_nn", time_features.compute_std_nn),
105	("pnn_20", time_features.compute_pnn20),
106	("cvnn", time_features.compute_cvnn),
107	("hrv_triangular_index", time_features.compute_hrv_triangular_index),
108	("tinn", time_features.compute_tinn),
109	("sdsd", time_features.compute_sdsd),
110	]:
111	try:	1✔
112	features[feature] = method()	1✔
113	except Exception as e:	×
114	features[feature] = np.nan	×
115	print(f"Error computing {feature}: {e}")	×
116
117	# Frequency-domain features
118	freq_features = FrequencyDomainFeatures(self.nn_intervals, fs=4)	1✔
119	for feature, method in [	1✔
120	# ("psd", freq_features.compute_psd),
121	("lf_power", freq_features.compute_lf),
122	("hf_power", freq_features.compute_hf),
123	("lf_hf_ratio", freq_features.compute_lf_hf_ratio),
124	("ulf_power", freq_features.compute_ulf),
125	("vlf_power", freq_features.compute_vlf),
126	("total_power", freq_features.compute_total_power),
127	("lfnu_power", freq_features.compute_lfnu),
128	("hfnu_power", freq_features.compute_hfnu),
129	]:
130	try:	1✔
131	features[feature] = method()	1✔
132	except Exception as e:	×
133	features[feature] = np.nan	×
134	print(f"Error computing {feature}: {e}")	×
135
136	# Nonlinear features
137	nonlinear_features = NonlinearFeatures(self.signal, self.fs)	1✔
138	for feature, method in [	1✔
139	# ("sample_entropy", nonlinear_features.compute_sample_entropy),
140	# ("approx_entropy", nonlinear_features.compute_approximate_entropy),
141	("fractal_dimension", nonlinear_features.compute_fractal_dimension),
142	("lyapunov_exponent", nonlinear_features.compute_lyapunov_exponent),
143	("dfa", nonlinear_features.compute_dfa),
144	(
145	"poincare",
146	lambda: nonlinear_features.compute_poincare_features(self.nn_intervals),
147	),
148	# ("recurrence", nonlinear_features.compute_recurrence_features),
149	]:
150	try:	1✔
151	if feature == "recurrence":	1!
152	recurrence_dict = method()	×
153	features["recurrence_rate"] = recurrence_dict["recurrence_rate"]	×
154	features["determinism"] = recurrence_dict["determinism"]	×
155	features["laminarity"] = recurrence_dict["laminarity"]	×
156	elif feature == "poincare":	1✔
157	poincare_sd1, poincare_sd2 = method()	1✔
158	features["poincare_sd1"] = poincare_sd1	1✔
159	features["poincare_sd2"] = poincare_sd2	1✔
160	else:
161	features[feature] = method()	1✔
162	except Exception as e:	×
163	features[feature] = np.nan	×
164	print(f"Error computing {feature}: {e}")	×
165
166	try:	1✔
167	if include_complex_methods:	1!
168	features["sample_entropy"] = nonlinear_features.compute_sample_entropy(	×
169	**kwargs
170	)
NEW 171	features["approximate_entropy"] = (	×
172	nonlinear_features.compute_approximate_entropy(**kwargs)
173	)
UNCOV 174	recurrence_dict = nonlinear_features.compute_recurrence_features(	×
175	**kwargs
176	)
177	features["recurrence_rate"] = recurrence_dict["recurrence_rate"]	×
178	features["determinism"] = recurrence_dict["determinism"]	×
179	features["laminarity"] = recurrence_dict["laminarity"]	×
180	else:
181	features["sample_entropy"] = None	1✔
182	features["approximate_entropy"] = None	1✔
183	features["recurrence_rate"] = None	1✔
184	features["determinism"] = None	1✔
185	features["laminarity"] = None	1✔
186	except Exception as e:	×
187	print(f"Error computing complex features: {e}")	×
188	features["sample_entropy"] = None	×
189	features["approximate_entropy"] = None	×
190	features["recurrence_rate"] = None	×
191	features["determinism"] = None	×
192	features["laminarity"] = None	×
193
194	return features	1✔

Oucru-Innovations / vital-DSP / 10909444855

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