#!/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