group.py

UnitSelectionParams

Bases: SpyglassMixin, Manual

Unit selection parameters for sorted spikes

Attributes:

Name	Type	Description
unit_filter_params_name	str	name of the unit selection parameters
include_labels	(List[str], optional)	list of labels to include, by default None
exclude_labels	(List[str], optional)	list of labels to exclude, by default None
unit_criteria	(dict, optional)	criteria on units table columns the unit must satisfy, by default None. See SortedSpikesGroup.filter_units_by_criteria

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@schema
class UnitSelectionParams(SpyglassMixin, dj.Manual):
    """Unit selection parameters for sorted spikes

    Attributes
    ----------
    unit_filter_params_name : str
        name of the unit selection parameters
    include_labels : List[str], optional
        list of labels to include, by default None
    exclude_labels : List[str], optional
        list of labels to exclude, by default None
    unit_criteria : dict, optional
        criteria on units table columns the unit must satisfy, by default None.
        See `SortedSpikesGroup.filter_units_by_criteria`
    """

    definition = """
    unit_filter_params_name: varchar(32)
    ---
    include_labels = Null: longblob
    exclude_labels = Null: longblob
    unit_criteria = Null: longblob # column -> criterion the unit must satisfy
    """
    # NOTE: pk reduced from 128 to 32 to avoid long primary key error
    contents = [
        {
            "unit_filter_params_name": "all_units",
            "include_labels": [],
            "exclude_labels": [],
        },
        {
            "unit_filter_params_name": "exclude_noise",
            "include_labels": [],
            "exclude_labels": ["noise", "mua"],
        },
        {
            "unit_filter_params_name": "default_exclusion",
            "include_labels": [],
            "exclude_labels": ["noise", "mua"],
        },
    ]

    @classmethod
    def insert_default(cls):
        """Insert default unit selection parameters"""
        cls.insert(cls.contents, skip_duplicates=True)

insert_default() classmethod

Insert default unit selection parameters

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@classmethod
def insert_default(cls):
    """Insert default unit selection parameters"""
    cls.insert(cls.contents, skip_duplicates=True)

SortedSpikesGroup

