common_task.py

Task

Bases: SpyglassMixin, Manual

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

@schema
class Task(SpyglassMixin, dj.Manual):
    definition = """
     task_name: varchar(80)
     ---
     task_description = NULL: varchar(2000)    # description of this task
     task_type = NULL: varchar(2000)           # type of task
     task_subtype = NULL: varchar(2000)        # subtype of task
     """

    # Tasks are shared across files, so an existing entry generated by
    # TaskEpoch is validated rather than reinserted.
    _expected_duplicates = True

    def insert_from_nwbfile(self, nwbf: pynwb.NWBFile):
        """Insert tasks from an NWB file.

        Parameters
        ----------
        nwbf : pynwb.NWBFile
            The source NWB file object.
        """
        tasks_mod = nwbf.processing.get("tasks")
        if tasks_mod is None:
            logger.warning(f"No tasks processing module found in {nwbf}\n")
            return
        for task in tasks_mod.data_interfaces.values():
            if self.is_nwb_task_table(task):
                self.insert_from_task_table(task)

    def insert_from_task_table(self, task_table: pynwb.core.DynamicTable):
        """Insert tasks from a pynwb DynamicTable containing task metadata.

        Duplicate tasks will check for matching secondary keys and not be added.

        Parameters
        ----------
        task_table : pynwb.core.DynamicTable
            The table representing task metadata.
        """
        taskdf = task_table.to_dataframe()

        task_dicts = taskdf.apply(
            lambda row: dict(
                task_name=row.task_name,
                task_description=row.task_description,
            ),
            axis=1,
        ).tolist()

        # Check if the task is already in the table
        # if so check that the secondary keys all match
        def unequal_vals(key, a, b):
            a, b = a.get(key) or "", b.get(key, "") or ""
            return a != b  # prevent false positive on None != ""

        inserts = []
        for task_dict in task_dicts:
            query = self & {"task_name": task_dict["task_name"]}
            if not query:
                inserts.append(task_dict)  # only append novel tasks
                continue
            existing = query.fetch1()
            for key in set(task_dict).union(existing):
                if not unequal_vals(key, task_dict, existing):
                    continue  # skip if values are equal
                if not accept_divergence(
                    key,
                    task_dict.get(key),
                    existing.get(key),
                    self._test_mode,
                    self.camel_name,
                ):
                    # If the user does not accept the divergence,
                    # raise an error to prevent data inconsistency
                    raise ValueError(
                        f"Task {task_dict['task_name']} already exists "
                        + f"with different values for {key}: "
                        + f"{task_dict.get(key)} != {existing.get(key)}"
                    )
        # Insert the tasks into the table
        self.insert(inserts)

    @classmethod
    def is_nwb_task_table(cls, task_table: pynwb.core.DynamicTable) -> bool:
        """Check format of pynwb DynamicTable containing task metadata.

        The table should be an instance of pynwb.core.DynamicTable and contain
        the columns 'task_name' and 'task_description'.

        Parameters
        ----------
        task_table : pynwb.core.DynamicTable
            The table representing task metadata.

        Returns
        -------
        bool
            Whether the DynamicTable conforms to the expected format for loading
            data into the Task table.
        """
        return (
            isinstance(task_table, pynwb.core.DynamicTable)
            and hasattr(task_table, "task_name")
            and hasattr(task_table, "task_description")
        )

insert_from_nwbfile(nwbf)

Insert tasks from an NWB file.

Parameters:

Name	Type	Description	Default
nwbf	NWBFile	The source NWB file object.	required

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

def insert_from_nwbfile(self, nwbf: pynwb.NWBFile):
    """Insert tasks from an NWB file.

    Parameters
    ----------
    nwbf : pynwb.NWBFile
        The source NWB file object.
    """
    tasks_mod = nwbf.processing.get("tasks")
    if tasks_mod is None:
        logger.warning(f"No tasks processing module found in {nwbf}\n")
        return
    for task in tasks_mod.data_interfaces.values():
        if self.is_nwb_task_table(task):
            self.insert_from_task_table(task)

insert_from_task_table(task_table)

Insert tasks from a pynwb DynamicTable containing task metadata.

Duplicate tasks will check for matching secondary keys and not be added.

