#!/usr/bin/env python3 """Join two synthetic healthcare labs without losing the service vocabulary version. Python 3.11+, standard library. Unzip the complete healthcare handoff lab first. No network, patient data, approval, payment, database or model calls. The question is an internal teaching envelope, not a complete FHIR request or a sent request. """ import argparse from copy import deepcopy from dataclasses import asdict, replace import json from pathlib import Path import runpy def load_lab(name): file = Path(__file__).resolve().with_name(name) if not file.is_file(): raise SystemExit(f"Missing {name}. Unzip fde-healthcare-handoff.zip; keep all three Python files together.") return runpy.run_path(str(file)) TERMINOLOGY = load_lab("fde-healthcare-terminology.py") ELIGIBILITY = load_lab("fde-healthcare-eligibility.py") UNANSWERED = ["authorization_approval", "claim_payment", "patient_responsibility"] def fixture(): question = ELIGIBILITY["Question"]() body = ELIGIBILITY["response"]() body["insurance"][0]["item"][0]["productOrService"]["coding"][0]["version"] = "payer-1" return {"source": TERMINOLOGY["coding"](), "concept_map": TERMINOLOGY["concept_map"](), "question": asdict(question), "http_status": 200, "response": body} def connect(case, *, ignore_response_version=False): """Interpret into a local candidate; expose a fact only after identity binding. Both previous interpreters are used unchanged. This extra boundary requires exactly one coding on the selected service item and an explicit vocabulary version. Those are workshop restrictions, not universal FHIR requirements. All inputs are trusted synthetic fixtures, not arbitrary external JSON. """ mapping = TERMINOLOGY["translate"](case["source"], case["concept_map"]) trace = ["mapping:" + mapping["reason"]] outgoing = None binding = None def finish(state, reason, owner, next_action, eligibility=None): return deepcopy({ "state": state, "reason": reason, "owner": owner, "next_action": next_action, "mapping": mapping, "question": outgoing, "binding": binding, "eligibility": eligibility, "authorization_required": eligibility["authorization_required"] if eligibility else None, "not_established": UNANSWERED + ([] if eligibility else ["authorization_requirement"]), "trace": trace + [reason], }) if mapping["state"] != "mapped": return finish("needs_terminology_review", "mapping_unresolved", "terminology_steward", "Settle the source mapping before asking the receiving system.") target = mapping["target"] # Patient, coverage, payer, date and request remain the caller's scoped question. # Only the service identity is replaced with the reviewed mapping result. question = replace(ELIGIBILITY["Question"](**case["question"]), code_system=target["system"], service_code=target["code"]) outgoing = {"context": asdict(question), "service": deepcopy(target), "sent": False} trace.append("versioned_question_prepared") # This candidate is private to this function. It is not yet a staff handoff. candidate = ELIGIBILITY["interpret"](question, case["response"], case["http_status"]) trace.append("source_interpretation:" + candidate["reason"]) if candidate["state"] != "reported": return finish(candidate["state"], candidate["reason"], candidate["owner"], candidate["next_action"]) body = case["response"] selected = [] for pi, plan in enumerate(body["insurance"]): if ELIGIBILITY["reference"](plan, "coverage") != question.coverage: continue for ii, item in enumerate(plan["item"]): service = item.get("productOrService") codes = service.get("coding") if isinstance(service, dict) else None if not isinstance(codes, list): continue if any(isinstance(code, dict) and code.get("system") == target["system"] and code.get("code") == target["code"] for code in codes): selected.append((f"insurance[{pi}].item[{ii}].productOrService.coding", codes)) # The earlier interpreter already rejects multiple matching items. Keep this # explicit because this boundary must not silently select an array position. if len(selected) != 1: return finish("needs_source", "response_item_not_unique", "integration_engineer", "Resolve the service item before binding a terminology version.") path, codes = selected[0] binding = {"path": path, "expected": {k: target[k] for k in ("system", "version", "code")}, "observed": deepcopy(codes), "policy": "ignore_version" if ignore_response_version else "require_version"} if len(codes) != 1: return finish("needs_source", "response_coding_not_unique", "integration_engineer", "This workshop accepts one coding on the selected item. Agree how to interpret multiple representations.") actual = codes[0].get("version") binding["path"] += "[0].version" if not ignore_response_version: if not isinstance(actual, str) or not actual.strip(): return finish("needs_source", "response_version_unreported", "integration_engineer", "Obtain the agreed vocabulary version; do not infer it from the request, label or resource revision.") if actual != target["version"]: return finish("needs_source", "response_version_mismatch", "integration_engineer", "Resolve the changed vocabulary with the terminology owner before using this service fact.") trace.append("response_version_ignored" if ignore_response_version else "response_version_matches") return finish("ready_for_staff", candidate["reason"], "eligibility_operations", "Hand over the scoped requirement fact and its evidence; approval and payment remain separate questions.", candidate) def scenarios(): cases = [] def add(name): item = {"name": name, **fixture()} cases.append(item) return item def service(case): return case["response"]["insurance"][0]["item"][0] def code(case): return service(case)["productOrService"]["coding"][0] add("matching-version") service(add("explicit-false"))["authorizationRequired"] = False code(add("renamed-response-display"))["display"] = "New readable label" code(add("changed-response-version"))["version"] = "payer-2" code(add("missing-response-version")).pop("version") case = add("resource-version-is-not-vocabulary-version") code(case).pop("version"); case["response"]["meta"] = {"versionId": "payer-1"} case = add("ambiguous-response-codings") service(case)["productOrService"]["coding"].append(dict(code(case), version="payer-2")) case = add("ambiguous-response-items") case["response"]["insurance"][0]["item"].append(deepcopy(service(case))) case["response"]["insurance"][0]["item"][1]["productOrService"]["coding"][0]["version"] = "payer-2" add("wrong-patient")["response"]["patient"]["reference"] = "Patient/SYNTHETIC-OTHER" add("wrong-request")["response"]["request"]["reference"] = "CoverageEligibilityRequest/SYNTHETIC-OTHER" service(add("missing-requirement")).pop("authorizationRequired") add("partial-response")["response"]["outcome"] = "partial" add("mapping-unresolved")["source"]["code"] = "ADMIN" add("transport-unavailable")["http_status"] = 503 return cases def report(ignore_response_version=False): return {"synthetic": True, "execution": "local Python only; no request sent and no records saved", "mode": "deliberately ignored response version" if ignore_response_version else "version-bound handoff", "cases": [{**case, "result": connect(case, ignore_response_version=ignore_response_version)} for case in scenarios()]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json", action="store_true") parser.add_argument("--ignore-response-version", action="store_true", help="Deliberately omit only the version check") args = parser.parse_args() output = report(args.ignore_response_version) if args.json: print(json.dumps(output, indent=2)) else: print(output["mode"]) for case in output["cases"]: print(f'{case["name"]:43} {case["result"]["state"]:27} {case["result"]["reason"]}') print("Exit 0 means the report ran, not that a service is approved or a learner passed.") if __name__ == "__main__": main()