Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_signal_findpeaks():
signal1 = np.cos(np.linspace(start=0, stop=30, num=1000))
info1 = nk.signal_findpeaks(signal1)
signal2 = np.concatenate([np.arange(0, 20, 0.1), np.arange(17, 30, 0.1), np.arange(30, 10, -0.1)])
info2 = nk.signal_findpeaks(signal2)
assert len(info1["Peaks"]) > len(info2["Peaks"])
def test_rsp_simulate():
rsp1 = nk.rsp_simulate(duration=20, length=3000, random_state=42)
assert len(rsp1) == 3000
rsp2 = nk.rsp_simulate(duration=20, length=3000, respiratory_rate=80, random_state=42)
# pd.DataFrame({"RSP1":rsp1, "RSP2":rsp2}).plot()
# pd.DataFrame({"RSP1":rsp1, "RSP2":rsp2}).hist()
assert len(nk.signal_findpeaks(rsp1, height_min=0.2)["Peaks"]) < len(
nk.signal_findpeaks(rsp2, height_min=0.2)["Peaks"]
)
rsp3 = nk.rsp_simulate(duration=20, length=3000, method="sinusoidal", random_state=42)
rsp4 = nk.rsp_simulate(duration=20, length=3000, method="breathmetrics", random_state=42)
# pd.DataFrame({"RSP3":rsp3, "RSP4":rsp4}).plot()
assert len(nk.signal_findpeaks(rsp3, height_min=0.2)["Peaks"]) > len(
nk.signal_findpeaks(rsp4, height_min=0.2)["Peaks"]
)
def test_eda_simulate():
eda1 = nk.eda_simulate(duration=10, length=None, scr_number=1, random_state=333)
assert len(nk.signal_findpeaks(eda1, height_min=0.6)["Peaks"]) == 1
eda2 = nk.eda_simulate(duration=10, length=None, scr_number=5, random_state=333)
assert len(nk.signal_findpeaks(eda2, height_min=0.6)["Peaks"]) == 5
# pd.DataFrame({"EDA1": eda1, "EDA2": eda2}).plot()
assert len(nk.signal_findpeaks(eda2, height_min=0.6)["Peaks"]) > len(
nk.signal_findpeaks(eda1, height_min=0.6)["Peaks"]
)
def test_signal_rate(): # since singal_rate wraps signal_period, the latter is tested as well
# Test with array.
duration = 10
sampling_rate = 1000
signal = nk.signal_simulate(duration=duration, sampling_rate=sampling_rate, frequency=1)
info = nk.signal_findpeaks(signal)
rate = nk.signal_rate(peaks=info["Peaks"], sampling_rate=1000, desired_length=len(signal))
assert rate.shape[0] == duration * sampling_rate
# Test with dictionary.produced from signal_findpeaks.
assert info[list(info.keys())[0]].shape == (info["Peaks"].shape[0],)
# Test with DataFrame.
duration = 120
sampling_rate = 1000
rsp = nk.rsp_simulate(
duration=duration, sampling_rate=sampling_rate, respiratory_rate=15, method="sinuosoidal", noise=0
)
rsp_cleaned = nk.rsp_clean(rsp, sampling_rate=sampling_rate)
signals, info = nk.rsp_peaks(rsp_cleaned)
rate = nk.signal_rate(signals, sampling_rate=sampling_rate, desired_length=duration * sampling_rate)
assert rate.shape == (signals.shape[0],)
def test_emg_simulate():
emg1 = nk.emg_simulate(duration=20, length=5000, burst_number=1)
assert len(emg1) == 5000
emg2 = nk.emg_simulate(duration=20, length=5000, burst_number=15)
assert scipy.stats.median_absolute_deviation(emg1) < scipy.stats.median_absolute_deviation(emg2)
emg3 = nk.emg_simulate(duration=20, length=5000, burst_number=1, burst_duration=2.0)
# pd.DataFrame({"EMG1":emg1, "EMG3": emg3}).plot()
assert len(nk.signal_findpeaks(emg3, height_min=1.0)["Peaks"]) > len(
nk.signal_findpeaks(emg1, height_min=1.0)["Peaks"]
)
def test_rsp_simulate():
rsp1 = nk.rsp_simulate(duration=20, length=3000, random_state=42)
assert len(rsp1) == 3000
rsp2 = nk.rsp_simulate(duration=20, length=3000, respiratory_rate=80, random_state=42)
# pd.DataFrame({"RSP1":rsp1, "RSP2":rsp2}).plot()
# pd.DataFrame({"RSP1":rsp1, "RSP2":rsp2}).hist()
assert len(nk.signal_findpeaks(rsp1, height_min=0.2)["Peaks"]) < len(
nk.signal_findpeaks(rsp2, height_min=0.2)["Peaks"]
)
rsp3 = nk.rsp_simulate(duration=20, length=3000, method="sinusoidal", random_state=42)
rsp4 = nk.rsp_simulate(duration=20, length=3000, method="breathmetrics", random_state=42)
# pd.DataFrame({"RSP3":rsp3, "RSP4":rsp4}).plot()
assert len(nk.signal_findpeaks(rsp3, height_min=0.2)["Peaks"]) > len(
nk.signal_findpeaks(rsp4, height_min=0.2)["Peaks"]
)
def test_rsp_simulate():
rsp1 = nk.rsp_simulate(duration=20, length=3000, random_state=42)
assert len(rsp1) == 3000
rsp2 = nk.rsp_simulate(duration=20, length=3000, respiratory_rate=80, random_state=42)
# pd.DataFrame({"RSP1":rsp1, "RSP2":rsp2}).plot()
# pd.DataFrame({"RSP1":rsp1, "RSP2":rsp2}).hist()
assert len(nk.signal_findpeaks(rsp1, height_min=0.2)["Peaks"]) < len(
nk.signal_findpeaks(rsp2, height_min=0.2)["Peaks"]
)
rsp3 = nk.rsp_simulate(duration=20, length=3000, method="sinusoidal", random_state=42)
rsp4 = nk.rsp_simulate(duration=20, length=3000, method="breathmetrics", random_state=42)
# pd.DataFrame({"RSP3":rsp3, "RSP4":rsp4}).plot()
assert len(nk.signal_findpeaks(rsp3, height_min=0.2)["Peaks"]) > len(
nk.signal_findpeaks(rsp4, height_min=0.2)["Peaks"]
)
def test_eda_simulate():
eda1 = nk.eda_simulate(duration=10, length=None, scr_number=1, random_state=333)
assert len(nk.signal_findpeaks(eda1, height_min=0.6)["Peaks"]) == 1
eda2 = nk.eda_simulate(duration=10, length=None, scr_number=5, random_state=333)
assert len(nk.signal_findpeaks(eda2, height_min=0.6)["Peaks"]) == 5
# pd.DataFrame({"EDA1": eda1, "EDA2": eda2}).plot()
assert len(nk.signal_findpeaks(eda2, height_min=0.6)["Peaks"]) > len(
nk.signal_findpeaks(eda1, height_min=0.6)["Peaks"]
)
data_rmse = pd.DataFrame(columns=["RMSE", "Index", "Participant", "Task", "Function"])
for i in range(4):
data = pd.read_csv("../../data/eogdb/eogdb_task" + str(i + 1) + ".csv")
for j, participant in enumerate(np.unique(data["Participant"])[1:3]):
segment = data[data["Participant"] == participant]
signal = segment["vEOG"]
cleaned = nk.eog_clean(signal, sampling_rate=200, method='neurokit')
blinks = nk.signal_findpeaks(cleaned, relative_height_min=1.5)["Peaks"]
events = nk.epochs_create(cleaned, blinks, sampling_rate=200, epochs_start=-0.4, epochs_end=0.6)
events = nk.epochs_to_array(events) # Convert to 2D array
# Rescale
for i in range(events.shape[1]):
events[:, i] = nk.rescale(events[:, i], to=[0, 1]) # Reshape to 0-1 scale
# RMSE - Gamma
rmse = pd.DataFrame({"RMSE": [nk.fit_rmse(events[:, i], template_gamma) for i in range(events.shape[1])],
"Index": range(events.shape[1]),
"Participant": [participant]*events.shape[1],
"Task": [data["Task"][0]]*events.shape[1],
"Function": ["Gamma"] * events.shape[1]})
rmse["Index"] = rmse["Participant"] + "_" + rmse["Task"] + "_" + rmse["Index"].astype(str)
data_rmse = pd.concat([data_rmse, rmse], axis=0)