Parameters:

Name	Type	Description	Default
task_table	DynamicTable	The table representing task metadata.	required

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

def insert_from_task_table(self, task_table: pynwb.core.DynamicTable):
    """Insert tasks from a pynwb DynamicTable containing task metadata.

    Duplicate tasks will check for matching secondary keys and not be added.

    Parameters
    ----------
    task_table : pynwb.core.DynamicTable
        The table representing task metadata.
    """
    taskdf = task_table.to_dataframe()

    task_dicts = taskdf.apply(
        lambda row: dict(
            task_name=row.task_name,
            task_description=row.task_description,
        ),
        axis=1,
    ).tolist()

    # Check if the task is already in the table
    # if so check that the secondary keys all match
    def unequal_vals(key, a, b):
        a, b = a.get(key) or "", b.get(key, "") or ""
        return a != b  # prevent false positive on None != ""

    inserts = []
    for task_dict in task_dicts:
        query = self & {"task_name": task_dict["task_name"]}
        if not query:
            inserts.append(task_dict)  # only append novel tasks
            continue
        existing = query.fetch1()
        for key in set(task_dict).union(existing):
            if not unequal_vals(key, task_dict, existing):
                continue  # skip if values are equal
            if not accept_divergence(
                key,
                task_dict.get(key),
                existing.get(key),
                self._test_mode,
                self.camel_name,
            ):
                # If the user does not accept the divergence,
                # raise an error to prevent data inconsistency
                raise ValueError(
                    f"Task {task_dict['task_name']} already exists "
                    + f"with different values for {key}: "
                    + f"{task_dict.get(key)} != {existing.get(key)}"
                )
    # Insert the tasks into the table
    self.insert(inserts)

is_nwb_task_table(task_table) classmethod

Check format of pynwb DynamicTable containing task metadata.

The table should be an instance of pynwb.core.DynamicTable and contain the columns 'task_name' and 'task_description'.

Parameters:

Name	Type	Description	Default
task_table	DynamicTable	The table representing task metadata.	required

Returns:

Type	Description
bool	Whether the DynamicTable conforms to the expected format for loading data into the Task table.

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

@classmethod
def is_nwb_task_table(cls, task_table: pynwb.core.DynamicTable) -> bool:
    """Check format of pynwb DynamicTable containing task metadata.

    The table should be an instance of pynwb.core.DynamicTable and contain
    the columns 'task_name' and 'task_description'.

    Parameters
    ----------
    task_table : pynwb.core.DynamicTable
        The table representing task metadata.

    Returns
    -------
    bool
        Whether the DynamicTable conforms to the expected format for loading
        data into the Task table.
    """
    return (
        isinstance(task_table, pynwb.core.DynamicTable)
        and hasattr(task_table, "task_name")
        and hasattr(task_table, "task_description")
    )

TaskEpoch

Bases: SpyglassIngestion, Imported

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

