feat(converter): refacto cisu_converter to use util functions

converter/converter/cisu_converter.py

@@ -3,7 +3,7 @@
 import random
 import string
 from datetime import datetime
-from .utils import add_object_to_initial_alert_notes, delete_paths, format_object, get_recipient, get_sender
+from .utils import add_to_initial_alert_notes, delete_paths, get_field_value, get_recipient, get_sender, is_field_completed
 
 class CISUConverterV3:
     """Handles CISU format conversions"""
@@ -41,6 +41,14 @@ class CISUConverterV3:
         "location.geometry.point.isAml"
     ]
 
+    CISU_PATHS_TO_ADD_TO_INITIAL_ALERT_NOTES =[
+        '$.qualification.victims',
+        '$.initialAlert.attachment',
+        '$.initialAlert.callTaker',
+        '$.freetext',
+        '$.newAlert'
+    ]
+
     @classmethod
     def from_cisu(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
         """
@@ -52,57 +60,42 @@ def from_cisu(cls, input_json: Dict[str, Any]) -> Dict[str, Any]:
         Returns:
             Converted EDXL Health JSON
         """
+        def add_location_detail(json_data: Dict[str,Any]):
+            if is_field_completed(json_data,'$.location.city.detail'):
+                if not is_field_completed(json_data, '$.location.freetext'):
+                    json_data['location']['freetext']=''
+                json_data['location']['freetext']+= " Détails de commune : " + json_data['location']['city']['detail']
+
+        def add_case_priority(json_data: Dict[str,Any]):
+            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'
 
         # 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']
 
-        # Create independent usecase copy for output
-        input_usecase_json = input_json['content'][0]['jsonContent']['embeddedJsonContent']['message']['createCase']
-        output_usecase_json = copy.deepcopy(input_usecase_json)
+        # Create independent use case copy for output
+        input_use_case_json = input_json['content'][0]['jsonContent']['embeddedJsonContent']['message']['createCase']
+        output_use_case_json = copy.deepcopy(input_use_case_json)
 
         # - Updates
         # Set owner to target recipient
-        output_usecase_json['owner'] = get_recipient(input_json)
-
-        # Handle victims information
-        if input_usecase_json.get('qualification', {}).get('victims'):
-            victims_text = format_object(input_usecase_json['qualification']['victims'])
-            add_object_to_initial_alert_notes(output_usecase_json, victims_text)
-
-        # Location
-        if input_usecase_json.get('location', {}).get('city', {}).get('detail'):
-            output_usecase_json['location']['freetext']=input_usecase_json['location']['city']['detail']
-
-        # Qualification - case detail priority
-        if input_usecase_json.get('initialAlert', {}).get('reporting'):
-            if 'details' not in output_usecase_json['qualification']:
-                output_usecase_json['qualification']['details']={}
-            output_usecase_json['qualification']['details']['priority']= 'P0' if input_usecase_json.get('caseDetails', {}).get('priority') =='ATTENTION' else 'P2'
-
-        # Initial Alert - free text information
-        if input_usecase_json.get('initialAlert', {}).get('attachment'):
-            attachments = format_object(input_usecase_json['initialAlert']['attachment'])
-            add_object_to_initial_alert_notes(output_usecase_json,attachments)
-
-        if input_usecase_json.get('initialAlert', {}).get('callTaker'):
-            call_taker = format_object(input_usecase_json['initialAlert']['callTaker'])
-            add_object_to_initial_alert_notes(output_usecase_json, call_taker)
+        output_use_case_json['owner'] = get_recipient(input_json)
 
-        if input_usecase_json.get('initialAlert', {}).get('freetext'):
-            description = format_object(input_usecase_json['initialAlert']['freetext'])
-            add_object_to_initial_alert_notes(output_usecase_json, description)
+        add_location_detail(output_use_case_json)
 
-        if input_usecase_json.get('newAlert'):
-            new_alert = format_object(input_usecase_json.get('newAlert'))
-            add_object_to_initial_alert_notes(output_usecase_json, new_alert)
+        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,cls.CISU_PATHS_TO_ADD_TO_INITIAL_ALERT_NOTES)
 
-        # - Deletions - must to be the last step
-        delete_paths(output_usecase_json, cls.CISU_PATHS_TO_DELETE)
+        # - 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_usecase_json
+        output_json['content'][0]['jsonContent']['embeddedJsonContent']['message']['createCaseHealth'] = output_use_case_json
         return output_json
 
     @classmethod

