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 ctx1aucigdepkljmnhbfo

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

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

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

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

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

38 if outcome == 'correct': 17 ctx1aucigdepkljmnhbfo

39 feedbackType[i] = 1 16 ctx1aucigdepkljmnhbf

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

41 feedbackType[i] = -1 17 ctx1aucigdepkljmnhbfo

42 return feedbackType 17 ctx1aucigdepkljmnhbfo

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 ctx1Dcdeb

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 ctx1Dcdeb

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

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

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

68 

69 return contrastLeft, contrastRight 5 ctx1Dcdeb

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 ctx1Kcidepkhb

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 ctx1azcqigdepkljmnhbfo

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

100 trial_nogo = np.array( 18 ctx1azcqigdepkljmnhbfo

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 ctx1azcqigdepkljmnhbfo

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

105 choice[trial_nogo] = 0 18 ctx1azcqigdepkljmnhbfo

106 choice = choice.astype(int) 18 ctx1azcqigdepkljmnhbfo

107 return choice 18 ctx1azcqigdepkljmnhbfo

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 ctx1Acdeb

125 if 'debias_trial' in trial: 5 ctx1Acdeb

126 return trial['debias_trial'] 

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

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

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 ctx1Acdeb

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

138 return repNum 5 ctx1Acdeb

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 ctx1aEcigdepkljmnhbfo

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

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

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

156 return reward_volume 17 ctx1aEcigdepkljmnhbfo

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 ctx1axcpb

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

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

176 for tr in data] 

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

178 for tr in data] 

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

180 for tr in data] 

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

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

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

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

185 

186 return np.array(merge) 5 ctx1axcpb

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: 16 ctx1scirgdekljmnhbfo

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

195 missed_bnc2 = 0 16 ctx1scirgdekljmnhbfo

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

197 

198 for ind, tr in enumerate(data): 16 ctx1scirgdekljmnhbfo

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

200 if not st: 16 ctx1scirgdekljmnhbfo

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) 16 ctx1scirgdekljmnhbfo

205 rw_times[ind] = tr['behavior_data']['States timestamps']['reward'][0][0] 16 ctx1scirgdekljmnhbfo

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

208 if missed_bnc2 == len(data): 16 ctx1scirgdekljmnhbfo

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

210 merge *= np.nan 16 ctx1scirgdekljmnhbfo

211 merge[~np.isnan(rw_times)] = rw_times[~np.isnan(rw_times)] 16 ctx1scirgdekljmnhbfo

212 merge[~np.isnan(err_sound_times)] = err_sound_times[~np.isnan(err_sound_times)] 16 ctx1scirgdekljmnhbfo

213 

214 return merge 16 ctx1scirgdekljmnhbfo

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'): 19 ctx1asxcirgdepkljmnhbfo

219 merge = self.get_feedback_times_ge5(self.session_path, task_collection=self.task_collection, data=self.bpod_trials) 16 ctx1scirgdekljmnhbfo

220 else: 

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

222 return np.array(merge) 19 ctx1asxcirgdepkljmnhbfo

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 ctx1aFcigdepkljmnhbfo

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

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

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 ctx1aIcigdepkljmnhbfo

254 for tr in self.bpod_trials]) 

255 return rt 17 ctx1aIcigdepkljmnhbfo

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 ctx1aGcigdepkljmnhbfo

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

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 ctx1aGcigdepkljmnhbfo

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), ]) 20 ctx1atcvirgdepkljmnhwbfo

332 for ind, tr in enumerate(self.bpod_trials): 20 ctx1atcvirgdepkljmnhwbfo

333 if raw.get_port_events(tr, 'BNC2'): 20 ctx1atcvirgdepkljmnhwbfo

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

335 if bnchigh: 19 ctx1tcvirgdepkljmnhwbfo

336 go_cue_times[ind] = bnchigh[0] 19 ctx1tcvirgdepkljmnhwbfo

337 continue 19 ctx1tcvirgdepkljmnhwbfo

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 ctx1atcb

345 

346 nmissing = np.sum(np.isnan(go_cue_times)) 20 ctx1atcvirgdepkljmnhwbfo

347 # Check if all stim_syncs have failed to be detected 

348 if np.all(np.isnan(go_cue_times)): 20 ctx1atcvirgdepkljmnhwbfo

349 _logger.warning( 4 ctx1atcb

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)): 19 ctx1tcvirgdepkljmnhwbfo

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

354 

355 return go_cue_times 20 ctx1atcvirgdepkljmnhwbfo

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 ctx1ayBCcigpkljmnhbfo

364 trials_included = self.get_included_trials_ge5( 14 ctx1yBcigkljmnhbfo

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

366 else: 

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

368 return trials_included 17 ctx1ayBCcigpkljmnhbfo

369 

370 @staticmethod (empty)

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

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

373 return trials_included 6 ctx1ayCcpb

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 ctx1yBcigkljmnhbfo

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

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 ctx1yBcigkljmnhbfo

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'] 19 ctx1aJcqirgdepkljmnhbfo

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

503 return stim_on_state[:, 0].T 19 ctx1aJcqirgdepkljmnhbfo

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