[Add Check]: Spike times not in unobserved intervals (#168)

* main work on check/tests

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Update nwbinspector/checks/ecephys.py

Co-authored-by: Ben Dichter <[email protected]>

* fix increment from auto commit

* added another test to further illustrate

* Update nwbinspector/checks/ecephys.py

Co-authored-by: Ben Dichter <[email protected]>

* add broken test

* fixed logic

Co-authored-by: Cody Baker <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Ben Dichter <[email protected]>

Author: 4 people

nwbinspector/checks/ecephys.py

@@ -50,3 +50,28 @@ def check_electrical_series_dims(electrical_series: ElectricalSeries):
 def check_electrical_series_reference_electrodes_table(electrical_series: ElectricalSeries):
     if electrical_series.electrodes.table.name != "electrodes":
         return InspectorMessage(message="electrodes does not  reference an electrodes table.")
+
+
+@register_check(importance=Importance.BEST_PRACTICE_VIOLATION, neurodata_type=Units)
+def check_spike_times_not_in_unobserved_interval(units_table: Units, nunits: int = 4):
+    """Check if a Units table has spike times that occur outside of observed intervals."""
+    if not units_table.obs_intervals:
+        return
+    for unit_spike_times, unit_obs_intervals in zip(
+        units_table["spike_times"][:nunits], units_table["obs_intervals"][:nunits]
+    ):
+        spike_times_array = np.array(unit_spike_times)
+        if not all(
+            sum(
+                [
+                    np.logical_and(start <= spike_times_array, spike_times_array <= stop)
+                    for start, stop in unit_obs_intervals
+                ]
+            )
+        ):
+            return InspectorMessage(
+                message=(
+                    "This Units table contains spike times that occur during periods of time not labeled as being "
+                    "observed intervals."
+                )
+            )

tests/unit_tests/test_ecephys.py

@@ -12,6 +12,7 @@
     check_negative_spike_times,
     check_electrical_series_dims,
     check_electrical_series_reference_electrodes_table,
+    check_spike_times_not_in_unobserved_interval,
 )
 from nwbinspector.register_checks import InspectorMessage, Importance
 
@@ -152,3 +153,64 @@ def test_trigger_check_electrical_series_reference_electrodes_table(self):
             check_electrical_series_reference_electrodes_table(electrical_series).message
             == "electrodes does not  reference an electrodes table."
         )
+
+
+def test_check_spike_times_not_in_unobserved_interval_pass():
+    units_table = Units(name="TestUnits")
+    units_table.add_unit(spike_times=[1, 2, 3], obs_intervals=[[0, 2.5], [2.7, 3.5]])
+    assert check_spike_times_not_in_unobserved_interval(units_table=units_table) is None
+
+
+def test_check_spike_times_not_in_unobserved_interval_pass_no_intervals():
+    units_table = Units(name="TestUnits")
+    units_table.add_unit(spike_times=[1, 2, 3])
+    assert check_spike_times_not_in_unobserved_interval(units_table=units_table) is None
+
+
+def test_check_spike_times_not_in_unobserved_interval_1():
+    units_table = Units(name="TestUnits")
+    units_table.add_unit(spike_times=[1, 2, 3], obs_intervals=[[0, 2.5], [3.5, 4]])
+    assert check_spike_times_not_in_unobserved_interval(units_table=units_table) == InspectorMessage(
+        message=(
+            "This Units table contains spike times that occur during periods of time not labeled as being "
+            "observed intervals."
+        ),
+        importance=Importance.BEST_PRACTICE_VIOLATION,
+        check_function_name="check_spike_times_not_in_unobserved_interval",
+        object_type="Units",
+        object_name="TestUnits",
+        location="/",
+    )
+
+
+def test_check_spike_times_not_in_unobserved_interval_2():
+    units_table = Units(name="TestUnits")
+    units_table.add_unit(spike_times=[1, 2, 3, 4, 5, 6], obs_intervals=[[0, 2.5], [3.5, 7]])
+    assert check_spike_times_not_in_unobserved_interval(units_table=units_table) == InspectorMessage(
+        message=(
+            "This Units table contains spike times that occur during periods of time not labeled as being "
+            "observed intervals."
+        ),
+        importance=Importance.BEST_PRACTICE_VIOLATION,
+        check_function_name="check_spike_times_not_in_unobserved_interval",
+        object_type="Units",
+        object_name="TestUnits",
+        location="/",
+    )
+
+
+def test_check_spike_times_not_in_unobserved_interval_multiple_units():
+    units_table = Units(name="TestUnits")
+    units_table.add_unit(spike_times=[1, 2, 3, 4, 5, 6], obs_intervals=[[0, 3.2], [3.5, 7]])
+    units_table.add_unit(spike_times=[6.5, 12, 13, 14, 15, 16], obs_intervals=[[10, 15.2], [15.8, 17]])
+    assert check_spike_times_not_in_unobserved_interval(units_table=units_table) == InspectorMessage(
+        message=(
+            "This Units table contains spike times that occur during periods of time not labeled as being "
+            "observed intervals."
+        ),
+        importance=Importance.BEST_PRACTICE_VIOLATION,
+        check_function_name="check_spike_times_not_in_unobserved_interval",
+        object_type="Units",
+        object_name="TestUnits",
+        location="/",
+    )