Merge pull request #325 from ansforge/refactor/converter/cisu-conversion-pipeline

feat(converter): RC-RI conversion

Author: SimonClo

converter/.gitignore

@@ -168,4 +168,7 @@ cython_debug/
 #.idea/
 
 # PyPI configuration file
-.pypirc
+.pypirc
+
+# Request cache
+*_cache.sqlite

converter/README.md

@@ -48,6 +48,8 @@ uv run pytest tests/test_utils.py -k test_format_object_primitive
 uv run pytest -rP
 ```
 
+Note : the tests download files (json samples & schemas) using the Github API. To avoid hammering the API leading to tests failure, a cache system has been added for the tests runner. To invalidate the cache and download the files again, delete the `schemas_and_samples_cache.sqlite` file at the root of the repository.
+
 ### Running the Service
 
 Development mode:

converter/converter/cisu/base_cisu_converter.py

@@ -0,0 +1,77 @@
+from converter.conversion_mixin import ConversionMixin
+from typing import Any, Dict
+
+
+class BaseCISUConverter(ConversionMixin):
+    def __init__(self):
+        raise ValueError(
+            "BaseMessageConverter is an abstract class and cannot be instantiated directly. Use a subclass instead."
+        )
+
+    @classmethod
+    def get_rs_message_type(cls) -> str:
+        raise NotImplementedError(
+            "Subclasses must implement this method to return the message type in RS specifications."
+        )
+
+    @classmethod
+    def get_cisu_message_type(cls) -> str:
+        raise NotImplementedError(
+            "Subclasses must implement this method to return the message type in RC specifications."
+        )
+
+    @classmethod
+    def from_rs_to_cisu(cls, edxl_json):
+        raise ValueError(
+            f"Traduction from '{cls.get_rs_message_type()}' to '{cls.get_cisu_message_type()}' is not supported."
+        )
+
+    @classmethod
+    def from_cisu_to_rs(cls, edxl_json):
+        raise ValueError(
+            f"Traduction from '{cls.get_cisu_message_type()}' to '{cls.get_rs_message_type()}' is not supported."
+        )
+
+    @classmethod
+    def copy_cisu_input_content(cls, edxl_json: Dict[str, Any]) -> Dict[str, Any]:
+        return cls._copy_input_content(edxl_json, cls.get_cisu_message_type())
+
+    @classmethod
+    def copy_cisu_input_use_case_content(
+        cls, edxl_json: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        return cls._copy_input_use_case_content(edxl_json, cls.get_cisu_message_type())
+
+    @classmethod
+    def format_cisu_output_json(
+        cls,
+        output_json: Dict[str, Any],
+        output_use_case_json: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return cls._format_output_json(
+            output_json,
+            output_use_case_json,
+            cls.get_cisu_message_type(),
+        )
+
+    @classmethod
+    def copy_rs_input_content(cls, edxl_json: Dict[str, Any]) -> Dict[str, Any]:
+        return cls._copy_input_content(edxl_json, cls.get_rs_message_type())
+
+    @classmethod
+    def copy_rs_input_use_case_content(
+        cls, edxl_json: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        return cls._copy_input_use_case_content(edxl_json, cls.get_rs_message_type())
+
+    @classmethod
+    def format_rs_output_json(
+        cls,
+        output_json: Dict[str, Any],
+        output_use_case_json: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return cls._format_output_json(
+            output_json,
+            output_use_case_json,
+            cls.get_rs_message_type(),
+        )

…nverter/converter/cisu/cisu_converter.py …reate_case/create_case_cisu_converter.pyconverter/converter/cisu/cisu_converter.py renamed to converter/converter/cisu/create_case/create_case_cisu_converter.py converter/converter/cisu/cisu_converter.py renamed to converter/converter/cisu/create_case/create_case_cisu_converter.py

@@ -6,19 +6,20 @@
 
 from yaml import dump
 
-from .utils import add_to_initial_alert_notes
-from ..constants import Constants
-from ..utils import (
+from converter.cisu.utils import add_to_initial_alert_notes
+from converter.constants import Constants
+from converter.utils import (
     delete_paths,
     get_field_value,
     get_recipient,
     get_sender,
     is_field_completed,
     translate_key_words,
 )
+from converter.cisu.base_cisu_converter import BaseCISUConverter
 
 
-class CISUConverterV3:
+class CreateCaseCISUConverter(BaseCISUConverter):
     """Handles CISU format conversions"""
 
     CISU_PATHS_TO_DELETE = [
@@ -70,7 +71,15 @@ class CISUConverterV3:
     DEFAULT_WHATS_HAPPEN = {"code": "C11.06.00", "label": "Autre nature de fait"}
 
     @classmethod
-    def from_cisu(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
+    def get_rs_message_type(cls) -> str:
+        return "createCaseHealth"
+
+    @classmethod
+    def get_cisu_message_type(cls) -> str:
+        return "createCase"
+
+    @classmethod
+    def from_cisu_to_rs(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
         """
         Convert from CISU to Health format
 
@@ -161,21 +170,11 @@ def add_object_to_medical_notes(
             json_data["medicalNote"].append(new_note)
 
         # Create independent envelope copy without usecase for output
-        output_json = copy.deepcopy(input_json)
-        if "createCase" not in input_json.get("content", [{}])[0].get(
-            "jsonContent", {}
-        ).get("embeddedJsonContent", {}).get("message", {}):
-            raise ValueError("Input JSON must contain 'createCase' key")
-        del output_json["content"][0]["jsonContent"]["embeddedJsonContent"]["message"][
-            "createCase"
-        ]
+        output_json = cls.copy_cisu_input_content(input_json)
 
         # Create independent use case copy for output
-        input_use_case_json = input_json["content"][0]["jsonContent"][
-            "embeddedJsonContent"
-        ]["message"]["createCase"]
         sender_id = get_sender(input_json)
-        output_use_case_json = copy.deepcopy(input_use_case_json)
+        output_use_case_json = cls.copy_cisu_input_use_case_content(input_json)
 
         # - Updates
         output_use_case_json["owner"] = get_recipient(input_json)
@@ -195,10 +194,7 @@ def add_object_to_medical_notes(
         # - Delete paths - /!\ It must be the last step
         delete_paths(output_use_case_json, cls.CISU_PATHS_TO_DELETE)
 
-        output_json["content"][0]["jsonContent"]["embeddedJsonContent"]["message"][
-            "createCaseHealth"
-        ] = output_use_case_json
-        return output_json
+        return cls.format_rs_output_json(output_json, output_use_case_json)
 
     @staticmethod
     def count_victims(json_data: Dict[str, Any]) -> int:
@@ -219,7 +215,7 @@ def get_victim_count(cls, json_data: Dict[str, Any]):
         return {"count": "BEAUCOUP"}
 
     @classmethod
-    def to_cisu(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
+    def from_rs_to_cisu(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
         """
         Convert from Health to CISU format
 
@@ -255,19 +251,10 @@ def add_default_external_info_type(json_data: Dict[str, Any]):
                         info["type"] = "AUTRE"
 
         # Create independent envelope copy without usecase for output
-        output_json = copy.deepcopy(input_json)
-        if "createCaseHealth" not in input_json.get("content", [{}])[0].get(
-            "jsonContent", {}
-        ).get("embeddedJsonContent", {}).get("message", {}):
-            raise ValueError("Input JSON must contain 'createCaseHealth' key")
-        del output_json["content"][0]["jsonContent"]["embeddedJsonContent"]["message"][
-            "createCaseHealth"
-        ]
+        output_json = cls.copy_rs_input_content(input_json)
 
         # Create independent usecase copy for output
-        input_usecase_json = input_json["content"][0]["jsonContent"][
-            "embeddedJsonContent"
-        ]["message"]["createCaseHealth"]
+        input_usecase_json = cls.copy_rs_input_use_case_content(input_json)
         output_usecase_json = copy.deepcopy(input_usecase_json)
 
         # Generate unique IDs
@@ -321,7 +308,4 @@ def add_default_external_info_type(json_data: Dict[str, Any]):
                 get_field_value(output_usecase_json, "$.location")
             )
 
-        output_json["content"][0]["jsonContent"]["embeddedJsonContent"]["message"][
-            "createCase"
-        ] = output_usecase_json
-        return output_json
+        return cls.format_cisu_output_json(output_json, output_usecase_json)

converter/converter/cisu/resources_info/resources_info_cisu_constants.py

@@ -0,0 +1,11 @@
+class ResourcesInfoCISUConstants:
+    RESOURCE_PATH = "$.resource"
+    STATE_PATH = "$.state"
+    VEHICLE_TYPE_PATH = "$.vehicleType"
+
+    PATIENT_ID_KEY = "patientId"
+
+    VEHICLE_TYPE_SIS = "SIS"
+    DEFAULT_CISU_STATE_VEHICLE_TYPE = "SMUR"
+
+    DEFAULT_CISU_STATE_STATUS = "DECISION"

converter/converter/cisu/resources_info/resources_info_cisu_converter.py

@@ -0,0 +1,79 @@
+import datetime
+
+from converter.cisu.base_cisu_converter import BaseCISUConverter
+from typing import Any, Dict
+
+from converter.cisu.resources_info.resources_info_cisu_constants import (
+    ResourcesInfoCISUConstants,
+)
+from converter.utils import get_field_value, set_value, delete_paths
+
+
+class ResourcesInfoCISUConverter(BaseCISUConverter):
+    @classmethod
+    def get_rs_message_type(cls) -> str:
+        return "resourcesInfo"
+
+    @classmethod
+    def get_cisu_message_type(cls) -> str:
+        return "resourcesInfoCisu"
+
+    @classmethod
+    def from_cisu_to_rs(cls, edxl_json: Dict[str, Any]) -> Dict[str, Any]:
+        output_json = cls.copy_cisu_input_content(edxl_json)
+        output_use_case_json = cls.copy_cisu_input_use_case_content(edxl_json)
+        resources = get_field_value(
+            output_use_case_json, ResourcesInfoCISUConstants.RESOURCE_PATH
+        )
+
+        for resource in resources:
+            state = get_field_value(resource, ResourcesInfoCISUConstants.STATE_PATH)
+            set_value(resource, ResourcesInfoCISUConstants.STATE_PATH, [state])
+
+        return cls.format_rs_output_json(output_json, output_use_case_json)
+
+    @classmethod
+    def from_rs_to_cisu(cls, edxl_json: Dict[str, Any]) -> Dict[str, Any]:
+        output_json = cls.copy_rs_input_content(edxl_json)
+        output_use_case_json = cls.copy_rs_input_use_case_content(edxl_json)
+
+        resources = get_field_value(
+            output_use_case_json, ResourcesInfoCISUConstants.RESOURCE_PATH
+        )
+        for resource in resources:
+            rs_vehicle_type = get_field_value(
+                resource, ResourcesInfoCISUConstants.VEHICLE_TYPE_PATH
+            )
+            cisu_vehicle_type = cls.translate_to_cisu_vehicle_type(rs_vehicle_type)
+            set_value(
+                resource,
+                ResourcesInfoCISUConstants.VEHICLE_TYPE_PATH,
+                cisu_vehicle_type,
+            )
+
+            cls.keep_last_state(resource)
+
+            delete_paths(resource, [ResourcesInfoCISUConstants.PATIENT_ID_KEY])
+
+        return cls.format_cisu_output_json(output_json, output_use_case_json)
+
+    @classmethod
+    def translate_to_cisu_vehicle_type(cls, rs_vehicle_type: str) -> str:
+        if rs_vehicle_type.startswith(ResourcesInfoCISUConstants.VEHICLE_TYPE_SIS):
+            return ResourcesInfoCISUConstants.VEHICLE_TYPE_SIS
+        return ResourcesInfoCISUConstants.DEFAULT_CISU_STATE_VEHICLE_TYPE
+
+    @classmethod
+    def keep_last_state(cls, resource: Dict[str, Any]) -> None:
+        states = get_field_value(resource, ResourcesInfoCISUConstants.STATE_PATH)
+        if states and len(states) > 0:
+            latest_state = sorted(states, key=lambda x: x.get("datetime", ""))[-1]
+        else:
+            latest_state = {
+                "datetime": datetime.datetime.now(datetime.timezone.utc).isoformat(
+                    timespec="seconds"
+                ),
+                "status": ResourcesInfoCISUConstants.DEFAULT_CISU_STATE_STATUS,
+            }
+
+        set_value(resource, ResourcesInfoCISUConstants.STATE_PATH, latest_state)

converter/converter/cisu/utils.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict, List
 
-from ..utils import concatenate_values, get_field_value, is_field_completed
+from converter.utils import concatenate_values, get_field_value, is_field_completed
 
 
 def add_object_to_initial_alert_notes(json_data: Dict[str, Any], note_text: str):

…r/converter/versions/conversion_mixin.py converter/converter/conversion_mixin.pyconverter/converter/versions/conversion_mixin.py renamed to converter/converter/conversion_mixin.py converter/converter/versions/conversion_mixin.py renamed to converter/converter/conversion_mixin.py

@@ -1,19 +1,18 @@
 import copy
 from typing import Dict, Any
 
-from converter.versions.base_message_converter import BaseMessageConverter
 
-
-class ConversionMixin(BaseMessageConverter):
+class ConversionMixin:
     CONTENT_KEY = "content"
     JSON_CONTENT_KEY = "jsonContent"
     EMBEDDED_JSON_CONTENT_KEY = "embeddedJsonContent"
     MESSAGE_KEY = "message"
 
     @classmethod
-    def copy_input_content(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
+    def _copy_input_content(
+        cls, input_json: Dict[str, Any], message_type: str
+    ) -> Dict[str, Any]:
         output_json = copy.deepcopy(input_json)
-        message_type = cls.get_message_type()
 
         message_content = (
             input_json.get(cls.CONTENT_KEY, [{}])[0]
@@ -30,18 +29,21 @@ def copy_input_content(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
         return output_json
 
     @classmethod
-    def copy_input_use_case_content(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
-        message_type = cls.get_message_type()
+    def _copy_input_use_case_content(
+        cls, input_json: Dict[str, Any], message_type: str
+    ) -> Dict[str, Any]:
         input_use_case_json = input_json[cls.CONTENT_KEY][0][cls.JSON_CONTENT_KEY][
             cls.EMBEDDED_JSON_CONTENT_KEY
         ][cls.MESSAGE_KEY][message_type]
         return copy.deepcopy(input_use_case_json)
 
     @classmethod
-    def format_output_json(
-        cls, output_json: Dict[str, Any], output_use_case_json: Dict[str, Any]
+    def _format_output_json(
+        cls,
+        output_json: Dict[str, Any],
+        output_use_case_json: Dict[str, Any],
+        message_type: str,
     ) -> Dict[str, Any]:
-        message_type = cls.get_message_type()
         output_json[cls.CONTENT_KEY][0][cls.JSON_CONTENT_KEY][
             cls.EMBEDDED_JSON_CONTENT_KEY
         ][cls.MESSAGE_KEY][message_type] = output_use_case_json