feat: add Python wrapper to run Synthea programmatically (#951)

.gitignore

@@ -51,3 +51,4 @@ src/main/resources/export/outpatient_rev_cntr_code_map.json
 
 # VS Code plugin files
 .history
+output/

scripts/python_wrapper/example.py

@@ -0,0 +1,24 @@
+from wrapper import SyntheaGenerator
+import os
+
+gen = SyntheaGenerator(
+    state="California",
+    gender="F",
+    age="30-40",
+    patients=2,
+    module="diabetes"
+)
+
+patients = gen.generate()
+
+
+if not patients:
+    print("No patients generated.")
+else:
+    print(f"Generated {len(patients)} patients")
+    print("Sample patient resourceType:", patients[0].get("resourceType"))
+
+gen.save("my_output/diabetes_test")
+
+saved_files = os.listdir("my_output/diabetes_test")
+print("Saved files:", saved_files)

scripts/python_wrapper/wrapper.py

@@ -0,0 +1,68 @@
+import subprocess
+import os
+import json
+
+class SyntheaGenerator:
+    def __init__(self, state="Massachusetts", gender=None, age=None, patients=1, module=None, synth_path=None):
+        if synth_path is None:
+            synth_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+
+        self.state = state
+        self.gender = gender
+        self.age = age
+        self.patients = patients
+        self.module = module
+        self.synth_path = synth_path
+        self.output_path = os.path.join(self.synth_path, "output", "fhir")
+
+    def _clear_output(self):
+        if os.path.exists(self.output_path):
+            for f in os.listdir(self.output_path):
+                if f.endswith(".json"):
+                    os.remove(os.path.join(self.output_path, f))
+
+    def _build_command(self):
+        script = "run_synthea.bat" if os.name == "nt" else "./run_synthea"
+        cmd = [script, self.state, "-p", str(self.patients)]
+        if self.age:
+            cmd += ["-a", self.age]
+        if self.gender:
+            cmd += ["-g", self.gender]
+        if self.module:
+            cmd += ["-m", self.module]
+        return cmd
+
+
+    def generate(self):
+        self._clear_output()
+        cmd = self._build_command()
+        try:
+            subprocess.run(cmd, cwd=self.synth_path, check=True)
+        except subprocess.CalledProcessError as e:
+            raise RuntimeError(f"[Synthea Error] CLI failed: {e}")
+
+        if not os.path.exists(self.output_path):
+            raise RuntimeError("No output folder found.")
+
+        json_files = [f for f in os.listdir(self.output_path) if f.endswith(".json")]
+        patients = []
+        for file in json_files:
+            with open(os.path.join(self.output_path, file), "r") as f:
+                data = json.load(f)
+                entries = data.get("entry", [])
+                has_patient = any(
+                    e.get("resource", {}).get("resourceType") == "Patient" for e in entries
+                )
+                if data.get("resourceType") == "Bundle" and has_patient:
+                    patients.append(data)
+
+        return patients
+
+    def save(self, output_dir):
+        os.makedirs(output_dir, exist_ok=True)
+        for file in os.listdir(self.output_path):
+            if file.endswith(".json"):
+                src = os.path.join(self.output_path, file)
+                dst = os.path.join(output_dir, file)
+                with open(src, "r") as f_in, open(dst, "w") as f_out:
+                    f_out.write(f_in.read())