From 7e40410a597da72abb0f92d48d620b139eb0603a Mon Sep 17 00:00:00 2001 From: Ben Dichter Date: Wed, 22 Jul 2020 14:07:17 -0400 Subject: [PATCH] add option to include source_channels for DecompositionSeries (#1258) * add option to include source_channels for DecompositionSeries and include round trip test * Use latest schema * Add unit test --- src/pynwb/misc.py | 16 ++++++--- tests/integration/hdf5/test_misc.py | 55 ++++++++++++++++++++++++++++- tests/unit/test_misc.py | 33 +++++++++++++++-- 3 files changed, 96 insertions(+), 8 deletions(-) diff --git a/src/pynwb/misc.py b/src/pynwb/misc.py index 18222bbd8..6124b18e3 100644 --- a/src/pynwb/misc.py +++ b/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) @@ -251,6 +251,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}) @@ -267,15 +268,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") diff --git a/tests/integration/hdf5/test_misc.py b/tests/integration/hdf5/test_misc.py index 396985993..abe9e9ef1 100644 --- a/tests/integration/hdf5/test_misc.py +++ b/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): @@ -134,3 +137,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) diff --git a/tests/unit/test_misc.py b/tests/unit/test_misc.py index 3417af97e..4412063ce 100644 --- a/tests/unit/test_misc.py +++ b/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):