Source code for ibllib.io.extractors.training_trials
importloggingimportnumpyasnpfromitertoolsimportaccumulatefrompackagingimportversionfromone.alf.ioimportAlfBunchimportibllib.io.raw_data_loadersasrawfromibllib.io.extractors.baseimportBaseBpodTrialsExtractor,run_extractor_classesfromibllib.io.extractors.training_wheelimportWheel_logger=logging.getLogger(__name__)__all__=['TrainingTrials']classFeedbackType(BaseBpodTrialsExtractor):""" Get the feedback that was delivered to subject. **Optional:** saves _ibl_trials.feedbackType.npy Checks in raw datafile for error and reward state. Will raise an error if more than one of the mutually exclusive states have been triggered. Sets feedbackType to -1 if error state was triggered (applies to no-go) Sets feedbackType to +1 if reward state was triggered """save_names='_ibl_trials.feedbackType.npy'var_names='feedbackType'def_extract(self):feedbackType=np.zeros(len(self.bpod_trials),np.int64)fori,tinenumerate(self.bpod_trials):state_names=['correct','error','no_go','omit_correct','omit_error','omit_no_go']outcome={sn:~np.isnan(t['behavior_data']['States timestamps'].get(sn,[[np.nan]])[0][0])forsninstate_names}assertnp.sum(list(outcome.values()))==1outcome=next(kforkinoutcomeifoutcome[k])ifoutcome=='correct':feedbackType[i]=1elifoutcomein['error','no_go']:feedbackType[i]=-1returnfeedbackTypeclassContrastLR(BaseBpodTrialsExtractor):""" Get left and right contrasts from raw datafile. Optionally, saves _ibl_trials.contrastLeft.npy and _ibl_trials.contrastRight.npy to alf folder. Uses signed_contrast to create left and right contrast vectors. """save_names=('_ibl_trials.contrastLeft.npy','_ibl_trials.contrastRight.npy')var_names=('contrastLeft','contrastRight')def_extract(self):# iblrigv8 has only flat values in the trial table so we can switch to parquet table when times come# and all the clutter here would fit in ~30 linesifisinstance(self.bpod_trials[0]['contrast'],float):contrastLeft=np.array([t['contrast']ifnp.sign(t['position'])<0elsenp.nanfortinself.bpod_trials])contrastRight=np.array([t['contrast']ifnp.sign(t['position'])>0elsenp.nanfortinself.bpod_trials])else:contrastLeft=np.array([t['contrast']['value']ifnp.sign(t['position'])<0elsenp.nanfortinself.bpod_trials])contrastRight=np.array([t['contrast']['value']ifnp.sign(t['position'])>0elsenp.nanfortinself.bpod_trials])returncontrastLeft,contrastRightclassProbabilityLeft(BaseBpodTrialsExtractor):save_names='_ibl_trials.probabilityLeft.npy'var_names='probabilityLeft'def_extract(self,**kwargs):returnnp.array([t['stim_probability_left']fortinself.bpod_trials])classChoice(BaseBpodTrialsExtractor):""" Get the subject's choice in every trial. **Optional:** saves _ibl_trials.choice.npy to alf folder. Uses signed_contrast and trial_correct. -1 is a CCW turn (towards the left) +1 is a CW turn (towards the right) 0 is a no_go trial If a trial is correct the choice of the animal was the inverse of the sign of the position. >>> choice[t] = -np.sign(position[t]) if trial_correct[t] """save_names='_ibl_trials.choice.npy'var_names='choice'def_extract(self):sitm_side=np.array([np.sign(t['position'])fortinself.bpod_trials])trial_correct=np.array([t['trial_correct']fortinself.bpod_trials])trial_nogo=np.array([~np.isnan(t['behavior_data']['States timestamps']['no_go'][0][0])fortinself.bpod_trials])choice=sitm_side.copy()choice[trial_correct]=-choice[trial_correct]choice[trial_nogo]=0choice=choice.astype(int)returnchoiceclassRepNum(BaseBpodTrialsExtractor):""" Count the consecutive repeated trials. **Optional:** saves _ibl_trials.repNum.npy to alf folder. Creates trial_repeated from trial['contrast']['type'] == 'RepeatContrast' >>> trial_repeated = [0, 1, 1, 0, 1, 0, 1, 1, 1, 0] >>> repNum = [0, 1, 2, 0, 1, 0, 1, 2, 3, 0] """save_names='_ibl_trials.repNum.npy'var_names='repNum'def_extract(self):defget_trial_repeat(trial):if'debias_trial'intrial:returntrial['debias_trial']elif'contrast'intrialandisinstance(trial['contrast'],dict):returntrial['contrast']['type']=='RepeatContrast'else:# For advanced choice world and its subclasses before version 8.19.0 there was no 'debias_trial' field# and no debiasing protocol applied, so simply return Falseassert(self.settings['PYBPOD_PROTOCOL'].startswith('_iblrig_tasks_advancedChoiceWorld')orself.settings['PYBPOD_PROTOCOL'].startswith('ccu_neuromodulatorChoiceWorld'))returnFalsetrial_repeated=np.fromiter(map(get_trial_repeat,self.bpod_trials),int)repNum=np.fromiter(accumulate(trial_repeated,lambdax,y:x+yifyelse0),int)returnrepNumclassRewardVolume(BaseBpodTrialsExtractor):""" Load reward volume delivered for each trial. **Optional:** saves _ibl_trials.rewardVolume.npy Uses reward_current to accumulate the amount of """save_names='_ibl_trials.rewardVolume.npy'var_names='rewardVolume'def_extract(self):trial_volume=[x['reward_amount']ifx['trial_correct']else0forxinself.bpod_trials]reward_volume=np.array(trial_volume).astype(np.float64)assertlen(reward_volume)==len(self.bpod_trials)returnreward_volumeclassFeedbackTimes(BaseBpodTrialsExtractor):""" Get the times the water or error tone was delivered to the animal. **Optional:** saves _ibl_trials.feedback_times.npy Gets reward and error state init times vectors, checks if the intersection of nans is empty, then merges the 2 vectors. """save_names='_ibl_trials.feedback_times.npy'var_names='feedback_times'@staticmethoddefget_feedback_times_lt5(session_path,task_collection='raw_behavior_data',data=False):ifnotdata:data=raw.load_data(session_path,task_collection=task_collection)rw_times=[tr['behavior_data']['States timestamps']['reward'][0][0]fortrindata]err_times=[tr['behavior_data']['States timestamps']['error'][0][0]fortrindata]nogo_times=[tr['behavior_data']['States timestamps']['no_go'][0][0]fortrindata]assertsum(np.isnan(rw_times)&np.isnan(err_times)&np.isnan(nogo_times))==0merge=np.array([np.array(times)[~np.isnan(times)]fortimesinzip(rw_times,err_times,nogo_times)]).squeeze()returnnp.array(merge)@staticmethoddefget_feedback_times_ge5(session_path,task_collection='raw_behavior_data',data=False):# ger err and no go trig times -- look for BNC2High of trial -- verify# only 2 onset times go tone and noise, select 2nd/-1 OR select the one# that is grater than the nogo or err trial onset timeifnotdata:data=raw.load_data(session_path,task_collection=task_collection)missed_bnc2=0rw_times,err_sound_times,merge=[np.zeros([len(data),])for_inrange(3)]forind,trinenumerate(data):st=tr['behavior_data']['Events timestamps'].get('BNC2High',None)ifnotst:st=np.array([np.nan,np.nan])missed_bnc2+=1# xonar soundcard duplicates events, remove consecutive events too close togetherst=np.delete(st,np.where(np.diff(st)<0.020)[0]+1)rw_times[ind]=tr['behavior_data']['States timestamps']['reward'][0][0]# get the error sound only if the reward is nanerr_sound_times[ind]=st[-1]ifst.size>=2andnp.isnan(rw_times[ind])elsenp.nanifmissed_bnc2==len(data):_logger.warning('No BNC2 for feedback times, filling error trials NaNs')merge*=np.nanmerge[~np.isnan(rw_times)]=rw_times[~np.isnan(rw_times)]merge[~np.isnan(err_sound_times)]=err_sound_times[~np.isnan(err_sound_times)]returnmergedef_extract(self):# Version checkifversion.parse(self.settings['IBLRIG_VERSION']or'100.0.0')>=version.parse('5.0.0'):merge=self.get_feedback_times_ge5(self.session_path,task_collection=self.task_collection,data=self.bpod_trials)else:merge=self.get_feedback_times_lt5(self.session_path,task_collection=self.task_collection,data=self.bpod_trials)returnnp.array(merge)classIntervals(BaseBpodTrialsExtractor):""" Trial start to trial end. Trial end includes 1 or 2 seconds after feedback, (depending on the feedback) and 0.5 seconds of iti. **Optional:** saves _ibl_trials.intervals.npy Uses the corrected Trial start and Trial end timestamp values form PyBpod. """save_names='_ibl_trials.intervals.npy'var_names='intervals'def_extract(self):starts=[t['behavior_data']['Trial start timestamp']fortinself.bpod_trials]ends=[t['behavior_data']['Trial end timestamp']fortinself.bpod_trials]returnnp.array([starts,ends]).TclassResponseTimes(BaseBpodTrialsExtractor):""" Time (in absolute seconds from session start) when a response was recorded. **Optional:** saves _ibl_trials.response_times.npy Uses the timestamp of the end of the closed_loop state. """save_names='_ibl_trials.response_times.npy'var_names='response_times'def_extract(self):rt=np.array([tr['behavior_data']['States timestamps']['closed_loop'][0][1]fortrinself.bpod_trials])returnrtclassItiDuration(BaseBpodTrialsExtractor):""" Calculate duration of iti from state timestamps. **Optional:** saves _ibl_trials.iti_duration.npy Uses Trial end timestamp and get_response_times to calculate iti. """save_names='_ibl_trials.itiDuration.npy'var_names='iti_dur'def_extract(self):rt,_=ResponseTimes(self.session_path).extract(save=False,task_collection=self.task_collection,bpod_trials=self.bpod_trials,settings=self.settings)ends=np.array([t['behavior_data']['Trial end timestamp']fortinself.bpod_trials])iti_dur=ends-rtreturniti_durclassGoCueTriggerTimes(BaseBpodTrialsExtractor):""" Get trigger times of goCue from state machine. Current software solution for triggering sounds uses PyBpod soft codes. Delays can be in the order of 10's of ms. This is the time when the command to play the sound was executed. To measure accurate time, either getting the sound onset from xonar soundcard sync pulse (latencies may vary). """save_names='_ibl_trials.goCueTrigger_times.npy'var_names='goCueTrigger_times'def_extract(self):ifversion.parse(self.settings['IBLRIG_VERSION']or'100.0.0')>=version.parse('5.0.0'):goCue=np.array([tr['behavior_data']['States timestamps']['play_tone'][0][0]fortrinself.bpod_trials])else:goCue=np.array([tr['behavior_data']['States timestamps']['closed_loop'][0][0]fortrinself.bpod_trials])returngoCueclassTrialType(BaseBpodTrialsExtractor):save_names='_ibl_trials.type.npy'var_name='trial_type'def_extract(self):trial_type=[]fortrinself.bpod_trials:if~np.isnan(tr["behavior_data"]["States timestamps"]["reward"][0][0]):trial_type.append(1)elif~np.isnan(tr["behavior_data"]["States timestamps"]["error"][0][0]):trial_type.append(-1)elif~np.isnan(tr["behavior_data"]["States timestamps"]["no_go"][0][0]):trial_type.append(0)else:_logger.warning("Trial is not in set {-1, 0, 1}, appending NaN to trialType")trial_type.append(np.nan)returnnp.array(trial_type)classGoCueTimes(BaseBpodTrialsExtractor):""" Get trigger times of goCue from state machine. Current software solution for triggering sounds uses PyBpod soft codes. Delays can be in the order of 10-100s of ms. This is the time when the command to play the sound was executed. To measure accurate time, either getting the sound onset from the future microphone OR the new xonar soundcard and setup developed by Sanworks guarantees a set latency (in testing). """save_names='_ibl_trials.goCue_times.npy'var_names='goCue_times'def_extract(self):go_cue_times=np.zeros([len(self.bpod_trials),])forind,trinenumerate(self.bpod_trials):ifraw.get_port_events(tr,'BNC2'):bnchigh=tr['behavior_data']['Events timestamps'].get('BNC2High',None)ifbnchigh:go_cue_times[ind]=bnchigh[0]continuebnclow=tr['behavior_data']['Events timestamps'].get('BNC2Low',None)ifbnclow:go_cue_times[ind]=bnclow[0]-0.1continuego_cue_times[ind]=np.nanelse:go_cue_times[ind]=np.nannmissing=np.sum(np.isnan(go_cue_times))# Check if all stim_syncs have failed to be detectedifnp.all(np.isnan(go_cue_times)):_logger.warning(f'{self.session_path}: Missing ALL !! BNC2 TTLs ({nmissing} trials)')# Check if any stim_sync has failed be detected for every trialelifnp.any(np.isnan(go_cue_times)):_logger.warning(f'{self.session_path}: Missing BNC2 TTLs on {nmissing} trials')returngo_cue_timesclassIncludedTrials(BaseBpodTrialsExtractor):save_names='_ibl_trials.included.npy'var_names='included'def_extract(self):ifversion.parse(self.settings['IBLRIG_VERSION']or'100.0.0')>=version.parse('5.0.0'):trials_included=self.get_included_trials_ge5(data=self.bpod_trials,settings=self.settings)else:trials_included=self.get_included_trials_lt5(data=self.bpod_trials)returntrials_included@staticmethoddefget_included_trials_lt5(data=False):trials_included=np.ones(len(data),dtype=bool)returntrials_included@staticmethoddefget_included_trials_ge5(data=False,settings=False):trials_included=np.array([Truefortindata])if('SUBJECT_DISENGAGED_TRIGGERED'insettings.keys()andsettings['SUBJECT_DISENGAGED_TRIGGERED']isnotFalse):idx=settings['SUBJECT_DISENGAGED_TRIALNUM']-1trials_included[idx:]=Falsereturntrials_includedclassItiInTimes(BaseBpodTrialsExtractor):var_names='itiIn_times'def_extract(self):ifversion.parse(self.settings["IBLRIG_VERSION"]or'100.0.0')<version.parse("5.0.0"):iti_in=np.ones(len(self.bpod_trials))*np.nanelse:iti_in=np.array([tr["behavior_data"]["States timestamps"]["exit_state"][0][0]fortrinself.bpod_trials])returniti_inclassErrorCueTriggerTimes(BaseBpodTrialsExtractor):var_names='errorCueTrigger_times'def_extract(self):errorCueTrigger_times=np.zeros(len(self.bpod_trials))*np.nanfori,trinenumerate(self.bpod_trials):nogo=tr["behavior_data"]["States timestamps"]["no_go"][0][0]error=tr["behavior_data"]["States timestamps"]["error"][0][0]ifnp.all(~np.isnan(nogo)):errorCueTrigger_times[i]=nogoelifnp.all(~np.isnan(error)):errorCueTrigger_times[i]=errorreturnerrorCueTrigger_timesclassStimFreezeTriggerTimes(BaseBpodTrialsExtractor):var_names='stimFreezeTrigger_times'def_extract(self):ifversion.parse(self.settings["IBLRIG_VERSION"]or'100.0.0')<version.parse("6.2.5"):returnnp.ones(len(self.bpod_trials))*np.nanfreeze_reward=np.array([Trueifnp.all(~np.isnan(tr["behavior_data"]["States timestamps"]["freeze_reward"][0]))elseFalsefortrinself.bpod_trials])freeze_error=np.array([Trueifnp.all(~np.isnan(tr["behavior_data"]["States timestamps"]["freeze_error"][0]))elseFalsefortrinself.bpod_trials])no_go=np.array([Trueifnp.all(~np.isnan(tr["behavior_data"]["States timestamps"]["no_go"][0]))elseFalsefortrinself.bpod_trials])assert(np.sum(freeze_error)+np.sum(freeze_reward)+np.sum(no_go)==len(self.bpod_trials))stimFreezeTrigger=np.array([])forr,e,n,trinzip(freeze_reward,freeze_error,no_go,self.bpod_trials):ifn:stimFreezeTrigger=np.append(stimFreezeTrigger,np.nan)continuestate="freeze_reward"ifrelse"freeze_error"stimFreezeTrigger=np.append(stimFreezeTrigger,tr["behavior_data"]["States timestamps"][state][0][0])returnstimFreezeTriggerclassStimOffTriggerTimes(BaseBpodTrialsExtractor):var_names='stimOffTrigger_times'save_names='_ibl_trials.stimOnTrigger_times.npy'def_extract(self):ifversion.parse(self.settings["IBLRIG_VERSION"]or'100.0.0')>=version.parse("6.2.5"):stim_off_trigger_state="hide_stim"elifversion.parse(self.settings["IBLRIG_VERSION"])>=version.parse("5.0.0"):stim_off_trigger_state="exit_state"else:stim_off_trigger_state="trial_start"stimOffTrigger_times=np.array([tr["behavior_data"]["States timestamps"][stim_off_trigger_state][0][0]fortrinself.bpod_trials])# If pre version 5.0.0 no specific nogo Off trigger was given, just return trial_startsifstim_off_trigger_state=="trial_start":returnstimOffTrigger_timesno_goTrigger_times=np.array([tr["behavior_data"]["States timestamps"]["no_go"][0][0]fortrinself.bpod_trials])# Stim off trigs are either in their own state or in the no_go state if the# mouse did not move, if the stim_off_trigger_state always exist# (exit_state or trial_start)# no NaNs will happen, NaNs might happen in at last trial if# session was stopped after response# if stim_off_trigger_state == "hide_stim":# assert all(~np.isnan(no_goTrigger_times) == np.isnan(stimOffTrigger_times))# Patch with the no_go states trig timesstimOffTrigger_times[~np.isnan(no_goTrigger_times)]=no_goTrigger_times[~np.isnan(no_goTrigger_times)]returnstimOffTrigger_timesclassStimOnTriggerTimes(BaseBpodTrialsExtractor):save_names='_ibl_trials.stimOnTrigger_times.npy'var_names='stimOnTrigger_times'def_extract(self):# Get the stim_on_state that triggers the onset of the stimstim_on_state=np.array([tr['behavior_data']['States timestamps']['stim_on'][0]fortrinself.bpod_trials])returnstim_on_state[:,0].TclassStimOnTimes_deprecated(BaseBpodTrialsExtractor):save_names='_ibl_trials.stimOn_times.npy'var_names='stimOn_times'def_extract(self):""" Find the time of the state machine command to turn on the stim (state stim_on start or rotary_encoder_event2) Find the next frame change from the photodiode after that TS. Screen is not displaying anything until then. (Frame changes are in BNC1 High and BNC1 Low) """# Version check_logger.warning("Deprecation Warning: this is an old version of stimOn extraction.""From version 5., use StimOnOffFreezeTimes")ifversion.parse(self.settings['IBLRIG_VERSION']or'100.0.0')>=version.parse('5.0.0'):stimOn_times=self.get_stimOn_times_ge5(self.session_path,data=self.bpod_trials,task_collection=self.task_collection)else:stimOn_times=self.get_stimOn_times_lt5(self.session_path,data=self.bpod_trials,task_collection=self.task_collection)returnnp.array(stimOn_times)@staticmethoddefget_stimOn_times_ge5(session_path,data=False,task_collection='raw_behavior_data'):""" Find first and last stim_sync pulse of the trial. stimOn_times should be the first after the stim_on state. (Stim updates are in BNC1High and BNC1Low - frame2TTL device) Check that all trials have frame changes. Find length of stim_on_state [start, stop]. If either check fails the HW device failed to detect the stim_sync square change Substitute that trial's missing or incorrect value with a NaN. return stimOn_times """ifnotdata:data=raw.load_data(session_path,task_collection=task_collection)# Get all stim_sync events detectedstim_sync_all=[raw.get_port_events(tr,'BNC1')fortrindata]stim_sync_all=[np.array(x)forxinstim_sync_all]# Get the stim_on_state that triggers the onset of the stimstim_on_state=np.array([tr['behavior_data']['States timestamps']['stim_on'][0]fortrindata])stimOn_times=np.array([])forsync,on,offinzip(stim_sync_all,stim_on_state[:,0],stim_on_state[:,1]):pulse=sync[np.where(np.bitwise_and((sync>on),(sync<=off)))]ifpulse.size==0:stimOn_times=np.append(stimOn_times,np.nan)else:stimOn_times=np.append(stimOn_times,pulse)nmissing=np.sum(np.isnan(stimOn_times))# Check if all stim_syncs have failed to be detectedifnp.all(np.isnan(stimOn_times)):_logger.error(f'{session_path}: Missing ALL BNC1 TTLs ({nmissing} trials)')# Check if any stim_sync has failed be detected for every trialifnp.any(np.isnan(stimOn_times)):_logger.warning(f'{session_path}: Missing BNC1 TTLs on {nmissing} trials')returnstimOn_times@staticmethoddefget_stimOn_times_lt5(session_path,data=False,task_collection='raw_behavior_data'):""" Find the time of the statemachine command to turn on the stim (state stim_on start or rotary_encoder_event2) Find the next frame change from the photodiode after that TS. Screen is not displaying anything until then. (Frame changes are in BNC1High and BNC1Low) """ifnotdata:data=raw.load_data(session_path,task_collection=task_collection)stim_on=[]bnc_h=[]bnc_l=[]fortrindata:stim_on.append(tr['behavior_data']['States timestamps']['stim_on'][0][0])if'BNC1High'intr['behavior_data']['Events timestamps'].keys():bnc_h.append(np.array(tr['behavior_data']['Events timestamps']['BNC1High']))else:bnc_h.append(np.array([np.NINF]))if'BNC1Low'intr['behavior_data']['Events timestamps'].keys():bnc_l.append(np.array(tr['behavior_data']['Events timestamps']['BNC1Low']))else:bnc_l.append(np.array([np.NINF]))stim_on=np.array(stim_on)bnc_h=np.array(bnc_h,dtype=object)bnc_l=np.array(bnc_l,dtype=object)count_missing=0stimOn_times=np.zeros_like(stim_on)foriinrange(len(stim_on)):hl=np.sort(np.concatenate([bnc_h[i],bnc_l[i]]))stot=hl[hl>stim_on[i]]ifnp.size(stot)==0:stot=np.array([np.nan])count_missing+=1stimOn_times[i]=stot[0]ifnp.all(np.isnan(stimOn_times)):_logger.error(f'{session_path}: Missing ALL BNC1 TTLs ({count_missing} trials)')ifcount_missing>0:_logger.warning(f'{session_path}: Missing BNC1 TTLs on {count_missing} trials')returnnp.array(stimOn_times)classStimOnOffFreezeTimes(BaseBpodTrialsExtractor):""" Extracts stim on / off and freeze times from Bpod BNC1 detected fronts. Each stimulus event is the first detected front of the BNC1 signal after the trigger state, but before the next trigger state. """save_names=('_ibl_trials.stimOn_times.npy','_ibl_trials.stimOff_times.npy',None)var_names=('stimOn_times','stimOff_times','stimFreeze_times')def_extract(self):choice=Choice(self.session_path).extract(bpod_trials=self.bpod_trials,task_collection=self.task_collection,settings=self.settings,save=False)[0]stimOnTrigger=StimOnTriggerTimes(self.session_path).extract(bpod_trials=self.bpod_trials,task_collection=self.task_collection,settings=self.settings,save=False)[0]stimFreezeTrigger=StimFreezeTriggerTimes(self.session_path).extract(bpod_trials=self.bpod_trials,task_collection=self.task_collection,settings=self.settings,save=False)[0]stimOffTrigger=StimOffTriggerTimes(self.session_path).extract(bpod_trials=self.bpod_trials,task_collection=self.task_collection,settings=self.settings,save=False)[0]f2TTL=[raw.get_port_events(tr,name='BNC1')fortrinself.bpod_trials]assertstimOnTrigger.size==stimFreezeTrigger.size==stimOffTrigger.size==choice.size==len(f2TTL)assertall(stimOnTrigger<np.nan_to_num(stimFreezeTrigger,nan=np.inf))and \
all(np.nan_to_num(stimFreezeTrigger,nan=-np.inf)<stimOffTrigger)stimOn_times=np.array([])stimOff_times=np.array([])stimFreeze_times=np.array([])has_freeze=version.parse(self.settings.get('IBLRIG_VERSION','0'))>=version.parse('6.2.5')fortr,on,freeze,off,cinzip(f2TTL,stimOnTrigger,stimFreezeTrigger,stimOffTrigger,choice):tr=np.array(tr)# stim onlim=freezeifhas_freezeelseoffidx,=np.where(np.logical_and(on<tr,tr<lim))stimOn_times=np.append(stimOn_times,tr[idx[0]]ifidx.size>0elsenp.nan)# stim offidx,=np.where(off<tr)stimOff_times=np.append(stimOff_times,tr[idx[0]]ifidx.size>0elsenp.nan)# stim freeze - take last event before off triggerifhas_freeze:idx,=np.where(np.logical_and(freeze<tr,tr<off))stimFreeze_times=np.append(stimFreeze_times,tr[idx[-1]]ifidx.size>0elsenp.nan)else:idx,=np.where(tr<=off)stimFreeze_times=np.append(stimFreeze_times,tr[idx[-1]]ifidx.size>0elsenp.nan)# In no_go trials no stimFreeze happens just stim OffstimFreeze_times[choice==0]=np.nanreturnstimOn_times,stimOff_times,stimFreeze_timesclassPhasePosQuiescence(BaseBpodTrialsExtractor):"""Extract stimulus phase, position and quiescence from Bpod data. For extraction of pre-generated events, use the ProbaContrasts extractor instead. """save_names=(None,None,'_ibl_trials.quiescencePeriod.npy')var_names=('phase','position','quiescence')def_extract(self,**kwargs):phase=np.array([t['stim_phase']fortinself.bpod_trials])position=np.array([t['position']fortinself.bpod_trials])quiescence=np.array([t['quiescent_period']fortinself.bpod_trials])returnphase,position,quiescenceclassPauseDuration(BaseBpodTrialsExtractor):"""Extract pause duration from raw trial data."""save_names=Nonevar_names='pause_duration'def_extract(self,**kwargs):# pausing logic added in version 8.9.0ver=version.parse(self.settings.get('IBLRIG_VERSION')or'0')default=0.ifver<version.parse('8.9.0')elsenp.nanreturnnp.fromiter((t.get('pause_duration',default)fortinself.bpod_trials),dtype=float)classTrialsTable(BaseBpodTrialsExtractor):""" Extracts the following into a table from Bpod raw data: intervals, goCue_times, response_times, choice, stimOn_times, contrastLeft, contrastRight, feedback_times, feedbackType, rewardVolume, probabilityLeft, firstMovement_times Additionally extracts the following wheel data: wheel_timestamps, wheel_position, wheel_moves_intervals, wheel_moves_peak_amplitude """save_names=('_ibl_trials.table.pqt',None,None,'_ibl_wheel.timestamps.npy','_ibl_wheel.position.npy','_ibl_wheelMoves.intervals.npy','_ibl_wheelMoves.peakAmplitude.npy',None,None)var_names=('table','stimOff_times','stimFreeze_times','wheel_timestamps','wheel_position','wheelMoves_intervals','wheelMoves_peakAmplitude','wheelMoves_peakVelocity_times','is_final_movement')def_extract(self,extractor_classes=None,**kwargs):base=[Intervals,GoCueTimes,ResponseTimes,Choice,StimOnOffFreezeTimes,ContrastLR,FeedbackTimes,FeedbackType,RewardVolume,ProbabilityLeft,Wheel]out,_=run_extractor_classes(base,session_path=self.session_path,bpod_trials=self.bpod_trials,settings=self.settings,save=False,task_collection=self.task_collection)table=AlfBunch({k:vfork,vinout.items()ifknotinself.var_names})assertlen(table.keys())==12returntable.to_df(),*(out.pop(x)forxinself.var_namesifx!='table')