Coverage for ibllib/pipes/misc.py: 59%

1"""Miscellaneous pipeline utility functions.""" 

2import ctypes (empty)

3import hashlib (empty)

4import json (empty)

5import os (empty)

6import re (empty)

7import shutil (empty)

8import subprocess (empty)

9import sys (empty)

10import time (empty)

11import logging (empty)

12import warnings (empty)

13from functools import wraps (empty)

14from pathlib import Path (empty)

15from typing import Union, List, Callable, Any (empty)

16from inspect import signature (empty)

17import uuid (empty)

18import socket (empty)

19import traceback (empty)

20import tempfile (empty)

21 

22import spikeglx (empty)

23from iblutil.io import hashfile, params (empty)

24from iblutil.util import range_str (empty)

25from one.alf.files import get_session_path (empty)

26from one.alf.spec import is_uuid_string, is_session_path, describe (empty)

27from one.api import ONE (empty)

28 

29import ibllib.io.flags as flags (empty)

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

31from ibllib.io.misc import delete_empty_folders (empty)

32import ibllib.io.session_params as sess_params (empty)

33 

34log = logging.getLogger(__name__) (empty)

35 

36DEVICE_FLAG_MAP = {'neuropixel': 'ephys', (empty)

37 'cameras': 'video', 

38 'widefield': 'widefield', 

39 'sync': 'sync'} 

40 

41 

42def subjects_data_folder(folder: Path, rglob: bool = False) -> Path: (empty)

43 """Given a root_data_folder will try to find a 'Subjects' data folder. 

44 

45 If Subjects folder is passed will return it directly.""" 

46 if not isinstance(folder, Path): 7 ctx1imfdage

47 folder = Path(folder) 4 ctx1fdge

48 if rglob: 7 ctx1imfdage

49 func = folder.rglob 7 ctx1imfdage

50 else: 

51 func = folder.glob 

52 

53 # Try to find Subjects folder one level 

54 if folder.name.lower() != 'subjects': 7 ctx1imfdage

55 # Try to find Subjects folder if folder.glob 

56 spath = [x for x in func('*') if x.name.lower() == 'subjects'] 7 ctx1imfdage

57 if not spath: 7 ctx1imfdage

58 raise ValueError('No "Subjects" folder in children folders') 

59 elif len(spath) > 1: 7 ctx1imfdage

60 raise ValueError(f'Multiple "Subjects" folder in children folders: {spath}') 

61 else: 

62 folder = folder / spath[0] 7 ctx1imfdage

63 

64 return folder 7 ctx1imfdage

65 

66 

67def cli_ask_default(prompt: str, default: str): (empty)

68 """ 

69 Prompt the user for input, display the default option and return user input or default 

70 :param prompt: String to display to user 

71 :param default: The default value to return if user doesn't enter anything 

72 :return: User input or default 

73 """ 

74 return input(f'{prompt} [default: {default}]: ') or default 1 ctx1j

75 

76 

77def cli_ask_options(prompt: str, options: list, default_idx: int = 0) -> str: (empty)

78 parsed_options = [str(x) for x in options] 

79 if default_idx is not None: 

80 parsed_options[default_idx] = f"[{parsed_options[default_idx]}]" 

81 options_str = " (" + " | ".join(parsed_options) + ")> " 

82 ans = input(prompt + options_str) or str(options[default_idx]) 

83 if ans not in [str(x) for x in options]: 

84 return cli_ask_options(prompt, options, default_idx=default_idx) 

85 return ans 

86 

87 

88def behavior_exists(session_path: str, include_devices=False) -> bool: (empty)

89 """ 

90 Returns True if the session has a task behaviour folder 

91 :param session_path: 

92 :return: 

93 """ 

94 session_path = Path(session_path) 7 ctx1zcfdage

95 if include_devices and session_path.joinpath("_devices").exists(): 7 ctx1zcfdage

96 return True 

97 if session_path.joinpath("raw_behavior_data").exists(): 7 ctx1zcfdage

98 return True 7 ctx1zcfdage

99 return any(session_path.glob('raw_task_data_*')) 3 ctx1zca

100 

101 

102def check_transfer(src_session_path, dst_session_path): (empty)

103 """ 

104 Check all the files in the source directory match those in the destination directory. Function 

105 will throw assertion errors/exceptions if number of files do not match, file names do not 

106 match, or if file sizes do not match. 

107 

108 :param src_session_path: The source directory that was copied 

109 :param dst_session_path: The copy target directory 

110 """ 

111 src_files = sorted([x for x in Path(src_session_path).rglob('*') if x.is_file()]) 1 ctx1y

112 dst_files = sorted([x for x in Path(dst_session_path).rglob('*') if x.is_file()]) 1 ctx1y

113 assert len(src_files) == len(dst_files), 'Not all files transferred' 1 ctx1y

114 for s, d in zip(src_files, dst_files): 1 ctx1y

115 assert s.name == d.name, 'file name mismatch' 1 ctx1y

