Merge pull request #452 from ansforge/feat/converter/nomenclatures/add-nomenclature-conversion

[ Converter / Nomenclatures ] Conversion de nomenclatures EDA

Author: bou3108

converter/converter/cisu/create_case/create_case_cisu_constants.py

@@ -74,6 +74,9 @@ class CreateCaseCISUConstants:
     QUALIFICATION_VICTIM_COUNT_PATH = f"{QUALIFICATION_VICTIMS_PATH}.count"
     QUALIFICATION_DETAILS_PRIORITY_PATH = f"{QUALIFICATION_PATH}.details.priority"
     QUALIFICATION_WHATS_HAPPEN_PATH = f"{QUALIFICATION_PATH}.whatsHappen"
+    QUALIFICATION_HEALTH_MOTIVE_PATH = f"{QUALIFICATION_PATH}.healthMotive"
+    QUALIFICATION_RISK_THREAT_PATH = f"{QUALIFICATION_PATH}.riskThreat"
+    QUALIFICATION_LOCATION_KIND_PATH = f"{QUALIFICATION_PATH}.locationKind"
 
     MEDICAL_NOTE_PATH = "$.medicalNote"
 

converter/converter/cisu/create_case/create_case_cisu_converter.py

@@ -24,6 +24,34 @@
     set_value,
 )
 from converter.cisu.base_cisu_converter import BaseCISUConverter
+from converter.nomenclatures.utils import (
+    apply_nomenclature_mapping,
+    apply_nomenclature_mapping_to_list,
+)
+from converter.nomenclatures.from_cisu_to_rs.whats_happen import (
+    MAP as CISU_TO_RS_WHATS_HAPPEN_MAP,
+)
+from converter.nomenclatures.from_cisu_to_rs.health_motive import (
+    MAP as CISU_TO_RS_HEALTH_MOTIVE_MAP,
+)
+from converter.nomenclatures.from_cisu_to_rs.risk_threat import (
+    MAP as CISU_TO_RS_RISK_THREAT_MAP,
+)
+from converter.nomenclatures.from_cisu_to_rs.location_kind import (
+    MAP as CISU_TO_RS_LOCATION_KIND_MAP,
+)
+from converter.nomenclatures.from_rs_to_cisu.whats_happen import (
+    MAP as RS_TO_CISU_WHATS_HAPPEN_MAP,
+)
+from converter.nomenclatures.from_rs_to_cisu.health_motive import (
+    MAP as RS_TO_CISU_HEALTH_MOTIVE_MAP,
+)
+from converter.nomenclatures.from_rs_to_cisu.risk_threat import (
+    MAP as RS_TO_CISU_RISK_THREAT_MAP,
+)
+from converter.nomenclatures.from_rs_to_cisu.location_kind import (
+    MAP as RS_TO_CISU_LOCATION_KIND_MAP,
+)
 import logging
 
 logger = logging.getLogger(__name__)
@@ -283,6 +311,29 @@ def add_call_taker_org_to_custom_map(json_data: Dict[str, Any]):
 
             return json_data
 
+        def apply_qualification_nomenclature_mappings(json_data: Dict[str, Any]):
+            logger.debug("Applying qualification nomenclature mappings")
+            apply_nomenclature_mapping(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_WHATS_HAPPEN_PATH,
+                CISU_TO_RS_WHATS_HAPPEN_MAP,
+            )
+            apply_nomenclature_mapping(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_HEALTH_MOTIVE_PATH,
+                CISU_TO_RS_HEALTH_MOTIVE_MAP,
+            )
+            apply_nomenclature_mapping_to_list(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_RISK_THREAT_PATH,
+                CISU_TO_RS_RISK_THREAT_MAP,
+            )
+            apply_nomenclature_mapping(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_LOCATION_KIND_PATH,
+                CISU_TO_RS_LOCATION_KIND_MAP,
+            )
+
         # Create independent envelope copy without usecase for output
         output_json = cls.copy_cisu_input_content(input_json)
 
@@ -300,6 +351,8 @@ def add_call_taker_org_to_custom_map(json_data: Dict[str, Any]):
         set_default_location_freetext(output_use_case_json)
         add_location_detail(output_use_case_json)
 
+        apply_qualification_nomenclature_mappings(output_use_case_json)
+
         add_case_priority(output_use_case_json)
         add_to_initial_alert_notes(
             output_use_case_json,
@@ -388,6 +441,29 @@ def add_default_external_info_type(json_data: Dict[str, Any]):
                         CreateCaseCISUConstants.DEFAULT_LOCATION_EXTERNAL_INFO_TYPE,
                     )
 
