pipeline.py

get_electrode_localization_dir(probe_insertion_key)

Return root directory of localization data for a given probe

Parameters:

Name	Type	Description	Default
probe_insertion_key	dict	key uniquely identifying one ephys.EphysRecording	required

Returns:

Name	Type	Description
path	str	Full path to localization data for either SpikeGLX or OpenEphys

Source code in workflow_array_ephys/paths.py

def get_electrode_localization_dir(probe_insertion_key: dict) -> str:
    """Return root directory of localization data for a given probe

    Args:
        probe_insertion_key (dict): key uniquely identifying one ephys.EphysRecording

    Returns:
        path (str): Full path to localization data for either SpikeGLX or OpenEphys
    """
    from .pipeline import ephys

    acq_software = (ephys.EphysRecording & probe_insertion_key).fetch1("acq_software")

    if acq_software == "SpikeGLX":
        spikeglx_meta_filepath = pathlib.Path(
            (
                ephys.EphysRecording.EphysFile
                & probe_insertion_key
                & 'file_path LIKE "%.ap.meta"'
            ).fetch1("file_path")
        )
        probe_dir = element_interface.utils.find_full_path(
            get_ephys_root_data_dir(), spikeglx_meta_filepath.parent
        )
    elif acq_software == "Open Ephys":
        probe_path = (ephys.EphysRecording.EphysFile & probe_insertion_key).fetch1(
            "file_path"
        )
        probe_dir = element_interface.utils.find_full_path(
            get_ephys_root_data_dir(), probe_path
        )

    return probe_dir

get_ephys_root_data_dir()

Return root directory for ephys from 'ephys_root_data_dir' in dj.config

Returns:

Name	Type	Description
path	any	List of path(s) if available or None

Source code in workflow_array_ephys/paths.py

def get_ephys_root_data_dir():
    """Return root directory for ephys from 'ephys_root_data_dir' in dj.config

    Returns:
        path (any): List of path(s) if available or None
    """

    data_dir = dj.config.get("custom", {}).get("ephys_root_data_dir", None)
    return pathlib.Path(data_dir) if data_dir else None

get_session_directory(session_key)

Return relative path from SessionDirectory table given key

Parameters:

Name	Type	Description	Default
session_key	dict	Key uniquely identifying a session	required

Returns:

Name	Type	Description
path	str	Relative path of session directory

Source code in workflow_array_ephys/paths.py

def get_session_directory(session_key: dict) -> str:
    """Return relative path from SessionDirectory table given key

    Args:
        session_key (dict): Key uniquely identifying a session

    Returns:
        path (str): Relative path of session directory
    """
    from .pipeline import session

    return (session.SessionDirectory & session_key).fetch1("session_dir")

ecephys_session_to_nwb(session_key, raw=True, spikes=True, lfp='source', end_frame=None, lab_key=None, project_key=None, protocol_key=None, nwbfile_kwargs=None)

Main function for converting ephys data to NWB

Parameters:

Name	Type	Description	Default
session_key	dict	key from Session table	required
raw	bool	Optional. Default True. Include the raw data from source. SpikeGLX & OpenEphys are supported	True
spikes	bool	Optional. Default True. Whether to include CuratedClustering	True
lfp	str	One of the following. "dj", read LFP data from ephys.LFP. "source", read LFP data from source (SpikeGLX supported). False, do not convert LFP.	'source'
end_frame	int	Optional limit on frames for small test conversions.	None
lab_key	dict	Optional key to add metadata from other Element Lab.	None
project_key	dict	Optional key to add metadata from other Element Lab.	None
protocol_key	dict	Optional key to add metadata from other Element Lab.	None
nwbfile_kwargs	dict	Optional. If Element Session is not used, this argument is required and must be a dictionary containing 'session_description' (str), 'identifier' (str), and 'session_start_time' (datetime), the required minimal data for instantiating an NWBFile object. If element-session is being used, this argument can optionally be used to overwrite NWBFile fields.	None

Source code in element_array_ephys/export/nwb/nwb.py

def ecephys_session_to_nwb(
    session_key,
    raw=True,
    spikes=True,
    lfp="source",
    end_frame=None,
    lab_key=None,
    project_key=None,
    protocol_key=None,
    nwbfile_kwargs=None,
):
    """Main function for converting ephys data to NWB

    Arguments:
        session_key (dict): key from Session table
        raw (bool): Optional. Default True. Include the raw data from source.
            SpikeGLX & OpenEphys are supported
        spikes (bool): Optional. Default True. Whether to include CuratedClustering
        lfp (str): One of the following.
            "dj", read LFP data from ephys.LFP.
            "source", read LFP data from source (SpikeGLX supported).
            False, do not convert LFP.
        end_frame (int): Optional limit on frames for small test conversions.
        lab_key (dict): Optional key to add metadata from other Element Lab.
        project_key (dict): Optional key to add metadata from other Element Lab.
        protocol_key (dict): Optional key to add metadata from other Element Lab.
        nwbfile_kwargs (dict): Optional. If Element Session is not used, this argument
            is required and must be a dictionary containing 'session_description' (str),
            'identifier' (str), and 'session_start_time' (datetime), the required
             minimal data for instantiating an NWBFile object. If element-session is
             being used, this argument can optionally be used to overwrite NWBFile
             fields.
    """

    session_to_nwb = getattr(ephys._linking_module, "session_to_nwb", False)

    if session_to_nwb:
        nwbfile = session_to_nwb(
            session_key,
            lab_key=lab_key,
            project_key=project_key,
            protocol_key=protocol_key,
            additional_nwbfile_kwargs=nwbfile_kwargs,
        )
    else:
        nwbfile = pynwb.NWBFile(**nwbfile_kwargs)

    ephys_root_data_dir = ephys.get_ephys_root_data_dir()

    if raw:
        add_ephys_recording_to_nwb(
            session_key,
            ephys_root_data_dir=ephys_root_data_dir,
            nwbfile=nwbfile,
            end_frame=end_frame,
        )

    if spikes:
        add_ephys_units_to_nwb(session_key, nwbfile)

    if lfp == "dj":
        add_ephys_lfp_from_dj_to_nwb(session_key, nwbfile)

    if lfp == "source":
        add_ephys_lfp_from_source_to_nwb(
            session_key,
            ephys_root_data_dir=ephys_root_data_dir,
            nwbfile=nwbfile,
            end_frame=end_frame,
        )

    return nwbfile

write_nwb(nwbfile, fname, check_read=True)

Export NWBFile

Parameters:

Name	Type	Description	Default
nwbfile	NWBFile	nwb file	required
fname	str	Absolute path including *.nwb extension.	required
check_read	bool	If True, PyNWB will try to read the produced NWB file and ensure that it can be read.	True

Source code in element_array_ephys/export/nwb/nwb.py

def write_nwb(nwbfile, fname, check_read=True):
    """Export NWBFile

    Arguments:
        nwbfile (NWBFile): nwb file
        fname (str): Absolute path including `*.nwb` extension.
        check_read (bool): If True, PyNWB will try to read the produced NWB file and
            ensure that it can be read.
    """
    with pynwb.NWBHDF5IO(fname, "w") as io:
        io.write(nwbfile)

    if check_read:
        with pynwb.NWBHDF5IO(fname, "r") as io:
            io.read()