Coverage for ibllib/io/extractors/training_trials.py: 80%

1import logging (empty)

2import numpy as np (empty)

3from itertools import accumulate (empty)

4from packaging import version (empty)

5from one.alf.io import AlfBunch (empty)

6 

7import ibllib.io.raw_data_loaders as raw (empty)

8from ibllib.io.extractors.base import BaseBpodTrialsExtractor, run_extractor_classes (empty)

9from ibllib.io.extractors.training_wheel import Wheel (empty)

10 

11 

12_logger = logging.getLogger(__name__) (empty)

13__all__ = ['TrainingTrials'] (empty)

14 

15 

16class FeedbackType(BaseBpodTrialsExtractor): (empty)

17 """ 

18 Get the feedback that was delivered to subject. 

19 **Optional:** saves _ibl_trials.feedbackType.npy 

20 

21 Checks in raw datafile for error and reward state. 

22 Will raise an error if more than one of the mutually exclusive states have 

23 been triggered. 

24 

25 Sets feedbackType to -1 if error state was triggered (applies to no-go) 

26 Sets feedbackType to +1 if reward state was triggered 

27 """ 

28 save_names = '_ibl_trials.feedbackType.npy' (empty)

29 var_names = 'feedbackType' (empty)

30 

31 def _extract(self): (empty)

32 feedbackType = np.zeros(len(self.bpod_trials), np.int64) 18 ctx1awcjhdeqlmknoifbgp

33 for i, t in enumerate(self.bpod_trials): 18 ctx1awcjhdeqlmknoifbgp

34 state_names = ['correct', 'error', 'no_go', 'omit_correct', 'omit_error', 'omit_no_go'] 18 ctx1awcjhdeqlmknoifbgp

35 outcome = {sn: ~np.isnan(t['behavior_data']['States timestamps'].get(sn, [[np.nan]])[0][0]) for sn in state_names} 18 ctx1awcjhdeqlmknoifbgp

36 assert np.sum(list(outcome.values())) == 1 18 ctx1awcjhdeqlmknoifbgp

37 outcome = next(k for k in outcome if outcome[k]) 18 ctx1awcjhdeqlmknoifbgp

38 if outcome == 'correct': 18 ctx1awcjhdeqlmknoifbgp

39 feedbackType[i] = 1 17 ctx1awcjhdeqlmknoifbg

40 elif outcome in ['error', 'no_go']: 18 ctx1awcjhdeqlmknoifbgp

41 feedbackType[i] = -1 18 ctx1awcjhdeqlmknoifbgp

42 return feedbackType 18 ctx1awcjhdeqlmknoifbgp

43 

44 

45class ContrastLR(BaseBpodTrialsExtractor): (empty)

46 """ 

47 Get left and right contrasts from raw datafile. Optionally, saves 

48 _ibl_trials.contrastLeft.npy and _ibl_trials.contrastRight.npy to alf folder. 

49 

50 Uses signed_contrast to create left and right contrast vectors. 

51 """ 

52 save_names = ('_ibl_trials.contrastLeft.npy', '_ibl_trials.contrastRight.npy') (empty)

53 var_names = ('contrastLeft', 'contrastRight') (empty)

54 

55 def _extract(self): (empty)

56 # iblrigv8 has only flat values in the trial table so we can switch to parquet table when times come 

57 # and all the clutter here would fit in ~30 lines 

58 if isinstance(self.bpod_trials[0]['contrast'], float): 6 ctx1Fcdefb

59 contrastLeft = np.array([t['contrast'] if np.sign( 1 ctx1f

60 t['position']) < 0 else np.nan for t in self.bpod_trials]) 

61 contrastRight = np.array([t['contrast'] if np.sign( 1 ctx1f

62 t['position']) > 0 else np.nan for t in self.bpod_trials]) 

63 else: 

64 contrastLeft = np.array([t['contrast']['value'] if np.sign( 5 ctx1Fcdeb

65 t['position']) < 0 else np.nan for t in self.bpod_trials]) 

66 contrastRight = np.array([t['contrast']['value'] if np.sign( 5 ctx1Fcdeb

67 t['position']) > 0 else np.nan for t in self.bpod_trials]) 

68 

69 return contrastLeft, contrastRight 6 ctx1Fcdefb

70 

71 

72class ProbabilityLeft(BaseBpodTrialsExtractor): (empty)

73 save_names = '_ibl_trials.probabilityLeft.npy' (empty)

74 var_names = 'probabilityLeft' (empty)

75 

76 def _extract(self, **kwargs): (empty)

77 return np.array([t['stim_probability_left'] for t in self.bpod_trials]) 10 ctx1Mcjdeqlifb

78 

79 

80class Choice(BaseBpodTrialsExtractor): (empty)

81 """ 

82 Get the subject's choice in every trial. 

83 **Optional:** saves _ibl_trials.choice.npy to alf folder. 

84 

85 Uses signed_contrast and trial_correct. 

86 -1 is a CCW turn (towards the left) 

87 +1 is a CW turn (towards the right) 

88 0 is a no_go trial 

89 If a trial is correct the choice of the animal was the inverse of the sign 

90 of the position. 

91 

92 >>> choice[t] = -np.sign(position[t]) if trial_correct[t] 

93 """ 

