common_ephys.py

ElectrodeGroup

Bases: SpyglassMixin, Imported

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

@schema
class ElectrodeGroup(SpyglassMixin, dj.Imported):
    definition = """
    # Grouping of electrodes corresponding to a physical probe.
    -> Session
    electrode_group_name: varchar(80)  # electrode group name from NWBFile
    ---
    -> BrainRegion
    -> [nullable] Probe
    description: varchar(2000)  # description of electrode group
    target_hemisphere = "Unknown": enum("Right", "Left", "Unknown")
    """

    def make(self, key):
        """Make without transaction

        Allows populate_all_common to work within a single transaction."""
        nwb_file_name = key["nwb_file_name"]
        nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)
        for electrode_group in nwbf.electrode_groups.values():
            key["electrode_group_name"] = electrode_group.name
            # add electrode group location if it does not exist, and fetch the row
            key["region_id"] = BrainRegion.fetch_add(
                region_name=electrode_group.location
            )
            if isinstance(electrode_group.device, ndx_franklab_novela.Probe):
                key["probe_id"] = electrode_group.device.probe_type
            key["description"] = electrode_group.description
            if isinstance(
                electrode_group, ndx_franklab_novela.NwbElectrodeGroup
            ):
                # Define target_hemisphere based on targeted x coordinate
                if (
                    electrode_group.targeted_x >= 0
                ):  # if positive or zero x coordinate
                    # define target location as right hemisphere
                    key["target_hemisphere"] = "Right"
                else:  # if negative x coordinate
                    # define target location as left hemisphere
                    key["target_hemisphere"] = "Left"
            self.insert1(key, skip_duplicates=True, allow_direct_insert=True)

make(key)

Make without transaction

Allows populate_all_common to work within a single transaction.

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

def make(self, key):
    """Make without transaction

    Allows populate_all_common to work within a single transaction."""
    nwb_file_name = key["nwb_file_name"]
    nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
    nwbf = get_nwb_file(nwb_file_abspath)
    for electrode_group in nwbf.electrode_groups.values():
        key["electrode_group_name"] = electrode_group.name
        # add electrode group location if it does not exist, and fetch the row
        key["region_id"] = BrainRegion.fetch_add(
            region_name=electrode_group.location
        )
        if isinstance(electrode_group.device, ndx_franklab_novela.Probe):
            key["probe_id"] = electrode_group.device.probe_type
        key["description"] = electrode_group.description
        if isinstance(
            electrode_group, ndx_franklab_novela.NwbElectrodeGroup
        ):
            # Define target_hemisphere based on targeted x coordinate
            if (
                electrode_group.targeted_x >= 0
            ):  # if positive or zero x coordinate
                # define target location as right hemisphere
                key["target_hemisphere"] = "Right"
            else:  # if negative x coordinate
                # define target location as left hemisphere
                key["target_hemisphere"] = "Left"
        self.insert1(key, skip_duplicates=True, allow_direct_insert=True)

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, calling_table=self.camel_name)

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

        electrode_constants = {
            "x_warped": 0,
            "y_warped": 0,
            "z_warped": 0,
            "contacts": "",
        }

        electrode_inserts = []
        electrodes = nwbf.electrodes.to_dataframe()
        for elect_id, elect_data in electrodes.iterrows():
            key.update(
                {
                    "electrode_id": elect_id,
                    "name": str(elect_id),
                    "electrode_group_name": elect_data.group_name,
                    "region_id": BrainRegion.fetch_add(
                        region_name=elect_data.group.location
                    ),
                    "x": elect_data.get("x"),
                    "y": elect_data.get("y"),
                    "z": elect_data.get("z"),
                    "filtering": elect_data.get("filtering", "unfiltered"),
                    "impedance": elect_data.get("imp"),
                    **electrode_constants,
                }
            )

            # 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.update(
                    {
                        "probe_id": elect_data.group.device.probe_type,
                        "probe_shank": elect_data.probe_shank,
                        "probe_electrode": elect_data.probe_electrode,
                        "bad_channel": (
                            "True" if elect_data.bad_channel else "False"
                        ),
                        "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])
            electrode_inserts.append(key.copy())

        self.insert(
            electrode_inserts,
            skip_duplicates=True,
            allow_direct_insert=True,  # for no_transaction, pop_all_common
        )

    @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, calling_table=cls.__name__)
        if "Electrode" not in config:
            return  # See #849

        # 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, calling_table=cls.__name__)
    if "Electrode" not in config:
        return  # See #849

    # 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."
            )