converter/converter/utils.py

@@ -1,5 +1,8 @@
 import re
 from typing import List, Dict, Any
+from jsonpath_ng import parse
+from yaml import dump
+
 
 def get_recipient(edxl_json: Dict[str, Any]) -> str:
     return edxl_json['descriptor']['explicitAddress']['explicitAddressValue']
@@ -88,8 +91,50 @@ def format_object(obj: Any, indent: int = 0) -> str:
     return f"{indent_str}{obj}"
 
 
-def add_object_to_initial_alert_notes(output_json, note_text):
-    if 'notes' not in output_json['initialAlert']:
-        output_json['initialAlert']['notes'] = []
+def add_object_to_initial_alert_notes(json_data: Dict[str, Any], note_text: str):
+    if not is_field_completed(json_data, '$.initialAlert.notes'):
+        json_data['initialAlert']['notes'] = []
+
+    json_data['initialAlert']['notes'].append({"freetext": note_text})
+
+
+def is_field_completed(json_data: Dict[str, Any], json_path:str):
+    try:
+        jsonpath_expr = parse(json_path)
+        return len(jsonpath_expr.find(json_data))>=1
+    except Exception as e:
+        print(f"Error raised in is_field_completed : {e}")
+        raise
+
+def get_field_value(json_data: Dict[str, Any], json_path: str):
+    try:
+        isCompleted = is_field_completed(json_data, json_path)
+
+        if not isCompleted:
+            return None
+
+        jsonpath_expr = parse(json_path)
+        matches = jsonpath_expr.find(json_data)
+
+        if len(matches) > 1:
+            return [match.value for match in matches]
+        return matches[0].value
 
-    output_json['initialAlert']['notes'].append({"freetext": note_text})
+    except Exception as e:
+        print(f"Error raised in is_field_completed : {e}")
+        raise
+
+
+def add_field_to_initial_alert_notes(data: Dict[str, Any], json_path: str):
+    field_value = get_field_value(data,json_path)
+
+    if field_value == None:
+        return
+
+    formatted_field_value = dump(field_value)
+    add_object_to_initial_alert_notes(data, formatted_field_value)
+
+
+def add_to_initial_alert_notes(data: Dict[str, Any], paths: List[str]):
+    for path in paths:
+        add_field_to_initial_alert_notes(data, path)

converter/tests/fixtures/json_data_fixture.json

@@ -0,0 +1,38 @@
+{
+  "caseId": "fr.health.samu770.DRFR154878900236",
+  "qualification": {
+    "whatsHappen": { "code": "C09.03.00", "label": "Fuite de gaz" },
+    "locationKind": {
+      "code": "L01.01.01",
+      "label": "Maison particulière, pavillon, à l'intérieur"
+    },
+    "riskThreat": [
+      { "code": "R13", "label": "Risque d'explosion, présence de gaz" },
+      { "code": "R12", "label": "Risque d'incendie" }
+    ],
+    "healthMotive": {
+      "code": "M03.10",
+      "label": "Malaise avec perte de connaissance initiale"
+    },
+    "details": { "priority": "P1" }
+  },
+  "location": {
+    "detailedAddress": {
+      "complete": "Rue des Marins",
+      "wayName": {
+        "complete": "Rue des Marins",
+        "type": "Rue",
+        "name": "Des Marins"
+      },
+      "number": "3"
+    },
+    "city": { "name": "Bouaye", "inseeCode": "44018" },
+    "freetext": "Maison aux volets verts",
+    "geometry": {
+      "point": {
+        "coord": { "lat": 47.147005, "lon": -1.690548, "precision": "ADRESSE" }
+      },
+      "datetime": "2024-05-18T18:17:00+02:00"
+    }
+  }
+}

converter/tests/test_utils.py

@@ -1,5 +1,13 @@
 import pytest
-from converter.utils import add_object_to_initial_alert_notes, format_object, delete_paths
+from converter.utils import add_object_to_initial_alert_notes, get_field_value, is_field_completed, format_object, delete_paths
+import unittest
+import json
+import os
+
+def load_json_file(filename):
+    file_path = os.path.join(os.path.dirname(__file__), filename)
+    with open(file_path, "r", encoding="utf-8") as file:
+        return json.load(file)
 
 class ExampleTestVictim:
     def __init__(self, count: str, condition: str):
@@ -82,6 +90,7 @@ def test_add_note_to_existing_notes():
     add_object_to_initial_alert_notes(output_json, note_text)
 
     assert {"freetext": "New note"} in output_json['initialAlert']['notes']
