import numpy as np
import wfdb
from wfdb import processing
class test_processing:
"""
Test processing functions
"""
def test_resample_single(self):
sig, fields = wfdb.rdsamp("sample-data/100")
ann = wfdb.rdann("sample-data/100", "atr")
fs = fields["fs"]
fs_target = 50
new_sig, new_ann = processing.resample_singlechan(
sig[:, 0], ann, fs, fs_target
)
expected_length = int(sig.shape[0] * fs_target / fs)
assert new_sig.shape[0] == expected_length
def test_resample_multi(self):
sig, fields = wfdb.rdsamp("sample-data/100")
ann = wfdb.rdann("sample-data/100", "atr")
fs = fields["fs"]
fs_target = 50
new_sig, new_ann = processing.resample_multichan(
sig, ann, fs, fs_target
)
expected_length = int(sig.shape[0] * fs_target / fs)
assert new_sig.shape[0] == expected_length
assert new_sig.shape[1] == sig.shape[1]
def test_normalize_bound(self):
sig, _ = wfdb.rdsamp("sample-data/100")
lb = -5
ub = 15
x = processing.normalize_bound(sig[:, 0], lb, ub)
assert x.shape[0] == sig.shape[0]
assert np.min(x) >= lb
assert np.max(x) <= ub
def test_find_peaks(self):
x = [0, 2, 1, 0, -10, -15, -15, -15, 9, 8, 0, 0, 1, 2, 10]
hp, sp = processing.find_peaks(x)
assert np.array_equal(hp, [1, 8])
assert np.array_equal(sp, [6, 10])
def test_find_peaks_empty(self):
x = []
hp, sp = processing.find_peaks(x)
assert hp.shape == (0,)
assert sp.shape == (0,)
def test_gqrs(self):
record = wfdb.rdrecord(
"sample-data/100",
channels=[0],
sampfrom=9998,
sampto=19998,
physical=False,
)
expected_peaks = [
271,
580,
884,
1181,
1469,
1770,
2055,
2339,
2634,
2939,
3255,
3551,
3831,
4120,
4412,
4700,
5000,
5299,
5596,
5889,
6172,
6454,
6744,
7047,
7347,
7646,
7936,
8216,
8503,
8785,
9070,
9377,
9682,
]
peaks = processing.gqrs_detect(
d_sig=record.d_signal[:, 0],
fs=record.fs,
adc_gain=record.adc_gain[0],
adc_zero=record.adc_zero[0],
threshold=1.0,
)
assert np.array_equal(peaks, expected_peaks)
def test_correct_peaks(self):
sig, fields = wfdb.rdsamp("sample-data/100")
ann = wfdb.rdann("sample-data/100", "atr")
fs = fields["fs"]
min_bpm = 10
max_bpm = 350
min_gap = fs * 60 / min_bpm
max_gap = fs * 60 / max_bpm
y_idxs = processing.correct_peaks(
sig=sig[:, 0],
peak_inds=ann.sample,
search_radius=int(max_gap),
smooth_window_size=150,
)
yz = np.zeros(sig.shape[0])
yz[y_idxs] = 1
yz = np.where(yz[:10000] == 1)[0]
expected_peaks = [
77,
370,
663,
947,
1231,
1515,
1809,
2045,
2403,
2706,
2998,
3283,
3560,
3863,
4171,
4466,
4765,
5061,
5347,
5634,
5919,
6215,
6527,
6824,
7106,
7393,
7670,
7953,
8246,
8539,
8837,
9142,
9432,
9710,
9998,
]
assert np.array_equal(yz, expected_peaks)
class test_qrs:
"""
Testing QRS detectors
"""
def test_xqrs(self):
"""
Run XQRS detector on record 100 and compare to reference annotations
"""
sig, fields = wfdb.rdsamp("sample-data/100", channels=[0])
ann_ref = wfdb.rdann("sample-data/100", "atr")
xqrs = processing.XQRS(sig=sig[:, 0], fs=fields["fs"])
xqrs.detect()
comparitor = processing.compare_annotations(
ann_ref.sample[1:], xqrs.qrs_inds, int(0.1 * fields["fs"])
)
assert comparitor.sensitivity > 0.99
assert comparitor.positive_predictivity > 0.99