common_ephys.py

Electrode

Bases: SpyglassMixin, Imported

Source code in src/spyglass/common/common_ephys.py

@schema
class Electrode(SpyglassMixin, dj.Imported):
    definition = """
    -> ElectrodeGroup
    electrode_id: int                      # the unique number for this electrode
    ---
    -> [nullable] Probe.Electrode
    -> BrainRegion
    name = "": varchar(200)                 # unique label for each contact
    original_reference_electrode = -1: int  # the configured reference electrode for this electrode
    x = NULL: float                         # the x coordinate of the electrode position in the brain
    y = NULL: float                         # the y coordinate of the electrode position in the brain
    z = NULL: float                         # the z coordinate of the electrode position in the brain
    filtering: varchar(2000)                # description of the signal filtering
    impedance = NULL: float                 # electrode impedance
    bad_channel = "False": enum("True", "False")  # if electrode is "good" or "bad" as observed during recording
    x_warped = NULL: float                  # x coordinate of electrode position warped to common template brain
    y_warped = NULL: float                  # y coordinate of electrode position warped to common template brain
    z_warped = NULL: float                  # z coordinate of electrode position warped to common template brain
    contacts: varchar(200)                  # label of electrode contacts used for a bipolar signal - current workaround
    """

    def make(self, key):
        nwb_file_name = key["nwb_file_name"]
        nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)
        config = get_config(nwb_file_abspath)

        if "Electrode" in config:
            electrode_config_dicts = {
                electrode_dict["electrode_id"]: electrode_dict
                for electrode_dict in config["Electrode"]
            }
        else:
            electrode_config_dicts = dict()

        electrodes = nwbf.electrodes.to_dataframe()
        for elect_id, elect_data in electrodes.iterrows():
            key["electrode_id"] = elect_id
            key["name"] = str(elect_id)
            key["electrode_group_name"] = elect_data.group_name
            key["region_id"] = BrainRegion.fetch_add(
                region_name=elect_data.group.location
            )
            key["x"] = elect_data.x
            key["y"] = elect_data.y
            key["z"] = elect_data.z
            key["x_warped"] = 0
            key["y_warped"] = 0
            key["z_warped"] = 0
            key["contacts"] = ""
            key["filtering"] = elect_data.filtering
            key["impedance"] = elect_data.get("imp")

            # rough check of whether the electrodes table was created by
            # rec_to_nwb and has the appropriate custom columns used by
            # rec_to_nwb

            # TODO this could be better resolved by making an extension for the
            # electrodes table

            if (
                isinstance(elect_data.group.device, ndx_franklab_novela.Probe)
                and "probe_shank" in elect_data
                and "probe_electrode" in elect_data
                and "bad_channel" in elect_data
                and "ref_elect_id" in elect_data
            ):
                key["probe_id"] = elect_data.group.device.probe_type
                key["probe_shank"] = elect_data.probe_shank
                key["probe_electrode"] = elect_data.probe_electrode
                key["bad_channel"] = (
                    "True" if elect_data.bad_channel else "False"
                )
                key["original_reference_electrode"] = elect_data.ref_elect_id

            # override with information from the config YAML based on primary
            # key (electrode id)

            if elect_id in electrode_config_dicts:
                # check whether the Probe.Electrode being referenced exists
                query = Probe.Electrode & electrode_config_dicts[elect_id]
                if len(query) == 0:
                    warnings.warn(
                        "No Probe.Electrode exists that matches the data: "
                        + f"{electrode_config_dicts[elect_id]}. "
                        "The config YAML for Electrode with electrode_id "
                        + f"{elect_id} will be ignored."
                    )
                else:
                    key.update(electrode_config_dicts[elect_id])

            self.insert1(key, skip_duplicates=True)

    @classmethod
    def create_from_config(cls, nwb_file_name: str):
        """Create or update Electrode entries from what is specified in the config YAML file.

        Parameters
        ----------
        nwb_file_name : str
            The name of the NWB file.
        """
        nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)
        config = get_config(nwb_file_abspath)
        if "Electrode" not in config:
            return

        # map electrode id to dictof electrode information from config YAML
        electrode_dicts = {
            electrode_dict["electrode_id"]: electrode_dict
            for electrode_dict in config["Electrode"]
        }

        electrodes = nwbf.electrodes.to_dataframe()
        for nwbfile_elect_id, elect_data in electrodes.iterrows():
            if nwbfile_elect_id in electrode_dicts:
                # use the information in the electrodes table to start and then
                # add (or overwrite) values from the config YAML

                key = dict()
                key["nwb_file_name"] = nwb_file_name
                key["name"] = str(nwbfile_elect_id)
                key["electrode_group_name"] = elect_data.group_name
                key["region_id"] = BrainRegion.fetch_add(
                    region_name=elect_data.group.location
                )
                key["x"] = elect_data.x
                key["y"] = elect_data.y
                key["z"] = elect_data.z
                key["x_warped"] = 0
                key["y_warped"] = 0
                key["z_warped"] = 0
                key["contacts"] = ""
                key["filtering"] = elect_data.filtering
                key["impedance"] = elect_data.get("imp")
                key.update(electrode_dicts[nwbfile_elect_id])
                query = Electrode & {"electrode_id": nwbfile_elect_id}
                if len(query):
                    cls.update1(key)
                    logger.info(
                        f"Updated Electrode with ID {nwbfile_elect_id}."
                    )
                else:
                    cls.insert1(
                        key, skip_duplicates=True, allow_direct_insert=True
                    )
                    logger.info(
                        f"Inserted Electrode with ID {nwbfile_elect_id}."
                    )
            else:
                warnings.warn(
                    f"Electrode ID {nwbfile_elect_id} exists in the NWB file "
                    + "but has no corresponding config YAML entry."
                )

