from typing import List, Dict, Any, Tuple, Optional, Union
import numpy as np
from allensdk import one
from allensdk.brain_observatory.behavior.data_files import SyncFile
from allensdk.brain_observatory.behavior.data_files import BehaviorStimulusFile
from allensdk.brain_observatory.behavior.data_objects import StimulusTimestamps
from allensdk.brain_observatory.behavior.data_objects.licks import Licks
from allensdk.brain_observatory.behavior.data_objects.rewards import Rewards
[docs]
class Trial:
def __init__(
self,
trial: dict,
start: float,
end: float,
behavior_stimulus_file: BehaviorStimulusFile,
index: int,
stimulus_timestamps: StimulusTimestamps,
licks: Licks,
rewards: Rewards,
stimuli: dict,
sync_file: Optional[SyncFile] = None):
"""
sync_file is an argument that will be used by
sub-classes that have a more subtle way of handling
monitor delay.
"""
self._trial = trial
self._start = start
self._end = self._calculate_trial_end(
trial_end=end, behavior_stimulus_file=behavior_stimulus_file)
self._index = index
self._stimulus_timestamps = stimulus_timestamps
self._sync_file = sync_file
self._data = self._match_to_sync_timestamps(
raw_stimulus_timestamps=stimulus_timestamps,
licks=licks,
rewards=rewards,
stimuli=stimuli)
@property
def data(self):
return self._data
def _match_to_sync_timestamps(
self,
raw_stimulus_timestamps: StimulusTimestamps,
licks: Licks,
rewards: Rewards,
stimuli: dict) -> Dict[str, Any]:
"""
raw_stimulus_timestamps include monitor_delay
"""
# need to separate out the monitor_delay from the
# un-corrected timestamps
stimulus_timestamps = raw_stimulus_timestamps.subtract_monitor_delay()
event_dict = {
(e[0], e[1]): {
'timestamp': stimulus_timestamps.value[e[3]],
'frame': e[3]} for e in self._trial['events']
}
tr_data = {"trial": self._trial["index"]}
lick_frames = licks.value['frame'].values
timestamps = stimulus_timestamps.value
reward_times = rewards.value['timestamps'].values
# this block of code is trying to mimic
# https://github.com/AllenInstitute/visual_behavior_analysis
# /blob/master/visual_behavior/translator/foraging2
# /stimulus_processing.py
# #L377-L381
# https://github.com/AllenInstitute/visual_behavior_analysis
# /blob/master/visual_behavior/translator/foraging2
# /extract_movies.py#L59-L94
# https://github.com/AllenInstitute/visual_behavior_analysis
# /blob/master/visual_behavior/translator/core/annotate.py#L11-L36
#
# In summary: there are cases where an "epilogue movie" is shown
# after the proper stimuli; we do not want licks that occur
# during this epilogue movie to be counted as belonging to
# the last trial
# https://github.com/AllenInstitute/visual_behavior_analysis
# /issues/482
# select licks that fall between trial_start and trial_end;
# licks on the boundary get assigned to the trial that is ending,
# rather than the trial that is starting
if self._end > 0:
valid_idx = np.where(np.logical_and(lick_frames > self._start,
lick_frames <= self._end))
else:
valid_idx = np.where(lick_frames > self._start)
valid_licks = lick_frames[valid_idx]
if len(valid_licks) > 0:
tr_data["lick_times"] = timestamps[valid_licks]
else:
tr_data["lick_times"] = np.array([], dtype=float)
tr_data["reward_time"] = self._get_reward_time(
reward_times,
event_dict[('trial_start', '')]['timestamp'],
event_dict[('trial_end', '')]['timestamp']
)
tr_data.update(self._get_trial_data())
tr_data.update(self._get_trial_timing(
event_dict,
tr_data['lick_times'],
tr_data['go'],
tr_data['catch'],
tr_data['auto_rewarded'],
tr_data['hit'],
tr_data['false_alarm'],
tr_data["aborted"],
))
tr_data.update(self._get_trial_image_names(stimuli))
self._validate_trial_condition_exclusivity(tr_data=tr_data)
return tr_data
@staticmethod
def _get_reward_time(rebased_reward_times,
start_time,
stop_time) -> float:
"""extract reward times in time range"""
reward_times = rebased_reward_times[np.where(np.logical_and(
rebased_reward_times >= start_time,
rebased_reward_times <= stop_time
))]
return float('nan') if len(reward_times) == 0 else one(
reward_times)
@staticmethod
def _calculate_trial_end(
trial_end,
behavior_stimulus_file: BehaviorStimulusFile) -> int:
if trial_end < 0:
bhv = behavior_stimulus_file.data['items']['behavior']['items']
if 'fingerprint' in bhv.keys():
trial_end = bhv['fingerprint']['starting_frame']
return trial_end
def _get_trial_data(self) -> Dict[str, Any]:
"""
Infer trial logic from trial log. Returns a dictionary.
* reward volume: volume of water delivered on the trial, in mL
Each of the following values is boolean:
Trial category values are mutually exclusive
* go: trial was a go trial (trial with a stimulus change)
* catch: trial was a catch trial (trial with a sham stimulus change)
stimulus_change/sham_change are mutually exclusive
* stimulus_change: did the stimulus change (True on 'go' trials)
* sham_change: stimulus did not change, but response was evaluated
(True on 'catch' trials)
Each trial can be one (and only one) of the following:
* hit (stimulus changed, animal responded in response window)
* miss (stimulus changed, animal did not respond in response window)
* false_alarm (stimulus did not change,
animal responded in response window)
* correct_reject (stimulus did not change,
animal did not respond in response window)
* aborted (animal responded before change time)
* auto_rewarded (reward was automatically delivered following the
change.
This will bias the animals choice and should not be
categorized as hit/miss)
"""
trial_event_names = [val[0] for val in self._trial['events']]
hit = 'hit' in trial_event_names
false_alarm = 'false_alarm' in trial_event_names
miss = 'miss' in trial_event_names
sham_change = 'sham_change' in trial_event_names
stimulus_change = 'stimulus_changed' in trial_event_names
aborted = 'abort' in trial_event_names
if aborted:
go = catch = auto_rewarded = False
else:
catch = self._trial["trial_params"]["catch"] is True
auto_rewarded = self._trial["trial_params"]["auto_reward"]
go = not catch and not auto_rewarded
correct_reject = catch and not false_alarm
if auto_rewarded:
hit = miss = correct_reject = false_alarm = False
return {
"reward_volume": sum([
r[0] for r in self._trial.get("rewards", [])]),
"hit": hit,
"false_alarm": false_alarm,
"miss": miss,
"sham_change": sham_change,
"stimulus_change": stimulus_change,
"aborted": aborted,
"go": go,
"catch": catch,
"auto_rewarded": auto_rewarded,
"correct_reject": correct_reject,
}
def _get_trial_timing(
self,
event_dict: dict,
licks: List[float],
go: bool,
catch: bool,
auto_rewarded: bool,
hit: bool,
false_alarm: bool,
aborted: bool) -> Dict[str, Any]:
"""
Extract a dictionary of trial timing data.
See trial_data_from_log for a description of the trial types.
Parameters
==========
event_dict: dict
Dictionary of trial events in the well-known `pkl` file
licks: List[float]
list of lick timestamps, from the `get_licks` response for
the BehaviorOphysExperiment.api.
go: bool
True if "go" trial, False otherwise. Mutually exclusive with
`catch`.
catch: bool
True if "catch" trial, False otherwise. Mutually exclusive
with `go.`
auto_rewarded: bool
True if "auto_rewarded" trial, False otherwise.
hit: bool
True if "hit" trial, False otherwise
false_alarm: bool
True if "false_alarm" trial, False otherwise
aborted: bool
True if "aborted" trial, False otherwise
Returns
=======
dict
start_time: float
The time the trial started (in seconds elapsed from
recording start)
stop_time: float
The time the trial ended (in seconds elapsed from
recording start)
trial_length: float
Duration of the trial in seconds
response_time: float
The response time, for non-aborted trials. This is equal
to the first lick in the trial. For aborted trials or trials
without licks, `response_time` is NaN.
change_frame: int
The frame number that the stimulus changed
change_time: float
The time in seconds that the stimulus changed
response_latency: float or None
The time in seconds between the stimulus change and the
animal's lick response, if the trial is a "go", "catch", or
"auto_rewarded" type. If the animal did not respond,
return `float("inf")`. In all other cases, return None.
Notes
=====
The following parameters are mutually exclusive (exactly one can
be true):
hit, miss, false_alarm, aborted, auto_rewarded
"""
assert not (aborted and (hit or false_alarm or auto_rewarded)), (
"'aborted' trials cannot be 'hit', 'false_alarm', "
"or 'auto_rewarded'")
assert not (hit and false_alarm), (
"both `hit` and `false_alarm` cannot be True, they are mutually "
"exclusive categories")
assert not (go and catch), (
"both `go` and `catch` cannot be True, they are mutually "
"exclusive "
"categories")
assert not (go and auto_rewarded), (
"both `go` and `auto_rewarded` cannot be True, they are mutually "
"exclusive categories")
def _get_response_time(licks: List[float], aborted: bool) -> float:
"""
Return the time the first lick occurred in a non-"aborted" trial.
A response time is not returned for on an "aborted trial", since by
definition, the animal licked before the change stimulus.
"""
if aborted:
return float("nan")
if len(licks):
return licks[0]
else:
return float("nan")
start_time = event_dict["trial_start", ""]['timestamp']
stop_time = event_dict["trial_end", ""]['timestamp']
response_time = _get_response_time(licks, aborted)
change_frame = self.calculate_change_frame(
event_dict=event_dict,
go=go,
catch=catch,
auto_rewarded=auto_rewarded)
result = {
"start_time": start_time,
"stop_time": stop_time,
"trial_length": stop_time - start_time,
"response_time": response_time,
"change_frame": change_frame
}
result, change_time = self.add_change_time(result)
if not (go or catch or auto_rewarded):
response_latency = None
elif len(licks) > 0:
response_latency = licks[0] - change_time
else:
response_latency = float("inf")
result["response_latency"] = response_latency
return result
[docs]
def calculate_change_frame(
self,
event_dict: dict,
go: bool,
catch: bool,
auto_rewarded: bool) -> Union[int, float]:
"""
Calculate the frame index of a stimulus change
associated with a specific event.
Parameters
----------
event_dict: dict
Dictionary of trial events in the well-known `pkl` file
go: bool
True if "go" trial, False otherwise. Mutually exclusive with
`catch`.
catch: bool
True if "catch" trial, False otherwise. Mutually exclusive
with `go.`
auto_rewarded: bool
True if "auto_rewarded" trial, False otherwise.
Returns
-------
change_frame: Union[int, float]
Index of the change frame; NaN if there is no change
Notes
-----
This is its own method so that child classes of Trial
can implement different logic as needed.
"""
if go or auto_rewarded:
change_frame = event_dict.get(('stimulus_changed', ''))['frame']
elif catch:
change_frame = event_dict.get(('sham_change', ''))['frame']
else:
change_frame = float("nan")
return change_frame
[docs]
def add_change_time(self, trial_dict: dict) -> Tuple[dict, float]:
"""
Add change_time to a dict representing a single trial.
This implementation will just take change_frame and
select the value of self._stimulus_timestamps corresponding
to that frame.
Parameters
----------
trial_dict:
dict containing all trial parameters except
change_time
Returns
-------
trial_dict:
Same as input, except change_time field has been
added
change_time: float
The change time value that was added
(this is returned separately so that child classes have the
option of naming the column something different than
'change_time')
Note
----
Modified trial_dict in-place, in addition to returning it
"""
change_frame = trial_dict['change_frame']
if np.isnan(change_frame):
change_time = np.nan
else:
change_frame = int(change_frame)
change_time = self._stimulus_timestamps.value[change_frame]
trial_dict['change_time'] = change_time
return trial_dict, change_time
def _get_trial_image_names(self, stimuli) -> Dict[str, str]:
"""
Gets the name of the stimulus presented at the beginning of the
trial and
what is it changed to at the end of the trial.
Parameters
----------
stimuli: The stimuli presentation log for the behavior session
Returns
-------
A dictionary indicating the starting_stimulus and what the
stimulus is
changed to.
"""
grating_oris = {'horizontal', 'vertical'}
trial_start_frame = self._trial["events"][0][3]
initial_image_category_name, _, initial_image_name = \
self._resolve_initial_image(
stimuli, trial_start_frame)
if len(self._trial["stimulus_changes"]) == 0:
change_image_name = initial_image_name
else:
((from_set, from_name),
(to_set, to_name),
_, _) = self._trial["stimulus_changes"][0]
# do this to fix names if the stimuli is a grating
if from_set in grating_oris:
from_name = f'gratings_{from_name}'
if to_set in grating_oris:
to_name = f'gratings_{to_name}'
assert from_name == initial_image_name
change_image_name = to_name
return {
"initial_image_name": initial_image_name,
"change_image_name": change_image_name
}
@staticmethod
def _resolve_initial_image(stimuli, start_frame) -> Tuple[str, str, str]:
"""Attempts to resolve the initial image for a given start_frame for
a trial
Parameters
----------
stimuli: Mapping
foraging2 shape stimuli mapping
start_frame: int
start frame of the trial
Returns
-------
initial_image_category_name: str
stimulus category of initial image
initial_image_group: str
group name of the initial image
initial_image_name: str
name of the initial image
"""
max_frame = float("-inf")
initial_image_group = ''
initial_image_name = ''
initial_image_category_name = ''
for stim_category_name, stim_dict in stimuli.items():
for set_event in stim_dict["set_log"]:
set_frame = set_event[3]
if start_frame >= set_frame >= max_frame:
# hack assumes initial_image_group == initial_image_name,
# only initial_image_name is present for natual_scenes
initial_image_group = initial_image_name = set_event[1]
initial_image_category_name = stim_category_name
if initial_image_category_name == 'grating':
initial_image_name = f'gratings_{initial_image_name}'
max_frame = set_frame
return initial_image_category_name, initial_image_group, \
initial_image_name
def _validate_trial_condition_exclusivity(self, tr_data: dict) -> None:
"""ensure that only one of N possible mutually
exclusive trial conditions is True"""
trial_conditions = {}
for key in ['hit',
'miss',
'false_alarm',
'correct_reject',
'auto_rewarded',
'aborted']:
trial_conditions[key] = tr_data[key]
on = []
for condition, value in trial_conditions.items():
if value:
on.append(condition)
if len(on) != 1:
all_conditions = list(trial_conditions.keys())
msg = f"expected exactly 1 trial condition out of " \
f"{all_conditions} "
msg += f"to be True, instead {on} were True (trial {self._index})"
raise AssertionError(msg)