Skip to content

Commit

Permalink
add static methods to find out available electrical series
Browse files Browse the repository at this point in the history
  • Loading branch information
h-mayorquin committed May 23, 2024
1 parent 0df2536 commit e6e5f35
Show file tree
Hide file tree
Showing 2 changed files with 75 additions and 12 deletions.
71 changes: 60 additions & 11 deletions src/spikeinterface/extractors/nwbextractors.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ def read_file_from_backend(
else:
import h5py

assert file is not None, "Unexpected, file is None"
assert file is not None, "Both file_path and file are None"
open_file = h5py.File(file, "r")

return open_file
Expand Down Expand Up @@ -125,11 +125,6 @@ def read_nwbfile(
nwbfile : NWBFile
The NWBFile object.
Raises
------
AssertionError
If ROS3 support is not enabled.
Notes
-----
This function can stream data from the "fsspec", and "rem" protocols.
Expand Down Expand Up @@ -277,14 +272,17 @@ def _retrieve_unit_table_pynwb(nwbfile: "NWBFile", unit_table_path: Optional[str


def _is_hdf5_file(filename_or_file):
# Source for magic numbers https://www.loc.gov/preservation/digital/formats/fdd/fdd000229.shtml
# We should find a better one though
if isinstance(filename_or_file, (str, Path)):
with open(filename_or_file, "rb") as f:
file_signature = f.read(8)
import h5py

filename = str(filename_or_file)
is_hdf5 = h5py.h5f.is_hdf5(filename.encode("utf-8"))
else:
file_signature = filename_or_file.read(8)
return file_signature == b"\x89HDF\r\n\x1a\n"
# Source of the magic number https://docs.hdfgroup.org/hdf5/develop/_f_m_t3.html
is_hdf5 = file_signature == b"\x89HDF\r\n\x1a\n"

return is_hdf5


def _get_backend_from_local_file(file_path: str | Path) -> str:
Expand Down Expand Up @@ -863,6 +861,57 @@ def _fetch_main_properties_backend(self):

return gains, offsets, locations, groups

@staticmethod
def fetch_available_electrical_series(
file_path: str | Path, stream_mode: Optional[Literal["fsspec", "remfile", "zarr"]] = None
) -> List[str]:
"""
Retrieves the paths to all ElectricalSeries objects within a neurodata file.
Parameters
----------
file_path : str | Path
The path to the neurodata file to be analyzed.
stream_mode : "fsspec" | "remfile" | "zarr" | None, optional
Determines the streaming mode for reading the file. Use this for optimized reading from
different sources, such as local disk or remote servers.
Returns
-------
list of str
A list of paths to all ElectricalSeries objects found in the file.
Notes
-----
The paths are returned as strings, and can be used to load the desired ElectricalSeries object.
Examples of paths are:
- "acquisition/ElectricalSeries1"
- "acquisition/ElectricalSeries2"
- "processing/ecephys/LFP/ElectricalSeries1"
- "processing/my_custom_module/MyContainer/ElectricalSeries2"
"""

if stream_mode is None:
backend = _get_backend_from_local_file(file_path)
else:
if stream_mode == "zarr":
backend = "zarr"
else:
backend = "hdf5"

file_handle = read_file_from_backend(
file_path=file_path,
stream_mode=stream_mode,
)

electrical_series_paths = _find_neurodata_type_from_backend(
file_handle,
neurodata_type="ElectricalSeries",
backend=backend,
)
return electrical_series_paths


class NwbRecordingSegment(BaseRecordingSegment):
def __init__(self, electrical_series_data, times_kwargs):
Expand Down
16 changes: 15 additions & 1 deletion src/spikeinterface/extractors/tests/test_nwbextractors.py
Original file line number Diff line number Diff line change
Expand Up @@ -319,6 +319,20 @@ def test_retrieving_from_processing(generate_nwbfile, use_pynwb):
assert np.array_equal(electrical_series_custom.data[:], recording_extractor_custom.get_traces())


def test_fetch_available_electrical_series(generate_nwbfile):
path_to_nwbfile, _ = generate_nwbfile
available_electrical_series = NwbRecordingExtractor.fetch_available_electrical_series(file_path=path_to_nwbfile)

expected_paths = [
"acquisition/ElectricalSeries1",
"acquisition/ElectricalSeries2",
"processing/ecephys/LFP/ElectricalSeries1",
"processing/my_custom_module/MyContainer/ElectricalSeries2",
]

assert available_electrical_series == expected_paths


@pytest.mark.parametrize("electrical_series_path", ["acquisition/ElectricalSeries1", "acquisition/ElectricalSeries2"])
def test_recording_equality_with_pynwb_and_backend(generate_nwbfile, electrical_series_path):
path_to_nwbfile, _ = generate_nwbfile
Expand All @@ -338,7 +352,7 @@ def test_recording_equality_with_pynwb_and_backend(generate_nwbfile, electrical_


@pytest.mark.parametrize("use_pynwb", [True, False])
def test_failure_with_wrong_electrical_series_name(generate_nwbfile, use_pynwb):
def test_failure_with_wrong_electrical_series_path(generate_nwbfile, use_pynwb):
"""Test that the extractor raises an error if the electrical series name is not found."""
path_to_nwbfile, _ = generate_nwbfile
with pytest.raises(ValueError):
Expand Down

0 comments on commit e6e5f35

Please sign in to comment.