create_from_config(nwb_file_name) classmethod

Create or update Electrode entries from what is specified in the config YAML file.

Parameters:

Name	Type	Description	Default
nwb_file_name	str	The name of the NWB file.	required

Source code in src/spyglass/common/common_ephys.py

@classmethod
def create_from_config(cls, nwb_file_name: str):
    """Create or update Electrode entries from what is specified in the config YAML file.

    Parameters
    ----------
    nwb_file_name : str
        The name of the NWB file.
    """
    nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
    nwbf = get_nwb_file(nwb_file_abspath)
    config = get_config(nwb_file_abspath)
    if "Electrode" not in config:
        return

    # map electrode id to dictof electrode information from config YAML
    electrode_dicts = {
        electrode_dict["electrode_id"]: electrode_dict
        for electrode_dict in config["Electrode"]
    }

    electrodes = nwbf.electrodes.to_dataframe()
    for nwbfile_elect_id, elect_data in electrodes.iterrows():
        if nwbfile_elect_id in electrode_dicts:
            # use the information in the electrodes table to start and then
            # add (or overwrite) values from the config YAML

            key = dict()
            key["nwb_file_name"] = nwb_file_name
            key["name"] = str(nwbfile_elect_id)
            key["electrode_group_name"] = elect_data.group_name
            key["region_id"] = BrainRegion.fetch_add(
                region_name=elect_data.group.location
            )
            key["x"] = elect_data.x
            key["y"] = elect_data.y
            key["z"] = elect_data.z
            key["x_warped"] = 0
            key["y_warped"] = 0
            key["z_warped"] = 0
            key["contacts"] = ""
            key["filtering"] = elect_data.filtering
            key["impedance"] = elect_data.get("imp")
            key.update(electrode_dicts[nwbfile_elect_id])
            query = Electrode & {"electrode_id": nwbfile_elect_id}
            if len(query):
                cls.update1(key)
                logger.info(
                    f"Updated Electrode with ID {nwbfile_elect_id}."
                )
            else:
                cls.insert1(
                    key, skip_duplicates=True, allow_direct_insert=True
                )
                logger.info(
                    f"Inserted Electrode with ID {nwbfile_elect_id}."
                )
        else:
            warnings.warn(
                f"Electrode ID {nwbfile_elect_id} exists in the NWB file "
                + "but has no corresponding config YAML entry."
            )

LFPSelection

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_ephys.py

