Source code for ibllib.pipes.scan_fix_passive_files

#!/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.files 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:"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:"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)"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"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