@schema
class TaskEpoch(SpyglassIngestion, dj.Imported):
    # Tasks, session and time intervals
    definition = """
     -> Session
     epoch: int  # the session epoch for this task and apparatus(1 based)
     ---
     -> Task
     -> [nullable] CameraDevice
     -> IntervalList
     task_environment = NULL: varchar(200)  # the environment the animal was in
     camera_names : blob # list of keys corresponding to entry in CameraDevice
     """

    _file_config = dict()  # config for the file being ingested
    _camera_cache = dict()  # nwb_file_name -> {camera id: camera name}
    _interval_cache = dict()  # nwb_file_name -> interval names in IntervalList

    _source_nwb_object_type = pynwb.core.DynamicTable

    # Only the task name comes straight off a row; see the override.
    table_key_to_obj_attr = {"self": {"task_name": "task_name"}}

    def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
        """Hold the caller's config, which `get_nwb_objects` also needs.

        The config is resolved by the caller -- `single_transaction_make`
        merges the file's own `_spyglass_config.yaml` into it. This table does
        not reload it. Kept on `self` only because `get_nwb_objects` and
        `_camera_name_map` take no config argument.
        """
        self._camera_cache, self._interval_cache = dict(), dict()
        self._file_config = config or dict()
        return super().insert_from_nwbfile(nwb_file_name, config, dry_run)

    def get_nwb_objects(self, nwb_file, nwb_file_name=None):
        """Return the file's task tables."""
        tasks_mod = nwb_file.processing.get("tasks")
        task_tables = (
            [
                table
                for table in tasks_mod.data_interfaces.values()
                if self.is_nwb_task_epoch(table)
            ]
            if tasks_mod is not None
            else []
        )

        if not task_tables and not self._file_config.get("Tasks", []):
            self._warn_msg(
                f"No tasks processing module found in {nwb_file} or config\n"
            )
            # Issue #1444: Check for orphaned ImageSeries
            self._check_videos_without_task(nwb_file, nwb_file_name)

        return task_tables

    def _camera_names(self, nwb_file_name) -> dict:
        """Return a file's camera id to name mapping, resolved once.

        Parameters
        ----------
        nwb_file_name : str
            The file being ingested.

        Returns
        -------
        dict
            Camera id to camera name.
        """
        if nwb_file_name not in self._camera_cache:
            self._camera_cache[nwb_file_name] = self._camera_name_map(
                get_nwb_file(Nwbfile().get_abs_path(nwb_file_name))
            )
        return self._camera_cache[nwb_file_name]

    def _camera_name_map(self, nwb_file) -> dict:
        """Map each camera id in the file to its camera name.

        Tasks refer to a camera_id unique within the NWB file, not to the
        CameraDevice primary key, so the id has to be resolved to a name.

        Parameters
        ----------
        nwb_file : pynwb.NWBFile
            The source file.

        Returns
        -------
        dict
            Camera id to camera name, from the file and then the config.
        """
        camera_names = dict()

        for device in nwb_file.devices.values():
            if is_nwb_obj_type(device, "CameraDevice"):
                camera_id = int(str.split(device.name)[1])
                camera_names[camera_id] = device.camera_name

        # Config entries are scalar per device: {camera_id: int,
        # camera_name: str}. The previous implementation zipped the two, which
        # inverted the mapping and raised TypeError on the int
        for device in self._file_config.get("CameraDevice") or []:
            if (camera_id := device.get("camera_id")) is not None:
                camera_names[camera_id] = device.get("camera_name")

        return camera_names

    def _session_intervals(self, nwb_file_name) -> list:
        """Return the interval names already held for a file, fetched once.

        Parameters
        ----------
        nwb_file_name : str
            The file being ingested.

        Returns
        -------
        list of str
            Interval names in IntervalList for this file.
        """
        if nwb_file_name not in self._interval_cache:
            self._interval_cache[nwb_file_name] = (
                IntervalList & {"nwb_file_name": nwb_file_name}
            ).fetch("interval_list_name")
        return self._interval_cache[nwb_file_name]

    def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
        """Generate a Task entry and one TaskEpoch entry per epoch.

        Called once per row of a task table. Task is returned first: it is
        TaskEpoch's parent and has to exist before the epoch rows land.
        """
        entries = super().generate_entries_from_nwb_object(nwb_obj, base_key)

        if hasattr(nwb_obj, "to_dataframe"):
            return entries  # the table itself; rows come back through here

        task_key = entries[self][0]
        nwb_file_name = task_key["nwb_file_name"]

        task_key["camera_names"] = self._get_valid_camera_names(
            nwb_obj.camera_id,
            self._camera_names(nwb_file_name),
            context=f" in NWB file {nwb_file_name}",
        )

        if hasattr(nwb_obj, "task_environment"):
            task_key["task_environment"] = nwb_obj.task_environment

        return {
            # The class, not an instance: entries from several task tables
            # are merged by dict key, and a fresh instance is a fresh key.
            Task: [
                dict(
                    task_name=nwb_obj.task_name,
                    task_description=nwb_obj.task_description,
                )
            ],
            self: self._process_task_epochs(
                task_key,
                nwb_obj.task_epochs,
                nwb_file_name,
                self._session_intervals(nwb_file_name),
            ),
        }

    def generate_entries_from_config(self, config, base_key=None):
        """Generate entries for tasks declared in the config file.

        The config names tasks in its own shape. A `Tasks` list of dicts,
        rather than as ready-made table keys, so this replaces the generic
        config handling rather than extending it.
        """
        base_key = base_key or dict()
        nwb_file_name = base_key.get("nwb_file_name")
        entries = []

        for task in config.get("Tasks", []):
            task_key = dict(
                base_key,
                task_name=task.get("task_name"),
                task_environment=task.get("task_environment", None),
            )
            task_key["camera_names"] = self._get_valid_camera_names(
                task.get("camera_id", []),
                self._camera_names(nwb_file_name),
                context=" in the config file",
            )

            entries.extend(
                self._process_task_epochs(
                    task_key,
                    task.get("task_epochs", []),
                    nwb_file_name,
                    self._session_intervals(nwb_file_name),
                )
            )

        return {self: entries} if entries else dict()

    @classmethod
    def _get_valid_camera_names(cls, camera_ids, camera_names, context=""):
        """Resolve camera IDs to the camera names TaskEpoch stores.

        `camera_names` is a required attribute, so an epoch that names no
        camera gets an empty list rather than no value.

        Parameters
        ----------
        camera_ids : list
            List of camera IDs to validate
        camera_names : dict
            Mapping of camera ID to camera name
        context : str, optional
            Context string for the error message

        Returns
        -------
        list
            List of camera name dicts, empty if no camera was named.

        Raises
        ------
        ValueError
            If a named camera ID has no device in the file or the config. A
            dangling reference is a data error, not something to skip.
        """
        camera_ids = [] if camera_ids is None else list(camera_ids)

        if unresolved := [
            camera_id
            for camera_id in camera_ids
            if camera_id not in camera_names
        ]:
            raise ValueError(
                f"No camera device found with ID {unresolved}{context}. "
                + f"Known camera IDs: {sorted(camera_names)}\n"
            )

        return [
            {"camera_name": camera_names[camera_id]} for camera_id in camera_ids
        ]

    @classmethod
    def _process_task_epochs(
        cls, base_key, task_epochs, nwb_file_name, session_intervals
    ):
        """Process task epochs and create TaskEpoch insert entries.

        Parameters
        ----------
        base_key : dict
            Base key dict with task_name, camera_names, etc.
        task_epochs : list
            List of epoch numbers/identifiers
        nwb_file_name : str
            Name of the NWB file
        session_intervals : list
            Available interval names from IntervalList

        Returns
        -------
        list
            List of dicts ready for TaskEpoch insertion
        """
        inserts = []
        for epoch in task_epochs:
            epoch_key = base_key.copy()
            epoch_key["epoch"] = epoch
            target_interval = cls.get_epoch_interval_name(
                epoch, session_intervals
            )
            if target_interval is None:
                continue
            epoch_key["interval_list_name"] = target_interval
            inserts.append(epoch_key)
        return inserts

    def make(self, key):
        """Deprecated in favor of insert_from_nwbfile."""
        raise NotImplementedError(
            "TaskEpoch.make is deprecated. Use insert_from_nwbfile."
        )

    @classmethod
    def get_epoch_interval_name(cls, epoch, session_intervals):
        """Get the interval name for a given epoch based on matching number.

        This method implements flexible matching to handle various epoch tag
        formats. It tries multiple formats to find a match:
        1. Exact match (e.g., "1")
        2. Two-digit zero-padded (e.g., "01")
        3. Three-digit zero-padded (e.g., "001")
        If multiple matches are found, the two-digit only match is prioritized if
        present. If no unique match is found, a warning is logged.

        Parameters
        ----------
        epoch : int or str
            The epoch number to search for
        session_intervals : list of str
            List of interval names from IntervalList

        Returns
        -------
        str or None
            The matching interval name, or None if no unique match is found

        Examples
        --------
        >>> session_intervals = ["1", "02", "003"]
        >>> TaskEpoch.get_epoch_interval_name(1, session_intervals)
        '1'
        >>> TaskEpoch.get_epoch_interval_name(2, session_intervals)
        '02'
        >>> TaskEpoch.get_epoch_interval_name(3, session_intervals)
        '003'
        """
        if epoch in session_intervals:
            return epoch

        two_digit_matches = [
            interval
            for interval in session_intervals
            if str(epoch).zfill(2) in interval
        ]
        if len(set(two_digit_matches)) == 1:
            return two_digit_matches[0]

        # Try multiple formats:
        possible_formats = [
            str(epoch),  # Try exact match first (e.g., "1")
            str(epoch).zfill(2),  # Try 2-digit zero-pad (e.g., "01")
            str(epoch).zfill(3),  # Try 3-digit zero-pad (e.g., "001")
        ]
        unique_formats = list(dict.fromkeys(possible_formats))

        # Find matches for any format, remove duplicates preserving order
        possible_targets = [
            interval
            for interval in session_intervals
            for target in unique_formats
            if target in interval
        ]

        if len(set(possible_targets)) == 1:
            return possible_targets[0]

        warn = "Multiple" if len(possible_targets) > 1 else "No"

        cls()._warn_msg(
            f"{warn} interval(s) found for epoch {epoch}. "
            f"Available intervals: {session_intervals}"
        )
        return None

    @staticmethod
    def _check_videos_without_task(nwbf, nwb_file_name):
        """Check for ImageSeries when no TaskEpoch entries exist.

        Issue #1444: VideoFile requires TaskEpoch entries.

        Parameters
        ----------
        nwbf : pynwb.NWBFile
            Already-open NWB file object
        nwb_file_name : str
            Name of the NWB file for error messages
        """
        video_names = [
            getattr(obj, "name", None)
            for obj in nwbf.objects.values()
            if isinstance(obj, pynwb.image.ImageSeries)
        ]

        if not video_names:  # No videos in NWB, nothing to warn about
            return

        logger.warning(
            f"{nwb_file_name} TaskEpoch Import Warning (Issue #1444)\n"
            f"Found {len(video_names)} ImageSeries without TaskEpochs:"
            f" {video_names}\n"
            f"VideoFile requires TaskEpoch associations to import videos.\n"
            f"To resolve this:\n"
            f"1. Add task information to your NWB file's processing['tasks']\n"
            f"2. Re-run populate_all_common() after adding task data\n\n"
        )

    @classmethod
    def update_entries(cls, restrict=True):
        """Update entries in the TaskEpoch table based on a restriction."""
        existing_entries = (cls & restrict).fetch("KEY")
        for row in existing_entries:
            if (cls & row).fetch1("camera_names"):
                continue
            row["camera_names"] = [
                {"camera_name": (cls & row).fetch1("camera_name")}
            ]
            cls.update1(row=row)

    @classmethod
    def is_nwb_task_epoch(cls, task_table: pynwb.core.DynamicTable) -> bool:
        """Check format of pynwb DynamicTable containing task metadata.

        The table should be an instance of pynwb.core.DynamicTable and contain
        the columns 'task_name', 'task_description', 'camera_id', 'and
        'task_epochs'.

        Parameters
        ----------
        task_table : pynwb.core.DynamicTable
            The table representing task metadata.

        Returns
        -------
        bool
            Whether the DynamicTable conforms to the expected format for
            loading data into the TaskEpoch table.
        """

        return (
            Task.is_nwb_task_table(task_table)
            and hasattr(task_table, "camera_id")
            and hasattr(task_table, "task_epochs")
        )