116 assert s.stat().st_size == d.stat().st_size, 'file size mismatch' 1 ctx1y

117 

118 

119def rename_session(session_path: str, new_subject=None, new_date=None, new_number=None, (empty)

120 ask: bool = False) -> Path: 

121 """Rename a session. Prompts the user for the new subject name, data and number and then moves 

122 session path to new session path. 

123 

124 :param session_path: A session path to rename 

125 :type session_path: str 

126 :param new_subject: A new subject name, if none provided, the user is prompted for one 

127 :param new_date: A new session date, if none provided, the user is prompted for one 

128 :param new_number: A new session number, if none provided, the user is prompted for one 

129 :param ask: used to ensure prompt input from user, defaults to False 

130 :type ask: bool 

131 :return: The renamed session path 

132 :rtype: Path 

133 """ 

134 session_path = get_session_path(session_path) 1 ctx1p

135 if session_path is None: 1 ctx1p

136 raise ValueError('Session path not valid ALF session folder') 1 ctx1p

137 mouse = session_path.parts[-3] 1 ctx1p

138 date = session_path.parts[-2] 1 ctx1p

139 sess = session_path.parts[-1] 1 ctx1p

140 new_mouse = new_subject or mouse 1 ctx1p

141 new_date = new_date or date 1 ctx1p

142 new_sess = new_number or sess 1 ctx1p

143 if ask: 1 ctx1p

144 new_mouse = input(f"Please insert subject NAME [current value: {mouse}]> ") 1 ctx1p

145 new_date = input(f"Please insert new session DATE [current value: {date}]> ") 1 ctx1p

146 new_sess = input(f"Please insert new session NUMBER [current value: {sess}]> ") 1 ctx1p

147 

148 new_session_path = Path(*session_path.parts[:-3]).joinpath(new_mouse, new_date, 1 ctx1p

149 new_sess.zfill(3)) 

150 assert is_session_path(new_session_path), 'invalid subject, date or number' 1 ctx1p

151 

152 if new_session_path.exists(): 1 ctx1p

153 ans = input(f'Warning: session path {new_session_path} already exists.\nWould you like to ' 

154 f'move {new_session_path} to a backup directory? [y/N] ') 

155 if (ans or 'n').lower() in ['n', 'no']: 

156 print(f'Manual intervention required, data exists in the following directory: ' 

157 f'{session_path}') 

158 return 

159 if backup_session(new_session_path): 

160 print(f'Backup was successful, removing directory {new_session_path}...') 

161 shutil.rmtree(str(new_session_path), ignore_errors=True) 

162 shutil.move(str(session_path), str(new_session_path)) 1 ctx1p

163 print(session_path, "--> renamed to:") 1 ctx1p

164 print(new_session_path) 1 ctx1p

165 

166 return new_session_path 1 ctx1p

167 

168 

169def backup_session(folder_path, root=None, extra=''): (empty)

170 """ 

171 Used to move the contents of a session to a backup folder, likely before the folder is 

172 removed. 

173 

174 Parameters 

175 ---------- 

176 folder_path : str, pathlib.Path 

177 The folder path to remove. 

178 root : str, pathlib.Path 

179 Copy folder tree relative to this. If None, copies from the session_path root. 

180 extra : str, pathlib.Path 

181 Extra folder parts to append to destination root path. 

182 

183 Returns 

184 ------- 

185 pathlib.Path 

186 The location of the backup data, if succeeded to copy. 

187 """ 

188 if not root: 1 ctx1q

189 if session_path := get_session_path(folder_path): 1 ctx1q

190 root = session_path.parents[2] 1 ctx1q

191 else: 

192 root = folder_path.parent 1 ctx1q

193 folder_path = Path(folder_path) 1 ctx1q

194 bk_path = Path(tempfile.gettempdir(), 'backup_sessions', extra, folder_path.relative_to(root)) 1 ctx1q

195 if folder_path.exists(): 1 ctx1q

196 if not folder_path.is_dir(): 1 ctx1q

197 log.error(f'The given folder path is not a directory: {folder_path}') 1 ctx1q

198 return 1 ctx1q

199 try: 1 ctx1q

200 log.debug(f'Created path: {bk_path.parent}') 1 ctx1q

201 bk_path = shutil.copytree(folder_path, bk_path) 1 ctx1q

202 log.info(f'Copied contents from {folder_path} to {bk_path}') 1 ctx1q

203 return bk_path 1 ctx1q

204 except FileExistsError: 1 ctx1q

205 log.error(f'A backup session for the given path already exists: {bk_path}, ' 1 ctx1q

206 f'manual intervention is necessary.') 

207 except shutil.Error as ex: 1 ctx1q

208 log.error('Failed to copy files from %s to %s: %s', folder_path, bk_path, ex) 1 ctx1q

209 else: 

210 log.error('The given session path does not exist: %s', folder_path) 1 ctx1q

211 

212 

213def copy_with_check(src, dst, **kwargs): (empty)

