Fhir Resources and Data generator

healthchain/data_generator/base_generators.py

@@ -0,0 +1,179 @@
+# generators.py
+
+import random
+import string
+
+from healthchain.fhir_resources.base_resources import (
+    booleanModel,
+    canonicalModel,
+    codeModel,
+    dateModel,
+    dateTimeModel,
+    decimalModel,
+    idModel,
+    instantModel,
+    integerModel,
+    markdownModel,
+    positiveIntModel,
+    stringModel,
+    timeModel,
+    unsignedIntModel,
+    uriModel,
+    urlModel,
+    uuidModel,
+)
+
+from faker import Faker
+
+faker = Faker()
+
+
+class Registry:
+    def __init__(self):
+        self.registry = {}
+
+    def register(self, cls):
+        self.registry[cls.__name__] = cls
+        return cls
+
+    def get(self, name):
+        if name not in self.registry:
+            raise ValueError(f"No generator registered for '{name}'")
+        return self.registry.get(name)
+
+
+generator_registry = Registry()
+
+
+def register_generator(cls):
+    generator_registry.register(cls)
+    return cls
+
+
+@register_generator
+class BaseGenerator:
+    @staticmethod
+    def generate():
+        raise NotImplementedError("Each generator must implement a 'generate' method.")
+
+
+@register_generator
+class booleanGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return booleanModel(random.choice(["true", "false"]))
+
+
+@register_generator
+class canonicalGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return canonicalModel(f"https://example/{faker.uri_path()}")
+
+
+@register_generator
+class codeGenerator(BaseGenerator):
+    # TODO: Codes can technically have whitespace but here I've left it out for simplicity
+    @staticmethod
+    def generate():
+        return codeModel(
+            "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
+        )
+
+
+@register_generator
+class dateGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return dateModel(faker.date())
+
+
+@register_generator
+class dateTimeGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return dateTimeModel(faker.date_time().isoformat())
+
+
+@register_generator
+class decimalGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return decimalModel(faker.random_number())
+
+
+@register_generator
+class idGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return idModel(faker.uuid4())
+
+
+@register_generator
+class instantGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return instantModel(faker.date_time().isoformat())
+
+
+@register_generator
+class integerGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return integerModel(faker.random_int())
+
+
+@register_generator
+class markdownGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return markdownModel(faker.text())
+
+
+@register_generator
+class positiveIntGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return positiveIntModel(faker.random_int(min=1))
+
+
+@register_generator
+class stringGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return stringModel(faker.word())
+
+
+@register_generator
+class timeGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return timeModel(faker.time())
+
+
+@register_generator
+class unsignedIntGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return unsignedIntModel(faker.random_int(min=0))
+
+
+@register_generator
+class uriGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return uriModel(f"https://example/{faker.uri_path()}")
+
+
+@register_generator
+class urlGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return urlModel(f"https://example/{faker.uri_path()}")
+
+
+@register_generator
+class uuidGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        return uuidModel(faker.uuid4())

healthchain/data_generator/encounter_generator.py

@@ -0,0 +1,92 @@
+from healthchain.fhir_resources.encounter_resources import EncounterModel
+from healthchain.fhir_resources.base_resources import CodingModel, CodeableConceptModel
+from healthchain.data_generator.base_generators import (
+    BaseGenerator,
+    generator_registry,
+    register_generator,
+)
+from typing import Optional
+from faker import Faker
+
+faker = Faker()
+
+
+@register_generator
+class ClassGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        patient_class_mapping = {"IMP": "inpatient", "AMB": "ambulatory"}
+        patient_class = faker.random_element(elements=("IMP", "AMB"))
+        return CodeableConceptModel(
+            coding=[
+                CodingModel(
+                    system="http://terminology.hl7.org/CodeSystem/v3-ActCode",
+                    code=patient_class,
+                    display=patient_class_mapping.get(patient_class),
+                )
+            ]
+        )
+
+
+@register_generator
+class EncounterTypeGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        encounter_type_mapping = {"11429006": "consultation", "50849002": "emergency"}
+        encounter_type = faker.random_element(elements=("11429006", "50849002"))
+        return CodeableConceptModel(
+            coding=[
+                CodingModel(
+                    system="http://snomed.info/sct",
+                    code=encounter_type,
+                    display=encounter_type_mapping.get(encounter_type),
+                )
+            ]
+        )
+
+
+@register_generator
+class EncounterPriorityGenerator(BaseGenerator):
+    @staticmethod
+    def generate():
+        encounter_priority_mapping = {"17621005": "normal", "24484000": "critical"}
+        encounter_priority = faker.random_element(elements=("17621005", "24484000"))
+        return CodeableConceptModel(
+            coding=[
+                CodingModel(
+                    system="http://snomed.info/sct",
+                    code=encounter_priority,
+                    display=encounter_priority_mapping.get(encounter_priority),
+                )
+            ]
+        )
+
+
+@register_generator
+class EncounterGenerator(BaseGenerator):
+    @staticmethod
+    def generate(patient_reference: Optional[str]):
+        if patient_reference is None:
+            patient_reference = "Patient/123"
+        return EncounterModel(
+            resourceType="Encounter",
+            id=generator_registry.get("idGenerator").generate(),
+            text={
+                "status": "generated",
+                "div": '<div xmlns="http://www.w3.org/1999/xhtml">Encounter with patient @example</div>',
+            },
+            status=faker.random_element(
+                elements=(
+                    "planned",
+                    "in-progress",
+                    "on-hold",
+                    "discharged",
+                    "cancelled",
+                )
+            ),
+            class_field=[generator_registry.get("ClassGenerator").generate()],
+            type_field=[generator_registry.get("EncounterTypeGenerator").generate()],
+            subject={"reference": patient_reference, "display": patient_reference},
+            participant=[],
+            reason=[],
+        )

healthchain/data_generator/generator_templates/templates.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class patient_template_1(BaseModel):
+    name: list = [{"family": "Doe", "given": ["John"], "prefix": ["Mr."]}]
+    birthDate: str = "1999-01-01"