feat: cleaning up s3 sync log and adding a sync_data_to_aws.py

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())

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

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:

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')
 
 

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):