import distutils.version
import uuid
import warnings
from copy import deepcopy
from pathlib import Path
from typing import Optional
import neo.io.baseio
import numpy as np
import pynwb
from pydantic import FilePath
from ..nwb_helpers import add_device_from_metadata
response_classes = dict(
voltage_clamp=pynwb.icephys.VoltageClampSeries,
current_clamp=pynwb.icephys.CurrentClampSeries,
izero=pynwb.icephys.IZeroClampSeries,
)
stim_classes = dict(
voltage_clamp=pynwb.icephys.VoltageClampStimulusSeries,
current_clamp=pynwb.icephys.CurrentClampStimulusSeries,
)
# TODO - get electrodes metadata
[docs]def get_number_of_electrodes(neo_reader) -> int:
"""
Get number of electrodes from Neo reader.
Returns
-------
int
The total number of electrodes in the recording.
"""
# TODO - take in account the case with multiple streams.
return len(neo_reader.header["signal_channels"])
[docs]def get_number_of_segments(neo_reader, block: int = 0) -> int:
"""
Get number of segments from Neo reader.
Parameters
----------
neo_reader : neo.io.baseio
The Neo reader object.
block : int, default: 0
Block index.
Returns
-------
int
The number of segments in the specified block.
"""
return neo_reader.header["nb_segment"][block]
[docs]def get_command_traces(neo_reader, segment: int = 0, cmd_channel: int = 0) -> tuple[list, str, str]:
"""
Get command traces (e.g. voltage clamp command traces).
Parameters
----------
neo_reader : neo.io.baseio
The Neo reader object.
segment : int, optional
Segment index. Defaults to 0.
cmd_channel : int, optional
ABF command channel (0 to 7). Defaults to 0.
Returns
-------
tuple[list, str, str]
A tuple containing:
- list: The command trace data
- str: The title of the command trace
- str: The units of the command trace
Notes
-----
This function only works for AxonIO interface.
"""
try:
traces, titles, units = neo_reader.read_raw_protocol()
return traces[segment][cmd_channel], titles[segment][cmd_channel], units[segment][cmd_channel]
except Exception as e:
msg = ".\n\n WARNING - get_command_traces() only works for AxonIO interface."
e.args = (str(e) + msg,)
return e
[docs]def get_conversion_from_unit(unit: str) -> float:
"""
Get conversion (to Volt or Ampere) from unit in string format.
Parameters
----------
unit : str
Unit as string. E.g. pA, mV, uV, etc...
Returns
-------
float
The conversion factor to convert to Ampere or Volt.
For example, for 'pA' returns 1e-12 to convert to Ampere.
"""
if unit in ["pA", "pV"]:
conversion = 1e-12
elif unit in ["nA", "nV"]:
conversion = 1e-9
elif unit in ["uA", "uV"]:
conversion = 1e-6
elif unit in ["mA", "mV"]:
conversion = 1e-3
elif unit in ["A", "V"]:
conversion = 1.0
else:
conversion = 1.0
warnings.warn("No valid units found for traces in the current file. Gain is set to 1, but this might be wrong.")
return float(conversion)
[docs]def add_icephys_electrode(neo_reader, nwbfile, metadata: dict = None):
"""
Add icephys electrodes to nwbfile object.
Will always ensure nwbfile has at least one icephys electrode.
Will auto-generate a linked device if the specified name does not exist in the nwbfile.
Parameters
----------
neo_reader : neo.io.baseio
nwbfile : NWBFile
NWBFile object to add the icephys electrode to.
metadata : dict, optional
Metadata info for constructing the nwb file.
Should be of the format::
metadata['Icephys']['Electrodes'] = [
{
'name': my_name,
'description': my_description,
'device_name': my_device_name
},
...
]
"""
metadata_copy = deepcopy(metadata) if metadata is not None else dict()
assert isinstance(nwbfile, pynwb.NWBFile), "'nwbfile' should be of type pynwb.NWBFile"
if len(nwbfile.devices) == 0:
warnings.warn("When adding Icephys Electrode, no Devices were found on nwbfile. Creating a Device now...")
add_device_from_metadata(nwbfile=nwbfile, modality="Icephys", metadata=metadata_copy)
if "Icephys" not in metadata_copy:
metadata_copy["Icephys"] = dict()
defaults = [
dict(
name=f"icephys_electrode_{elec_id}",
description="no description",
device_name=[i.name for i in nwbfile.devices.values()][0],
)
for elec_id in range(get_number_of_electrodes(neo_reader))
]
if "Electrodes" not in metadata_copy["Icephys"] or len(metadata_copy["Icephys"]["Electrodes"]) == 0:
metadata_copy["Icephys"]["Electrodes"] = defaults
assert all(
[isinstance(x, dict) for x in metadata_copy["Icephys"]["Electrodes"]]
), "Expected metadata['Icephys']['Electrodes'] to be a list of dictionaries!"
# Create Icephys electrode from metadata
for elec in metadata_copy["Icephys"]["Electrodes"]:
if elec.get("name", defaults[0]["name"]) not in nwbfile.icephys_electrodes:
device_name = elec.pop("device_name", None) or elec.pop("device", defaults[0]["device_name"])
# elec.pop("device_name", 0)
if device_name not in nwbfile.devices:
new_device_metadata = dict(Ecephys=dict(Device=[dict(name=device_name)]))
add_device_from_metadata(nwbfile, modality="Icephys", metadata=new_device_metadata)
warnings.warn(
f"Device '{device_name}' not detected in "
"attempted link to icephys electrode! Automatically generating."
)
electrode_kwargs = elec
electrode_kwargs.update(
name=elec.get("name", defaults[0]["name"]),
description=elec.get("description", defaults[0]["description"]),
device=nwbfile.devices[device_name],
)
nwbfile.create_icephys_electrode(**electrode_kwargs)
[docs]def add_icephys_recordings(
neo_reader,
nwbfile: pynwb.NWBFile,
metadata: dict = None,
icephys_experiment_type: str = "voltage_clamp",
stimulus_type: str = "not described",
skip_electrodes: tuple[int] = (),
):
"""
Add icephys recordings (stimulus/response pairs) to nwbfile object.
Parameters
----------
neo_reader : neo.io.baseio
nwbfile : NWBFile
metadata : dict, optional
icephys_experiment_type : {'voltage_clamp', 'current_clamp', 'izero'}
Type of icephys recording.
stimulus_type : str, default: 'not described'
skip_electrodes : tuple, default: ()
Electrode IDs to skip.
"""
n_segments = get_number_of_segments(neo_reader, block=0)
# Check for protocol data (only ABF2), necessary for stimuli data
if neo_reader._axon_info["fFileVersionNumber"] < 2:
n_commands = 0
warnings.warn(
f"Protocol section is only present in ABF2 files. {neo_reader.filename} has version "
f"{neo_reader._axon_info['fFileVersionNumber']}. Saving experiment as 'i_zero'..."
)
else:
protocol = neo_reader.read_raw_protocol()
n_commands = len(protocol[0])
if n_commands == 0:
icephys_experiment_type = "izero"
warnings.warn(
f"No command data found by neo reader in file {neo_reader.filename}. Saving experiment as 'i_zero'..."
)
else:
assert (
n_commands == n_segments
), f"File contains inconsistent number of segments ({n_segments}) and commands ({n_commands})"
assert icephys_experiment_type in ["voltage_clamp", "current_clamp", "izero"], (
f"'icephys_experiment_type' should be 'voltage_clamp', 'current_clamp' or 'izero', but received value "
f"{icephys_experiment_type}"
)
# Check and auto-create electrodes, in case they don't exist yet in nwbfile
if len(nwbfile.icephys_electrodes) == 0:
warnings.warn(
"When adding Icephys Recording, no Icephys Electrodes were found on nwbfile. Creating Electrodes now..."
)
add_icephys_electrode(
neo_reader=neo_reader,
nwbfile=nwbfile,
metadata=metadata,
)
if getattr(nwbfile, "intracellular_recordings", None):
ri = max(nwbfile.intracellular_recordings["responses"].index)
else:
ri = -1
if getattr(nwbfile, "icephys_simultaneous_recordings", None):
simultaneous_recordings_offset = len(nwbfile.icephys_simultaneous_recordings)
else:
simultaneous_recordings_offset = 0
if getattr(nwbfile, "icephys_sequential_recordings", None):
sessions_offset = len(nwbfile.icephys_sequential_recordings)
else:
sessions_offset = 0
relative_session_start_time = deepcopy(
metadata["Icephys"]["Sessions"][sessions_offset]["relative_session_start_time"]
)
session_stimulus_type = deepcopy(metadata["Icephys"]["Sessions"][sessions_offset]["stimulus_type"])
# Sequential icephys recordings
simultaneous_recordings = list()
for si in range(n_segments):
# Parallel icephys recordings
recordings = list()
for ei, electrode in enumerate(
list(nwbfile.icephys_electrodes.values())[: len(neo_reader.header["signal_channels"]["units"])]
):
if ei in skip_electrodes:
continue
# Starting time is the signal starting time within .abf file + time
# relative to first session (first .abf file)
ri += 1
starting_time = neo_reader.get_signal_t_start(block_index=0, seg_index=si)
starting_time = starting_time + relative_session_start_time
sampling_rate = neo_reader.get_signal_sampling_rate()
response_unit = neo_reader.header["signal_channels"]["units"][ei]
response_conversion = get_conversion_from_unit(unit=response_unit)
response_gain = neo_reader.header["signal_channels"]["gain"][ei]
response_name = f"{icephys_experiment_type}-response-{si + 1 + simultaneous_recordings_offset:02}-ch-{ei}"
response = response_classes[icephys_experiment_type](
name=response_name,
description=f"Response to: {session_stimulus_type}",
electrode=electrode,
data=neo_reader.get_analogsignal_chunk(block_index=0, seg_index=si, channel_indexes=ei),
starting_time=starting_time,
rate=sampling_rate,
conversion=response_conversion * response_gain,
gain=np.nan,
)
if icephys_experiment_type != "izero":
stim_unit = protocol[2][ei]
stim_conversion = get_conversion_from_unit(unit=stim_unit)
stimulus = stim_classes[icephys_experiment_type](
name=f"stimulus-{si + 1 + simultaneous_recordings_offset:02}-ch-{ei}",
description=f"Stim type: {session_stimulus_type}",
electrode=electrode,
data=protocol[0][si][ei],
rate=sampling_rate,
starting_time=starting_time,
conversion=stim_conversion,
gain=np.nan,
)
icephys_recording = nwbfile.add_intracellular_recording(
electrode=electrode, response=response, stimulus=stimulus
)
else:
icephys_recording = nwbfile.add_intracellular_recording(electrode=electrode, response=response)
recordings.append(icephys_recording)
sim_rec = nwbfile.add_icephys_simultaneous_recording(recordings=recordings)
simultaneous_recordings.append(sim_rec)
nwbfile.add_icephys_sequential_recording(
simultaneous_recordings=simultaneous_recordings, stimulus_type=stimulus_type
)
# TODO
# # Add a list of sequential recordings table indices as a repetition
# run_index = nwbfile.add_icephys_repetition(
# sequential_recordings=[
# seq_rec,
# ]
# )
# # Add a list of repetition table indices as a experimental condition
# nwbfile.add_icephys_experimental_condition(
# repetitions=[
# run_index,
# ]
# )
[docs]def add_neo_to_nwb(
neo_reader,
nwbfile: pynwb.NWBFile,
metadata: dict = None,
icephys_experiment_type: str = "voltage_clamp",
stimulus_type: Optional[str] = None,
skip_electrodes: tuple[int] = (),
):
"""
Auxiliary static method for nwbextractor.
Adds all recording related information from recording object and metadata to the nwbfile object.
Parameters
----------
neo_reader: Neo reader object
nwbfile: NWBFile
nwb file to which the recording information is to be added
metadata: dict
metadata info for constructing the nwb file (optional).
Check the auxiliary function docstrings for more information
about metadata format.
icephys_experiment_type: str (optional)
Type of Icephys experiment. Allowed types are: 'voltage_clamp', 'current_clamp' and 'izero'.
If no value is passed, 'voltage_clamp' is used as default.
stimulus_type: str, optional
skip_electrodes: tuple, optional
Electrode IDs to skip. Defaults to ().
"""
assert isinstance(nwbfile, pynwb.NWBFile), "'nwbfile' should be of type pynwb.NWBFile"
# TODO: remove completely after 10/1/2024
add_device_from_metadata(nwbfile=nwbfile, modality="Icephys", metadata=metadata)
add_icephys_electrode(
neo_reader=neo_reader,
nwbfile=nwbfile,
metadata=metadata,
)
add_icephys_recordings(
neo_reader=neo_reader,
nwbfile=nwbfile,
metadata=metadata,
icephys_experiment_type=icephys_experiment_type,
stimulus_type=stimulus_type,
skip_electrodes=skip_electrodes,
)
[docs]def write_neo_to_nwb(
neo_reader: neo.io.baseio.BaseIO,
save_path: Optional[FilePath] = None, # pragma: no cover
overwrite: bool = False,
nwbfile=None,
metadata: dict = None,
icephys_experiment_type: Optional[str] = None,
stimulus_type: Optional[str] = None,
skip_electrodes: Optional[tuple] = (),
):
"""
Primary method for writing a Neo reader object to an NWBFile.
Parameters
----------
neo_reader: Neo reader
save_path: PathType
Required if an nwbfile is not passed. Must be the path to the nwbfile
being appended, otherwise one is created and written.
overwrite: bool
If using save_path, whether to overwrite the NWBFile if it already exists.
nwbfile: NWBFile
Required if a save_path is not specified. If passed, this function
will fill the relevant fields within the nwbfile.
metadata: dict
metadata info for constructing the nwb file (optional). Should be of the format::
metadata['Ecephys'] = {}
with keys of the forms::
metadata['Ecephys']['Device'] = [
{
'name': my_name,
'description': my_description
},
...
]
metadata['Ecephys']['ElectrodeGroup'] = [
{
'name': my_name,
'description': my_description,
'location': electrode_location,
'device': my_device_name
},
...
]
metadata['Ecephys']['Electrodes'] = [
{
'name': my_name,
'description': my_description
},
...
]
metadata['Ecephys']['ElectricalSeries'] = {
'name': my_name,
'description': my_description
}
Note that data intended to be added to the electrodes table of the NWBFile should be set as channel
properties in the RecordingExtractor object.
icephys_experiment_type: str (optional)
Type of Icephys experiment. Allowed types are: 'voltage_clamp', 'current_clamp' and 'izero'.
If no value is passed, 'voltage_clamp' is used as default.
stimulus_type: str, optional
skip_electrodes: tuple, optional
Electrode IDs to skip. Defaults to ().
"""
if nwbfile is not None:
assert isinstance(nwbfile, pynwb.NWBFile), "'nwbfile' should be of type pynwb.NWBFile"
assert (
distutils.version.LooseVersion(pynwb.__version__) >= "1.3.3"
), "'write_neo_to_nwb' not supported for version < 1.3.3. Run pip install --upgrade pynwb"
assert save_path is None or nwbfile is None, "Either pass a save_path location, or nwbfile object, but not both!"
if metadata is None:
metadata = get_nwb_metadata(neo_reader=neo_reader)
kwargs = dict(
neo_reader=neo_reader,
metadata=metadata,
icephys_experiment_type=icephys_experiment_type,
stimulus_type=stimulus_type,
skip_electrodes=skip_electrodes,
)
if nwbfile is None:
if Path(save_path).is_file() and not overwrite:
read_mode = "r+"
else:
read_mode = "w"
with pynwb.NWBHDF5IO(str(save_path), mode=read_mode) as io:
if read_mode == "r+":
nwbfile = io.read()
else:
nwbfile_kwargs = dict(
session_description="Auto-generated by NwbRecordingExtractor without description.",
identifier=str(uuid.uuid4()),
)
if metadata is not None and "NWBFile" in metadata:
nwbfile_kwargs.update(metadata["NWBFile"])
nwbfile = pynwb.NWBFile(**nwbfile_kwargs)
add_neo_to_nwb(nwbfile=nwbfile, **kwargs)
io.write(nwbfile)
else:
add_neo_to_nwb(nwbfile=nwbfile, **kwargs)