@schema
class LFPSelection(SpyglassMixin, dj.Manual):
    definition = """
     -> Session
     """

    class LFPElectrode(SpyglassMixin, dj.Part):
        definition = """
        -> LFPSelection
        -> Electrode
        """

    def set_lfp_electrodes(self, nwb_file_name, electrode_list):
        """Removes all electrodes for the specified nwb file and then adds back the electrodes in the list

        Parameters
        ----------
        nwb_file_name : str
            The name of the nwb file for the desired session
        electrode_list : list
            list of electrodes to be used for LFP

        """
        # remove the session and then recreate the session and Electrode list
        (LFPSelection() & {"nwb_file_name": nwb_file_name}).delete()
        # check to see if the user allowed the deletion
        if (
            len((LFPSelection() & {"nwb_file_name": nwb_file_name}).fetch())
            == 0
        ):
            LFPSelection().insert1({"nwb_file_name": nwb_file_name})

            # TODO: do this in a better way
            all_electrodes = (
                Electrode() & {"nwb_file_name": nwb_file_name}
            ).fetch(as_dict=True)
            primary_key = Electrode.primary_key
            for e in all_electrodes:
                # create a dictionary so we can insert new elects
                if e["electrode_id"] in electrode_list:
                    lfpelectdict = {
                        k: v for k, v in e.items() if k in primary_key
                    }
                    LFPSelection().LFPElectrode.insert1(
                        lfpelectdict, replace=True
                    )

set_lfp_electrodes(nwb_file_name, electrode_list)

Removes all electrodes for the specified nwb file and then adds back the electrodes in the list

Parameters:

Name	Type	Description	Default
nwb_file_name	str	The name of the nwb file for the desired session	required
electrode_list	list	list of electrodes to be used for LFP	required

Source code in src/spyglass/common/common_ephys.py

def set_lfp_electrodes(self, nwb_file_name, electrode_list):
    """Removes all electrodes for the specified nwb file and then adds back the electrodes in the list

    Parameters
    ----------
    nwb_file_name : str
        The name of the nwb file for the desired session
    electrode_list : list
        list of electrodes to be used for LFP

    """
    # remove the session and then recreate the session and Electrode list
    (LFPSelection() & {"nwb_file_name": nwb_file_name}).delete()
    # check to see if the user allowed the deletion
    if (
        len((LFPSelection() & {"nwb_file_name": nwb_file_name}).fetch())
        == 0
    ):
        LFPSelection().insert1({"nwb_file_name": nwb_file_name})

        # TODO: do this in a better way
        all_electrodes = (
            Electrode() & {"nwb_file_name": nwb_file_name}
        ).fetch(as_dict=True)
        primary_key = Electrode.primary_key
        for e in all_electrodes:
            # create a dictionary so we can insert new elects
            if e["electrode_id"] in electrode_list:
                lfpelectdict = {
                    k: v for k, v in e.items() if k in primary_key
                }
                LFPSelection().LFPElectrode.insert1(
                    lfpelectdict, replace=True
                )

LFPBandSelection

Bases: SpyglassMixin, Manual

Source code in src/spyglass/common/common_ephys.py

