feat(examples): add an offline ValueSet validate-code example (#4657)

CHANGELOG.md

@@ -19,6 +19,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add an offline ValueSet validate-code example that checks a member, a
+  non-member and an excluded code against a synthetic ValueSet, and shows the
+  `annotate`, `drop` and `downgrade` CodeableConcept policies, with a smoke
+  test (#4657).
+
 - Add a runnable TypeScript REST client example that de-identifies a note
   with an injected `fetch` and handles `OpenMedApiError` by its code, with
   offline stubbed-fetch tests (#5429).

docs/cookbook.md

@@ -51,6 +51,7 @@ using the same workflow with real data.
 | Run the v3 five-source synthetic Journey gate | [`examples/v3_golden_journey.py`](https://github.com/maziyarpanahi/openmed/blob/master/examples/v3_golden_journey.py) |
 | De-identify, extract, ground, and export a FHIR Bundle | [`examples/ground_then_export_fhir.py`](https://github.com/maziyarpanahi/openmed/blob/master/examples/ground_then_export_fhir.py) |
 | Ground a synthetic mention offline | [`examples/offline_grounding.py`](https://github.com/maziyarpanahi/openmed/blob/master/examples/offline_grounding.py) |
+| Check codes against a local ValueSet offline, with a synthetic vocabulary | [`examples/valueset_validate_code.py`](https://github.com/maziyarpanahi/openmed/blob/master/examples/valueset_validate_code.py) |
 | Prepare a local OpenMRS de-identified handoff | [`examples/openmrs_deid_handoff.py`](https://github.com/maziyarpanahi/openmed/blob/master/examples/openmrs_deid_handoff.py) |
 | Build a de-identified DHIS2 district export | [`examples/dhis2_district_export.py`](https://github.com/maziyarpanahi/openmed/blob/master/examples/dhis2_district_export.py) |
 | Extend the SDOH status vocabulary with a local domain | [`examples/custom_status_vocabulary.py`](https://github.com/maziyarpanahi/openmed/blob/master/examples/custom_status_vocabulary.py) |

examples/valueset_validate_code.py

@@ -0,0 +1,100 @@
+"""Check codes against a local ValueSet, offline, with a synthetic vocabulary.
+
+``validate_code()`` and ``validate_codeable_concept()`` evaluate FHIR ValueSet
+membership from a ValueSet document you provide. They never contact a
+terminology server. The system and codes below are invented for this example;
+no real vocabulary content is used.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from openmed.clinical.exporters.valueset import (
+    validate_code,
+    validate_codeable_concept,
+)
+
+SYSTEM = "urn:openmed:example:visit-priority"
+
+# Three codes are included, and one of them is then excluded again.
+VALUESET: dict[str, Any] = {
+    "resourceType": "ValueSet",
+    "url": "urn:openmed:example:valueset:visit-priority",
+    "status": "draft",
+    "compose": {
+        "include": [
+            {
+                "system": SYSTEM,
+                "concept": [
+                    {"code": "routine"},
+                    {"code": "soon"},
+                    {"code": "immediate"},
+                ],
+            }
+        ],
+        "exclude": [{"system": SYSTEM, "concept": [{"code": "immediate"}]}],
+    },
+}
+
+# One coding is a member of the ValueSet and one is not.
+CONCEPT: dict[str, Any] = {
+    "text": "Visit priority",
+    "coding": [
+        {"system": SYSTEM, "code": "soon"},
+        {"system": SYSTEM, "code": "unlisted"},
+    ],
+}
+
+
+def main() -> list[dict[str, Any]]:
+    """Validate a member, a non-member and an excluded code, then each policy.
+
+    Prints one JSON line per case and returns the same records.
+    """
+
+    records: list[dict[str, Any]] = []
+
+    for case, code in (
+        ("member", "routine"),
+        ("non_member", "unlisted"),
+        ("excluded", "immediate"),
+    ):
+        result = validate_code(SYSTEM, code, VALUESET)
+        records.append(
+            {
+                "case": case,
+                "code": code,
+                "valid": result.valid,
+                "message": result.message,
+            }
+        )
+
+    # The policy decides what happens to a coding that is not a member:
+    #   annotate  keeps it and adds a validation extension (the default)
+    #   drop      removes it
+    #   downgrade keeps it and reports the finding as a warning, not an error
+    for policy in ("annotate", "drop", "downgrade"):
+        validated = validate_codeable_concept(CONCEPT, VALUESET, policy=policy)
+        records.append(
+            {
+                "case": f"codeable_concept_{policy}",
+                "valid": validated.valid,
+                "codes": [coding["code"] for coding in validated["coding"]],
+                "annotated_codes": [
+                    coding["code"]
+                    for coding in validated["coding"]
+                    if "extension" in coding
+                ],
+                "severities": [issue["severity"] for issue in validated.issues],
+            }
+        )
+
+    for record in records:
+        print(json.dumps(record, sort_keys=True))
+    return records
+
+
+if __name__ == "__main__":
+    main()

tests/unit/examples/test_valueset_validate_code_example.py

@@ -0,0 +1,102 @@
+"""Tests for the offline ValueSet validate-code example."""
+
+from __future__ import annotations
+
+import json
+import socket
+
+import pytest
+
+from examples import valueset_validate_code as example
+
+EXAMPLE_SYSTEM_PREFIX = "urn:openmed:example:"
+
+
+def _by_case(records: list[dict]) -> dict[str, dict]:
+    return {record["case"]: record for record in records}
+
+
+def test_example_prints_one_json_line_per_case(capsys) -> None:
+    records = example.main()
+
+    lines = capsys.readouterr().out.strip().splitlines()
+    assert [json.loads(line) for line in lines] == records
+    assert [record["case"] for record in records] == [
+        "member",
+        "non_member",
+        "excluded",
+        "codeable_concept_annotate",
+        "codeable_concept_drop",
+        "codeable_concept_downgrade",
+    ]
+
+
+def test_member_is_valid_and_non_member_is_not(capsys) -> None:
+    cases = _by_case(example.main())
+    capsys.readouterr()
+
+    assert cases["member"]["valid"] is True
+    assert cases["non_member"]["valid"] is False
+    assert "not a member" in cases["non_member"]["message"]
+
+
+def test_excluded_code_is_reported_as_a_non_member(capsys) -> None:
+    cases = _by_case(example.main())
+    capsys.readouterr()
+
+    assert cases["excluded"]["valid"] is False
+    assert "explicitly excluded" in cases["excluded"]["message"]
+
+
+def test_each_policy_treats_the_non_member_coding_differently(capsys) -> None:
+    cases = _by_case(example.main())
+    capsys.readouterr()
+
+    annotate = cases["codeable_concept_annotate"]
+    assert annotate["codes"] == ["soon", "unlisted"]
+    assert annotate["annotated_codes"] == ["unlisted"]
+    assert annotate["severities"] == ["error"]
+
+    drop = cases["codeable_concept_drop"]
+    assert drop["codes"] == ["soon"]
+    assert drop["annotated_codes"] == []
+    assert drop["severities"] == ["error"]
+
+    downgrade = cases["codeable_concept_downgrade"]
+    assert downgrade["codes"] == ["soon", "unlisted"]
+    assert downgrade["annotated_codes"] == ["unlisted"]
+    assert downgrade["severities"] == ["warning"]
+
+    assert all(case["valid"] is False for case in (annotate, drop, downgrade))
+
+
+def test_example_uses_only_the_invented_example_system() -> None:
+    systems = {include["system"] for include in example.VALUESET["compose"]["include"]}
+    systems |= {exclude["system"] for exclude in example.VALUESET["compose"]["exclude"]}
+    systems |= {coding["system"] for coding in example.CONCEPT["coding"]}
+    systems.add(example.SYSTEM)
+
+    assert systems
+    assert all(system.startswith(EXAMPLE_SYSTEM_PREFIX) for system in systems)
+
+
+def test_example_makes_no_network_connection(monkeypatch, capsys) -> None:
+    def refuse(*args, **kwargs):
+        raise AssertionError("the example must not open a network connection")
+
+    monkeypatch.setattr(socket, "socket", refuse)
+    monkeypatch.setattr(socket, "create_connection", refuse)
+
+    records = example.main()
+    capsys.readouterr()
+
+    assert len(records) == 6
+
+
+@pytest.mark.parametrize("policy", ["annotate", "drop", "downgrade"])
+def test_documented_policies_are_the_ones_the_library_accepts(policy) -> None:
+    from openmed.clinical.exporters.valueset import validate_codeable_concept
+
+    result = validate_codeable_concept(example.CONCEPT, example.VALUESET, policy=policy)
+
+    assert result.valid is False