behavior_session.py

import datetime
import pathlib
import warnings
from typing import Any, Dict, List, Optional, Type

import numpy as np
import pandas as pd
import pynwb
import pytz
from allensdk import OneResultExpectedError
from allensdk.brain_observatory import sync_utilities
from allensdk.brain_observatory.behavior.data_files import (
    BehaviorStimulusFile,
    MappingStimulusFile,
    ReplayStimulusFile,
    SyncFile,
)
from allensdk.brain_observatory.behavior.data_files.eye_tracking_file import (
    EyeTrackingFile,
)
from allensdk.brain_observatory.behavior.data_files.eye_tracking_metadata_file import (  # noqa: E501
    EyeTrackingMetadataFile,
)
from allensdk.brain_observatory.behavior.data_files.eye_tracking_video import (
    EyeTrackingVideo,
)
from allensdk.brain_observatory.behavior.data_files.stimulus_file import (
    StimulusFileLookup,
    stimulus_lookup_from_json,
)
from allensdk.brain_observatory.behavior.data_objects import (
    BehaviorSessionId,
    RunningAcquisition,
    RunningSpeed,
    StimulusTimestamps,
)
from allensdk.brain_observatory.behavior.data_objects.eye_tracking.eye_tracking_table import (  # noqa: E501
    EyeTrackingTable,
)
from allensdk.brain_observatory.behavior.data_objects.eye_tracking.rig_geometry import (  # noqa: E501
    RigGeometry as EyeTrackingRigGeometry,
)
from allensdk.brain_observatory.behavior.data_objects.licks import Licks
from allensdk.brain_observatory.behavior.data_objects.metadata.behavior_metadata.behavior_metadata import (  # noqa: E501
    BehaviorMetadata,
    get_expt_description,
)
from allensdk.brain_observatory.behavior.data_objects.metadata.behavior_metadata.date_of_acquisition import (  # noqa: E501
    DateOfAcquisition,
)
from allensdk.brain_observatory.behavior.data_objects.metadata.behavior_metadata.project_code import (  # noqa: E501
    ProjectCode,
)
from allensdk.brain_observatory.behavior.data_objects.rewards import Rewards
from allensdk.brain_observatory.behavior.data_objects.stimuli.presentations import (  # noqa: E501
    Presentations,
)
from allensdk.brain_observatory.behavior.data_objects.stimuli.stimuli import (
    Stimuli,
)
from allensdk.brain_observatory.behavior.data_objects.stimuli.templates import (  # noqa: E501
    Templates,
)
from allensdk.brain_observatory.behavior.data_objects.task_parameters import (
    TaskParameters,
)
from allensdk.brain_observatory.behavior.data_objects.trials.trials import (
    Trials,
)
from allensdk.brain_observatory.sync_dataset import Dataset as SyncDataset
from allensdk.core import (
    DataObject,
    JsonReadableInterface,
    LimsReadableInterface,
    NwbReadableInterface,
    NwbWritableInterface,
)
from allensdk.core.auth_config import LIMS_DB_CREDENTIAL_MAP
from allensdk.internal.api import PostgresQueryMixin, db_connection_creator
from pynwb import NWBFile


