added fixed and intersection metric

susrisha · susrisha · commit 1edfd81a94e2 · 2024-09-17T12:40:14.000+05:30
diff --git a/src/assets/messages/incoming.json b/src/assets/messages/incoming.json
@@ -3,7 +3,8 @@
     "messageId": "message-id-from-msg",
     "data": {
       "jobId": "0b41ebc5-350c-42d3-90af-3af4ad3628fb",
-      "data_file": "https://tdeisamplestorage.blob.core.windows.net/osw/test_upload/Archive.zip",
-      "algorithms": "fixed"
+      "data_file": "https://tdeisamplestorage.blob.core.windows.net/osw/test/wenatchee.zip",
+      "algorithm": "fixed",
+      "sub_regions_file":""
     }
   }
diff --git a/src/calculators/__init__.py b/src/calculators/__init__.py
@@ -1,3 +1,3 @@
 from .qm_calculator import QMCalculator
-from .qm_fixed_calculator import QMFixedCalculator, QMRandomCalculator
-from .xn_qm_lib import QMIXNCalculator
+from .qm_fixed_calculator import QMFixedCalculator
+from .qm_xn_lib_calculator import QMXNLibCalculator
diff --git a/src/calculators/qm_fixed_calculator.py b/src/calculators/qm_fixed_calculator.py
@@ -1,4 +1,4 @@
-from qm_calculator import QMCalculator, QualityMetricResult
+from src.calculators.qm_calculator import QMCalculator, QualityMetricResult
 import random
 import geopandas as gpd
 import sys
diff --git a/src/calculators/qm_xn_lib_calculator.py b/src/calculators/qm_xn_lib_calculator.py
@@ -1,12 +1,12 @@
-from qm_calculator import QMCalculator, QualityMetricResult
+from src.calculators.qm_calculator import QMCalculator, QualityMetricResult
 import geopandas as gpd
 import sys
 import warnings
 import networkx as nx
 import traceback
 import geonetworkx as gnx
 import osmnx as ox
-import dask_geopandas
+import dask_geopandas 
 from shapely import Point, LineString, MultiLineString, Polygon, MultiPolygon
 from shapely.ops import voronoi_diagram
 import itertools
diff --git a/src/config.py b/src/config.py
@@ -1,7 +1,7 @@
 import os
 from dotenv import load_dotenv
 from pydantic_settings import BaseSettings
-from src.calculators import QMFixedCalculator, QMRandomCalculator, QMIXNCalculator
+from src.calculators import QMFixedCalculator, QMXNLibCalculator
 
 load_dotenv()
 
@@ -12,7 +12,7 @@ class Config(BaseSettings):
     incoming_topic_subscription: str = os.environ.get('QUALITY_REQ_SUB', '')
     outgoing_topic_name: str = os.environ.get('QUALITY_RES_TOPIC', '')
     storage_container_name: str = os.environ.get('CONTAINER_NAME', 'osw')
-    algorithm_dictionary: dict = {"fixed":QMFixedCalculator,"ixn":QMIXNCalculator}
+    algorithm_dictionary: dict = {"fixed":QMFixedCalculator,"ixn":QMXNLibCalculator}
     max_concurrent_messages: int = os.environ.get('MAX_CONCURRENT_MESSAGES', 1)
 
     def get_download_folder(self) -> str:
diff --git a/src/models/quality_request.py b/src/models/quality_request.py
@@ -6,8 +6,8 @@
 class RequestData:
     jobId: str
     data_file: str
-    algorithms: str
-    intersectionFile: Optional[str] = None
+    algorithm: str
+    sub_regions_file: Optional[str] = None
 
 
 @dataclass
diff --git a/src/services/osw_qm_calculator_service.py b/src/services/osw_qm_calculator_service.py
@@ -1,6 +1,6 @@
 import zipfile
 from src.config import Config
-from src.calculators import QMIXNCalculator
+from src.calculators import QMXNLibCalculator, QMFixedCalculator, QMCalculator
 import json
 import os
 import tempfile