214 dst = Path(dst) 

215 if dst.exists() and Path(src).stat().st_size == dst.stat().st_size: 

216 return dst 

217 elif dst.exists(): 

218 dst.unlink() 

219 return shutil.copy2(src, dst, **kwargs) 

220 

221 

222def transfer_session_folders(local_sessions: list, remote_subject_folder, subfolder_to_transfer): (empty)

223 """ 

224 Used to determine which local session folders should be transferred to which remote session folders, will prompt the user 

225 when necessary. 

226 

227 Parameters 

228 ---------- 

229 local_sessions : list 

230 Required list of local session folder paths to sync to local server. 

231 remote_subject_folder : str, pathlib.Path 

232 The remote location of the subject folder (typically pulled from the params). 

233 subfolder_to_transfer : str 

234 Which subfolder to sync 

235 

236 Returns 

237 ------- 

238 list of tuples 

239 For each session, a tuple of (source, destination) of attempted file transfers. 

240 list of bool 

241 A boolean True/False for success/failure of the transfer. 

242 """ 

243 transfer_list = [] # list of sessions to transfer 6 ctx1cfdage

244 skip_list = "" # "list" of sessions to skip and the reason for the skip 6 ctx1cfdage

245 # Iterate through all local sessions in the given list 

246 for local_session in local_sessions: 6 ctx1cfdage

247 # Set expected remote_session location and perform simple error state checks 

248 remote_session = remote_subject_folder.joinpath(*local_session.parts[-3:]) 6 ctx1cfdage

249 # Skip session if ... 

250 if subfolder_to_transfer: 6 ctx1cfdage

251 if not local_session.joinpath(subfolder_to_transfer).exists(): 6 ctx1cfdage

252 msg = f"{local_session} - skipping session, no '{subfolder_to_transfer}' folder found locally" 

253 log.warning(msg) 

254 skip_list += msg + "\n" 

255 continue 

256 if not remote_session.parent.exists(): 6 ctx1cfdage

257 msg = f"{local_session} - no matching remote session date folder found for the given local session" 2 ctx1ca

258 log.info(msg) 2 ctx1ca

259 skip_list += msg + "\n" 2 ctx1ca

260 continue 2 ctx1ca

261 if not behavior_exists(remote_session): 6 ctx1cfdage

262 msg = f"{local_session} - skipping session, no behavior data found in remote folder {remote_session}" 2 ctx1ca

263 log.warning(msg) 2 ctx1ca

264 skip_list += msg + "\n" 2 ctx1ca

265 continue 2 ctx1ca

266 

267 # Determine if there are multiple session numbers from the date path 

268 local_sessions_for_date = get_session_numbers_from_date_path(local_session.parent) 6 ctx1cfdage

269 remote_sessions_for_date = get_session_numbers_from_date_path(remote_session.parent) 6 ctx1cfdage

270 remote_session_pick = None 6 ctx1cfdage

271 if len(local_sessions_for_date) > 1 or len(remote_sessions_for_date) > 1: 6 ctx1cfdage

272 # Format folder size output for end user to review 

273 local_session_numbers_with_size = remote_session_numbers_with_size = "" 2 ctx1ca

274 for lsfd in local_sessions_for_date: 2 ctx1ca

275 size_in_gb = round(get_directory_size(local_session.parent / lsfd, in_gb=True), 2) 2 ctx1ca

276 local_session_numbers_with_size += lsfd + " (" + str(size_in_gb) + " GB)\n" 2 ctx1ca

277 for rsfd in remote_sessions_for_date: 2 ctx1ca

278 size_in_gb = round(get_directory_size(remote_session.parent / rsfd, in_gb=True), 2) 2 ctx1ca

279 remote_session_numbers_with_size += rsfd + " (" + str(size_in_gb) + " GB)\n" 2 ctx1ca

280 log.info(f"\n\nThe following local session folder(s) were found on this acquisition PC:\n\n" 2 ctx1ca

281 f"{''.join(local_session_numbers_with_size)}\nThe following remote session folder(s) were found on the " 

282 f"server:\n\n{''.join(remote_session_numbers_with_size)}\n") 

283 

284 def _remote_session_picker(sessions_for_date): 2 ctx1ca

285 resp = "s" 2 ctx1ca

286 resp_invalid = True 2 ctx1ca

287 while resp_invalid: # loop until valid user input 2 ctx1ca

288 resp = input(f"\n\n--- USER INPUT NEEDED ---\nWhich REMOTE session number would you like to transfer your " 2 ctx1ca

289 f"local session to? Options {range_str(map(int, sessions_for_date))} or " 

290 f"[s]kip/[h]elp/[e]xit> ").strip().lower() 

291 if resp == "h": 2 ctx1ca

292 print("An example session filepath:\n") 

293 describe("number") # Explain what a session number is 

294 input("Press enter to continue") 

295 elif resp == "s" or resp == "e": # exit loop 2 ctx1ca