class BehaviorSession(
    DataObject,
    LimsReadableInterface,
    NwbReadableInterface,
    JsonReadableInterface,
    NwbWritableInterface,
):
    """Represents data from a single Visual Behavior behavior session.
    Initialize by using class methods `from_lims` or `from_nwb_path`.
    """

    def __init__(
        self,
        behavior_session_id: BehaviorSessionId,
        stimulus_timestamps: StimulusTimestamps,
        running_acquisition: RunningAcquisition,
        raw_running_speed: RunningSpeed,
        running_speed: RunningSpeed,
        licks: Licks,
        rewards: Rewards,
        stimuli: Stimuli,
        task_parameters: TaskParameters,
        trials: Trials,
        metadata: BehaviorMetadata,
        date_of_acquisition: DateOfAcquisition,
        eye_tracking_table: Optional[EyeTrackingTable] = None,
        eye_tracking_rig_geometry: Optional[EyeTrackingRigGeometry] = None,
    ):
        super().__init__(
            name="behavior_session", value=None, is_value_self=True
        )

        self._behavior_session_id = behavior_session_id
        self._licks = licks
        self._rewards = rewards
        self._running_acquisition = running_acquisition
        self._running_speed = running_speed
        self._raw_running_speed = raw_running_speed
        self._stimuli = stimuli
        self._stimulus_timestamps = stimulus_timestamps
        self._task_parameters = task_parameters
        self._trials = trials
        self._metadata = metadata
        self._date_of_acquisition = date_of_acquisition
        self._eye_tracking = eye_tracking_table
        self._eye_tracking_rig_geometry = eye_tracking_rig_geometry

    # ==================== class and utility methods ======================

    @classmethod
    def from_json(
        cls,
        session_data: dict,
        read_stimulus_presentations_table_from_file=False,
        stimulus_presentation_columns: Optional[List[str]] = None,
        stimulus_presentation_exclude_columns: Optional[List[str]] = None,
        eye_tracking_z_threshold: float = 3.0,
        eye_tracking_dilation_frames: int = 2,
        eye_tracking_drop_frames: bool = False,
        sync_file_permissive: bool = False,
        running_speed_load_from_multiple_stimulus_files: bool = False,
    ) -> "BehaviorSession":
        """

        Parameters
        ----------
        session_data
            Dict of input data necessary to construct a session
        read_stimulus_presentations_table_from_file
            Whether to read the stimulus table from a file rather than
            construct it here
        stimulus_presentation_columns
            Columns to include in the stimulus presentation table. This also
            specifies the order of the columns.
        stimulus_presentation_exclude_columns
            Optional list of columns to exclude from stimulus presentations
            table
        eye_tracking_z_threshold
            See `BehaviorSession.from_nwb`
        eye_tracking_dilation_frames
            See `BehaviorSession.from_nwb`
        eye_tracking_drop_frames
            See `drop_frames` arg in `allensdk.brain_observatory.behavior.
            data_objects.eye_tracking.eye_tracking_table.EyeTrackingTable.
            from_data_file`
        sync_file_permissive
            See `permissive` arg in `SyncFile` constructor
        running_speed_load_from_multiple_stimulus_files
            Whether to load running speed from multiple stimulus files
            If False, will just load from a single behavior stimulus file

        Returns
        -------
        `BehaviorSession` instance

        """
        if "monitor_delay" not in session_data:
            monitor_delay = cls._get_monitor_delay()
        else:
            monitor_delay = session_data["monitor_delay"]

        behavior_session_id = BehaviorSessionId.from_json(
            dict_repr=session_data
        )

        stimulus_file_lookup = stimulus_lookup_from_json(
            dict_repr=session_data
        )

        if "sync_file" in session_data:
            sync_file = SyncFile.from_json(
                dict_repr=session_data, permissive=sync_file_permissive
            )
        else:
            sync_file = None

        if running_speed_load_from_multiple_stimulus_files:
            running_acquisition = (
                RunningAcquisition.from_multiple_stimulus_files(
                    behavior_stimulus_file=(
                        BehaviorStimulusFile.from_json(dict_repr=session_data)
                    ),
                    mapping_stimulus_file=MappingStimulusFile.from_json(
                        dict_repr=session_data
                    ),
                    replay_stimulus_file=ReplayStimulusFile.from_json(
                        dict_repr=session_data
                    ),
                    sync_file=SyncFile.from_json(dict_repr=session_data),
                )
            )
            raw_running_speed = RunningSpeed.from_multiple_stimulus_files(
                behavior_stimulus_file=(
                    BehaviorStimulusFile.from_json(dict_repr=session_data)
                ),
                mapping_stimulus_file=MappingStimulusFile.from_json(
                    dict_repr=session_data
                ),
                replay_stimulus_file=ReplayStimulusFile.from_json(
                    dict_repr=session_data
                ),
                sync_file=SyncFile.from_json(dict_repr=session_data),
                filtered=False,
            )
            running_speed = RunningSpeed.from_multiple_stimulus_files(
                behavior_stimulus_file=(
                    BehaviorStimulusFile.from_json(dict_repr=session_data)
                ),
                mapping_stimulus_file=MappingStimulusFile.from_json(
                    dict_repr=session_data
                ),
                replay_stimulus_file=ReplayStimulusFile.from_json(
                    dict_repr=session_data
                ),
                sync_file=SyncFile.from_json(dict_repr=session_data),
                filtered=True,
            )
        else:
            behavior_stimulus_file = (
                stimulus_file_lookup.behavior_stimulus_file
            )

            running_acquisition = RunningAcquisition.from_stimulus_file(
                behavior_stimulus_file=behavior_stimulus_file,
                sync_file=sync_file,
            )

            raw_running_speed = RunningSpeed.from_stimulus_file(
                behavior_stimulus_file=behavior_stimulus_file,
                sync_file=sync_file,
                filtered=False,
            )

            running_speed = RunningSpeed.from_stimulus_file(
                behavior_stimulus_file=behavior_stimulus_file,
                sync_file=sync_file,
            )

        metadata = BehaviorMetadata.from_json(dict_repr=session_data)

        (
            stimulus_timestamps,
            licks,
            rewards,
            stimuli,
            task_parameters,
            trials,
        ) = cls._read_data_from_stimulus_file(
            stimulus_file_lookup=stimulus_file_lookup,
            behavior_session_id=behavior_session_id.value,
            sync_file=sync_file,
            monitor_delay=monitor_delay,
            include_stimuli=(not read_stimulus_presentations_table_from_file),
            stimulus_presentation_columns=stimulus_presentation_columns,
        )

        if read_stimulus_presentations_table_from_file:
            stimuli = Stimuli(
                presentations=Presentations.from_path(
                    path=session_data["stim_table_file"],
                    behavior_session_id=session_data["behavior_session_id"],
                    exclude_columns=stimulus_presentation_exclude_columns,
                    trials=trials,
                ),
                templates=Templates.from_stimulus_file(
                    stimulus_file=stimulus_file_lookup.behavior_stimulus_file
                ),
            )
        date_of_acquisition = DateOfAcquisition.from_json(
            dict_repr=session_data
        ).validate(
            stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            behavior_session_id=behavior_session_id.value,
        )

        try:
            eye_tracking_file = EyeTrackingFile.from_json(
                dict_repr=session_data
            )
        except KeyError:
            eye_tracking_file = None

        if eye_tracking_file is None:
            # Return empty data to match what is returned by from_nwb.
            eye_tracking_table = EyeTrackingTable(
                eye_tracking=EyeTrackingTable._get_empty_df()
            )
            eye_tracking_rig_geometry = None
        else:
            try:
                eye_tracking_metadata_file = EyeTrackingMetadataFile.from_json(
                    dict_repr=session_data
                )
            except KeyError:
                eye_tracking_metadata_file = None

            eye_tracking_table = cls._read_eye_tracking_table(
                eye_tracking_file=eye_tracking_file,
                eye_tracking_metadata_file=eye_tracking_metadata_file,
                sync_file=sync_file,
                z_threshold=eye_tracking_z_threshold,
                dilation_frames=eye_tracking_dilation_frames,
            )

            eye_tracking_rig_geometry = EyeTrackingRigGeometry.from_json(
                dict_repr=session_data
            )

        return cls(
            behavior_session_id=behavior_session_id,
            stimulus_timestamps=stimulus_timestamps,
            running_acquisition=running_acquisition,
            raw_running_speed=raw_running_speed,
            running_speed=running_speed,
            metadata=metadata,
            licks=licks,
            rewards=rewards,
            stimuli=stimuli,
            task_parameters=task_parameters,
            trials=trials,
            date_of_acquisition=date_of_acquisition,
            eye_tracking_table=eye_tracking_table,
            eye_tracking_rig_geometry=eye_tracking_rig_geometry,
        )

    @classmethod
    def from_lims(
        cls,
        behavior_session_id: int,
        lims_db: Optional[PostgresQueryMixin] = None,
        sync_file: Optional[SyncFile] = None,
        monitor_delay: Optional[float] = None,
        date_of_acquisition: Optional[DateOfAcquisition] = None,
        eye_tracking_z_threshold: float = 3.0,
        eye_tracking_dilation_frames: int = 2,
    ) -> "BehaviorSession":
        """

        Parameters
        ----------
        behavior_session_id : int
            Behavior session id
        lims_db : PostgresQueryMixin, Optional
            Database connection. If not provided will create a new one.
        sync_file : SyncFile, Optional
            If provided, will be used to compute the stimulus timestamps
            associated with this session. Otherwise, the stimulus timestamps
            will be computed from the stimulus file.
        monitor_delay : float, Optional
            Monitor delay. If not provided, will use an estimate.
            To provide this value, see for example
            allensdk.brain_observatory.behavior.data_objects.stimuli.util.
            calculate_monitor_delay
        date_of_acquisition : DateOfAcquisition, Optional
            Date of acquisition. If not provided, will read from
            behavior_sessions table.
        eye_tracking_z_threshold : float
            See `BehaviorSession.from_nwb`, default 3.0
        eye_tracking_dilation_frames : int
            See `BehaviorSession.from_nwb`, default 2

        Returns
        -------
        `BehaviorSession` instance
        """
        if lims_db is None:
            lims_db = db_connection_creator(
                fallback_credentials=LIMS_DB_CREDENTIAL_MAP
            )

        if monitor_delay is None:
            monitor_delay = cls._get_monitor_delay()

        if sync_file is None:
            try:
                sync_file = SyncFile.from_lims(
                    db=lims_db, behavior_session_id=behavior_session_id
                )
            except OneResultExpectedError:
                sync_file = None

        behavior_session_id = BehaviorSessionId(behavior_session_id)

        stimulus_file_lookup = StimulusFileLookup()

        stimulus_file_lookup.behavior_stimulus_file = (
            BehaviorStimulusFile.from_lims(
                db=lims_db, behavior_session_id=behavior_session_id.value
            )
        )

        running_acquisition = RunningAcquisition.from_stimulus_file(
            behavior_stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            sync_file=sync_file,
        )

        raw_running_speed = RunningSpeed.from_stimulus_file(
            behavior_stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            sync_file=sync_file,
            filtered=False,
        )

        running_speed = RunningSpeed.from_stimulus_file(
            behavior_stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            sync_file=sync_file,
            filtered=True,
        )

        behavior_metadata = BehaviorMetadata.from_lims(
            behavior_session_id=behavior_session_id, lims_db=lims_db
        )

        (
            stimulus_timestamps,
            licks,
            rewards,
            stimuli,
            task_parameters,
            trials,
        ) = cls._read_data_from_stimulus_file(
            behavior_session_id=behavior_session_id.value,
            stimulus_file_lookup=stimulus_file_lookup,
            sync_file=sync_file,
            monitor_delay=monitor_delay,
            project_code=ProjectCode.from_lims(
                behavior_session_id=behavior_session_id.value, lims_db=lims_db
            ),
        )

        if date_of_acquisition is None:
            date_of_acquisition = DateOfAcquisition.from_lims(
                behavior_session_id=behavior_session_id.value, lims_db=lims_db
            )
        date_of_acquisition = date_of_acquisition.validate(
            stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            behavior_session_id=behavior_session_id.value,
        )

        eye_tracking_file = EyeTrackingFile.from_lims(
            db=lims_db, behavior_session_id=behavior_session_id.value
        )

        if eye_tracking_file is None:
            # Return empty data to match what is returned by from_nwb.
            eye_tracking_table = EyeTrackingTable(
                eye_tracking=EyeTrackingTable._get_empty_df()
            )
            eye_tracking_rig_geometry = None
        else:
            eye_tracking_video = EyeTrackingVideo.from_lims(
                db=lims_db, behavior_session_id=behavior_session_id.value
            )

            eye_tracking_metadata_file = None

            eye_tracking_table = cls._read_eye_tracking_table(
                eye_tracking_file=eye_tracking_file,
                eye_tracking_metadata_file=eye_tracking_metadata_file,
                eye_tracking_video=eye_tracking_video,
                sync_file=sync_file,
                z_threshold=eye_tracking_z_threshold,
                dilation_frames=eye_tracking_dilation_frames,
            )

            eye_tracking_rig_geometry = EyeTrackingRigGeometry.from_lims(
                behavior_session_id=behavior_session_id.value, lims_db=lims_db
            )

        return BehaviorSession(
            behavior_session_id=behavior_session_id,
            stimulus_timestamps=stimulus_timestamps,
            running_acquisition=running_acquisition,
            raw_running_speed=raw_running_speed,
            running_speed=running_speed,
            metadata=behavior_metadata,
            licks=licks,
            rewards=rewards,
            stimuli=stimuli,
            task_parameters=task_parameters,
            trials=trials,
            date_of_acquisition=date_of_acquisition,
            eye_tracking_table=eye_tracking_table,
            eye_tracking_rig_geometry=eye_tracking_rig_geometry,
        )

    @classmethod
    def from_nwb(
        cls,
        nwbfile: NWBFile,
        add_is_change_to_stimulus_presentations_table=True,
        eye_tracking_z_threshold: float = 3.0,
        eye_tracking_dilation_frames: int = 2,
    ) -> "BehaviorSession":
        """

        Parameters
        ----------
        nwbfile
        add_is_change_to_stimulus_presentations_table: Whether to add a column
            denoting whether the stimulus presentation represented a change
            event. May not be needed in case this column is precomputed
        eye_tracking_z_threshold : float, optional
            The z-threshold when determining which frames likely contain
            outliers for eye or pupil areas. Influences which frames
            are considered 'likely blinks'. By default 3.0
        eye_tracking_dilation_frames : int, optional
            Determines the number of adjacent frames that will be marked
            as 'likely_blink' when performing blink detection for
            `eye_tracking` data, by default 2

        Returns
        -------

        """
        behavior_session_id = BehaviorSessionId.from_nwb(nwbfile)
        stimulus_timestamps = StimulusTimestamps.from_nwb(nwbfile)
        running_acquisition = RunningAcquisition.from_nwb(nwbfile)
        raw_running_speed = RunningSpeed.from_nwb(nwbfile, filtered=False)
        running_speed = RunningSpeed.from_nwb(nwbfile)
        metadata = BehaviorMetadata.from_nwb(nwbfile)
        licks = Licks.from_nwb(nwbfile=nwbfile)
        rewards = Rewards.from_nwb(nwbfile=nwbfile)
        stimuli = Stimuli.from_nwb(
            nwbfile=nwbfile,
            add_is_change_to_presentations_table=(
                add_is_change_to_stimulus_presentations_table
            ),
        )
        task_parameters = TaskParameters.from_nwb(nwbfile=nwbfile)
        trials = cls._trials_class().from_nwb(nwbfile=nwbfile)
        date_of_acquisition = DateOfAcquisition.from_nwb(nwbfile=nwbfile)

        with warnings.catch_warnings():
            warnings.filterwarnings(
                action="ignore",
                message="This nwb file with identifier ",
                category=UserWarning,
            )
            eye_tracking_rig_geometry = EyeTrackingRigGeometry.from_nwb(
                nwbfile=nwbfile
            )
        with warnings.catch_warnings():
            warnings.filterwarnings(
                action="ignore",
                message="This nwb file with identifier ",
                category=UserWarning,
            )
            eye_tracking_table = EyeTrackingTable.from_nwb(
                nwbfile=nwbfile,
                z_threshold=eye_tracking_z_threshold,
                dilation_frames=eye_tracking_dilation_frames,
            )

        return cls(
            behavior_session_id=behavior_session_id,
            stimulus_timestamps=stimulus_timestamps,
            running_acquisition=running_acquisition,
            raw_running_speed=raw_running_speed,
            running_speed=running_speed,
            metadata=metadata,
            licks=licks,
            rewards=rewards,
            stimuli=stimuli,
            task_parameters=task_parameters,
            trials=trials,
            date_of_acquisition=date_of_acquisition,
            eye_tracking_table=eye_tracking_table,
            eye_tracking_rig_geometry=eye_tracking_rig_geometry,
        )

    @classmethod
    def from_nwb_path(cls, nwb_path: str, **kwargs) -> "BehaviorSession":
        """

        Parameters
        ----------
        nwb_path
            Path to nwb file
        kwargs
            Kwargs to be passed to `from_nwb`

        Returns
        -------
        An instantiation of a `BehaviorSession`
        """
        nwb_path = str(nwb_path)
        with pynwb.NWBHDF5IO(nwb_path, "r", load_namespaces=True) as read_io:
            nwbfile = read_io.read()
            return cls.from_nwb(nwbfile=nwbfile, **kwargs)

    def to_nwb(
        self,
        add_metadata=True,
        include_experiment_description=True,
        stimulus_presentations_stimulus_column_name: str = "stimulus_name",
    ) -> NWBFile:
        """

        Parameters
        ----------
        add_metadata
            Set this to False to prevent adding metadata to the nwb
            instance.
        include_experiment_description: Whether to include a description of the
            experiment in the nwbfile
        stimulus_presentations_stimulus_column_name: Name of the column
            denoting the stimulus name in the presentations table
        """
        if include_experiment_description:
            experiment_description = get_expt_description(
                session_type=self._get_session_type()
            )
        else:
            experiment_description = None

        nwbfile = NWBFile(
            session_description=self._get_session_type(),
            identifier=self._get_identifier(),
            session_start_time=self._date_of_acquisition.value,
            file_create_date=pytz.utc.localize(datetime.datetime.now()),
            institution="Allen Institute for Brain Science",
            keywords=self._get_keywords(),
            experiment_description=experiment_description,
        )

        self._stimulus_timestamps.to_nwb(nwbfile=nwbfile)
        self._running_acquisition.to_nwb(nwbfile=nwbfile)
        self._raw_running_speed.to_nwb(nwbfile=nwbfile)
        self._running_speed.to_nwb(nwbfile=nwbfile)

        if add_metadata:
            self._metadata.to_nwb(nwbfile=nwbfile)

        self._licks.to_nwb(nwbfile=nwbfile)
        self._rewards.to_nwb(nwbfile=nwbfile)
        self._stimuli.to_nwb(
            nwbfile=nwbfile,
            presentations_stimulus_column_name=(
                stimulus_presentations_stimulus_column_name
            ),
        )
        self._task_parameters.to_nwb(nwbfile=nwbfile)
        self._trials.to_nwb(nwbfile=nwbfile)
        if self._eye_tracking is not None:
            self._eye_tracking.to_nwb(nwbfile=nwbfile)
        if self._eye_tracking_rig_geometry is not None:
            self._eye_tracking_rig_geometry.to_nwb(nwbfile=nwbfile)

        return nwbfile

    def list_data_attributes_and_methods(self) -> List[str]:
        """Convenience method for end-users to list attributes and methods
        that can be called to access data for a BehaviorSession.

        NOTE: Because BehaviorOphysExperiment inherits from BehaviorSession,
        this method will also be available there.

        Returns
        -------
        List[str]
            A list of attributes and methods that end-users can access or call
            to get data.
        """
        attrs_and_methods_to_ignore: set = {
            "from_json",
            "from_lims",
            "from_nwb_path",
            "list_data_attributes_and_methods",
        }
        attrs_and_methods_to_ignore.update(dir(NwbReadableInterface))
        attrs_and_methods_to_ignore.update(dir(NwbWritableInterface))
        attrs_and_methods_to_ignore.update(dir(DataObject))
        class_dir = dir(self)
        attrs_and_methods = [
            r
            for r in class_dir
            if (r not in attrs_and_methods_to_ignore and not r.startswith("_"))
        ]
        return attrs_and_methods

    # ========================= 'get' methods ==========================

    def get_reward_rate(self) -> np.ndarray:
        """Get the reward rate of the subject for the task calculated over a
        25 trial rolling window and provides a measure of the rewards
        earned per unit time (in units of rewards/minute).

        Returns
        -------
        np.ndarray
            The reward rate (rewards/minute) of the subject for the
            task calculated over a 25 trial rolling window.
        """
        return self._trials.calculate_reward_rate()

    def get_rolling_performance_df(self) -> pd.DataFrame:
        """Return a DataFrame containing trial by trial behavior response
        performance metrics.

        Returns
        -------
        pd.DataFrame
            A pandas DataFrame containing:
                trials_id [index]: (int)
                    Index of the trial. All trials, including aborted trials,
                    are assigned an index starting at 0 for the first trial.
                reward_rate: (float)
                    Rewards earned in the previous 25 trials, normalized by
                    the elapsed time of the same 25 trials. Units are
                    rewards/minute.
                hit_rate_raw: (float)
                    Fraction of go trials where the mouse licked in the
                    response window, calculated over the previous 100
                    non-aborted trials. Without trial count correction applied.
                hit_rate: (float)
                    Fraction of go trials where the mouse licked in the
                    response window, calculated over the previous 100
                    non-aborted trials. With trial count correction applied.
                false_alarm_rate_raw: (float)
                    Fraction of catch trials where the mouse licked in the
                    response window, calculated over the previous 100
                    non-aborted trials. Without trial count correction applied.
                false_alarm_rate: (float)
                    Fraction of catch trials where the mouse licked in
                    the response window, calculated over the previous 100
                    non-aborted trials. Without trial count correction applied.
                rolling_dprime: (float)
                    d prime calculated using the rolling hit_rate and
                    rolling false_alarm _rate.

        """
        return self._trials.rolling_performance

    def get_performance_metrics(
        self, engaged_trial_reward_rate_threshold: float = 2.0
    ) -> dict:
        """Get a dictionary containing a subject's behavior response
        summary data.

        Parameters
        ----------
        engaged_trial_reward_rate_threshold : float, optional
            The number of rewards per minute that needs to be attained
            before a subject is considered 'engaged', by default 2.0

        Returns
        -------
        dict
            Returns a dict of performance metrics with the following fields:
                trial_count: (int)
                    The length of the trial dataframe
                    (including all 'go', 'catch', and 'aborted' trials)
                go_trial_count: (int)
                    Number of 'go' trials in a behavior session
                catch_trial_count: (int)
                    Number of 'catch' trial types during a behavior session
                hit_trial_count: (int)
                    Number of trials with a hit behavior response
                    type in a behavior session
                miss_trial_count: (int)
                    Number of trials with a miss behavior response
                    type in a behavior session
                false_alarm_trial_count: (int)
                    Number of trials where the mouse had a false alarm
                    behavior response
                correct_reject_trial_count: (int)
                    Number of trials with a correct reject behavior
                    response during a behavior session
                auto_reward_count:
                    Number of trials where the mouse received an auto
                    reward of water.
                earned_reward_count:
                    Number of trials where the mouse was eligible to receive a
                    water reward ('go' trials) and did receive an earned
                    water reward
                total_reward_count:
                    Number of trials where the mouse received a
                    water reward (earned or auto rewarded)
                total_reward_volume: (float)
                    Volume of all water rewards received during a
                    behavior session (earned and auto rewarded)
                maximum_reward_rate: (float)
                    The peak of the rolling reward rate (rewards/minute)
                engaged_trial_count: (int)
                    Number of trials where the mouse is engaged
                    (reward rate > 2 rewards/minute)
                mean_hit_rate: (float)
                    The mean of the rolling hit_rate
                mean_hit_rate_uncorrected:
                    The mean of the rolling hit_rate_raw
                mean_hit_rate_engaged: (float)
                    The mean of the rolling hit_rate, excluding epochs
                    when the rolling reward rate was below 2 rewards/minute
                mean_false_alarm_rate: (float)
                    The mean of the rolling false_alarm_rate, excluding
                    epochs when the rolling reward rate was below 2
                    rewards/minute
                mean_false_alarm_rate_uncorrected: (float)
                    The mean of the rolling false_alarm_rate_raw
                mean_false_alarm_rate_engaged: (float)
                    The mean of the rolling false_alarm_rate,
                    excluding epochs when the rolling reward rate
                    was below 2 rewards/minute
                mean_dprime: (float)
                    The mean of the rolling d_prime
                mean_dprime_engaged: (float)
                    The mean of the rolling d_prime, excluding
                    epochs when the rolling reward rate was
                    below 2 rewards/minute
                max_dprime: (float)
                    The peak of the rolling d_prime
                max_dprime_engaged: (float)
                    The peak of the rolling d_prime, excluding epochs
                    when the rolling reward rate was below 2 rewards/minute
        """
        performance_metrics = {
            "trial_count": self._trials.trial_count,
            "go_trial_count": self._trials.go_trial_count,
            "catch_trial_count": self._trials.catch_trial_count,
            "hit_trial_count": self._trials.hit_trial_count,
            "miss_trial_count": self._trials.miss_trial_count,
            "false_alarm_trial_count": self._trials.false_alarm_trial_count,
            "correct_reject_trial_count": self._trials.correct_reject_trial_count,  # noqa: E501
            "auto_reward_count": self.trials.auto_rewarded.sum(),
            "earned_reward_count": self.trials.hit.sum(),
            "total_reward_count": len(self.rewards),
            "total_reward_volume": self.rewards.volume.sum(),
        }
        # Although 'earned_reward_count' will currently have the same value as
        # 'hit_trial_count', in the future there may be variants of the
        # task where rewards are withheld. In that case the
        # 'earned_reward_count' will be smaller than (and different from)
        # the 'hit_trial_count'.

        rpdf = self.get_rolling_performance_df()
        engaged_trial_mask = (
            rpdf["reward_rate"] > engaged_trial_reward_rate_threshold
        )
        performance_metrics["maximum_reward_rate"] = np.nanmax(
            rpdf["reward_rate"].values
        )
        performance_metrics[
            "engaged_trial_count"
        ] = self._trials.get_engaged_trial_count(
            engaged_trial_reward_rate_threshold=(
                engaged_trial_reward_rate_threshold
            )
        )
        performance_metrics["mean_hit_rate"] = rpdf["hit_rate"].mean()
        performance_metrics["mean_hit_rate_uncorrected"] = rpdf[
            "hit_rate_raw"
        ].mean()
        performance_metrics["mean_hit_rate_engaged"] = rpdf["hit_rate"][
            engaged_trial_mask
        ].mean()
        performance_metrics["mean_false_alarm_rate"] = rpdf[
            "false_alarm_rate"
        ].mean()
        performance_metrics["mean_false_alarm_rate_uncorrected"] = rpdf[
            "false_alarm_rate_raw"
        ].mean()
        performance_metrics["mean_false_alarm_rate_engaged"] = rpdf[
            "false_alarm_rate"
        ][engaged_trial_mask].mean()
        performance_metrics["mean_dprime"] = rpdf["rolling_dprime"].mean()
        performance_metrics["mean_dprime_engaged"] = rpdf["rolling_dprime"][
            engaged_trial_mask
        ].mean()
        performance_metrics["max_dprime"] = rpdf["rolling_dprime"].max()
        performance_metrics["max_dprime_engaged"] = rpdf["rolling_dprime"][
            engaged_trial_mask
        ].max()

        return performance_metrics

    # ====================== properties ========================

    @property
    def behavior_session_id(self) -> int:
        """Unique identifier for a behavioral session.
        :rtype: int
        """
        return self._behavior_session_id.value

    @property
    def eye_tracking(self) -> Optional[pd.DataFrame]:
        """A dataframe containing ellipse fit parameters for the eye, pupil
        and corneal reflection (cr). Fits are derived from tracking points
        from a DeepLabCut model applied to video frames of a subject's
        right eye. Raw tracking points and raw video frames are not exposed
        by the SDK.

        Notes:
        - All columns starting with 'pupil_' represent ellipse fit parameters
          relating to the pupil.
        - All columns starting with 'eye_' represent ellipse fit parameters
          relating to the eyelid.
        - All columns starting with 'cr_' represent ellipse fit parameters
          relating to the corneal reflection, which is caused by an infrared
          LED positioned near the eye tracking camera.
        - All positions are in units of pixels.
        - All areas are in units of pixels^2
        - All values are in the coordinate space of the eye tracking camera,
          NOT the coordinate space of the stimulus display (i.e. this is not
          gaze location), with (0, 0) being the upper-left corner of the
          eye-tracking image.
        - The 'likely_blink' column is True for any row (frame) where the pupil
          fit failed OR eye fit failed OR an outlier fit was identified on the
          pupil or eye fit.
        - The pupil_area, cr_area, eye_area, and pupil/eye_width, height, phi
          columns are set to NaN wherever 'likely_blink' == True.
        - The pupil_area_raw, cr_area_raw, eye_area_raw columns contains all
          pupil fit values (including where 'likely_blink' == True).
        - All ellipse fits are derived from tracking points that were output by
          a DeepLabCut model that was trained on hand-annotated data from a
          subset of imaging sessions on optical physiology rigs.
        - Raw DeepLabCut tracking points are not publicly available.

        :rtype: pandas.DataFrame
        """
        return (
            self._eye_tracking.value
            if self._eye_tracking is not None
            else None
        )

    @property
    def eye_tracking_rig_geometry(self) -> dict:
        """the eye tracking equipment geometry associated with a
        given behavior session.

        Returns
        -------
        dict
            dictionary with the following keys:
                camera_eye_position_mm (array of float)
                camera_rotation_deg (array of float)
                equipment (string)
                led_position (array of float)
                monitor_position_mm (array of float)
                monitor_rotation_deg (array of float)
        """
        if self._eye_tracking_rig_geometry is None:
            return dict()
        return self._eye_tracking_rig_geometry.to_dict()["rig_geometry"]

    @property
    def licks(self) -> pd.DataFrame:
        """A dataframe containing lick timestmaps and frames, sampled
        at 60 Hz.

        NOTE: For BehaviorSessions, returned timestamps are not
        aligned to external 'synchronization' reference timestamps.
        Synchronized timestamps are only available for
        BehaviorOphysExperiments.

        Returns
        -------
        np.ndarray
            A dataframe containing lick timestamps.
            dataframe columns:
                timestamps: (float)
                    time of lick, in seconds
                frame: (int)
                    frame of lick

        """
        return self._licks.value

    @property
    def rewards(self) -> pd.DataFrame:
        """Retrieves rewards from data file saved at the end of the
        behavior session.

        NOTE: For BehaviorSessions, returned timestamps are not
        aligned to external 'synchronization' reference timestamps.
        Synchronized timestamps are only available for
        BehaviorOphysExperiments.

        Returns
        -------
        pd.DataFrame
            A dataframe containing timestamps of delivered rewards.
            Timestamps are sampled at 60Hz.

            dataframe columns:
                volume: (float)
                    volume of individual water reward in ml.
                    0.007 if earned reward, 0.005 if auto reward.
                timestamps: (float)
                    time in seconds
                auto_rewarded: (bool)
                    True if free reward was delivered for that trial.
                    Occurs during the first 5 trials of a session and
                    throughout as needed

        """
        return self._rewards.value

    @property
    def running_speed(self) -> pd.DataFrame:
        """Running speed and timestamps, sampled at 60Hz. By default
        applies a 10Hz low pass filter to the data. To get the
        running speed without the filter, use `raw_running_speed`.

        NOTE: For BehaviorSessions, returned timestamps are not
        aligned to external 'synchronization' reference timestamps.
        Synchronized timestamps are only available for
        BehaviorOphysExperiments.

        Returns
        -------
        pd.DataFrame
            Dataframe containing running speed and timestamps
            dataframe columns:
                timestamps: (float)
                    time in seconds
                speed: (float)
                    speed in cm/sec
        """
        return self._running_speed.value

    @property
    def raw_running_speed(self) -> pd.DataFrame:
        """Get unfiltered running speed data. Sampled at 60Hz.

        NOTE: For BehaviorSessions, returned timestamps are not
        aligned to external 'synchronization' reference timestamps.
        Synchronized timestamps are only available for
        BehaviorOphysExperiments.

        Returns
        -------
        pd.DataFrame
            Dataframe containing unfiltered running speed and timestamps
            dataframe columns:
                timestamps: (float)
                    time in seconds
                speed: (float)
                    speed in cm/sec
        """
        return self._raw_running_speed.value

    @property
    def stimulus_presentations(self) -> pd.DataFrame:
        """Table whose rows are stimulus presentations (e.g. a given image,
        for a given duration, typically 250 ms) and whose columns are
        presentation characteristics.

        For VBO, the stimulus presentations tables now contain data from all
        stimuli not just the active/passive image stimulus. These new stimuli
        are delineated by stimulus_block and stimulus_block_name. The
        previously released image behavior stimulus is accessible as the block
        with a name containing "change_detection". Use the following example
        code snippet to retrieve the original stimulus block from the pandas
        table:
            `stimulus_presentations[
                stimulus_presentations.stimulus_block_name.str.contains(
                    'change_detection'
                )
            ]`

        Returns
        -------
        pd.DataFrame
            Table whose rows are stimulus presentations
            (e.g. a given image, for a given duration, typically 250 ms)
            and whose columns are presentation characteristics.

            dataframe columns:
                stimulus_presentations_id [index]: (int)
                    identifier for a stimulus presentation
                    (presentation of an image)
                stimulus_block: (int)
                    Index of the different stimuli presented to the mouse
                    in order.
                stimulus_block_name [VBO only]: (int)
                    Name of the individual stimulus blocks describing briefly
                    what was shown to the mouse.
                image_index [VBO only]: (int)
                    image index (0-7) for a given session,
                    corresponding to each image name
                image_name: (string)
                    Name of the image presented to the mouse if applicable.
                    This indexes in to the stimulus_templates object.
                movie_frame_index [VBO only]: (int)
                    Index into the natural movie presented. use with the
                    method get_raw_natural_movie or if you need the movie
                    frames stretched and warped to the monitor use
                    get_natural_movie_template. (Warning this is a very compute
                    intensive process).
                duration: (float)
                    duration of an image presentation (flash)
                    in seconds (stop_time - start_time). NaN if omitted
                start_time: (float)
                    image presentation start time in seconds
                end_time: (float)
                    image presentation end time in seconds
                start_frame: (int)
                    image presentation start frame
                end_frame: (int)
                    image presentation end frame
                is_change: (bool)
                    True if the image stimulus has changed. <NA> if not
                    an image stimulus block.
                is_image_novel: (bool)
                    True if the image stimulus is novel. <NA> if not
                    an image stimulus block.
                omitted: (bool)
                    True if no image was shown for this stimulus
                    presentation. <NA> if not an image stimulus block.
                movie_repeat [VBO only]: (int)
                    Number of times the movie has been repeated during the
                    natural movie block. -99 if not a movie block.
                flashes_since_change: (bool)
                    Number of times the image has been flashed in between when
                    it has changed.
                trials_id: (int)
                    Id to match to the table Index of the trials table.
                active: (bool)
                    Stimulus block overlaps with the trials.
                is_sham_change: (bool)
                    True if the stimulus overlaps with a catch trial.
                stimulus_index [VBN only]: (int)
                    0 for gabor_20_deg_250ms, 1 for flash_250ms. -99 otherwise
                position_x [VBN only]: (float)
                    X position of gabor grating when shown. NaN if not
                    applicable.
                position_y [VBN only]: (float)
                    Y position of gabor grating when shown. NaN if not
                    applicable.
                color [VBN only]: (float)
                    Value of the color int flash_250ms stimulus. Either -1 or
                    1. NaN if not applicable.
                rewarded [VBN only]: (bool)
                    Merge of trials data. True if the mouse was rewarded during
                    the stimulus. <NA> if not applicable to stimulus block.
                orientation [VBN only]: (float)
                    Orientation of the gabor grating in degrees.
                temporal_frequency [VBN only]: (float)
                    Temporal frequency of the gabor grating. NaN if not
                    applicable.
                spatial_frequency [VBN only]: (float)
                    Spatial frequency of the gabor grating. NaN if not
                    applicable.
                contrast [VBN only]: float
                    Contrast value of the gabor grating. NaN if not applicable.
        """
        table = self._stimuli.presentations.value
        table = table.drop(columns=["image_set", "index"], errors="ignore")
        table = table.rename(columns={"stop_time": "end_time"})

        return table

    @property
    def stimulus_templates(self) -> Optional[pd.DataFrame]:
        """Get stimulus templates (scenes) for behavior session.

        Returns
        -------
        pd.DataFrame
            A pandas DataFrame object containing the stimulus images for the
            experiment.

            dataframe columns:
                image_name [index]: (string)
                    name of image presented, if 'omitted'
                    then no image was presented
                unwarped: (array of int)
                    image array of unwarped stimulus image
                warped: (array of int)
                    image array of warped stimulus image

        """
        if self._stimuli.templates.image_template_key is not None:
            return self._stimuli.templates.value[
                self._stimuli.templates.image_template_key
            ].to_dataframe()
        else:
            return None

    @property
    def stimulus_timestamps(self) -> np.ndarray:
        """Timestamps associated with the stimulus presetntation on
        the monitor retrieveddata file saved at the end of the
        behavior session. Sampled at 60Hz.

        NOTE: For BehaviorSessions, returned timestamps are not
        aligned to external 'synchronization' reference timestamps.
        Synchronized timestamps are only available for
        BehaviorOphysExperiments.

        Returns
        -------
        np.ndarray
            Timestamps associated with stimulus presentations on the monitor
        """
        return self._stimulus_timestamps.value

    @property
    def task_parameters(self) -> dict:
        """Get task parameters from data file saved at the end of
        the behavior session file.

        Returns
        -------
        dict
            A dictionary containing parameters used to define the task runtime
            behavior.
                auto_reward_volume: (float)
                    Volume of auto rewards in ml.
                blank_duration_sec : (list of floats)
                    Duration in seconds of inter stimulus interval.
                    Inter-stimulus interval chosen as a uniform random value.
                    between the range defined by the two values.
                    Values are ignored if `stimulus_duration_sec` is null.
                response_window_sec: (list of floats)
                    Range of period following an image change, in seconds,
                    where mouse response influences trial outcome.
                    First value represents response window start.
                    Second value represents response window end.
                    Values represent time before display lag is
                    accounted for and applied.
                n_stimulus_frames: (int)
                    Total number of visual stimulus frames presented during
                    a behavior session.
                task: (string)
                    Type of visual stimulus task.
                session_type: (string)
                    Visual stimulus type run during behavior session.
                omitted_flash_fraction: (float)
                    Probability that a stimulus image presentations is omitted.
                    Change stimuli, and the stimulus immediately preceding the
                    change, are never omitted.
                stimulus_distribution: (string)
                    Distribution for drawing change times.
                    Either 'exponential' or 'geometric'.
                stimulus_duration_sec: (float)
                    Duration in seconds of each stimulus image presentation
                reward_volume: (float)
                    Volume of earned water reward in ml.
                stimulus: (string)
                    Stimulus type ('gratings' or 'images').

        """
        return self._task_parameters.to_dict()["task_parameters"]

    @property
    def trials(self) -> pd.DataFrame:
        """Get trials from data file saved at the end of the
        behavior session.

        Returns
        -------
        pd.DataFrame
            A dataframe containing trial and behavioral response data.

            dataframe columns:
                trials_id [index]: (int)
                    trial identifier
                lick_times: (array of float)
                    array of lick times in seconds during that trial.
                    Empty array if no licks occured during the trial.
                reward_time: (NaN or float)
                    Time the reward is delivered following a correct
                    response or on auto rewarded trials.
                reward_volume: (float)
                    volume of reward in ml. 0.005 for auto reward
                    0.007 for earned reward
                hit: (bool)
                    Behavior response type. On catch trial mouse licks
                    within reward window.
                false_alarm: (bool)
                    Behavior response type. On catch trial mouse licks
                    within reward window.
                miss: (bool)
                    Behavior response type. On a go trial, mouse either
                    does not lick at all, or licks after reward window
                is_change: (bool)
                    True if an image change occurs during the trial
                    (if the trial was both a 'go' trial and the trial
                    was not aborted)
                aborted: (bool)
                    Behavior response type. True if the mouse licks
                    before the scheduled change time.
                go: (bool)
                    Trial type. True if there was a change in stimulus
                    image identity on this trial
                catch: (bool)
                    Trial type. True if there was not a change in stimulus
                    identity on this trial
                auto_rewarded: (bool)
                    True if free reward was delivered for that trial.
                    Occurs during the first 5 trials of a session and
                    throughout as needed.
                correct_reject: (bool)
                    Behavior response type. On a catch trial, mouse
                    either does not lick at all or licks after reward
                    window
                start_time: (float)
                    start time of the trial in seconds
                stop_time: (float)
                    end time of the trial in seconds
                trial_length: (float)
                    duration of trial in seconds (stop_time -start_time)
                response_time: (float)
                    time of first lick in trial in seconds and NaN if
                    trial aborted
                initial_image_name: (string)
                    name of image presented at start of trial
                change_image_name: (string)
                    name of image that is changed to at the change time,
                    on go trials
                change_time: (float)
                    Time in the session at which the change occurred.
                response_latency [VBO only]: (float)
                    Delay between change and first lick in seconds. NaN if
                    no change occurred.
                change_frame: (int)
                    Frame in the session at which the change occurred. -99 if
                    no change.
                change_time_no_display_delay [VBN only]: (float)
                    Time of the change in seconds, before the display lag is
                    accounted for and applied.
        """
        return self._trials.data

    @property
    def metadata(self) -> Dict[str, Any]:
        """metadata for a given session

        Returns
        -------
        Dict
            A dictionary containing behavior session specific metadata
            dictionary keys:
                age_in_days: (int)
                    age of mouse in days
                behavior_session_uuid: (int)
                    unique identifier for a behavior session
                behavior_session_id: (int)
                    unique identifier for a behavior session
                cre_line: (string)
                    cre driver line for a transgenic mouse
                date_of_acquisition: (date time object)
                    date and time of experiment acquisition,
                    yyyy-mm-dd hh:mm:ss
                driver_line: (list of string)
                    all driver lines for a transgenic mouse
                equipment_name: (string)
                    identifier for equipment data was collected on
                full_genotype: (string)
                    full genotype of transgenic mouse
                mouse_id: (int)
                    unique identifier for a mouse
                project_code: (string)
                    String of project session is associated with.
                reporter_line: (string)
                    reporter line for a transgenic mouse
                session_type: (string)
                    visual stimulus type displayed during behavior
                    session
                sex: (string)
                    sex of the mouse
                stimulus_frame_rate: (float)
                    frame rate (Hz) at which the visual stimulus is
                    displayed
        """
        return self._get_metadata(behavior_metadata=self._metadata)

    @classmethod
    def _read_licks(
        cls,
        stimulus_file_lookup: StimulusFileLookup,
        sync_file: Optional[SyncFile],
        monitor_delay: float,
    ) -> Licks:
        """
        Construct the Licks data object for this session

        Note: monitor_delay is a part of the call signature so that
        it can be used in sub-class implementations of this method.
        """

        stimulus_timestamps = cls._read_behavior_stimulus_timestamps(
            sync_file=sync_file,
            stimulus_file_lookup=stimulus_file_lookup,
            monitor_delay=0.0,
        )

        return Licks.from_stimulus_file(
            stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            stimulus_timestamps=stimulus_timestamps,
        )

    @classmethod
    def _read_rewards(
        cls,
        stimulus_file_lookup: StimulusFileLookup,
        sync_file: Optional[SyncFile],
    ) -> Rewards:
        """
        Construct the Rewards data object for this session
        """
        stimulus_timestamps = cls._read_behavior_stimulus_timestamps(
            sync_file=sync_file,
            stimulus_file_lookup=stimulus_file_lookup,
            monitor_delay=0.0,
        )

        return Rewards.from_stimulus_file(
            stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            stimulus_timestamps=stimulus_timestamps.subtract_monitor_delay(),
        )

    @classmethod
    def _read_stimuli(
        cls,
        stimulus_file_lookup: StimulusFileLookup,
        behavior_session_id: int,
        sync_file: Optional[SyncFile],
        monitor_delay: float,
        trials: Trials,
        stimulus_presentation_columns: Optional[List[str]] = None,
        project_code: Optional[ProjectCode] = None,
    ) -> Stimuli:
        """
        Construct the Stimuli data object for this session
        """

        stimulus_timestamps = cls._read_behavior_stimulus_timestamps(
            sync_file=sync_file,
            stimulus_file_lookup=stimulus_file_lookup,
            monitor_delay=monitor_delay,
        )

        return Stimuli.from_stimulus_file(
            behavior_session_id=behavior_session_id,
            stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            stimulus_timestamps=stimulus_timestamps,
            presentation_columns=stimulus_presentation_columns,
            project_code=project_code,
            trials=trials,
        )

    @classmethod
    def _read_trials(
        cls,
        stimulus_file_lookup: StimulusFileLookup,
        sync_file: Optional[SyncFile],
        monitor_delay: float,
        licks: Licks,
        rewards: Rewards,
    ) -> Trials:
        """
        Construct the Trials data object for this session
        """

        stimulus_timestamps = cls._read_behavior_stimulus_timestamps(
            sync_file=sync_file,
            stimulus_file_lookup=stimulus_file_lookup,
            monitor_delay=monitor_delay,
        )

        return cls._trials_class().from_stimulus_file(
            stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
            stimulus_timestamps=stimulus_timestamps,
            licks=licks,
            rewards=rewards,
        )

    @classmethod
    def _read_behavior_stimulus_timestamps(
        cls,
        stimulus_file_lookup: StimulusFileLookup,
        sync_file: Optional[SyncFile],
        monitor_delay: float,
    ) -> StimulusTimestamps:
        """
        Assemble the StimulusTimestamps from the SyncFile.
        If a SyncFile is not available, use the
        behavior_stimulus_file
        """
        if sync_file is not None:
            stimulus_timestamps = StimulusTimestamps.from_sync_file(
                sync_file=sync_file, monitor_delay=monitor_delay
            )
        else:
            stimulus_timestamps = StimulusTimestamps.from_stimulus_file(
                stimulus_file=stimulus_file_lookup.behavior_stimulus_file,
                monitor_delay=monitor_delay,
            )

        return stimulus_timestamps

    @classmethod
    def _read_session_timestamps(
        cls,
        stimulus_file_lookup: StimulusFileLookup,
        sync_file: Optional[SyncFile],
        monitor_delay: float,
    ) -> StimulusTimestamps:
        """
        Assemble the StimulusTimestamps (with monitor delay) that will
        be associated with this session
        """
        return cls._read_behavior_stimulus_timestamps(
            stimulus_file_lookup=stimulus_file_lookup,
            sync_file=sync_file,
            monitor_delay=monitor_delay,
        )

    @classmethod
    def _read_data_from_stimulus_file(
        cls,
        stimulus_file_lookup: StimulusFileLookup,
        behavior_session_id: int,
        sync_file: Optional[SyncFile],
        monitor_delay: float,
        include_stimuli: bool = True,
        stimulus_presentation_columns: Optional[List[str]] = None,
        project_code: Optional[ProjectCode] = None,
    ):
        """Helper method to read data from stimulus file"""

        licks = cls._read_licks(
            stimulus_file_lookup=stimulus_file_lookup,
            sync_file=sync_file,
            monitor_delay=monitor_delay,
        )

        rewards = cls._read_rewards(
            stimulus_file_lookup=stimulus_file_lookup, sync_file=sync_file
        )

        session_stimulus_timestamps = cls._read_session_timestamps(
            stimulus_file_lookup=stimulus_file_lookup,
            sync_file=sync_file,
            monitor_delay=monitor_delay,
        )

        trials = cls._read_trials(
            stimulus_file_lookup=stimulus_file_lookup,
            sync_file=sync_file,
            monitor_delay=monitor_delay,
            licks=licks,
            rewards=rewards,
        )

        if include_stimuli:
            stimuli = cls._read_stimuli(
                stimulus_file_lookup=stimulus_file_lookup,
                behavior_session_id=behavior_session_id,
                sync_file=sync_file,
                monitor_delay=monitor_delay,
                trials=trials,
                stimulus_presentation_columns=stimulus_presentation_columns,
                project_code=project_code,
            )
        else:
            stimuli = None

        task_parameters = TaskParameters.from_stimulus_file(
            stimulus_file=stimulus_file_lookup.behavior_stimulus_file
        )

        return (
            session_stimulus_timestamps.subtract_monitor_delay(),
            licks,
            rewards,
            stimuli,
            task_parameters,
            trials,
        )

    @classmethod
    def _read_eye_tracking_table(
        cls,
        eye_tracking_file: EyeTrackingFile,
        sync_file: SyncFile,
        z_threshold: float,
        dilation_frames: int,
        eye_tracking_metadata_file: Optional[EyeTrackingMetadataFile] = None,
        eye_tracking_video: Optional[EyeTrackingVideo] = None,
    ) -> EyeTrackingTable:
        # this is possible if instantiating from_lims
        if sync_file is None:
            msg = (
                "sync_file is None for this session; "
                "do not know how to create an eye tracking "
                "table without a sync_file"
            )
            raise RuntimeError(msg)

        sync_path = pathlib.Path(sync_file.filepath)

        frame_times = sync_utilities.get_synchronized_frame_times(
            session_sync_file=sync_path,
            sync_line_label_keys=SyncDataset.EYE_TRACKING_KEYS,
            drop_frames=None,
            trim_after_spike=False,
        )

        stimulus_timestamps = StimulusTimestamps(
            timestamps=frame_times.to_numpy(), monitor_delay=0.0
        )

        return EyeTrackingTable.from_data_file(
            data_file=eye_tracking_file,
            metadata_file=eye_tracking_metadata_file,
            video=eye_tracking_video,
            stimulus_timestamps=stimulus_timestamps,
            z_threshold=z_threshold,
            dilation_frames=dilation_frames,
            empty_on_fail=True,
        )

    def _get_metadata(self, behavior_metadata: BehaviorMetadata) -> dict:
        """Returns dict of metadata"""
        return {
            "equipment_name": behavior_metadata.equipment.value,
            "sex": behavior_metadata.subject_metadata.sex,
            "age_in_days": behavior_metadata.subject_metadata.age_in_days,
            "stimulus_frame_rate": behavior_metadata.stimulus_frame_rate,
            "session_type": behavior_metadata.session_type,
            "date_of_acquisition": self._date_of_acquisition.value,
            "reporter_line": behavior_metadata.subject_metadata.reporter_line,
            "cre_line": behavior_metadata.subject_metadata.cre_line,
            "behavior_session_uuid": behavior_metadata.behavior_session_uuid,
            "driver_line": behavior_metadata.subject_metadata.driver_line,
            "mouse_id": behavior_metadata.subject_metadata.mouse_id,
            "project_code": behavior_metadata.project_code,
            "full_genotype": behavior_metadata.subject_metadata.full_genotype,
            "behavior_session_id": behavior_metadata.behavior_session_id,
        }

    def _get_identifier(self) -> str:
        return str(self._behavior_session_id.value)

    def _get_session_type(self) -> str:
        return self._metadata.session_type

    @staticmethod
    def _get_keywords():
        """Keywords for NWB file"""
        return ["visual", "behavior", "task"]

    @staticmethod
    def _get_monitor_delay():
        # This is the median estimate across all rigs
        # as discussed in
        # https://github.com/AllenInstitute/AllenSDK/issues/1318
        return 0.02115

    @classmethod
    def _trials_class(cls) -> Type[Trials]:
        return Trials