#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @Author: Niccolò Bonacchi
# @Date: Thursday, September 24th 2020, 2:10:29 pm
import json
import logging
import shutil
from pathlib import Path, PureWindowsPath
from one.alf.path import get_session_path
log = logging.getLogger('ibllib')
[docs]
def load_settings_file(filepath, key=None):
filepath = Path(filepath)
if filepath.stat().st_size == 0:
return
with open(filepath, 'r') as f:
settings = json.load(f)
return settings.get(key, None) if key else settings
[docs]
def find_pairs(root_data_folder):
"""Find all passive sessions that needs transfer and where to"""
root_data_folder = Path(root_data_folder)
settings_files = list(root_data_folder.rglob('_iblrig_taskSettings.raw.json'))
n_settings_files = len(settings_files)
if n_settings_files == 0:
log.warning(f'Found {n_settings_files} sessions')
else:
log.info(f'Found {n_settings_files} sessions')
# Load the corresponding ephys session path form settings file if exists
pairs = []
for sf in settings_files:
# Get session path form settings file
source_spath = get_session_path(sf)
if source_spath is None:
continue
# Find the root_data_path for session
subjects_folder_path = Path(*Path(source_spath).parts[:-3])
# Load reference to corresponding ephys session (ces) which comes in windows format
ces = load_settings_file(sf, key='CORRESPONDING_EPHYS_SESSION')
# if CORRESPONDING_EPHYS_SESSION does not exist, it's not a passive session
if ces is None:
continue
# Convert windows path to corresponding session name (csn) in native Path format
csn = Path(*PureWindowsPath(ces).parts[-3:])
target_spath = subjects_folder_path / csn
pairs.append((source_spath, target_spath))
# Remove sessions that are already transferred i.e. source and destination files are equal
from_to_pairs = [(x, y) for x, y in pairs if x != y]
n_pairs = len(from_to_pairs)
if n_pairs == 0:
log.warning(f'Found {n_pairs} passive sessions to move')
else:
log.info(f'Found {n_pairs} passive sessions to move')
return from_to_pairs
[docs]
def move_rename_pairs(from_to_pairs):
""""""
moved_ok = []
for i, (src, dst) in enumerate(from_to_pairs):
src = Path(src)
dst = Path(dst)
log.info(f'Moving {i + 1} of {len(from_to_pairs)}: \n{src}\n--> {dst}')
try:
shutil.move(str(src / 'raw_behavior_data'), str(dst / 'raw_passive_data'))
ffile = src.joinpath('passive_data_for_ephys.flag')
if ffile.exists():
ffile.unlink()
ffile.parent.rmdir()
moved_ok.append(True)
except BaseException as e:
log.error(f'Failed to move {src} to {dst}:\n {e}')
moved_ok.append(False)
continue
log.info(f'Moved {sum(moved_ok)} of {len(from_to_pairs)}')
return moved_ok
[docs]
def execute(root_data_folder, dry=True):
from_to_pairs = find_pairs(root_data_folder)
if dry:
return from_to_pairs, [False] * len(from_to_pairs)
moved_ok = move_rename_pairs(from_to_pairs)
return from_to_pairs, moved_ok