Updates for neo and pi
@@ -81,6 +81,10 @@ class OpenEphysBinaryRecordingExtractor(NeoBaseRecordingExtractor):
----------
folder_path: str
The folder path to load the recordings from.
+ experiment_name: str, list, or None
+ If multiple experiments are available, this argument allows users to select one
+ or more experiments. If None, all experiements are loaded as blocks.
+ E.g. `experiment_names="experiment2"`, `experiment_names=["experiment1", "experiment2"]`
stream_id: str, optional
If there are several streams, specify the stream id you want to load.
stream_name: str, optional
@@ -96,17 +100,33 @@ class OpenEphysBinaryRecordingExtractor(NeoBaseRecordingExtractor):
name = "openephys"
has_default_locations = True
- def __init__(self, folder_path, stream_id=None, stream_name=None, block_index=None, all_annotations=False):
- neo_kwargs = self.map_to_neo_kwargs(folder_path)
+ def __init__(self, folder_path, experiment_names=None, stream_id=None, stream_name=None, block_index=None, all_annotations=False):
+ neo_kwargs = self.map_to_neo_kwargs(folder_path, experiment_names)
NeoBaseRecordingExtractor.__init__(self, stream_id=stream_id,
stream_name=stream_name,
block_index=block_index,
all_annotations=all_annotations,
**neo_kwargs)
+ # get streams to find correct probe
+ stream_names, stream_ids = self.get_streams(folder_path)
+ if stream_name is None and stream_id is None:
+ stream_name = stream_names[0]
+ elif stream_name is None:
+ stream_name = stream_names[stream_ids.index(stream_id)]
+
+ # do not load probe for NIDQ stream
+ if "NI-DAQmx" not in stream_name:
+ # find settings file
+ record_node = stream_name.split("#")[0]
+ if block_index is None:
+ settings_file = self.neo_reader.folder_structure[record_node]["experiments"][0]["settings_file"]
+ else:
+ settings_file = self.neo_reader.folder_structure[record_node]["experiments"][block_index]["settings_file"]
- probe = pi.read_openephys(folder_path, raise_error=False)
+ probe = pi.read_openephys(folder_path, settings_file=settings_file,
+ stream_name=stream_name, raise_error=False)
if probe is not None:
- self.set_probe(probe, in_place=True)
+ self = self.set_probe(probe, in_place=True)
probe_name = probe.annotations["probe_name"]
# load num_channels_per_adc depending on probe type
if "2.0" in probe_name:
@@ -115,13 +135,12 @@ class OpenEphysBinaryRecordingExtractor(NeoBaseRecordingExtractor):
num_channels_per_adc = 12
sample_shifts = get_neuropixels_sample_shifts(self.get_num_channels(), num_channels_per_adc)
self.set_property("inter_sample_shift", sample_shifts)
-
-
self._kwargs .update(dict(folder_path=str(folder_path)))
+
@classmethod
- def map_to_neo_kwargs(cls, folder_path):
- neo_kwargs = {'dirname': str(folder_path)}
+ def map_to_neo_kwargs(cls, folder_path, experiment_names=None):
+ neo_kwargs = {'dirname': str(folder_path), 'experiment_names': experiment_names}
return neo_kwargs
