Add DecompositionSeries unit and source_timeseries checks

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### New Checks
 
 * Added `check_spatial_series_unit` to validate that SpatialSeries objects (outside of CompassDirection) have a recognized unit string from a curated allowlist of SI length units, angular units, and pixels. [#685](https://github.com/NeurodataWithoutBorders/nwbinspector/pull/685)
+* Added `check_decomposition_series_unit` to validate that DecompositionSeries has an appropriate unit for its metric (e.g. "radians" or "degrees" for phase, matching the source signal unit for amplitude). Also added `check_decomposition_series_source_timeseries` to suggest linking the source TimeSeries for provenance. [#676](https://github.com/NeurodataWithoutBorders/nwbinspector/pull/676)
 * Added `check_imaging_plane_location_allen_ccf`, `check_electrodes_location_allen_ccf`, and `check_intracellular_electrode_location_allen_ccf` to validate location fields against Allen Mouse Brain CCF ontology terms when subject species is mouse. [#671](https://github.com/NeurodataWithoutBorders/nwbinspector/pull/671)
 * Added `check_time_intervals_start_time_not_constant` to flag TimeIntervals tables where all start_time values are identical, indicating times were likely not set relative to session start. [#677](https://github.com/NeurodataWithoutBorders/nwbinspector/issues/677)
 * Added `check_units_resolution_is_set` to flag when the Units table has spike_times but resolution is not set to a meaningful positive float. [#686](https://github.com/NeurodataWithoutBorders/nwbinspector/pull/686)

docs/best_practices/best_practices_index.rst

@@ -31,5 +31,6 @@ Authors: Oliver Ruebel, Andrew Tritt, Ryan Ly, Cody Baker and Ben Dichter
    ogen
    image_series
    images
+   misc
    simulated_data
    extensions

docs/best_practices/misc.rst

@@ -0,0 +1,34 @@
+Miscellaneous
+=============
+
+
+DecompositionSeries
+-------------------
+
+:ref:`nwb-schema:sec-DecompositionSeries` stores the product of spectral analysis (e.g., wavelet decomposition, short-time
+Fourier transform). The data is 3D (time x channels x frequency bands) and has a ``metric`` field indicating what the
+values represent (recommended values: "phase", "amplitude", "power").
+
+.. _best_practice_decomposition_series_unit:
+
+DecompositionSeries Unit
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+The unit of a :ref:`nwb-schema:sec-DecompositionSeries` should be appropriate for its metric. When the metric is "phase",
+the unit should be "radians" or "degrees". When the metric is "amplitude", the unit should match the unit of the source
+signal (e.g., "volts" if the source is an ElectricalSeries). The PyNWB default of "no unit" should be replaced with
+the actual unit.
+
+Check function: :py:meth:`~nwbinspector.checks._misc.check_decomposition_series_unit`
+
+
+.. _best_practice_decomposition_series_source:
+
+DecompositionSeries Source TimeSeries
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A :ref:`nwb-schema:sec-DecompositionSeries` should link to the source TimeSeries that was decomposed via the
+``source_timeseries`` field. This provides provenance about the origin of the decomposed signal and enables
+validation of the unit field.
+
+Check function: :py:meth:`~nwbinspector.checks._misc.check_decomposition_series_source_timeseries`

src/nwbinspector/checks/__init__.py

@@ -40,6 +40,10 @@
     check_order_of_images_len,
     check_order_of_images_unique,
 )
+from ._misc import (
+    check_decomposition_series_source_timeseries,
+    check_decomposition_series_unit,
+)
 from ._nwb_containers import (
     check_empty_string_for_optional_attribute,
     check_large_dataset_compression,
@@ -194,4 +198,6 @@
     "check_electrodes_location_allen_ccf",
     "check_intracellular_electrode_location_allen_ccf",
     "check_units_table_has_spikes",
+    "check_decomposition_series_unit",
+    "check_decomposition_series_source_timeseries",
 ]

src/nwbinspector/checks/_misc.py

@@ -0,0 +1,70 @@
+"""Checks specific to neurodata types in the pynwb.misc module."""
+
+from typing import Optional
+
+from pynwb.misc import DecompositionSeries
+
+from .._registration import Importance, InspectorMessage, register_check
+
+
+@register_check(importance=Importance.BEST_PRACTICE_VIOLATION, neurodata_type=DecompositionSeries)
+def check_decomposition_series_unit(
+    decomposition_series: DecompositionSeries,
+) -> Optional[InspectorMessage]:
+    """
+    Check that DecompositionSeries has an appropriate unit for its metric.
+
+    Best Practice: :ref:`best_practice_decomposition_series_unit`
+    """
+    unit = decomposition_series.unit
+    metric = decomposition_series.metric
+
+    if unit in ("no unit", ""):
+        return InspectorMessage(
+            message=(
+                f"DecompositionSeries is missing a valid unit (current value: '{unit}'). "
+                f"Please specify the unit appropriate for the metric '{metric}' "
+                "(e.g., 'radians' or 'degrees' for phase, or the source signal unit for amplitude)."
+            )
+        )
+
+    if metric == "phase" and unit not in ("radians", "degrees"):
+        return InspectorMessage(
+            message=(
+                f"DecompositionSeries with metric 'phase' should have unit 'radians' or 'degrees', "
+                f"but has unit '{unit}'."
+            )
+        )
+
+    if metric == "amplitude" and decomposition_series.source_timeseries is not None:
+        source_unit = decomposition_series.source_timeseries.unit
+        if unit != source_unit:
+            return InspectorMessage(
+                message=(
+                    f"DecompositionSeries with metric 'amplitude' should have the same unit as the source "
+                    f"TimeSeries ('{source_unit}'), but has unit '{unit}'."
+                )
+            )
+
+    return None
+
+
+@register_check(importance=Importance.BEST_PRACTICE_SUGGESTION, neurodata_type=DecompositionSeries)
+def check_decomposition_series_source_timeseries(
+    decomposition_series: DecompositionSeries,
+) -> Optional[InspectorMessage]:
+    """
+    Check that DecompositionSeries has a source_timeseries linked.
+
+    Best Practice: :ref:`best_practice_decomposition_series_source`
+    """
+    if decomposition_series.source_timeseries is None:
+        return InspectorMessage(
+            message=(
+                "DecompositionSeries does not have a source_timeseries linked. "
+                "Setting source_timeseries provides provenance about which signal was decomposed "
+                "and enables validation of the unit field."
+            )
+        )
+
+    return None

tests/unit_tests/test_misc.py

@@ -0,0 +1,109 @@
+import warnings
+
+import numpy as np
+from pynwb.misc import DecompositionSeries
+from pynwb.testing.mock.ecephys import mock_ElectricalSeries
+
+from nwbinspector import Importance, InspectorMessage
+from nwbinspector.checks import (
+    check_decomposition_series_source_timeseries,
+    check_decomposition_series_unit,
+)
+
+
+def _make_decomposition_series(metric="phase", unit="radians", source_timeseries=None):
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        return DecompositionSeries(
+            name="test",
+            data=np.ones((10, 2, 3)),
+            metric=metric,
+            unit=unit,
+            rate=1.0,
+            source_timeseries=source_timeseries,
+        )
+
+
+def test_pass_decomposition_series_unit_phase_radians():
+    ds = _make_decomposition_series(metric="phase", unit="radians")
+    assert check_decomposition_series_unit(ds) is None
+
+
+def test_pass_decomposition_series_unit_phase_degrees():
+    ds = _make_decomposition_series(metric="phase", unit="degrees")
+    assert check_decomposition_series_unit(ds) is None
+
+
+def test_fail_decomposition_series_unit_phase_wrong():
+    ds = _make_decomposition_series(metric="phase", unit="volts")
+    result = check_decomposition_series_unit(ds)
+    assert result == InspectorMessage(
+        message=(
+            "DecompositionSeries with metric 'phase' should have unit 'radians' or 'degrees', " "but has unit 'volts'."
+        ),
+        importance=Importance.BEST_PRACTICE_VIOLATION,
+        check_function_name="check_decomposition_series_unit",
+        object_type="DecompositionSeries",
+        object_name="test",
+        location="/",
+    )
+
+
+def test_fail_decomposition_series_unit_no_unit():
+    ds = _make_decomposition_series(metric="amplitude", unit="no unit")
+    result = check_decomposition_series_unit(ds)
+    assert result is not None
+    assert "no unit" in result.message
+    assert "amplitude" in result.message
+
+
+def test_fail_decomposition_series_unit_empty_string():
+    ds = _make_decomposition_series(metric="power", unit="")
+    result = check_decomposition_series_unit(ds)
+    assert result is not None
+
+
+def test_pass_decomposition_series_unit_amplitude_matches_source():
+    source = mock_ElectricalSeries()
+    ds = _make_decomposition_series(metric="amplitude", unit="volts", source_timeseries=source)
+    assert check_decomposition_series_unit(ds) is None
+
+
+def test_fail_decomposition_series_unit_amplitude_mismatch():
+    source = mock_ElectricalSeries()
+    ds = _make_decomposition_series(metric="amplitude", unit="watts", source_timeseries=source)
+    result = check_decomposition_series_unit(ds)
+    assert result == InspectorMessage(
+        message=(
+            "DecompositionSeries with metric 'amplitude' should have the same unit as the source "
+            "TimeSeries ('volts'), but has unit 'watts'."
+        ),
+        importance=Importance.BEST_PRACTICE_VIOLATION,
+        check_function_name="check_decomposition_series_unit",
+        object_type="DecompositionSeries",
+        object_name="test",
+        location="/",
+    )
+
+
+def test_pass_decomposition_series_source_set():
+    source = mock_ElectricalSeries()
+    ds = _make_decomposition_series(source_timeseries=source)
+    assert check_decomposition_series_source_timeseries(ds) is None
+
+
+def test_fail_decomposition_series_source_not_set():
+    ds = _make_decomposition_series()
+    result = check_decomposition_series_source_timeseries(ds)
+    assert result == InspectorMessage(
+        message=(
+            "DecompositionSeries does not have a source_timeseries linked. "
+            "Setting source_timeseries provides provenance about which signal was decomposed "
+            "and enables validation of the unit field."
+        ),
+        importance=Importance.BEST_PRACTICE_SUGGESTION,
+        check_function_name="check_decomposition_series_source_timeseries",
+        object_type="DecompositionSeries",
+        object_name="test",
+        location="/",
+    )