296 resp_invalid = False 

297 elif len(resp) <= 3: 2 ctx1ca

298 resp_invalid = False if [i for i in sessions_for_date if int(resp) == int(i)] else None 2 ctx1ca

299 else: 

300 print("Invalid response. Please try again.") 

301 return resp 2 ctx1ca

302 

303 log.info(f"Evaluation for local session " 2 ctx1ca

304 f"{local_session.parts[-3]}/{local_session.parts[-2]}/{local_session.parts[-1]}...") 

305 user_response = _remote_session_picker(remote_sessions_for_date) 2 ctx1ca

306 if user_response == "s": 2 ctx1ca

307 msg = f"{local_session} - Local session skipped due to user input" 

308 log.info(msg) 

309 skip_list += msg + "\n" 

310 continue 

311 elif user_response == "e": 2 ctx1ca

312 log.info("Exiting, no files transferred.") 

313 return 

314 else: 

315 remote_session_pick = remote_session.parent / user_response.zfill(3) 2 ctx1ca

316 

317 # Append to the transfer_list 

318 transfer_tuple = (local_session, remote_session_pick) if remote_session_pick else (local_session, remote_session) 6 ctx1cfdage

319 transfer_list.append(transfer_tuple) 6 ctx1cfdage

320 log.info(f"{transfer_tuple[0]}, {transfer_tuple[1]} - Added to the transfer list") 6 ctx1cfdage

321 

322 # Verify that the number of local transfer_list entries match the number of remote transfer_list entries 

323 if len(transfer_list) != len(set(dst for _, dst in transfer_list)): 6 ctx1cfdage

324 raise RuntimeError( 

325 "An invalid combination of sessions were picked; the most likely cause of this error is multiple local " 

326 "sessions being selected for a single remote session. Please rerun the script." 

327 ) 

328 

329 # Call rsync/rdiff function for every entry in the transfer list 

330 success = [] 6 ctx1cfdage

331 for src, dst in transfer_list: 6 ctx1cfdage

332 if subfolder_to_transfer: 6 ctx1cfdage

333 success.append(rsync_paths(src / subfolder_to_transfer, dst / subfolder_to_transfer)) 6 ctx1cfdage

334 else: 

335 success.append(rsync_paths(src, dst)) 

336 if not success[-1]: 6 ctx1cfdage

337 log.error("File transfer failed, check log for reason.") 

338 

339 # Notification to user for any transfers were skipped 

340 log.warning(f"Video transfers that were not completed:\n\n{skip_list}") if skip_list else log.info("No transfers skipped.") 6 ctx1cfdage

341 return transfer_list, success 6 ctx1cfdage

342 

343 

344def transfer_folder(src: Path, dst: Path, force: bool = False) -> None: (empty)

345 """functionality has been replaced by transfer_session_folders function""" 

346 print(f"Attempting to copy:\n{src}\n--> {dst}") 

347 if force: 

348 print(f"Removing {dst}") 

349 shutil.rmtree(dst, ignore_errors=True) 

350 else: 

351 try: 

352 check_transfer(src, dst) 

353 print("All files already copied, use force=True to re-copy") 

354 return 

355 except AssertionError: 

356 pass 

357 print(f"Copying all files:\n{src}\n--> {dst}") 

358 # rsync_folder(src, dst, '**transfer_me.flag') 

359 if sys.version_info.minor < 8: 

360 # dirs_exist_ok kwarg not supported in < 3.8 

361 shutil.rmtree(dst, ignore_errors=True) 

362 shutil.copytree(src, dst, copy_function=copy_with_check) 

363 else: 

364 shutil.copytree(src, dst, dirs_exist_ok=True, copy_function=copy_with_check) 

365 # If folder was created delete the src_flag_file 

366 if check_transfer(src, dst) is None: 

367 print("All files copied") 

368 # rdiff-backup --compare /tmp/tmpw9o1zgn0 /tmp/tmp82gg36rm 

369 # No changes found. Directory matches archive data. 

370 

371 

372def load_params_dict(params_fname: str) -> dict: (empty)

373 params_fpath = Path(params.getfile(params_fname)) 

374 if not params_fpath.exists(): 

375 return None 

376 with open(params_fpath, "r") as f: 

377 out = json.load(f) 

378 return out 

379 

380 

381def load_videopc_params(): (empty)

382 """(DEPRECATED) This will be removed in favour of iblrigv8 functions.""" 

383 warnings.warn('load_videopc_params will be removed in favour of iblrigv8', FutureWarning) 

384 if not load_params_dict("videopc_params"): 

385 create_videopc_params() 

386 return load_params_dict("videopc_params") 

387 

388 

389def load_ephyspc_params(): (empty)

390 if not load_params_dict("ephyspc_params"): 

391 create_ephyspc_params() 

392 return load_params_dict("ephyspc_params") 

393 

394 

395def create_basic_transfer_params(param_str='transfer_params', local_data_path=None, (empty)

396 remote_data_path=None, clobber=False, **kwargs): 

