feat(converter): conversion RC-RI/RS-RI

converter/converter/cisu/base_cisu_converter.py

@@ -1,4 +1,8 @@
-class BaseCISUConverter:
+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."
@@ -27,3 +31,47 @@ 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(),
+        )

converter/converter/cisu/create_case/create_case_cisu_converter.py

@@ -170,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)
@@ -204,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:
@@ -264,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
@@ -330,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

@@ -1,4 +1,12 @@
+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):
@@ -9,3 +17,63 @@ def get_rs_message_type(cls) -> str:
     @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/versions/conversion_mixin.py → 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

converter/converter/versions/base_message_converter.py

@@ -1,7 +1,10 @@
+from converter.conversion_mixin import ConversionMixin
+from typing import Dict, Any
+
 version_order_list = ["v1", "v2", "v3"]
 
 
-class BaseMessageConverter:
+class BaseMessageConverter(ConversionMixin):
     def __init__(self):
         raise ValueError(
             "BaseMessageConverter is an abstract class and cannot be instantiated directly. Use a subclass instead."
@@ -112,3 +115,21 @@ def raise_conversion_impossible_error(cls, source_version, target_version):
         raise ValueError(
             f"Version conversion from {source_version} to {target_version} is not possible."
         )
+
+    @classmethod
+    def copy_input_content(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
+        return cls._copy_input_content(input_json, cls.get_message_type())
+
+    @classmethod
+    def copy_input_use_case_content(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
+        return cls._copy_input_use_case_content(input_json, cls.get_message_type())
+
+    @classmethod
+    def format_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_message_type()
+        )

converter/converter/versions/create_case_health/create_case_health_converter.py

@@ -7,7 +7,7 @@
     is_field_completed,
     map_to_new_value,
 )
-from converter.versions.conversion_mixin import ConversionMixin
+from converter.versions.base_message_converter import BaseMessageConverter
 from converter.versions.create_case_health.v1_v2.constants import V1V2Constants
 from converter.versions.create_case_health.v2_v3.constants import V2V3Constants
 from converter.versions.utils import (
@@ -17,7 +17,7 @@
 )
 
 
-class CreateHealthCaseConverter(ConversionMixin):
+class CreateHealthCaseConverter(BaseMessageConverter):
     @staticmethod
     def get_message_type() -> str:
         return "createCaseHealth"

converter/converter/versions/document_link/document_link_converter.py

@@ -1,13 +1,13 @@
 from typing import Dict, Any
 
 from converter.utils import delete_paths
-from converter.versions.conversion_mixin import ConversionMixin
+from converter.versions.base_message_converter import BaseMessageConverter
 from converter.versions.document_link.document_link_constants import (
     DocumentLinkConstants,
 )
 
 
-class DocumentLinkConverter(ConversionMixin):
+class DocumentLinkConverter(BaseMessageConverter):
     @staticmethod
     def get_message_type():
         return "documentLink"

converter/converter/versions/geo_positions_update/geo_positions_update_converter.py

@@ -1,14 +1,14 @@
 from typing import Dict, Any
 
 from converter.utils import get_field_value, update_json_value, delete_paths
-from converter.versions.conversion_mixin import ConversionMixin
+from converter.versions.base_message_converter import BaseMessageConverter
 from converter.versions.geo_positions_update.geo_positions_update_constants import (
     GeoPositionsUpdateConstants,
 )
 from converter.versions.utils import convert_to_float, convert_to_str
 
 
-class GeoPositionsUpdateConverter(ConversionMixin):
+class GeoPositionsUpdateConverter(BaseMessageConverter):
     @staticmethod
     def get_message_type():
         return "geoPositionsUpdate"

converter/converter/versions/geo_resources_details/geo_resources_details_converter.py

@@ -1,14 +1,14 @@
 from typing import Dict, Any
 
 from converter.utils import get_field_value, map_to_new_value, update_json_value
-from converter.versions.conversion_mixin import ConversionMixin
+from converter.versions.base_message_converter import BaseMessageConverter
 from converter.versions.geo_resources_details.geo_resources_details_constants import (
     GeoResourcesDetailsConstants,
 )
 from converter.versions.utils import reverse_map_to_new_value
 
 
-class GeoResourcesDetailsConverter(ConversionMixin):
+class GeoResourcesDetailsConverter(BaseMessageConverter):
     @staticmethod
     def get_message_type():
         return "geoResourcesDetails"

converter/converter/versions/intervention_report/intervention_report_converter.py

@@ -6,14 +6,14 @@
     map_to_new_value,
     set_value,
 )
-from converter.versions.conversion_mixin import ConversionMixin
+from converter.versions.base_message_converter import BaseMessageConverter
 from converter.versions.intervention_report.intervention_report_constants import (
     InterventionReportConstants,
 )
 from converter.versions.utils import reverse_map_to_new_value
 
 
-class InterventionReportConverter(ConversionMixin):
+class InterventionReportConverter(BaseMessageConverter):
     @staticmethod
     def get_message_type():
         return "interventionReport"

converter/converter/versions/reference/reference_converter.py

@@ -1,10 +1,9 @@
 from converter.utils import delete_paths
-from converter.versions.conversion_mixin import ConversionMixin
 from converter.versions.identical_message_converter import IdenticalMessageConverter
 from converter.versions.reference.reference_constants import ReferenceConstants
 
 
-class ReferenceConverter(IdenticalMessageConverter, ConversionMixin):
+class ReferenceConverter(IdenticalMessageConverter):
     @staticmethod
     def get_message_type():
         return "reference"