Merge pull request #49 from CBIIT/CRDCDH-2169-001-pgu

CRDCDH 2169

Author: n2iw

requirements.txt

@@ -1,4 +1,5 @@
 pyyaml
 boto3
 requests
-requests_aws4auth
+requests_aws4auth
+tqdm

src/common/graphql_client.py

@@ -166,7 +166,9 @@ def get_data_file_config(self, submissionID):
             if status == 200:
                 results = response.json()
                 if results.get("errors"):
-                    self.log.error(f'Get data file config failed: {results.get("errors")[0].get("message")}.')
+                    msg = f'Get data file config failed: {results.get("errors")[0].get("message")}.'
+                    self.log.error(msg)
+                    return False, None
                 else:
                     data_file_config = results.get("data").get("retrieveFileNodeConfig")
                     if data_file_config:

src/common/s3util.py

@@ -1,5 +1,8 @@
 #!/usr/bin/env python
+import os
 import boto3
+from boto3.s3.transfer import TransferConfig, S3Transfer
+from tqdm import tqdm  # For progress bar
 from botocore.exceptions import ClientError
 
 from bento.common.utils import get_logger
@@ -55,14 +58,29 @@ def file_exists_on_s3(self, key):
                 self.log.exception(e)
                 return False, msg  
 
-    def put_file_obj(self, key, data, md5_base64):
-        return self.bucket.put_object(Key=key,
-                                      Body=data,
-                                      ContentMD5=md5_base64,
-                                      ACL= BUCKET_OWNER_ACL)
+    def put_file_obj(self, file_size, key, data, md5_base64):
+        # Initialize the progress bar
+        progress = create_progress_bar(file_size)
+        chunk_size = 1024 * 1024 if file_size >= 1024 * 1024 else file_size #chunk data for display progress for small metadata file < 4,500,000,000 bytes
+        try:
+            # Upload the file in chunks
+            for chunk in iter(lambda: data.read(chunk_size), b''):
+                self.bucket.put_object(Key=key,
+                                        Body=chunk,
+                                        ContentMD5=md5_base64,
+                                        ACL= BUCKET_OWNER_ACL,
+                                        )
+                # Update the progress bar
+                progress.update(len(chunk))
+        finally:
+            # Close the progress bar
+            progress.close()
 
-    def upload_file_obj(self, key, data, config=None, extra_args={'ACL': BUCKET_OWNER_ACL}):
-        return self.bucket.upload_fileobj(data, key, ExtraArgs=extra_args, Config=config)
+    def upload_file_obj(self, file_size, key, data, config=None, extra_args={'ACL': BUCKET_OWNER_ACL}):
+        self.bucket.upload_fileobj(
+            data, key, ExtraArgs=extra_args, Config=config,
+            Callback=ProgressPercentage(file_size)
+        )
 
     def get_object_size(self, key):
         try:
@@ -90,7 +108,9 @@ def same_size_file_exists(self, key, file_size):
 
     def download_object(self, key, local_file_path):
         try:
-            self.bucket.download_file( key, local_file_path)
+            file_size, msg = self.get_object_size(key)
+            self.bucket.download_file(key, local_file_path,
+                                      Callback=ProgressPercentage(file_size))
             return True, None
         except ClientError as ce:
             msg = None
@@ -115,5 +135,26 @@ def close(self):
         self.client = None
         self.bucket = None
         self.s3 = None
+
+"""
+S3 util class to display upload progress
+"""
+class ProgressPercentage(object):
+    def __init__(self, file_size):
+        self._size = file_size
+        self._seen_so_far = 0
+        self._progress = create_progress_bar(file_size)
+
+    def __call__(self, bytes_amount):
+        self._seen_so_far += bytes_amount
+        self._progress.update(bytes_amount)
+
+    def __del__(self):
+        self._progress.close() 
+
+def create_progress_bar(file_size):
+    progress_bar = tqdm(total= file_size, unit='B', unit_scale=True, desc="Progress", smoothing=0.0,
+                              bar_format="{l_bar}\033[1;32m{bar}\033[0m| {n_fmt}/{total_fmt} [elapsed: {elapsed} | remaining: {remaining}, {rate_fmt}]")
+    return progress_bar
 
 