Raw

Bases: SpyglassMixin, Imported

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

@schema
class Raw(SpyglassMixin, dj.Imported):
    definition = """
    # Raw voltage timeseries data, ElectricalSeries in NWB.
    -> Session
    ---
    -> IntervalList
    raw_object_id: varchar(40)      # the NWB object ID for loading this object from the file
    sampling_rate: float            # Sampling rate calculated from data, in Hz
    comments: varchar(2000)
    description: varchar(2000)
    """

    _nwb_table = Nwbfile

    def make(self, key):
        """Make without transaction

        Allows populate_all_common to work within a single transaction."""
        nwb_file_name = key["nwb_file_name"]
        nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)
        raw_interval_name = "raw data valid times"

        # get the acquisition object
        try:
            # TODO this assumes there is a single item in NWBFile.acquisition
            rawdata = nwbf.get_acquisition()
            assert isinstance(rawdata, pynwb.ecephys.ElectricalSeries)
        except (ValueError, AssertionError):
            warnings.warn(
                f"Unable to get acquisition object in: {nwb_file_abspath}\n\t"
                + f"Skipping entry in {self.full_table_name}"
            )
            return

        if rawdata.rate is not None:
            key["sampling_rate"] = rawdata.rate
        else:
            logger.info("Estimating sampling rate...")
            # NOTE: Only use first 1e6 timepoints to save time
            key["sampling_rate"] = estimate_sampling_rate(
                np.asarray(rawdata.timestamps[: int(1e6)]), 1.5, verbose=True
            )

        interval_dict = {
            "nwb_file_name": key["nwb_file_name"],
            "interval_list_name": raw_interval_name,
        }

        if rawdata.rate is not None:
            interval_dict["valid_times"] = np.array(
                [[0, len(rawdata.data) / rawdata.rate]]
            )
        else:
            # get the list of valid times given the specified sampling rate.
            interval_dict["valid_times"] = get_valid_intervals(
                timestamps=np.asarray(rawdata.timestamps),
                sampling_rate=key["sampling_rate"],
                gap_proportion=1.75,
                min_valid_len=0,
            )
        IntervalList().insert1(interval_dict, skip_duplicates=True)

        # now insert each of the electrodes as an individual row, but with the
        # same nwb_object_id

        logger.info(
            f'Importing raw data: Sampling rate:\t{key["sampling_rate"]} Hz\n\t'
            + f'Number of valid intervals:\t{len(interval_dict["valid_times"])}'
        )

        key.update(
            {
                "raw_object_id": rawdata.object_id,
                "interval_list_name": raw_interval_name,
                "comments": rawdata.comments,
                "description": rawdata.description,
            }
        )

        self.insert1(
            key,
            skip_duplicates=True,
            allow_direct_insert=True,
        )

    def nwb_object(self, key):
        # TODO return the nwb_object; FIX: this should be replaced with a fetch
        # call. Note that we're using the raw file so we can modify the other
        # one.

        nwb_file_name = key["nwb_file_name"]
        nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)
        raw_object_id = (self & {"nwb_file_name": key["nwb_file_name"]}).fetch1(
            "raw_object_id"
        )
        return nwbf.objects[raw_object_id]

make(key)

Make without transaction

Allows populate_all_common to work within a single transaction.

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