@schema
class LFPBandSelection(SpyglassMixin, dj.Manual):
    definition = """
    -> LFP
    -> FirFilterParameters                   # the filter to use for the data
    -> IntervalList.proj(target_interval_list_name='interval_list_name')  # the original set of times to be filtered
    lfp_band_sampling_rate: int    # the sampling rate for this band
    ---
    min_interval_len = 1: float  # the minimum length of a valid interval to filter
    """

    class LFPBandElectrode(SpyglassMixin, dj.Part):
        definition = """
        -> LFPBandSelection
        -> LFPSelection.LFPElectrode  # the LFP electrode to be filtered
        reference_elect_id = -1: int  # the reference electrode to use; -1 for no reference
        ---
        """

    def set_lfp_band_electrodes(
        self,
        nwb_file_name,
        electrode_list,
        filter_name,
        interval_list_name,
        reference_electrode_list,
        lfp_band_sampling_rate,
    ):
        """
        Adds an entry for each electrode in the electrode_list with the specified filter, interval_list, and
        reference electrode.
        Also removes any entries that have the same filter, interval list and reference electrode but are not
        in the electrode_list.
        :param nwb_file_name: string - the name of the nwb file for the desired session
        :param electrode_list: list of LFP electrodes to be filtered
        :param filter_name: the name of the filter (from the FirFilterParameters schema)
        :param interval_name: the name of the interval list (from the IntervalList schema)
        :param reference_electrode_list: A single electrode id corresponding to the reference to use for all
        electrodes or a list with one element per entry in the electrode_list
        :param lfp_band_sampling_rate: The output sampling rate to be used for the filtered data; must be an
        integer divisor of the LFP sampling rate
        :return: none
        """
        # Error checks on parameters
        # electrode_list
        query = LFPSelection().LFPElectrode() & {"nwb_file_name": nwb_file_name}
        available_electrodes = query.fetch("electrode_id")
        if not np.all(np.isin(electrode_list, available_electrodes)):
            raise ValueError(
                "All elements in electrode_list must be valid electrode_ids in the LFPSelection table"
            )
        # sampling rate
        lfp_sampling_rate = (LFP() & {"nwb_file_name": nwb_file_name}).fetch1(
            "lfp_sampling_rate"
        )
        decimation = lfp_sampling_rate // lfp_band_sampling_rate
        if lfp_sampling_rate // decimation != lfp_band_sampling_rate:
            raise ValueError(
                f"lfp_band_sampling rate {lfp_band_sampling_rate} is not an integer divisor of lfp "
                f"samping rate {lfp_sampling_rate}"
            )
        # filter
        query = FirFilterParameters() & {
            "filter_name": filter_name,
            "filter_sampling_rate": lfp_sampling_rate,
        }
        if not query:
            raise ValueError(
                f"filter {filter_name}, sampling rate {lfp_sampling_rate} is not in the FirFilterParameters table"
            )
        # interval_list
        query = IntervalList() & {
            "nwb_file_name": nwb_file_name,
            "interval_name": interval_list_name,
        }
        if not query:
            raise ValueError(
                f"interval list {interval_list_name} is not in the IntervalList table; the list must be "
                "added before this function is called"
            )
        # reference_electrode_list
        if len(reference_electrode_list) != 1 and len(
            reference_electrode_list
        ) != len(electrode_list):
            raise ValueError(
                "reference_electrode_list must contain either 1 or len(electrode_list) elements"
            )
        # add a -1 element to the list to allow for the no reference option
        available_electrodes = np.append(available_electrodes, [-1])
        if not np.all(np.isin(reference_electrode_list, available_electrodes)):
            raise ValueError(
                "All elements in reference_electrode_list must be valid electrode_ids in the LFPSelection "
                "table"
            )

        # make a list of all the references
        ref_list = np.zeros((len(electrode_list),))
        ref_list[:] = reference_electrode_list

        key = dict()
        key["nwb_file_name"] = nwb_file_name
        key["filter_name"] = filter_name
        key["filter_sampling_rate"] = lfp_sampling_rate
        key["target_interval_list_name"] = interval_list_name
        key["lfp_band_sampling_rate"] = lfp_sampling_rate // decimation
        # insert an entry into the main LFPBandSelectionTable
        self.insert1(key, skip_duplicates=True)

        # get all of the current entries and delete any that are not in the list
        elect_id, ref_id = (self.LFPBandElectrode() & key).fetch(
            "electrode_id", "reference_elect_id"
        )
        for e, r in zip(elect_id, ref_id):
            if not len(np.where((electrode_list == e) & (ref_list == r))[0]):
                key["electrode_id"] = e
                key["reference_elect_id"] = r
                (self.LFPBandElectrode() & key).delete()

        # iterate through all of the new elements and add them
        for e, r in zip(electrode_list, ref_list):
            key["electrode_id"] = e
            query = Electrode & {
                "nwb_file_name": nwb_file_name,
                "electrode_id": e,
            }
            key["electrode_group_name"] = query.fetch1("electrode_group_name")
            key["reference_elect_id"] = r
            self.LFPBandElectrode().insert1(key, skip_duplicates=True)

set_lfp_band_electrodes(nwb_file_name, electrode_list, filter_name, interval_list_name, reference_electrode_list, lfp_band_sampling_rate)

Adds an entry for each electrode in the electrode_list with the specified filter, interval_list, and reference electrode. Also removes any entries that have the same filter, interval list and reference electrode but are not in the electrode_list. :param nwb_file_name: string - the name of the nwb file for the desired session :param electrode_list: list of LFP electrodes to be filtered :param filter_name: the name of the filter (from the FirFilterParameters schema) :param interval_name: the name of the interval list (from the IntervalList schema) :param reference_electrode_list: A single electrode id corresponding to the reference to use for all electrodes or a list with one element per entry in the electrode_list :param lfp_band_sampling_rate: The output sampling rate to be used for the filtered data; must be an integer divisor of the LFP sampling rate :return: none

Source code in src/spyglass/common/common_ephys.py

