Updated converter to handle urls. Added snomed validation

Author: robertnovac1

delta_backend/poetry.lock

@@ -13,6 +13,7 @@ python = "~3.10"
 boto3 = "~1.26.90"
 mypy-boto3-dynamodb = "^1.26.164"
 moto = "~4.2.11"
+python-stdnum = "^1.20"
 
 [tool.poetry.group.dev.dependencies]
 coverage = "^7.8.0"

delta_backend/pyproject.toml

@@ -153,7 +153,7 @@
       }
     },
     {
-      "fieldNameFHIR": "extension|0|valueCodeableConcept|coding|0|code",
+      "fieldNameFHIR": "extension|#:https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure|valueCodeableConcept|coding|#:http://snomed.info/sct|code",
       "fieldNameFlat": "VACCINATION_PROCEDURE_CODE",
       "expression": {
         "expressionName": "Not Empty",
@@ -185,7 +185,7 @@
       "expression": {
         "expressionName": "Not Empty",
         "expressionType": "SNOMED",
-        "expressionRule": ""
+        "expressionRule": "validate-code"
       }
     },
     {

delta_backend/src/ConversionLayout.py

@@ -52,7 +52,7 @@ def _getSchemaParser(self, schemafile):
         return schemaParser
 
     # Convert data against converter schema
-    def _convertData(self, ConversionValidate, expression, dataParser, json_data):
+    def _convertData(self, ConversionValidate: ConversionChecker, expression, dataParser: FHIRParser, json_data):
 
         FHIRFieldName = expression["fieldNameFHIR"]
         FlatFieldName = expression["fieldNameFlat"]
@@ -61,7 +61,7 @@ def _convertData(self, ConversionValidate, expression, dataParser, json_data):
         expressionRule = expression["expression"]["expressionRule"]
 
         try:
-            conversionValues = dataParser.getKeyValue(FHIRFieldName)
+            conversionValues = dataParser.getKeyValue(FHIRFieldName, expressionType, expressionRule)
         except Exception as e:
             message = "Data get value Unexpected exception [%s]: %s" % (e.__class__.__name__, e)
             error = self._log_error(message, code=ExceptionMessages.PARSING_ERROR)

delta_backend/src/Converter.py

@@ -1,6 +1,6 @@
 # FHIR JSON importer and data access
 import json
-
+from utils import is_valid_simple_snomed
 
 class FHIRParser:
     # parser variables
@@ -10,6 +10,16 @@ class FHIRParser:
     def parseFHIRData(self, fhirData):
         self.FHIRFile = json.loads(fhirData) if isinstance(fhirData, str) else fhirData
 
+    def _validate_expression_rule(self, expression_type, expression_rule, key_value_pair):
+        if not expression_rule:
+            return True
+        
+        elif expression_type == "SNOMED" and expression_rule == "validate-code":
+            if key_value_pair.get("code"):
+                return is_valid_simple_snomed(key_value_pair["code"])
+        
+        return False
+        
     # scan for a key name or a value
     def _scanValuesForMatch(self, parent, matchValue):
         try:
@@ -21,18 +31,21 @@ def _scanValuesForMatch(self, parent, matchValue):
             return False
 
     # locate an index for an item in a list
-    def _locateListId(self, parent, locator):
-        fieldList = locator.split(":")
+    def _locateListId(self, parent, locator, expression_type, expression_rule: str = ""):
+        fieldList = locator.split(":", 1)
         nodeId = 0
         index = 0
         try:
             while index < len(parent):
-                for key in parent[index]:
-                    if (parent[index][key] == fieldList[1]) or (key == fieldList[1]):
+                for key, value in parent[index].items():
+                    if (
+                        (value == fieldList[1] or key == fieldList[1])
+                        and self._validate_expression_rule(expression_type, expression_rule, parent[index])
+                    ):
                         nodeId = index
                         break
                     else:
-                        if self._scanValuesForMatch(parent[index][key], fieldList[1]):
+                        if self._scanValuesForMatch(value, fieldList[1]):
                             nodeId = index
                             break
                 index += 1
@@ -54,26 +67,26 @@ def _getNode(self, parent, child):
         return result
 
     # locate a value for a key
-    def _scanForValue(self, FHIRFields):
+    def _scanForValue(self, FHIRFields, expression_type, expression_rule: str = ""):
         fieldList = FHIRFields.split("|")
         # get root field before we iterate
         rootfield = self.FHIRFile[fieldList[0]]
         del fieldList[0]
         try:
             for field in fieldList:
                 if field.startswith("#"):
-                    rootfield = self._locateListId(rootfield, field)  # check here for default index??
+                    rootfield = self._locateListId(rootfield, field, expression_type, expression_rule)  # check here for default index??
                 else:
                     rootfield = self._getNode(rootfield, field)
         except:
             rootfield = ""
         return rootfield
 
     # get the value for a key
-    def getKeyValue(self, fieldName):
+    def getKeyValue(self, fieldName, expression_type, expression_rule: str = ""):
         value = []
         try:
-            responseValue = self._scanForValue(fieldName)
+            responseValue = self._scanForValue(fieldName, expression_type, expression_rule)
         except:
             responseValue = ""
 

delta_backend/src/FHIRParser.py

@@ -0,0 +1,17 @@
+
+from stdnum.verhoeff import validate
+
+def is_valid_simple_snomed(simple_snomed: str) -> bool:
+    "check the snomed code valid or not."
+    min_snomed_length = 6
+    max_snomed_length = 18
+    try: 
+        return (
+            simple_snomed is not None
+            and simple_snomed.isdigit()
+            and min_snomed_length <= len(simple_snomed) <= max_snomed_length
+            and validate(simple_snomed)
+            and (simple_snomed[-3:-1] in ("00", "10"))
+        )
+    except:
+        return False

delta_backend/src/utils.py

@@ -43,15 +43,25 @@
     {
       "url": "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure",
       "valueCodeableConcept": {
-        "coding": [
-          {
-            "system": "http://snomed.info/sct",
-            "code": "13246810000001",
-            "display": "Administration of first dose of severe acute respiratory syndrome coronavirus 2 vaccine (procedure)"
-          }
-        ]
+          "coding": [
+              {
+                  "code": "956951000000105",
+                  "display": "Seasonal influenza vaccination (procedure)",
+                  "system": "http://snomed.info/test"
+              },
+              {
+                  "code": "956951000000104",
+                  "display": "Seasonal influenza vaccination (procedure)",
+                  "system": "http://snomed.info/sct"
+              },
+              {
+                  "code": "NEG",
+                  "display": "Seasonal influenza vaccination (procedure)",
+                  "system": "https://acme.lab/resultcodes"
+              }
+          ]
+        }
       }
-    }
   ],
   "identifier": [
     {
@@ -62,6 +72,11 @@
   "status": "completed",
   "vaccineCode": {
     "coding": [
+      {
+        "system": "http://snomed.info/sct",
+        "code": "39114911000001104",
+        "display": "COVID-19 Vaccine Vaxzevria (ChAdOx1 S [recombinant]) not less than 2.5x100,000,000 infectious units/0.5ml dose suspension for injection multidose vials (AstraZeneca UK Ltd) (product)"
+      },
       {
         "system": "http://snomed.info/sct",
         "code": "39114911000001105",