Integration of data generator and api (dotimplement#13)

Co-authored-by: jenniferajiang <[email protected]>

.gitignore

@@ -158,3 +158,5 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+output/

healthchain/__init__.py

@@ -1,7 +1,12 @@
 import logging
 from .utils.logger import add_handlers
-
+from healthchain.decorators import ehr, api, sandbox
+from healthchain.data_generator.data_generator import DataGenerator
+from healthchain.models.requests.cdsrequest import CDSRequest
 
 logger = logging.getLogger(__name__)
 add_handlers(logger)
 logger.setLevel(logging.INFO)
+
+# Export them at the top level
+__all__ = ["ehr", "api", "sandbox", "DataGenerator", "CDSRequest"]

healthchain/base.py

@@ -76,10 +76,6 @@ class BaseStrategy(ABC):
     - the format of the request it constructs (CDS Hook or NoteReader workflows)
     """
 
-    @abstractmethod
-    def _validate_data(self, data) -> bool:
-        pass
-
     @abstractmethod
     def construct_request(self, data, workflow: Workflow) -> Dict:
         pass

healthchain/cli.py

@@ -0,0 +1,28 @@
+import argparse
+import subprocess
+
+
+def run_file(filename):
+    try:
+        result = subprocess.run(["poetry", "run", "python", filename], check=True)
+        print(result.stdout)
+    except subprocess.CalledProcessError as e:
+        print(f"An error occurred while trying to run the file: {e}")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="HealthChain command-line interface")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    # Subparser for the 'run' command
+    run_parser = subparsers.add_parser("run", help="Run a specified file")
+    run_parser.add_argument("filename", type=str, help="The filename to run")
+
+    args = parser.parse_args()
+
+    if args.command == "run":
+        run_file(args.filename)
+
+
+if __name__ == "__main__":
+    main()

healthchain/clients.py

@@ -3,6 +3,7 @@
 
 from typing import Any, Callable, List, Dict
 
+from .models.requests.cdsrequest import CDSRequest
 from .base import BaseStrategy, BaseClient, Workflow
 
 log = logging.getLogger(__name__)
@@ -24,7 +25,7 @@ def __init__(
         self.data_generator_func: Callable[..., Any] = func
         self.workflow: Workflow = workflow
         self.strategy: BaseStrategy = strategy
-        self.request_data: List[Dict] = []
+        self.request_data: List[CDSRequest] = []
 
     def generate_request(self, *args: Any, **kwargs: Any) -> None:
         """
@@ -58,7 +59,7 @@ async def send_request(self, url: str) -> List[Dict]:
             for request in self.request_data:
                 try:
                     response = await client.post(
-                        url=url, data=request.model_dump_json(exclude_none=True)
+                        url=url, json=request.model_dump(exclude_none=True)
                     )
                     json_responses.append(response.json())
                 except Exception as e:

healthchain/data_generator/__init__.py

@@ -0,0 +1,19 @@
+from .encounter_generators import EncounterGenerator
+from .condition_generators import ConditionGenerator
+from .patient_generators import PatientGenerator
+from .practitioner_generators import PractitionerGenerator
+from .procedure_generators import ProcedureGenerator
+from .medication_administration_generators import MedicationAdministrationGenerator
+from .medication_request_generators import MedicationRequestGenerator
+from .data_generator import DataGenerator
+
+__all__ = [
+    "EncounterGenerator",
+    "ConditionGenerator",
+    "PatientGenerator",
+    "PractitionerGenerator",
+    "ProcedureGenerator",
+    "MedicationAdministrationGenerator",
+    "MedicationRequestGenerator",
+    "DataGenerator",
+]

healthchain/data_generator/base_generators.py

@@ -22,7 +22,10 @@
     urlModel,
     uuidModel,
 )
-
+from healthchain.fhir_resources.general_purpose_resources import (
+    CodeableConceptModel,
+    CodingModel,
+)
 from faker import Faker
 
 faker = Faker()
@@ -39,7 +42,7 @@ def register(self, cls):
     def get(self, name):
         if name not in self.registry:
             raise ValueError(f"No generator registered for '{name}'")
-        return self.registry.get(name)
+        return self.registry.get(name)()
 
 
 generator_registry = Registry()
