Coverage for ibllib/tests/fixtures/utils.py: 98%

1#!/usr/bin/env python 

2# -*- coding:utf-8 -*- 

3# @Author: Niccolò Bonacchi 

4# @Date: Friday, October 9th 2020, 12:02:56 pm 

5import json (empty)

6from pathlib import Path (empty)

7 

8from ibllib.io import session_params (empty)

9 

10 

11def create_fake_session_folder( (empty)

12 root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True 

13): 

14 root_data_path = Path(root_data_path) 8 ctx1adbefcgh

15 session_path = root_data_path / lab / "Subjects" / mouse / date / num 8 ctx1adbefcgh

16 if session_path.exists() and increment: 8 ctx1adbefcgh

17 num = str(int(num) + 1).zfill(3) 3 ctx1abc

18 return create_fake_session_folder( 3 ctx1abc

19 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment 

20 ) 

21 session_path = root_data_path / lab / "Subjects" / mouse / date / num 8 ctx1adbefcgh

22 

23 session_path.mkdir(exist_ok=True, parents=True) 8 ctx1adbefcgh

24 return session_path 8 ctx1adbefcgh

25 

26 

27def create_fake_raw_ephys_data_folder(session_path, populate=True): (empty)

28 """create_fake_raw_ephys_data_folder creates raw_ephys_data folder 

29 can populate with empty files with expected names 

30 

31 :param session_path: [description] 

32 :type session_path: [type] 

33 :param populate: [description], defaults to True 

34 :type populate: bool, optional 

35 :return: [description] 

36 :rtype: [type] 

37 """ 

38 session_path = Path(session_path) 2 ctx1ai

39 raw_ephys_data_path = session_path / "raw_ephys_data" 2 ctx1ai

40 raw_ephys_data_path.mkdir(exist_ok=True, parents=True) 2 ctx1ai

41 if populate: 2 ctx1ai

42 file_list = [ 2 ctx1ai

43 "probe00/_spikeglx_ephysData_g0_t0.imec0.sync.npy", 

44 "probe00/_spikeglx_ephysData_g0_t0.imec0.lf.meta", 

45 "probe00/_spikeglx_ephysData_g0_t0.imec0.ap.cbin", 

46 "probe00/_spikeglx_ephysData_g0_t0.imec0.timestamps.npy", 

47 "probe00/_spikeglx_ephysData_g0_t0.imec0.lf.cbin", 

48 "probe00/_spikeglx_ephysData_g0_t0.imec0.ap.ch", 

49 "probe00/_spikeglx_ephysData_g0_t0.imec0.wiring.json", 

50 "probe00/_spikeglx_sync.times.probe00.npy", 

51 "probe00/_spikeglx_sync.channels.probe00.npy", 

52 "probe00/_spikeglx_sync.polarities.probe00.npy", 

53 "probe01/_spikeglx_ephysData_g0_t0.imec1.sync.npy", 

54 "probe01/_spikeglx_ephysData_g0_t0.imec1.lf.meta", 

55 "probe01/_spikeglx_ephysData_g0_t0.imec1.timestamps.cbin", 

56 "probe01/_spikeglx_ephysData_g0_t0.imec1.ap.cbin", 

57 "probe01/_spikeglx_ephysData_g0_t0.imec1.lf.cbin", 

58 "probe01/_spikeglx_ephysData_g0_t0.imec1.ap.ch", 

59 "probe01/_spikeglx_ephysData_g0_t0.imec1.wiring.json", 

60 "probe02/_spikeglx_ephysData_g0_t0.imec.ap.meta", # 3A 

61 "probe02/_spikeglx_ephysData_g0_t0.imec.lf.meta", 

62 "probe02/_spikeglx_ephysData_g0_t0.imec.ap.bin", 

63 "probe02/_spikeglx_ephysData_g0_t0.imec.lf.bin", 

64 "_spikeglx_ephysData_g0_t0.nidq.sync.npy", 

65 "_spikeglx_ephysData_g0_t0.nidq.meta", 

66 "_spikeglx_ephysData_g0_t0.nidq.cbin", 

67 "_spikeglx_ephysData_g0_t0.nidq.ch", 

68 "_spikeglx_ephysData_g0_t0.wiring.json", 

69 ] 

70 for f in file_list: 2 ctx1ai

71 fpath = raw_ephys_data_path / Path(f) 2 ctx1ai

72 fpath.parent.mkdir(parents=True, exist_ok=True) 2 ctx1ai

73 fpath.touch() 2 ctx1ai

74 

75 return session_path 2 ctx1ai

76 

77 

78def populate_raw_spikeglx(session_path, (empty)

79 model='3B', legacy=False, user_label='my_run', n_probes=2): 