@@ -40,24 +40,58 @@ def calculate_quality_metric(self, input_file, algorithm_names, output_path, ixn
             None
 
         """
-        with zipfile.ZipFile(input_file, 'r') as input_zip:
-            input_unzip_folder = tempfile.TemporaryDirectory()
-            file_list = self.extract_zip(input_zip, input_unzip_folder.name)
-        logger.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()
-        logger.info(f"Started calculating quality metrics for input files: {input_files_path}")
-        for input_file in input_files_path:
-            qm_calculated_json = self.parse_and_calculate_quality_metric(input_file, algorithm_names,ixn_file)
-            qm_calculated_file_path = os.path.join(output_unzip_folder.name, os.path.basename(input_file))
-            with open(qm_calculated_file_path, 'w') as qm_file:
-                json.dump(qm_calculated_json, qm_file)
-        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)
-        logger.info(f'Cleaning up temporary folders.')
-        input_unzip_folder.cleanup()
-        output_unzip_folder.cleanup()
+        try:
+            with zipfile.ZipFile(input_file, 'r') as input_zip:
+                input_unzip_folder = tempfile.TemporaryDirectory()
+                input_files_path = self.extract_zip(input_zip, input_unzip_folder.name)
+            # 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}")
+            # 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:
+                raise Exception('Edges file not found in input files.')
+            edges_file_path = edges_file_path[0]
+            edges_file_basename = os.path.basename(edges_file_path)
+            edges_file_without_extension = os.path.splitext(edges_file_basename)[0]
+            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)
+                qm_calculator.calculate_quality_metric()
+            # Copy the rest of the files from input to output
+            for file_path in input_files_path:
+                if 'edges' in file_path:
+                    continue
+                file_basename = os.path.basename(file_path)
+                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}')
+            self.zip_folder(output_unzip_folder.name, output_path)
+            logging.info(f'Cleaning up temporary folders.')
+            input_unzip_folder.cleanup()
+            output_unzip_folder.cleanup()
+        except Exception as e:
+            logging.error(f'Error calculating quality metrics: {e}')
+            raise e
+
+    def get_osw_qm_calculator(self, algorithm_name:str, ixn_file:str=None, edges_file:str=None, output_file:str=None) -> QMCalculator:
+        """
+        Returns an instance of the specified quality metric calculator.
+
+        Args:
+            algorithm_name (str): The name of the quality metric calculator.
+
+        Returns:
+            QMCalculator: An instance of the specified quality metric calculator.
+
+        """
+        if algorithm_name == 'ixn':
+            return QMXNLibCalculator(edges_file, output_file, ixn_file)
+        else:
+            return QMFixedCalculator(edges_file, output_file)
 
     def zip_folder(self, input_folder, output_zip):
         """
@@ -71,7 +105,6 @@ def zip_folder(self, input_folder, output_zip):
             None
 
         """
-        print(output_zip)
         with zipfile.ZipFile(output_zip, 'w') as output_zip:
             for root, dirs, files in os.walk(input_folder):
                 for file in files:
@@ -90,12 +123,12 @@ def extract_zip(self, input_zip, unzip_folder) -> [str]:
 
         """
         input_zip.extractall(unzip_folder)
-        # walk through and collect all files
-        input_files = []
+        input_file_paths = []
+        # Get the file paths for all the files using os.walk
         for root, dirs, files in os.walk(unzip_folder):
             for file in files:
-                input_files.append(os.path.join(root, file))
-        return input_files
+                input_file_paths.append(os.path.join(root, file))
+        return input_file_paths
 
     def parse_and_calculate_quality_metric(self, input_file, algorithm_names, ixn_file=None):
         """
@@ -117,15 +150,18 @@ def parse_and_calculate_quality_metric(self, input_file, algorithm_names, ixn_fi
             if not algo_name in config.algorithm_dictionary:
                 logger.warning('Algorithm not found : ' + algo_name)
             else:
-                algo_instances.append(config.algorithm_dictionary[algo_name]())
+                if algo_name != 'ixn':
+                    algo_instances.append(config.algorithm_dictionary[algo_name]())
         with open(input_file, 'r') as input_file:
             input_json = json.load(input_file)
             for feature in input_json['features']:
                 for calculator in algo_instances:
                     feature[calculator.qm_metric_tag()] = calculator.calculate_quality_metric(feature)
 
         if 'ixn' in algorithm_names:
-            ixn_calculator = QMIXNCalculator()
+            # get the edges file from the input
+
+            ixn_calculator = QMXNLibCalculator()
             input_json = ixn_calculator.calculate_quality_metric(input_file, ixn_file)
              
         end_time = time.time()
diff --git a/src/services/servicebus_service.py b/src/services/servicebus_service.py
@@ -28,68 +28,89 @@ def __new__(cls):
 
     def __init__(self) -> None:
         self.config = Config()
-        self.core = Core()
-        self.incoming_topic = self.core.get_topic(self.config.incoming_topic_name,
-                                                  max_concurrent_messages=self.config.max_concurrent_messages)
-        self.storage_service = StorageService(core=self.core)
+        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.listening_thread = threading.Thread(target=self.incoming_topic.subscribe, args=[self.config.incoming_topic_subscription, self.process_message])
         # Start listening to the things
-        self.listening_thread = threading.Thread(target=self.listen)
+        # self.incoming_topic.subscribe(self.config.incoming_topic_subscription, self.handle_message)
         self.listening_thread.start()
         pass
-    
-    def listen(self):
-        self.incoming_topic.subscribe(self.config.incoming_topic_subscription, self.handle_message)
-        pass
 
-    def handle_message(self, msg: QueueMessage):
-        # Logs and creates a thread for processing
-        self.process_message(msg=msg)
+    # def handle_message(self, msg: QueueMessage):
+    #     # Logs and creates a thread for processing
+    #     process_thread = threading.Thread(target=self.process_message, args=[msg])
+    #     process_thread.start()
 
     def process_message(self, msg: QueueMessage):
-        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
-        input_file_url = quality_request.data.data_file
-        parsed_url = urlparse(input_file_url)
-        file_name = os.path.basename(parsed_url.path)
-        input_dir_path = parsed_url.path
-        download_folder = os.path.join(self.config.get_download_folder(),msg.messageId)
-        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)
+        try:
+            logging.info(f"Processing message {msg.messageId}")
+            # Parse the message
+            quality_request = QualityRequest(messageType=msg.messageType,messageId=msg.messageId,data=msg.data)
+            # Download the file
+            input_file_url = quality_request.data.data_file
+            parsed_url = urlparse(input_file_url)
+            file_name = os.path.basename(parsed_url.path)
+            input_dir_path = parsed_url.path
+            download_folder = os.path.join(self.config.get_download_folder(),msg.messageId)
+            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)
 
-        # intersection file
-        ixn_file_url = quality_request.data.intersectionFile
-        ixn_file_path = None
-        if 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)
-            # quality_request.data.intersectionFile = ixn_file_path
+            # intersection file
+            ixn_file_url = quality_request.data.sub_regions_file
+            ixn_file_path = None
+            if ixn_file_url or 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)
+                # quality_request.data.intersectionFile = ixn_file_path
 
-        # Process the file
-        output_folder = os.path.join(download_folder,'qm')
-        os.makedirs(output_folder,exist_ok=True)
-        output_file_local_path = os.path.join(output_folder,'qm-output.zip')
-        qm_calculator = OswQmCalculator()
-        algorithm_names = quality_request.data.algorithms.split(',')
-        qm_calculator.calculate_quality_metric(download_path, algorithm_names,output_file_local_path,ixn_file_path)
-        # 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}')
+            # Process the file
+            output_folder = os.path.join(download_folder,'qm')
+            os.makedirs(output_folder,exist_ok=True)
+            output_file_local_path = os.path.join(output_folder,'qm-output.zip')
+            qm_calculator = OswQmCalculator()
+            algorithm_names = quality_request.data.algorithm.split(',')
+            qm_calculator.calculate_quality_metric(download_path, algorithm_names,output_file_local_path,ixn_file_path)
+            # 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}')
 
-        response = QualityMetricResponse(
-            messageType=msg.messageType,
-            messageId=msg.messageId,
-            data=response_data
-        )
-        self.send_response(msg=response)
-        # Process the message
-        # Clean up the download_folder
-        logger.info('Cleaning up download folder')
-        shutil.rmtree(download_folder)
+            response_data = {
+                'status':'success',
+                'message':'Quality metrics calculated successfully',
+                'success':True,
+                'dataset_url':input_file_url,
+                'qm_dataset_url':output_file_url
+            }
+            response = QualityMetricResponse(
+                messageType=msg.messageType,
+                messageId=msg.messageId,
+                data=  response_data
+            )
+            self.send_response(response)
+            # Process the message
+            # Clean up the download_folder
+            logging.info('Cleaning up download folder')
+            shutil.rmtree(download_folder)
+        except Exception as e:
+            logging.error(f'Error processing message {msg.messageId} : {e}')
+            response_data = {
+                'status':'failed',
+                'message':str(e),
+                'success':False,
+                'dataset_url':input_file_url,
+                'qm_dataset_url':None
+            }
+            response = QualityMetricResponse(
+                messageType=msg.messageType,
+                messageId=msg.messageId,
+                data=  response_data
+            )
+            self.send_response(response)
         pass
 
     def send_response(self, msg: QueueMessage):
diff --git a/test.py b/test.py

Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,8 @@`
`3`	`3`	`"messageId": "message-id-from-msg",`
`4`	`4`	`"data": {`
`5`	`5`	`"jobId": "0b41ebc5-350c-42d3-90af-3af4ad3628fb",`
`6`		`- "data_file": "https://tdeisamplestorage.blob.core.windows.net/osw/test_upload/Archive.zip",`
`7`		`- "algorithms": "fixed"`
	`6`	`+ "data_file": "https://tdeisamplestorage.blob.core.windows.net/osw/test/wenatchee.zip",`
	`7`	`+ "algorithm": "fixed",`
	`8`	`+ "sub_regions_file":""`
`8`	`9`	`}`
`9`	`10`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-from qm_calculator import QMCalculator, QualityMetricResult`
	`1`	`+from src.calculators.qm_calculator import QMCalculator, QualityMetricResult`
`2`	`2`	`import random`
`3`	`3`	`import geopandas as gpd`
`4`	`4`	`import sys`