bugfix: handle failure in openrao when return failure status

Author: m-karo

integrations/rmq.py

@@ -2,6 +2,7 @@
 import time
 import pika
 import config
+import traceback
 from typing import List
 from concurrent.futures import ThreadPoolExecutor
 from loguru import logger
@@ -375,9 +376,9 @@ def _process_messages(self, basic_deliver, properties, body):
                 properties.content_type = content_type
                 logger.info(f"Message converted")
             except Exception as error:
-                logger.error(f"Message conversion failed: {error}", exc_info=True)
+                logger.error(f"Message conversion failed: {error}\n{traceback.format_exc()}")
                 ack = False
-                self._channel.basic_reject(basic_deliver.delivery_tag, requeue=True)
+                self._channel.basic_reject(basic_deliver.delivery_tag, requeue=False)
                 # self.connection.close()
                 # self.stop()
 
@@ -388,9 +389,10 @@ def _process_messages(self, basic_deliver, properties, body):
                     logger.info(f"Handling message with handler: {message_handler.__class__.__name__}")
                     body, properties = message_handler.handle(body, properties=properties, channel=self._channel)
                 except Exception as error:
-                    logger.error(f"Message handling failed: {error}", exc_info=True)
+                    logger.error(f"Message handling failed: {error}\n{traceback.format_exc()}")
+                    logger.exception("Message handling failed, see traceback in document")
                     ack = False
-                    self._channel.basic_reject(basic_deliver.delivery_tag, requeue=True)
+                    self._channel.basic_reject(basic_deliver.delivery_tag, requeue=False)
                     # self.connection.close()
                     # self.stop()
 

rao/handlers.py

@@ -262,6 +262,12 @@ def handle(self, message: bytes, properties: object, **kwargs):
 
             # Serialize results to json
             results = optimizer.results.to_json()
+            if results['computationStatus'] == 'failure':
+                logger.error(f"Optimizer failed computation: {results}")
+                logger.error(f"Enable pypowsybl logs for more information")
+                continue
+
+            # Check if there are any optimized remedial actions
             if not results['networkActionResults'] and not results['rangeActionResults']:
                 logger.warning(f"No possible actions proposed by optimizer")
 

remedial_action_schedules/config.properties

@@ -3,5 +3,5 @@ RMQ_QUEUE_IN = object-storage.schedules.csa
 
 [HANDLER]
 CONVERTER_KEY_MODE = local
-ELASTIC_SCHEDULES_INDEX = rao-results
+ELASTIC_SCHEDULES_INDEX = csa-remedial-action-schedules
 

remedial_action_schedules/handlers.py

@@ -1,5 +1,6 @@
 from pika import BasicProperties
 import uuid
+import json
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Set, Tuple, Union
@@ -326,13 +327,16 @@ def handle(self, message: bytes, properties: BasicProperties,  **kwargs):
         # JSON normalize and transform to DataFrame
         df = self.normalize_cim_payload(data)
 
+        # TODO need to get CO and RA from object storage and merge
+
         # Convert to dictionary
         data_to_send = df.to_dict(orient='records')
 
-        response = self.elastic_service.send_to_elastic(
+        response = self.elastic_service.send_to_elastic_bulk(
             index=ELASTIC_METADATA_INDEX,
-            json_message=data_to_send,
-            id=metadata_object.get('identifier', None)
+            json_message_list=data_to_send,
+            id_from_metadata=True,
+            id_metadata_list=["@id"],
         )
 
         logger.info(f"Message sending to Elastic successful: {response}")
@@ -341,23 +345,50 @@ def handle(self, message: bytes, properties: BasicProperties,  **kwargs):
 
 
 if __name__ == "__main__":
-    rdf_xml = r"C:\Users\martynas.karobcikas\Downloads\ras-example.xml"
-    # rdf_xml = r"C:\Users\martynas.karobcikas\Documents\Python projects\RAO\test-data\TC1_assessed_elements.xml"
-    # rdf_xml = r"C:\Users\martynas.karobcikas\Documents\Python projects\RAO\test-data\TC1_contingencies.xml"
-    # rdf_xml = r"C:\Users\martynas.karobcikas\Documents\Python projects\RAO\test-data\TC1_remedial_actions.xml"
-    g = Graph()
-    g.parse(rdf_xml, format="xml")  # your RDF/XML file
-
-    result = convert_cim_rdf_to_json(rdf_xml, root_class=["RemedialActionSchedule"], key_mode='local')
-    # result = convert_cim_rdf_to_json(rdf_xml, root_class=["RemedialActionSchedule"], key_mode='qualified')
-    # result = convert_cim_rdf_to_json(rdf_xml, root_class=["GridStateAlterationRemedialAction"], key_mode='local')
-    # result = convert_cim_rdf_to_json(rdf_xml, root_class=["OrdinaryContingency", "ExceptionalContingency"], key_mode='local')
-
-    import json
-    print(json.dumps(result, indent=2))
-
-    with open("test.json", "w") as f:
-        json.dump(result, f, ensure_ascii=False, indent=4)
+    # rdf_xml = r"C:\Users\martynas.karobcickas\Downloads\ras-example.xml"
+    # rdf_xml = r"C:\Users\martynas.karobcickas\Documents\Python projects\RAO\test-data\TC1_assessed_elements.xml"
+    # rdf_xml = r"C:\Users\martynas.karobcickas\Documents\Python projects\RAO\test-data\TC1_contingencies.xml"
+    # rdf_xml = r"C:\Users\martynas.karobcickas\Documents\Python projects\RAO\test-data\TC1_remedial_actions.xml"
+    # g = Graph()
+    # g.parse(rdf_xml, format="xml")  # Put your RDF/XML file
+
+    # result = convert_cim_rdf_to_json(rdf_xml, root_class=["RemedialActionSchedule"], key_mode="local")
+    # result = convert_cim_rdf_to_json(rdf_xml, root_class=["RemedialActionSchedule"], key_mode="qualified")
+    # result = convert_cim_rdf_to_json(rdf_xml, root_class=["GridStateAlterationRemedialAction"], key_mode="local")
+    # result = convert_cim_rdf_to_json(rdf_xml, root_class=["OrdinaryContingency", "ExceptionalContingency"], key_mode="local")
+
+    # print json
+    # print(json.dumps(result, indent=2))
+
+    # with open("test.json", "w") as f:
+    #     json.dump(result, f, ensure_ascii=False, indent=4)
+    # df = RemedialActionScheduleToElasticHandler.normalize_cim_payload(result)
+    # print(df.head())
+
+    # Define RMQ test message
+    headers = {
+        "baCorrelationID": f"{uuid.uuid4()}",
+        "baseMessageID": f"{uuid.uuid4()}",
+        "businessType": "CSA-INPUT",
+        "messageID": f"{uuid.uuid4()}",
+        "sendTimestamp": datetime.utcnow().isoformat(),
+        "sender": "TSOX",
+        "senderApplication": "APPX",
+        "service": "INPUT-DATA",
+    }
 
-    df = RemedialActionScheduleToElasticHandler.normalize_cim_payload(result)
-    print(df.head())
+    properties = BasicProperties(
+        content_type="application/octet-stream",
+        delivery_mode=2,
+        priority=4,
+        message_id=f"{uuid.uuid4()}",
+        timestamp=147728025,
+        headers=headers,
+    )
+
+    with open(r"C:\Users\martynas.karobcickas\Downloads\ras-example.xml", "rb") as file:
+        file_bytes = file.read()
+
+    # Create instance
+    service = RemedialActionScheduleToElasticHandler()
+    result = service.handle(message=file_bytes, properties=properties)