Bases: SpyglassMixin, Manual

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@schema
class SortedSpikesGroup(SpyglassMixin, dj.Manual):
    definition = """
    -> Session
    -> UnitSelectionParams
    sorted_spikes_group_name: varchar(80)
    """

    class Units(SpyglassMixinPart):
        definition = """
        -> master
        -> SpikeSortingOutput.proj(spikesorting_merge_id='merge_id')
        """

    def create_group(
        self,
        group_name: str,
        nwb_file_name: str,
        unit_filter_params_name: str = "all_units",
        keys: list[dict] = [],
    ):
        """Create a new group of sorted spikes"""
        group_key = {
            "sorted_spikes_group_name": group_name,
            "nwb_file_name": nwb_file_name,
            "unit_filter_params_name": unit_filter_params_name,
        }
        if self & group_key:
            if test_mode:
                return
            raise ValueError(
                f"Group {nwb_file_name}: {group_name} already exists "
                + "please delete the group before creating a new one",
            )

        parts_insert = [{**key, **group_key} for key in keys]

        self.insert1(
            group_key,
            skip_duplicates=True,
        )
        self.Units.insert(parts_insert, skip_duplicates=True)

    @staticmethod
    def filter_units(
        labels: list[list[str]],
        include_labels: list[str],
        exclude_labels: list[str],
    ) -> np.ndarray:
        """
        Filter units based on labels

        labels: list of list of strings
            list of labels for each unit
        include_labels: list of strings
            if provided, only units with any of these labels will be included
        exclude_labels: list of strings
            if provided, units with any of these labels will be excluded
        """
        include_labels = np.unique(include_labels)
        exclude_labels = np.unique(exclude_labels)

        if include_labels.size == 0 and exclude_labels.size == 0:
            # if no labels are provided, include all units
            return np.ones(len(labels), dtype=bool)

        include_mask = np.zeros(len(labels), dtype=bool)
        for ind, unit_labels in enumerate(labels):
            if isinstance(unit_labels, str):
                unit_labels = [unit_labels]
            if (
                include_labels.size > 0
                and np.all(~np.isin(unit_labels, include_labels))
            ) or np.any(np.isin(unit_labels, exclude_labels)):
                # if the unit does not have any of the include labels
                # or has any of the exclude labels, skip
                continue
            include_mask[ind] = True
        return include_mask

    @staticmethod
    def filter_units_by_criteria(
        units_df,
        unit_criteria: Optional[dict] = None,
        strict: bool = True,
    ) -> np.ndarray:
        """Filter units on arbitrary columns of the units table

        Parameters
        ----------
        units_df : pd.DataFrame
            units table of one sorting, one row per unit
        unit_criteria : dict, optional
            column name to {operator: value}, or to a bare value or list of
            values as shorthand for {"isin": value}. A unit is included only if
            it satisfies every criterion. Operators are ">", ">=", "<", "<=",
            "==", "!=", "between" (matches the inclusive [low, high] pair) and
            "outside" (its exact complement), "isin" and "notin". By default
            None, which includes every unit.
        strict : bool, optional
            by default True, raise if a criterion names a column this units
            table does not have. If False, skip that criterion with a warning
            and apply the rest, which lets **every** unit of this sorting pass
            it. `fetch_spike_data` passes False so that it can report every
            sorting in the group missing the column, then raises itself.

        Returns
        -------
        np.ndarray
            boolean mask of shape (n_units,), True for each unit satisfying
            all criteria

        Raises
        ------
        ValueError
            if a criterion holds no operator at all, if an operator is not one
            of those listed above, if "between" or "outside" is given anything
            but a [low, high] pair, if any operator but "isin" or "notin" is
            applied to a column holding a list per unit, or, when strict, if a
            criterion names a column not in units_df

        Notes
        -----
        Units missing a value (NaN, or potentially None from an imported
        units table) fail every criterion on that column, negated ones
        ("!=", "notin") included: a metric that was never computed is no
        evidence that the unit is good. "isin" and "notin" also work on columns
        holding a list per unit (e.g. the curation labels), matching if any
        item of the list is in the target. An empty list is a value, not a
        missing one, so a unit carrying no labels passes "notin".

        Each sorting in a group has its own units table, and those tables may
        not share the same columns (e.g. if they were curated differently), so
        a criterion may apply to only some of them. Passing a criteria column
        that is not in the units table raises an error. Pass strict=False to
        skip criteria on missing columns (passing all units for that criterion)
        with a warning instead of erroring.
        """
        include_mask = np.ones(len(units_df), dtype=bool)

        for column, criterion in (unit_criteria or {}).items():
            if column not in units_df:
                if strict:
                    raise ValueError(
                        f"Unit criteria column '{column}' not in units table. "
                        + f"Columns are {list(units_df.columns)}. Pass "
                        + "strict=False to skip criteria on missing columns."
                    )
                logger.warning(
                    f"Unit criteria column '{column}' not in units table. "
                    + "Skipping this criterion: every unit of this sorting "
                    + "passes it unfiltered."
                )
                continue

            if not isinstance(criterion, dict):  # shorthand for membership
                criterion = {"isin": criterion}
            if not criterion:
                raise ValueError(
                    f"Empty criterion for column '{column}': no operator to "
                    + "apply. Drop the column from the criteria to include "
                    + "every unit."
                )
            values = units_df[column].to_numpy()

            for operator, value in criterion.items():
                if operator not in UNIT_CRITERIA_OPERATORS:
                    raise ValueError(
                        f"Invalid unit criteria operator '{operator}' for "
                        + f"column '{column}'. Expected one of "
                        + f"{list(UNIT_CRITERIA_OPERATORS)}"
                    )
                if operator in RANGE_OPERATORS and (
                    not isinstance(value, (list, tuple, np.ndarray))
                    or len(value) != 2
                ):
                    raise ValueError(
                        f"Unit criteria operator '{operator}' for column "
                        + f"'{column}' takes a [low, high] pair, got "
                        + f"{value!r}."
                    )
                if operator in SCALAR_ONLY_OPERATORS and _is_list_valued(
                    values
                ):
                    raise ValueError(
                        f"Unit criteria operator '{operator}' cannot filter "
                        + f"column '{column}', which holds a list per unit. "
                        + "Comparing a list to the criterion would mask every "
                        + "unit in or out at once. Use 'isin' or 'notin', "
                        + "which match any item of the list."
                    )
                include_mask &= UNIT_CRITERIA_OPERATORS[operator](values, value)

        return include_mask

    @classmethod
    def fetch_spike_data(
        cls,
        key: dict,
        time_slice: Union[list[float], slice] = None,
        return_unit_ids: bool = False,
    ) -> Union[list[np.ndarray], Optional[list[dict]]]:
        """fetch spike times for units in the group

        Parameters
        ----------
        key : dict
            dictionary containing the group key
        time_slice : list of float or slice, optional
            if provided, filter for spikes occurring in the interval
            [start, stop], by default None
        return_unit_ids : bool, optional
            if True, return the unit_ids along with the spike times, by default
            False. Unit ids defined as a list of dictionaries with keys
            'spikesorting_merge_id' and 'unit_number'

        Returns
        -------
        list of np.ndarray
            list of spike times for each unit in the group
        """
        key = cls.get_fully_defined_key(key)

        # get merge_ids for SpikeSortingOutput
        merge_ids = (
            (
                SortedSpikesGroup.Units
                & {
                    "nwb_file_name": key["nwb_file_name"],
                    "sorted_spikes_group_name": key["sorted_spikes_group_name"],
                }
            )
        ).fetch("spikesorting_merge_id")

        # get the filtering parameters. unit_criteria is fetched by `get` so
        # that this still works against a table that has not yet been altered
        # to add it
        filter_params = (UnitSelectionParams & key).fetch1()
        include_labels = filter_params["include_labels"]
        exclude_labels = filter_params["exclude_labels"]
        unit_criteria = filter_params.get("unit_criteria")

        # where each criteria column was, and was not, applied, as criteria
        # column -> merge_id of every sorting whose units table has
        # (applied_to) or lacks (skipped_by) it. Both are needed to tell a
        # column missing everywhere (potentially a typo) from one missing
        # here and there (a result mixing gated and un-gated units), so they
        # are checked once every sorting has been seen
        criteria_columns = set(unit_criteria or {})
        applied_to = {column: [] for column in criteria_columns}
        skipped_by = {column: [] for column in criteria_columns}

        # get the spike times for each merge_id
        spike_times = []
        unit_ids = []
        merge_keys = [dict(merge_id=merge_id) for merge_id in merge_ids]
        nwb_file_list, merge_ids = (SpikeSortingOutput & merge_keys).fetch_nwb(
            return_merge_ids=True
        )
        for nwb_file, merge_id in zip(nwb_file_list, merge_ids):
            nwb_field_name = _get_spike_obj_name(nwb_file, allow_empty=True)

            if nwb_field_name is None:
                logger.warning(f"No spike object found for {merge_id}")
                # case where no units found or curation removed all units
                continue

            units_df = nwb_file[nwb_field_name]
            sorting_spike_times = units_df["spike_times"].to_list()
            file_unit_ids = [
                {"spikesorting_merge_id": merge_id, "unit_id": unit_id}
                for unit_id in range(len(sorting_spike_times))
            ]

            include_unit = np.ones(len(sorting_spike_times), dtype=bool)

            # filter the spike times based on the curation labels if present
            group_col = next(
                (c for c in units_df.columns if c in CURATION_LABEL_COLUMNS),
                None,
            )
            if group_col is not None and not test_mode:
                include_unit &= SortedSpikesGroup.filter_units(
                    units_df[group_col].to_list(),
                    include_labels,
                    exclude_labels,
                )

            # filter on arbitrary criteria over the units table columns
            for column in criteria_columns:
                seen = applied_to if column in units_df else skipped_by
                seen[column].append(merge_id)
            include_unit &= SortedSpikesGroup.filter_units_by_criteria(
                units_df, unit_criteria, strict=False
            )

            if not include_unit.all():
                sorting_spike_times = list(
                    compress(sorting_spike_times, include_unit)
                )
                file_unit_ids = list(compress(file_unit_ids, include_unit))

            # filter the spike times based on the time slice if provided
            if time_slice is not None:
                if isinstance(time_slice, (list, tuple)):
                    time_slice = slice(*time_slice)
                sorting_spike_times = [
                    times[
                        np.logical_and(
                            times >= time_slice.start, times <= time_slice.stop
                        )
                    ]
                    for times in sorting_spike_times
                ]

            # append the approved spike times to the list
            spike_times.extend(sorting_spike_times)
            unit_ids.extend(file_unit_ids)

        if any(skipped_by.values()):
            raise _skipped_criteria_error(skipped_by, applied_to)

        if return_unit_ids:
            return spike_times, unit_ids

        return spike_times

    @classmethod
    def get_spike_indicator(
        cls,
        key: dict,
        time: np.ndarray,
        return_unit_ids: bool = False,
    ) -> np.ndarray:
        """Get spike indicator matrix for the group

        Parameters
        ----------
        key : dict
            key to identify the group
        time : np.ndarray
            time vector for which to calculate the spike indicator matrix
        return_unit_ids : bool, optional
            if True, return the unit ids along with the spike indicator matrix,
            by default False. Unit ids defined as a list of dictionaries with
            keys 'spikesorting_merge_id' and 'unit_number'

        Returns
        -------
        np.ndarray
            spike indicator matrix with shape (len(time), n_units)
        list of dict, optional
            if return_unit_ids is True, returns a list of dictionaries with
            keys 'spikesorting_merge_id' and 'unit_number' for each unit
        """
        time = np.asarray(time)
        min_time, max_time = time[[0, -1]]
        spike_times, unit_ids = cls.fetch_spike_data(key, return_unit_ids=True)

        spike_indicator = np.zeros((len(time), len(spike_times)))

        for ind, times in enumerate(spike_times):
            times = times[np.logical_and(times >= min_time, times <= max_time)]
            spike_indicator[:, ind] = np.bincount(
                np.digitize(times, time[1:-1]),
                minlength=time.shape[0],
            )

        if spike_indicator.ndim == 1:
            spike_indicator = spike_indicator[:, np.newaxis]
        if return_unit_ids:
            return spike_indicator, unit_ids
        return spike_indicator

    @classmethod
    def get_firing_rate(
        cls,
        key: dict,
        time: np.ndarray,
        multiunit: bool = False,
        smoothing_sigma: float = 0.015,
        return_unit_ids: bool = False,
    ) -> np.ndarray:
        """Get time-dependent firing rate for units in the group

        Parameters
        ----------
        key : dict
            key to identify the group
        time : np.ndarray
            time vector for which to calculate the firing rate
        multiunit : bool, optional
            if True, return the multiunit firing rate for units in the group,
            by default False
        smoothing_sigma : float, optional
            standard deviation of gaussian filter to smooth firing rates in
            seconds, by default 0.015
        return_unit_ids : bool, optional
            if True, return the unit ids along with the firing rate, by default
            False. Unit ids defined as a list of dictionaries with keys
            'spikesorting_merge_id' and 'unit_number'

        Returns
        -------
        np.ndarray
            time-dependent firing rate with shape (len(time), n_units)
        list of dict, optional
            if return_unit_ids is True, returns a list of dictionaries with
            keys 'spikesorting_merge_id' and 'unit_number' for each unit
        """
        spike_indicator, unit_ids = cls.get_spike_indicator(
            key, time, return_unit_ids=True
        )
        firing_rate = firing_rate_from_spike_indicator(
            spike_indicator=spike_indicator,
            time=time,
            multiunit=multiunit,
            smoothing_sigma=smoothing_sigma,
        )
        if return_unit_ids:
            return firing_rate, unit_ids
        return firing_rate

