diff --git a/converter/converter/cisu_converter.py b/converter/converter/cisu_converter.py index e8713e185..da39ab8a6 100644 --- a/converter/converter/cisu_converter.py +++ b/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,6 +60,17 @@ 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) @@ -59,50 +78,24 @@ def from_cisu(cls, input_json: Dict[str, Any]) -> Dict[str, Any]: 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 diff --git a/converter/converter/utils.py b/converter/converter/utils.py index 66dcd6db5..cf105bb7d 100644 --- a/converter/converter/utils.py +++ b/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) diff --git a/converter/tests/fixtures/json_data_fixture.json b/converter/tests/fixtures/json_data_fixture.json new file mode 100644 index 000000000..8245275b6 --- /dev/null +++ b/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" + } + } +} diff --git a/converter/tests/test_utils.py b/converter/tests/test_utils.py index 1b84515ae..350bc5c7f 100644 --- a/converter/tests/test_utils.py +++ b/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()