insert_from_nwbfile(nwb_file_name, config=None, dry_run=False)

Hold the caller's config, which get_nwb_objects also needs.

The config is resolved by the caller -- single_transaction_make merges the file's own _spyglass_config.yaml into it. This table does not reload it. Kept on self only because get_nwb_objects and _camera_name_map take no config argument.

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

def insert_from_nwbfile(self, nwb_file_name, config=None, dry_run=False):
    """Hold the caller's config, which `get_nwb_objects` also needs.

    The config is resolved by the caller -- `single_transaction_make`
    merges the file's own `_spyglass_config.yaml` into it. This table does
    not reload it. Kept on `self` only because `get_nwb_objects` and
    `_camera_name_map` take no config argument.
    """
    self._camera_cache, self._interval_cache = dict(), dict()
    self._file_config = config or dict()
    return super().insert_from_nwbfile(nwb_file_name, config, dry_run)

get_nwb_objects(nwb_file, nwb_file_name=None)

Return the file's task tables.

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

def get_nwb_objects(self, nwb_file, nwb_file_name=None):
    """Return the file's task tables."""
    tasks_mod = nwb_file.processing.get("tasks")
    task_tables = (
        [
            table
            for table in tasks_mod.data_interfaces.values()
            if self.is_nwb_task_epoch(table)
        ]
        if tasks_mod is not None
        else []
    )

    if not task_tables and not self._file_config.get("Tasks", []):
        self._warn_msg(
            f"No tasks processing module found in {nwb_file} or config\n"
        )
        # Issue #1444: Check for orphaned ImageSeries
        self._check_videos_without_task(nwb_file, nwb_file_name)

    return task_tables