create_group(group_name, nwb_file_name, unit_filter_params_name='all_units', keys=[])

Create a new group of sorted spikes

Source code in src/spyglass/spikesorting/analysis/v1/group.py

def create_group(
    self,
    group_name: str,
    nwb_file_name: str,
    unit_filter_params_name: str = "all_units",
    keys: list[dict] = [],
):
    """Create a new group of sorted spikes"""
    group_key = {
        "sorted_spikes_group_name": group_name,
        "nwb_file_name": nwb_file_name,
        "unit_filter_params_name": unit_filter_params_name,
    }
    if self & group_key:
        if test_mode:
            return
        raise ValueError(
            f"Group {nwb_file_name}: {group_name} already exists "
            + "please delete the group before creating a new one",
        )

    parts_insert = [{**key, **group_key} for key in keys]

    self.insert1(
        group_key,
        skip_duplicates=True,
    )
    self.Units.insert(parts_insert, skip_duplicates=True)

filter_units(labels, include_labels, exclude_labels) staticmethod

Filter units based on labels

labels: list of list of strings list of labels for each unit include_labels: list of strings if provided, only units with any of these labels will be included exclude_labels: list of strings if provided, units with any of these labels will be excluded

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@staticmethod
def filter_units(
    labels: list[list[str]],
    include_labels: list[str],
    exclude_labels: list[str],
) -> np.ndarray:
    """
    Filter units based on labels

    labels: list of list of strings
        list of labels for each unit
    include_labels: list of strings
        if provided, only units with any of these labels will be included
    exclude_labels: list of strings
        if provided, units with any of these labels will be excluded
    """
    include_labels = np.unique(include_labels)
    exclude_labels = np.unique(exclude_labels)

    if include_labels.size == 0 and exclude_labels.size == 0:
        # if no labels are provided, include all units
        return np.ones(len(labels), dtype=bool)

    include_mask = np.zeros(len(labels), dtype=bool)
    for ind, unit_labels in enumerate(labels):
        if isinstance(unit_labels, str):
            unit_labels = [unit_labels]
        if (
            include_labels.size > 0
            and np.all(~np.isin(unit_labels, include_labels))
        ) or np.any(np.isin(unit_labels, exclude_labels)):
            # if the unit does not have any of the include labels
            # or has any of the exclude labels, skip
            continue
        include_mask[ind] = True
    return include_mask