def make(self, key):
    """Make without transaction

    Allows populate_all_common to work within a single transaction."""
    nwb_file_name = key["nwb_file_name"]
    nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
    nwbf = get_nwb_file(nwb_file_abspath)
    raw_interval_name = "raw data valid times"

    # get the acquisition object
    try:
        # TODO this assumes there is a single item in NWBFile.acquisition
        rawdata = nwbf.get_acquisition()
        assert isinstance(rawdata, pynwb.ecephys.ElectricalSeries)
    except (ValueError, AssertionError):
        warnings.warn(
            f"Unable to get acquisition object in: {nwb_file_abspath}\n\t"
            + f"Skipping entry in {self.full_table_name}"
        )
        return

    if rawdata.rate is not None:
        key["sampling_rate"] = rawdata.rate
    else:
        logger.info("Estimating sampling rate...")
        # NOTE: Only use first 1e6 timepoints to save time
        key["sampling_rate"] = estimate_sampling_rate(
            np.asarray(rawdata.timestamps[: int(1e6)]), 1.5, verbose=True
        )

    interval_dict = {
        "nwb_file_name": key["nwb_file_name"],
        "interval_list_name": raw_interval_name,
    }

    if rawdata.rate is not None:
        interval_dict["valid_times"] = np.array(
            [[0, len(rawdata.data) / rawdata.rate]]
        )
    else:
        # get the list of valid times given the specified sampling rate.
        interval_dict["valid_times"] = get_valid_intervals(
            timestamps=np.asarray(rawdata.timestamps),
            sampling_rate=key["sampling_rate"],
            gap_proportion=1.75,
            min_valid_len=0,
        )
    IntervalList().insert1(interval_dict, skip_duplicates=True)

    # now insert each of the electrodes as an individual row, but with the
    # same nwb_object_id

    logger.info(
        f'Importing raw data: Sampling rate:\t{key["sampling_rate"]} Hz\n\t'
        + f'Number of valid intervals:\t{len(interval_dict["valid_times"])}'
    )

    key.update(
        {
            "raw_object_id": rawdata.object_id,
            "interval_list_name": raw_interval_name,
            "comments": rawdata.comments,
            "description": rawdata.description,
        }
    )

    self.insert1(
        key,
        skip_duplicates=True,
        allow_direct_insert=True,
    )

SampleCount

Bases: SpyglassMixin, Imported

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

@schema
class SampleCount(SpyglassMixin, dj.Imported):
    definition = """
    # Sample count :s timestamp timeseries
    -> Session
    ---
    sample_count_object_id: varchar(40)      # the NWB object ID for loading this object from the file
    """

    _nwb_table = Nwbfile

    def make(self, key):
        """Make without transaction

        Allows populate_all_common to work within a single transaction."""
        nwb_file_name = key["nwb_file_name"]
        nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
        nwbf = get_nwb_file(nwb_file_abspath)
        # get the sample count object
        # TODO: change name when nwb file is changed
        sample_count = get_data_interface(nwbf, "sample_count")
        if sample_count is None:
            logger.info(
                "Unable to import SampleCount: no data interface named "
                + f'"sample_count" found in {nwb_file_name}.'
            )
            return  # see #849
        key["sample_count_object_id"] = sample_count.object_id
        self.insert1(key, allow_direct_insert=True)

make(key)

Make without transaction

Allows populate_all_common to work within a single transaction.

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

def make(self, key):
    """Make without transaction

    Allows populate_all_common to work within a single transaction."""
    nwb_file_name = key["nwb_file_name"]
    nwb_file_abspath = Nwbfile.get_abs_path(nwb_file_name)
    nwbf = get_nwb_file(nwb_file_abspath)
    # get the sample count object
    # TODO: change name when nwb file is changed
    sample_count = get_data_interface(nwbf, "sample_count")
    if sample_count is None:
        logger.info(
            "Unable to import SampleCount: no data interface named "
            + f'"sample_count" found in {nwb_file_name}.'
        )
        return  # see #849
    key["sample_count_object_id"] = sample_count.object_id
    self.insert1(key, allow_direct_insert=True)

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(
            safemode=not test_mode
        )
        # 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(
        safemode=not test_mode
    )
    # 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"sampling 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"sampling 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)