VED-241 SNOMED converter logic and validation (currently disabled)

Author: robertnovac1

delta_backend/.coveragerc

@@ -0,0 +1,3 @@
+[report]
+omit =
+    tests/*

delta_backend/Makefile

@@ -8,4 +8,16 @@ package: build
 test:
 	python -m unittest
 
+check-conversion:
+	python tests/check_conversion.py
+
+coverage-run:
+	coverage run -m unittest discover -v
+
+coverage-report:
+	coverage report -m 
+
+coverage-html:
+	coverage html  
+
 .PHONY: build package 

delta_backend/poetry.lock

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

delta_backend/pyproject.toml

@@ -153,11 +153,11 @@
       }
     },
     {
-      "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",
-        "expressionType": "NOTEMPTY",
+        "expressionType": "SNOMED",
         "expressionRule": ""
       }
     },
@@ -184,7 +184,7 @@
       "fieldNameFlat": "VACCINE_PRODUCT_CODE",
       "expression": {
         "expressionName": "Not Empty",
-        "expressionType": "NOTEMPTY",
+        "expressionType": "SNOMED",
         "expressionRule": ""
       }
     },
@@ -229,7 +229,7 @@
       "fieldNameFlat": "SITE_OF_VACCINATION_CODE",
       "expression": {
         "expressionName": "Not Empty",
-        "expressionType": "NOTEMPTY",
+        "expressionType": "SNOMED",
         "expressionRule": ""
       }
     },
@@ -247,7 +247,7 @@
       "fieldNameFlat": "ROUTE_OF_VACCINATION_CODE",
       "expression": {
         "expressionName": "Not Empty",
-        "expressionType": "NOTEMPTY",
+        "expressionType": "SNOMED",
         "expressionRule": ""
       }
     },
@@ -292,7 +292,7 @@
       "fieldNameFlat": "INDICATION_CODE",
       "expression": {
         "expressionName": "Not Empty",
-        "expressionType": "NOTEMPTY",
+        "expressionType": "SNOMED",
         "expressionRule": ""
       }
     },

delta_backend/src/ConversionLayout.py

@@ -51,7 +51,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"]
@@ -60,7 +60,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,26 @@ 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):
+        """
+        Applies expression rules for filtering key-value pairs during searches.
+
+        This method provides a flexible foundation for implementing various filtering 
+        or validation rules, enabling more dynamic and configurable search behavior. 
+        While it currently supports only SNOMED code validation, the structure opens 
+        the door to applying a wide range of expression rules in the future.
+
+        For example, when processing a list of items, this method helps determine 
+        which item(s) satisfy specific criteria based on the logic defined by the 
+        expression type and rule.
+        """
+        if expression_type == "SNOMED" and expression_rule == "validate-code":
+            if key_value_pair.get("code"):
+                return is_valid_simple_snomed(key_value_pair["code"])
+        
+        return True
+        
     # scan for a key name or a value
     def _scanValuesForMatch(self, parent, matchValue):
         try:
@@ -21,18 +41,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 +77,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: str = "", 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,20 @@
+
+from stdnum.verhoeff import validate
+
+def is_valid_simple_snomed(simple_snomed: str) -> bool:
+    """
+    This utility is designed for reuse and should be packaged as part of a 
+    shared validation module or service.
+    """
+    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",