filter_units_by_criteria(units_df, unit_criteria=None, strict=True) staticmethod

Filter units on arbitrary columns of the units table

Parameters:

Name	Type	Description	Default
units_df	DataFrame	units table of one sorting, one row per unit	required
unit_criteria	dict	column name to {operator: value}, or to a bare value or list of values as shorthand for {"isin": value}. A unit is included only if it satisfies every criterion. Operators are ">", ">=", "<", "<=", "==", "!=", "between" (matches the inclusive [low, high] pair) and "outside" (its exact complement), "isin" and "notin". By default None, which includes every unit.	None
strict	bool	by default True, raise if a criterion names a column this units table does not have. If False, skip that criterion with a warning and apply the rest, which lets every unit of this sorting pass it. fetch_spike_data passes False so that it can report every sorting in the group missing the column, then raises itself.	True

Returns:

Type	Description
ndarray	boolean mask of shape (n_units,), True for each unit satisfying all criteria

Raises:

Type	Description
ValueError	if a criterion holds no operator at all, if an operator is not one of those listed above, if "between" or "outside" is given anything but a [low, high] pair, if any operator but "isin" or "notin" is applied to a column holding a list per unit, or, when strict, if a criterion names a column not in units_df

Notes

Units missing a value (NaN, or potentially None from an imported units table) fail every criterion on that column, negated ones ("!=", "notin") included: a metric that was never computed is no evidence that the unit is good. "isin" and "notin" also work on columns holding a list per unit (e.g. the curation labels), matching if any item of the list is in the target. An empty list is a value, not a missing one, so a unit carrying no labels passes "notin".

