#!/usr/bin/env python3 """Original synthetic FHIR R4 terminology exercise. Standard library; no network. This accepts a deliberately small ConceptMap subset, not arbitrary FHIR. The static contract stands in for a separately reviewed terminology agreement. No clinical code set, patient record, terminology server or approval is involved. """ import argparse from copy import deepcopy import json SOURCE = "https://example.invalid/local-services" TARGET = "https://example.invalid/synthetic-services" CONTRACT = { "map_url": "https://example.invalid/maps/local-to-payer", "map_version": "workshop-1", "source_system": SOURCE, "source_version": "local-1", "target_system": TARGET, "target_version": "payer-1", "allowed_targets": ["SYNTHETIC-VISIT"], } def coding(): return {"system": SOURCE, "version": "local-1", "code": "VISIT", "display": "Visit"} def concept_map(): return { "resourceType": "ConceptMap", "url": CONTRACT["map_url"], "version": "workshop-1", "status": "active", "group": [{"source": SOURCE, "sourceVersion": "local-1", "target": TARGET, "targetVersion": "payer-1", "element": [{"code": "VISIT", "display": "Visit", "target": [ {"code": "SYNTHETIC-VISIT", "display": "Office encounter", "equivalence": "equal"} ]}]}], } def has_modifier(value): if isinstance(value, dict): return "modifierExtension" in value or any(has_modifier(v) for v in value.values()) return isinstance(value, list) and any(has_modifier(v) for v in value) def translate(source, mapping, contract=CONTRACT, *, match_display=False): """Resolve one code under the narrow contract; never change the supplied records.""" evidence = [] def finish(reason, target=None): return { "state": "mapped" if target else "needs_terminology_review", "reason": reason, "source": deepcopy(source), "target": target, "map": {"url": mapping.get("url"), "version": mapping.get("version")} if isinstance(mapping, dict) else None, "evidence": deepcopy(evidence), "owner": "eligibility_operations" if target else "terminology_steward", "next_action": "Use the mapped identity in a separately scoped eligibility question." if target else "Resolve the mapping evidence before forming the eligibility question.", "not_established": ["coverage", "authorization_requirement", "authorization_approval", "payment"], } if not isinstance(source, dict) or any(not isinstance(source.get(k), str) or not source[k].strip() for k in ("system", "version", "code")): return finish("source_identity_incomplete") if (source["system"], source["version"]) != (contract["source_system"], contract["source_version"]): return finish("source_context_not_accepted") if not isinstance(mapping, dict) or mapping.get("resourceType") != "ConceptMap": return finish("not_a_concept_map") if has_modifier(source) or has_modifier(mapping): return finish("unsupported_modifier") if (mapping.get("url"), mapping.get("version")) != (contract["map_url"], contract["map_version"]): return finish("unreviewed_map_identity") if mapping.get("status") != "active": return finish("map_not_active") evidence.append({"path": "url + version", "value": [mapping["url"], mapping["version"]]}) groups = mapping.get("group") if not isinstance(groups, list) or any(not isinstance(g, dict) for g in groups): return finish("invalid_map_shape") context = (source["system"], source["version"], contract["target_system"], contract["target_version"]) groups = [(i, g) for i, g in enumerate(groups) if tuple(g.get(k) for k in ("source", "sourceVersion", "target", "targetVersion")) == context] if not groups: return finish("mapping_context_not_found") matches = [] for gi, group in groups: elements = group.get("element") if not isinstance(elements, list) or any(not isinstance(e, dict) for e in elements): return finish("invalid_map_shape") for ei, element in enumerate(elements): # DELIBERATE DEFECT: a label collision can select a different concept. field = "display" if match_display else "code" if source.get(field) and element.get(field) == source[field]: matches.append((f"group[{gi}].element[{ei}]", element)) if not matches: return finish("source_not_mapped") # No fallback from group.unmapped in this subset. targets = [] for path, element in matches: values = element.get("target", []) if not isinstance(values, list) or any(not isinstance(t, dict) for t in values): return finish("invalid_map_shape") targets.extend((f"{path}.target[{i}]", t) for i, t in enumerate(values)) # Aggregate matching entries across groups; never silently take the first. if len(targets) != 1: return finish("target_not_unique") path, target = targets[0] evidence.append({"path": path, "value": deepcopy(target)}) if target.get("dependsOn") or target.get("product"): return finish("additional_context_required") if target.get("equivalence") not in ("equal", "equivalent"): return finish("relationship_not_accepted") if not isinstance(target.get("code"), str) or target["code"] not in contract["allowed_targets"]: return finish("target_outside_agreed_set") return finish("mapped_identity", {"system": contract["target_system"], "version": contract["target_version"], "code": target["code"], **({"display": target["display"]} if "display" in target else {})}) def scenarios(): cases = [] def add(name): source, mapping = coding(), concept_map() cases.append({"name": name, "source": source, "concept_map": mapping}) return source, mapping add("exact-identity") source, _ = add("renamed-display"); source["display"] = "Consultation label" source, _ = add("same-label-different-code"); source["code"] = "ADMIN" source, _ = add("same-code-different-system"); source["system"] = "https://example.invalid/unrelated" source, _ = add("missing-source-version"); source.pop("version") _, mapping = add("unreviewed-map-version"); mapping["version"] = "workshop-2" _, mapping = add("retired-map"); mapping["status"] = "retired" _, mapping = add("changed-target-version"); mapping["group"][0]["targetVersion"] = "payer-2" _, mapping = add("wider-target"); mapping["group"][0]["element"][0]["target"][0]["equivalence"] = "wider" _, mapping = add("conditional-map") mapping["group"][0]["element"][0]["target"][0]["dependsOn"] = [{"property": "https://example.invalid/setting", "value": "office"}] _, mapping = add("duplicate-mapping-group"); mapping["group"].append(deepcopy(mapping["group"][0])) _, mapping = add("outside-target-set"); mapping["group"][0]["element"][0]["target"][0]["code"] = "SYNTHETIC-OTHER" source, mapping = add("unmapped-default"); source["code"] = "UNKNOWN"; source["display"] = "Unknown" mapping["group"][0]["unmapped"] = {"mode": "fixed", "code": "SYNTHETIC-VISIT"} source, _ = add("reverse-direction"); source.update(system=TARGET, version="payer-1", code="SYNTHETIC-VISIT") return cases def report(match_display=False): return {"synthetic": True, "execution": "local Python only; no terminology server or clinical data", "mode": "deliberately broken display matching" if match_display else "guarded identity matching", "contract": deepcopy(CONTRACT), "cases": [ {**case, "result": translate(case["source"], case["concept_map"], match_display=match_display)} for case in scenarios()]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json", action="store_true") parser.add_argument("--match-display", action="store_true", help="Run the intentionally wrong label comparison") args = parser.parse_args() result = report(args.match_display) if args.json: print(json.dumps(result, indent=2)) else: print(result["mode"]) for case in result["cases"]: print(f'{case["name"]:30} {case["result"]["reason"]}') print("Exit 0 means the report ran, not that a learner passed or a service was approved.") if __name__ == "__main__": main()