@@ -177,3 +180,25 @@ class UuidGenerator(BaseGenerator):
     @staticmethod
     def generate():
         return uuidModel(faker.uuid4())
+
+
+class CodeableConceptGenerator(BaseGenerator):
+    @staticmethod
+    def generate_from_valueset(ValueSet):
+        value_set_instance = ValueSet()
+        return CodeableConceptModel(
+            coding=[
+                CodingModel(
+                    system=value_set_instance.system,
+                    code=faker.random_element(value_set_instance.value_set)["code"],
+                    display=faker.random_element(value_set_instance.value_set)[
+                        "display"
+                    ],
+                    # extension=[ExtensionModel(value_set_instance.extension)],
+                )
+            ]
+        )
+
+    @staticmethod
+    def generate():
+        pass

healthchain/data_generator/condition_generators.py

@@ -2,6 +2,7 @@
     BaseGenerator,
     generator_registry,
     register_generator,
+    CodeableConceptGenerator,
 )
 from healthchain.fhir_resources.general_purpose_resources import (
     CodeableConceptModel,
@@ -13,6 +14,10 @@
     Condition_StageModel,
     Condition_ParticipantModel,
 )
+from healthchain.data_generator.value_sets.condition import (
+    ConditionCodeSimple,
+    ConditionCodeComplex,
+)
 from typing import Optional
 from faker import Faker
 
@@ -95,18 +100,18 @@ def generate():
 
 
 @register_generator
-class SnomedCodeGenerator(BaseGenerator):
-    @staticmethod
-    def generate():
-        return CodeableConceptModel(
-            coding=[
-                CodingModel(
-                    system="http://snomed.info/sct",
-                    code=faker.random_element(elements=("386661006")),
-                    display=faker.random_element(elements=("Fever")),
-                )
-            ]
-        )
+class SnomedCodeGenerator(CodeableConceptGenerator):
+    def __init__(self) -> None:
+        super().__init__()
+
+    # @staticmethod
+    def generate(self, constraints: Optional[list] = None):
+        # TODO: Factor out the code generation logic to a central place
+        constraints = constraints or []
+        if "complex-condition" not in constraints:
+            return self.generate_from_valueset(ConditionCodeSimple)
+        elif "complex-condition" in constraints:
+            return self.generate_from_valueset(ConditionCodeComplex)
 
 
 @register_generator
@@ -135,11 +140,19 @@ def generate():
 
 
 @register_generator
-class ConditionModelGenerator(BaseGenerator):
+class ConditionGenerator(BaseGenerator):
     @staticmethod
-    def generate(subject_reference: Optional[str], encounter_reference: Optional[str]):
+    def generate(
+        subject_reference: Optional[str] = None,
+        encounter_reference: Optional[str] = None,
+        constraints: Optional[list] = None,
+    ):
         subject_reference = subject_reference or "Patient/123"
         encounter_reference = encounter_reference or "Encounter/123"
+        # TODO - Check whether this is the correct way to handle params
+        code = generator_registry.get("SnomedCodeGenerator").generate(
+            constraints=constraints
+        )
         return ConditionModel(
             id=generator_registry.get("IdGenerator").generate(),
             clinicalStatus=generator_registry.get("ClinicalStatusGenerator").generate(),
@@ -148,12 +161,13 @@ def generate(subject_reference: Optional[str], encounter_reference: Optional[str
             ).generate(),
             category=[generator_registry.get("CategoryGenerator").generate()],
             severity=generator_registry.get("SeverityGenerator").generate(),
-            code=generator_registry.get("SnomedCodeGenerator").generate(),
+            code=code,
             bodySite=[generator_registry.get("BodySiteGenerator").generate()],
             subject=ReferenceModel(reference=subject_reference),
             encounter=ReferenceModel(reference=encounter_reference),
-            onsetDateTime=generator_registry.get(
+            onsetDateTime=generator_registry.get("DateGenerator").generate(),
+            abatementDateTime=generator_registry.get(
                 "DateGenerator"
-            ).generate(),  ## Are there more plausible dates to use?
+            ).generate(),  ## TODO: Constraint abatementDateTime to be after onsetDateTime
             recordedDate=generator_registry.get("DateGenerator").generate(),
         )