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

1import logging (empty)

2import numpy as np (empty)

3from pkg_resources import parse_version (empty)

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

5 

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

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

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

9 

10 

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

12__all__ = ['TrainingTrials', 'extract_all'] (empty)

13 

14 

15class FeedbackType(BaseBpodTrialsExtractor): (empty)

16 """ 

17 Get the feedback that was delivered to subject. 

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

19 

20 Checks in raw datafile for error and reward state. 

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

22 been triggered. 

23 

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

25 Sets feedbackType to +1 if reward state was triggered 

26 """ 

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

28 var_names = 'feedbackType' (empty)

29 

30 def _extract(self): (empty)

31 feedbackType = np.zeros(len(self.bpod_trials), np.int64) 25 ctx1adFcotjkfgiqrsuplmnwxbehv

32 for i, t in enumerate(self.bpod_trials): 25 ctx1adFcotjkfgiqrsuplmnwxbehv

33 state_names = ['correct', 'error', 'no_go', 'omit_correct', 'omit_error', 'omit_no_go'] 25 ctx1adFcotjkfgiqrsuplmnwxbehv

34 outcome = {sn: ~np.isnan(t['behavior_data']['States timestamps'].get(sn, [[np.NaN]])[0][0]) for sn in state_names} 25 ctx1adFcotjkfgiqrsuplmnwxbehv

35 assert np.sum(list(outcome.values())) == 1 25 ctx1adFcotjkfgiqrsuplmnwxbehv

36 outcome = next(k for k in outcome if outcome[k]) 25 ctx1adFcotjkfgiqrsuplmnwxbehv

37 if outcome == 'correct': 25 ctx1adFcotjkfgiqrsuplmnwxbehv

38 feedbackType[i] = 1 24 ctx1adFcotjkfgiqrsuplmnwxbeh

39 elif outcome in ['error', 'no_go']: 25 ctx1adFcotjkfgiqrsuplmnwxbehv

40 feedbackType[i] = -1 25 ctx1adFcotjkfgiqrsuplmnwxbehv

41 return feedbackType 25 ctx1adFcotjkfgiqrsuplmnwxbehv

42 

43 

44class ContrastLR(BaseBpodTrialsExtractor): (empty)

45 """ 

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

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

48 

49 Uses signed_contrast to create left and right contrast vectors. 

50 """ 

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

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

53 

54 def _extract(self): (empty)

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

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

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

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

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

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

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

62 else: 