397 """Create some basic parameters common to all acquisition rigs. 

398 

399 Namely prompt user for the local root data path and the remote (lab server) data path. 

400 NB: All params stored in uppercase by convention. 

401 

402 Parameters 

403 ---------- 

404 param_str : str 

405 The name of the parameters to load/save. 

406 local_data_path : str, pathlib.Path 

407 The local root data path, stored with the DATA_FOLDER_PATH key. If None, user is prompted. 

408 remote_data_path : str, pathlib.Path, bool 

409 The local root data path, stored with the REMOTE_DATA_FOLDER_PATH key. If None, user is prompted. 

410 If False, the REMOTE_DATA_PATH key is not updated or is set to False if clobber = True. 

411 clobber : bool 

412 If True, any parameters in existing parameter file not found as keyword args will be removed, 

413 otherwise the user is prompted for these also. 

414 **kwargs 

415 Extra parameters to set. If value is None, the user is prompted. 

416 

417 Returns 

418 ------- 

419 dict 

420 The parameters written to disc. 

421 

422 Examples 

423 -------- 

424 Set up basic transfer parameters for modality acquisition PC 

425 

426 >>> par = create_basic_transfer_params() 

427 

428 Set up basic transfer paramers without prompting the user 

429 

430 >>> par = create_basic_transfer_params( 

431 ... local_data_path='/iblrig_data/Subjects', 

432 ... remote_data_path='/mnt/iblserver.champalimaud.pt/ibldata/Subjects') 

433 

434 Prompt user for extra parameter using custom prompt (will call function with current default) 

435 

436 >>> from functools import partial 

437 >>> par = create_basic_transfer_params( 

438 ... custom_arg=partial(cli_ask_default, 'Please enter custom arg value')) 

439 

440 Set up with no remote path (NB: if not the first time, use clobber=True to save param key) 

441 

442 >>> par = create_basic_transfer_params(remote_data_path=False) 

443 

444 """ 

445 parameters = params.as_dict(params.read(param_str, {})) or {} 7 ctx1jrsfdge

446 if local_data_path is None: 7 ctx1jrsfdge

447 local_data_path = parameters.get('DATA_FOLDER_PATH') 3 ctx1jrs

448 if not local_data_path or clobber: 3 ctx1jrs

449 local_data_path = cli_ask_default("Where's your LOCAL 'Subjects' data folder?", local_data_path) 1 ctx1j

450 parameters['DATA_FOLDER_PATH'] = local_data_path 7 ctx1jrsfdge

451 

452 if remote_data_path is None: 7 ctx1jrsfdge

453 remote_data_path = parameters.get('REMOTE_DATA_FOLDER_PATH') 3 ctx1jrs

454 if remote_data_path in (None, '') or clobber: 3 ctx1jrs

455 remote_data_path = cli_ask_default("Where's your REMOTE 'Subjects' data folder?", remote_data_path) 1 ctx1j

456 if remote_data_path is not False: 7 ctx1jrsfdge

457 parameters['REMOTE_DATA_FOLDER_PATH'] = remote_data_path 7 ctx1jrsfdge

458 elif 'REMOTE_DATA_FOLDER_PATH' not in parameters or clobber: 1 ctx1j

459 parameters['REMOTE_DATA_FOLDER_PATH'] = False # Always assume no remote path 1 ctx1j

460 

461 # Deal with extraneous parameters 

462 for k, v in kwargs.items(): 7 ctx1jrsfdge

463 if callable(v): # expect function handle with default value as input 1 ctx1j

464 n_pars = len(signature(v).parameters) 1 ctx1j

465 parameters[k.upper()] = v(parameters.get(k.upper())) if n_pars > 0 else v() 1 ctx1j

466 elif v is None: # generic prompt for key 1 ctx1j

467 parameters[k.upper()] = cli_ask_default( 1 ctx1j

468 f'Enter a value for parameter {k.upper()}', parameters.get(k.upper()) 

469 ) 

470 else: # assign value to parameter 

471 parameters[k.upper()] = str(v) 1 ctx1j

472 

473 defined = list(map(str.upper, ('DATA_FOLDER_PATH', 'REMOTE_DATA_FOLDER_PATH', 'TRANSFER_LABEL', *kwargs.keys()))) 7 ctx1jrsfdge

474 if clobber: 7 ctx1jrsfdge

475 # Delete any parameters in parameter dict that were not passed as keyword args into function 

476 parameters = {k: v for k, v in parameters.items() if k in defined} 1 ctx1j

477 else: 

478 # Prompt for any other parameters that weren't passed into function 

479 for k in filter(lambda x: x not in defined, map(str.upper, parameters.keys())): 7 ctx1jrsfdge

480 parameters[k] = cli_ask_default(f'Enter a value for parameter {k}', parameters.get(k)) 1 ctx1j

481 

482 if 'TRANSFER_LABEL' not in parameters: 7 ctx1jrsfdge