generate_entries_from_nwb_object(nwb_obj, base_key=None)

Generate a Task entry and one TaskEpoch entry per epoch.

Called once per row of a task table. Task is returned first: it is TaskEpoch's parent and has to exist before the epoch rows land.

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

def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
    """Generate a Task entry and one TaskEpoch entry per epoch.

    Called once per row of a task table. Task is returned first: it is
    TaskEpoch's parent and has to exist before the epoch rows land.
    """
    entries = super().generate_entries_from_nwb_object(nwb_obj, base_key)

    if hasattr(nwb_obj, "to_dataframe"):
        return entries  # the table itself; rows come back through here

    task_key = entries[self][0]
    nwb_file_name = task_key["nwb_file_name"]

    task_key["camera_names"] = self._get_valid_camera_names(
        nwb_obj.camera_id,
        self._camera_names(nwb_file_name),
        context=f" in NWB file {nwb_file_name}",
    )

    if hasattr(nwb_obj, "task_environment"):
        task_key["task_environment"] = nwb_obj.task_environment

    return {
        # The class, not an instance: entries from several task tables
        # are merged by dict key, and a fresh instance is a fresh key.
        Task: [
            dict(
                task_name=nwb_obj.task_name,
                task_description=nwb_obj.task_description,
            )
        ],
        self: self._process_task_epochs(
            task_key,
            nwb_obj.task_epochs,
            nwb_file_name,
            self._session_intervals(nwb_file_name),
        ),
    }