def set_lfp_band_electrodes(
    self,
    nwb_file_name,
    electrode_list,
    filter_name,
    interval_list_name,
    reference_electrode_list,
    lfp_band_sampling_rate,
):
    """
    Adds an entry for each electrode in the electrode_list with the specified filter, interval_list, and
    reference electrode.
    Also removes any entries that have the same filter, interval list and reference electrode but are not
    in the electrode_list.
    :param nwb_file_name: string - the name of the nwb file for the desired session
    :param electrode_list: list of LFP electrodes to be filtered
    :param filter_name: the name of the filter (from the FirFilterParameters schema)
    :param interval_name: the name of the interval list (from the IntervalList schema)
    :param reference_electrode_list: A single electrode id corresponding to the reference to use for all
    electrodes or a list with one element per entry in the electrode_list
    :param lfp_band_sampling_rate: The output sampling rate to be used for the filtered data; must be an
    integer divisor of the LFP sampling rate
    :return: none
    """
    # Error checks on parameters
    # electrode_list
    query = LFPSelection().LFPElectrode() & {"nwb_file_name": nwb_file_name}
    available_electrodes = query.fetch("electrode_id")
    if not np.all(np.isin(electrode_list, available_electrodes)):
        raise ValueError(
            "All elements in electrode_list must be valid electrode_ids in the LFPSelection table"
        )
    # sampling rate
    lfp_sampling_rate = (LFP() & {"nwb_file_name": nwb_file_name}).fetch1(
        "lfp_sampling_rate"
    )
    decimation = lfp_sampling_rate // lfp_band_sampling_rate
    if lfp_sampling_rate // decimation != lfp_band_sampling_rate:
        raise ValueError(
            f"lfp_band_sampling rate {lfp_band_sampling_rate} is not an integer divisor of lfp "
            f"samping rate {lfp_sampling_rate}"
        )
    # filter
    query = FirFilterParameters() & {
        "filter_name": filter_name,
        "filter_sampling_rate": lfp_sampling_rate,
    }
    if not query:
        raise ValueError(
            f"filter {filter_name}, sampling rate {lfp_sampling_rate} is not in the FirFilterParameters table"
        )
    # interval_list
    query = IntervalList() & {
        "nwb_file_name": nwb_file_name,
        "interval_name": interval_list_name,
    }
    if not query:
        raise ValueError(
            f"interval list {interval_list_name} is not in the IntervalList table; the list must be "
            "added before this function is called"
        )
    # reference_electrode_list
    if len(reference_electrode_list) != 1 and len(
        reference_electrode_list
    ) != len(electrode_list):
        raise ValueError(
            "reference_electrode_list must contain either 1 or len(electrode_list) elements"
        )
    # add a -1 element to the list to allow for the no reference option
    available_electrodes = np.append(available_electrodes, [-1])
    if not np.all(np.isin(reference_electrode_list, available_electrodes)):
        raise ValueError(
            "All elements in reference_electrode_list must be valid electrode_ids in the LFPSelection "
            "table"
        )

    # make a list of all the references
    ref_list = np.zeros((len(electrode_list),))
    ref_list[:] = reference_electrode_list

    key = dict()
    key["nwb_file_name"] = nwb_file_name
    key["filter_name"] = filter_name
    key["filter_sampling_rate"] = lfp_sampling_rate
    key["target_interval_list_name"] = interval_list_name
    key["lfp_band_sampling_rate"] = lfp_sampling_rate // decimation
    # insert an entry into the main LFPBandSelectionTable
    self.insert1(key, skip_duplicates=True)

    # get all of the current entries and delete any that are not in the list
    elect_id, ref_id = (self.LFPBandElectrode() & key).fetch(
        "electrode_id", "reference_elect_id"
    )
    for e, r in zip(elect_id, ref_id):
        if not len(np.where((electrode_list == e) & (ref_list == r))[0]):
            key["electrode_id"] = e
            key["reference_elect_id"] = r
            (self.LFPBandElectrode() & key).delete()

    # iterate through all of the new elements and add them
    for e, r in zip(electrode_list, ref_list):
        key["electrode_id"] = e
        query = Electrode & {
            "nwb_file_name": nwb_file_name,
            "electrode_id": e,
        }
        key["electrode_group_name"] = query.fetch1("electrode_group_name")
        key["reference_elect_id"] = r
        self.LFPBandElectrode().insert1(key, skip_duplicates=True)