add option to include source_channels for DecompositionSeries

src/pynwb/misc.py

@@ -7,7 +7,7 @@
 
 from . import register_class, CORE_NAMESPACE
 from .base import TimeSeries
-from hdmf.common import DynamicTable
+from hdmf.common import DynamicTable, DynamicTableRegion
 
 
 @register_class('AnnotationSeries', CORE_NAMESPACE)
@@ -248,6 +248,7 @@ class DecompositionSeries(TimeSeries):
 
     __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'},
                      {'name': 'bands',
                       'doc': 'the bands that the signal is decomposed into', 'child': True})
 
@@ -262,15 +263,20 @@ class DecompositionSeries(TimeSeries):
              '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},
+            {'name': 'source_channels', 'type': DynamicTableRegion, 'doc': 'the channels that provided the source data',
+             'default': None},
             *get_docval(TimeSeries.__init__, 'resolution', 'conversion', 'timestamps', 'starting_time', 'rate',
                         'comments', 'control', 'control_description'))
     def __init__(self, **kwargs):
-        metric, source_timeseries, bands = popargs('metric', 'source_timeseries', 'bands', kwargs)
+        metric, source_timeseries, bands, source_channels = popargs('metric', 'source_timeseries', 'bands',
+                                                                    'source_channels', kwargs)
         super(DecompositionSeries, self).__init__(**kwargs)
         self.source_timeseries = source_timeseries
-        if self.source_timeseries is None:
-            warnings.warn("It is best practice to set `source_timeseries` if it is present to document "
-                          "where the DecompositionSeries was derived from. (Optional)")
+        self.source_channels = source_channels
+        if self.source_timeseries is None and self.source_channels is None:
+            warnings.warn("Neither source_timeseries nor source_channels is present in DecompositionSeries. It is "
+                          "recommended to indicate the source timeseries if it is present, or else to link to the "
+                          "corresponding source_channels. (Optional)")
         self.metric = metric
         if bands is None:
             bands = DynamicTable("bands", "data about the frequency bands that the signal was decomposed into")

tests/integration/hdf5/test_misc.py

@@ -1,9 +1,12 @@
 import numpy as np
 
-from hdmf.common import DynamicTable, VectorData
+from hdmf.common import DynamicTable, VectorData, DynamicTableRegion
 from pynwb import TimeSeries
 from pynwb.misc import Units, DecompositionSeries
 from pynwb.testing import NWBH5IOMixin, AcquisitionH5IOMixin, TestCase
+from pynwb.ecephys import ElectrodeGroup
+from pynwb.device import Device
+from pynwb.file import ElectrodeTable as get_electrode_table
 
 
 class TestUnitsIO(AcquisitionH5IOMixin, TestCase):
@@ -107,3 +110,53 @@ def addContainer(self, nwbfile):
     def getContainer(self, nwbfile):
         """ Return the test DecompositionSeries from the given NWBFile """
         return nwbfile.processing['test_mod']['LFPSpectralAnalysis']
+
+
+class TestDecompositionSeriesWithSourceChannelsIO(AcquisitionH5IOMixin, TestCase):
+
+    @staticmethod
+    def make_electrode_table(self):
+        """ Make an electrode table, electrode group, and device """
+        self.table = get_electrode_table()
+        self.dev1 = Device(name='dev1')
+        self.group = ElectrodeGroup(name='tetrode1',
+                                    description='tetrode description',
+                                    location='tetrode location',
+                                    device=self.dev1)
+        for i in range(4):
+            self.table.add_row(x=i, y=2.0, z=3.0, imp=-1.0, location='CA1', filtering='none', group=self.group,
+                               group_name='tetrode1')
+
+    def setUpContainer(self):
+        """ Return the test ElectricalSeries to read/write """
+        self.make_electrode_table(self)
+        region = DynamicTableRegion(name='source_channels',
+                                    data=[0, 2],
+                                    description='the first and third electrodes',
+                                    table=self.table)
+        data = np.random.randn(100, 2, 30)
+        timestamps = np.arange(100)/100
+        ds = DecompositionSeries(name='test_DS',
+                                 data=data,
+                                 source_channels=region,
+                                 timestamps=timestamps,
+                                 metric='amplitude')
+        return ds
+
+    def addContainer(self, nwbfile):
+        """ Add the test ElectricalSeries and related objects to the given NWBFile """
+        nwbfile.add_device(self.dev1)
+        nwbfile.add_electrode_group(self.group)
+        nwbfile.set_electrode_table(self.table)
+        nwbfile.add_acquisition(self.container)
+
+    def test_eg_ref(self):
+        """
+        Test that the electrode DynamicTableRegion references of the read ElectricalSeries have a group that
+        correctly resolves to ElectrodeGroup instances.
+        """
+        read = self.roundtripContainer()
+        row1 = read.source_channels[0]
+        row2 = read.source_channels[1]
+        self.assertIsInstance(row1.iloc[0]['group'], ElectrodeGroup)
+        self.assertIsInstance(row2.iloc[0]['group'], ElectrodeGroup)

tests/unit/test_misc.py

@@ -1,9 +1,9 @@
 import numpy as np
 
-from hdmf.common import DynamicTable, VectorData
+from hdmf.common import DynamicTable, VectorData, DynamicTableRegion
 
 from pynwb.misc import AnnotationSeries, AbstractFeatureSeries, IntervalSeries, Units, DecompositionSeries
-from pynwb.file import TimeSeries
+from pynwb.file import TimeSeries, ElectrodeTable as get_electrode_table
 from pynwb.device import Device
 from pynwb.ecephys import ElectrodeGroup
 from pynwb.testing import TestCase
@@ -74,6 +74,35 @@ def test_init_delayed_bands(self):
         self.assertEqual(spec_anal.source_timeseries, timeseries)
         self.assertEqual(spec_anal.metric, 'amplitude')
 
+    @staticmethod
+    def make_electrode_table(self):
+        """ Make an electrode table, electrode group, and device """
+        self.table = get_electrode_table()
+        self.dev1 = Device(name='dev1')
+        self.group = ElectrodeGroup(name='tetrode1',
+                                    description='tetrode description',
+                                    location='tetrode location',
+                                    device=self.dev1)
+        for i in range(4):
+            self.table.add_row(x=i, y=2.0, z=3.0, imp=-1.0, location='CA1', filtering='none', group=self.group,
+                               group_name='tetrode1')
+
+    def test_init_with_source_channels(self):
+        self.make_electrode_table(self)
+        region = DynamicTableRegion(name='source_channels',
+                                    data=[0, 2],
+                                    description='the first and third electrodes',
+                                    table=self.table)
+        data = np.random.randn(100, 2, 30)
+        timestamps = np.arange(100)/100
+        ds = DecompositionSeries(name='test_DS',
+                                 data=data,
+                                 source_channels=region,
+                                 timestamps=timestamps,
+                                 metric='amplitude')
+
+        self.assertIs(ds.source_channels, region)
+
 
 class IntervalSeriesConstructor(TestCase):
     def test_init(self):