generate_entries_from_config(config, base_key=None)

Generate entries for tasks declared in the config file.

The config names tasks in its own shape. A Tasks list of dicts, rather than as ready-made table keys, so this replaces the generic config handling rather than extending it.

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

def generate_entries_from_config(self, config, base_key=None):
    """Generate entries for tasks declared in the config file.

    The config names tasks in its own shape. A `Tasks` list of dicts,
    rather than as ready-made table keys, so this replaces the generic
    config handling rather than extending it.
    """
    base_key = base_key or dict()
    nwb_file_name = base_key.get("nwb_file_name")
    entries = []

    for task in config.get("Tasks", []):
        task_key = dict(
            base_key,
            task_name=task.get("task_name"),
            task_environment=task.get("task_environment", None),
        )
        task_key["camera_names"] = self._get_valid_camera_names(
            task.get("camera_id", []),
            self._camera_names(nwb_file_name),
            context=" in the config file",
        )

        entries.extend(
            self._process_task_epochs(
                task_key,
                task.get("task_epochs", []),
                nwb_file_name,
                self._session_intervals(nwb_file_name),
            )
        )

    return {self: entries} if entries else dict()

make(key)

Deprecated in favor of insert_from_nwbfile.

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

def make(self, key):
    """Deprecated in favor of insert_from_nwbfile."""
    raise NotImplementedError(
        "TaskEpoch.make is deprecated. Use insert_from_nwbfile."
    )

get_epoch_interval_name(epoch, session_intervals) classmethod

Get the interval name for a given epoch based on matching number.

This method implements flexible matching to handle various epoch tag formats. It tries multiple formats to find a match: 1. Exact match (e.g., "1") 2. Two-digit zero-padded (e.g., "01") 3. Three-digit zero-padded (e.g., "001") If multiple matches are found, the two-digit only match is prioritized if present. If no unique match is found, a warning is logged.

Parameters:

Name	Type	Description	Default
epoch	int or str	The epoch number to search for	required
session_intervals	list of str	List of interval names from IntervalList	required

Returns:

Type	Description
str or None	The matching interval name, or None if no unique match is found

Examples:

>>> session_intervals = ["1", "02", "003"]
>>> TaskEpoch.get_epoch_interval_name(1, session_intervals)
'1'
>>> TaskEpoch.get_epoch_interval_name(2, session_intervals)
'02'
>>> TaskEpoch.get_epoch_interval_name(3, session_intervals)
'003'

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

@classmethod
def get_epoch_interval_name(cls, epoch, session_intervals):
    """Get the interval name for a given epoch based on matching number.

    This method implements flexible matching to handle various epoch tag
    formats. It tries multiple formats to find a match:
    1. Exact match (e.g., "1")
    2. Two-digit zero-padded (e.g., "01")
    3. Three-digit zero-padded (e.g., "001")
    If multiple matches are found, the two-digit only match is prioritized if
    present. If no unique match is found, a warning is logged.

    Parameters
    ----------
    epoch : int or str
        The epoch number to search for
    session_intervals : list of str
        List of interval names from IntervalList

    Returns
    -------
    str or None
        The matching interval name, or None if no unique match is found

    Examples
    --------
    >>> session_intervals = ["1", "02", "003"]
    >>> TaskEpoch.get_epoch_interval_name(1, session_intervals)
    '1'
    >>> TaskEpoch.get_epoch_interval_name(2, session_intervals)
    '02'
    >>> TaskEpoch.get_epoch_interval_name(3, session_intervals)
    '003'
    """
    if epoch in session_intervals:
        return epoch

    two_digit_matches = [
        interval
        for interval in session_intervals
        if str(epoch).zfill(2) in interval
    ]
    if len(set(two_digit_matches)) == 1:
        return two_digit_matches[0]

    # Try multiple formats:
    possible_formats = [
        str(epoch),  # Try exact match first (e.g., "1")
        str(epoch).zfill(2),  # Try 2-digit zero-pad (e.g., "01")
        str(epoch).zfill(3),  # Try 3-digit zero-pad (e.g., "001")
    ]
    unique_formats = list(dict.fromkeys(possible_formats))

    # Find matches for any format, remove duplicates preserving order
    possible_targets = [
        interval
        for interval in session_intervals
        for target in unique_formats
        if target in interval
    ]

    if len(set(possible_targets)) == 1:
        return possible_targets[0]

    warn = "Multiple" if len(possible_targets) > 1 else "No"

    cls()._warn_msg(
        f"{warn} interval(s) found for epoch {epoch}. "
        f"Available intervals: {session_intervals}"
    )
    return None

update_entries(restrict=True) classmethod

Update entries in the TaskEpoch table based on a restriction.

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

@classmethod
def update_entries(cls, restrict=True):
    """Update entries in the TaskEpoch table based on a restriction."""
    existing_entries = (cls & restrict).fetch("KEY")
    for row in existing_entries:
        if (cls & row).fetch1("camera_names"):
            continue
        row["camera_names"] = [
            {"camera_name": (cls & row).fetch1("camera_name")}
        ]
        cls.update1(row=row)

is_nwb_task_epoch(task_table) classmethod

Check format of pynwb DynamicTable containing task metadata.

The table should be an instance of pynwb.core.DynamicTable and contain the columns 'task_name', 'task_description', 'camera_id', 'and 'task_epochs'.

Parameters:

Name	Type	Description	Default
task_table	DynamicTable	The table representing task metadata.	required

Returns:

Type	Description
bool	Whether the DynamicTable conforms to the expected format for loading data into the TaskEpoch table.

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

@classmethod
def is_nwb_task_epoch(cls, task_table: pynwb.core.DynamicTable) -> bool:
    """Check format of pynwb DynamicTable containing task metadata.

    The table should be an instance of pynwb.core.DynamicTable and contain
    the columns 'task_name', 'task_description', 'camera_id', 'and
    'task_epochs'.

    Parameters
    ----------
    task_table : pynwb.core.DynamicTable
        The table representing task metadata.

    Returns
    -------
    bool
        Whether the DynamicTable conforms to the expected format for
        loading data into the TaskEpoch table.
    """

    return (
        Task.is_nwb_task_table(task_table)
        and hasattr(task_table, "camera_id")
        and hasattr(task_table, "task_epochs")
    )