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

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) 17 ctx1avcigdepkljmnhbfo

33 for i, t in enumerate(self.bpod_trials): 17 ctx1avcigdepkljmnhbfo

34 state_names = ['correct', 'error', 'no_go', 'omit_correct', 'omit_error', 'omit_no_go'] 17 ctx1avcigdepkljmnhbfo

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

36 assert np.sum(list(outcome.values())) == 1 17 ctx1avcigdepkljmnhbfo

37 outcome = next(k for k in outcome if outcome[k]) 17 ctx1avcigdepkljmnhbfo

38 if outcome == 'correct': 17 ctx1avcigdepkljmnhbfo

39 feedbackType[i] = 1 16 ctx1avcigdepkljmnhbf

40 elif outcome in ['error', 'no_go']: 17 ctx1avcigdepkljmnhbfo

41 feedbackType[i] = -1 17 ctx1avcigdepkljmnhbfo

42 return feedbackType 17 ctx1avcigdepkljmnhbfo

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): 5 ctx1Ecdeb

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

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

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

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 ctx1Ecdeb

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

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

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

68 

69 return contrastLeft, contrastRight 5 ctx1Ecdeb

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]) 9 ctx1Lcidepkhb

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]) 18 ctx1aAcqigdepkljmnhbfo

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

100 trial_nogo = np.array( 18 ctx1aAcqigdepkljmnhbfo

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

102 for t in self.bpod_trials]) 

103 choice = sitm_side.copy() 18 ctx1aAcqigdepkljmnhbfo

104 choice[trial_correct] = -choice[trial_correct] 18 ctx1aAcqigdepkljmnhbfo

105 choice[trial_nogo] = 0 18 ctx1aAcqigdepkljmnhbfo

106 choice = choice.astype(int) 18 ctx1aAcqigdepkljmnhbfo

107 return choice 18 ctx1aAcqigdepkljmnhbfo

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): 5 ctx1Bcdeb

125 if 'debias_trial' in trial: 5 ctx1Bcdeb

126 return trial['debias_trial'] 

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

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

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) 5 ctx1Bcdeb

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

138 return repNum 5 ctx1Bcdeb

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'] 17 ctx1aFcigdepkljmnhbfo

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

154 reward_volume = np.array(trial_volume).astype(np.float64) 17 ctx1aFcigdepkljmnhbfo

155 assert len(reward_volume) == len(self.bpod_trials) 17 ctx1aFcigdepkljmnhbfo

156 return reward_volume 17 ctx1aFcigdepkljmnhbfo

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 ctx1aycpb

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

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

176 for tr in data] 

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

178 for tr in data] 

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

180 for tr in data] 

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

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

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

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

185 

186 return np.array(merge) 5 ctx1aycpb

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: 17 ctx1tcirgdeksljmnhbfo

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

195 missed_bnc2 = 0 17 ctx1tcirgdeksljmnhbfo

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

197 

198 for ind, tr in enumerate(data): 17 ctx1tcirgdeksljmnhbfo

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

200 if not st: 17 ctx1tcirgdeksljmnhbfo

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

202 missed_bnc2 += 1 2 ctx1bf

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) 17 ctx1tcirgdeksljmnhbfo

205 rw_times[ind] = tr['behavior_data']['States timestamps']['reward'][0][0] 17 ctx1tcirgdeksljmnhbfo

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 17 ctx1tcirgdeksljmnhbfo

208 if missed_bnc2 == len(data): 17 ctx1tcirgdeksljmnhbfo

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

210 merge *= np.nan 17 ctx1tcirgdeksljmnhbfo

211 merge[~np.isnan(rw_times)] = rw_times[~np.isnan(rw_times)] 17 ctx1tcirgdeksljmnhbfo

212 merge[~np.isnan(err_sound_times)] = err_sound_times[~np.isnan(err_sound_times)] 17 ctx1tcirgdeksljmnhbfo

213 

214 return merge 17 ctx1tcirgdeksljmnhbfo

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'): 20 ctx1atycirgdepksljmnhbfo

219 merge = self.get_feedback_times_ge5(self.session_path, task_collection=self.task_collection, data=self.bpod_trials) 17 ctx1tcirgdeksljmnhbfo

220 else: 

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

222 return np.array(merge) 20 ctx1atycirgdepksljmnhbfo

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] 17 ctx1aGcigdepkljmnhbfo

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

239 return np.array([starts, ends]).T 17 ctx1aGcigdepkljmnhbfo

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] 17 ctx1aJcigdepkljmnhbfo

254 for tr in self.bpod_trials]) 

255 return rt 17 ctx1aJcigdepkljmnhbfo

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'): 17 ctx1aHcigdepkljmnhbfo

290 goCue = np.array([tr['behavior_data']['States timestamps'] 15 ctx1Hcigdekljmnhbfo

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

292 else: 

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

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

295 return goCue 17 ctx1aHcigdepkljmnhbfo

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), ]) 21 ctx1aucwirgdepksljmnhxbfo

332 for ind, tr in enumerate(self.bpod_trials): 21 ctx1aucwirgdepksljmnhxbfo

333 if raw.get_port_events(tr, 'BNC2'): 21 ctx1aucwirgdepksljmnhxbfo

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

335 if bnchigh: 20 ctx1ucwirgdepksljmnhxbfo

336 go_cue_times[ind] = bnchigh[0] 20 ctx1ucwirgdepksljmnhxbfo

337 continue 20 ctx1ucwirgdepksljmnhxbfo

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

339 if bnclow: 2 ctx1bf

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

341 continue 2 ctx1bf

