Merge pull request #376 from NHSDigital/VED-163-FHIR-FLAT-JSON

VED-163-fhir to flat json date conversion

Author: Akol125

delta_backend/.coverage

@@ -0,0 +1,2 @@
+output.json
+output.csv

delta_backend/.gitignore

@@ -5,4 +5,7 @@ package: build
 	mkdir -p build
 	docker run --rm -v $(shell pwd)/build:/build delta-lambda-build
 
+test:
+	python -m unittest
+
 .PHONY: build package 

delta_backend/Makefile

@@ -0,0 +1,71 @@
+# 🩺 FHIR to Flat JSON Conversion Engine
+
+This project is designed to convert FHIR-compliant JSON data (e.g., Immunization records) into a flat JSON format based on a configurable schema layout. It is intended to support synchronization of Immunisation API generated data from external sources to DPS (Data Processing System) data system
+
+---
+
+## 📁 File Structure Overview
+
+| File Name              | What It Does |
+|------------------------|---------------|
+| **`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). |
+| **`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. |
+
+---
+
+
+## 🛠️ Key Features
+
+- Schema-driven field extraction and formatting
+- Support for custom date formats like `YYYYMMDD`, and CSV-safe UTC timestamps
+- Robust handling of patient, practitioner, and address data using time-aware logic
+- Extendable structure with static helper methods and modular architecture
+
+---
+
+## 📦 Example Use Case
+
+- Input: FHIR `Immunization` resource (with nested fields)
+- Output: Flat JSON object with 34 standardized key-value pairs
+- Purpose: To export into CSV or push into downstream ETL systems
+
+---
+
+## ✅ Getting Started with `check_conversion.py`
+
+To quickly test your conversion, use the provided `check_conversion.py` script.
+This script loads sample FHIR data, runs it through the converter, and automatically saves the output in both JSON and CSV formats.
+
+### 🔄 How to Use It
+
+1. Add your FHIR data (e.g., a dictionary or sample JSON) into the `fhir_sample` variable inside `check_conversion.py`
+2. Ensure the field mapping in `ConversionLayout.py` matches your desired output
+3. Run the script from the `tests` folder:
+
+```bash
+python check_conversion.py
+```
+
+### 📁 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:
+
+```
+/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
+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/README.md

@@ -11,6 +11,9 @@ boto3 = "~1.26.90"
 mypy-boto3-dynamodb = "^1.26.164"
 moto = "~4.2.11"
 
+[tool.poetry.group.dev.dependencies]
+coverage = "^7.8.0"
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

delta_backend/__init__.py

@@ -1,13 +1,15 @@
+
+# Handles the transformation logic for each field based on the schema
 # Root and base type expression checker functions
 import ExceptionMessages
-import datetime
-import uuid
+from datetime import datetime,timedelta
+from zoneinfo import ZoneInfo
 import re
 from LookUpData import LookUpData
 
 
 # --------------------------------------------------------------------------------------------------------
-# record exception capture
+# Custom error type to handle validation failures
 class RecordError(Exception):
 
     def __init__(self, code=None, message=None, details=None):
@@ -24,6 +26,7 @@ def __repr__(self):
 
 # ---------------------------------------------------------------------------------------------------------
 # main conversion checker
+# Conversion engine for expression-based field transformation
 class ConversionChecker:
     # checker settings
     summarise = False
@@ -37,13 +40,17 @@ def __init__(self, dataParser, summarise, report_unexpected_exception):
         self.summarise = summarise  # instance attribute
         self.report_unexpected_exception = report_unexpected_exception  # instance attribute
 
-    # exposed functions
+    # Main entry point called by converter.py
     def convertData(self, expressionType, expressionRule, fieldName, fieldValue):
         match expressionType:
             case "DATECONVERT":
                 return self._convertToDate(
                     expressionRule, fieldName, fieldValue, self.summarise, self.report_unexpected_exception
                 )
+            case "DATETIME":
+                return self._convertToDateTime(
+                    expressionRule, fieldName, fieldValue, self.summarise, self.report_unexpected_exception
+                )
             case "NOTEMPTY":
                 return self._convertToNotEmpty(
                     expressionRule, fieldName, fieldValue, self.summarise, self.report_unexpected_exception
@@ -75,15 +82,71 @@ def convertData(self, expressionType, expressionRule, fieldName, fieldValue):
             case _:
                 return "Schema expression not found! Check your expression type : " + expressionType
 
-    # iso8086 date time validate
+    # Convert ISO date string to a specific format (e.g. YYYYMMDD)
     def _convertToDate(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):
+        if not fieldValue:
+            return ""
+
+        if not isinstance(fieldValue, str):
+            raise RecordError(
+                ExceptionMessages.RECORD_CHECK_FAILED,
+                f"{fieldName} rejected: not a string.",
+                f"Received: {type(fieldValue)}",
+        )
+        # 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}",
+            )
         try:
-            convertDate = datetime.datetime.fromisoformat(fieldValue)
-            return convertDate.strftime(expressionRule)
-        except Exception as e:
+            dt = datetime.fromisoformat(fieldValue)
+            format_str = expressionRule.replace("format:", "")
+            return dt.strftime(format_str)
+        except ValueError:
             if report_unexpected_exception:
-                message = ExceptionMessages.MESSAGES[ExceptionMessages.UNEXPECTED_EXCEPTION] % (e.__class__.__name__, e)
-                return message
+                return f"Unexpected format: {fieldValue}"
+
+    # Convert FHIR datetime into CSV-safe UTC format
+    def _convertToDateTime(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):
+        if not 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 datetime not accepted.",
+                f"Invalid partial datetime: {fieldValue}",
+            )
+        try:
+            dt = datetime.fromisoformat(fieldValue)
+        except ValueError:
+            if report_unexpected_exception:
+                return f"Unexpected format: {fieldValue}"
+
+        # Allow only +00:00 or +01:00 offsets (UTC and BST) and reject unsupported timezones
+        offset = dt.utcoffset()
+        allowed_offsets = [ZoneInfo("UTC").utcoffset(dt),
+                           ZoneInfo("Europe/London").utcoffset(dt)]
+        if offset not in allowed_offsets:
+            raise RecordError(
+                ExceptionMessages.RECORD_CHECK_FAILED,
+                f"{fieldName} rejected: unsupported timezone.",
+                f"Unsupported offset: {offset}",
+            )
+
+        # Convert to UTC
+        dt_utc = dt.astimezone(ZoneInfo("UTC")).replace(microsecond=0)
+
+        format_str = expressionRule.replace("format:", "")
+
+        if format_str == "csv-utc":
+            formatted = dt_utc.strftime("%Y%m%dT%H%M%S%z")
+            return formatted.replace("+0000", "00").replace("+0100", "01")
+
+        return dt_utc.strftime(format_str)
 
     # Not Empty Validate
     def _convertToNotEmpty(self, expressionRule, fieldName, fieldValue, summarise, report_unexpected_exception):

delta_backend/poetry.lock

@@ -1,6 +1,6 @@
 
-#This is the base layout for converting from FHIR to Flat JSON
-#See the readme for an explanation of how this works
+# This file holds the schema/base layout that maps FHIR fields to flat JSON fields
+# Each entry tells the converter how to extract and transform a specific value
 
 ConvertLayout = {
   "id": "7d78e9a6-d859-45d3-bb05-df9c405acbdb",
@@ -67,8 +67,8 @@
       "fieldNameFlat": "DATE_AND_TIME",
       "expression": {
         "expressionName": "Date Convert",
-        "expressionType": "DATECONVERT",
-        "expressionRule": "%Y%m%dT%H%M%S"
+        "expressionType": "DATETIME",
+        "expressionRule": "format:csv-utc"
       }
     },
     {
@@ -140,7 +140,7 @@
       "expression": {
         "expressionName": "Date Convert",
         "expressionType": "DATECONVERT",
-        "expressionRule": "%Y%m%d"
+        "expressionRule": "format:%Y%m%d"
       }
     },
     {
@@ -221,7 +221,7 @@
       "expression": {
         "expressionName": "Date Convert",
         "expressionType": "DATECONVERT",
-        "expressionRule": "%Y%m%d"
+        "expressionRule": "format:%Y%m%d"
       }
     },
     {

delta_backend/pyproject.toml

@@ -29,13 +29,14 @@ def __init__(self, fhir_data):
         self.FHIRData = fhir_data  # Store JSON data directly
         self.SchemaFile = ConversionLayout.ConvertLayout
 
-    # create a FHIR  parser - uses fhir json data from delta
+    # create a FHIR  parser - uses fhir json data from delta 
+    # (helper methods to extract values from the nested FHIR structure)
     def _getFHIRParser(self, fhir_data):
         fhirParser = FHIRParser()
         fhirParser.parseFHIRData(fhir_data)
         return fhirParser
 
-    # create a schema parser
+    # create a schema parser - parses the schema that defines how FHIR fields should be mapped into flat fields.
     def _getSchemaParser(self, schemafile):
         schemaParser = SchemaParser()
         schemaParser.parseSchema(schemafile)
@@ -120,7 +121,7 @@ def extract_patient_details(self, json_data, FlatFieldName):
             self._cached_values = {}
 
         if not self._cached_values:
-            occurrence_time = datetime.strptime(json_data.get("occurrenceDateTime", ""), "%Y-%m-%dT%H:%M:%S%z")
+            occurrence_time = datetime.fromisoformat(json_data.get("occurrenceDateTime", ""))
             patient = get_patient(json_data)
             if not patient:
                 return None