483 parameters['TRANSFER_LABEL'] = f'{socket.gethostname()}_{uuid.getnode()}' 1 ctx1j

484 

485 # Write parameters 

486 params.write(param_str, parameters) 7 ctx1jrsfdge

487 return parameters 7 ctx1jrsfdge

488 

489 

490def create_videopc_params(force=False, silent=False): (empty)

491 """(DEPRECATED) This will be removed in favour of iblrigv8 functions.""" 

492 url = 'https://github.com/int-brain-lab/iblrig/blob/videopc/docs/source/video.rst' 

493 warnings.warn(f'create_videopc_params is deprecated, see {url}', FutureWarning) 

494 if Path(params.getfile("videopc_params")).exists() and not force: 

495 print(f"{params.getfile('videopc_params')} exists already, exiting...") 

496 print(Path(params.getfile("videopc_params")).exists()) 

497 return 

498 if silent: 

499 data_folder_path = r"D:\iblrig_data\Subjects" 

500 remote_data_folder_path = r"\\iblserver.champalimaud.pt\ibldata\Subjects" 

501 body_cam_idx = 0 

502 left_cam_idx = 1 

503 right_cam_idx = 2 

504 else: 

505 data_folder_path = cli_ask_default( 

506 r"Where's your LOCAL 'Subjects' data folder?", r"D:\iblrig_data\Subjects" 

507 ) 

508 remote_data_folder_path = cli_ask_default( 

509 r"Where's your REMOTE 'Subjects' data folder?", 

510 r"\\iblserver.champalimaud.pt\ibldata\Subjects", 

511 ) 

512 body_cam_idx = cli_ask_default("Please select the index of the BODY camera", "0") 

513 left_cam_idx = cli_ask_default("Please select the index of the LEFT camera", "1") 

514 right_cam_idx = cli_ask_default("Please select the index of the RIGHT camera", "2") 

515 

516 param_dict = { 

517 "DATA_FOLDER_PATH": data_folder_path, 

518 "REMOTE_DATA_FOLDER_PATH": remote_data_folder_path, 

519 "BODY_CAM_IDX": body_cam_idx, 

520 "LEFT_CAM_IDX": left_cam_idx, 

521 "RIGHT_CAM_IDX": right_cam_idx, 

522 } 

523 params.write("videopc_params", param_dict) 

524 print(f"Created {params.getfile('videopc_params')}") 

525 print(param_dict) 

526 return param_dict 

527 

528 

529def create_ephyspc_params(force=False, silent=False): (empty)

530 if Path(params.getfile("ephyspc_params")).exists() and not force: 

531 print(f"{params.getfile('ephyspc_params')} exists already, exiting...") 

532 print(Path(params.getfile("ephyspc_params")).exists()) 

533 return 

534 if silent: 

535 data_folder_path = r"D:\iblrig_data\Subjects" 

536 remote_data_folder_path = r"\\iblserver.champalimaud.pt\ibldata\Subjects" 

537 probe_types = {"PROBE_TYPE_00": "3A", "PROBE_TYPE_01": "3B"} 

538 else: 

539 data_folder_path = cli_ask_default( 

540 r"Where's your LOCAL 'Subjects' data folder?", r"D:\iblrig_data\Subjects" 

541 ) 

542 remote_data_folder_path = cli_ask_default( 

543 r"Where's your REMOTE 'Subjects' data folder?", 

544 r"\\iblserver.champalimaud.pt\ibldata\Subjects", 

545 ) 

546 n_probes = int(cli_ask_default("How many probes are you using?", '2')) 

547 assert 100 > n_probes > 0, 'Please enter number between 1, 99 inclusive' 

548 probe_types = {} 

549 for i in range(n_probes): 

550 probe_types[f'PROBE_TYPE_{i:02}'] = cli_ask_options( 

551 f"What's the type of PROBE {i:02}?", ["3A", "3B"]) 

552 param_dict = { 

553 "DATA_FOLDER_PATH": data_folder_path, 

554 "REMOTE_DATA_FOLDER_PATH": remote_data_folder_path, 

555 **probe_types 

556 } 

557 params.write("ephyspc_params", param_dict) 

558 print(f"Created {params.getfile('ephyspc_params')}") 

559 print(param_dict) 

560 return param_dict 

561 

562 

563def rdiff_install() -> bool: (empty)

564 """ 

565 For windows: 

566 * if the rdiff-backup executable does not already exist on the system 

567 * downloads rdiff-backup zip file 

568 * copies the executable to the C:\tools folder 

569 

570 For linux/mac: 

571 * runs a pip install rdiff-backup 

572 

573 Returns: 

574 True when install is successful, False when an error is encountered 

575 """ 

576 if os.name == "nt": 2 ctx1Bi

577 # ensure tools folder exists 

578 tools_folder = "C:\\tools\\" 

579 os.mkdir(tools_folder) if not Path(tools_folder).exists() else None 

580 

581 rdiff_cmd_loc = tools_folder + "rdiff-backup.exe" 

