Coverage for ibllib/pipes/transfer_rig_data.py: 89%

1#!/usr/bin/env python 

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

3# @Author: Niccolò Bonacchi 

4# @Date: Wednesday, January 16th 2019, 2:03:59 pm 

5import argparse (empty)

6import logging (empty)

7import shutil (empty)

8from pathlib import Path (empty)

9from shutil import ignore_patterns as ig (empty)

10 

11import ibllib.io.extractors.base (empty)

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

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

14 

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

16log.setLevel(logging.INFO) (empty)

17 

18 

19def main(local_folder: str, remote_folder: str, force: bool = False) -> None: (empty)

20 local_folder = Path(local_folder) 3 ctx1cba

21 remote_folder = Path(remote_folder) 3 ctx1cba

22 

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

24 

25 if not src_session_paths: 3 ctx1cba

26 log.info("Nothing to transfer, exiting...") 2 ctx1ba

27 return 2 ctx1ba

28 

29 # Create all dst paths 

30 dst_session_paths = [] 3 ctx1cba

31 for s in src_session_paths: 3 ctx1cba

32 mouse = s.parts[-3] 3 ctx1cba

33 date = s.parts[-2] 3 ctx1cba

34 sess = s.parts[-1] 3 ctx1cba

35 d = remote_folder / mouse / date / sess 3 ctx1cba

36 dst_session_paths.append(d) 3 ctx1cba

37 

38 for src, dst in zip(src_session_paths, dst_session_paths): 3 ctx1cba

39 src_flag_file = src / "transfer_me.flag" 3 ctx1cba

40 flag = flags.read_flag_file(src_flag_file) 3 ctx1cba

41 if isinstance(flag, list): 3 ctx1cba

42 raise NotImplementedError 

43 else: 

44 if force: 3 ctx1cba

45 shutil.rmtree(dst, ignore_errors=True) 2 ctx1ba

46 log.info(f"Copying {src}...") 3 ctx1cba

47 shutil.copytree(src, dst, ignore=ig(str(src_flag_file.name))) 3 ctx1cba

48 # finally if folder was created delete the src flag_file and create compress_me.flag 

49 if dst.exists(): 3 ctx1cba

50 task_type = ibllib.io.extractors.base.get_session_extractor_type(Path(src)) 3 ctx1cba

51 if task_type not in ['ephys', 'ephys_sync', 'ephys_mock']: 3 ctx1cba

52 flags.write_flag_file(dst.joinpath('raw_session.flag')) 2 ctx1ba

53 settings = raw.load_settings(dst) 2 ctx1ba

54 if 'ephys' in settings['PYBPOD_BOARD']: # Any traing task on an ephys rig 2 ctx1ba

55 dst.joinpath('raw_session.flag').unlink() 1 ctx1a

56 log.info(f"Copied to {remote_folder}: Session {src_flag_file.parent}") 3 ctx1cba

57 src_flag_file.unlink() 3 ctx1cba

58 

59 # Cleanup 

60 src_video_file = src / 'raw_video_data' / '_iblrig_leftCamera.raw.avi' 3 ctx1cba

61 dst_video_file = dst / 'raw_video_data' / '_iblrig_leftCamera.raw.avi' 3 ctx1cba

62 src_audio_file = src / 'raw_behavior_data' / '_iblrig_micData.raw.wav' 3 ctx1cba

63 dst_audio_file = dst / 'raw_behavior_data' / '_iblrig_micData.raw.wav' 3 ctx1cba

64 

65 if src_audio_file.exists() and \ 3 ctx1cba

66 src_audio_file.stat().st_size == dst_audio_file.stat().st_size: 

67 src_audio_file.unlink() 3 ctx1cba

68 

69 if src_video_file.exists() and \ 3 ctx1cba

70 src_video_file.stat().st_size == dst_video_file.stat().st_size: 

71 src_video_file.unlink() 3 ctx1cba

72 

73 

74if __name__ == "__main__": (empty)

75 parser = argparse.ArgumentParser( 

76 description='Transfer files to IBL local server') 

77 parser.add_argument( 

78 'local_folder', help='Local iblrig_data/Subjects folder') 

79 parser.add_argument( 

80 'remote_folder', help='Remote iblrig_data/Subjects folder') 

81 args = parser.parse_args() 

82 main(args.local_folder, args.remote_folder)