Merge pull request #29 from NHSDigital/AMB-1689-Map-batch-CSV-file-to-immunisation-object

 AMB-1689-Map-batch-CSV-file-to-immunisation-object

Author: almas-liaqat

lambdas/batch_processing/batch_processing.py

@@ -0,0 +1,56 @@
+import json
+from time import time
+from boto3 import resource
+from boto3 import client
+from lambdas.batch_processing.csv_to_model import read_csv_to_immunizations
+from lambdas.batch_processing.model_to_fhir import convert_to_fhir
+
+
+# This is yet to be integrated with main logic
+def lambda_handler(_event, _context):
+    s3_client = client("s3")
+    records = _event.get("Records", [])
+    s3_event = records[0].get("s3", {})
+    source_bucket_name = s3_event.get("bucket", {}).get("name")
+    bucket_key_name = s3_event.get("object", {}).get("key")
+
+    dest_bucket_name = source_bucket_name.replace("source", "destination")
+    output_bucket = resource("s3").Bucket(dest_bucket_name)
+
+    # Read the contents of the CSV file from S3
+    try:
+        response = s3_client.get_object(Bucket=source_bucket_name, Key=bucket_key_name)
+        csv_data = response["Body"].read().decode("utf-8")
+
+        immunizations = read_csv_to_immunizations(csv_data)
+
+        for immunization in immunizations:
+
+            fhir_imms = convert_to_fhir(immunization)
+
+            expected_json = json.loads(fhir_imms.get_immunization())
+            expected_json["patient"] = json.loads(fhir_imms.get_patient())
+            expected_json["reportOrigin"] = json.loads(fhir_imms.get_report_origin())
+            expected_json["reasonCode"] = json.loads(fhir_imms.get_reason_code())
+            expected_json["recorded"] = fhir_imms.get_recorded()
+            expected_json["manufacturer"] = json.loads(fhir_imms.get_manufacturer())
+            expected_json["performer"] = fhir_imms.get_actor()
+            # Add "resourceType" to the "location" element
+            expected_json["location"]["resourceType"] = "Location"
+            print(json.dumps(expected_json))
+
+        # Write some placeholder bytestring data to a file in the bucket
+        filename = f"output_report_{int(time())}.txt"
+        data = (
+            f"Test file to see if the lambda writes to the correct S3 bucket. "
+            f"This was the name of the original file: {bucket_key_name}. "
+            f"content of file: {immunizations}. "
+            f"If our AWS bill skyrockets, this file has been written to the wrong bucket!"
+        )
+
+        output_bucket.put_object(Body=data, Key=filename)
+
+        return {"statusCode": 200}
+    except Exception as e:
+        print(f"Error reading CSV file: {str(e)}")
+        return {"statusCode": 500, "body": "Error reading CSV file from S3."}

lambdas/batch_processing/csv_to_model.py

@@ -0,0 +1,42 @@
+import csv
+from models.immunization import ImmunizationModel
+import datetime
+
+
+def parse_iso8601_datetime(iso_datetime_str):
+    try:
+        # Parse ISO 8601 datetime string with timezone offset
+        dt = datetime.fromisoformat(iso_datetime_str)
+
+        # Convert the parsed datetime to the system's local timezone
+        dt = dt.astimezone()
+
+        return dt
+    except ValueError:
+        return None
+
+
+# Define a function to read and transform the CSV file
+def read_csv_to_immunizations(csv_data):
+    immunizations = [] = []
+
+    csv_reader = csv.DictReader(csv_data.splitlines())
+    for row in csv_reader:
+        # Parse datetime fields if needed
+        if row["PERSON_DOB"]:
+            row["PERSON_DOB"] = datetime.strptime(row["PERSON_DOB"], "%Y-%m-%d").date()
+        if row["DATE_AND_TIME"]:
+            row["DATE_AND_TIME"] = parse_iso8601_datetime(row["DATE_AND_TIME"])
+        if row["RECORDED_DATE"]:
+            row["RECORDED_DATE"] = datetime.strptime(
+                row["RECORDED_DATE"], "%Y-%m-%d"
+            ).date()
+        if row["EXPIRY_DATE"]:
+            row["EXPIRY_DATE"] = datetime.strptime(
+                row["EXPIRY_DATE"], "%Y-%m-%d"
+            ).date()
+        immunization = ImmunizationModel(**row)
+        if immunization is not None:
+            immunizations.append(immunization)
+
+    return immunizations