Each sorting in a group has its own units table, and those tables may not share the same columns (e.g. if they were curated differently), so a criterion may apply to only some of them. Passing a criteria column that is not in the units table raises an error. Pass strict=False to skip criteria on missing columns (passing all units for that criterion) with a warning instead of erroring.

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@staticmethod
def filter_units_by_criteria(
    units_df,
    unit_criteria: Optional[dict] = None,
    strict: bool = True,
) -> np.ndarray:
    """Filter units on arbitrary columns of the units table

    Parameters
    ----------
    units_df : pd.DataFrame
        units table of one sorting, one row per unit
    unit_criteria : dict, optional
        column name to {operator: value}, or to a bare value or list of
        values as shorthand for {"isin": value}. A unit is included only if
        it satisfies every criterion. Operators are ">", ">=", "<", "<=",
        "==", "!=", "between" (matches the inclusive [low, high] pair) and
        "outside" (its exact complement), "isin" and "notin". By default
        None, which includes every unit.
    strict : bool, optional
        by default True, raise if a criterion names a column this units
        table does not have. If False, skip that criterion with a warning
        and apply the rest, which lets **every** unit of this sorting pass
        it. `fetch_spike_data` passes False so that it can report every
        sorting in the group missing the column, then raises itself.

    Returns
    -------
    np.ndarray
        boolean mask of shape (n_units,), True for each unit satisfying
        all criteria

    Raises
    ------
    ValueError
        if a criterion holds no operator at all, if an operator is not one
        of those listed above, if "between" or "outside" is given anything
        but a [low, high] pair, if any operator but "isin" or "notin" is
        applied to a column holding a list per unit, or, when strict, if a
        criterion names a column not in units_df

    Notes
    -----
    Units missing a value (NaN, or potentially None from an imported
    units table) fail every criterion on that column, negated ones
    ("!=", "notin") included: a metric that was never computed is no
    evidence that the unit is good. "isin" and "notin" also work on columns
    holding a list per unit (e.g. the curation labels), matching if any
    item of the list is in the target. An empty list is a value, not a
    missing one, so a unit carrying no labels passes "notin".

    Each sorting in a group has its own units table, and those tables may
    not share the same columns (e.g. if they were curated differently), so
    a criterion may apply to only some of them. Passing a criteria column
    that is not in the units table raises an error. Pass strict=False to
    skip criteria on missing columns (passing all units for that criterion)
    with a warning instead of erroring.
    """
    include_mask = np.ones(len(units_df), dtype=bool)

    for column, criterion in (unit_criteria or {}).items():
        if column not in units_df:
            if strict:
                raise ValueError(
                    f"Unit criteria column '{column}' not in units table. "
                    + f"Columns are {list(units_df.columns)}. Pass "
                    + "strict=False to skip criteria on missing columns."
                )
            logger.warning(
                f"Unit criteria column '{column}' not in units table. "
                + "Skipping this criterion: every unit of this sorting "
                + "passes it unfiltered."
            )
            continue

        if not isinstance(criterion, dict):  # shorthand for membership
            criterion = {"isin": criterion}
        if not criterion:
            raise ValueError(
                f"Empty criterion for column '{column}': no operator to "
                + "apply. Drop the column from the criteria to include "
                + "every unit."
            )
        values = units_df[column].to_numpy()

        for operator, value in criterion.items():
            if operator not in UNIT_CRITERIA_OPERATORS:
                raise ValueError(
                    f"Invalid unit criteria operator '{operator}' for "
                    + f"column '{column}'. Expected one of "
                    + f"{list(UNIT_CRITERIA_OPERATORS)}"
                )
            if operator in RANGE_OPERATORS and (
                not isinstance(value, (list, tuple, np.ndarray))
                or len(value) != 2
            ):
                raise ValueError(
                    f"Unit criteria operator '{operator}' for column "
                    + f"'{column}' takes a [low, high] pair, got "
                    + f"{value!r}."
                )
            if operator in SCALAR_ONLY_OPERATORS and _is_list_valued(
                values
            ):
                raise ValueError(
                    f"Unit criteria operator '{operator}' cannot filter "
                    + f"column '{column}', which holds a list per unit. "
                    + "Comparing a list to the criterion would mask every "
                    + "unit in or out at once. Use 'isin' or 'notin', "
                    + "which match any item of the list."
                )
            include_mask &= UNIT_CRITERIA_OPERATORS[operator](values, value)

    return include_mask