94 save_names = '_ibl_trials.choice.npy' (empty)

95 var_names = 'choice' (empty)

96 

97 def _extract(self): (empty)

98 sitm_side = np.array([np.sign(t['position']) for t in self.bpod_trials]) 19 ctx1aBcrjhdeqlmknoifbgp

99 trial_correct = np.array([t['trial_correct'] for t in self.bpod_trials]) 19 ctx1aBcrjhdeqlmknoifbgp

100 trial_nogo = np.array( 19 ctx1aBcrjhdeqlmknoifbgp

101 [~np.isnan(t['behavior_data']['States timestamps']['no_go'][0][0]) 

102 for t in self.bpod_trials]) 

103 choice = sitm_side.copy() 19 ctx1aBcrjhdeqlmknoifbgp

104 choice[trial_correct] = -choice[trial_correct] 19 ctx1aBcrjhdeqlmknoifbgp

105 choice[trial_nogo] = 0 19 ctx1aBcrjhdeqlmknoifbgp

106 choice = choice.astype(int) 19 ctx1aBcrjhdeqlmknoifbgp

107 return choice 19 ctx1aBcrjhdeqlmknoifbgp

108 

109 

110class RepNum(BaseBpodTrialsExtractor): (empty)

111 """ 

112 Count the consecutive repeated trials. 

113 **Optional:** saves _ibl_trials.repNum.npy to alf folder. 

114 

115 Creates trial_repeated from trial['contrast']['type'] == 'RepeatContrast' 

116 

117 >>> trial_repeated = [0, 1, 1, 0, 1, 0, 1, 1, 1, 0] 

118 >>> repNum = [0, 1, 2, 0, 1, 0, 1, 2, 3, 0] 

119 """ 

120 save_names = '_ibl_trials.repNum.npy' (empty)

121 var_names = 'repNum' (empty)

122 

123 def _extract(self): (empty)

124 def get_trial_repeat(trial): 6 ctx1Ccdefb

125 if 'debias_trial' in trial: 6 ctx1Ccdefb

126 return trial['debias_trial'] 1 ctx1f

127 elif 'contrast' in trial and isinstance(trial['contrast'], dict): 5 ctx1Ccdeb

128 return trial['contrast']['type'] == 'RepeatContrast' 5 ctx1Ccdeb

129 else: 

130 # For advanced choice world and its subclasses before version 8.19.0 there was no 'debias_trial' field 

131 # and no debiasing protocol applied, so simply return False 

132 assert (self.settings['PYBPOD_PROTOCOL'].startswith('_iblrig_tasks_advancedChoiceWorld') or 

133 self.settings['PYBPOD_PROTOCOL'].startswith('ccu_neuromodulatorChoiceWorld')) 

134 return False 

135 

136 trial_repeated = np.fromiter(map(get_trial_repeat, self.bpod_trials), int) 6 ctx1Ccdefb

137 repNum = np.fromiter(accumulate(trial_repeated, lambda x, y: x + y if y else 0), int) 6 ctx1Ccdefb

138 return repNum 6 ctx1Ccdefb

139 

140 

141class RewardVolume(BaseBpodTrialsExtractor): (empty)

142 """ 

143 Load reward volume delivered for each trial. 

144 **Optional:** saves _ibl_trials.rewardVolume.npy 

145 

146 Uses reward_current to accumulate the amount of 

147 """ 

148 save_names = '_ibl_trials.rewardVolume.npy' (empty)

149 var_names = 'rewardVolume' (empty)

150 

151 def _extract(self): (empty)

152 trial_volume = [x['reward_amount'] 18 ctx1aGcjhdeqlmknoifbgp

153 if x['trial_correct'] else 0 for x in self.bpod_trials] 

154 reward_volume = np.array(trial_volume).astype(np.float64) 18 ctx1aGcjhdeqlmknoifbgp

155 assert len(reward_volume) == len(self.bpod_trials) 18 ctx1aGcjhdeqlmknoifbgp

156 return reward_volume 18 ctx1aGcjhdeqlmknoifbgp

157 

158 

159class FeedbackTimes(BaseBpodTrialsExtractor): (empty)

160 """ 

161 Get the times the water or error tone was delivered to the animal. 

162 **Optional:** saves _ibl_trials.feedback_times.npy 

163 

164 Gets reward and error state init times vectors, 

165 checks if the intersection of nans is empty, then 

166 merges the 2 vectors. 

167 """ 

168 save_names = '_ibl_trials.feedback_times.npy' (empty)

169 var_names = 'feedback_times' (empty)

170 

171 @staticmethod (empty)

172 def get_feedback_times_lt5(session_path, task_collection='raw_behavior_data', data=False): (empty)

173 if not data: 5 ctx1azcqb

174 data = raw.load_data(session_path, task_collection=task_collection) 

