add unit test for timing

py_neuromodulation/nm_generator.py

@@ -25,13 +25,13 @@ def raw_data_generator(
     offset_time = settings["segment_length_features_ms"]
     offset_start = offset_time / 1000 * sfreq
 
-    cnt_fsnew = 0
+    ratio_samples_features = sfreq / sfreq_new
+
+    ratio_counter = 0
     for cnt in range(data.shape[1]+1):  # shape + 1 guarantees that the last sample is also included
-        if cnt < offset_start:
-            cnt_fsnew += 1
-            continue
 
-        cnt_fsnew += 1
-        if cnt_fsnew >= (sfreq / sfreq_new):
-            cnt_fsnew = 0
+        if (cnt - offset_start) >= ratio_samples_features * ratio_counter:
+
+            ratio_counter += 1
+
             yield data[:, np.floor(cnt-offset_start).astype(int) : cnt]

tests/test_timing.py

@@ -0,0 +1,80 @@
+import py_neuromodulation as pn
+from py_neuromodulation import (
+    nm_settings,
+)
+
+import numpy as np
+
+
+def test_setting_computation_time():
+    """Intantiate test for feature computation with fixed time duration.
+    The number of output features should match the ratio of the sampling rate of the data and raw signal sampling rate.
+    """
+
+    data_duration_s = 5
+    sampling_rate_features_hz = 200
+    fs = 1000
+    data = np.random.random((1, int(data_duration_s * fs)))
+
+    settings = nm_settings.get_default_settings()
+    settings = nm_settings.set_settings_fast_compute(settings)
+    settings["segment_length_features_ms"] = 1000  # start afte 1 second
+    settings["features"]["fft"] = False
+    settings["features"]["raw_hjorth"] = True
+    stream = pn.Stream(
+        sfreq=fs,
+        data=data,
+        sampling_rate_features_hz=sampling_rate_features_hz,
+        settings=settings,
+    )
+
+    features = stream.run()
+
+    # test if features up till the last sample was computed
+    assert (
+        data_duration_s * 1000 - features["time"].iloc[-1]
+    ) < 1000 / sampling_rate_features_hz
+
+    # test that the time difference between two samples is the feature sampling rate
+    assert (
+        features["time"].iloc[1] - features["time"].iloc[0]
+    ) == 1000 / sampling_rate_features_hz
+
+    assert features["time"].iloc[0] == settings["segment_length_features_ms"]
+
+
+def test_float_fs():
+    """Change sampling rate here to be float, s.t. rounding issues are not affecting overall number of
+    computed features.
+    """
+
+    data_duration_s = 5
+    sampling_rate_features_hz = 200
+    fs = 1111.111
+    data = np.random.random((1, int(data_duration_s * fs)))
+
+    settings = nm_settings.get_default_settings()
+    settings = nm_settings.set_settings_fast_compute(settings)
+    settings["segment_length_features_ms"] = 333  # start afte 1 second
+
+    settings["features"]["fft"] = False
+    settings["features"]["raw_hjorth"] = True
+    stream = pn.Stream(
+        sfreq=fs,
+        data=data,
+        sampling_rate_features_hz=sampling_rate_features_hz,
+        settings=settings,
+    )
+
+    features = stream.run()
+
+    # test if features up till the last sample was computed
+    assert (
+        data_duration_s * 1000 - features["time"].iloc[-1]
+    ) < 1000 / sampling_rate_features_hz
+
+    assert (
+        features["time"].iloc[1] - features["time"].iloc[0]
+    ) == 1000 / sampling_rate_features_hz
+
+    assert features["time"].iloc[0] == settings["segment_length_features_ms"]