Coverage for ibllib/pipes/scan_fix_passive_files.py: 97%

1#!/usr/bin/env python 

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

3# @Author: Niccolò Bonacchi 

4# @Date: Thursday, September 24th 2020, 2:10:29 pm 

5import json (empty)

6import logging (empty)

7import shutil (empty)

8from pathlib import Path, PureWindowsPath (empty)

9 

10from one.alf.path import get_session_path (empty)

11 

12log = logging.getLogger("ibllib") (empty)

13 

14 

15def load_settings_file(filepath, key=None): (empty)

16 filepath = Path(filepath) 3 ctx1cba

17 if filepath.stat().st_size == 0: 3 ctx1cba

18 return 3 ctx1cba

19 

20 with open(filepath, "r") as f: 3 ctx1cba

21 settings = json.load(f) 3 ctx1cba

22 return settings.get(key, None) if key else settings 3 ctx1cba

23 

24 

25def find_pairs(root_data_folder): (empty)

26 """Find all passive sessions that needs transfer and where to""" 

27 root_data_folder = Path(root_data_folder) 3 ctx1cba

28 settings_files = list(root_data_folder.rglob("_iblrig_taskSettings.raw.json")) 3 ctx1cba

29 n_settings_files = len(settings_files) 3 ctx1cba

30 if n_settings_files == 0: 3 ctx1cba

31 log.warning(f"Found {n_settings_files} sessions") 

32 else: 

33 log.info(f"Found {n_settings_files} sessions") 3 ctx1cba

34 

35 # Load the corresponding ephys session path form settings file if exists 

36 pairs = [] 3 ctx1cba

37 for sf in settings_files: 3 ctx1cba

38 # Get session path form settings file 

39 source_spath = get_session_path(sf) 3 ctx1cba

40 if source_spath is None: 3 ctx1cba

41 continue 

42 # Find the root_data_path for session 

43 subjects_folder_path = Path(*Path(source_spath).parts[:-3]) 3 ctx1cba

44 # Load reference to corresponding ephys session (ces) which comes in windows format 

45 ces = load_settings_file(sf, key="CORRESPONDING_EPHYS_SESSION") 3 ctx1cba

46 # if CORRESPONDING_EPHYS_SESSION does not exist, it's not a passive session 

47 if ces is None: 3 ctx1cba

48 continue 3 ctx1cba

49 # Convert windows path to corresponding session name (csn) in native Path format 

50 csn = Path(*PureWindowsPath(ces).parts[-3:]) 3 ctx1cba

51 target_spath = subjects_folder_path / csn 3 ctx1cba

52 

53 pairs.append((source_spath, target_spath)) 3 ctx1cba

54 # Remove sessions that are already transferred i.e. source and destination files are equal 

55 from_to_pairs = [(x, y) for x, y in pairs if x != y] 3 ctx1cba

56 n_pairs = len(from_to_pairs) 3 ctx1cba

57 if n_pairs == 0: 3 ctx1cba

58 log.warning(f"Found {n_pairs} passive sessions to move") 1 ctx1a

59 else: 

60 log.info(f"Found {n_pairs} passive sessions to move") 3 ctx1cba

61 

62 return from_to_pairs 3 ctx1cba

63 

64 

65def move_rename_pairs(from_to_pairs): (empty)

66 """""" 

67 moved_ok = [] 2 ctx1ba

68 for i, (src, dst) in enumerate(from_to_pairs): 2 ctx1ba

69 src = Path(src) 2 ctx1ba

70 dst = Path(dst) 2 ctx1ba

71 log.info(f"Moving {i + 1} of {len(from_to_pairs)}: \n{src}\n--> {dst}") 2 ctx1ba

72 try: 2 ctx1ba

73 shutil.move(str(src / "raw_behavior_data"), str(dst / "raw_passive_data")) 2 ctx1ba

74 ffile = src.joinpath("passive_data_for_ephys.flag") 2 ctx1ba

75 if ffile.exists(): 2 ctx1ba

76 ffile.unlink() 2 ctx1ba

77 ffile.parent.rmdir() 2 ctx1ba

78 moved_ok.append(True) 2 ctx1ba

79 except BaseException as e: 1 ctx1b

80 log.error(f"Failed to move {src} to {dst}:\n {e}") 1 ctx1b

81 moved_ok.append(False) 1 ctx1b

82 continue 1 ctx1b

83 log.info(f"Moved {sum(moved_ok)} of {len(from_to_pairs)}") 2 ctx1ba

84 return moved_ok 2 ctx1ba

85 

86 

87def execute(root_data_folder, dry=True): (empty)

88 from_to_pairs = find_pairs(root_data_folder) 1 ctx1a

89 if dry: 1 ctx1a

90 return from_to_pairs, [False] * len(from_to_pairs) 1 ctx1a

91 moved_ok = move_rename_pairs(from_to_pairs) 1 ctx1a

92 return from_to_pairs, moved_ok 1 ctx1a