Add config option to define own location-type coding system url

importer/README.md

@@ -72,6 +72,7 @@ The coverage report `coverage.html` will be at the working directory
 - The seventh and eighth columns are the location's type and typeCode, respectively
 - The ninth column is the administrative level, that shows the hierarchical level of the location. Root location would have a `level 0` and all child locations will have a level `parent_admin_level + 1`
 - The tenth and eleventh columns are the location's physicalType and physicalTypeCode, respectively
+- You can pass in `--location_type_coding_system` to define your own location type coding system url (not required)
 
 ### 2. Create users in bulk
 - Run `python3 main.py --csv_file csv/users.csv --resource_type users --log_level info`

importer/json_payloads/locations_payload.json

@@ -23,7 +23,7 @@
             {
                 "coding": [
                     {
-                        "system": "http://terminology.hl7.org/CodeSystem/location-type",
+                        "system": "$t_system",
                         "code": "$t_code",
                         "display": "$t_display"
                     }

importer/main.py

@@ -286,6 +286,8 @@ def identify_coding_object_index(array, current_system):
         list_of_systems = value["coding"][0]["system"]
         if current_system in list_of_systems:
             return index
+        else:
+            return -1
 
 
 def check_parent_admin_level(locationParentId):
@@ -309,7 +311,7 @@ def check_parent_admin_level(locationParentId):
 
 
 # custom extras for locations
-def location_extras(resource, payload_string):
+def location_extras(resource, payload_string, location_coding_system):
     try:
         (
             locationName,
@@ -355,7 +357,8 @@ def location_extras(resource, payload_string):
         payload_string = json.dumps(obj, indent=4)
 
     try:
-        if locationType and locationType != "type":
+        payload_string = payload_string.replace("$t_system", location_coding_system)
+        if len(locationType.strip()) > 0 and locationType != "type":
             payload_string = payload_string.replace("$t_display", locationType)
         if locationTypeCode and locationTypeCode != "typeCode":
             payload_string = payload_string.replace("$t_code", locationTypeCode)
@@ -1028,7 +1031,7 @@ def check_for_nulls(resource: list) -> list:
 
 # This function builds a json payload
 # which is posted to the api to create resources
-def build_payload(resource_type, resources, resource_payload_file, created_resources=None):
+def build_payload(resource_type, resources, resource_payload_file, created_resources=None, location_coding_system=None):
     logging.info("Building request payload")
     initial_string = """{"resourceType": "Bundle","type": "transaction","entry": [ """
     final_string = group_type = " "
@@ -1090,7 +1093,7 @@ def build_payload(resource_type, resources, resource_payload_file, created_resou
             if resource_type == "organizations":
                 ps = organization_extras(resource, ps)
             elif resource_type == "locations":
-                ps = location_extras(resource, ps)
+                ps = location_extras(resource, ps, location_coding_system)
             elif resource_type == "careTeams":
                 ps = care_team_extras(resource, ps, "orgs & users")
             elif resource_type == "Group":
@@ -1862,6 +1865,10 @@ def filter(self, record):
     required=False,
     default="DIRECT",
 )
+@click.option(
+    "--location_type_coding_system",
+    required=False,
+    default="http://terminology.hl7.org/CodeSystem/location-type")
 def main(
     csv_file,
     json_file,
@@ -1884,6 +1891,7 @@ def main(
     resources_count,
     list_resource_id,
     sync,
+    location_type_coding_system,
 ):
     if log_level == "DEBUG":
         logging.basicConfig(
@@ -1955,7 +1963,8 @@ def main(
         elif resource_type == "locations":
             logging.info("Processing locations")
             json_payload = build_payload(
-                "locations", resource_list, "json_payloads/locations_payload.json"
+                "locations", resource_list, "json_payloads/locations_payload.json", None,
+                location_type_coding_system
             )
             final_response = handle_request("POST", json_payload, config.fhir_base_url)
             logging.info("Processing complete!")

importer/test_main.py

@@ -136,7 +136,8 @@ def test_build_payload_locations(
         csv_file = "csv/locations/locations_full.csv"
         resource_list = read_csv(csv_file)
         payload = build_payload(
-            "locations", resource_list, "json_payloads/locations_payload.json"
+            "locations", resource_list, "json_payloads/locations_payload.json",
+            None, "http://terminology.hl7.org/CodeSystem/location-type"
         )
         payload_obj = json.loads(payload)
         self.assertIsInstance(payload_obj, dict)
@@ -225,7 +226,8 @@ def test_build_payload_locations(
         csv_file = "csv/locations/locations_min.csv"
         resource_list = read_csv(csv_file)
         payload = build_payload(
-            "locations", resource_list, "json_payloads/locations_payload.json"
+            "locations", resource_list, "json_payloads/locations_payload.json",
+            None, "http://terminology.hl7.org/CodeSystem/location-type"
         )
         payload_obj = json.loads(payload)
         self.assertIsInstance(payload_obj, dict)
@@ -612,7 +614,8 @@ def test_uuid_generated_for_locations_is_unique_and_repeatable(self):
         ]
 
         payload = build_payload(
-            "locations", resources, "json_payloads/locations_payload.json"
+            "locations", resources, "json_payloads/locations_payload.json",
+            None, "http://terminology.hl7.org/CodeSystem/location-type"
         )
         payload_obj = json.loads(payload)
         location1 = payload_obj["entry"][0]["resource"]["id"]
@@ -1521,6 +1524,22 @@ def test_build_resource_type_map(self):
         self.assertIsInstance(mapping, dict)
         self.assertEqual(mapping, mapped_resources)
 
+    @patch("main.check_parent_admin_level")
+    @patch("main.get_resource")
+    def test_define_own_location_type_coding_system_url(self, mock_get_resource, mock_check_parent_admin_level):
+        mock_get_resource.return_value = "1"
+        mock_check_parent_admin_level.return_value = "3"
+        test_system_code = "http://terminology.hl7.org/CodeSystem/test_location-type"
+
+        csv_file = "csv/locations/locations_full.csv"
+        resource_list = read_csv(csv_file)
+        payload = build_payload(
+            "locations", resource_list, "json_payloads/locations_payload.json",
+            None, test_system_code
+        )
+        payload_obj = json.loads(payload)
+        self.assertEqual(payload_obj["entry"][0]["resource"]["type"][0]["coding"][0]["system"], test_system_code)
+
 
 if __name__ == "__main__":
     unittest.main()