80 """ 

81 Touch file tree to emulate files saved by SpikeGLX 

82 :param session_path: The raw ephys data path to place files 

83 :param model: Probe model file structure ('3A' or '3B') 

84 :param legacy: If true, the emulate older SpikeGLX version where all files are saved 

85 into a single folder 

86 :param user_label: User may input any name into SpikeGLX and filenames will include this 

87 :param n_probes: Number of probe datafiles to touch 

88 :return: 

89 

90 Examples: 

91 populate_raw_spikeglx('3A_folder', model='3A', legacy=True, n_probes=1) 

92 3A_folder 

93 └───raw_ephys_folder 

94 my_run_probe00_g0_t0.imec.ap.bin 

95 my_run_probe00_g0_t0.imec.ap.meta 

96 my_run_probe00_g0_t0.imec.lf.bin 

97 my_run_probe00_g0_t0.imec.lf.meta 

98 

99 populate_raw_spikeglx('3B_folder', model='3B', n_probes=3) 

100 3B_folder 

101 └───my_run_g0_t0 

102 my_run_g0_t0.imec0.ap.bin 

103 my_run_g0_t0.imec0.ap.meta 

104 my_run_g0_t0.imec0.lf.bin 

105 my_run_g0_t0.imec0.lf.meta 

106 my_run_g0_t0.imec1.ap.bin 

107 my_run_g0_t0.imec1.ap.meta 

108 my_run_g0_t0.imec1.lf.bin 

109 my_run_g0_t0.imec1.lf.meta 

110 my_run_g0_t0.imec2.ap.bin 

111 my_run_g0_t0.imec2.ap.meta 

112 my_run_g0_t0.imec2.lf.bin 

113 my_run_g0_t0.imec2.lf.meta 

114 my_run_g0_t0.nidq.bin 

115 my_run_g0_t0.nidq.meta 

116 

117 See also: http://billkarsh.github.io/SpikeGLX/help/parsing/ 

118 """ 

119 for i in range(n_probes): (empty)

120 label = (user_label + f'_probe{i:02}') if legacy and model == '3A' else user_label (empty)

121 root = session_path.joinpath('raw_ephys_folder' if legacy else f'{user_label}_g0_t0') (empty)

122 root.mkdir(exist_ok=True, parents=True) (empty)

123 for ext in ('meta', 'bin'): (empty)

124 for freq in ('lf', 'ap'): (empty)

125 filename = f'{label}_g0_t0.imec{i if model == "3B" else ""}.{freq}.{ext}' (empty)

126 root.joinpath(filename).touch() (empty)

127 if model == '3B': (empty)

128 root.joinpath(f'{label}_g0_t0.nidq.{ext}').touch() (empty)

129 

130 

131def create_fake_raw_video_data_folder(session_path, populate=True, write_pars_stub=False): (empty)

132 """ 

133 Create the folder structure for a raw video session with three cameras. 

134 Creates a raw_video_data folder and optionally, touches some files and writes a experiment 

135 description stub to a _devices folder. 

136 

137 Parameters 

138 ---------- 

139 session_path : str, pathlib.Path 

140 The session path in which to create the folders. 

141 populate : bool 

142 If true, touch some raw video files. 

143 write_pars_stub : bool, str, dict 

144 If true, write an experiment description stub containing behaviour settings. If a string, 

145 the stub filename will contain this. If a dict, the key is used as the filename; the value, 

146 the file contents. 

147 

148 Example 

149 ------- 

150 >>> create_fake_raw_video_data_folder(session_path, populate=False, write_pars_stub=False) 

151 >>> create_fake_raw_video_data_folder(session_path, write_pars_stub='hostname_19826354') 

152 """ 

153 session_path = Path(session_path) 3 ctx1abc

154 raw_video_data_path = session_path / "raw_video_data" 3 ctx1abc

155 raw_video_data_path.mkdir(exist_ok=True, parents=True) 3 ctx1abc

156 if populate: 3 ctx1abc

157 file_list = [ 3 ctx1abc

158 "_iblrig_leftCamera.raw.mp4", 

159 "_iblrig_rightCamera.raw.mp4", 

160 "_iblrig_bodyCamera.raw.mp4", 

161 "_iblrig_leftCamera.timestamps.ssv", 

162 "_iblrig_rightCamera.timestamps.ssv", 

163 "_iblrig_bodyCamera.timestamps.ssv", 

164 "_iblrig_leftCamera.GPIO.bin", 

165 "_iblrig_rightCamera.GPIO.bin", 

166 "_iblrig_bodyCamera.GPIO.bin", 

167 "_iblrig_leftCamera.frame_counter.bin", 

168 "_iblrig_rightCamera.frame_counter.bin", 

169 "_iblrig_bodyCamera.frame_counter.bin", 

170 "_iblrig_VideoCodeFiles.raw.zip", 

171 ] 

