VED-166-FHIR-FLAT-JSON2

delta_backend/README.md

@@ -10,11 +10,11 @@ This project is designed to convert FHIR-compliant JSON data (e.g., Immunization
 |------------------------|---------------|
 | **`converter.py`**     | 🧠 The main brain — applies the schema, runs conversions, handles errors. |
 | **`FHIRParser.py`**    | 🪜 Knows how to dig into nested FHIR structures and pull out values like dates, IDs, and patient names. |
-| **`SchemaParser.py`**  | 📐 Reads your schema layout and tells the converter which FHIR fields to extract and how to rename/format them. |
-| **`ConversionLayout.py`** | ✍️ A plain Python list that defines which fields you want, and how they should be formatted (e.g. date format, renaming rules). |
+| **`SchemaParser.py`**  | Reads your schema layout and tells the converter which FHIR fields to extract and how to rename/format them. |
+| **`ConversionLayout.py`** | A plain Python list that defines which fields you want, and how they should be formatted (e.g. date format, renaming rules). |
 | **`ConversionChecker.py`** | 🔧 Handles transformation logic — e.g. turning a FHIR datetime into `YYYY-MM-DD`, applying lookups, gender codes, defaults, etc. |
-| **`Extractor.py`**     | 🎣 Specialized logic to pull practitioner names, site codes, addresses, and apply time-aware rules. |
-| **`ExceptionMessages.py`** | 🚨 Holds reusable error messages and codes for clean debugging and validation feedback. |
+| **`Extractor.py`**     | Specialized logic to pull practitioner names, site codes, addresses, and apply time-aware rules. |
+| **`ExceptionMessages.py`** | Holds reusable error messages and codes for clean debugging and validation feedback. |
 
 ---
 
@@ -28,7 +28,7 @@ This project is designed to convert FHIR-compliant JSON data (e.g., Immunization
 
 ---
 
-## 📦 Example Use Case
+## Example Use Case
 
 - Input: FHIR `Immunization` resource (with nested fields)
 - Output: Flat JSON object with 34 standardized key-value pairs
@@ -51,19 +51,19 @@ This script loads sample FHIR data, runs it through the converter, and automatic
 python check_conversion.py
 ```
 
-### 📁 Output Location
+### Output Location
 When the script runs, it will automatically:
 - Save a **flat JSON file** as `output.json`
 - Save a **CSV file** as `output.csv`
 
-These will be located one level up from the `src/` folder:
+These will be located one level up from the `tests/` folder:
 
 ```
 /mnt/c/Users/USER/desktop/shn/immunisation-fhir-api/delta_backend/output.json
 /mnt/c/Users/USER/desktop/shn/immunisation-fhir-api/delta_backend/output.csv
 ```
 
-### 👀 Visualization
+### Visualization
 You can now:
 - Open `output.csv` in Excel or Google Sheets to view cleanly structured records
 - Inspect `output.json` to validate the flat key-value output programmatically

delta_backend/src/ConversionChecker.py

@@ -39,6 +39,7 @@ def __init__(self, dataParser, summarise, report_unexpected_exception):
         self.dataLookUp = LookUpData()  # used for generic look up
         self.summarise = summarise  # instance attribute
         self.report_unexpected_exception = report_unexpected_exception  # instance attribute
+        self.errorRecords = []  # Store all errors here
 
     # Main entry point called by converter.py
     def convertData(self, expressionType, expressionRule, fieldName, fieldValue):
@@ -55,6 +56,10 @@ def convertData(self, expressionType, expressionRule, fieldName, fieldValue):
                 return self._convertToNotEmpty(
                     expressionRule, fieldName, fieldValue, self.summarise, self.report_unexpected_exception
                 )
+            case "DOSESEQUENCE":
+                return self._convertToDose(
+                    expressionRule, fieldName, fieldValue, self.summarise, self.report_unexpected_exception
+                )
             case "GENDER":
                 return self._convertToGender(
                     expressionRule, fieldName, fieldValue, self.summarise, self.report_unexpected_exception
@@ -88,18 +93,10 @@ def _convertToDate(self, expressionRule, fieldName, fieldValue, summarise, repor
             return ""
 
         if not isinstance(fieldValue, str):
-            raise RecordError(
-                ExceptionMessages.RECORD_CHECK_FAILED,
-                f"{fieldName} rejected: not a string.",
-                f"Received: {type(fieldValue)}",
-        )
+            return ""
         # Reject partial dates like "2024" or "2024-05"
         if re.match(r"^\d{4}(-\d{2})?$", fieldValue):
-            raise RecordError(
-                ExceptionMessages.RECORD_CHECK_FAILED,
-                f"{fieldName} rejected: partial date not accepted.",
-                f"Invalid partial date: {fieldValue}",
-            )
+            return ""
         try:
             dt = datetime.fromisoformat(fieldValue)
             format_str = expressionRule.replace("format:", "")
@@ -148,7 +145,7 @@ def _convertToDateTime(self, expressionRule, fieldName, fieldValue, summarise, r
 
         return dt_utc.strftime(format_str)
 
-    # Not Empty Validate
+    # Not Empty Validate - Returns exactly what is in the extracted fields no parsing or logic needed
     def _convertToNotEmpty(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):
         try:
             if len(str(fieldValue)) > 0:
@@ -161,39 +158,70 @@ def _convertToNotEmpty(self, expressionRule, fieldName, fieldValue, summarise, r
 
     # NHSNumber Validate
     def _convertToNHSNumber(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):
+        """
+        Validates that the NHS Number is exactly 10 digits.
+        """
+        # If it is outright empty, return back an empty string
+        if not fieldValue:
+            return ""
+
         try:
-            regexRule = "^6[0-9]{10}$"
-            result = re.search(regexRule, fieldValue)
-            if not result:
-                raise RecordError(
-                    ExceptionMessages.RECORD_CHECK_FAILED,
-                    "NHS Number check failed",
-                    "NHS Number does not meet regex rules, data- " + fieldValue,
-                )
+            regexRule = r"^\d{10}$"
+            if isinstance(fieldValue, str) and re.fullmatch(regexRule, fieldValue):
+                return fieldValue
+            raise ValueError(f"NHS Number must be exactly 10 digits: {fieldValue}")
         except Exception as e:
             if report_unexpected_exception:
                 message = ExceptionMessages.MESSAGES[ExceptionMessages.UNEXPECTED_EXCEPTION] % (e.__class__.__name__, e)
-                return message
+                self.errorRecords.append({
+                "field": fieldName,
+                "value": fieldValue,
+                "message": message
+            })
+        return ""
 
     # Gender Validate
     def _convertToGender(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):
+        """
+        Converts gender string to numeric representation.
+        Mapping:
+            - "male" → "1"
+            - "female" → "2"
+            - "other" → "9"
+            - "unknown" → "0"
+        """
         try:
-            genderlist = {"male": "1", "female": "2", "other": "9", "unknown": "0"}
-            genderNumber = genderlist[fieldValue]
-            return genderNumber
+            gender_map = {
+                "male": "1",
+                "female": "2",
+                "other": "9",
+                "unknown": "0"
+            }
+
+            # Normalize input
+            normalized_gender = str(fieldValue).lower()
+
+            if normalized_gender not in gender_map:
+                return ""
+            return gender_map[normalized_gender]
+
         except Exception as e:
             if report_unexpected_exception:
-                message = ExceptionMessages.MESSAGES[ExceptionMessages.UNEXPECTED_EXCEPTION] % (e.__class__.__name__, e)
-                return message
+                return f"Unexpected exception [{e.__class__.__name__}]: {str(e)}"
 
-    # Change to Validate
+    # Code for converting Action Flag
     def _convertToChangeTo(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):
         try:
             return expressionRule
         except Exception as e:
             if report_unexpected_exception:
                 message = ExceptionMessages.MESSAGES[ExceptionMessages.UNEXPECTED_EXCEPTION] % (e.__class__.__name__, e)
                 return message
+    # Code for converting Dose Sequence
+    def _convertToDose(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):
+        if isinstance(fieldValue, (int, float)) and 1 <= fieldValue <= 9:
+            return fieldValue
+        return "" 
 
     # Change to Lookup
     def _convertToLookUp(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):

delta_backend/src/ConversionLayout.py

@@ -12,8 +12,8 @@
       "fieldNameFHIR": "contained|#:Patient|identifier|#:https://fhir.nhs.uk/Id/nhs-number|value",
       "fieldNameFlat": "NHS_NUMBER",
       "expression": {
-        "expressionName": "Not Empty",
-        "expressionType": "NOTEMPTY",
+        "expressionName": "NHS NUMBER",
+        "expressionType": "NHSNUMBER",
         "expressionRule": ""
       }
     },
@@ -175,7 +175,7 @@
       "fieldNameFlat": "DOSE_SEQUENCE",
       "expression": {
         "expressionName": "Not Empty",
-        "expressionType": "NOTEMPTY",
+        "expressionType": "DOSESEQUENCE",
         "expressionRule": ""
       }
     },

delta_backend/tests/check_conversion.py

@@ -9,6 +9,7 @@
 # Sample FHIR Immunization resource (minimal test data)
 fhir_sample = os.path.join(os.path.dirname(__file__),"sample_data", "fhir_sample.json")
 
+
 with open(fhir_sample, "r", encoding="utf-8") as f:
     json_data = json.load(f)
 

delta_backend/tests/sample_data/fhir_sample.json

@@ -27,11 +27,14 @@
           "given": ["Sarah"]
         }
       ],
-      "gender": "unknown",
+      "gender": "other",
       "birthDate": "1965-02-28",
       "address": [
         {
-          "postalCode": "EC1A 1BB"
+          "use": "home",
+          "line": ["123 High Street"],
+          "city": "London",
+          "postalCode": "SW1A 1AA"
         }
       ]
     }
@@ -147,7 +150,7 @@
           ]
         }
       ],
-      "doseNumberPositiveInt": 1
+      "doseNumberPositiveInt": 2
     }
   ]
 }

delta_backend/tests/test_convert_to_flat_json.py

@@ -312,13 +312,21 @@ def test_convert_to_nhs_number(self, MockLookUpData):
         dataParser = Mock()
 
         checker = ConversionChecker(dataParser, summarise=False, report_unexpected_exception=True)
+
+         # Test empty NHS number
+        empty_nhs_number = ""
+        result = checker._convertToNHSNumber(None, "fieldName", empty_nhs_number, False, True)
+        self.assertEqual(result, "", "Expected empty string for empty NHS number input")
 
+        # Test valid NHS number
         valid_nhs_number = "6000000000"
-        result = checker._convertToNHSNumber(None, "fieldName", valid_nhs_number, False, True)
-        self.assertTrue("NHS Number does not meet regex " in result)
+        result = checker._convertToNHSNumber("NHSNUMBER", "fieldName", valid_nhs_number, False, True)
+        self.assertEqual(result, "6000000000", "Valid NHS number should be returned as-is")
 
-        invalid_nhs_number = "1234567890"
-        result = checker._convertToNHSNumber(None, "fieldName", invalid_nhs_number, False, True)
+        # Test invalid NHS number
+        invalid_nhs_number = "1234567890243"
+        result = checker._convertToNHSNumber("NHSNUMBER","fieldName", invalid_nhs_number, False, True)
+        self.assertEqual(result, "", "Invalid NHS number should return empty string")
 
     @patch("ConversionChecker.LookUpData")
     def test_convert_to_date(self, MockLookUpData):
@@ -360,6 +368,25 @@ def test_convert_to_date_time(self, MockLookUpData):
         result = checker._convertToDateTime("format:%Y%m%dT%H%M%S", "fieldName", "", False, True)
         self.assertEqual(result, "")
 
+    #check for dose sequence
+    @patch("ConversionChecker.LookUpData")
+    def test_convert_to_dose(self, MockLookUpData):
+        dataParser = Mock()
+
+        checker = ConversionChecker(dataParser, summarise=False, report_unexpected_exception=True)
+        # Valid dose
+        for dose in [1, 4, 9]:
+            with self.subTest(dose=dose):
+                result = checker._convertToDose("DOSESEQUENCE", "DOSE_AMOUNT", dose, False, True)
+                self.assertEqual(result, dose)
+
+        # Invalid dose
+        invalid_doses = [10, 10.1, 100, 9.0001]
+        for dose in invalid_doses:
+            with self.subTest(dose=dose):
+                result = checker._convertToDose("DOSESEQUENCE", "DOSE_AMOUNT", dose, False, True)
+                self.assertEqual(result, "", f"Expected empty string for invalid dose {dose}")
+
     def clear_table(self):
         scan = self.table.scan()
         with self.table.batch_writer() as batch: