Source code for ibllib.tests.fixtures.utils

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @Author: Niccolò Bonacchi
# @Date: Friday, October 9th 2020, 12:02:56 pm
import json
from pathlib import Path


[docs]def create_fake_session_folder( root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True ): root_data_path = Path(root_data_path) session_path = root_data_path / lab / "Subjects" / mouse / date / num if session_path.exists() and increment: num = str(int(num) + 1).zfill(3) return create_fake_session_folder( root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment ) session_path = root_data_path / lab / "Subjects" / mouse / date / num session_path.mkdir(exist_ok=True, parents=True) return session_path
[docs]def create_fake_raw_ephys_data_folder(session_path, populate=True): """create_fake_raw_ephys_data_folder creates raw_ephys_data folder can populate with empty files with expected names :param session_path: [description] :type session_path: [type] :param populate: [description], defaults to True :type populate: bool, optional :return: [description] :rtype: [type] """ session_path = Path(session_path) raw_ephys_data_path = session_path / "raw_ephys_data" raw_ephys_data_path.mkdir(exist_ok=True, parents=True) if populate: file_list = [ "probe00/_spikeglx_ephysData_g0_t0.imec0.sync.npy", "probe00/_spikeglx_ephysData_g0_t0.imec0.lf.meta", "probe00/_spikeglx_ephysData_g0_t0.imec0.ap.cbin", "probe00/_spikeglx_ephysData_g0_t0.imec0.timestamps.npy", "probe00/_spikeglx_ephysData_g0_t0.imec0.lf.cbin", "probe00/_spikeglx_ephysData_g0_t0.imec0.ap.ch", "probe00/_spikeglx_ephysData_g0_t0.imec0.wiring.json", "probe00/_spikeglx_sync.times.probe00.npy", "probe00/_spikeglx_sync.channels.probe00.npy", "probe00/_spikeglx_sync.polarities.probe00.npy", "probe01/_spikeglx_ephysData_g0_t0.imec1.sync.npy", "probe01/_spikeglx_ephysData_g0_t0.imec1.lf.meta", "probe01/_spikeglx_ephysData_g0_t0.imec1.timestamps.cbin", "probe01/_spikeglx_ephysData_g0_t0.imec1.ap.cbin", "probe01/_spikeglx_ephysData_g0_t0.imec1.lf.cbin", "probe01/_spikeglx_ephysData_g0_t0.imec1.ap.ch", "probe01/_spikeglx_ephysData_g0_t0.imec1.wiring.json", "probe02/_spikeglx_ephysData_g0_t0.imec.ap.meta", # 3A "probe02/_spikeglx_ephysData_g0_t0.imec.lf.meta", "probe02/_spikeglx_ephysData_g0_t0.imec.ap.bin", "probe02/_spikeglx_ephysData_g0_t0.imec.lf.bin", "_spikeglx_ephysData_g0_t0.nidq.sync.npy", "_spikeglx_ephysData_g0_t0.nidq.meta", "_spikeglx_ephysData_g0_t0.nidq.cbin", "_spikeglx_ephysData_g0_t0.nidq.ch", "_spikeglx_ephysData_g0_t0.wiring.json", ] for f in file_list: fpath = raw_ephys_data_path / Path(f) fpath.parent.mkdir(parents=True, exist_ok=True) fpath.touch() return session_path
[docs]def populate_raw_spikeglx(session_path, model='3B', legacy=False, user_label='my_run', n_probes=2): """ Touch file tree to emulate files saved by SpikeGLX :param session_path: The raw ephys data path to place files :param model: Probe model file structure ('3A' or '3B') :param legacy: If true, the emulate older SpikeGLX version where all files are saved into a single folder :param user_label: User may input any name into SpikeGLX and filenames will include this :param n_probes: Number of probe datafiles to touch :return: Examples: populate_raw_spikeglx('3A_folder', model='3A', legacy=True, n_probes=1) 3A_folder └───raw_ephys_folder my_run_probe00_g0_t0.imec.ap.bin my_run_probe00_g0_t0.imec.ap.meta my_run_probe00_g0_t0.imec.lf.bin my_run_probe00_g0_t0.imec.lf.meta populate_raw_spikeglx('3B_folder', model='3B', n_probes=3) 3B_folder └───my_run_g0_t0 my_run_g0_t0.imec0.ap.bin my_run_g0_t0.imec0.ap.meta my_run_g0_t0.imec0.lf.bin my_run_g0_t0.imec0.lf.meta my_run_g0_t0.imec1.ap.bin my_run_g0_t0.imec1.ap.meta my_run_g0_t0.imec1.lf.bin my_run_g0_t0.imec1.lf.meta my_run_g0_t0.imec2.ap.bin my_run_g0_t0.imec2.ap.meta my_run_g0_t0.imec2.lf.bin my_run_g0_t0.imec2.lf.meta my_run_g0_t0.nidq.bin my_run_g0_t0.nidq.meta See also: http://billkarsh.github.io/SpikeGLX/help/parsing/ """ for i in range(n_probes): label = (user_label + f'_probe{i:02}') if legacy and model == '3A' else user_label root = session_path.joinpath('raw_ephys_folder' if legacy else f'{user_label}_g0_t0') root.mkdir(exist_ok=True, parents=True) for ext in ('meta', 'bin'): for freq in ('lf', 'ap'): filename = f'{label}_g0_t0.imec{i if model == "3B" else ""}.{freq}.{ext}' root.joinpath(filename).touch() if model == '3B': root.joinpath(f'{label}_g0_t0.nidq.{ext}').touch()
[docs]def create_fake_raw_video_data_folder(session_path, populate=True): session_path = Path(session_path) raw_video_data_path = session_path / "raw_video_data" raw_video_data_path.mkdir(exist_ok=True, parents=True) if populate: file_list = [ "_iblrig_leftCamera.raw.mp4", "_iblrig_rightCamera.raw.mp4", "_iblrig_bodyCamera.raw.mp4", "_iblrig_leftCamera.timestamps.ssv", "_iblrig_rightCamera.timestamps.ssv", "_iblrig_bodyCamera.timestamps.ssv", "_iblrig_leftCamera.GPIO.bin", "_iblrig_rightCamera.GPIO.bin", "_iblrig_bodyCamera.GPIO.bin", "_iblrig_leftCamera.frame_counter.bin", "_iblrig_rightCamera.frame_counter.bin", "_iblrig_bodyCamera.frame_counter.bin", "_iblrig_VideoCodeFiles.raw.zip", ] for f in file_list: fpath = raw_video_data_path / Path(f) fpath.parent.mkdir(parents=True, exist_ok=True) fpath.touch()
[docs]def create_fake_alf_folder_dlc_data(session_path, populate=True): session_path = Path(session_path) alf_path = session_path / "alf" alf_path.mkdir(exist_ok=True, parents=True) if populate: file_list = [ "_ibl_leftCamera.dlc.pqt", "_ibl_rightCamera.dlc.pqt", "_ibl_bodyCamera.dlc.pqt", "_ibl_leftCamera.times.npy", "_ibl_rightCamera.times.npy", "_ibl_bodyCamera.times.npy", ] for f in file_list: fpath = alf_path / Path(f) fpath.parent.mkdir(parents=True, exist_ok=True) fpath.touch()
[docs]def create_fake_raw_behavior_data_folder( session_path, populate=True, task="ephysCW", folder="raw_behavior_data" ): raw_behavior_data_path = session_path / folder raw_behavior_data_path.mkdir(exist_ok=True, parents=True) ephysCW = [ "_iblrig_ambientSensorData.raw.jsonable", "_iblrig_encoderEvents.raw.ssv", "_iblrig_encoderPositions.raw.ssv", "_iblrig_encoderTrialInfo.raw.ssv", "_iblrig_micData.raw.wav", "_iblrig_stimPositionScreen.raw.csv", "_iblrig_syncSquareUpdate.raw.csv", "_iblrig_taskCodeFiles.raw.zip", "_iblrig_taskData.raw.jsonable", "_iblrig_taskSettings.raw.json", "online_plot.png", ] passiveCW = [ "_iblrig_encoderEvents.raw.ssv", "_iblrig_encoderPositions.raw.ssv", "_iblrig_encoderTrialInfo.raw.ssv", "_iblrig_RFMapStim.raw.bin", "_iblrig_stimPositionScreen.raw.csv", "_iblrig_syncSquareUpdate.raw.csv", "_iblrig_taskCodeFiles.raw.zip", "_iblrig_taskSettings.raw.json", ] if populate: file_list = [] if "ephys" in task: file_list = ephysCW elif "passive" in task: file_list = passiveCW (session_path / "passive_data_for_ephys.flag").touch() else: print(f"Not implemented: Task {task}") for f in file_list: fpath = raw_behavior_data_path / Path(f) fpath.parent.mkdir(parents=True, exist_ok=True) fpath.touch() return session_path
[docs]def populate_task_settings(fpath: Path, patch: dict): with fpath.open("w") as f: json.dump(patch, f, indent=1)
[docs]def create_fake_complete_ephys_session( root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True ): session_path = create_fake_session_folder( root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment ) _mouse, _date, _num = session_path.parts[-3:] create_fake_raw_ephys_data_folder(session_path, populate=True) create_fake_raw_video_data_folder(session_path, populate=True) create_fake_raw_behavior_data_folder(session_path, populate=True, task="ephys") create_fake_raw_behavior_data_folder( session_path, populate=True, task="passive", folder="raw_passive_data" ) fpath = Path(session_path) / "raw_passive_data" / "_iblrig_taskSettings.raw.json" passive_settings = { "CORRESPONDING_EPHYS_SESSION": f"C:\\some\\root\\folder\\Subjects\\{_mouse}\\{_date}\\{_num}" } populate_task_settings(fpath, passive_settings) if session_path.joinpath("passive_data_for_ephys.flag").exists(): session_path.joinpath("passive_data_for_ephys.flag").unlink() return session_path
[docs]def create_fake_ephys_recording_bad_passive_transfer_sessions( root_data_path, lab="fakelab", mouse="fakemouse", date="1900-01-01", num="001", increment=True ): session_path = create_fake_session_folder( root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment ) _mouse, _date, _num = session_path.parts[-3:] create_fake_raw_ephys_data_folder(session_path, populate=True) create_fake_raw_video_data_folder(session_path, populate=True) create_fake_raw_behavior_data_folder(session_path, populate=True, task="ephys") passive_session_path = create_fake_session_folder( root_data_path, lab=lab, mouse=mouse, date=date, num=num, increment=increment ) create_fake_raw_behavior_data_folder(passive_session_path, populate=True, task="passive") fpath = Path(passive_session_path) / "raw_behavior_data" / "_iblrig_taskSettings.raw.json" passive_settings = { "CORRESPONDING_EPHYS_SESSION": f"C:\\some\\root\\folder\\Subjects\\{_mouse}\\{_date}\\{_num}" } populate_task_settings(fpath, passive_settings) return session_path, passive_session_path