import numpy as np
from wfdb.io.record import rdrecord
from wfdb.io.convert.edf import read_edf
class TestConvert:
def test_edf_uniform(self):
"""
EDF format conversion to MIT for uniform sample rates.
"""
# Uniform sample rates
record_MIT = rdrecord("sample-data/n16").__dict__
record_EDF = read_edf("sample-data/n16.edf").__dict__
fields = list(record_MIT.keys())
# Original MIT format method of checksum is outdated, sometimes
# the same value though
fields.remove("checksum")
# Original MIT format units are less comprehensive since they
# default to mV if unknown.. therefore added more default labels
fields.remove("units")
test_results = []
for field in fields:
# Signal value will be slightly off due to C to Python type conversion
if field == "p_signal":
true_array = np.array(record_MIT[field]).flatten()
pred_array = np.array(record_EDF[field]).flatten()
# Prevent divide by zero warning
for i, v in enumerate(true_array):
if v == 0:
true_array[i] = 1
pred_array[i] = 1
sig_diff = np.abs((pred_array - true_array) / true_array)
sig_diff[sig_diff == -np.inf] = 0
sig_diff[sig_diff == np.inf] = 0
sig_diff = np.nanmean(sig_diff, 0)
# 5% tolerance
if np.max(sig_diff) <= 5:
test_results.append(True)
else:
test_results.append(False)
elif field == "init_value":
signal_diff = [
abs(record_MIT[field][i] - record_EDF[field][i])
for i in range(len(record_MIT[field]))
]
if abs(max(min(signal_diff), max(signal_diff), key=abs)) <= 2:
test_results.append(True)
else:
test_results.append(False)
else:
test_results.append(record_MIT[field] == record_MIT[field])
target_results = len(fields) * [True]
assert np.array_equal(test_results, target_results)
def test_edf_non_uniform(self):
"""
EDF format conversion to MIT for non-uniform sample rates.
"""
# Non-uniform sample rates
record_MIT = rdrecord("sample-data/wave_4").__dict__
record_EDF = read_edf("sample-data/wave_4.edf").__dict__
fields = list(record_MIT.keys())
# Original MIT format method of checksum is outdated, sometimes
# the same value though
fields.remove("checksum")
# Original MIT format units are less comprehensive since they
# default to mV if unknown.. therefore added more default labels
fields.remove("units")
test_results = []
for field in fields:
# Signal value will be slightly off due to C to Python type conversion
if field == "p_signal":
true_array = np.array(record_MIT[field]).flatten()
pred_array = np.array(record_EDF[field]).flatten()
# Prevent divide by zero warning
for i, v in enumerate(true_array):
if v == 0:
true_array[i] = 1
pred_array[i] = 1
sig_diff = np.abs((pred_array - true_array) / true_array)
sig_diff[sig_diff == -np.inf] = 0
sig_diff[sig_diff == np.inf] = 0
sig_diff = np.nanmean(sig_diff, 0)
# 5% tolerance
if np.max(sig_diff) <= 5:
test_results.append(True)
else:
test_results.append(False)
elif field == "init_value":
signal_diff = [
abs(record_MIT[field][i] - record_EDF[field][i])
for i in range(len(record_MIT[field]))
]
if abs(max(min(signal_diff), max(signal_diff), key=abs)) <= 2:
test_results.append(True)
else:
test_results.append(False)
else:
test_results.append(record_MIT[field] == record_MIT[field])
target_results = len(fields) * [True]
assert np.array_equal(test_results, target_results)