+    assert {"freetext": "Existing note"} in output_json['initialAlert']['notes']
     assert len(output_json['initialAlert']['notes']) == 2
 
 def test_add_note_to_empty_notes():
@@ -96,3 +105,81 @@ def test_add_note_to_empty_notes():
 
     assert {"freetext": "New note"} in output_json['initialAlert']['notes']
     assert len(output_json['initialAlert']['notes']) == 1
+
+class TestIsFieldCompleted(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.json_data = load_json_file("./fixtures/json_data_fixture.json")
+
+    def test_existing_path(self):
+         self.assertTrue(is_field_completed(self.json_data, "$.caseId"))
+         self.assertTrue(is_field_completed(self.json_data, "$.location.detailedAddress.wayName.type"))
+         self.assertTrue(is_field_completed(self.json_data, "$.qualification.riskThreat[0].label"))
+         self.assertTrue(is_field_completed(self.json_data, "$.qualification.riskThreat[0]"))
+         self.assertTrue(is_field_completed(self.json_data, "$.qualification.riskThreat"))
+         self.assertTrue(is_field_completed(self.json_data, "$.qualification"))
+         self.assertTrue(is_field_completed(self.json_data, "$.qualification.healthMotive"))
+
+    def test_non_existing_path(self):
+        self.assertFalse(is_field_completed(self.json_data, "$.caseId.name"))
+        self.assertFalse(is_field_completed(self.json_data, "$.name"))
+        self.assertFalse(is_field_completed(self.json_data, "$.location.detailedAddress.city.id"))
+
+    def test_empty_json_data(self):
+        self.assertFalse(is_field_completed({}, "$.caseId"))
+
+    def test_invalid_jsonpath(self):
+        with self.assertRaises(Exception):
+            is_field_completed(self.json_data, "$..")
+        with self.assertRaises(Exception):
+            is_field_completed(self.json_data, "$toto")
+
+class TestGetFieldValue(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.json_data = load_json_file("./fixtures/json_data_fixture.json")
+
+    def test_existing_path(self):
+        self.assertEqual(get_field_value(self.json_data, "$.caseId"), "fr.health.samu770.DRFR154878900236")
+        self.assertEqual(get_field_value(self.json_data, "$.location.detailedAddress.wayName.type"), "Rue")
+        self.assertEqual(get_field_value(self.json_data, "$.qualification.riskThreat[0].label"), "Risque d'explosion, présence de gaz")
+        self.assertEqual(get_field_value(self.json_data, "$.qualification.riskThreat[1].label"), "Risque d'incendie")
+        self.assertEqual(get_field_value(self.json_data, "$.qualification.riskThreat[0]"), { "code": "R13", "label": "Risque d'explosion, présence de gaz" })
+        self.assertEqual(get_field_value(self.json_data, "$.qualification.riskThreat[*]"),[
+                { "code": "R13", "label": "Risque d'explosion, présence de gaz" },
+                { "code": "R12", "label": "Risque d'incendie" }
+                ])
+        self.assertEqual(get_field_value(self.json_data, "$.qualification"), {
+                "whatsHappen": { "code": "C09.03.00", "label": "Fuite de gaz" },
+                "locationKind": {
+                "code": "L01.01.01",
+                "label": "Maison particulière, pavillon, à l'intérieur"
+                },
+                "riskThreat": [
+                { "code": "R13", "label": "Risque d'explosion, présence de gaz" },
+                { "code": "R12", "label": "Risque d'incendie" }
+                ],
+                "healthMotive": {
+                "code": "M03.10",
+                "label": "Malaise avec perte de connaissance initiale"
+                },
+                "details": { "priority": "P1" }
+            })
+
+    def test_non_existing_path(self):
+        self.assertIsNone(get_field_value(self.json_data, "$.caseId.name"))
+        self.assertIsNone(get_field_value(self.json_data, "$.name"))
+        self.assertIsNone(get_field_value(self.json_data, "$.location.detailedAddress.city.id"))
+
+    def test_empty_json_data(self):
+        self.assertIsNone(get_field_value({}, "$.caseId"))
+
+    def test_invalid_jsonpath(self):
+        with self.assertRaises(Exception):
+            get_field_value(self.json_data, "$..")
+        with self.assertRaises(Exception):
+            get_field_value(self.json_data, "$toto")
+
+
+if __name__ == "__main__":
+    unittest.main()