src/common/utils.py

@@ -4,6 +4,7 @@
 from uuid import UUID
 from datetime import datetime
 from common.constants import S3_START
+import os
 
 """ 
 clean_up_key_value(dict)
@@ -94,4 +95,54 @@ def extract_s3_info_from_url(url):
     prefix = "/".join(split_list[1:])
     return bucket_name, prefix
 
+def format_size(size_in_bytes):
+    """
+    Convert a size in bytes to a human-readable format with proper units.
+    """
+    # Define the units and their respective thresholds
+    units = ["Bytes", "KB", "MB", "GB", "TB", "PB"]
+    unit_index = 0
+
+    # Convert to the appropriate unit
+    while size_in_bytes >= 1024 and unit_index < len(units) - 1:
+        size_in_bytes /= 1024
+        unit_index += 1
+
+    # Return the formatted string with 2 decimal places
+    return f"{size_in_bytes:.2f} {units[unit_index]}"
+
+def calculate_eclipse_time(file_size, upload_speed):
+    """
+    Calculate the upload time in hh:mm:ss format given the file size and upload speed.
+    
+    :param file_size: Size of the file in bytes.
+    :param upload_speed: Upload speed in bps (megabits per second).
+    :return: Upload time in hh:mm:ss format.
+    """
+    # Convert file size from bytes to bits (1 byte = 8 bits)
+    file_size_bits = file_size * 8
+    
+    # Calculate total upload time in seconds
+    upload_time_seconds = file_size_bits / upload_speed
+
+    # Format the time as hh:mm:ss
+    return format_time(upload_time_seconds)
+
+def format_time(seconds):
+    """
+    Format seconds into hh:mm:ss format.
+
+    :param seconds: Time in seconds.
+    :return: Formatted time string in hh:mm:ss format.
+    """
+    if seconds < 1:
+        return "less than 1 sec"
+    hours = int(seconds // 3600)
+    remaining_seconds = seconds % 3600
+    minutes = int(remaining_seconds // 60)
+    seconds = int(remaining_seconds % 60)
+    return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
+                           
+    
+
 

src/copier.py

@@ -102,12 +102,12 @@ def copy_file(self, file_info, overwrite, dryrun):
             else: 
                 file_info[SKIPPED] = False
             #self.log.info(f'Copying from {org_url} to s3://{self.bucket_name}/{key.strip("/")} ...')
-            self.log.info(f'Copying from {org_url} to destination folder in S3 bucket ...')
-            dest_size, msg = self._upload_obj(org_url, key, org_size)
+            self.log.info(f'Uploading file, "{org_url}" to destination...')
+            dest_size = self._upload_obj(org_url, key, org_size)
             if dest_size != org_size:
                 self.log.error(f'Uploading “{file_name}” failed - uploading was not complete. Please try again and contact the helpdesk if this error persists.')
                 return {self.STATUS: False}
-
+            
             return succeed
         except ClientError as ce:
             self.log.debug(ce)
@@ -137,13 +137,13 @@ def _upload_obj(self, org_url, key, org_size):
             t_config = TransferConfig(multipart_threshold=self.MULTI_PART_THRESHOLD,
                                         multipart_chunksize=chunk_size)
             with open(org_url, 'rb') as stream:
-                self.bucket.upload_file_obj(key, stream, t_config)
+                self.bucket.upload_file_obj(org_size, key, stream, t_config)
         else: #small file
             md5_obj = get_md5_hex_n_base64(org_url)
             md5_base64 = md5_obj['base64']
             with open(org_url, 'rb') as data:
-                self.bucket.put_file_obj(key, data, md5_base64)
-
+                self.bucket.put_file_obj(org_size, key, data, md5_base64 )
+            
         self.files_copied += 1
-        self.log.info(f'Copying file {key} SUCCEEDED!')
-        return self.bucket.get_object_size(key)
+        size, msg =  self.bucket.get_object_size(key)
+        return size