Source code for pynwb.io.base

from hdmf.build import LinkBuilder

from .core import NWBContainerMapper
from .. import register_map
from ..base import TimeSeries, ProcessingModule


[docs] @register_map(ProcessingModule) class ModuleMap(NWBContainerMapper): def __init__(self, spec): super().__init__(spec) containers_spec = self.spec.get_neurodata_type('NWBDataInterface') table_spec = self.spec.get_neurodata_type('DynamicTable') self.map_spec('data_interfaces', containers_spec) self.map_spec('data_interfaces', table_spec)
[docs] @register_map(TimeSeries) class TimeSeriesMap(NWBContainerMapper): def __init__(self, spec): super().__init__(spec) data_spec = self.spec.get_dataset('data') self.map_spec('unit', data_spec.get_attribute('unit')) self.map_spec('resolution', data_spec.get_attribute('resolution')) self.map_spec('conversion', data_spec.get_attribute('conversion')) self.map_spec('offset', data_spec.get_attribute('offset')) self.map_spec('continuity', data_spec.get_attribute('continuity')) timestamps_spec = self.spec.get_dataset('timestamps') self.map_spec('timestamps_unit', timestamps_spec.get_attribute('unit')) self.map_spec('interval', timestamps_spec.get_attribute('interval')) startingtime_spec = self.spec.get_dataset('starting_time') self.map_spec('starting_time_unit', startingtime_spec.get_attribute('unit')) self.map_spec('rate', startingtime_spec.get_attribute('rate')) # TODO map the sync group to something sync_spec = self.spec.get_group('sync') self.unmap(sync_spec)
[docs] @NWBContainerMapper.object_attr("timestamps") def timestamps_attr(self, container, manager): ret = container.fields.get('timestamps') if isinstance(ret, TimeSeries): owner = ret curr = owner.fields.get('timestamps') while isinstance(curr, TimeSeries): owner = curr curr = owner.fields.get('timestamps') ts_builder = manager.build(owner) tstamps_builder = ts_builder['timestamps'] ret = LinkBuilder(tstamps_builder, 'timestamps') return ret
[docs] @NWBContainerMapper.constructor_arg("timestamps") def timestamps_carg(self, builder, manager): tstamps_builder = builder.get('timestamps') if tstamps_builder is None: return None if isinstance(tstamps_builder, LinkBuilder): # if the parent of our target is available, return the parent object # Otherwise, return the dataset in the target builder # # NOTE: it is not available when data is externally linked # and we haven't explicitly read that file target = tstamps_builder.builder if target.parent is not None: return manager.construct(target.parent) else: return target.data else: return tstamps_builder.data
[docs] @NWBContainerMapper.object_attr("data") def data_attr(self, container, manager): ret = container.fields.get('data') if isinstance(ret, TimeSeries): owner = ret curr = owner.fields.get('data') while isinstance(curr, TimeSeries): owner = curr curr = owner.fields.get('data') data_builder = manager.build(owner) ret = LinkBuilder(data_builder['data'], 'data') return ret
[docs] @NWBContainerMapper.constructor_arg("data") def data_carg(self, builder, manager): # handle case where a TimeSeries is read and missing data timeseries_cls = manager.get_cls(builder) data_builder = builder.get('data') if data_builder is None: return timeseries_cls.DEFAULT_DATA if isinstance(data_builder, LinkBuilder): # NOTE: parent is not available when data is externally linked # and we haven't explicitly read that file target = data_builder.builder if target.parent is not None: return manager.construct(target.parent) else: return target.data return data_builder.data
[docs] @NWBContainerMapper.constructor_arg("unit") def unit_carg(self, builder, manager): # handle case where a TimeSeries is read and missing unit timeseries_cls = manager.get_cls(builder) data_builder = builder.get('data') if data_builder is None: return timeseries_cls.DEFAULT_UNIT if isinstance(data_builder, LinkBuilder): # NOTE: parent is not available when data is externally linked # and we haven't explicitly read that file target = data_builder.builder if target.parent is not None: data_builder = manager.construct(target.parent) else: data_builder = target if isinstance(data_builder, TimeSeries): # Data linked in another timeseries unit_value = data_builder.unit else: # DatasetBuilder owned by this timeseries unit_value = data_builder.attributes.get('unit') if unit_value is None: return timeseries_cls.DEFAULT_UNIT return unit_value