diff --git a/lochness/__init__.py b/lochness/__init__.py index 2f905d01..98c8a283 100644 --- a/lochness/__init__.py +++ b/lochness/__init__.py @@ -394,7 +394,8 @@ def attempt(f, Lochness, *args, **kwargs): ''' try: - f(Lochness, *args, **kwargs) + output = f(Lochness, *args, **kwargs) + return output except Exception as e: logger.warn(e) logger.debug(tb.format_exc().strip()) diff --git a/lochness/net/__init__.py b/lochness/net/__init__.py index 1875b9bc..7eee62f0 100644 --- a/lochness/net/__init__.py +++ b/lochness/net/__init__.py @@ -13,7 +13,7 @@ def wrapped_f(*args, **kwargs): attempt = 1 while True: try: - f(*args, **kwargs) + return f(*args, **kwargs) break except requests.exceptions.ConnectionError as e: attempt += 1 diff --git a/lochness/redcap/__init__.py b/lochness/redcap/__init__.py index 737adb4b..82b6c065 100644 --- a/lochness/redcap/__init__.py +++ b/lochness/redcap/__init__.py @@ -176,10 +176,10 @@ def initialize_metadata(Lochness: 'Lochness object', # Redcap default information subject_dict['REDCap'] = \ f'redcap.{project_name}:{subject_id}' + # subject_dict['REDCap'] += \ + # f';redcap.UPENN:{row[redcap_id_colname]}' # UPENN REDCAP subject_dict['REDCap'] += \ - f';redcap.UPENN:{row[redcap_id_colname]}' # UPENN REDCAP - subject_dict['REDCap'] += \ - f';redcap.UPENN_new:{row[redcap_id_colname]}' # UPENN REDCAP + f';redcap.UPENN_nda:{row[redcap_id_colname]}' # UPENN REDCAP subject_dict['Box'] = f'box.{study_name}:{subject_id}' subject_dict['XNAT'] = f'xnat.{study_name}:*:{subject_id}' @@ -333,8 +333,8 @@ def initialize_metadata_rm(Lochness: 'Lochness object', # Redcap default information subject_dict['REDCap'] = \ f'redcap.{project_name}:{subject_id}' - subject_dict['REDCap'] += \ - f';redcap.UPENN:{subject_id}' # UPENN REDCAP + # subject_dict['REDCap'] += \ + # f';redcap.UPENN:{subject_id}' # UPENN REDCAP subject_dict['REDCap'] += \ f';redcap.UPENN_new:{subject_id}' # UPENN REDCAP subject_dict['Box'] = f'box.{study_name}:{subject_id}' @@ -683,9 +683,11 @@ def sync(Lochness, subject, dry=False): _debug_tup = (redcap_instance, redcap_project, redcap_subject) if 'UPENN' in redcap_instance: + logger.debug(f"redcap_instance: {redcap_instance}") if redcap_instance == 'redcap.UPENN': upenn_id_colname = 'session_subid' - elif redcap_instance == 'redcap.UPENN_new': + continue # skip downloading from the original UPENN redcap + elif redcap_instance == 'redcap.UPENN_nda': upenn_id_colname = 'src_subject_id' else: logger.warning('Wrong upenn_id_colname. Check REDCap code') @@ -901,8 +903,9 @@ def post_to_redcap(api_url, data, debug_tup): # verify response content integrity if 'content-length' not in r.headers: - logger.warn('server did not return a content-length header, ' - f'can\'t verify response integrity for {debug_tup}') + logger.warn( + f'server ({api_url}) did not return a content-length ' + f'header, can\'t verify response integrity for {debug_tup}') else: expected_len = int(r.headers['content-length']) if content_len != expected_len: diff --git a/lochness/transfer/__init__.py b/lochness/transfer/__init__.py index ce8967ba..9e4eb737 100644 --- a/lochness/transfer/__init__.py +++ b/lochness/transfer/__init__.py @@ -349,7 +349,12 @@ def lochness_to_lochness_transfer_s3(Lochness, command = f"aws s3 cp \ {metadata_file} \ s3://{s3_bucket_name}/{s3_phoenix_metadata}" - os.popen(command).read() + result = subprocess.run( + command, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True) logger.debug('Syncing metadata files completed') logger.debug('Syncing source files') @@ -373,26 +378,7 @@ def lochness_to_lochness_transfer_s3(Lochness, sites): continue - s3_phoenix_root_dtype = re.sub(Lochness['phoenix_root'], - s3_phoenix_root, - str(source_directory)) - - command = f'aws s3 sync \ - {source_directory}/ \ - s3://{s3_bucket_name}/{s3_phoenix_root_dtype}' - - # save aws 3 sync cmd stdout to a file - s3_sync_stdout = Path(Lochness['phoenix_root']) / 'aws_s3_sync_stdouts.log' - now = datetime.now() - current_time = now.strftime("%Y-%m-%d %H:%M:%S") - with open(s3_sync_stdout, 'a') as fp: - command_str = '\n'.join([f'{current_time} {x}' for x in - os.popen(command).read().split('\n') - if 'upload' in x]) + '\n' - fp.write(command_str) - - logger.debug(command_str) - logger.debug('aws rsync completed') + lochness_s3_dir(Lochness, source_directory) # sync redcap dictionary redcap_dict = Path(Lochness['phoenix_root']) / 'GENERAL/redcap_metadata.csv' @@ -445,7 +431,9 @@ def create_s3_transfer_table(Lochness, rewrite=False) -> None: df = pd.DataFrame() with open(log_file, 'r') as fp: - for line in fp.readlines(): + text = fp.read() + + for line in text.splitlines(): if not 'upload' in line: continue @@ -526,6 +514,70 @@ def get_ctime_or_nan(x): df.to_csv(out_file) +def lochness_s3_dir(Lochness, + source_directory, + interview_run_sheets_only=False): + '''Lochness aws s3 sync for protected data + + Key arguments: + Lochness: Lochness config.load object + dir_path: Path of the directory to be synced + + Requirements: + - AWS CLI needs to be set with the correct credentials before executing + this module. + $ aws configure + - s3 bucket needs to be linked to the ID + - The name of the s3 bucket needs to be in the config.yml + eg) AWS_BUCKET_NAME: ampscz-dev + AWS_BUCKET_PHOENIX_ROOT: TEST_PHOENIX_ROOT + + Notes: + - do not share .mp3 files from phone data + ''' + s3_bucket_name = Lochness['AWS_BUCKET_NAME'] + s3_phoenix_root = Lochness['AWS_BUCKET_ROOT'] + + # save aws 3 sync cmd stdout to a file + s3_sync_stdout = Path(Lochness['phoenix_root']) / 'aws_s3_sync_stdouts.log' + + s3_phoenix_root_dtype = re.sub(Lochness['phoenix_root'], + s3_phoenix_root, + str(source_directory)) + + if interview_run_sheets_only: + command = f"aws s3 sync \ + {interview_dir} \ + s3://{s3_bucket_name}/{s3_target} \ + --follow-symlinks \ + --exclude='*' --include='*Run_sheet_interviews_*.csv'" + else: + command = f"aws s3 sync \ + {source_directory}/ \ + s3://{s3_bucket_name}/{s3_phoenix_root_dtype} \ + --follow-symlinks \ + --exclude '*.mp3' --exclude '.check_sum*'" + command = re.sub(r'\s+', r' ', command) + logger.debug(f's3 command: {command}') + + + now = datetime.now() + current_time = now.strftime("%Y-%m-%d %H:%M:%S") + with open(s3_sync_stdout, 'a') as fp: + result = subprocess.run( + command, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True) + if 'upload' in str(result.stdout): + output_str = '\n'.join([f'{current_time} {x}' for x in + result.stdout.split('\n') + if 'upload' in x]) + '\n' + fp.write(output_str) + logger.debug(f'aws rsync completed: {source_directory}') + + def lochness_to_lochness_transfer_s3_protected(Lochness, sites: List[str], sources: List[str]): @@ -572,65 +624,29 @@ def lochness_to_lochness_transfer_s3_protected(Lochness, Lochness['phoenix_root'], sites): pass - #continue - - s3_phoenix_root_dtype = re.sub(Lochness['phoenix_root'], - s3_phoenix_root, - str(source_directory)) - command = f"aws s3 sync \ - {source_directory}/ \ - s3://{s3_bucket_name}/{s3_phoenix_root_dtype} \ - --exclude '*.mp3' --exclude '.check_sum*'" - - logger.debug(re.sub(r'\s+', r' ', command)) - - now = datetime.now() - current_time = now.strftime("%Y-%m-%d %H:%M:%S") - with open(s3_sync_stdout, 'a') as fp: - command_str = '\n'.join( - [f'{current_time} {x}' for x in - os.popen(command).read().split('\n') - if 'upload' in x]) + '\n' - fp.write(command_str) + lochness_s3_dir(Lochness, source_directory) + logger.debug(f'aws rsync completed: {source_directory}') logger.debug(f'aws rsync completed "{datatype}" datatype') # interview run sheets # phoenix_root / PROTECTED / site / raw / subject / datatype + if not is_datatype_in_sources('interviews', sources): + return + interview_dirs = Path(Lochness['phoenix_root']).glob( f'PROTECTED/*/raw/*/interviews') - if not is_datatype_in_sources('interviews', sources): - return for interview_dir in interview_dirs: if not is_phoenix_path_from_sitelist(interview_dir, Lochness['phoenix_root'], sites): continue - - s3_target = re.sub(Lochness['phoenix_root'], - s3_phoenix_root, - str(interview_dir)) - - - command = f"aws s3 sync \ - {interview_dir} \ - s3://{s3_bucket_name}/{s3_target} \ - --exclude='*' --include='*Run_sheet_interviews_*.csv'" - - logger.debug(re.sub(r'\s+', r' ', command)) - - now = datetime.now() - current_time = now.strftime("%Y-%m-%d %H:%M:%S") - with open(s3_sync_stdout, 'a') as fp: - command_str = '\n'.join( - [f'{current_time} {x}' for x in - os.popen(command).read().split('\n') - if 'upload' in x]) + '\n' - fp.write(command_str) - + lochness_s3_dir(Lochness, + str(interview_dir), + interview_run_sheets_only=True) logger.debug(f'aws rsync completed for interview run sheets') diff --git a/lochness/xnat/__init__.py b/lochness/xnat/__init__.py index e94fe15f..04663522 100644 --- a/lochness/xnat/__init__.py +++ b/lochness/xnat/__init__.py @@ -15,6 +15,7 @@ import lochness.tree as tree import lochness.config as config from lochness.cleaner import is_transferred_and_removed +from lochness.transfer import lochness_s3_dir yaml.SafeDumper.add_representer( col.OrderedDict, yaml.representer.SafeRepresenter.represent_dict) @@ -210,7 +211,7 @@ def clear_stored_credentials(dataorc_bindary_path: Path): command = ' '.join(command_array) logger.info(f'Executing command: {command}') - child = pexpect.spawn(command) + child = pexpect.spawn(command, use_poll=True) child.expect("Please enter your username:", timeout=10) child.sendline(xnat_username) child.expect("Please enter your password:", timeout=10) @@ -219,7 +220,7 @@ def clear_stored_credentials(dataorc_bindary_path: Path): @net.retry(max_attempts=5) -def sync_xnatpy(Lochness, subject, dry=False): +def sync_xnatpy(Lochness, subject, dry=False, args=None): """A new sync function with XNATpy""" logger.debug('exploring {0}/{1}'.format(subject.study, subject.id)) @@ -228,13 +229,17 @@ def sync_xnatpy(Lochness, subject, dry=False): # remove xnatpy tmp files logger.debug('Cleaning up xnatpy files') for tmp_file in Path(tmp_dir).glob('*generated_xnat.py'): - os.remove(tmp_file) + try: + os.remove(tmp_file) + except OSError as e: + logger.warning(f'Failed to remove {tmp_file}: {e}') for tmp_file in Path(tmp_dir).glob('tmp_xnat*'): try: os.rmdir(tmp_file) except OSError as e: logger.warning(f'Failed to remove {tmp_file}: {e}') + logger.debug('Cleaning up xnatpy files - completed') for alias, xnat_uids in iter(subject.xnat.items()): @@ -283,6 +288,7 @@ def sync_xnatpy(Lochness, subject, dry=False): if os.path.exists(dst): logger.debug('Already downloaded') + lochness_s3_dir(Lochness, Path(dst).parent) continue message = 'downloading {PROJECT}/{LABEL} to {FOLDER}' @@ -317,7 +323,29 @@ def sync_xnatpy(Lochness, subject, dry=False): shutil.move(downloaded_file_path, dst) os.chmod(dst, 0o0755) - kill_process_by_name('dataorc') + if args.lochness_sync_send and args.s3: + if 'mri' in Lochness['s3_selective_sync']: + logger.debug(f'Syncing {dst}') + lochness_s3_dir(Lochness, Path(dst).parent) + save_last_downloaded_subject(Lochness, subject.id) + return 'completed' + + +def save_last_downloaded_subject(Lochness, subject_label:str): + log_file = Path(Lochness['phoenix_root']).parent / '.last_xnat_sync_id' + with open(log_file, 'w') as fp: + fp.write(f"{subject_label}") + + +def load_last_downloaded_subject(Lochness) -> tuple: + log_file = Path(Lochness['phoenix_root']).parent / '.last_xnat_sync_id' + if not log_file.is_file(): + return '' + + with open(log_file, 'r') as fp: + subject_label = fp.readline().strip() + + return subject_label def check_consistency(d, experiment): diff --git a/scripts/sync.py b/scripts/sync.py index 02504c30..670a8790 100755 --- a/scripts/sync.py +++ b/scripts/sync.py @@ -30,6 +30,7 @@ from lochness.transfer import lochness_to_lochness_transfer_s3_protected from lochness.transfer import create_s3_transfer_table from lochness.transfer import lochness_to_lochness_transfer_receive_sftp +from lochness.xnat import load_last_downloaded_subject from lochness.email import send_out_daily_updates from datetime import datetime, date from lochness.cleaner import rm_transferred_files_under_phoenix @@ -279,8 +280,13 @@ def do(args, Lochness): lochness.initialize_metadata(Lochness, args, multiple_site, upenn_redcap) + n = 0 + last_subject_synced = load_last_downloaded_subject(Lochness) + keep_loop = True + output = None for subject in lochness.read_phoenix_metadata(Lochness, args.studies): + if n == 0: save_redcap_metadata(Lochness, subject) @@ -300,15 +306,26 @@ def do(args, Lochness): else: for source, Module in zip(args.input_sources, args.source): if source == 'xnat': - lochness.attempt(Module.sync_xnatpy, Lochness, - subject, dry=args.dry) + # if last_subject_synced == '': + # keep_loop = True + # elif subject.id == last_subject_synced: + # logger.debug(f'Last subject downloaded: {subject.id}') + # keep_loop = True + # continue # skip last synced + # elif not keep_loop: + if not keep_loop: + continue # skip subjects before last synced + + output = lochness.attempt(Module.sync_xnatpy, Lochness, + subject, dry=args.dry, args=args) + if output is not None: # expecting 'completed' + keep_loop = False + else: lochness.attempt(Module.sync, Lochness, subject, dry=args.dry) n += 1 - # anonymize PII - #if Lochness['s3_selective_sync']: # dpanonymize.lock_lochness( # Lochness, @@ -344,6 +361,9 @@ def do(args, Lochness): logger.info('A round of sync completed') + if 'xnat' == source: + return + if __name__ == '__main__': main()