+        def apply_qualification_nomenclature_mappings(json_data: Dict[str, Any]):
+            logger.debug("Applying qualification nomenclature mappings")
+            apply_nomenclature_mapping(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_WHATS_HAPPEN_PATH,
+                RS_TO_CISU_WHATS_HAPPEN_MAP,
+            )
+            apply_nomenclature_mapping(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_HEALTH_MOTIVE_PATH,
+                RS_TO_CISU_HEALTH_MOTIVE_MAP,
+            )
+            apply_nomenclature_mapping_to_list(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_RISK_THREAT_PATH,
+                RS_TO_CISU_RISK_THREAT_MAP,
+            )
+            apply_nomenclature_mapping(
+                json_data,
+                CreateCaseCISUConstants.QUALIFICATION_LOCATION_KIND_PATH,
+                RS_TO_CISU_LOCATION_KIND_MAP,
+            )
+
         # Create independent envelope copy without usecase for output
         output_json = cls.copy_rs_input_content(input_json)
 
@@ -420,6 +496,8 @@ def add_default_external_info_type(json_data: Dict[str, Any]):
             CreateCaseCISUConstants.DEFAULT_LOCATION_COUNTRY,
         )
 
+        apply_qualification_nomenclature_mappings(output_usecase_json)
+
         if not is_field_completed(
             output_usecase_json, CreateCaseCISUConstants.QUALIFICATION_WHATS_HAPPEN_PATH
         ):

converter/converter/nomenclatures/utils.py

@@ -0,0 +1,60 @@
+from typing import Any, Dict, Optional
+
+from converter.utils import delete_paths, get_field_value, set_value
+
+
+def apply_nomenclature_mapping(
+    json_data: Dict[str, Any],
+    field_path: str,
+    mapping: Dict[str, Optional[Dict[str, str]]],
+) -> None:
+    """
+    Looks up json_data[field_path].code in the given nomenclature mapping and
+    updates json_data accordingly:
+      - no entry for the code: json_data is left untouched
+      - entry maps to a dict: the field at field_path is replaced by that dict
+      - entry maps to None: the field at field_path is removed
+    """
+    code = get_field_value(json_data, f"{field_path}.code")
+    if code is None or code not in mapping:
+        return
+
+    new_value = mapping[code]
+    if new_value is None:
+        delete_paths(json_data, [field_path])
+    else:
+        set_value(json_data, field_path, new_value)
+
+
+def apply_nomenclature_mapping_to_list(
+    json_data: Dict[str, Any],
+    field_path: str,
+    mapping: Dict[str, Optional[Dict[str, str]]],
+) -> None:
+    """
+    Same semantics as apply_nomenclature_mapping, applied item by item to the
+    array field at field_path:
+      - an item's code has no entry in mapping: the item is left untouched
+      - entry maps to a dict: the item is replaced by that dict
+      - entry maps to None: the item is removed from the array
+    If the array ends up empty, the field itself is removed.
+    """
+    items = get_field_value(json_data, field_path)
+    if not isinstance(items, list):
+        return
+
+    updated_items = []
+    for item in items:
+        code = item.get("code") if isinstance(item, dict) else None
+        if code is None or code not in mapping:
+            updated_items.append(item)
+            continue
+
+        new_value = mapping[code]
+        if new_value is not None:
+            updated_items.append(new_value)
+
+    if updated_items:
+        set_value(json_data, field_path, updated_items)
+    else:
+        delete_paths(json_data, [field_path])

converter/tests/cisu/test_qualification_nomenclature_mapping.py