175 rw_times = [tr['behavior_data']['States timestamps']['reward'][0][0] 5 ctx1azcqb

176 for tr in data] 

177 err_times = [tr['behavior_data']['States timestamps']['error'][0][0] 5 ctx1azcqb

178 for tr in data] 

179 nogo_times = [tr['behavior_data']['States timestamps']['no_go'][0][0] 5 ctx1azcqb

180 for tr in data] 

181 assert sum(np.isnan(rw_times) & 5 ctx1azcqb

182 np.isnan(err_times) & np.isnan(nogo_times)) == 0 

183 merge = np.array([np.array(times)[~np.isnan(times)] for times in 5 ctx1azcqb

184 zip(rw_times, err_times, nogo_times)]).squeeze() 

185 

186 return np.array(merge) 5 ctx1azcqb

187 

188 @staticmethod (empty)

189 def get_feedback_times_ge5(session_path, task_collection='raw_behavior_data', data=False): (empty)

190 # ger err and no go trig times -- look for BNC2High of trial -- verify 

191 # only 2 onset times go tone and noise, select 2nd/-1 OR select the one 

192 # that is grater than the nogo or err trial onset time 

193 if not data: 18 ctx1ucjshdeltmknoifbgp

194 data = raw.load_data(session_path, task_collection=task_collection) 

195 missed_bnc2 = 0 18 ctx1ucjshdeltmknoifbgp

196 rw_times, err_sound_times, merge = [np.zeros([len(data), ]) for _ in range(3)] 18 ctx1ucjshdeltmknoifbgp

197 

198 for ind, tr in enumerate(data): 18 ctx1ucjshdeltmknoifbgp

199 st = tr['behavior_data']['Events timestamps'].get('BNC2High', None) 18 ctx1ucjshdeltmknoifbgp

200 if not st: 18 ctx1ucjshdeltmknoifbgp

201 st = np.array([np.nan, np.nan]) 2 ctx1bg

202 missed_bnc2 += 1 2 ctx1bg

203 # xonar soundcard duplicates events, remove consecutive events too close together 

204 st = np.delete(st, np.where(np.diff(st) < 0.020)[0] + 1) 18 ctx1ucjshdeltmknoifbgp

205 rw_times[ind] = tr['behavior_data']['States timestamps']['reward'][0][0] 18 ctx1ucjshdeltmknoifbgp

206 # get the error sound only if the reward is nan 

207 err_sound_times[ind] = st[-1] if st.size >= 2 and np.isnan(rw_times[ind]) else np.nan 18 ctx1ucjshdeltmknoifbgp

208 if missed_bnc2 == len(data): 18 ctx1ucjshdeltmknoifbgp

209 _logger.warning('No BNC2 for feedback times, filling error trials NaNs') 

210 merge *= np.nan 18 ctx1ucjshdeltmknoifbgp

211 merge[~np.isnan(rw_times)] = rw_times[~np.isnan(rw_times)] 18 ctx1ucjshdeltmknoifbgp

212 merge[~np.isnan(err_sound_times)] = err_sound_times[~np.isnan(err_sound_times)] 18 ctx1ucjshdeltmknoifbgp

213 

214 return merge 18 ctx1ucjshdeltmknoifbgp

215 

216 def _extract(self): (empty)

217 # Version check 

218 if version.parse(self.settings['IBLRIG_VERSION'] or '100.0.0') >= version.parse('5.0.0'): 21 ctx1auzcjshdeqltmknoifbgp

219 merge = self.get_feedback_times_ge5(self.session_path, task_collection=self.task_collection, data=self.bpod_trials) 18 ctx1ucjshdeltmknoifbgp

220 else: 

221 merge = self.get_feedback_times_lt5(self.session_path, task_collection=self.task_collection, data=self.bpod_trials) 5 ctx1azcqb

222 return np.array(merge) 21 ctx1auzcjshdeqltmknoifbgp

223 

224 

225class Intervals(BaseBpodTrialsExtractor): (empty)

226 """ 

227 Trial start to trial end. Trial end includes 1 or 2 seconds after feedback, 

228 (depending on the feedback) and 0.5 seconds of iti. 

229 **Optional:** saves _ibl_trials.intervals.npy 

230 

231 Uses the corrected Trial start and Trial end timestamp values form PyBpod. 

232 """ 

233 save_names = '_ibl_trials.intervals.npy' (empty)

234 var_names = 'intervals' (empty)

235 

236 def _extract(self): (empty)

237 starts = [t['behavior_data']['Trial start timestamp'] for t in self.bpod_trials] 18 ctx1aHcjhdeqlmknoifbgp

238 ends = [t['behavior_data']['Trial end timestamp'] for t in self.bpod_trials] 18 ctx1aHcjhdeqlmknoifbgp

239 return np.array([starts, ends]).T 18 ctx1aHcjhdeqlmknoifbgp

240 

241 

242class ResponseTimes(BaseBpodTrialsExtractor): (empty)

243 """ 

244 Time (in absolute seconds from session start) when a response was recorded. 

245 **Optional:** saves _ibl_trials.response_times.npy 

246 

247 Uses the timestamp of the end of the closed_loop state. 

248 """ 

