from typing import List, Optional, Tuple
import numpy as np
import pandas as pd
import pynwb
from allensdk.brain_observatory import dict_to_indexed_array
from allensdk.brain_observatory.ecephys.nwb import EcephysProbe, \
EcephysElectrodeGroup
ELECTRODE_TABLE_DEFAULT_COLUMNS = [
("probe_vertical_position",
"Length-wise position of electrode/channel on device (microns)"),
("probe_horizontal_position",
"Width-wise position of electrode/channel on device (microns)"),
("probe_id", "The unique id of this electrode's/channel's device"),
("probe_channel_number",
"The local index of electrode/channel on device"),
("valid_data", "Whether data from this electrode/channel is usable")
]
[docs]def add_ragged_data_to_dynamic_table(
table, data, column_name, column_description=""
):
""" Builds the index and data vectors required for writing ragged array
data to a pynwb dynamic table
Parameters
----------
table : pynwb.core.DynamicTable
table to which data will be added (as VectorData / VectorIndex)
data : dict
each key-value pair describes some grouping of data
column_name : str
used to set the name of this column
column_description : str, optional
used to set the description of this column
Returns
-------
nwbfile : pynwb.NWBFile
"""
idx, values = dict_to_indexed_array(data, table.id.data)
del data
table.add_column(
name=column_name,
description=column_description,
data=values,
index=idx
)
[docs]def add_probe_to_nwbfile(nwbfile, probe_id, sampling_rate, lfp_sampling_rate,
has_lfp_data, name,
location="See electrode locations"):
""" Creates objects required for representation of a single
extracellular ephys probe within an NWB file.
Parameters
----------
nwbfile : pynwb.NWBFile
file to which probe information will be assigned.
probe_id : int
unique identifier for this probe
sampling_rate: float,
sampling rate of the neuropixels probe
lfp_sampling_rate: float
sampling rate of LFP
has_lfp_data: bool
True if LFP data is available for the probe, otherwise False
name : str, optional
human-readable name for this probe.
Practically, we use tags like "probeA" or "probeB"
location : str, optional
A required field for the `EcephysElectrodeGroup`. Because the group
contains a number of electrodes/channels along the neuropixels probe,
location will vary significantly. Thus by default this field is:
"See electrode locations" where the nwbfile.electrodes table will
provide much more detailed location information.
Returns
------
nwbfile : pynwb.NWBFile
the updated file object
probe_nwb_device : pynwb.device.Device
device object corresponding to this probe
probe_nwb_electrode_group : pynwb.ecephys.ElectrodeGroup
electrode group object corresponding to this probe
"""
probe_nwb_device = EcephysProbe(name=name,
description="Neuropixels 1.0 Probe",
manufacturer="imec",
probe_id=probe_id,
sampling_rate=sampling_rate)
probe_nwb_electrode_group = EcephysElectrodeGroup(
name=name,
description="Ecephys Electrode Group", # required field
probe_id=probe_id,
location=location,
device=probe_nwb_device,
lfp_sampling_rate=lfp_sampling_rate,
has_lfp_data=has_lfp_data
)
nwbfile.add_device(probe_nwb_device)
nwbfile.add_electrode_group(probe_nwb_electrode_group)
return nwbfile, probe_nwb_device, probe_nwb_electrode_group
[docs]def add_ecephys_electrodes(
nwbfile: pynwb.NWBFile,
channels: List[dict],
electrode_group: EcephysElectrodeGroup,
channel_number_whitelist: Optional[np.ndarray] = None):
"""Add electrode information to an ecephys nwbfile electrode table.
Parameters
----------
nwbfile : pynwb.NWBFile
The nwbfile to add electrodes data to
channels : List[dict]
A list of 'channel' dictionaries containing the following fields:
id: The unique id for a given electrode/channel
probe_id: The unique id for an electrode's/channel's device
valid_data: Whether the data for an electrode/channel is usable
local_index: The local index of an electrode/channel on a
given device
probe_vertical_position: Length-wise position of electrode/channel
on device (microns)
probe_horizontal_position: Width-wise position of electrode/channel
on device (microns)
structure_id: The LIMS id associated with an anatomical
structure
structure_acronym: Acronym associated with an anatomical
structure
anterior_posterior_ccf_coordinate
dorsal_ventral_ccf_coordinate
left_right_ccf_coordinate
Optional fields which may be used in the future:
impedence: The impedence of a given channel.
filtering: The type of hardware filtering done a channel.
(e.g. "1000 Hz low-pass filter")
electrode_group : EcephysElectrodeGroup
The pynwb electrode group that electrodes should be associated with
channel_number_whitelist : Optional[np.ndarray], optional
If provided, only add electrodes (a.k.a. channels) specified by the
whitelist (and in order specified), by default None
"""
_add_ecephys_electrode_columns(nwbfile)
channel_table = pd.DataFrame(channels)
if channel_number_whitelist is not None:
channel_table.set_index("probe_channel_number", inplace=True)
channel_table = channel_table.loc[channel_number_whitelist, :]
channel_table.reset_index(inplace=True)
for _, row in channel_table.iterrows():
x = row["anterior_posterior_ccf_coordinate"]
y = row["dorsal_ventral_ccf_coordinate"]
z = row["left_right_ccf_coordinate"]
nwbfile.add_electrode(
id=row["id"],
x=(np.nan if x is None else x), # Not all probes have CCF coords
y=(np.nan if y is None else y),
z=(np.nan if z is None else z),
probe_vertical_position=row["probe_vertical_position"],
probe_horizontal_position=row["probe_horizontal_position"],
probe_channel_number=row["probe_channel_number"],
valid_data=row["valid_data"],
probe_id=row["probe_id"],
group=electrode_group,
location=row["structure_acronym"],
imp=row.get("impedence", row.get("impedance")),
filtering=row["filtering"]
)
def _add_ecephys_electrode_columns(nwbfile: pynwb.NWBFile,
columns_to_add:
Optional[List[Tuple[str, str]]] = None):
"""Add additional columns to ecephys nwbfile electrode table.
Parameters
----------
nwbfile : pynwb.NWBFile
An nwbfile to add additional electrode columns to
columns_to_add : Optional[List[Tuple[str, str]]]
A list of (column_name, column_description) tuples to be added
to the nwbfile electrode table, by default None. If None, default
columns are added.
"""
if columns_to_add is None:
columns_to_add = ELECTRODE_TABLE_DEFAULT_COLUMNS
for col_name, col_description in columns_to_add:
if (not nwbfile.electrodes) or \
(col_name not in nwbfile.electrodes.colnames):
nwbfile.add_electrode_column(name=col_name,
description=col_description)