FrequencyBandsTable

CHANGELOG.md

@@ -15,6 +15,7 @@
 - Added dictionary-like operations directly on `ProcessingModule` objects (e.g., `len(processing_module)`). @bendichter [#2020](https://github.com/NeurodataWithoutBorders/pynwb/pull/2020)
 - When an external file is detected when initializing an ImageSeries and no format is provided, automatically set format to "external" instead of raising an error. @stephprince [#2060](https://github.com/NeurodataWithoutBorders/pynwb/pull/2060)
 - Added mask_type option to `mock_PlaneSegmentation`. @pauladkisson [#2067](https://github.com/NeurodataWithoutBorders/pynwb/pull/2067)
+- Formally defined bands within `DecompositionSeries` as the neurodatatype `FrequencyBandsTable`. @mavaylon1 [#2063](https://github.com/NeurodataWithoutBorders/pynwb/pull/2063)
 
 ### Bug fixes
 - Fixed `add_data_interface` functionality that was mistakenly removed in PyNWB 3.0. @stephprince [#2052](https://github.com/NeurodataWithoutBorders/pynwb/pull/2052)

src/pynwb/io/misc.py

@@ -1,7 +1,23 @@
 from hdmf.common.io.table import DynamicTableMap
+from .base import TimeSeriesMap
 
 from .. import register_map
-from pynwb.misc import Units
+from pynwb.misc import Units, DecompositionSeries
+
+
+@register_map(DecompositionSeries)
+class DecompositionSeriesMap(TimeSeriesMap):
+    def __init__(self, spec):
+        super().__init__(spec)
+
+    @TimeSeriesMap.constructor_arg('bands')
+    def bands(self, builder, manager):
+        if builder.groups['bands'].attributes['neurodata_type'] != 'FrequencyBandsTable':
+            builder.groups['bands'].attributes['neurodata_type'] = 'FrequencyBandsTable'
+            builder.groups['bands'].attributes['namespace'] = 'core'
+        manager.clear_cache()
+        new_container =  manager.construct(builder.groups['bands'])
+        return new_container
 
 
 @register_map(Units)

src/pynwb/misc.py

@@ -9,7 +9,7 @@
 from . import register_class, CORE_NAMESPACE
 from .base import TimeSeries
 from .ecephys import ElectrodeGroup
-from hdmf.common import DynamicTable, DynamicTableRegion
+from hdmf.common import DynamicTable, DynamicTableRegion, VectorData
 
 __all__ = [
     'AnnotationSeries',
@@ -253,13 +253,50 @@ def get_unit_obs_intervals(self, **kwargs):
         index = getargs('index', kwargs)
         return np.asarray(self['obs_intervals'][index])
 
+@register_class('FrequencyBandsTable', CORE_NAMESPACE)
+class FrequencyBandsTable(DynamicTable):
+    """
+    Table for describing the bands that DecompositionSeries was generated from.
+    """
+    __columns__ = (
+        {'name': 'band_name', 'description': 'Name of the band, e.g. theta.', 'required': True},
+        {'name': 'band_limits', 'description': 'Low and high limit of each band in Hz.', 'required': True},
+        {'name': 'band_mean', 'description': 'The mean Gaussian filters, in Hz.', 'required': False},
+        {'name': 'band_stdev', 'description': 'The standard deviation Gaussian filters, in Hz.', 'required': False})
+
+    @docval({'name': 'band_mean', 'type': VectorData, 'doc': 'The mean Gaussian filters, in Hz.', 'default': None},
+            {'name': 'band_stdev', 'type': VectorData, 'doc': 'The standard deviation Gaussian filters, in Hz.',
+             'default': None},
+            *get_docval(DynamicTable.__init__, 'id', 'columns', 'colnames'))
+    def __init__(self, **kwargs):
+        kwargs['name'] = 'bands'
+        kwargs['description'] = 'Table for describing the bands that DecompositionSeries was generated from.'
+
+        # optional fields
+        keys_to_set = ('band_mean', 'band_stdev')
+        args_to_set = popargs_to_dict(keys_to_set, kwargs)
+        for key, val in args_to_set.items():
+            setattr(self, key, val)
+
+        super().__init__(**kwargs)
+
+    @docval({'name': 'band_name', 'type': str, 'doc': 'Name of the band, e.g. theta.'},
+            {'name': 'band_limits', 'type': ('array_data', 'data'), 'shape': [None, 2],
+             'doc': 'Low and high limit of each band in Hz.'},
+            {'name': 'band_mean', 'type': float, 'doc': 'The mean Gaussian filters, in Hz.',
+             'default': None},
+            {'name': 'band_stdev', 'type': float, 'doc': 'The standard deviation Gaussian filters, in Hz.',
+             'default': None},
+            allow_extra=True)
+    def add_band(self, **kwargs):
+        super().add_row(**kwargs)
+
 
 @register_class('DecompositionSeries', CORE_NAMESPACE)
 class DecompositionSeries(TimeSeries):
     """
     Stores product of spectral analysis
     """
-
     __nwbfields__ = ('metric',
                      {'name': 'source_timeseries', 'child': False, 'doc': 'the input TimeSeries from this analysis'},
                      {'name': 'source_channels', 'child': True, 'doc': 'the channels that provided the source data'},
@@ -278,7 +315,7 @@ class DecompositionSeries(TimeSeries):
             {'name': 'metric', 'type': str,  # required
              'doc': "metric of analysis. recommended - 'phase', 'amplitude', 'power'"},
             {'name': 'unit', 'type': str, 'doc': 'SI unit of measurement', 'default': 'no unit'},
-            {'name': 'bands', 'type': DynamicTable,
+            {'name': 'bands', 'type': FrequencyBandsTable,
              'doc': 'a table for describing the frequency bands that the signal was decomposed into', 'default': None},
             {'name': 'source_timeseries', 'type': TimeSeries,
              'doc': 'the input TimeSeries from this analysis', 'default': None},
@@ -302,16 +339,9 @@ def __init__(self, **kwargs):
                           "corresponding source_channels. (Optional)")
         self.metric = metric
         if bands is None:
-            bands = DynamicTable(
-                name="bands",
-                description="data about the frequency bands that the signal was decomposed into"
-            )
+            bands = FrequencyBandsTable()
         self.bands = bands
 
-    def __check_column(self, name, desc):
-        if name not in self.bands.colnames:
-            self.bands.add_column(name, desc)
-
     @docval({'name': 'band_name', 'type': str, 'doc': 'the name of the frequency band',
              'default': None},
             {'name': 'band_limits', 'type': ('array_data', 'data'), 'default': None,
@@ -327,14 +357,8 @@ def add_band(self, **kwargs):
         """
         band_name, band_limits, band_mean, band_stdev = getargs('band_name', 'band_limits', 'band_mean', 'band_stdev',
                                                                 kwargs)
-        if band_name is not None:
-            self.__check_column('band_name', "the name of the frequency band (recommended: 'alpha', 'beta', 'gamma', "
-                                             "'delta', 'high gamma'")
-        if band_name is not None:
-            self.__check_column('band_limits', 'low and high frequencies of bandpass filter in Hz')
-        if band_mean is not None:
-            self.__check_column('band_mean', 'the mean of Gaussian filters in Hz')
-        if band_stdev is not None:
-            self.__check_column('band_stdev', 'the standard deviation of Gaussian filters in Hz')
-
-        self.bands.add_row({k: v for k, v in kwargs.items() if v is not None})
+
+        self.bands.add_band(band_name=band_name,
+                            band_limits=band_limits,
+                            band_mean=band_mean,
+                            band_stdev=band_stdev)

tests/integration/hdf5/test_misc.py

@@ -1,8 +1,8 @@
 import numpy as np
 
-from hdmf.common import DynamicTable, VectorData, DynamicTableRegion
-from pynwb import TimeSeries
-from pynwb.misc import Units, DecompositionSeries
+from hdmf.common import VectorData, DynamicTableRegion
+from pynwb import TimeSeries, NWBHDF5IO
+from pynwb.misc import Units, DecompositionSeries, FrequencyBandsTable
 from pynwb.testing import NWBH5IOMixin, AcquisitionH5IOMixin, TestCase
 from pynwb.ecephys import ElectrodeGroup, ElectrodesTable
 from pynwb.device import Device
@@ -104,6 +104,18 @@ def test_to_dataframe(self):
         units.to_dataframe()
 
 
+class TestBackCompatDecompositionSeries(TestCase):
+    def test_read_nwbfile(self):
+        """
+        Test that reads an NWBFile with a DecompositionSeries that has a DynamicTable for bands.
+        """
+        io = NWBHDF5IO("tests/back_compat/3.0.0_DecompositionSeries.nwb", mode="r")
+        nwbfile = io.read()
+        # Read FrequencyBandsTable to ensure it uses the schema type
+        bands = nwbfile.processing['test_mod']['LFPSpectralAnalysis'].bands
+        self.assertTrue(isinstance(bands, FrequencyBandsTable))
+
+
 class TestDecompositionSeriesIO(NWBH5IOMixin, TestCase):
 
     def setUpContainer(self):
@@ -115,9 +127,7 @@ def setUpContainer(self):
             unit='flibs',
             timestamps=np.ones((3,)),
         )
-        bands = DynamicTable(
-            name='bands',
-            description='band info for LFPSpectralAnalysis',
+        bands = FrequencyBandsTable(
             columns=[
                 VectorData(name='band_name', description='name of bands', data=['alpha', 'beta', 'gamma']),
                 VectorData(name='band_limits', description='low and high cutoffs in Hz', data=np.ones((3, 2))),
@@ -179,9 +189,7 @@ def setUpContainer(self):
         )
         data = np.random.randn(100, 2, 30)
         timestamps = np.arange(100)/100
-        bands = DynamicTable(
-            name='bands',
-            description='band info for LFPSpectralAnalysis',
+        bands = FrequencyBandsTable(
             columns=[
                 VectorData(name='band_name', description='name of bands', data=['alpha', 'beta', 'gamma']),
                 VectorData(name='band_limits', description='low and high cutoffs in Hz', data=np.ones((3, 2))),

tests/unit/test_misc.py

@@ -1,8 +1,10 @@
 import numpy as np
 
-from hdmf.common import DynamicTable, VectorData, DynamicTableRegion
+from hdmf.common import VectorData, DynamicTableRegion
 
-from pynwb.misc import AnnotationSeries, AbstractFeatureSeries, IntervalSeries, Units, DecompositionSeries
+from pynwb.misc import (
+    AnnotationSeries, AbstractFeatureSeries, IntervalSeries, Units, DecompositionSeries, FrequencyBandsTable
+)
 from pynwb.file import TimeSeries
 from pynwb.device import Device
 from pynwb.ecephys import ElectrodeGroup, ElectrodesTable
@@ -29,21 +31,51 @@ def test_init(self):
         aFS.add_features(2.0, [1.])
 
 
+class FrequencyBandsTableConstructor(TestCase):
+    def setUp(self):
+        self.bands = FrequencyBandsTable(
+            columns=[
+                VectorData(name='band_name', description='name of bands', data=['alpha', 'beta', 'gamma']),
+                VectorData(name='band_limits', description='low and high cutoffs in Hz', data=np.ones((3, 2))),
+                VectorData(name='band_mean', description='mean gaussian filters in Hz', data=np.ones((3,))),
+                VectorData(
+                    name='band_stdev',
+                    description='standard deviation of gaussian filters in Hz',
+                    data=np.ones((3,))
+                ),
+            ],
+        )
+
+    def test_init(self):
+        self.assertEqual(self.bands['band_name'].data, ['alpha', 'beta', 'gamma'])
+        np.testing.assert_equal(self.bands['band_limits'].data, np.ones((3, 2)))
+        np.testing.assert_equal(self.bands['band_mean'].data, np.ones((3,)))
+        np.testing.assert_equal(self.bands['band_stdev'].data, np.ones((3,)))
+
+    def test_add_row(self):
+        self.bands.add_band(band_name='band_name', band_limits=np.array([[1., 1.]]), band_mean=1., band_stdev=1.)
+        np.testing.assert_equal(self.bands['band_limits'].data, np.ones((4, 2)))
+        np.testing.assert_equal(self.bands['band_mean'].data, np.ones((4,)))
+        np.testing.assert_equal(self.bands['band_stdev'].data, np.ones((4,)))
+
+
 class DecompositionSeriesConstructor(TestCase):
     def test_init(self):
         timeseries = TimeSeries(name='dummy timeseries', description='desc',
                                 data=np.ones((3, 3)), unit='Volts',
                                 timestamps=[1., 2., 3.])
-        bands = DynamicTable(name='bands', description='band info for LFPSpectralAnalysis', columns=[
-            VectorData(name='band_name', description='name of bands', data=['alpha', 'beta', 'gamma']),
-            VectorData(name='band_limits', description='low and high cutoffs in Hz', data=np.ones((3, 2))),
-            VectorData(name='band_mean', description='mean gaussian filters in Hz', data=np.ones((3,))),
-            VectorData(
-                name='band_stdev',
-                description='standard deviation of gaussian filters in Hz',
-                data=np.ones((3,))
-            ),
-        ])
+        bands = FrequencyBandsTable(
+            columns=[
+                VectorData(name='band_name', description='name of bands', data=['alpha', 'beta', 'gamma']),
+                VectorData(name='band_limits', description='low and high cutoffs in Hz', data=np.ones((3, 2))),
+                VectorData(name='band_mean', description='mean gaussian filters in Hz', data=np.ones((3,))),
+                VectorData(
+                    name='band_stdev',
+                    description='standard deviation of gaussian filters in Hz',
+                    data=np.ones((3,))
+                ),
+            ],
+        )
         spec_anal = DecompositionSeries(name='LFPSpectralAnalysis',
                                         description='my description',
                                         data=np.ones((3, 3, 3)),
@@ -74,16 +106,15 @@ def test_init_delayed_bands(self):
                                         source_timeseries=timeseries,
                                         metric='amplitude')
         for band_name in ['alpha', 'beta', 'gamma']:
-            spec_anal.add_band(band_name=band_name, band_limits=(1., 1.), band_mean=1., band_stdev=1.)
-
+            spec_anal.add_band(band_name=band_name, band_limits=np.array([[1., 1.]]), band_mean=1., band_stdev=1.)
         self.assertEqual(spec_anal.name, 'LFPSpectralAnalysis')
         self.assertEqual(spec_anal.description, 'my description')
         np.testing.assert_equal(spec_anal.data, np.ones((3, 3, 3)))
         np.testing.assert_equal(spec_anal.timestamps, [1., 2., 3.])
-        self.assertEqual(spec_anal.bands['band_name'].data, ['alpha', 'beta', 'gamma'])
-        np.testing.assert_equal(spec_anal.bands['band_limits'].data, np.ones((3, 2)))
         self.assertEqual(spec_anal.source_timeseries, timeseries)
         self.assertEqual(spec_anal.metric, 'amplitude')
+        self.assertEqual(spec_anal.bands['band_name'].data, ['alpha', 'beta', 'gamma'])
+        np.testing.assert_equal(spec_anal.bands['band_limits'].data, [np.array([[1., 1.]]) for _ in range(3)])
 
     @staticmethod
     def make_electrode_table(self):