249 save_names = '_ibl_trials.response_times.npy' (empty)

250 var_names = 'response_times' (empty)

251 

252 def _extract(self): (empty)

253 rt = np.array([tr['behavior_data']['States timestamps']['closed_loop'][0][1] 18 ctx1aKcjhdeqlmknoifbgp

254 for tr in self.bpod_trials]) 

255 return rt 18 ctx1aKcjhdeqlmknoifbgp

256 

257 

258class ItiDuration(BaseBpodTrialsExtractor): (empty)

259 """ 

260 Calculate duration of iti from state timestamps. 

261 **Optional:** saves _ibl_trials.iti_duration.npy 

262 

263 Uses Trial end timestamp and get_response_times to calculate iti. 

264 """ 

265 save_names = '_ibl_trials.itiDuration.npy' (empty)

266 var_names = 'iti_dur' (empty)

267 

268 def _extract(self): (empty)

269 rt, _ = ResponseTimes(self.session_path).extract( 

270 save=False, task_collection=self.task_collection, bpod_trials=self.bpod_trials, settings=self.settings) 

271 ends = np.array([t['behavior_data']['Trial end timestamp'] for t in self.bpod_trials]) 

272 iti_dur = ends - rt 

273 return iti_dur 

274 

275 

276class GoCueTriggerTimes(BaseBpodTrialsExtractor): (empty)

277 """ 

278 Get trigger times of goCue from state machine. 

279 

280 Current software solution for triggering sounds uses PyBpod soft codes. 

281 Delays can be in the order of 10's of ms. This is the time when the command 

282 to play the sound was executed. To measure accurate time, either getting the 

283 sound onset from xonar soundcard sync pulse (latencies may vary). 

284 """ 

285 save_names = '_ibl_trials.goCueTrigger_times.npy' (empty)

286 var_names = 'goCueTrigger_times' (empty)

287 

288 def _extract(self): (empty)

289 if version.parse(self.settings['IBLRIG_VERSION'] or '100.0.0') >= version.parse('5.0.0'): 18 ctx1aIcjhdeqlmknoifbgp

290 goCue = np.array([tr['behavior_data']['States timestamps'] 16 ctx1Icjhdelmknoifbgp

291 ['play_tone'][0][0] for tr in self.bpod_trials]) 

292 else: 

293 goCue = np.array([tr['behavior_data']['States timestamps'] 4 ctx1acqb

294 ['closed_loop'][0][0] for tr in self.bpod_trials]) 

295 return goCue 18 ctx1aIcjhdeqlmknoifbgp

296 

297 

298class TrialType(BaseBpodTrialsExtractor): (empty)

299 save_names = '_ibl_trials.type.npy' (empty)

300 var_name = 'trial_type' (empty)

301 

302 def _extract(self): (empty)

303 trial_type = [] 

304 for tr in self.bpod_trials: 

305 if ~np.isnan(tr["behavior_data"]["States timestamps"]["reward"][0][0]): 

306 trial_type.append(1) 

307 elif ~np.isnan(tr["behavior_data"]["States timestamps"]["error"][0][0]): 

308 trial_type.append(-1) 

309 elif ~np.isnan(tr["behavior_data"]["States timestamps"]["no_go"][0][0]): 

310 trial_type.append(0) 

311 else: 

312 _logger.warning("Trial is not in set {-1, 0, 1}, appending NaN to trialType") 

313 trial_type.append(np.nan) 

314 return np.array(trial_type) 

315 

316 

317class GoCueTimes(BaseBpodTrialsExtractor): (empty)

318 """ 

319 Get trigger times of goCue from state machine. 

320 

321 Current software solution for triggering sounds uses PyBpod soft codes. 

322 Delays can be in the order of 10-100s of ms. This is the time when the command 

323 to play the sound was executed. To measure accurate time, either getting the 

324 sound onset from the future microphone OR the new xonar soundcard and 

325 setup developed by Sanworks guarantees a set latency (in testing). 

326 """ 

327 save_names = '_ibl_trials.goCue_times.npy' (empty)

328 var_names = 'goCue_times' (empty)

329 

330 def _extract(self): (empty)

331 go_cue_times = np.zeros([len(self.bpod_trials), ]) 22 ctx1avcxjshdeqltmknoifybgp

332 for ind, tr in enumerate(self.bpod_trials): 22 ctx1avcxjshdeqltmknoifybgp

333 if raw.get_port_events(tr, 'BNC2'): 22 ctx1avcxjshdeqltmknoifybgp

334 bnchigh = tr['behavior_data']['Events timestamps'].get('BNC2High', None) 21 ctx1vcxjshdeqltmknoifybgp

335 if bnchigh: 21 ctx1vcxjshdeqltmknoifybgp

336 go_cue_times[ind] = bnchigh[0] 21 ctx1vcxjshdeqltmknoifybgp

337 continue 21 ctx1vcxjshdeqltmknoifybgp