342 go_cue_times[ind] = np.nan 

343 else: 

344 go_cue_times[ind] = np.nan 4 ctx1aucb

345 

346 nmissing = np.sum(np.isnan(go_cue_times)) 21 ctx1aucwirgdepksljmnhxbfo

347 # Check if all stim_syncs have failed to be detected 

348 if np.all(np.isnan(go_cue_times)): 21 ctx1aucwirgdepksljmnhxbfo

349 _logger.warning( 4 ctx1aucb

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)): 20 ctx1ucwirgdepksljmnhxbfo

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

354 

355 return go_cue_times 21 ctx1aucwirgdepksljmnhxbfo

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 ctx1azCDcigpkljmnhbfo

364 trials_included = self.get_included_trials_ge5( 14 ctx1zCcigkljmnhbfo

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

366 else: 

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

368 return trials_included 17 ctx1azCDcigpkljmnhbfo

369 

370 @staticmethod (empty)

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

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

373 return trials_included 6 ctx1azDcpb

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 ctx1zCcigkljmnhbfo

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

379 'SUBJECT_DISENGAGED_TRIGGERED'] is not False): 

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

381 trials_included[idx:] = False 8 ctx1igljmnbf

382 return trials_included 14 ctx1zCcigkljmnhbfo

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"): 16 ctx1acigdepkljmnhbfo

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

391 else: 

392 iti_in = np.array( 14 ctx1cigdekljmnhbfo

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

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

395 ) 

396 return iti_in 16 ctx1acigdepkljmnhbfo

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 16 ctx1acigdepkljmnhbfo

404 for i, tr in enumerate(self.bpod_trials): 16 ctx1acigdepkljmnhbfo

405 nogo = tr["behavior_data"]["States timestamps"]["no_go"][0][0] 16 ctx1acigdepkljmnhbfo

406 error = tr["behavior_data"]["States timestamps"]["error"][0][0] 16 ctx1acigdepkljmnhbfo

407 if np.all(~np.isnan(nogo)): 16 ctx1acigdepkljmnhbfo

408 errorCueTrigger_times[i] = nogo 10 ctx1cgdelmnhbf

409 elif np.all(~np.isnan(error)): 16 ctx1acigdepkljmnhbfo

410 errorCueTrigger_times[i] = error 16 ctx1acigdepkljmnhbfo

411 return errorCueTrigger_times 16 ctx1acigdepkljmnhbfo

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"): 17 ctx1acqigdepkljmnhbfo

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

420 freeze_reward = np.array( 8 ctx1igdekjhf

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( 8 ctx1igdekjhf

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( 8 ctx1igdekjhf

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) + 8 ctx1igdekjhf

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

446 stimFreezeTrigger = np.array([]) 8 ctx1igdekjhf

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

448 if n: 8 ctx1igdekjhf

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

450 continue 5 ctx1gdehf

451 state = "freeze_reward" if r else "freeze_error" 8 ctx1igdekjhf

452 stimFreezeTrigger = np.append( 8 ctx1igdekjhf

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

454 ) 

455 return stimFreezeTrigger 8 ctx1igdekjhf

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"): 17 ctx1acqigdepkljmnhbfo

464 stim_off_trigger_state = "hide_stim" 8 ctx1igdekjhf

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

466 stim_off_trigger_state = "exit_state" 8 ctx1cqlmnbfo

467 else: 

468 stim_off_trigger_state = "trial_start" 5 ctx1acqpb

469 

470 stimOffTrigger_times = np.array( 17 ctx1acqigdepkljmnhbfo

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": 17 ctx1acqigdepkljmnhbfo

476 return stimOffTrigger_times 5 ctx1acqpb

477 

478 no_goTrigger_times = np.array( 15 ctx1cqigdekljmnhbfo

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[ 15 ctx1cqigdekljmnhbfo

490 ~np.isnan(no_goTrigger_times) 

491 ] 

492 return stimOffTrigger_times 15 ctx1cqigdekljmnhbfo

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'] 20 ctx1aKcqirgdepksljmnhbfo

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

503 return stim_on_state[:, 0].T 20 ctx1aKcqirgdepksljmnhbfo

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( 17 ctx1acqigdepkljmnhbfo

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( 17 ctx1acqigdepkljmnhbfo

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( 17 ctx1acqigdepkljmnhbfo

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( 17 ctx1acqigdepkljmnhbfo

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] 17 ctx1acqigdepkljmnhbfo

644 assert stimOnTrigger.size == stimFreezeTrigger.size == stimOffTrigger.size == choice.size == len(f2TTL) 17 ctx1acqigdepkljmnhbfo

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

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

647 

648 stimOn_times = np.array([]) 17 ctx1acqigdepkljmnhbfo

649 stimOff_times = np.array([]) 17 ctx1acqigdepkljmnhbfo

650 stimFreeze_times = np.array([]) 17 ctx1acqigdepkljmnhbfo

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

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

653 tr = np.array(tr) 17 ctx1acqigdepkljmnhbfo

654 # stim on 

655 lim = freeze if has_freeze else off 17 ctx1acqigdepkljmnhbfo

656 idx, = np.where(np.logical_and(on < tr, tr < lim)) 17 ctx1acqigdepkljmnhbfo

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

658 # stim off 

659 idx, = np.where(off < tr) 17 ctx1acqigdepkljmnhbfo

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

661 # stim freeze - take last event before off trigger 

662 if has_freeze: 17 ctx1acqigdepkljmnhbfo

663 idx, = np.where(np.logical_and(freeze < tr, tr < off)) 8 ctx1igdekjhf

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