refactor(converter): use standardized modification functions in create case cisu converter

Author: SimonClo

converter/converter/cisu/create_case/create_case_cisu_constants.py

@@ -46,3 +46,37 @@ class CreateCaseCISUConstants:
     }
 
     DEFAULT_WHATS_HAPPEN = {"code": "C11.06.00", "label": "Autre nature de fait"}
+
+    DEFAULT_LOCATION_EXTERNAL_INFO_TYPE = "AUTRE"
+
+    DEFAULT_LOCATION_COUNTRY = "FR"
+
+    LOCATION_PATH = "$.location"
+    LOCATION_FREETEXT_PATH = f"{LOCATION_PATH}.freetext"
+    LOCATION_CITY_DETAIL_PATH = f"{LOCATION_PATH}.city.detail"
+    LOCATION_EXTERNAL_INFO_PATH = f"{LOCATION_PATH}.externalInfo"
+    LOCATION_EXTERNAL_INFO_TYPE_PATH = "$.type"
+    LOCATION_LOC_ID_PATH = f"{LOCATION_PATH}.locID"
+    LOCATION_COUNTRY_PATH = f"{LOCATION_PATH}.country"
+
+    INITIAL_ALERT_PATH = "$.initialAlert"
+    INITIAL_ALERT_REPORTING_PATH = f"{INITIAL_ALERT_PATH}.reporting"
+    INITIAL_ALERT_NOTES_PATH = f"{INITIAL_ALERT_PATH}.notes"
+    INITIAL_ALERT_ID_PATH = f"{INITIAL_ALERT_PATH}.id"
+    INITIAL_ALERT_CALL_TAKER_PATH = f"{INITIAL_ALERT_PATH}.callTaker"
+    INITIAL_ALERT_RECEPTION_PATH = f"{INITIAL_ALERT_PATH}.reception"
+    INITIAL_ALERT_QUALIFICATION_PATH = f"{INITIAL_ALERT_PATH}.qualification"
+    INITIAL_ALERT_LOCATION_PATH = f"{INITIAL_ALERT_PATH}.location"
+
+    QUALIFICATION_PATH = "$.qualification"
+    QUALIFICATION_VICTIMS_PATH = f"{QUALIFICATION_PATH}.victims"
+    QUALIFICATION_DETAILS_PRIORITY_PATH = f"{QUALIFICATION_PATH}.details.priority"
+    QUALIFICATION_WHATS_HAPPEN_PATH = f"{QUALIFICATION_PATH}.whatsHappen"
+
+    MEDICAL_NOTE_PATH = "$.medicalNote"
+
+    OWNER_PATH = "$.owner"
+
+    REFERENCE_VERSION_PATH = "$.referenceVersion"
+
+    CREATION_PATH = "$.creation"

converter/converter/cisu/create_case/create_case_cisu_converter.py

@@ -18,6 +18,7 @@
     get_sender,
     is_field_completed,
     translate_key_words,
+    set_value,
 )
 from converter.cisu.base_cisu_converter import BaseCISUConverter
 import logging