338 bnclow = tr['behavior_data']['Events timestamps'].get('BNC2Low', None) 2 ctx1bg

339 if bnclow: 2 ctx1bg

340 go_cue_times[ind] = bnclow[0] - 0.1 2 ctx1bg

341 continue 2 ctx1bg

342 go_cue_times[ind] = np.nan 

343 else: 

344 go_cue_times[ind] = np.nan 4 ctx1avcb

345 

346 nmissing = np.sum(np.isnan(go_cue_times)) 22 ctx1avcxjshdeqltmknoifybgp

347 # Check if all stim_syncs have failed to be detected 

348 if np.all(np.isnan(go_cue_times)): 22 ctx1avcxjshdeqltmknoifybgp

349 _logger.warning( 4 ctx1avcb

350 f'{self.session_path}: Missing ALL !! BNC2 TTLs ({nmissing} trials)') 

351 # Check if any stim_sync has failed be detected for every trial 

352 elif np.any(np.isnan(go_cue_times)): 21 ctx1vcxjshdeqltmknoifybgp

353 _logger.warning(f'{self.session_path}: Missing BNC2 TTLs on {nmissing} trials') 

354 

355 return go_cue_times 22 ctx1avcxjshdeqltmknoifybgp

356 

357 

358class IncludedTrials(BaseBpodTrialsExtractor): (empty)

359 save_names = '_ibl_trials.included.npy' (empty)

360 var_names = 'included' (empty)

361 

362 def _extract(self): (empty)

363 if version.parse(self.settings['IBLRIG_VERSION'] or '100.0.0') >= version.parse('5.0.0'): 17 ctx1aADEcjhqlmknoibgp

364 trials_included = self.get_included_trials_ge5( 14 ctx1ADcjhlmknoibgp

365 data=self.bpod_trials, settings=self.settings) 

366 else: 

367 trials_included = self.get_included_trials_lt5(data=self.bpod_trials) 6 ctx1aAEcqb

368 return trials_included 17 ctx1aADEcjhqlmknoibgp

369 

370 @staticmethod (empty)

371 def get_included_trials_lt5(data=False): (empty)

372 trials_included = np.ones(len(data), dtype=bool) 6 ctx1aAEcqb

373 return trials_included 6 ctx1aAEcqb

374 

375 @staticmethod (empty)

376 def get_included_trials_ge5(data=False, settings=False): (empty)

377 trials_included = np.array([True for t in data]) 14 ctx1ADcjhlmknoibgp

378 if ('SUBJECT_DISENGAGED_TRIGGERED' in settings.keys() and settings[ 14 ctx1ADcjhlmknoibgp

379 'SUBJECT_DISENGAGED_TRIGGERED'] is not False): 

380 idx = settings['SUBJECT_DISENGAGED_TRIALNUM'] - 1 8 ctx1jhmknobg

381 trials_included[idx:] = False 8 ctx1jhmknobg

382 return trials_included 14 ctx1ADcjhlmknoibgp

383 

384 

385class ItiInTimes(BaseBpodTrialsExtractor): (empty)

386 var_names = 'itiIn_times' (empty)

387 

388 def _extract(self): (empty)

389 if version.parse(self.settings["IBLRIG_VERSION"] or '100.0.0') < version.parse("5.0.0"): 17 ctx1acjhdeqlmknoifbgp

390 iti_in = np.ones(len(self.bpod_trials)) * np.nan 4 ctx1acqb

391 else: 

392 iti_in = np.array( 15 ctx1cjhdelmknoifbgp

393 [tr["behavior_data"]["States timestamps"] 

394 ["exit_state"][0][0] for tr in self.bpod_trials] 

395 ) 

396 return iti_in 17 ctx1acjhdeqlmknoifbgp

397 

398 

399class ErrorCueTriggerTimes(BaseBpodTrialsExtractor): (empty)

400 var_names = 'errorCueTrigger_times' (empty)

401 

402 def _extract(self): (empty)

403 errorCueTrigger_times = np.zeros(len(self.bpod_trials)) * np.nan 17 ctx1acjhdeqlmknoifbgp

404 for i, tr in enumerate(self.bpod_trials): 17 ctx1acjhdeqlmknoifbgp

405 nogo = tr["behavior_data"]["States timestamps"]["no_go"][0][0] 17 ctx1acjhdeqlmknoifbgp

406 error = tr["behavior_data"]["States timestamps"]["error"][0][0] 17 ctx1acjhdeqlmknoifbgp

407 if np.all(~np.isnan(nogo)): 17 ctx1acjhdeqlmknoifbgp

408 errorCueTrigger_times[i] = nogo 10 ctx1chdemnoibg

409 elif np.all(~np.isnan(error)): 17 ctx1acjhdeqlmknoifbgp

410 errorCueTrigger_times[i] = error 17 ctx1acjhdeqlmknoifbgp

411 return errorCueTrigger_times 17 ctx1acjhdeqlmknoifbgp

412 

413 

414class StimFreezeTriggerTimes(BaseBpodTrialsExtractor): (empty)

415 var_names = 'stimFreezeTrigger_times' (empty)