fetch_spike_data(key, time_slice=None, return_unit_ids=False) classmethod

fetch spike times for units in the group

Parameters:

Name	Type	Description	Default
key	dict	dictionary containing the group key	required
time_slice	list of float or slice	if provided, filter for spikes occurring in the interval [start, stop], by default None	None
return_unit_ids	bool	if True, return the unit_ids along with the spike times, by default False. Unit ids defined as a list of dictionaries with keys 'spikesorting_merge_id' and 'unit_number'	False

Returns:

Type	Description
list of np.ndarray	list of spike times for each unit in the group

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@classmethod
def fetch_spike_data(
    cls,
    key: dict,
    time_slice: Union[list[float], slice] = None,
    return_unit_ids: bool = False,
) -> Union[list[np.ndarray], Optional[list[dict]]]:
    """fetch spike times for units in the group

    Parameters
    ----------
    key : dict
        dictionary containing the group key
    time_slice : list of float or slice, optional
        if provided, filter for spikes occurring in the interval
        [start, stop], by default None
    return_unit_ids : bool, optional
        if True, return the unit_ids along with the spike times, by default
        False. Unit ids defined as a list of dictionaries with keys
        'spikesorting_merge_id' and 'unit_number'

    Returns
    -------
    list of np.ndarray
        list of spike times for each unit in the group
    """
    key = cls.get_fully_defined_key(key)

    # get merge_ids for SpikeSortingOutput
    merge_ids = (
        (
            SortedSpikesGroup.Units
            & {
                "nwb_file_name": key["nwb_file_name"],
                "sorted_spikes_group_name": key["sorted_spikes_group_name"],
            }
        )
    ).fetch("spikesorting_merge_id")

    # get the filtering parameters. unit_criteria is fetched by `get` so
    # that this still works against a table that has not yet been altered
    # to add it
    filter_params = (UnitSelectionParams & key).fetch1()
    include_labels = filter_params["include_labels"]
    exclude_labels = filter_params["exclude_labels"]
    unit_criteria = filter_params.get("unit_criteria")

    # where each criteria column was, and was not, applied, as criteria
    # column -> merge_id of every sorting whose units table has
    # (applied_to) or lacks (skipped_by) it. Both are needed to tell a
    # column missing everywhere (potentially a typo) from one missing
    # here and there (a result mixing gated and un-gated units), so they
    # are checked once every sorting has been seen
    criteria_columns = set(unit_criteria or {})
    applied_to = {column: [] for column in criteria_columns}
    skipped_by = {column: [] for column in criteria_columns}

    # get the spike times for each merge_id
    spike_times = []
    unit_ids = []
    merge_keys = [dict(merge_id=merge_id) for merge_id in merge_ids]
    nwb_file_list, merge_ids = (SpikeSortingOutput & merge_keys).fetch_nwb(
        return_merge_ids=True
    )
    for nwb_file, merge_id in zip(nwb_file_list, merge_ids):
        nwb_field_name = _get_spike_obj_name(nwb_file, allow_empty=True)

        if nwb_field_name is None:
            logger.warning(f"No spike object found for {merge_id}")
            # case where no units found or curation removed all units
            continue

        units_df = nwb_file[nwb_field_name]
        sorting_spike_times = units_df["spike_times"].to_list()
        file_unit_ids = [
            {"spikesorting_merge_id": merge_id, "unit_id": unit_id}
            for unit_id in range(len(sorting_spike_times))
        ]

        include_unit = np.ones(len(sorting_spike_times), dtype=bool)

        # filter the spike times based on the curation labels if present
        group_col = next(
            (c for c in units_df.columns if c in CURATION_LABEL_COLUMNS),
            None,
        )
        if group_col is not None and not test_mode:
            include_unit &= SortedSpikesGroup.filter_units(
                units_df[group_col].to_list(),
                include_labels,
                exclude_labels,
            )

        # filter on arbitrary criteria over the units table columns
        for column in criteria_columns:
            seen = applied_to if column in units_df else skipped_by
            seen[column].append(merge_id)
        include_unit &= SortedSpikesGroup.filter_units_by_criteria(
            units_df, unit_criteria, strict=False
        )

        if not include_unit.all():
            sorting_spike_times = list(
                compress(sorting_spike_times, include_unit)
            )
            file_unit_ids = list(compress(file_unit_ids, include_unit))

        # filter the spike times based on the time slice if provided
        if time_slice is not None:
            if isinstance(time_slice, (list, tuple)):
                time_slice = slice(*time_slice)
            sorting_spike_times = [
                times[
                    np.logical_and(
                        times >= time_slice.start, times <= time_slice.stop
                    )
                ]
                for times in sorting_spike_times
            ]

        # append the approved spike times to the list
        spike_times.extend(sorting_spike_times)
        unit_ids.extend(file_unit_ids)

    if any(skipped_by.values()):
        raise _skipped_criteria_error(skipped_by, applied_to)

    if return_unit_ids:
        return spike_times, unit_ids

    return spike_times