63 contrastLeft = np.array([t['contrast']['value'] if np.sign( 6 ctx1dNcfgb

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

65 contrastRight = np.array([t['contrast']['value'] if np.sign( 6 ctx1dNcfgb

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

67 

68 return contrastLeft, contrastRight 6 ctx1dNcfgb

69 

70 

71class ProbabilityLeft(BaseBpodTrialsExtractor): (empty)

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

73 var_names = 'probabilityLeft' (empty)

74 

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

76 return np.array([t['stim_probability_left'] for t in self.bpod_trials]) 24 ctx1adUcojkfgiqrsuplmnwxbehv

77 

78 

79class Choice(BaseBpodTrialsExtractor): (empty)

80 """ 

81 Get the subject's choice in every trial. 

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

83 

84 Uses signed_contrast and trial_correct. 

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

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

87 0 is a no_go trial 

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

89 of the position. 

90 

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

92 """ 

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

94 var_names = 'choice' (empty)

95 

96 def _extract(self): (empty)

97 sitm_side = np.array([np.sign(t['position']) for t in self.bpod_trials]) 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

98 trial_correct = np.array([t['trial_correct'] for t in self.bpod_trials]) 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

99 trial_nogo = np.array( 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

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

101 for t in self.bpod_trials]) 

102 choice = sitm_side.copy() 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

103 choice[trial_correct] = -choice[trial_correct] 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

104 choice[trial_nogo] = 0 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

105 choice = choice.astype(int) 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

106 return choice 26 ctx1adKcAotjkfgiqrsuplmnwxbehv

107 

108 

109class RepNum(BaseBpodTrialsExtractor): (empty)

110 """ 

111 Count the consecutive repeated trials. 

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

113 

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

115 

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

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

118 """ 

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

120 var_names = 'repNum' (empty)

121 

122 def _extract(self): (empty)

123 def get_trial_repeat(trial): 6 ctx1dDcfgb

124 if 'debias_trial' in trial: 6 ctx1dDcfgb

125 return trial['debias_trial'] 

126 else: 

127 return trial['contrast']['type'] == 'RepeatContrast' 6 ctx1dDcfgb

128 

129 trial_repeated = np.array(list(map(get_trial_repeat, self.bpod_trials))).astype(int) 6 ctx1dDcfgb

130 repNum = trial_repeated.copy() 6 ctx1dDcfgb

131 c = 0 6 ctx1dDcfgb

132 for i in range(len(trial_repeated)): 6 ctx1dDcfgb

133 if trial_repeated[i] == 0: 6 ctx1dDcfgb

134 c = 0 6 ctx1dDcfgb

135 repNum[i] = 0 6 ctx1dDcfgb

136 continue 6 ctx1dDcfgb

137 c += 1 6 ctx1dDcfgb

138 repNum[i] = c 6 ctx1dDcfgb

139 return repNum 6 ctx1dDcfgb

140 

141 

142class RewardVolume(BaseBpodTrialsExtractor): (empty)

143 """ 

144 Load reward volume delivered for each trial. 

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

146 

147 Uses reward_current to accumulate the amount of 

148 """ 

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

150 var_names = 'rewardVolume' (empty)

151 

152 def _extract(self): (empty)

153 trial_volume = [x['reward_amount'] 25 ctx1adOcotjkfgiqrsuplmnwxbehv

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

155 reward_volume = np.array(trial_volume).astype(np.float64) 25 ctx1adOcotjkfgiqrsuplmnwxbehv

156 assert len(reward_volume) == len(self.bpod_trials) 25 ctx1adOcotjkfgiqrsuplmnwxbehv

157 return reward_volume 25 ctx1adOcotjkfgiqrsuplmnwxbehv

158 

159 

160class FeedbackTimes(BaseBpodTrialsExtractor): (empty)

161 """ 

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

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

164 

165 Gets reward and error state init times vectors, 

166 checks if theintersection of nans is empty, then 

167 merges the 2 vectors. 

168 """ 

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

170 var_names = 'feedback_times' (empty)

171 

172 @staticmethod (empty)

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

174 if not data: 9 ctx1adIciwxbe

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

176 rw_times = [tr['behavior_data']['States timestamps']['reward'][0][0] 9 ctx1adIciwxbe

177 for tr in data] 

178 err_times = [tr['behavior_data']['States timestamps']['error'][0][0] 9 ctx1adIciwxbe

179 for tr in data] 

180 nogo_times = [tr['behavior_data']['States timestamps']['no_go'][0][0] 9 ctx1adIciwxbe

181 for tr in data] 

182 assert sum(np.isnan(rw_times) & 9 ctx1adIciwxbe

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

184 merge = np.array([np.array(times)[~np.isnan(times)] for times in 9 ctx1adIciwxbe

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

186 

187 return np.array(merge) 9 ctx1adIciwxbe

188 

189 @staticmethod (empty)

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

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

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

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

194 if not data: 22 ctx1dCcotjykfgqrsuplmnbehv

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

196 missed_bnc2 = 0 22 ctx1dCcotjykfgqrsuplmnbehv

197 rw_times, err_sound_times, merge = [np.zeros([len(data), ]) for _ in range(3)] 22 ctx1dCcotjykfgqrsuplmnbehv

198 

199 for ind, tr in enumerate(data): 22 ctx1dCcotjykfgqrsuplmnbehv

200 st = tr['behavior_data']['Events timestamps'].get('BNC2High', None) 22 ctx1dCcotjykfgqrsuplmnbehv

201 if not st: 22 ctx1dCcotjykfgqrsuplmnbehv

202 st = np.array([np.nan, np.nan]) 2 ctx1eh

203 missed_bnc2 += 1 2 ctx1eh

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

205 st = np.delete(st, np.where(np.diff(st) < 0.020)[0] + 1) 22 ctx1dCcotjykfgqrsuplmnbehv

206 rw_times[ind] = tr['behavior_data']['States timestamps']['reward'][0][0] 22 ctx1dCcotjykfgqrsuplmnbehv

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

208 err_sound_times[ind] = st[-1] if st.size >= 2 and np.isnan(rw_times[ind]) else np.nan 22 ctx1dCcotjykfgqrsuplmnbehv

209 if missed_bnc2 == len(data): 22 ctx1dCcotjykfgqrsuplmnbehv

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

211 merge *= np.nan 22 ctx1dCcotjykfgqrsuplmnbehv

212 merge[~np.isnan(rw_times)] = rw_times[~np.isnan(rw_times)] 22 ctx1dCcotjykfgqrsuplmnbehv

213 merge[~np.isnan(err_sound_times)] = err_sound_times[~np.isnan(err_sound_times)] 22 ctx1dCcotjykfgqrsuplmnbehv

214 

215 return merge 22 ctx1dCcotjykfgqrsuplmnbehv

216 

217 def _extract(self): (empty)

218 # Version check 

219 if parse_version(self.settings['IBLRIG_VERSION_TAG']) >= parse_version('5.0.0'): 27 ctx1adCIcotjykfgiqrsuplmnwxbehv

220 merge = self.get_feedback_times_ge5(self.session_path, task_collection=self.task_collection, data=self.bpod_trials) 22 ctx1dCcotjykfgqrsuplmnbehv

221 else: 

222 merge = self.get_feedback_times_lt5(self.session_path, task_collection=self.task_collection, data=self.bpod_trials) 9 ctx1adIciwxbe

223 return np.array(merge) 27 ctx1adCIcotjykfgiqrsuplmnwxbehv

224 

225 

226class Intervals(BaseBpodTrialsExtractor): (empty)

227 """ 

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

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

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

231 

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

233 """ 

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

235 var_names = 'intervals' (empty)

236 

237 def _extract(self): (empty)

238 starts = [t['behavior_data']['Trial start timestamp'] for t in self.bpod_trials] 26 ctx1adPcotjykfgiqrsuplmnwxbehv

239 ends = [t['behavior_data']['Trial end timestamp'] for t in self.bpod_trials] 26 ctx1adPcotjykfgiqrsuplmnwxbehv

240 return np.array([starts, ends]).T 26 ctx1adPcotjykfgiqrsuplmnwxbehv

241 

242 

243class ResponseTimes(BaseBpodTrialsExtractor): (empty)

244 """ 

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

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

247 

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

249 """ 

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

251 var_names = 'response_times' (empty)

252 

253 def _extract(self): (empty)

254 rt = np.array([tr['behavior_data']['States timestamps']['closed_loop'][0][1] 25 ctx1adScotjkfgiqrsuplmnwxbehv

255 for tr in self.bpod_trials]) 

256 return rt 25 ctx1adScotjkfgiqrsuplmnwxbehv

257 

258 

259class ItiDuration(BaseBpodTrialsExtractor): (empty)

260 """ 

261 Calculate duration of iti from state timestamps. 

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

263 

264 Uses Trial end timestamp and get_response_times to calculate iti. 

265 """ 

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

267 var_names = 'iti_dur' (empty)

268 

269 def _extract(self): (empty)

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

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

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

273 iti_dur = ends - rt 

274 return iti_dur 

275 

276 

277class GoCueTriggerTimes(BaseBpodTrialsExtractor): (empty)

278 """ 

279 Get trigger times of goCue from state machine. 

280 

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

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

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

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

285 """ 

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

287 var_names = 'goCueTrigger_times' (empty)

288 

289 def _extract(self): (empty)

290 if parse_version(self.settings['IBLRIG_VERSION_TAG']) >= parse_version('5.0.0'): 26 ctx1adQcotjkfgiqRrsuplmnwxbehv

291 goCue = np.array([tr['behavior_data']['States timestamps'] 22 ctx1dQcotjkfgqRrsuplmnbehv

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

293 else: 

294 goCue = np.array([tr['behavior_data']['States timestamps'] 8 ctx1adciwxbe

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

296 return goCue 26 ctx1adQcotjkfgiqRrsuplmnwxbehv

297 

298 

299class TrialType(BaseBpodTrialsExtractor): (empty)

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

301 var_name = 'trial_type' (empty)

302 

303 def _extract(self): (empty)

304 trial_type = [] 

305 for tr in self.bpod_trials: 

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

307 trial_type.append(1) 

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

309 trial_type.append(-1) 

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

311 trial_type.append(0) 

312 else: 

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

314 trial_type.append(np.nan) 

315 return np.array(trial_type) 

316 

317 

318class GoCueTimes(BaseBpodTrialsExtractor): (empty)

319 """ 

320 Get trigger times of goCue from state machine. 

321 

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

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

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

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

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

327 """ 

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

329 var_names = 'goCue_times' (empty)

330 

331 def _extract(self): (empty)

332 go_cue_times = np.zeros([len(self.bpod_trials), ]) 28 ctx1adEcotGjykfgiqrsuplmnwxHbehv

333 for ind, tr in enumerate(self.bpod_trials): 28 ctx1adEcotGjykfgiqrsuplmnwxHbehv

334 if raw.get_port_events(tr, 'BNC2'): 28 ctx1adEcotGjykfgiqrsuplmnwxHbehv

335 bnchigh = tr['behavior_data']['Events timestamps'].get('BNC2High', None) 25 ctx1dEcotGjykfgiqrsuplmnHbehv

336 if bnchigh: 25 ctx1dEcotGjykfgiqrsuplmnHbehv

337 go_cue_times[ind] = bnchigh[0] 25 ctx1dEcotGjykfgiqrsuplmnHbehv

338 continue 25 ctx1dEcotGjykfgiqrsuplmnHbehv

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

340 if bnclow: 2 ctx1eh

341 go_cue_times[ind] = bnclow[0] - 0.1 2 ctx1eh

342 continue 2 ctx1eh

343 go_cue_times[ind] = np.nan 

344 else: 

345 go_cue_times[ind] = np.nan 8 ctx1adEcwxbe

346 

347 nmissing = np.sum(np.isnan(go_cue_times)) 28 ctx1adEcotGjykfgiqrsuplmnwxHbehv

348 # Check if all stim_syncs have failed to be detected 

349 if np.all(np.isnan(go_cue_times)): 28 ctx1adEcotGjykfgiqrsuplmnwxHbehv

350 _logger.warning( 8 ctx1adEcwxbe

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

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

353 elif np.any(np.isnan(go_cue_times)): 25 ctx1dEcotGjykfgiqrsuplmnHbehv

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

355 

356 return go_cue_times 28 ctx1adEcotGjykfgiqrsuplmnwxHbehv

357 

358 

359class IncludedTrials(BaseBpodTrialsExtractor): (empty)

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

361 var_names = 'included' (empty)

362 

363 def _extract(self): (empty)

364 if parse_version(self.settings['IBLRIG_VERSION_TAG']) >= parse_version('5.0.0'): 21 ctx1dJLMcojkiqrsuplmnbehv

365 trials_included = self.get_included_trials_ge5( 19 ctx1dJLcojkqrsuplmnbehv

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

367 else: 

368 trials_included = self.get_included_trials_lt5(data=self.bpod_trials) 5 ctx1JMcib

369 return trials_included 21 ctx1dJLMcojkiqrsuplmnbehv

370 

371 @staticmethod (empty)

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

373 trials_included = np.array([True for t in data]) 5 ctx1JMcib

374 return trials_included 5 ctx1JMcib

375 

376 @staticmethod (empty)

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

378 trials_included = np.array([True for t in data]) 19 ctx1dJLcojkqrsuplmnbehv

379 if ('SUBJECT_DISENGAGED_TRIGGERED' in settings.keys() and settings[ 19 ctx1dJLcojkqrsuplmnbehv

380 'SUBJECT_DISENGAGED_TRIGGERED'] is not False): 

381 idx = settings['SUBJECT_DISENGAGED_TRIALNUM'] - 1 8 ctx1jkupmneh

382 trials_included[idx:] = False 8 ctx1jkupmneh

383 return trials_included 19 ctx1dJLcojkqrsuplmnbehv

384 

385 

386class ItiInTimes(BaseBpodTrialsExtractor): (empty)

387 var_names = 'itiIn_times' (empty)

388 

389 def _extract(self): (empty)

390 if parse_version(self.settings["IBLRIG_VERSION_TAG"]) < parse_version("5.0.0"): 21 ctx1dcotjkfgiqrsuplmnbehv

391 iti_in = np.ones(len(self.bpod_trials)) * np.nan 2 ctx1ib

392 else: 

393 iti_in = np.array( 20 ctx1dcotjkfgqrsuplmnbehv

394 [tr["behavior_data"]["States timestamps"] 

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

396 ) 

397 return iti_in 21 ctx1dcotjkfgiqrsuplmnbehv

398 

399 

400class ErrorCueTriggerTimes(BaseBpodTrialsExtractor): (empty)

401 var_names = 'errorCueTrigger_times' (empty)

402 

403 def _extract(self): (empty)

404 errorCueTrigger_times = np.zeros(len(self.bpod_trials)) * np.nan 21 ctx1dcotjkfgiqrsuplmnbehv

405 for i, tr in enumerate(self.bpod_trials): 21 ctx1dcotjkfgiqrsuplmnbehv

406 nogo = tr["behavior_data"]["States timestamps"]["no_go"][0][0] 21 ctx1dcotjkfgiqrsuplmnbehv

407 error = tr["behavior_data"]["States timestamps"]["error"][0][0] 21 ctx1dcotjkfgiqrsuplmnbehv

408 if np.all(~np.isnan(nogo)): 21 ctx1dcotjkfgiqrsuplmnbehv

409 errorCueTrigger_times[i] = nogo 13 ctx1dctkfgrsulbeh

410 elif np.all(~np.isnan(error)): 21 ctx1dcotjkfgiqrsuplmnbehv

411 errorCueTrigger_times[i] = error 21 ctx1dcotjkfgiqrsuplmnbehv

412 return errorCueTrigger_times 21 ctx1dcotjkfgiqrsuplmnbehv

413 

414 

415class StimFreezeTriggerTimes(BaseBpodTrialsExtractor): (empty)

416 var_names = 'stimFreezeTrigger_times' (empty)

417 

418 def _extract(self): (empty)

419 if parse_version(self.settings["IBLRIG_VERSION_TAG"]) < parse_version("6.2.5"): 21 ctx1dcotjkfgiqrsuplmnbehv

420 return np.ones(len(self.bpod_trials)) * np.nan 10 ctx1dcirsubehv

421 freeze_reward = np.array( 12 ctx1otjkfgqplmnh

422 [ 

423 True 

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

425 else False 

426 for tr in self.bpod_trials 

427 ] 

428 ) 

429 freeze_error = np.array( 12 ctx1otjkfgqplmnh

430 [ 

431 True 

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

433 else False 

434 for tr in self.bpod_trials 

435 ] 

436 ) 

437 no_go = np.array( 12 ctx1otjkfgqplmnh

438 [ 

439 True 

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

441 else False 

442 for tr in self.bpod_trials 

443 ] 

444 ) 

445 assert (np.sum(freeze_error) + np.sum(freeze_reward) + 12 ctx1otjkfgqplmnh

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

447 stimFreezeTrigger = np.array([]) 12 ctx1otjkfgqplmnh

448 for r, e, n, tr in zip(freeze_reward, freeze_error, no_go, self.bpod_trials): 12 ctx1otjkfgqplmnh

449 if n: 12 ctx1otjkfgqplmnh

450 stimFreezeTrigger = np.append(stimFreezeTrigger, np.nan) 6 ctx1tkfglh

451 continue 6 ctx1tkfglh

452 state = "freeze_reward" if r else "freeze_error" 12 ctx1otjkfgqplmnh

453 stimFreezeTrigger = np.append( 12 ctx1otjkfgqplmnh

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

455 ) 

456 return stimFreezeTrigger 12 ctx1otjkfgqplmnh

457 

458 

459class StimOffTriggerTimes(BaseBpodTrialsExtractor): (empty)

460 var_names = 'stimOffTrigger_times' (empty)

461 

462 def _extract(self): (empty)

463 if parse_version(self.settings["IBLRIG_VERSION_TAG"]) >= parse_version("6.2.5"): 21 ctx1dcotjkfgiqrsuplmnbehv

464 stim_off_trigger_state = "hide_stim" 12 ctx1otjkfgqplmnh

465 elif parse_version(self.settings["IBLRIG_VERSION_TAG"]) >= parse_version("5.0.0"): 10 ctx1dcirsubehv

466 stim_off_trigger_state = "exit_state" 9 ctx1dcrsubehv

467 else: 

468 stim_off_trigger_state = "trial_start" 2 ctx1ib

469 

470 stimOffTrigger_times = np.array( 21 ctx1dcotjkfgiqrsuplmnbehv

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": 21 ctx1dcotjkfgiqrsuplmnbehv

476 return stimOffTrigger_times 2 ctx1ib

477 

478 no_goTrigger_times = np.array( 20 ctx1dcotjkfgqrsuplmnbehv

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[ 20 ctx1dcotjkfgqrsuplmnbehv

490 ~np.isnan(no_goTrigger_times) 

491 ] 

492 return stimOffTrigger_times 20 ctx1dcotjkfgqrsuplmnbehv

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'] 23 ctx1dTcotjykfgiqrsuplmnbehv

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

503 return stim_on_state[:, 0].T 23 ctx1dTcotjykfgiqrsuplmnbehv

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." 10 ctx1adBzciwxbe

520 "From version 5., use StimOnOffFreezeTimes") 

521 if parse_version(self.settings['IBLRIG_VERSION_TAG']) >= parse_version('5.0.0'): 10 ctx1adBzciwxbe

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

523 task_collection=self.task_collection) 

524 else: 

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

526 task_collection=self.task_collection) 

527 return np.array(stimOn_times) 10 ctx1adBzciwxbe

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: 1 ctx1B

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] 1 ctx1B

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

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

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

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

549 

550 stimOn_times = np.array([]) 1 ctx1B

551 for sync, on, off in zip( 1 ctx1B

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

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

554 if pulse.size == 0: 1 ctx1B

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

556 else: 

557 stimOn_times = np.append(stimOn_times, pulse) 1 ctx1B

558 

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

560 # Check if all stim_syncs have failed to be detected 

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

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)): 1 ctx1B

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

567 

568 return stimOn_times 1 ctx1B

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 hte stim 

574 (state stim_on start or rotary_encoder_event2) 

575 Find the next frame change from the photodiodeafter that TS. 

576 Screen is not displaying anything until then. 

577 (Frame changes are in BNC1High and BNC1Low) 

578 """ 

579 if not data: 9 ctx1adzciwxbe

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

581 stim_on = [] 9 ctx1adzciwxbe

582 bnc_h = [] 9 ctx1adzciwxbe

583 bnc_l = [] 9 ctx1adzciwxbe

584 for tr in data: 9 ctx1adzciwxbe

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

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

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

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

589 else: 

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

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

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

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

594 else: 

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

596 

597 stim_on = np.array(stim_on) 9 ctx1adzciwxbe

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

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

600 

601 count_missing = 0 9 ctx1adzciwxbe

602 stimOn_times = np.zeros_like(stim_on) 9 ctx1adzciwxbe

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

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

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

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

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

608 count_missing += 1 6 ctx1azcwxe

609 stimOn_times[i] = stot[0] 9 ctx1adzciwxbe

610 

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

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

613 

614 if count_missing > 0: 9 ctx1adzciwxbe

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

616 

617 return np.array(stimOn_times) 9 ctx1adzciwxbe

618 

619 

620class StimOnOffFreezeTimes(BaseBpodTrialsExtractor): (empty)

621 """ 

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

623 """ 

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

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

626 

627 def _extract(self): (empty)

628 choice = Choice(self.session_path).extract( 25 ctx1adcAotjkfgiqrsuplmnwxbehv

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

630 )[0] 

631 f2TTL = [raw.get_port_events(tr, name='BNC1') for tr in self.bpod_trials] 25 ctx1adcAotjkfgiqrsuplmnwxbehv

632 

633 stimOn_times = np.array([]) 25 ctx1adcAotjkfgiqrsuplmnwxbehv

634 stimOff_times = np.array([]) 25 ctx1adcAotjkfgiqrsuplmnwxbehv

635 stimFreeze_times = np.array([]) 25 ctx1adcAotjkfgiqrsuplmnwxbehv

636 for tr in f2TTL: 25 ctx1adcAotjkfgiqrsuplmnwxbehv

637 if tr and len(tr) == 2: 25 ctx1adcAotjkfgiqrsuplmnwxbehv

638 stimOn_times = np.append(stimOn_times, tr[0]) 14 ctx1Aotjfgirslmnbh

639 stimOff_times = np.append(stimOff_times, tr[-1]) 14 ctx1Aotjfgirslmnbh

640 stimFreeze_times = np.append(stimFreeze_times, np.nan) 14 ctx1Aotjfgirslmnbh

641 elif tr and len(tr) >= 3: 25 ctx1adcAotjkfgiqrsuplmnwxbehv

642 stimOn_times = np.append(stimOn_times, tr[0]) 22 ctx1dcAotjkfgiqrsuplmnbehv

643 stimOff_times = np.append(stimOff_times, tr[-1]) 22 ctx1dcAotjkfgiqrsuplmnbehv

644 stimFreeze_times = np.append(stimFreeze_times, tr[-2]) 22 ctx1dcAotjkfgiqrsuplmnbehv

645 else: 

646 stimOn_times = np.append(stimOn_times, np.nan) 7 ctx1aAjkwxe

647 stimOff_times = np.append(stimOff_times, np.nan) 7 ctx1aAjkwxe

648 stimFreeze_times = np.append(stimFreeze_times, np.nan) 7 ctx1aAjkwxe

649 

650 # In no_go trials no stimFreeze happens just stim Off 

651 stimFreeze_times[choice == 0] = np.nan 25 ctx1adcAotjkfgiqrsuplmnwxbehv

652 # Check for trigger times 

653 # 2nd order criteria: 

654 # stimOn -> Closest one to stimOnTrigger? 

655 # stimOff -> Closest one to stimOffTrigger? 

656 # stimFreeze -> Closest one to stimFreezeTrigger? 

657 

658 return stimOn_times, stimOff_times, stimFreeze_times 25 ctx1adcAotjkfgiqrsuplmnwxbehv

659 

660 

661class PhasePosQuiescence(BaseBpodTrialsExtractor): (empty)

662 """Extracts stimulus phase, position and quiescence from Bpod data. 

663 For extraction of pre-generated events, use the ProbaContrasts extractor instead. 

664 """ 

665 save_names = (None, None, '_ibl_trials.quiescencePeriod.npy') (empty)

666 var_names = ('phase', 'position', 'quiescence') (empty)

667 

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

669 phase = np.array([t['stim_phase'] for t in self.bpod_trials]) 23 ctx1adcojkfgiqrsuplmnwxbehv

670 position = np.array([t['position'] for t in self.bpod_trials]) 23 ctx1adcojkfgiqrsuplmnwxbehv