582 if not Path(rdiff_cmd_loc).exists(): 

583 import requests 

584 import zipfile 

585 from io import BytesIO 

586 

587 url = "https://github.com/rdiff-backup/rdiff-backup/releases/download/v2.0.5/rdiff-backup-2.0.5.win32exe.zip" 

588 log.info("Downloading zip file for rdiff-backup.") 

589 # Download the file by sending the request to the URL, ensure success by status code 

590 if requests.get(url).status_code == 200: 

591 log.info("Download complete for rdiff-backup zip file.") 

592 # extracting the zip file contents 

593 zipfile = zipfile.ZipFile(BytesIO(requests.get(url).content)) 

594 zipfile.extractall("C:\\Temp") 

595 rdiff_folder_name = zipfile.namelist()[0] # attempting a bit of future-proofing 

596 # move the executable to the C:\tools folder 

597 shutil.copy("C:\\Temp\\" + rdiff_folder_name + "rdiff-backup.exe", rdiff_cmd_loc) 

598 shutil.rmtree("C:\\Temp\\" + rdiff_folder_name) # cleanup temp folder 

599 try: # attempt to call the rdiff command 

600 subprocess.run([rdiff_cmd_loc, "--version"], check=True) 

601 except (FileNotFoundError, subprocess.CalledProcessError) as e: 

602 log.error("rdiff-backup installation did not complete.\n", e) 

603 return False 

604 return True 

605 else: 

606 log.error("Download request status code not 200, something did not go as expected.") 

607 return False 

608 else: # anything not Windows 

609 try: # package should not be installed via the requirements.txt to accommodate windows 2 ctx1Bi

610 subprocess.run(["pip", "install", "rdiff-backup"], check=True) 2 ctx1Bi

611 except subprocess.CalledProcessError as e: 

612 log.error("rdiff-backup pip install did not complete.\n", e) 

613 return False 

614 return True 2 ctx1Bi

615 

616 

617def get_directory_size(dir_path: Path, in_gb=False) -> float: (empty)

618 """ 

619 Used to determine total size of all files in a given session_path, including all child directories 

620 

621 Args: 

622 dir_path (Path): path we want to get the total size of 

623 in_gb (bool): set to True for returned value to be in gigabytes 

624 

625 Returns: 

626 float: sum of all files in the given directory path (in bytes by default, in GB if specified) 

627 """ 

628 total = 0 2 ctx1ca

629 with iter(os.scandir(dir_path)) as it: 2 ctx1ca

630 for entry in it: 2 ctx1ca

631 if entry.is_file(): 2 ctx1ca

632 total += entry.stat().st_size 2 ctx1ca

633 elif entry.is_dir(): 2 ctx1ca

634 total += get_directory_size(entry.path) 2 ctx1ca

635 if in_gb: 2 ctx1ca

636 return total / 1024 / 1024 / 1024 # in GB 2 ctx1ca

637 return total # in bytes 2 ctx1ca

638 

639 

640def get_session_numbers_from_date_path(date_path: Path) -> list: (empty)

641 """ 

642 Retrieves session numbers when given a date path 

643 

644 Args: 

645 date_path (Path): path to date, i.e. \\\\server\\some_lab\\Subjects\\Date" 

646 

647 Returns: 

648 (list): Found sessions as a sorted list 

649 """ 

650 contents = Path(date_path).glob('*') 6 ctx1cfdage

651 folders = filter(lambda x: x.is_dir() and re.match(r'^\d{3}$', x.name), contents) 6 ctx1cfdage

652 sessions_as_set = set(map(lambda x: x.name, folders)) 6 ctx1cfdage

653 sessions_as_sorted_list = sorted(sessions_as_set) 6 ctx1cfdage

654 return sessions_as_sorted_list 6 ctx1cfdage

655 

656 

657def rsync_paths(src: Path, dst: Path) -> bool: (empty)

658 """ 

659 Used to run the rsync algorithm via a rdiff-backup command on the paths contained on the provided source and destination. 

660 This function relies on the rdiff-backup package and is run from the command line, i.e. subprocess.run(). Full documentation 

661 can be found here - https://rdiff-backup.net/docs/rdiff-backup.1.html 

662 

663 Parameters 

664 ---------- 

665 src : Path 

666 source path that contains data to be transferred 

667 dst : Path 

668 destination path that will receive the transferred data 

669 

670 Returns 

671 ------- 

672 bool 

673 True for success, False for failure 

674 

675 Raises 

676 ------ 

677 FileNotFoundError, subprocess.CalledProcessError 

678 """ 

679 # Set rdiff_cmd_loc based on OS type (assuming C:\tools is not in Windows PATH environ) 

680 rdiff_cmd_loc = "C:\\tools\\rdiff-backup.exe" if os.name == "nt" else "rdiff-backup" 9 ctx1tcimfdage

681 try: # Check if rdiff-backup command is available 9 ctx1tcimfdage