get_spike_indicator(key, time, return_unit_ids=False) classmethod

Get spike indicator matrix for the group

Parameters:

Name	Type	Description	Default
key	dict	key to identify the group	required
time	ndarray	time vector for which to calculate the spike indicator matrix	required
return_unit_ids	bool	if True, return the unit ids along with the spike indicator matrix, by default False. Unit ids defined as a list of dictionaries with keys 'spikesorting_merge_id' and 'unit_number'	False

Returns:

Type	Description
ndarray	spike indicator matrix with shape (len(time), n_units)
list of dict, optional	if return_unit_ids is True, returns a list of dictionaries with keys 'spikesorting_merge_id' and 'unit_number' for each unit

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@classmethod
def get_spike_indicator(
    cls,
    key: dict,
    time: np.ndarray,
    return_unit_ids: bool = False,
) -> np.ndarray:
    """Get spike indicator matrix for the group

    Parameters
    ----------
    key : dict
        key to identify the group
    time : np.ndarray
        time vector for which to calculate the spike indicator matrix
    return_unit_ids : bool, optional
        if True, return the unit ids along with the spike indicator matrix,
        by default False. Unit ids defined as a list of dictionaries with
        keys 'spikesorting_merge_id' and 'unit_number'

    Returns
    -------
    np.ndarray
        spike indicator matrix with shape (len(time), n_units)
    list of dict, optional
        if return_unit_ids is True, returns a list of dictionaries with
        keys 'spikesorting_merge_id' and 'unit_number' for each unit
    """
    time = np.asarray(time)
    min_time, max_time = time[[0, -1]]
    spike_times, unit_ids = cls.fetch_spike_data(key, return_unit_ids=True)

    spike_indicator = np.zeros((len(time), len(spike_times)))

    for ind, times in enumerate(spike_times):
        times = times[np.logical_and(times >= min_time, times <= max_time)]
        spike_indicator[:, ind] = np.bincount(
            np.digitize(times, time[1:-1]),
            minlength=time.shape[0],
        )

    if spike_indicator.ndim == 1:
        spike_indicator = spike_indicator[:, np.newaxis]
    if return_unit_ids:
        return spike_indicator, unit_ids
    return spike_indicator