416 

417 def _extract(self): (empty)

418 if version.parse(self.settings["IBLRIG_VERSION"] or '100.0.0') < version.parse("6.2.5"): 18 ctx1acrjhdeqlmknoifbgp

419 return np.ones(len(self.bpod_trials)) * np.nan 10 ctx1acrqmnobgp

420 freeze_reward = np.array( 9 ctx1jhdelkifg

421 [ 

422 True 

423 if np.all(~np.isnan(tr["behavior_data"]["States timestamps"]["freeze_reward"][0])) 

424 else False 

425 for tr in self.bpod_trials 

426 ] 

427 ) 

428 freeze_error = np.array( 9 ctx1jhdelkifg

429 [ 

430 True 

431 if np.all(~np.isnan(tr["behavior_data"]["States timestamps"]["freeze_error"][0])) 

432 else False 

433 for tr in self.bpod_trials 

434 ] 

435 ) 

436 no_go = np.array( 9 ctx1jhdelkifg

437 [ 

438 True 

439 if np.all(~np.isnan(tr["behavior_data"]["States timestamps"]["no_go"][0])) 

440 else False 

441 for tr in self.bpod_trials 

442 ] 

443 ) 

444 assert (np.sum(freeze_error) + np.sum(freeze_reward) + 9 ctx1jhdelkifg

445 np.sum(no_go) == len(self.bpod_trials)) 

446 stimFreezeTrigger = np.array([]) 9 ctx1jhdelkifg

447 for r, e, n, tr in zip(freeze_reward, freeze_error, no_go, self.bpod_trials): 9 ctx1jhdelkifg

448 if n: 9 ctx1jhdelkifg

449 stimFreezeTrigger = np.append(stimFreezeTrigger, np.nan) 5 ctx1hdeig

450 continue 5 ctx1hdeig

451 state = "freeze_reward" if r else "freeze_error" 9 ctx1jhdelkifg

452 stimFreezeTrigger = np.append( 9 ctx1jhdelkifg

453 stimFreezeTrigger, tr["behavior_data"]["States timestamps"][state][0][0] 

454 ) 

455 return stimFreezeTrigger 9 ctx1jhdelkifg

456 

457 

458class StimOffTriggerTimes(BaseBpodTrialsExtractor): (empty)

459 var_names = 'stimOffTrigger_times' (empty)

460 save_names = '_ibl_trials.stimOnTrigger_times.npy' (empty)

461 

462 def _extract(self): (empty)

463 if version.parse(self.settings["IBLRIG_VERSION"] or '100.0.0') >= version.parse("6.2.5"): 18 ctx1acrjhdeqlmknoifbgp

464 stim_off_trigger_state = "hide_stim" 9 ctx1jhdelkifg

465 elif version.parse(self.settings["IBLRIG_VERSION"]) >= version.parse("5.0.0"): 10 ctx1acrqmnobgp

466 stim_off_trigger_state = "exit_state" 8 ctx1crmnobgp

467 else: 

468 stim_off_trigger_state = "trial_start" 5 ctx1acrqb

469 

470 stimOffTrigger_times = np.array( 18 ctx1acrjhdeqlmknoifbgp

471 [tr["behavior_data"]["States timestamps"][stim_off_trigger_state][0][0] 

472 for tr in self.bpod_trials] 

473 ) 

474 # If pre version 5.0.0 no specific nogo Off trigger was given, just return trial_starts 

475 if stim_off_trigger_state == "trial_start": 18 ctx1acrjhdeqlmknoifbgp

476 return stimOffTrigger_times 5 ctx1acrqb

477 

478 no_goTrigger_times = np.array( 16 ctx1crjhdelmknoifbgp

479 [tr["behavior_data"]["States timestamps"]["no_go"][0][0] for tr in self.bpod_trials] 

480 ) 

481 # Stim off trigs are either in their own state or in the no_go state if the 

482 # mouse did not move, if the stim_off_trigger_state always exist 

483 # (exit_state or trial_start) 

484 # no NaNs will happen, NaNs might happen in at last trial if 

485 # session was stopped after response 

486 # if stim_off_trigger_state == "hide_stim": 

487 # assert all(~np.isnan(no_goTrigger_times) == np.isnan(stimOffTrigger_times)) 

488 # Patch with the no_go states trig times 

489 stimOffTrigger_times[~np.isnan(no_goTrigger_times)] = no_goTrigger_times[ 16 ctx1crjhdelmknoifbgp

490 ~np.isnan(no_goTrigger_times) 

491 ] 

492 return stimOffTrigger_times 16 ctx1crjhdelmknoifbgp

493 

494 

495class StimOnTriggerTimes(BaseBpodTrialsExtractor): (empty)

496 save_names = '_ibl_trials.stimOnTrigger_times.npy' (empty)

497 var_names = 'stimOnTrigger_times' (empty)

498 

499 def _extract(self): (empty)

500 # Get the stim_on_state that triggers the onset of the stim 

