From feaf6efc825d40ad9e3a520ccd66aa0eb247ac71 Mon Sep 17 00:00:00 2001 From: Andrew Barros Date: Wed, 10 Jul 2024 10:42:58 -0400 Subject: [PATCH 1/5] Return an empty array rather than throwing an exception if no QRS data is found. --- neurokit2/ecg/ecg_findpeaks.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/neurokit2/ecg/ecg_findpeaks.py b/neurokit2/ecg/ecg_findpeaks.py index 6a1f34e61b..b1dfd65c19 100644 --- a/neurokit2/ecg/ecg_findpeaks.py +++ b/neurokit2/ecg/ecg_findpeaks.py @@ -274,6 +274,9 @@ def _ecg_findpeaks_neurokit( qrs = smoothgrad > gradthreshold beg_qrs = np.where(np.logical_and(np.logical_not(qrs[0:-1]), qrs[1:]))[0] end_qrs = np.where(np.logical_and(qrs[0:-1], np.logical_not(qrs[1:])))[0] + + if len(beg_qrs) == 0: + return [] # Throw out QRS-ends that precede first QRS-start. end_qrs = end_qrs[end_qrs > beg_qrs[0]] From 847bb1994be5402497c067a67d60beac98487feb Mon Sep 17 00:00:00 2001 From: Andrew Barros Date: Fri, 12 Jul 2024 09:18:34 -0400 Subject: [PATCH 2/5] add tests for all peak detection methods handling empty input --- tests/tests_ecg_findpeaks.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/tests_ecg_findpeaks.py b/tests/tests_ecg_findpeaks.py index c2ca3f4a1c..e6169ccd0b 100644 --- a/tests/tests_ecg_findpeaks.py +++ b/tests/tests_ecg_findpeaks.py @@ -13,6 +13,7 @@ _ecg_findpeaks_MWA, _ecg_findpeaks_peakdetect, _ecg_findpeaks_hamilton, + _ecg_findpeaks_findmethod, ) @@ -23,6 +24,23 @@ def _read_csv_column(csv_name, column): csv_data = pd.read_csv(csv_path, header=None) return csv_data[column].to_numpy() +def test_ecg_findpeaks_all_methods_handle_empty_input(): + METHODS = ["neurokit", "pantompkins", "nabian", "gamboa", + "slopesumfunction", "wqrs", "hamilton", "christov", + "engzee", "manikandan", "elgendi", "kalidas", + "martinez", "rodrigues", "vgraph"] + + failed_methods = [] + for method in METHODS: + try: + method_func = _ecg_findpeaks_findmethod(method) + _ = method_func(np.zeros(12*240), sampling_rate=240) + except Exception: + failed_methods.append(method) + continue + + np.testing.assert_equal(failed_methods, []) + def test_ecg_findpeaks_MWA(): np.testing.assert_array_equal( From 8738072ecc33a8aaa1bcfdf5a25e9ffb992e8154 Mon Sep 17 00:00:00 2001 From: Andrew Barros Date: Fri, 12 Jul 2024 09:40:58 -0400 Subject: [PATCH 3/5] fix empty input errors in other findpeak methods --- neurokit2/ecg/ecg_findpeaks.py | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/neurokit2/ecg/ecg_findpeaks.py b/neurokit2/ecg/ecg_findpeaks.py index b1dfd65c19..df3af656c6 100644 --- a/neurokit2/ecg/ecg_findpeaks.py +++ b/neurokit2/ecg/ecg_findpeaks.py @@ -276,7 +276,8 @@ def _ecg_findpeaks_neurokit( end_qrs = np.where(np.logical_and(qrs[0:-1], np.logical_not(qrs[1:])))[0] if len(beg_qrs) == 0: - return [] + return np.array([]) + # Throw out QRS-ends that precede first QRS-start. end_qrs = end_qrs[end_qrs > beg_qrs[0]] @@ -503,6 +504,14 @@ def _ecg_findpeaks_zong(signal, sampling_rate=1000, cutoff=16, window=0.13, **kw ret = np.pad(clt, (window_size - 1, 0), "constant", constant_values=(0, 0)) ret = np.convolve(ret, np.ones(window_size), "valid") + # Check that ret is at least as large as the window + if len(ret) < window_size: + warn( + f"The signal must be at least {window_size} samples long for peak detection with the Zong method. ", + category=NeuroKitWarning, + ) + return np.array([]) + for i in range(1, window_size): ret[i - 1] = ret[i - 1] / i ret[window_size - 1 :] = ret[window_size - 1 :] / window_size @@ -638,7 +647,8 @@ def _ecg_findpeaks_christov(signal, sampling_rate=1000, **kwargs): if len(RR) > 5: RR.pop(0) Rm = int(np.mean(RR)) - + if len(QRS) == 0: + return np.array([]) QRS.pop(0) QRS = np.array(QRS, dtype="int") return QRS @@ -919,6 +929,9 @@ def _ecg_findpeaks_engzee(signal, sampling_rate=1000, **kwargs): thi = False thf = False + if len(r_peaks) == 0: + return np.array([]) + r_peaks.pop( 0 ) # removing the 1st detection as it 1st needs the QRS complex amplitude for the threshold @@ -959,6 +972,12 @@ def running_mean(x, N): # Eq. 1: First-order differencing difference dn = np.append(filtered[1:], 0) - filtered + + # If the signal is flat then return an empty array rather than error out + # with a divide by zero error. + if np.max(abs(dn)) == 0: + return np.array([]) + # Eq. 2 dtn = dn / (np.max(abs(dn))) From 3cce0cf80528fce4c253a3cbef8c38b87bbd12b4 Mon Sep 17 00:00:00 2001 From: Andrew Barros Date: Fri, 12 Jul 2024 10:49:06 -0400 Subject: [PATCH 4/5] add better error message to the test, remove vgraph from testing set --- tests/tests_ecg_findpeaks.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/tests_ecg_findpeaks.py b/tests/tests_ecg_findpeaks.py index e6169ccd0b..8d7a5aa1ec 100644 --- a/tests/tests_ecg_findpeaks.py +++ b/tests/tests_ecg_findpeaks.py @@ -28,7 +28,7 @@ def test_ecg_findpeaks_all_methods_handle_empty_input(): METHODS = ["neurokit", "pantompkins", "nabian", "gamboa", "slopesumfunction", "wqrs", "hamilton", "christov", "engzee", "manikandan", "elgendi", "kalidas", - "martinez", "rodrigues", "vgraph"] + "martinez", "rodrigues"] failed_methods = [] for method in METHODS: @@ -38,8 +38,8 @@ def test_ecg_findpeaks_all_methods_handle_empty_input(): except Exception: failed_methods.append(method) continue - - np.testing.assert_equal(failed_methods, []) + if failed_methods: + raise Exception(f"Failed methods: {failed_methods}") def test_ecg_findpeaks_MWA(): From f0c35dcf7b1ef49ae01839ad477410b39e4fab90 Mon Sep 17 00:00:00 2001 From: Andrew Barros Date: Fri, 12 Jul 2024 12:37:02 -0400 Subject: [PATCH 5/5] Refactor the test to use parameterized testing. --- tests/tests_ecg_findpeaks.py | 23 +++++++++-------------- 1 file changed, 9 insertions(+), 14 deletions(-) diff --git a/tests/tests_ecg_findpeaks.py b/tests/tests_ecg_findpeaks.py index 8d7a5aa1ec..f80ee0c2be 100644 --- a/tests/tests_ecg_findpeaks.py +++ b/tests/tests_ecg_findpeaks.py @@ -3,6 +3,7 @@ import numpy as np import pandas as pd +import pytest # Trick to directly access internal functions for unit testing. # @@ -24,22 +25,16 @@ def _read_csv_column(csv_name, column): csv_data = pd.read_csv(csv_path, header=None) return csv_data[column].to_numpy() -def test_ecg_findpeaks_all_methods_handle_empty_input(): - METHODS = ["neurokit", "pantompkins", "nabian", "gamboa", +#vgraph is not included because it currently causes CI to fail (issue 1007) +@pytest.mark.parametrize("method",["neurokit", "pantompkins", "nabian", "gamboa", "slopesumfunction", "wqrs", "hamilton", "christov", "engzee", "manikandan", "elgendi", "kalidas", - "martinez", "rodrigues"] - - failed_methods = [] - for method in METHODS: - try: - method_func = _ecg_findpeaks_findmethod(method) - _ = method_func(np.zeros(12*240), sampling_rate=240) - except Exception: - failed_methods.append(method) - continue - if failed_methods: - raise Exception(f"Failed methods: {failed_methods}") + "martinez", "rodrigues",]) +def test_ecg_findpeaks_all_methods_handle_empty_input(method): + method_func = _ecg_findpeaks_findmethod(method) + # The test here is implicit: no exceptions means that it passed, + # even if the output is nonsense. + _ = method_func(np.zeros(12*240), sampling_rate=240) def test_ecg_findpeaks_MWA():