172 for f in file_list: 3 ctx1abc

173 fpath = raw_video_data_path / Path(f) 3 ctx1abc

174 fpath.parent.mkdir(parents=True, exist_ok=True) 3 ctx1abc

175 fpath.touch() 3 ctx1abc

176 

177 if write_pars_stub: 3 ctx1abc

178 if isinstance(write_pars_stub, dict): (empty)

179 (name, data), = write_pars_stub.items() 

180 else: 

181 name = write_pars_stub if isinstance(write_pars_stub, str) else 'video' (empty)

182 d = {'collection': 'raw_video_data', 'sync_label': 'frame2ttl'} (empty)

183 data = { (empty)

184 'devices': {'cameras': {k: d.copy() for k in ('body', 'left', 'right')}}, 

185 'version': session_params.SPEC_VERSION 

186 } 

187 file_device = session_path.joinpath(f'_ibl_experiment.description_{name}.yaml') (empty)

188 file_device.parent.mkdir(exist_ok=True) (empty)

189 session_params.write_yaml(file_device, data) (empty)

190 return raw_video_data_path 3 ctx1abc

191 

192 

193def create_fake_alf_folder_dlc_data(session_path, populate=True): (empty)

194 session_path = Path(session_path) (empty)

195 alf_path = session_path / "alf" (empty)

196 alf_path.mkdir(exist_ok=True, parents=True) (empty)

197 if populate: (empty)

198 file_list = [ (empty)

199 "_ibl_leftCamera.dlc.pqt", 

200 "_ibl_rightCamera.dlc.pqt", 

201 "_ibl_bodyCamera.dlc.pqt", 

202 "_ibl_leftCamera.times.npy", 

203 "_ibl_rightCamera.times.npy", 

204 "_ibl_bodyCamera.times.npy", 

205 "_ibl_leftCamera.features.npy", 

206 "_ibl_rightCamera.features.npy", 

207 ] 

208 for f in file_list: (empty)

209 fpath = alf_path / Path(f) (empty)

210 fpath.parent.mkdir(parents=True, exist_ok=True) (empty)

211 fpath.touch() (empty)

212 

213 

214def create_fake_raw_behavior_data_folder( (empty)

215 session_path, populate=True, task="ephysCW", folder="raw_behavior_data", write_pars_stub=False 

216): 

217 """Create the folder structure for a raw behaviour session. 

218 

219 Creates a raw_behavior_data folder and optionally, touches some files and writes a experiment 

220 description stub to a `_devices` folder. 

221 

222 Parameters 

223 ---------- 

224 session_path : pathlib.Path 

225 The session path in which to create the folders. 

226 populate : bool 

227 If true, touch some raw behaviour files. 

228 task : str 

229 The name of the task protocol, if 'ephys' or 'passive' extra files are touched. 

230 write_pars_stub : bool, str, dict 

231 If true, write an experiment description stub containing behaviour settings. If a string, 

232 the stub will be named as such. If a dict, the key is used as the filename; the value, 

233 the file contents. 

234 

235 Returns 

236 ------- 

237 pathlib.Path 

238 The raw behaviour data path. 

239 """ 

240 raw_behavior_data_path = session_path / folder 8 ctx1adbefcgh

241 raw_behavior_data_path.mkdir(exist_ok=True, parents=True) 8 ctx1adbefcgh

242 ephysCW = [ 8 ctx1adbefcgh

243 "_iblrig_ambientSensorData.raw.jsonable", 

244 "_iblrig_encoderEvents.raw.ssv", 

245 "_iblrig_encoderPositions.raw.ssv", 

246 "_iblrig_encoderTrialInfo.raw.ssv", 

247 "_iblrig_micData.raw.wav", 

248 "_iblrig_stimPositionScreen.raw.csv", 

249 "_iblrig_syncSquareUpdate.raw.csv", 

250 "_iblrig_taskCodeFiles.raw.zip", 

251 "_iblrig_taskData.raw.jsonable", 

252 "_iblrig_taskSettings.raw.json", 

253 "online_plot.png", 

254 ] 

255 passiveCW = [ 8 ctx1adbefcgh

256 "_iblrig_encoderEvents.raw.ssv", 

257 "_iblrig_encoderPositions.raw.ssv", 

258 "_iblrig_encoderTrialInfo.raw.ssv", 

259 "_iblrig_RFMapStim.raw.bin", 

260 "_iblrig_stimPositionScreen.raw.csv", 

261 "_iblrig_syncSquareUpdate.raw.csv", 

262 "_iblrig_taskCodeFiles.raw.zip", 

263 "_iblrig_taskSettings.raw.json", 

264 ] 

265 

266 if populate: 8 ctx1adbefcgh