get_firing_rate(key, time, multiunit=False, smoothing_sigma=0.015, return_unit_ids=False) classmethod

Get time-dependent firing rate for units in the group

Parameters:

Name	Type	Description	Default
key	dict	key to identify the group	required
time	ndarray	time vector for which to calculate the firing rate	required
multiunit	bool	if True, return the multiunit firing rate for units in the group, by default False	False
smoothing_sigma	float	standard deviation of gaussian filter to smooth firing rates in seconds, by default 0.015	0.015
return_unit_ids	bool	if True, return the unit ids along with the firing rate, by default False. Unit ids defined as a list of dictionaries with keys 'spikesorting_merge_id' and 'unit_number'	False

Returns:

Type	Description
ndarray	time-dependent firing rate with shape (len(time), n_units)
list of dict, optional	if return_unit_ids is True, returns a list of dictionaries with keys 'spikesorting_merge_id' and 'unit_number' for each unit

Source code in src/spyglass/spikesorting/analysis/v1/group.py

@classmethod
def get_firing_rate(
    cls,
    key: dict,
    time: np.ndarray,
    multiunit: bool = False,
    smoothing_sigma: float = 0.015,
    return_unit_ids: bool = False,
) -> np.ndarray:
    """Get time-dependent firing rate for units in the group

    Parameters
    ----------
    key : dict
        key to identify the group
    time : np.ndarray
        time vector for which to calculate the firing rate
    multiunit : bool, optional
        if True, return the multiunit firing rate for units in the group,
        by default False
    smoothing_sigma : float, optional
        standard deviation of gaussian filter to smooth firing rates in
        seconds, by default 0.015
    return_unit_ids : bool, optional
        if True, return the unit ids along with the firing rate, by default
        False. Unit ids defined as a list of dictionaries with keys
        'spikesorting_merge_id' and 'unit_number'

    Returns
    -------
    np.ndarray
        time-dependent firing rate with shape (len(time), n_units)
    list of dict, optional
        if return_unit_ids is True, returns a list of dictionaries with
        keys 'spikesorting_merge_id' and 'unit_number' for each unit
    """
    spike_indicator, unit_ids = cls.get_spike_indicator(
        key, time, return_unit_ids=True
    )
    firing_rate = firing_rate_from_spike_indicator(
        spike_indicator=spike_indicator,
        time=time,
        multiunit=multiunit,
        smoothing_sigma=smoothing_sigma,
    )
    if return_unit_ids:
        return firing_rate, unit_ids
    return firing_rate