/
ecephys.py
398 lines (341 loc) · 19.8 KB
/
ecephys.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
from collections import Iterable
from hdmf.utils import docval, getargs, popargs, call_docval_func
from hdmf.data_utils import DataChunkIterator, assertEqualShape
from . import register_class, CORE_NAMESPACE
from .base import TimeSeries, _default_resolution, _default_conversion
from .core import NWBContainer, NWBDataInterface, MultiContainerInterface, DynamicTableRegion
from .device import Device
@register_class('ElectrodeGroup', CORE_NAMESPACE)
class ElectrodeGroup(NWBContainer):
"""
"""
__nwbfields__ = ('name',
'description',
'location',
'device')
@docval({'name': 'name', 'type': str, 'doc': 'the name of this electrode'},
{'name': 'description', 'type': str, 'doc': 'description of this electrode group'},
{'name': 'location', 'type': str, 'doc': 'description of location of this electrode group'},
{'name': 'device', 'type': Device, 'doc': 'the device that was used to record from this electrode group'},
{'name': 'parent', 'type': 'NWBContainer',
'doc': 'The parent NWBContainer for this NWBContainer', 'default': None})
def __init__(self, **kwargs):
call_docval_func(super(ElectrodeGroup, self).__init__, kwargs)
description, location, device = popargs("description", "location", "device", kwargs)
self.description = description
self.location = location
self.device = device
_et_docval = [
{'name': 'id', 'type': int, 'doc': 'a unique identifier for the electrode'},
{'name': 'x', 'type': float, 'doc': 'the x coordinate of the position'},
{'name': 'y', 'type': float, 'doc': 'the y coordinate of the position'},
{'name': 'z', 'type': float, 'doc': 'the z coordinate of the position'},
{'name': 'imp', 'type': float, 'doc': 'the impedance of the electrode'},
{'name': 'location', 'type': str, 'doc': 'the location of electrode within the subject e.g. brain region'},
{'name': 'filtering', 'type': str, 'doc': 'description of hardware filtering'},
{'name': 'description', 'type': str, 'doc': 'a brief description of what this electrode is'},
{'name': 'group', 'type': ElectrodeGroup, 'doc': 'the ElectrodeGroup object to add to this NWBFile'},
{'name': 'group_name', 'type': str, 'doc': 'the ElectrodeGroup object to add to this NWBFile', 'default': None}
]
@register_class('ElectricalSeries', CORE_NAMESPACE)
class ElectricalSeries(TimeSeries):
"""
Stores acquired voltage data from extracellular recordings. The data field of an ElectricalSeries
is an int or float array storing data in Volts. TimeSeries::data array structure: [num times] [num
channels] (or [num_times] for single electrode).
"""
__nwbfields__ = ({'name': 'electrodes', 'required_name': 'electrodes',
'doc': 'the electrodes that generated this electrical series', 'child': True},)
__help = "Stores acquired voltage data from extracellular recordings."
@docval({'name': 'name', 'type': str, 'doc': 'The name of this TimeSeries dataset'},
{'name': 'data', 'type': ('array_data', 'data', TimeSeries),
'shape': ((None, ), (None, None), (None, None, None)),
'doc': 'The data this TimeSeries dataset stores. Can also store binary data e.g. image frames'},
{'name': 'electrodes', 'type': DynamicTableRegion,
'doc': 'the table region corresponding to the electrodes from which this series was recorded'},
{'name': 'resolution', 'type': float,
'doc': 'The smallest meaningful difference (in specified unit) between values in data',
'default': _default_resolution},
{'name': 'conversion', 'type': float,
'doc': 'Scalar to multiply each element by to convert to volts', 'default': _default_conversion},
{'name': 'timestamps', 'type': ('array_data', 'data', TimeSeries),
'doc': 'Timestamps for samples stored in data', 'default': None},
{'name': 'starting_time', 'type': float, 'doc': 'The timestamp of the first sample', 'default': None},
{'name': 'rate', 'type': float, 'doc': 'Sampling rate in Hz', 'default': None},
{'name': 'comments', 'type': str,
'doc': 'Human-readable comments about this TimeSeries dataset', 'default': 'no comments'},
{'name': 'description', 'type': str,
'doc': 'Description of this TimeSeries dataset', 'default': 'no description'},
{'name': 'control', 'type': Iterable,
'doc': 'Numerical labels that apply to each element in data', 'default': None},
{'name': 'control_description', 'type': Iterable,
'doc': 'Description of each control value', 'default': None},
{'name': 'parent', 'type': 'NWBContainer',
'doc': 'The parent NWBContainer for this NWBContainer', 'default': None})
def __init__(self, **kwargs):
name, electrodes, data = popargs('name', 'electrodes', 'data', kwargs)
super(ElectricalSeries, self).__init__(name, data, 'volt', **kwargs)
self.electrodes = electrodes
@register_class('SpikeEventSeries', CORE_NAMESPACE)
class SpikeEventSeries(ElectricalSeries):
"""
Stores "snapshots" of spike events (i.e., threshold crossings) in data. This may also be raw data,
as reported by ephys hardware. If so, the TimeSeries::description field should describing how
events were detected. All SpikeEventSeries should reside in a module (under EventWaveform
interface) even if the spikes were reported and stored by hardware. All events span the same
recording channels and store snapshots of equal duration. TimeSeries::data array structure:
[num events] [num channels] [num samples] (or [num events] [num samples] for single
electrode).
"""
__nwbfields__ = ()
__help = "Snapshots of spike events from data."
@docval({'name': 'name', 'type': str, 'doc': 'The name of this TimeSeries dataset'},
{'name': 'data', 'type': ('array_data', 'data', TimeSeries),
'doc': 'The data this TimeSeries dataset stores. Can also store binary data e.g. image frames'},
{'name': 'timestamps', 'type': ('array_data', 'data', TimeSeries),
'doc': 'Timestamps for samples stored in data'},
{'name': 'electrodes', 'type': DynamicTableRegion,
'doc': 'the table region corresponding to the electrodes from which this series was recorded'},
{'name': 'resolution', 'type': float,
'doc': 'The smallest meaningful difference (in specified unit) between values in data',
'default': _default_resolution},
{'name': 'conversion', 'type': float,
'doc': 'Scalar to multiply each element by to convert to volts', 'default': _default_conversion},
{'name': 'comments', 'type': str,
'doc': 'Human-readable comments about this TimeSeries dataset', 'default': 'no comments'},
{'name': 'description', 'type': str,
'doc': 'Description of this TimeSeries dataset', 'default': 'no description'},
{'name': 'control', 'type': Iterable,
'doc': 'Numerical labels that apply to each element in data', 'default': None},
{'name': 'control_description', 'type': Iterable,
'doc': 'Description of each control value', 'default': None},
{'name': 'parent', 'type': 'NWBContainer',
'doc': 'The parent NWBContainer for this NWBContainer', 'default': None})
def __init__(self, **kwargs):
name, data, electrodes = popargs('name', 'data', 'electrodes', kwargs)
timestamps = getargs('timestamps', kwargs)
if not (isinstance(data, TimeSeries) and isinstance(timestamps, TimeSeries)):
if not (isinstance(data, DataChunkIterator) and isinstance(timestamps, DataChunkIterator)):
if len(data) != len(timestamps):
raise Exception('Must provide the same number of timestamps and spike events')
else:
# TODO: add check when we have DataChunkIterators
pass
super(SpikeEventSeries, self).__init__(name, data, electrodes, **kwargs)
@register_class('EventDetection', CORE_NAMESPACE)
class EventDetection(NWBDataInterface):
"""
Detected spike events from voltage trace(s).
"""
__nwbfields__ = ('detection_method',
'source_electricalseries',
'source_idx',
'times')
__help = ("Description of how events were detected, such as voltage "
"threshold, or dV/dT threshold, as well as relevant values.")
@docval({'name': 'detection_method', 'type': str,
'doc': 'Description of how events were detected, such as voltage threshold, or dV/dT threshold, \
as well as relevant values.'},
{'name': 'source_electricalseries', 'type': ElectricalSeries, 'doc': 'The source electrophysiology data'},
{'name': 'source_idx', 'type': ('array_data', 'data'),
'doc': 'Indices (zero-based) into source ElectricalSeries::data array corresponding \
to time of event. Module description should define what is meant by time of event \
(e.g., .25msec before action potential peak, zero-crossing time, etc). \
The index points to each event from the raw data'},
{'name': 'times', 'type': ('array_data', 'data'), 'doc': 'Timestamps of events, in Seconds'},
{'name': 'name', 'type': str, 'doc': 'the name of this container', 'default': 'EventDetection'})
def __init__(self, **kwargs):
detection_method, source_electricalseries, source_idx, times = popargs(
'detection_method', 'source_electricalseries', 'source_idx', 'times', kwargs)
super(EventDetection, self).__init__(**kwargs)
self.detection_method = detection_method
# do not set parent, since this is a link
self.source_electricalseries = source_electricalseries
self.source_idx = source_idx
self.times = times
self.unit = 'Seconds'
@register_class('EventWaveform', CORE_NAMESPACE)
class EventWaveform(MultiContainerInterface):
"""
Spike data for spike events detected in raw data
stored in this NWBFile, or events detect at acquisition
"""
__clsconf__ = {
'attr': 'spike_event_series',
'type': SpikeEventSeries,
'add': 'add_spike_event_series',
'get': 'get_spike_event_series',
'create': 'create_spike_event_series'
}
__help = "Waveform of detected extracellularly recorded spike events"
@register_class('Clustering', CORE_NAMESPACE)
class Clustering(NWBDataInterface):
"""
DEPRECATED in favor of :py:meth:`~pynwb.misc.Units`.
Specifies cluster event times and cluster metric for maximum ratio of
waveform peak to RMS on any channel in cluster.
"""
__nwbfields__ = (
'description',
'num',
'peak_over_rms',
'times'
)
__help = ("[DEPRECATED] Clustered spike data, whether from automatic clustering "
"tools (eg, klustakwik) or as a result of manual sorting.")
@docval({'name': 'description', 'type': str,
'doc': 'Description of clusters or clustering, (e.g. cluster 0 is noise, \
clusters curated using Klusters, etc).'},
{'name': 'num', 'type': ('array_data', 'data'), 'doc': 'Cluster number of each event.', 'shape': (None, )},
{'name': 'peak_over_rms', 'type': Iterable, 'shape': (None, ),
'doc': 'Maximum ratio of waveform peak to RMS on any channel in the cluster\
(provides a basic clustering metric).'},
{'name': 'times', 'type': ('array_data', 'data'), 'doc': 'Times of clustered events, in seconds.',
'shape': (None,)},
{'name': 'name', 'type': str, 'doc': 'the name of this container', 'default': 'Clustering'})
def __init__(self, **kwargs):
import warnings
warnings.warn("use pynwb.misc.Units or NWBFile.units instead", DeprecationWarning)
description, num, peak_over_rms, times = popargs(
'description', 'num', 'peak_over_rms', 'times', kwargs)
super(Clustering, self).__init__(**kwargs)
self.description = description
self.num = num
self.peak_over_rms = list(peak_over_rms)
self.times = times
@register_class('ClusterWaveforms', CORE_NAMESPACE)
class ClusterWaveforms(NWBDataInterface):
"""
DEPRECATED. `ClusterWaveforms` was deprecated in Oct 27, 2018 and will be removed in a future release.
Please use the `Units` table to store waveform mean and standard deviation
e.g. `NWBFile.units.add_unit(..., waveform_mean=..., waveform_sd=...)`
Describe cluster waveforms by mean and standard deviation for at each sample.
"""
__nwbfields__ = ('clustering_interface',
'waveform_filtering',
'waveform_mean',
'waveform_sd')
__help = ("[DEPRECATED] Mean waveform shape of clusters. Waveforms should be "
"high-pass filtered (ie, not the same bandpass filter "
"used waveform analysis and clustering)")
@docval({'name': 'clustering_interface', 'type': Clustering,
'doc': 'the clustered spike data used as input for computing waveforms'},
{'name': 'waveform_filtering', 'type': str,
'doc': 'filter applied to data before calculating mean and standard deviation'},
{'name': 'waveform_mean', 'type': Iterable, 'shape': (None, None),
'doc': 'the mean waveform for each cluster'},
{'name': 'waveform_sd', 'type': Iterable, 'shape': (None, None),
'doc': 'the standard deviations of waveforms for each cluster'},
{'name': 'name', 'type': str, 'doc': 'the name of this container', 'default': 'ClusterWaveforms'})
def __init__(self, **kwargs):
import warnings
warnings.warn("use pynwb.misc.Units or NWBFile.units instead", DeprecationWarning)
clustering_interface, waveform_filtering, waveform_mean, waveform_sd = popargs(
'clustering_interface', 'waveform_filtering', 'waveform_mean', 'waveform_sd', kwargs)
super(ClusterWaveforms, self).__init__(**kwargs)
self.clustering_interface = clustering_interface
self.waveform_filtering = waveform_filtering
self.waveform_mean = waveform_mean
self.waveform_sd = waveform_sd
@register_class('LFP', CORE_NAMESPACE)
class LFP(MultiContainerInterface):
"""
LFP data from one or more channels. The electrode map in each published ElectricalSeries will
identify which channels are providing LFP data. Filter properties should be noted in the
ElectricalSeries description or comments field.
"""
__clsconf__ = [
{'attr': 'electrical_series',
'type': ElectricalSeries,
'add': 'add_electrical_series',
'get': 'get_electrical_series',
'create': 'create_electrical_series'}]
__help = ("LFP data from one or more channels. Filter properties "
"should be noted in the ElectricalSeries")
@register_class('FilteredEphys', CORE_NAMESPACE)
class FilteredEphys(MultiContainerInterface):
"""
Ephys data from one or more channels that has been subjected to filtering. Examples of filtered
data include Theta and Gamma (LFP has its own interface). FilteredEphys modules publish an
ElectricalSeries for each filtered channel or set of channels. The name of each ElectricalSeries is
arbitrary but should be informative. The source of the filtered data, whether this is from analysis
of another time series or as acquired by hardware, should be noted in each's
TimeSeries::description field. There is no assumed 1::1 correspondence between filtered ephys
signals and electrodes, as a single signal can apply to many nearby electrodes, and one
electrode may have different filtered (e.g., theta and/or gamma) signals represented.
"""
__help = ("Ephys data from one or more channels that is subjected to filtering, such as "
"for gamma or theta oscillations (LFP has its own interface). Filter properties should "
"be noted in the ElectricalSeries")
__clsconf__ = {
'attr': 'electrical_series',
'type': ElectricalSeries,
'add': 'add_electrical_series',
'get': 'get_electrical_series',
'create': 'create_electrical_series'
}
@register_class('FeatureExtraction', CORE_NAMESPACE)
class FeatureExtraction(NWBDataInterface):
"""
Features, such as PC1 and PC2, that are extracted from signals stored in a SpikeEvent
TimeSeries or other source.
"""
__nwbfields__ = ('description',
{'name': 'electrodes', 'child': True},
'times',
'features')
__help = "Container for salient features of detected events"
@docval({'name': 'electrodes', 'type': DynamicTableRegion,
'doc': 'the table region corresponding to the electrodes from which this series was recorded'},
{'name': 'description', 'type': ('array_data', 'data'),
'doc': 'A description for each feature extracted', 'shape': (None, )},
{'name': 'times', 'type': ('array_data', 'data'), 'shape': (None, ),
'doc': 'The times of events that features correspond to'},
{'name': 'features', 'type': ('array_data', 'data'), 'shape': (None, None, None),
'doc': 'Features for each channel'},
{'name': 'name', 'type': str, 'doc': 'the name of this container', 'default': 'FeatureExtraction'})
def __init__(self, **kwargs):
# get the inputs
electrodes, description, times, features = popargs(
'electrodes', 'description', 'times', 'features', kwargs)
# Validate the shape of the inputs
# Validate event times compared to features
shape_validators = []
shape_validators.append(assertEqualShape(data1=features,
data2=times,
axes1=0,
axes2=0,
name1='feature_shape',
name2='times',
ignore_undetermined=True))
# Validate electrodes compared to features
shape_validators.append(assertEqualShape(data1=features,
data2=electrodes,
axes1=1,
axes2=0,
name1='feature_shape',
name2='electrodes',
ignore_undetermined=True))
# Valided description compared to features
shape_validators.append(assertEqualShape(data1=features,
data2=description,
axes1=2,
axes2=0,
name1='feature_shape',
name2='description',
ignore_undetermined=True))
# Raise an error if any of the shapes do not match
raise_error = False
error_msg = ""
for sv in shape_validators:
raise_error |= not sv.result
if not sv.result:
error_msg += sv.message + "\n"
if raise_error:
raise ValueError(error_msg)
# Initialize the object
super(FeatureExtraction, self).__init__(**kwargs)
self.electrodes = electrodes
self.description = description
self.times = list(times)
self.features = features