267 file_list = [] 8 ctx1adbefcgh

268 if "ephys" in task: 8 ctx1adbefcgh

269 file_list = ephysCW 8 ctx1adbefcgh

270 elif "passive" in task: (empty)

271 file_list = passiveCW (empty)

272 (session_path / "passive_data_for_ephys.flag").touch() (empty)

273 else: 

274 print(f"Not implemented: Task {task}") 

275 

276 for f in file_list: 8 ctx1adbefcgh

277 fpath = raw_behavior_data_path / Path(f) 8 ctx1adbefcgh

278 fpath.parent.mkdir(parents=True, exist_ok=True) 8 ctx1adbefcgh

279 fpath.touch() 8 ctx1adbefcgh

280 

281 if write_pars_stub: 8 ctx1adbefcgh

282 if isinstance(write_pars_stub, dict): (empty)

283 (name, data), = write_pars_stub.items() 

284 else: 

285 name = write_pars_stub if isinstance(write_pars_stub, str) else 'behaviour' (empty)

286 data = { (empty)

287 'devices': {'microphone': {'microphone': {'collection': folder, 'sync_label': None}}}, 

288 'procedures': ['Behavior training/tasks'], 

289 'projects': ['ibl_neuropixel_brainwide_01'], 

290 'tasks': [{task: {'collection': folder}}], 

291 'version': session_params.SPEC_VERSION 

292 } 

293 if 'ephys' in task: (empty)

294 data['tasks'][0][task]['sync_label'] = 'frame2ttl' (empty)

295 else: 

296 data['sync'] = {'bpod': {'collection': 'raw_behavior_data', 'extension': 'jsonable'}} (empty)

297 data['tasks'][0][task]['sync_label'] = 'bpod' (empty)

298 

299 file_device = session_path.joinpath(f'_ibl_experiment.description_{name}.yaml') (empty)

300 file_device.parent.mkdir(exist_ok=True) (empty)

301 session_params.write_yaml(file_device, data) (empty)

302 

303 return raw_behavior_data_path 8 ctx1adbefcgh

304 

305 

306def populate_task_settings(fpath: Path, patch: dict): (empty)

307 with fpath.open("w") as f: 2 ctx1aj

308 json.dump(patch, f, indent=1) 2 ctx1aj

309 

310 

311def create_fake_complete_ephys_session( (empty)

312 root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True 

313): 

314 session_path = create_fake_session_folder( (empty)

315 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment 

316 ) 

317 _mouse, _date, _num = session_path.parts[-3:] (empty)

318 create_fake_raw_ephys_data_folder(session_path, populate=True) (empty)

319 create_fake_raw_video_data_folder(session_path, populate=True) (empty)

320 create_fake_raw_behavior_data_folder(session_path, populate=True, task="ephys") (empty)

321 create_fake_raw_behavior_data_folder( (empty)

322 session_path, populate=True, task="passive", folder="raw_passive_data" 

323 ) 

324 fpath = Path(session_path) / "raw_passive_data" / "_iblrig_taskSettings.raw.json" (empty)

325 passive_settings = { (empty)

326 "CORRESPONDING_EPHYS_SESSION": 

327 f"C:\\some\\root\\folder\\Subjects\\{_mouse}\\{_date}\\{_num}" 

328 } 

329 populate_task_settings(fpath, passive_settings) (empty)

330 if session_path.joinpath("passive_data_for_ephys.flag").exists(): (empty)

331 session_path.joinpath("passive_data_for_ephys.flag").unlink() (empty)

332 

333 return session_path (empty)

334 

335 

336def create_fake_ephys_recording_bad_passive_transfer_sessions( (empty)

337 root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True 

338): 

339 session_path = create_fake_session_folder( (empty)

340 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment 

341 ) 

342 _mouse, _date, _num = session_path.parts[-3:] (empty)

343 create_fake_raw_ephys_data_folder(session_path, populate=True) (empty)

344 create_fake_raw_video_data_folder(session_path, populate=True) (empty)

345 create_fake_raw_behavior_data_folder(session_path, populate=True, task="ephys") (empty)

346 

347 passive_session_path = create_fake_session_folder( (empty)

348 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment 

349 ) 

350 create_fake_raw_behavior_data_folder(passive_session_path, populate=True, task="passive") (empty)

351 fpath = Path(passive_session_path) / "raw_behavior_data" / "_iblrig_taskSettings.raw.json" (empty)

352 passive_settings = { (empty)

353 "CORRESPONDING_EPHYS_SESSION": 

354 f"C:\\some\\root\\folder\\Subjects\\{_mouse}\\{_date}\\{_num}" 

355 } 

356 populate_task_settings(fpath, passive_settings) (empty)

357 

358 return session_path, passive_session_path (empty)