Coverage for ibllib/io/extractors/biased_trials.py: 99%

1from pathlib import Path, PureWindowsPath (empty)

2 

3from packaging import version (empty)

4import numpy as np (empty)

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

6 

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

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

9from ibllib.io.extractors.training_trials import ( (empty)

10 Choice, FeedbackTimes, FeedbackType, GoCueTimes, GoCueTriggerTimes, 

11 IncludedTrials, Intervals, ProbabilityLeft, ResponseTimes, RewardVolume, 

12 StimOnTriggerTimes, StimOnOffFreezeTimes, ItiInTimes, 

13 StimOffTriggerTimes, StimFreezeTriggerTimes, ErrorCueTriggerTimes, PhasePosQuiescence) 

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

15 

16__all__ = ['BiasedTrials', 'EphysTrials'] (empty)

17 

18 

19class ContrastLR(BaseBpodTrialsExtractor): (empty)

20 """Get left and right contrasts from raw datafile.""" 

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

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

23 

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

25 contrastLeft = np.array([t['contrast'] if np.sign( 9 ctx1plmqnocdb

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

27 contrastRight = np.array([t['contrast'] if np.sign( 9 ctx1plmqnocdb

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

29 return contrastLeft, contrastRight 9 ctx1plmqnocdb

30 

31 

32class ProbaContrasts(BaseBpodTrialsExtractor): (empty)

33 """Bpod pre-generated values for probabilityLeft, contrastLR, phase, quiescence.""" 

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

35 '_ibl_trials.probabilityLeft.npy', '_ibl_trials.quiescencePeriod.npy') 

36 var_names = ('contrastLeft', 'contrastRight', 'phase', (empty)

37 'position', 'probabilityLeft', 'quiescence') 

38 

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

40 """Extracts positions, contrasts, quiescent delay, stimulus phase and probability left 

41 from pregenerated session files. Used in ephysChoiceWorld extractions. 

42 Optional: saves alf contrastLR and probabilityLeft npy files""" 

43 pe = self.get_pregenerated_events(self.bpod_trials, self.settings) 11 ctx1akicjfgdbeh

44 return [pe[k] for k in sorted(pe.keys())] 11 ctx1akicjfgdbeh

45 

46 @staticmethod (empty)

47 def get_pregenerated_events(bpod_trials, settings): (empty)

48 for k in ['PRELOADED_SESSION_NUM', 'PREGENERATED_SESSION_NUM', 'SESSION_TEMPLATE_ID']: 11 ctx1akicjfgdbeh

49 num = settings.get(k, None) 11 ctx1akicjfgdbeh

50 if num is not None: 11 ctx1akicjfgdbeh

51 break 5 ctx1kijde

52 if num is None: 11 ctx1akicjfgdbeh

53 fn = settings.get('SESSION_LOADED_FILE_PATH', '') 7 ctx1acfgbeh

54 fn = PureWindowsPath(fn).name 7 ctx1acfgbeh

55 num = ''.join([d for d in fn if d.isdigit()]) 7 ctx1acfgbeh

56 if num == '': 7 ctx1acfgbeh

57 raise ValueError("Can't extract left probability behaviour.") 

58 # Load the pregenerated file 

59 ntrials = len(bpod_trials) 11 ctx1akicjfgdbeh

60 sessions_folder = Path(raw.__file__).parent.joinpath( 11 ctx1akicjfgdbeh

61 "extractors", "ephys_sessions") 

62 fname = f"session_{num}_ephys_pcqs.npy" 11 ctx1akicjfgdbeh

63 pcqsp = np.load(sessions_folder.joinpath(fname)) 11 ctx1akicjfgdbeh

64 pos = pcqsp[:, 0] 11 ctx1akicjfgdbeh

65 con = pcqsp[:, 1] 11 ctx1akicjfgdbeh

66 pos = pos[: ntrials] 11 ctx1akicjfgdbeh

67 con = con[: ntrials] 11 ctx1akicjfgdbeh

68 contrastRight = con.copy() 11 ctx1akicjfgdbeh

69 contrastLeft = con.copy() 11 ctx1akicjfgdbeh

70 contrastRight[pos < 0] = np.nan 11 ctx1akicjfgdbeh

71 contrastLeft[pos > 0] = np.nan 11 ctx1akicjfgdbeh

72 qui = pcqsp[:, 2] 11 ctx1akicjfgdbeh

73 qui = qui[: ntrials] 11 ctx1akicjfgdbeh

74 phase = pcqsp[:, 3] 11 ctx1akicjfgdbeh

75 phase = phase[: ntrials] 11 ctx1akicjfgdbeh

76 pLeft = pcqsp[:, 4] 11 ctx1akicjfgdbeh

77 pLeft = pLeft[: ntrials] 11 ctx1akicjfgdbeh

78 

79 phase_path = sessions_folder.joinpath(f"session_{num}_stim_phase.npy") 11 ctx1akicjfgdbeh

80 is_patched_version = version.parse( 11 ctx1akicjfgdbeh

81 settings.get('IBLRIG_VERSION') or '0') > version.parse('6.4.0') 

82 if phase_path.exists() and is_patched_version: 11 ctx1akicjfgdbeh

83 phase = np.load(phase_path)[:ntrials] 4 ctx1kijd

84 

85 return {'position': pos, 'quiescence': qui, 'phase': phase, 'probabilityLeft': pLeft, 11 ctx1akicjfgdbeh

86 'contrastRight': contrastRight, 'contrastLeft': contrastLeft} 

87 

88 

89class TrialsTableBiased(BaseBpodTrialsExtractor): (empty)

90 """ 

91 Extracts the following into a table from Bpod raw data: 

92 intervals, goCue_times, response_times, choice, stimOn_times, contrastLeft, contrastRight, 

93 feedback_times, feedbackType, rewardVolume, probabilityLeft, firstMovement_times 

94 Additionally extracts the following wheel data: 

95 wheel_timestamps, wheel_position, wheelMoves_intervals, wheelMoves_peakAmplitude 

96 """ 

97 save_names = ('_ibl_trials.table.pqt', None, None, '_ibl_wheel.timestamps.npy', '_ibl_wheel.position.npy', (empty)

98 '_ibl_wheelMoves.intervals.npy', '_ibl_wheelMoves.peakAmplitude.npy', None, None) 

99 var_names = ('table', 'stimOff_times', 'stimFreeze_times', 'wheel_timestamps', 'wheel_position', 'wheelMoves_intervals', (empty)

100 'wheelMoves_peakAmplitude', 'wheelMoves_peakVelocity_times', 'is_final_movement') 

101 

102 def _extract(self, extractor_classes=None, **kwargs): (empty)

103 extractor_classes = extractor_classes or [] 6 ctx1lmnodb

104 base = [Intervals, GoCueTimes, ResponseTimes, Choice, StimOnOffFreezeTimes, ContrastLR, FeedbackTimes, FeedbackType, 6 ctx1lmnodb

105 RewardVolume, ProbabilityLeft, Wheel] 

106 out, _ = run_extractor_classes( 6 ctx1lmnodb

107 base + extractor_classes, session_path=self.session_path, bpod_trials=self.bpod_trials, 

108 settings=self.settings, save=False, task_collection=self.task_collection) 

109 

110 table = AlfBunch({k: out.pop(k) for k in list(out.keys()) if k not in self.var_names}) 6 ctx1lmnodb

111 assert len(table.keys()) == 12 6 ctx1lmnodb

112 

113 return table.to_df(), *(out.pop(x) for x in self.var_names if x != 'table') 6 ctx1lmnodb

114 

115 

116class TrialsTableEphys(BaseBpodTrialsExtractor): (empty)

117 """ 

118 Extracts the following into a table from Bpod raw data: 

119 intervals, goCue_times, response_times, choice, stimOn_times, contrastLeft, contrastRight, 

120 feedback_times, feedbackType, rewardVolume, probabilityLeft, firstMovement_times 

121 Additionally extracts the following wheel data: 

122 wheel_timestamps, wheel_position, wheelMoves_intervals, wheelMoves_peakAmplitude 

123 """ 

124 save_names = ('_ibl_trials.table.pqt', None, None, '_ibl_wheel.timestamps.npy', '_ibl_wheel.position.npy', (empty)

125 '_ibl_wheelMoves.intervals.npy', '_ibl_wheelMoves.peakAmplitude.npy', None, 

126 None, None, None, '_ibl_trials.quiescencePeriod.npy') 

127 var_names = ('table', 'stimOff_times', 'stimFreeze_times', 'wheel_timestamps', 'wheel_position', 'wheelMoves_intervals', (empty)

128 'wheelMoves_peakAmplitude', 'wheelMoves_peakVelocity_times', 'is_final_movement', 

129 'phase', 'position', 'quiescence') 

130 

131 def _extract(self, extractor_classes=None, **kwargs): (empty)

132 extractor_classes = extractor_classes or [] 9 ctx1aicjfgbeh

133 base = [Intervals, GoCueTimes, ResponseTimes, Choice, StimOnOffFreezeTimes, ProbaContrasts, 9 ctx1aicjfgbeh

134 FeedbackTimes, FeedbackType, RewardVolume, Wheel] 

135 # Exclude from trials table 

136 out, _ = run_extractor_classes( 9 ctx1aicjfgbeh

137 base + extractor_classes, session_path=self.session_path, bpod_trials=self.bpod_trials, 

138 settings=self.settings, save=False, task_collection=self.task_collection) 

139 table = AlfBunch({k: v for k, v in out.items() if k not in self.var_names}) 9 ctx1aicjfgbeh

140 assert len(table.keys()) == 12 9 ctx1aicjfgbeh

141 

142 return table.to_df(), *(out.pop(x) for x in self.var_names if x != 'table') 9 ctx1aicjfgbeh

143 

144 

145class BiasedTrials(BaseBpodTrialsExtractor): (empty)

146 """ 

147 Same as training_trials.TrainingTrials except... 

148 - there is no RepNum 

149 - ContrastLR is extracted differently 

150 - IncludedTrials is only extracted for 5.0.0 or greater 

151 """ 

152 save_names = ('_ibl_trials.goCueTrigger_times.npy', '_ibl_trials.stimOnTrigger_times.npy', None, (empty)

153 '_ibl_trials.stimOffTrigger_times.npy', None, None, '_ibl_trials.table.pqt', 

154 '_ibl_trials.stimOff_times.npy', None, '_ibl_wheel.timestamps.npy', '_ibl_wheel.position.npy', 

155 '_ibl_wheelMoves.intervals.npy', '_ibl_wheelMoves.peakAmplitude.npy', None, None, 

156 '_ibl_trials.included.npy', None, None, '_ibl_trials.quiescencePeriod.npy') 

157 var_names = ('goCueTrigger_times', 'stimOnTrigger_times', 'itiIn_times', 'stimOffTrigger_times', 'stimFreezeTrigger_times', (empty)

158 'errorCueTrigger_times', 'table', 'stimOff_times', 'stimFreeze_times', 'wheel_timestamps', 'wheel_position', 

159 'wheelMoves_intervals', 'wheelMoves_peakAmplitude', 'wheelMoves_peakVelocity_times', 'is_final_movement', 

160 'included', 'phase', 'position', 'quiescence') 

161 

162 def _extract(self, extractor_classes=None, **kwargs) -> dict: (empty)

163 extractor_classes = extractor_classes or [] 5 ctx1lmnob

164 base = [GoCueTriggerTimes, StimOnTriggerTimes, ItiInTimes, StimOffTriggerTimes, StimFreezeTriggerTimes, 5 ctx1lmnob

165 ErrorCueTriggerTimes, TrialsTableBiased, IncludedTrials, PhasePosQuiescence] 

166 # Exclude from trials table 

167 out, _ = run_extractor_classes( 5 ctx1lmnob

168 base + extractor_classes, session_path=self.session_path, bpod_trials=self.bpod_trials, 

169 settings=self.settings, save=False, task_collection=self.task_collection) 

170 return {k: out[k] for k in self.var_names} 5 ctx1lmnob

171 

172 

173class EphysTrials(BaseBpodTrialsExtractor): (empty)

174 """ 

175 Same as BiasedTrials except... 

176 - Contrast, phase, position, probabilityLeft and quiescence is extracted differently 

177 """ 

178 save_names = ('_ibl_trials.goCueTrigger_times.npy', '_ibl_trials.stimOnTrigger_times.npy', None, (empty)

179 '_ibl_trials.stimOffTrigger_times.npy', None, None, 

180 '_ibl_trials.table.pqt', '_ibl_trials.stimOff_times.npy', None, '_ibl_wheel.timestamps.npy', 

181 '_ibl_wheel.position.npy', '_ibl_wheelMoves.intervals.npy', '_ibl_wheelMoves.peakAmplitude.npy', None, None, 

182 '_ibl_trials.included.npy', None, None, '_ibl_trials.quiescencePeriod.npy') 

183 var_names = ('goCueTrigger_times', 'stimOnTrigger_times', 'itiIn_times', 'stimOffTrigger_times', 'stimFreezeTrigger_times', (empty)

184 'errorCueTrigger_times', 'table', 'stimOff_times', 'stimFreeze_times', 'wheel_timestamps', 'wheel_position', 

185 'wheelMoves_intervals', 'wheelMoves_peakAmplitude', 'wheelMoves_peakVelocity_times', 'is_final_movement', 

186 'included', 'phase', 'position', 'quiescence') 

187 

188 def _extract(self, extractor_classes=None, **kwargs) -> dict: (empty)

189 extractor_classes = extractor_classes or [] 10 ctx1aicjfgdbeh

190 

191 # For iblrig v8 we use the biased trials table instead. ContrastLeft, ContrastRight and ProbabilityLeft are 

192 # filled from the values in the bpod data itself rather than using the pregenerated session number 

193 iblrig_version = self.settings.get('IBLRIG_VERSION', self.settings.get('IBLRIG_VERSION_TAG', '0')) 10 ctx1aicjfgdbeh

194 if version.parse(iblrig_version) >= version.parse('8.0.0'): 10 ctx1aicjfgdbeh

195 TrialsTable = TrialsTableBiased 1 ctx1d

196 else: 

197 TrialsTable = TrialsTableEphys 9 ctx1aicjfgbeh

198 

199 base = [GoCueTriggerTimes, StimOnTriggerTimes, ItiInTimes, StimOffTriggerTimes, StimFreezeTriggerTimes, 10 ctx1aicjfgdbeh

200 ErrorCueTriggerTimes, TrialsTable, IncludedTrials, PhasePosQuiescence] 

201 # Get all detected TTLs. These are stored for QC purposes 

202 self.frame2ttl, self.audio = raw.load_bpod_fronts(self.session_path, data=self.bpod_trials) 10 ctx1aicjfgdbeh

203 # Exclude from trials table 

204 out, _ = run_extractor_classes( 10 ctx1aicjfgdbeh

205 base + extractor_classes, session_path=self.session_path, bpod_trials=self.bpod_trials, 

206 settings=self.settings, save=False, task_collection=self.task_collection) 

207 return {k: out[k] for k in self.var_names} 10 ctx1aicjfgdbeh