501 stim_on_state = np.array([tr['behavior_data']['States timestamps'] 21 ctx1aLcrjshdeqltmknoifbgp

502 ['stim_on'][0] for tr in self.bpod_trials]) 

503 return stim_on_state[:, 0].T 21 ctx1aLcrjshdeqltmknoifbgp

504 

505 

506class StimOnTimes_deprecated(BaseBpodTrialsExtractor): (empty)

507 save_names = '_ibl_trials.stimOn_times.npy' (empty)

508 var_names = 'stimOn_times' (empty)

509 

510 def _extract(self): (empty)

511 """ 

512 Find the time of the state machine command to turn on the stim 

513 (state stim_on start or rotary_encoder_event2) 

514 Find the next frame change from the photodiode after that TS. 

515 Screen is not displaying anything until then. 

516 (Frame changes are in BNC1 High and BNC1 Low) 

517 """ 

518 # Version check 

519 _logger.warning("Deprecation Warning: this is an old version of stimOn extraction." 

520 "From version 5., use StimOnOffFreezeTimes") 

521 if version.parse(self.settings['IBLRIG_VERSION'] or '100.0.0') >= version.parse('5.0.0'): 

522 stimOn_times = self.get_stimOn_times_ge5(self.session_path, data=self.bpod_trials, 

523 task_collection=self.task_collection) 

524 else: 

525 stimOn_times = self.get_stimOn_times_lt5(self.session_path, data=self.bpod_trials, 

526 task_collection=self.task_collection) 

527 return np.array(stimOn_times) 

528 

529 @staticmethod (empty)

530 def get_stimOn_times_ge5(session_path, data=False, task_collection='raw_behavior_data'): (empty)

531 """ 

532 Find first and last stim_sync pulse of the trial. 

533 stimOn_times should be the first after the stim_on state. 

534 (Stim updates are in BNC1High and BNC1Low - frame2TTL device) 

535 Check that all trials have frame changes. 

536 Find length of stim_on_state [start, stop]. 

537 If either check fails the HW device failed to detect the stim_sync square change 

538 Substitute that trial's missing or incorrect value with a NaN. 

539 return stimOn_times 

540 """ 

541 if not data: 

542 data = raw.load_data(session_path, task_collection=task_collection) 

543 # Get all stim_sync events detected 

544 stim_sync_all = [raw.get_port_events(tr, 'BNC1') for tr in data] 

545 stim_sync_all = [np.array(x) for x in stim_sync_all] 

546 # Get the stim_on_state that triggers the onset of the stim 

547 stim_on_state = np.array([tr['behavior_data']['States timestamps'] 

548 ['stim_on'][0] for tr in data]) 

549 

550 stimOn_times = np.array([]) 

551 for sync, on, off in zip( 

552 stim_sync_all, stim_on_state[:, 0], stim_on_state[:, 1]): 

553 pulse = sync[np.where(np.bitwise_and((sync > on), (sync <= off)))] 

554 if pulse.size == 0: 

555 stimOn_times = np.append(stimOn_times, np.nan) 

556 else: 

557 stimOn_times = np.append(stimOn_times, pulse) 

558 

559 nmissing = np.sum(np.isnan(stimOn_times)) 

560 # Check if all stim_syncs have failed to be detected 

561 if np.all(np.isnan(stimOn_times)): 

562 _logger.error(f'{session_path}: Missing ALL BNC1 TTLs ({nmissing} trials)') 

563 

564 # Check if any stim_sync has failed be detected for every trial 

565 if np.any(np.isnan(stimOn_times)): 

566 _logger.warning(f'{session_path}: Missing BNC1 TTLs on {nmissing} trials') 

567 

568 return stimOn_times 

569 

570 @staticmethod (empty)

571 def get_stimOn_times_lt5(session_path, data=False, task_collection='raw_behavior_data'): (empty)

572 """ 

573 Find the time of the statemachine command to turn on the stim 

574 (state stim_on start or rotary_encoder_event2) 

575 Find the next frame change from the photodiode after that TS. 

576 Screen is not displaying anything until then. 

577 (Frame changes are in BNC1High and BNC1Low) 

578 """ 

579 if not data: 

580 data = raw.load_data(session_path, task_collection=task_collection) 

581 stim_on = [] 

582 bnc_h = [] 

583 bnc_l = [] 

584 for tr in data: 

585 stim_on.append(tr['behavior_data']['States timestamps']['stim_on'][0][0]) 

586 if 'BNC1High' in tr['behavior_data']['Events timestamps'].keys(): 

587 bnc_h.append(np.array(tr['behavior_data'] 

588 ['Events timestamps']['BNC1High'])) 

589 else: 

590 bnc_h.append(np.array([np.NINF])) 

591 if 'BNC1Low' in tr['behavior_data']['Events timestamps'].keys(): 

592 bnc_l.append(np.array(tr['behavior_data'] 

593 ['Events timestamps']['BNC1Low'])) 

594 else: 

595 bnc_l.append(np.array([np.NINF])) 

596 

597 stim_on = np.array(stim_on) 

598 bnc_h = np.array(bnc_h, dtype=object) 