682 subprocess.run([rdiff_cmd_loc, "--version"], check=True) 9 ctx1tcimfdage

683 except (FileNotFoundError, subprocess.CalledProcessError) as e: 1 ctx1i

684 if not rdiff_install(): # Attempt to install rdiff 1 ctx1i

685 log.error("rdiff-backup command is unavailable, transfers can not continue.\n", e) 

686 raise 

687 

688 log.info("Attempting to transfer data: " + str(src) + " -> " + str(dst)) 9 ctx1tcimfdage

689 WindowsInhibitor().inhibit() if os.name == "nt" else None # prevent Windows from going to sleep 9 ctx1tcimfdage

690 try: 9 ctx1tcimfdage

691 rsync_command = [rdiff_cmd_loc, "--verbosity", str(0), 9 ctx1tcimfdage

692 "--create-full-path", "--backup-mode", "--no-acls", "--no-eas", 

693 "--no-file-statistics", "--exclude", "**transfer_me.flag", 

694 str(src), str(dst)] 

695 subprocess.run(rsync_command, check=True) 9 ctx1tcimfdage

696 time.sleep(1) # give rdiff-backup a second to complete all logging operations 9 ctx1tcimfdage

697 except (FileNotFoundError, subprocess.CalledProcessError) as e: 

698 log.error("Transfer failed with code %i.\n", e.returncode) 

699 if e.stderr: 

700 log.error(e.stderr) 

701 return False 

702 log.info("Validating transfer completed...") 9 ctx1tcimfdage

703 try: # Validate the transfers succeeded 9 ctx1tcimfdage

704 rsync_validate = [rdiff_cmd_loc, "--verify", str(dst)] 9 ctx1tcimfdage

705 subprocess.run(rsync_validate, check=True) 9 ctx1tcimfdage

706 except (FileNotFoundError, subprocess.CalledProcessError) as e: 

707 log.error(f"Validation for destination {dst} failed.\n", e) 

708 return False 

709 log.info("Cleaning up rdiff files...") 9 ctx1tcimfdage

710 shutil.rmtree(dst / "rdiff-backup-data") 9 ctx1tcimfdage

711 WindowsInhibitor().uninhibit() if os.name == 'nt' else None # allow Windows to go to sleep 9 ctx1tcimfdage

712 return True 9 ctx1tcimfdage

713 

714 

715def confirm_ephys_remote_folder(local_folder=False, remote_folder=False, force=False, iblscripts_folder=False, (empty)

716 session_path=None): 

717 """ 

718 :param local_folder: The full path to the local Subjects folder 

719 :param remote_folder: the full path to the remote Subjects folder 

720 :param force: 

721 :param iblscripts_folder: 

722 :return: 

723 """ 

724 # FIXME: session_path can be relative 

725 pars = load_ephyspc_params() 

726 if not iblscripts_folder: 

727 import deploy 

728 iblscripts_folder = Path(deploy.__file__).parent.parent 

729 if not local_folder: 

730 local_folder = pars["DATA_FOLDER_PATH"] 

731 if not remote_folder: 

732 remote_folder = pars["REMOTE_DATA_FOLDER_PATH"] 

733 local_folder = Path(local_folder) 

734 remote_folder = Path(remote_folder) 

735 # Check for Subjects folder 

736 local_folder = subjects_data_folder(local_folder, rglob=True) 

737 remote_folder = subjects_data_folder(remote_folder, rglob=True) 

738 

739 log.info(f"local folder: {local_folder}") 

740 log.info(f"remote folder: {remote_folder}") 

741 if session_path is None: 

742 src_session_paths = [x.parent for x in local_folder.rglob("transfer_me.flag")] 

743 else: 

744 src_session_paths = session_path if isinstance(session_path, list) else [session_path] 

745 

746 if not src_session_paths: 

747 log.info("Nothing to transfer, exiting...") 

748 return 

749 for session_path in src_session_paths: 

750 log.info(f"Found : {session_path}") 

751 log.info(f"Found: {len(src_session_paths)} sessions to transfer, starting transferring now") 

752 

753 for session_path in src_session_paths: 

754 log.info(f"Transferring session: {session_path}") 

755 # Rename ephys files 

756 # FIXME: if transfer has failed and wiring file is there renaming will fail! 

757 rename_ephys_files(str(session_path)) 

758 # Move ephys files 

759 move_ephys_files(str(session_path)) 

760 # Copy wiring files 

761 copy_wiring_files(str(session_path), iblscripts_folder) 

762 try: 

763 create_alyx_probe_insertions(str(session_path)) 

764 except BaseException: 

765 log.error(traceback.print_exc()) 

766 log.info("Probe creation failed, please create the probe insertions manually. Continuing transfer...") 

767 msg = f"Transfer {session_path }to {remote_folder} with the same name?" 

768 resp = input(msg + "\n[y]es/[r]ename/[s]kip/[e]xit\n ^\n> ") or "y" 

769 resp = resp.lower()