@@ -51,40 +52,60 @@ def from_cisu_to_rs(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
 
         def set_default_location_freetext(json_data: Dict[str, Any]):
             logger.debug("Setting default location freetext")
-            if not is_field_completed(json_data, "$.location.freetext"):
-                json_data["location"]["freetext"] = (
-                    ""  # need at least empty value to pass validation
-                )
+            if not is_field_completed(
+                json_data, CreateCaseCISUConstants.LOCATION_FREETEXT_PATH
+            ):
+                set_value(json_data, CreateCaseCISUConstants.LOCATION_FREETEXT_PATH, "")
 
         def add_location_detail(json_data: Dict[str, Any]):
             logger.debug("Adding location detail to freetext")
-            if is_field_completed(json_data, "$.location.city.detail"):
-                set_default_location_freetext(json_data)
-                json_data["location"]["freetext"] += (
-                    "\nDétails de commune : " + json_data["location"]["city"]["detail"]
-                )
+            location_city_detail = get_field_value(
+                json_data, CreateCaseCISUConstants.LOCATION_CITY_DETAIL_PATH
+            )
+            if not location_city_detail:
+                return
+
+            current_location_freetext = get_field_value(
+                json_data, CreateCaseCISUConstants.LOCATION_FREETEXT_PATH
+            )
+            updated_location_freetext = current_location_freetext + (
+                "\nDétails de commune : " + location_city_detail
+            )
+            set_value(
+                json_data,
+                CreateCaseCISUConstants.LOCATION_FREETEXT_PATH,
+                updated_location_freetext,
+            )
 
         def add_case_priority(json_data: Dict[str, Any]):
             logger.debug("Adding case priority")
-            if is_field_completed(json_data, "$.initialAlert.reporting"):
-                if not is_field_completed(json_data, "$.qualification.details"):
-                    json_data["qualification"]["details"] = {}
-                json_data["qualification"]["details"]["priority"] = (
-                    "P0"
-                    if get_field_value(json_data, "$.initialAlert.reporting")
-                    == "ATTENTION"
-                    else "P2"
-                )
+            initial_alert_reporting = get_field_value(
+                json_data, CreateCaseCISUConstants.INITIAL_ALERT_REPORTING_PATH
+            )
+            if not initial_alert_reporting:
+                return
+
+            new_priority_value = (
+                "P0" if initial_alert_reporting == "ATTENTION" else "P2"
+            )
+            set_value(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_DETAILS_PRIORITY_PATH,
+                new_priority_value,
+            )
 
         def merge_notes_freetext(json_data: Dict[str, Any]):
             logger.debug("Merging freetext notes")
-            if not is_field_completed(json_data, "$.initialAlert.notes"):
+            initial_alert_notes = get_field_value(
+                json_data, CreateCaseCISUConstants.INITIAL_ALERT_NOTES_PATH
+            )
+            if not initial_alert_notes:
                 return json_data
 
             merged_texts = []
             other_notes = []
 
-            for note in json_data["initialAlert"]["notes"]:
+            for note in initial_alert_notes:
                 if "freetext" in note:
                     merged_texts.append(note["freetext"])
                 else:
@@ -94,7 +115,11 @@ def merge_notes_freetext(json_data: Dict[str, Any]):
 
             result_notes = [merged_note] if merged_note else []
             result_notes.extend(other_notes)
-            json_data["initialAlert"]["notes"] = result_notes
+            set_value(
+                json_data,
+                CreateCaseCISUConstants.INITIAL_ALERT_NOTES_PATH,
+                result_notes,
+            )
 
             return json_data
 
@@ -104,20 +129,18 @@ def add_victims_to_medical_notes(json_data: Dict[str, Any], sender_id: str):
 
             if field_value is None:
                 return
-            else:
-                formatted_field_value = dump(field_value, allow_unicode=True)
-                translated_text = translate_key_words(
-                    formatted_field_value,
-                    CreateCaseCISUConstants.MEDICAL_NOTE_KEY_TRANSLATIONS,
-                )
-                add_object_to_medical_notes(json_data, translated_text, sender_id)
+
+            formatted_field_value = dump(field_value, allow_unicode=True)
+            translated_text = translate_key_words(
+                formatted_field_value,
+                CreateCaseCISUConstants.MEDICAL_NOTE_KEY_TRANSLATIONS,
+            )
+            add_object_to_medical_notes(json_data, translated_text, sender_id)
 
         def add_object_to_medical_notes(
             json_data: Dict[str, Any], note_text: str, sender_id: str
         ):
             logger.debug("Adding object to medical notes")
-            if not is_field_completed(json_data, "$.medicalNote"):
-                json_data["medicalNote"] = []
 
             random_str = "".join(
                 random.choices(
@@ -133,7 +156,14 @@ def add_object_to_medical_notes(
                 "operator": {"role": "AUTRE"},
             }
 
-            json_data["medicalNote"].append(new_note)
+            medical_notes = (
+                get_field_value(json_data, CreateCaseCISUConstants.MEDICAL_NOTE_PATH)
+                or []
+            )
+            medical_notes.append(new_note)
+            set_value(
+                json_data, CreateCaseCISUConstants.MEDICAL_NOTE_PATH, medical_notes
+            )
 
         # Create independent envelope copy without usecase for output
         output_json = cls.copy_cisu_input_content(input_json)
@@ -143,18 +173,21 @@ def add_object_to_medical_notes(
         output_use_case_json = cls.copy_cisu_input_use_case_content(input_json)
 
         # - Updates
-        output_use_case_json["owner"] = get_recipient(input_json)
+        set_value(
+            output_use_case_json,
+            CreateCaseCISUConstants.OWNER_PATH,
+            get_recipient(input_json),
+        )
 
         set_default_location_freetext(output_use_case_json)
         add_location_detail(output_use_case_json)
 
-        if is_field_completed(output_use_case_json, "$.initialAlert"):
-            add_case_priority(output_use_case_json)
-            add_to_initial_alert_notes(
-                output_use_case_json,
-                CreateCaseCISUConstants.CISU_PATHS_TO_ADD_TO_INITIAL_ALERT_NOTES,
-            )
-            merge_notes_freetext(output_use_case_json)
+        add_case_priority(output_use_case_json)
+        add_to_initial_alert_notes(
+            output_use_case_json,
+            CreateCaseCISUConstants.CISU_PATHS_TO_ADD_TO_INITIAL_ALERT_NOTES,
+        )
+        merge_notes_freetext(output_use_case_json)
 
         add_victims_to_medical_notes(output_use_case_json, sender_id)
 
@@ -197,10 +230,10 @@ def from_rs_to_cisu(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
 
         def add_victim_information(json_data: Dict[str, Any]):
             logger.debug("Adding victim information")
-            if not is_field_completed(json_data, "$.qualification"):
-                json_data["qualification"] = {}
-            json_data["qualification"]["victims"] = cls.get_victim_count(
-                cls, input_usecase_json
+            set_value(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_VICTIMS_PATH,
+                cls.get_victim_count(cls, input_usecase_json),
             )
 
         def get_call_taker_information(json_data: Dict[str, Any]):
@@ -216,11 +249,20 @@ def get_call_taker_information(json_data: Dict[str, Any]):
 
         def add_default_external_info_type(json_data: Dict[str, Any]):
             logger.debug("Adding default external info type")
-            external_info = get_field_value(json_data, "$.location.externalInfo")
-            if external_info is not None:
-                for info in external_info:
-                    if not is_field_completed(info, "$.type"):
-                        info["type"] = "AUTRE"
+            external_info = get_field_value(
+                json_data, CreateCaseCISUConstants.LOCATION_EXTERNAL_INFO_PATH
+            )
+            if external_info is None:
+                return
+            for info in external_info:
+                if not is_field_completed(
+                    info, CreateCaseCISUConstants.LOCATION_EXTERNAL_INFO_TYPE_PATH
+                ):
+                    set_value(
+                        info,
+                        CreateCaseCISUConstants.LOCATION_EXTERNAL_INFO_TYPE_PATH,
+                        CreateCaseCISUConstants.DEFAULT_LOCATION_EXTERNAL_INFO_TYPE,
+                    )
 
         # Create independent envelope copy without usecase for output
         output_json = cls.copy_rs_input_content(input_json)
@@ -237,19 +279,30 @@ def add_default_external_info_type(json_data: Dict[str, Any]):
 
         # - Updates
         # ToDo: pass this by ConfigMap and based on the version of the model
-        output_usecase_json["referenceVersion"] = "2.0"
+        set_value(
+            output_usecase_json, CreateCaseCISUConstants.REFERENCE_VERSION_PATH, "2.0"
+        )
         add_victim_information(output_usecase_json)
 
-        if not is_field_completed(output_usecase_json, "$.location"):
-            output_usecase_json["location"] = {}
-
-        output_usecase_json["location"]["locID"] = f"LOC-{timestamp}-{random_str}"
+        set_value(
+            output_usecase_json,
+            CreateCaseCISUConstants.LOCATION_LOC_ID_PATH,
+            f"LOC-{timestamp}-{random_str}",
+        )
         # ToDo: get country from INSEE code | Ref.: https://www.insee.fr/fr/information/7766585#titre-bloc-25
-        output_usecase_json["location"]["country"] = "FR"  # Default value
+        set_value(
+            output_usecase_json,
+            CreateCaseCISUConstants.LOCATION_COUNTRY_PATH,
+            CreateCaseCISUConstants.DEFAULT_LOCATION_COUNTRY,
+        )
 
-        if not is_field_completed(output_usecase_json, "$.qualification.whatsHappen"):
-            output_usecase_json["qualification"]["whatsHappen"] = (
-                CreateCaseCISUConstants.DEFAULT_WHATS_HAPPEN
+        if not is_field_completed(
+            output_usecase_json, CreateCaseCISUConstants.QUALIFICATION_WHATS_HAPPEN_PATH
+        ):
+            set_value(
+                output_usecase_json,
+                CreateCaseCISUConstants.QUALIFICATION_WHATS_HAPPEN_PATH,
+                CreateCaseCISUConstants.DEFAULT_WHATS_HAPPEN,
             )
 
         add_default_external_info_type(output_usecase_json)
@@ -260,27 +313,57 @@ def add_default_external_info_type(json_data: Dict[str, Any]):
             output_usecase_json, CreateCaseCISUConstants.HEALTH_PATHS_TO_DELETE
         )
 
-        if is_field_completed(input_usecase_json, "$.initialAlert"):
-            output_usecase_json["initialAlert"]["id"] = f"INAL-{timestamp}-{random_str}"
-            output_usecase_json["initialAlert"]["callTaker"] = (
-                get_call_taker_information(input_json)
+        if is_field_completed(
+            input_usecase_json, CreateCaseCISUConstants.INITIAL_ALERT_PATH
+        ):
+            set_value(
+                output_usecase_json,
+                CreateCaseCISUConstants.INITIAL_ALERT_ID_PATH,
+                f"INAL-{timestamp}-{random_str}",
             )
-            output_usecase_json["initialAlert"]["reception"] = get_field_value(
-                input_usecase_json, "$.creation"
+
+            set_value(
+                output_usecase_json,
+                CreateCaseCISUConstants.INITIAL_ALERT_CALL_TAKER_PATH,
+                get_call_taker_information(input_json),
+            )
+
+            set_value(
+                output_usecase_json,
+                CreateCaseCISUConstants.INITIAL_ALERT_RECEPTION_PATH,
+                get_field_value(
+                    input_usecase_json, CreateCaseCISUConstants.CREATION_PATH
+                ),
             )
-            output_usecase_json["initialAlert"]["reporting"] = (
+
+            new_initial_alert_reporting = (
                 "ATTENTION"
                 if get_field_value(
                     input_usecase_json, "$.qualification.details.priority"
                 )
                 in ["P0", "P1"]
                 else "STANDARD"
             )
-            output_usecase_json["initialAlert"]["qualification"] = copy.deepcopy(
-                get_field_value(output_usecase_json, "$.qualification")
+            set_value(
+                output_usecase_json,
+                CreateCaseCISUConstants.INITIAL_ALERT_REPORTING_PATH,
+                new_initial_alert_reporting,
+            )
+
+            set_value(
+                output_usecase_json,
+                CreateCaseCISUConstants.INITIAL_ALERT_QUALIFICATION_PATH,
+                get_field_value(
+                    output_usecase_json, CreateCaseCISUConstants.QUALIFICATION_PATH
+                ),
             )
-            output_usecase_json["initialAlert"]["location"] = copy.deepcopy(
-                get_field_value(output_usecase_json, "$.location")
+
+            set_value(
+                output_usecase_json,
+                CreateCaseCISUConstants.INITIAL_ALERT_LOCATION_PATH,
+                get_field_value(
+                    output_usecase_json, CreateCaseCISUConstants.LOCATION_PATH
+                ),
             )
 
         return cls.format_cisu_output_json(output_json, output_usecase_json)