599 bnc_l = np.array(bnc_l, dtype=object) 

600 

601 count_missing = 0 

602 stimOn_times = np.zeros_like(stim_on) 

603 for i in range(len(stim_on)): 

604 hl = np.sort(np.concatenate([bnc_h[i], bnc_l[i]])) 

605 stot = hl[hl > stim_on[i]] 

606 if np.size(stot) == 0: 

607 stot = np.array([np.nan]) 

608 count_missing += 1 

609 stimOn_times[i] = stot[0] 

610 

611 if np.all(np.isnan(stimOn_times)): 

612 _logger.error(f'{session_path}: Missing ALL BNC1 TTLs ({count_missing} trials)') 

613 

614 if count_missing > 0: 

615 _logger.warning(f'{session_path}: Missing BNC1 TTLs on {count_missing} trials') 

616 

617 return np.array(stimOn_times) 

618 

619 

620class StimOnOffFreezeTimes(BaseBpodTrialsExtractor): (empty)

621 """ 

622 Extracts stim on / off and freeze times from Bpod BNC1 detected fronts. 

623 

624 Each stimulus event is the first detected front of the BNC1 signal after the trigger state, but before the next 

625 trigger state. 

626 """ 

627 save_names = ('_ibl_trials.stimOn_times.npy', '_ibl_trials.stimOff_times.npy', None) (empty)

628 var_names = ('stimOn_times', 'stimOff_times', 'stimFreeze_times') (empty)

629 

630 def _extract(self): (empty)

631 choice = Choice(self.session_path).extract( 18 ctx1acrjhdeqlmknoifbgp

632 bpod_trials=self.bpod_trials, task_collection=self.task_collection, settings=self.settings, save=False 

633 )[0] 

634 stimOnTrigger = StimOnTriggerTimes(self.session_path).extract( 18 ctx1acrjhdeqlmknoifbgp

635 bpod_trials=self.bpod_trials, task_collection=self.task_collection, settings=self.settings, save=False 

636 )[0] 

637 stimFreezeTrigger = StimFreezeTriggerTimes(self.session_path).extract( 18 ctx1acrjhdeqlmknoifbgp

638 bpod_trials=self.bpod_trials, task_collection=self.task_collection, settings=self.settings, save=False 

639 )[0] 

640 stimOffTrigger = StimOffTriggerTimes(self.session_path).extract( 18 ctx1acrjhdeqlmknoifbgp

641 bpod_trials=self.bpod_trials, task_collection=self.task_collection, settings=self.settings, save=False 

642 )[0] 

643 f2TTL = [raw.get_port_events(tr, name='BNC1') for tr in self.bpod_trials] 18 ctx1acrjhdeqlmknoifbgp

644 assert stimOnTrigger.size == stimFreezeTrigger.size == stimOffTrigger.size == choice.size == len(f2TTL) 18 ctx1acrjhdeqlmknoifbgp

645 assert all(stimOnTrigger < np.nan_to_num(stimFreezeTrigger, nan=np.inf)) and \ 18 ctx1acrjhdeqlmknoifbgp

646 all(np.nan_to_num(stimFreezeTrigger, nan=-np.inf) < stimOffTrigger) 

647 

648 stimOn_times = np.array([]) 18 ctx1acrjhdeqlmknoifbgp

649 stimOff_times = np.array([]) 18 ctx1acrjhdeqlmknoifbgp

650 stimFreeze_times = np.array([]) 18 ctx1acrjhdeqlmknoifbgp

651 has_freeze = version.parse(self.settings.get('IBLRIG_VERSION', '0')) >= version.parse('6.2.5') 18 ctx1acrjhdeqlmknoifbgp

652 for tr, on, freeze, off, c in zip(f2TTL, stimOnTrigger, stimFreezeTrigger, stimOffTrigger, choice): 18 ctx1acrjhdeqlmknoifbgp

653 tr = np.array(tr) 18 ctx1acrjhdeqlmknoifbgp

654 # stim on 

655 lim = freeze if has_freeze else off 18 ctx1acrjhdeqlmknoifbgp

656 idx, = np.where(np.logical_and(on < tr, tr < lim)) 18 ctx1acrjhdeqlmknoifbgp

657 stimOn_times = np.append(stimOn_times, tr[idx[0]] if idx.size > 0 else np.nan) 18 ctx1acrjhdeqlmknoifbgp

658 # stim off 

659 idx, = np.where(off < tr) 18 ctx1acrjhdeqlmknoifbgp

660 stimOff_times = np.append(stimOff_times, tr[idx[0]] if idx.size > 0 else np.nan) 18 ctx1acrjhdeqlmknoifbgp

661 # stim freeze - take last event before off trigger 

662 if has_freeze: 18 ctx1acrjhdeqlmknoifbgp

663 idx, = np.where(np.logical_and(freeze < tr, tr < off)) 9 ctx1jhdelkifg

664 stimFreeze_times = np.append(stimFreeze_times, tr[idx[-1]] if idx.size > 0 else np.nan) 9 ctx1jhdelkifg