@@ -0,0 +1,165 @@
+"""
+Integration tests exercising CreateCaseCISUConverter.from_cisu_to_rs / from_rs_to_cisu
+end-to-end against real nomenclature map entries, for the qualification.whatsHappen,
+qualification.healthMotive, qualification.riskThreat and qualification.locationKind
+fields. Complements the unit tests in tests/nomenclatures/test_utils.py (which cover
+apply_nomenclature_mapping/apply_nomenclature_mapping_to_list in isolation) by proving
+the maps are actually wired into the converter and that real fixture-shaped messages
+produce the expected replace/remove behaviour.
+"""
+
+import copy
+
+from converter.cisu.create_case.create_case_cisu_converter import (
+    CreateCaseCISUConverter,
+)
+from tests.cisu.helpers import get_edxl_message
+from tests.constants import TestConstants
+from tests.test_helpers import TestHelper
+
+RC_EDA_BASE = TestHelper.load_json_file(
+    "tests/fixtures/RC-EDA/RC-EDA_required_fields.json"
+)
+RS_EDA_BASE = TestHelper.load_json_file(
+    "tests/fixtures/RS-EDA/cisu_case/RS-EDA_required_fields.json"
+)
+
+
+def cisu_to_rs_message(qualification_overrides: dict) -> dict:
+    content = copy.deepcopy(RC_EDA_BASE)
+    content["createCase"]["qualification"].update(qualification_overrides)
+    edxl = TestHelper.combine_edxl_envelope_and_message(
+        TestConstants.EDXL_FIRE_TO_HEALTH_ENVELOPE_PATH, content
+    )
+    result = CreateCaseCISUConverter.from_cisu_to_rs(edxl)
+    return get_edxl_message(result)["createCaseHealth"]["qualification"]
+
+
+def rs_to_cisu_message(qualification_overrides: dict) -> dict:
+    content = copy.deepcopy(RS_EDA_BASE)
+    content["createCaseHealth"]["qualification"].update(qualification_overrides)
+    edxl = TestHelper.combine_edxl_envelope_and_message(
+        TestConstants.EDXL_HEALTH_TO_FIRE_ENVELOPE_PATH, content
+    )
+    result = CreateCaseCISUConverter.from_rs_to_cisu(edxl)
+    return get_edxl_message(result)["createCase"]["qualification"]
+
+
+# --- from_cisu_to_rs: code has no entry in the map -> field untouched ---
+
+
+def test_from_cisu_to_rs_whats_happen_unmapped_code_is_untouched():
+    unmapped = {"code": "C99.99.99", "label": "Code inconnu"}
+    qualification = cisu_to_rs_message({"whatsHappen": unmapped})
+    assert qualification["whatsHappen"] == unmapped
+
+
+# --- from_cisu_to_rs: code has a non-None entry -> field replaced ---
+
+
+def test_from_cisu_to_rs_whats_happen_mapped_code_is_replaced():
+    qualification = cisu_to_rs_message(
+        {"whatsHappen": {"code": "C02.05.03", "label": "Menace de suicide"}}
+    )
+    assert qualification["whatsHappen"] == {
+        "code": "C02.18.00",
+        "label": "Problème psychiatrique, menace de suicide",
+    }
+
+
+def test_from_cisu_to_rs_health_motive_mapped_code_is_replaced():
+    qualification = cisu_to_rs_message(
+        {"healthMotive": {"code": "M20.00", "label": "Inconscience"}}
+    )
+    assert qualification["healthMotive"] == {
+        "code": "M08.00.00",
+        "label": "Personne inconsciente ",
+    }
+
+
+def test_from_cisu_to_rs_location_kind_mapped_code_is_replaced():
+    qualification = cisu_to_rs_message(
+        {
+            "locationKind": {
+                "code": "L02.02.04",
+                "label": "Aire de repos sur voie rapide",
+            }
+        }
+    )
+    assert qualification["locationKind"] == {
+        "code": "L02.02.03",
+        "label": "Aire de repos ou de service sur voie rapide",
+    }
+
+
+# --- from_cisu_to_rs: riskThreat is an array -> per-item behaviour ---
+
+
+def test_from_cisu_to_rs_risk_threat_mapped_to_none_removes_item_and_field():
+    qualification = cisu_to_rs_message(
+        {"riskThreat": [{"code": "R36", "label": "Menace"}]}
+    )
+    assert "riskThreat" not in qualification
+
+
+def test_from_cisu_to_rs_risk_threat_removes_only_the_none_mapped_item():
+    unmapped = {"code": "R99", "label": "Autre menace"}
+    qualification = cisu_to_rs_message(
+        {"riskThreat": [{"code": "R36", "label": "Menace"}, unmapped]}
+    )
+    assert qualification["riskThreat"] == [unmapped]
+
+
+# --- from_rs_to_cisu: same three cases, using the reverse maps ---
+
+
+def test_from_rs_to_cisu_whats_happen_unmapped_code_is_untouched():
+    unmapped = {"code": "C99.99.99", "label": "Code inconnu"}
+    qualification = rs_to_cisu_message({"whatsHappen": unmapped})
+    assert qualification["whatsHappen"] == unmapped
+
+
+def test_from_rs_to_cisu_whats_happen_mapped_code_is_replaced():
+    qualification = rs_to_cisu_message(
+        {"whatsHappen": {"code": "C02.17.00", "label": "Mort suspecte/cadavre"}}
+    )
+    assert qualification["whatsHappen"] == {
+        "code": "C10.02.01",
+        "label": "Mort suspecte",
+    }
+
+
+def test_from_rs_to_cisu_health_motive_mapped_code_is_replaced():
+    qualification = rs_to_cisu_message(
+        {"healthMotive": {"code": "M08.00.00", "label": "Personne inconsciente"}}
+    )
+    assert qualification["healthMotive"] == {
+        "code": "M22.00",
+        "label": "Personne inconsciente",
+    }
+
+
+def test_from_rs_to_cisu_location_kind_mapped_code_is_replaced():
+    qualification = rs_to_cisu_message(
+        {"locationKind": {"code": "L01.01.05", "label": "Pavillon"}}
+    )
+    assert qualification["locationKind"] == {
+        "code": "L01.01.00",
+        "label": "Maison particulière, pavillon",
+    }
+
+
+def test_from_rs_to_cisu_risk_threat_mapped_code_replaces_matching_item_only():
+    unmapped = {"code": "R99", "label": "Autre menace"}
+    qualification = rs_to_cisu_message(
+        {
+            "riskThreat": [
+                {"code": "R17", "label": "Situation instable"},
+                unmapped,
+            ]
+        }
+    )
+    assert qualification["riskThreat"] == [
+        {"code": "R37", "label": "Situation instable/indéterminée"},
+        unmapped,
+    ]