from marshmallow import Schema, fields
from datetime import datetime, date
import numpy as np
[docs]def annotate_change_detect(trials):
""" adds `change` and `detect` columns to dataframe
Parameters
----------
trials : pandas DataFrame
dataframe of trials
inplace : bool, optional
modify `trials` in place. if False, returns a copy. default: True
See Also
--------
io.load_trials
"""
trials['change'] = trials['trial_type'] == 'go'
trials['detect'] = trials['response'] == 1.0
return trials
[docs]def assign_session_id(trials):
""" adds a column with a unique ID for the session defined as
a combination of the mouse ID and startdatetime
Parameters
----------
trials : pandas DataFrame
dataframe of trials
inplace : bool, optional
modify `trials` in place. if False, returns a copy. default: True
See Also
--------
io.load_trials
"""
trials['session_id'] = (trials['mouse_id']
+ '_'
+ trials['startdatetime'].map(
lambda x: x.isoformat()))
return trials
[docs]def fix_change_time(trials):
""" forces `None` values in the `change_time` column to numpy NaN
Parameters
----------
trials : pandas DataFrame
dataframe of trials
inplace : bool, optional
modify `trials` in place. if False, returns a copy. default: True
See Also
--------
io.load_trials
"""
trials['change_time'] = trials['change_time'].map(
lambda x: np.nan if x is None else x)
return trials
[docs]def explode_response_window(trials):
""" explodes the `response_window` column in lower & upper columns
Parameters
----------
trials : pandas DataFrame
dataframe of trials
inplace : bool, optional
modify `trials` in place. if False, returns a copy. default: True
See Also
--------
io.load_trials
"""
trials['response_window_lower'] = \
trials['response_window'].map(lambda x: x[0])
trials['response_window_upper'] = \
trials['response_window'].map(lambda x: x[1])
return trials
[docs]def annotate_trials(trials):
""" performs multiple annotatations:
- annotate_change_detect
- fix_change_time
- explode_response_window
Parameters
----------
trials : pandas DataFrame
dataframe of trials
inplace : bool, optional
modify `trials` in place. if False, returns a copy. default: True
See Also
--------
io.load_trials
"""
# build arrays for change detection
trials = annotate_change_detect(trials)
# assign a session ID to each row
trials = assign_session_id(trials)
# calculate reaction times
trials = fix_change_time(trials)
# unwrap the response window
trials = explode_response_window(trials)
return trials
[docs]class FriendlyDateTime(fields.DateTime):
def _deserialize(self, value, attr, data, **kwargs):
if isinstance(value, datetime):
return value
result = super(FriendlyDateTime, self)._deserialize(value, attr, data)
return result
[docs]class FriendlyDate(fields.Date):
def _deserialize(self, value, attr, data, **kwargs):
if isinstance(value, date):
return value
result = super(FriendlyDate, self)._deserialize(value, attr, data)
return result
[docs]class ExtendedTrialSchema(Schema):
"""
This schema describes the edf core trial structure
"""
index = fields.Int(
description='Trial number in this session',
required=True,
)
startframe = fields.Int(
description='frame when this trial starts',
required=True,
)
starttime = fields.Float(
description='time in seconds when this trial starts',
required=True,
)
endframe = fields.Int(
description='frame when this trial ends',
required=True,
)
endtime = fields.Float(
description='time in seconds when this trial ends',
required=True,
)
trial_length = fields.Float(
required=True,
)
# timing paramters
change_frame = fields.Float(
description='The stimulus frame when the change occured on this trial',
required=True,
allow_nan=True,
)
scheduled_change_time = fields.Float(
description=("The time when the change was scheduled to occur "
"on this trial"),
required=True,
)
change_time = fields.Float(
description='The time when the change occured on this trial',
required=True,
allow_nan=True,
)
# image parameters
initial_image_category = fields.String(
description='The category of the initial images on this trial',
required=True,
allow_none=True,
)
initial_image_name = fields.String(
description=("The name of the last initial image before the "
"change on this trial"),
required=True,
allow_none=True,
)
change_image_category = fields.String(
description='The category of the change images on this trial',
required=True,
allow_none=True,
)
change_image_name = fields.String(
description='The name of the first change image on this trial',
required=True,
allow_none=True,
)
# oriented gratings paramters
initial_contrast = fields.Float(
description='The contrast of the initial orientation on this trial',
required=True,
allow_none=True,
)
change_contrast = fields.Float(
description='The contrast of the change orientation on this trial',
required=True,
allow_none=True,
)
initial_ori = fields.Float(
description='The orientation of the initial orientation on this trial',
required=True,
allow_none=True,
)
change_ori = fields.Float(
description='The orientation of the change orientation on this trial',
required=True,
allow_none=True,
allow_nan=True,
)
delta_ori = fields.Float(
description=("The difference between the initial and change "
"orientations on this trial"),
required=True,
allow_none=True,
)
# licks
lick_times = fields.List(
fields.Float,
description='times of licks on this trial',
required=True,
)
response_latency = fields.Float(
description=("The latency between the change and the first lick "
"on this trial"),
required=True,
allow_nan=True,
)
response_time = fields.List(
fields.Float,
description='need to check this with Doug',
required=True,
)
reward_frames = fields.List(
fields.Int,
required=True,
)
reward_times = fields.List(
fields.Float,
required=True,
)
reward_volume = fields.Float(
required=True,
)
rewarded = fields.Bool(
required=True,
)
auto_rewarded = fields.Bool(
description='whether this trial was an auto_rewarded trial',
required=True,
allow_none=True,
)
cumulative_reward_number = fields.Int(
description=("the cumulative number of rewards in the session at "
"trial end"),
required=True,
)
cumulative_volume = fields.Float(
description='the total volume of rewards in the session at trial end',
required=True,
)
# optogenetics
optogenetics = fields.Bool(
description=("whether optogenetic stimulation was applied "
"on this trial"),
required=True,
)
blank_duration_range = fields.List(
fields.Float,
required=True,
)
blank_screen_timeout = fields.Bool(
required=True,
)
color = fields.String(
required=True,
)
computer_name = fields.String(
required=True,
)
distribution_mean = fields.Float(
required=True,
)
LDT_mode = fields.String(
required=True,
)
lick_frames = fields.List(
fields.Integer(strict=True),
required=True,
)
mouse_id = fields.String(
required=True,
)
number_of_rewards = fields.Integer(
required=True,
strict=True,
)
prechange_minimum = fields.Float(
required=True,
)
response = fields.Float(
required=True,
)
response_type = fields.String(
required=True,
)
response_window = fields.List(
fields.Float,
required=True,
)
reward_licks = fields.List(
fields.Float,
required=True,
allow_none=True,
)
reward_lick_count = fields.Integer(
required=True,
# strict=True,
allow_none=True,
)
reward_lick_latency = fields.Float(
allow_none=True,
allow_nan=True,
)
reward_rate = fields.Float(
allow_none=True,
allow_nan=True,
)
rig_id = fields.String(
required=True,
)
session_duration = fields.Float(
required=True,
)
stage = fields.String(
required=True,
)
stim_duration = fields.Float(
required=True,
)
stimulus = fields.String(
required=True,
)
stimulus_distribution = fields.String(
required=True,
)
task = fields.String(
required=True,
)
trial_type = fields.String(
required=True,
)
user_id = fields.String(
required=True,
)
startdatetime = FriendlyDateTime(
required=True,
strict=True,
)
date = FriendlyDate(
required=True,
)
year = fields.Integer(
strict=True
)
month = fields.Integer(
required=True,
strict=True,
)
day = fields.Integer(
required=True,
strict=True,
)
hour = fields.Integer(
required=True,
strict=True,
)
dayofweek = fields.Integer(
strict=True,
required=True,
)
behavior_session_uuid = fields.UUID(
required=True,
)