Coverage for ibllib/tests/fixtures/utils.py: 88%
146 statements
« prev ^ index » next coverage.py v7.7.0, created at 2025-03-17 15:25 +0000
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
6import random
7import string
8import logging
9from pathlib import Path
11from one.registration import RegistrationClient
13from ibllib.io import session_params
15_logger = logging.getLogger(__name__)
18def create_fake_session_folder(
19 root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True
20):
21 root_data_path = Path(root_data_path)
22 session_path = root_data_path / lab / "Subjects" / mouse / date / num
23 if session_path.exists() and increment:
24 num = str(int(num) + 1).zfill(3)
25 return create_fake_session_folder(
26 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment
27 )
28 session_path = root_data_path / lab / "Subjects" / mouse / date / num
30 session_path.mkdir(exist_ok=True, parents=True)
31 return session_path
34def create_fake_raw_ephys_data_folder(session_path, populate=True):
35 """create_fake_raw_ephys_data_folder creates raw_ephys_data folder
36 can populate with empty files with expected names
38 :param session_path: [description]
39 :type session_path: [type]
40 :param populate: [description], defaults to True
41 :type populate: bool, optional
42 :return: [description]
43 :rtype: [type]
44 """
45 session_path = Path(session_path) 1ab
46 raw_ephys_data_path = session_path / "raw_ephys_data" 1ab
47 raw_ephys_data_path.mkdir(exist_ok=True, parents=True) 1ab
48 if populate: 1ab
49 file_list = [ 1ab
50 "probe00/_spikeglx_ephysData_g0_t0.imec0.sync.npy",
51 "probe00/_spikeglx_ephysData_g0_t0.imec0.lf.meta",
52 "probe00/_spikeglx_ephysData_g0_t0.imec0.ap.cbin",
53 "probe00/_spikeglx_ephysData_g0_t0.imec0.timestamps.npy",
54 "probe00/_spikeglx_ephysData_g0_t0.imec0.lf.cbin",
55 "probe00/_spikeglx_ephysData_g0_t0.imec0.ap.ch",
56 "probe00/_spikeglx_ephysData_g0_t0.imec0.wiring.json",
57 "probe00/_spikeglx_sync.times.probe00.npy",
58 "probe00/_spikeglx_sync.channels.probe00.npy",
59 "probe00/_spikeglx_sync.polarities.probe00.npy",
60 "probe01/_spikeglx_ephysData_g0_t0.imec1.sync.npy",
61 "probe01/_spikeglx_ephysData_g0_t0.imec1.lf.meta",
62 "probe01/_spikeglx_ephysData_g0_t0.imec1.timestamps.cbin",
63 "probe01/_spikeglx_ephysData_g0_t0.imec1.ap.cbin",
64 "probe01/_spikeglx_ephysData_g0_t0.imec1.lf.cbin",
65 "probe01/_spikeglx_ephysData_g0_t0.imec1.ap.ch",
66 "probe01/_spikeglx_ephysData_g0_t0.imec1.wiring.json",
67 "probe02/_spikeglx_ephysData_g0_t0.imec.ap.meta", # 3A
68 "probe02/_spikeglx_ephysData_g0_t0.imec.lf.meta",
69 "probe02/_spikeglx_ephysData_g0_t0.imec.ap.bin",
70 "probe02/_spikeglx_ephysData_g0_t0.imec.lf.bin",
71 "_spikeglx_ephysData_g0_t0.nidq.sync.npy",
72 "_spikeglx_ephysData_g0_t0.nidq.meta",
73 "_spikeglx_ephysData_g0_t0.nidq.cbin",
74 "_spikeglx_ephysData_g0_t0.nidq.ch",
75 "_spikeglx_ephysData_g0_t0.wiring.json",
76 ]
77 for f in file_list: 1ab
78 fpath = raw_ephys_data_path / Path(f) 1ab
79 fpath.parent.mkdir(parents=True, exist_ok=True) 1ab
80 fpath.touch() 1ab
82 return session_path 1ab
85def populate_raw_spikeglx(session_path,
86 model='3B', legacy=False, user_label='my_run', n_probes=2):
87 """
88 Touch file tree to emulate files saved by SpikeGLX
89 :param session_path: The raw ephys data path to place files
90 :param model: Probe model file structure ('3A' or '3B')
91 :param legacy: If true, emulate older SpikeGLX version where all files are saved
92 into a single folder
93 :param user_label: User may input any name into SpikeGLX and filenames will include this
94 :param n_probes: Number of probe datafiles to touch
95 :return:
97 Examples:
98 populate_raw_spikeglx('3A_folder', model='3A', legacy=True, n_probes=1)
99 3A_folder
100 └───raw_ephys_folder
101 my_run_probe00_g0_t0.imec.ap.bin
102 my_run_probe00_g0_t0.imec.ap.meta
103 my_run_probe00_g0_t0.imec.lf.bin
104 my_run_probe00_g0_t0.imec.lf.meta
106 populate_raw_spikeglx('3B_folder', model='3B', n_probes=3)
107 3B_folder
108 └───my_run_g0_t0
109 my_run_g0_t0.imec0.ap.bin
110 my_run_g0_t0.imec0.ap.meta
111 my_run_g0_t0.imec0.lf.bin
112 my_run_g0_t0.imec0.lf.meta
113 my_run_g0_t0.imec1.ap.bin
114 my_run_g0_t0.imec1.ap.meta
115 my_run_g0_t0.imec1.lf.bin
116 my_run_g0_t0.imec1.lf.meta
117 my_run_g0_t0.imec2.ap.bin
118 my_run_g0_t0.imec2.ap.meta
119 my_run_g0_t0.imec2.lf.bin
120 my_run_g0_t0.imec2.lf.meta
121 my_run_g0_t0.nidq.bin
122 my_run_g0_t0.nidq.meta
124 populate_raw_spikeglx('3B_folder', model='3B', n_probes=0)
125 3B_folder
126 └───my_run_g0_t0
127 my_run_g0_t0.nidq.bin
128 my_run_g0_t0.nidq.meta
130 See also: http://billkarsh.github.io/SpikeGLX/help/parsing/
131 """
132 for i in range(n_probes):
133 label = (user_label + f'_probe{i:02}') if legacy and model == '3A' else user_label
134 root = session_path.joinpath('raw_ephys_folder' if legacy else f'{user_label}_g0_t0')
135 root.mkdir(exist_ok=True, parents=True)
136 for ext in ('meta', 'bin'):
137 for freq in ('lf', 'ap'):
138 filename = f'{label}_g0_t0.imec{i if model == "3B" else ""}.{freq}.{ext}'
139 root.joinpath(filename).touch()
140 if model == '3B':
141 root.joinpath(f'{label}_g0_t0.nidq.{ext}').touch()
142 if n_probes == 0:
143 if model != '3B':
144 raise NotImplementedError
145 root = session_path.joinpath(f'{user_label}_g0_t0')
146 root.mkdir(exist_ok=True, parents=True)
147 for ext in ('meta', 'bin'):
148 root.joinpath(f'{user_label}_g0_t0.nidq.{ext}').touch()
151def create_fake_raw_video_data_folder(session_path, populate=True, write_pars_stub=False):
152 """
153 Create the folder structure for a raw video session with three cameras.
154 Creates a raw_video_data folder and optionally, touches some files and writes a experiment
155 description stub to a _devices folder.
157 Parameters
158 ----------
159 session_path : str, pathlib.Path
160 The session path in which to create the folders.
161 populate : bool
162 If true, touch some raw video files.
163 write_pars_stub : bool, str, dict
164 If true, write an experiment description stub containing behaviour settings. If a string,
165 the stub filename will contain this. If a dict, the key is used as the filename; the value,
166 the file contents.
168 Example
169 -------
170 >>> create_fake_raw_video_data_folder(session_path, populate=False, write_pars_stub=False)
171 >>> create_fake_raw_video_data_folder(session_path, write_pars_stub='hostname_19826354')
172 """
173 session_path = Path(session_path)
174 raw_video_data_path = session_path / "raw_video_data"
175 raw_video_data_path.mkdir(exist_ok=True, parents=True)
176 if populate:
177 file_list = [
178 "_iblrig_leftCamera.raw.mp4",
179 "_iblrig_rightCamera.raw.mp4",
180 "_iblrig_bodyCamera.raw.mp4",
181 "_iblrig_leftCamera.timestamps.ssv",
182 "_iblrig_rightCamera.timestamps.ssv",
183 "_iblrig_bodyCamera.timestamps.ssv",
184 "_iblrig_leftCamera.GPIO.bin",
185 "_iblrig_rightCamera.GPIO.bin",
186 "_iblrig_bodyCamera.GPIO.bin",
187 "_iblrig_leftCamera.frame_counter.bin",
188 "_iblrig_rightCamera.frame_counter.bin",
189 "_iblrig_bodyCamera.frame_counter.bin",
190 "_iblrig_VideoCodeFiles.raw.zip",
191 ]
192 for f in file_list:
193 fpath = raw_video_data_path / Path(f)
194 fpath.parent.mkdir(parents=True, exist_ok=True)
195 fpath.touch()
197 if write_pars_stub:
198 if isinstance(write_pars_stub, dict):
199 (name, data), = write_pars_stub.items()
200 else:
201 name = write_pars_stub if isinstance(write_pars_stub, str) else 'video'
202 d = {'collection': 'raw_video_data', 'sync_label': 'frame2ttl'}
203 data = {
204 'devices': {'cameras': {k: d.copy() for k in ('body', 'left', 'right')}},
205 'version': session_params.SPEC_VERSION
206 }
207 file_device = session_path.joinpath(f'_ibl_experiment.description_{name}.yaml')
208 file_device.parent.mkdir(exist_ok=True)
209 session_params.write_yaml(file_device, data)
210 return raw_video_data_path
213def create_fake_alf_folder_dlc_data(session_path, populate=True):
214 session_path = Path(session_path)
215 alf_path = session_path / "alf"
216 alf_path.mkdir(exist_ok=True, parents=True)
217 if populate:
218 file_list = [
219 "_ibl_leftCamera.dlc.pqt",
220 "_ibl_rightCamera.dlc.pqt",
221 "_ibl_bodyCamera.dlc.pqt",
222 "_ibl_leftCamera.times.npy",
223 "_ibl_rightCamera.times.npy",
224 "_ibl_bodyCamera.times.npy",
225 "_ibl_leftCamera.features.npy",
226 "_ibl_rightCamera.features.npy",
227 ]
228 for f in file_list:
229 fpath = alf_path / Path(f)
230 fpath.parent.mkdir(parents=True, exist_ok=True)
231 fpath.touch()
234def create_fake_raw_behavior_data_folder(
235 session_path, populate=True, task="ephysCW", folder="raw_behavior_data", write_pars_stub=False
236):
237 """Create the folder structure for a raw behaviour session.
239 Creates a raw_behavior_data folder and optionally, touches some files and writes an experiment
240 description stub to a `_devices` folder.
242 Parameters
243 ----------
244 session_path : pathlib.Path
245 The session path in which to create the folders.
246 populate : bool
247 If true, touch some raw behaviour files.
248 task : str
249 The name of the task protocol, if 'ephys' or 'passive' extra files are touched.
250 write_pars_stub : bool, str, dict
251 If true, write an experiment description stub containing behaviour settings. If a string,
252 the stub will be named as such. If a dict, the key is used as the filename; the value,
253 the file contents.
255 Returns
256 -------
257 pathlib.Path
258 The raw behaviour data path.
259 """
260 raw_behavior_data_path = session_path / folder
261 raw_behavior_data_path.mkdir(exist_ok=True, parents=True)
262 ephysCW = [
263 "_iblrig_ambientSensorData.raw.jsonable",
264 "_iblrig_encoderEvents.raw.ssv",
265 "_iblrig_encoderPositions.raw.ssv",
266 "_iblrig_encoderTrialInfo.raw.ssv",
267 "_iblrig_micData.raw.wav",
268 "_iblrig_stimPositionScreen.raw.csv",
269 "_iblrig_syncSquareUpdate.raw.csv",
270 "_iblrig_taskCodeFiles.raw.zip",
271 "_iblrig_taskData.raw.jsonable",
272 "_iblrig_taskSettings.raw.json",
273 "online_plot.png",
274 ]
275 passiveCW = [
276 "_iblrig_encoderEvents.raw.ssv",
277 "_iblrig_encoderPositions.raw.ssv",
278 "_iblrig_encoderTrialInfo.raw.ssv",
279 "_iblrig_RFMapStim.raw.bin",
280 "_iblrig_stimPositionScreen.raw.csv",
281 "_iblrig_syncSquareUpdate.raw.csv",
282 "_iblrig_taskCodeFiles.raw.zip",
283 "_iblrig_taskSettings.raw.json",
284 ]
286 if populate:
287 file_list = []
288 if "ephys" in task:
289 file_list = ephysCW
290 elif "passive" in task:
291 file_list = passiveCW
292 (session_path / "passive_data_for_ephys.flag").touch()
293 else:
294 print(f"Not implemented: Task {task}")
296 for f in file_list:
297 fpath = raw_behavior_data_path / Path(f)
298 fpath.parent.mkdir(parents=True, exist_ok=True)
299 fpath.touch()
301 if write_pars_stub:
302 if isinstance(write_pars_stub, dict):
303 (name, data), = write_pars_stub.items()
304 else:
305 name = write_pars_stub if isinstance(write_pars_stub, str) else 'behaviour'
306 data = {
307 'devices': {'microphone': {'microphone': {'collection': folder, 'sync_label': None}}},
308 'procedures': ['Behavior training/tasks'],
309 'projects': ['ibl_neuropixel_brainwide_01'],
310 'tasks': [{task: {'collection': folder}}],
311 'version': session_params.SPEC_VERSION
312 }
313 if 'ephys' in task:
314 data['tasks'][0][task]['sync_label'] = 'frame2ttl'
315 else:
316 data['sync'] = {'bpod': {'collection': 'raw_behavior_data', 'extension': 'jsonable'}}
317 data['tasks'][0][task]['sync_label'] = 'bpod'
319 file_device = session_path.joinpath(f'_ibl_experiment.description_{name}.yaml')
320 file_device.parent.mkdir(exist_ok=True)
321 session_params.write_yaml(file_device, data)
323 return raw_behavior_data_path
326def populate_task_settings(fpath: Path, patch: dict):
327 """
328 Populate a task settings JSON file.
330 Parameters
331 ----------
332 fpath : pathlib.Path
333 A path to a raw task settings folder or the full settings file path.
334 patch : dict
335 The settings dict to write to file.
337 Returns
338 -------
339 pathlib.Path
340 The full settings file path.
341 """
342 if fpath.is_dir(): 1ad
343 fpath /= '_iblrig_taskSettings.raw.json'
344 with fpath.open('w') as f: 1ad
345 json.dump(patch, f, indent=1) 1ad
346 return fpath 1ad
349def create_fake_complete_ephys_session(
350 root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True
351):
352 session_path = create_fake_session_folder(
353 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment
354 )
355 _mouse, _date, _num = session_path.parts[-3:]
356 create_fake_raw_ephys_data_folder(session_path, populate=True)
357 create_fake_raw_video_data_folder(session_path, populate=True)
358 create_fake_raw_behavior_data_folder(session_path, populate=True, task="ephys")
359 create_fake_raw_behavior_data_folder(
360 session_path, populate=True, task="passive", folder="raw_passive_data"
361 )
362 fpath = Path(session_path) / "raw_passive_data" / "_iblrig_taskSettings.raw.json"
363 passive_settings = {
364 "CORRESPONDING_EPHYS_SESSION":
365 f"C:\\some\\root\\folder\\Subjects\\{_mouse}\\{_date}\\{_num}"
366 }
367 populate_task_settings(fpath, passive_settings)
368 if session_path.joinpath("passive_data_for_ephys.flag").exists():
369 session_path.joinpath("passive_data_for_ephys.flag").unlink()
371 return session_path
374def create_fake_ephys_recording_bad_passive_transfer_sessions(
375 root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True
376):
377 session_path = create_fake_session_folder(
378 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment
379 )
380 _mouse, _date, _num = session_path.parts[-3:]
381 create_fake_raw_ephys_data_folder(session_path, populate=True)
382 create_fake_raw_video_data_folder(session_path, populate=True)
383 create_fake_raw_behavior_data_folder(session_path, populate=True, task="ephys")
385 passive_session_path = create_fake_session_folder(
386 root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment
387 )
388 create_fake_raw_behavior_data_folder(passive_session_path, populate=True, task="passive")
389 fpath = Path(passive_session_path) / "raw_behavior_data" / "_iblrig_taskSettings.raw.json"
390 passive_settings = {
391 "CORRESPONDING_EPHYS_SESSION":
392 f"C:\\some\\root\\folder\\Subjects\\{_mouse}\\{_date}\\{_num}"
393 }
394 populate_task_settings(fpath, passive_settings)
396 return session_path, passive_session_path
399def register_new_session(one, subject=None, date=None):
400 """
401 Register a new test session.
403 NB: This creates the session path on disk, using `one.cache_dir`.
405 Parameters
406 ----------
407 one : one.api.OneAlyx
408 An instance of ONE.
409 subject : str
410 The subject name. If None, a new random subject is created.
411 date : str
412 An ISO date string. If None, a random one is created.
414 Returns
415 -------
416 pathlib.Path
417 New local session path.
418 uuid.UUID
419 The experiment UUID.
420 """
421 if not date: 1ac
422 date = f'20{random.randint(0, 99):02}-{random.randint(1, 12):02}-{random.randint(1, 28):02}' 1ac
423 if not subject: 1ac
424 subject = ''.join(random.choices(string.ascii_letters, k=10))
425 one.alyx.rest('subjects', 'create', data={'lab': 'mainenlab', 'nickname': subject})
427 session_path, eid = RegistrationClient(one).create_new_session(subject, date=str(date)[:10]) 1ac
428 _logger.debug('Registered session %s with eid %s', session_path, eid) 1ac
429 return session_path, eid 1ac