"""
Purge data from acquisition PC.
Steps:
- Find all files by rglob
- Find all sessions of the found files
- Check Alyx if corresponding datasetTypes have been registered as existing
sessions and files on Flatiron
- Delete local raw file if found on Flatiron
"""
import argparse
from pathlib import Path
from one.api import ONE
from one.alf.files import get_session_path
[docs]
def session_name(path) -> str:
"""Returns the session name (subject/date/number) string for any filepath
using session_path"""
return '/'.join(get_session_path(path).parts[-3:])
[docs]
def purge_local_data(local_folder, file_name, lab=None, dry=False):
# Figure out datasetType from file_name or file path
file_name = Path(file_name).name
alf_parts = file_name.split('.')
dstype = '.'.join(alf_parts[:2])
print(f'Looking for file <{file_name}> in folder <{local_folder}>')
# Get all paths for file_name in local folder
local_folder = Path(local_folder)
files = list(local_folder.rglob(f'*{file_name}'))
print(f'Found {len(files)} files')
print(f'Checking on Flatiron for datsetType: {dstype}...')
# Get all sessions and details from Alyx that have the dstype
one = ONE(cache_rest=None)
if lab is None:
eid, det = one.search(dataset_types=[dstype], details=True)
else:
eid, det = one.search(dataset_types=[dstype], lab=lab, details=True)
urls = []
for d in det:
urls.extend([x['data_url'] for x in d['data_dataset_session_related']
if x['dataset_type'] == dstype])
# Remove None answers when session is registered but dstype not htere yet
urls = [u for u in urls if u is not None]
print(f'Found files on Flatiron: {len(urls)}')
to_remove = []
for f in files:
sess_name = session_name(f)
for u in urls:
if sess_name in u:
to_remove.append(f)
print(f'Local files to remove: {len(to_remove)}')
for f in to_remove:
print(f)
if dry:
continue
else:
f.unlink()
return
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Delete files from rig')
parser.add_argument('folder', help='Local iblrig_data folder')
parser.add_argument(
'file', help='File name to search and destroy for every session')
parser.add_argument('-lab', required=False, default=None,
help='Lab name, search on Alyx faster. default: None')
parser.add_argument('--dry', required=False, default=False,
action='store_true', help='Dry run? default: False')
args = parser.parse_args()
purge_local_data(args.folder, args.file, lab=args.lab, dry=args.dry)
print('Done\n')