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
[docs]
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 ======================
[docs]
@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,
)
[docs]
@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,
)
[docs]
@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,
)
[docs]
@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)
[docs]
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
[docs]
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 ==========================
[docs]
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()
# ====================== 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