Merge pull request #20 from TaskarCenterAtUW/feature-1063

Fixed the issues with messages

Author: susrisha

src/services/osw_qm_calculator_service.py

@@ -47,7 +47,7 @@ def calculate_quality_metric(self, input_file, algorithm_names, output_path, ixn
             # logging.info(f"Extracted input files: {file_list}")
             # input_files_path = [os.path.join(input_unzip_folder.name, file) for file in file_list]
             output_unzip_folder = tempfile.TemporaryDirectory()
-            logging.info(f"Started calculating quality metrics for input files: {input_files_path}")
+            logger.info(f"Started calculating quality metrics for input files: {input_files_path}")
             # Get only the edges file out of the input files
             edges_file_path = [file_path for file_path in input_files_path if 'edges' in file_path]
             if len(edges_file_path) == 0:
@@ -58,7 +58,10 @@ def calculate_quality_metric(self, input_file, algorithm_names, output_path, ixn
             for algorithm_name in algorithm_names:
                 qm_edges_output_path = os.path.join(output_unzip_folder.name, f'{algorithm_name}_{edges_file_without_extension}.geojson')
                 qm_calculator = self.get_osw_qm_calculator(algorithm_name, ixn_file, edges_file_path, qm_edges_output_path)
+                start_time = time.time()
                 qm_calculator.calculate_quality_metric()
+                end_time = time.time()
+                logger.info(f"Time taken to calculate quality metrics for {algorithm_name}: {end_time - start_time} seconds")
             # Copy the rest of the files from input to output
             for file_path in input_files_path:
                 if 'edges' in file_path:
@@ -67,10 +70,10 @@ def calculate_quality_metric(self, input_file, algorithm_names, output_path, ixn
                 output_file_path = os.path.join(output_unzip_folder.name, file_basename)
                 os.rename(file_path, output_file_path)
 
-            logging.info(f"Finished calculating quality metrics for input files: {input_files_path}")
-            logging.info(f'Zipping output files to {output_path}')
+            logger.info(f"Finished calculating quality metrics for input files: {input_files_path}")
+            logger.info(f'Zipping output files to {output_path}')
             self.zip_folder(output_unzip_folder.name, output_path)
-            logging.info(f'Cleaning up temporary folders.')
+            logger.info(f'Cleaning up temporary folders.')
             input_unzip_folder.cleanup()
             output_unzip_folder.cleanup()
         except Exception as e:

src/services/servicebus_service.py

@@ -28,10 +28,10 @@ def __new__(cls):
 
     def __init__(self) -> None:
         self.config = Config()
-        core = Core()
-        self.incoming_topic = core.get_topic(self.config.incoming_topic_name)
-        self.outgoing_topic = core.get_topic(self.config.outgoing_topic_name)
-        self.storage_service = StorageService(core)
+        self.core = Core()
+        self.incoming_topic = self.core.get_topic(self.config.incoming_topic_name)
+        self.outgoing_topic = self.core.get_topic(self.config.outgoing_topic_name)
+        self.storage_service = StorageService(self.core)
         self.listening_thread = threading.Thread(target=self.incoming_topic.subscribe, args=[self.config.incoming_topic_subscription, self.process_message])
         # Start listening to the things
         # self.incoming_topic.subscribe(self.config.incoming_topic_subscription, self.handle_message)
@@ -44,8 +44,9 @@ def __init__(self) -> None:
     #     process_thread.start()
 
     def process_message(self, msg: QueueMessage):
+        logger.info(f"Processing message {msg}")
         try:
-            logging.info(f"Processing message {msg.messageId}")
+            logger.info(f"Processing message {msg.messageId}")
             # Parse the message
             quality_request = QualityRequest(messageType=msg.messageType,messageId=msg.messageId,data=msg.data)
             # Download the file
@@ -57,11 +58,12 @@ def process_message(self, msg: QueueMessage):
             os.makedirs(download_folder,exist_ok=True)
             download_path = os.path.join(download_folder,file_name)
             self.storage_service.download_remote_file(input_file_url, download_path)
-
+            logger.info(f'Downloaded file to {download_path}')
             # intersection file
             ixn_file_url = quality_request.data.sub_regions_file
             ixn_file_path = None
-            if ixn_file_url or ixn_file_url != '':
+            if ixn_file_url is not None:
+                logger.info(f'Downloading intersection file {ixn_file_url}')
                 ixn_file_name = os.path.basename(ixn_file_url)
                 ixn_file_path = os.path.join(download_folder,ixn_file_name)
                 self.storage_service.download_remote_file(ixn_file_url, ixn_file_path)
@@ -77,7 +79,7 @@ def process_message(self, msg: QueueMessage):
             # Upload the file
             output_file_remote_path = f'{self.get_directory_path(input_file_url)}/qm-{quality_request.data.jobId}-output.zip'
             output_file_url = self.storage_service.upload_local_file(output_file_local_path,output_file_remote_path)
-            logging.info(f'Uploaded file to {output_file_url}')
+            logger.info(f'Uploaded file to {output_file_url}')
 
             response_data = {
                 'status':'success',
@@ -94,10 +96,10 @@ def process_message(self, msg: QueueMessage):
             self.send_response(response)
             # Process the message
             # Clean up the download_folder
-            logging.info('Cleaning up download folder')
+            logger.info('Cleaning up download folder')
             shutil.rmtree(download_folder)
         except Exception as e:
-            logging.error(f'Error processing message {msg.messageId} : {e}')
+            logger.error(f'Error processing message {msg.messageId} : {e}')
             response_data = {
                 'status':'failed',
                 'message':str(e),