fix: accept Define-XML extensions of extensible CDISC codelists in conformance checks

pointblank/metadata/_conformance/ct.py

@@ -19,9 +19,15 @@ class ControlledTerminology:
         The CT package identifiers that were loaded (for provenance).
     """
 
-    def __init__(self, codelists: dict[str, set[str]], packages: list[str]) -> None:
+    def __init__(
+        self,
+        codelists: dict[str, set[str]],
+        packages: list[str],
+        extensible: set[str] | None = None,
+    ) -> None:
         self._codelists = codelists
         self.packages = packages
+        self._extensible = {name.upper() for name in (extensible or set())}
 
     @classmethod
     def load(cls, packages: list[str]) -> ControlledTerminology:
@@ -40,6 +46,7 @@ def load(cls, packages: list[str]) -> ControlledTerminology:
             the same codelist appears in both.
         """
         codelists: dict[str, set[str]] = {}
+        extensible: set[str] = set()
         loaded: list[str] = []
         for pkg in packages:
             path = _CT_DIR / f"{pkg}.json"
@@ -52,8 +59,9 @@ def load(cls, packages: list[str]) -> ControlledTerminology:
             data: dict = json.loads(path.read_text(encoding="utf-8"))
             for name, terms in data.get("codelists", {}).items():
                 codelists[name.upper()] = set(str(t) for t in terms)
+            extensible |= {name.upper() for name in data.get("extensible", [])}
             loaded.append(pkg)
-        return cls(codelists, loaded)
+        return cls(codelists, loaded, extensible)
 
     @classmethod
     def load_default(cls) -> ControlledTerminology:
@@ -82,6 +90,10 @@ def is_valid(self, codelist: str, value) -> bool:
             return True  # unknown codelist: don't flag
         return str(value) in terms
 
+    def is_extensible(self, name: str) -> bool:
+        """Whether a codelist is extensible (sponsors may add terms, declared in Define-XML)."""
+        return name.upper() in self._extensible
+
     def get_codelist(self, name: str) -> set[str] | None:
         """Return the set of permitted values for a codelist, or `None` if unknown."""
         return self._codelists.get(name.upper())

pointblank/metadata/_conformance/operations.py

@@ -91,13 +91,27 @@ def _op_codelist_check(
         return df.with_columns(nw.lit(True).alias(result_col))
     # Build case-insensitive lookup; SAS/XPT missing values arrive as "" — treat as null.
     upper_terms = {t.upper() for t in terms}
+    # Sponsors may extend an extensible codelist, declaring the new terms in Define-XML
+    if ct.is_extensible(codelist):
+        upper_terms |= {str(v).upper() for v in _define_extended_values(define_meta, col)}
     values = df[col].to_list()
     mask = [
         True if (v is None or str(v) == "") else (str(v).upper() in upper_terms) for v in values
     ]
     return df.with_columns(_new_bool_series(result_col, mask, df))
 
 
+def _define_extended_values(define_meta: Any, col: str) -> list:
+    """Get the codelist extensions (`def:ExtendedValue="Yes"`) Define-XML declares for a variable."""
+    if define_meta is None:
+        return []
+    var_meta = next((v for v in define_meta.variables if v.name.upper() == col.upper()), None)
+    if var_meta is None or not var_meta.codelist_ref:
+        return []
+    codelist = define_meta.codelists.get(var_meta.codelist_ref)
+    return codelist.extended_values() if codelist is not None else []
+
+
 def _op_consistency_check(
     df: nw.DataFrame,
     params: dict,

pointblank/metadata/_readers_cdisc.py

@@ -200,6 +200,13 @@ def _read_define_xml_metadata(
     )
 
 
+def _is_extended_value(item, ns: dict[str, str]) -> bool:
+    """Whether a `CodeListItem`/`EnumeratedItem` is a sponsor extension (`def:ExtendedValue="Yes"`)."""
+    def_ns = ns.get("def")
+    value = item.get(f"{{{def_ns}}}ExtendedValue") if def_ns else None
+    return (value or "").strip().lower() == "yes"
+
+
 def _parse_codelists(mdv, ns: dict[str, str]) -> dict[str, Codelist]:
     """Parse `CodeList` elements from `MetaDataVersion`.
 
@@ -241,9 +248,9 @@ def _parse_codelists(mdv, ns: dict[str, str]) -> dict[str, Codelist]:
             decode_el = item.find("odm:Decode/odm:TranslatedText", ns)
             label = decode_el.text if decode_el is not None and decode_el.text else value
 
-            # Check for NCI code (extensible attribute)
-            # nci:ExtCodeID or similar
-            entry = CodelistEntry(value=value, label=str(label))
+            entry = CodelistEntry(
+                value=value, label=str(label), is_extended=_is_extended_value(item, ns)
+            )
             entries.append(entry)
 
         # Also check for EnumeratedItem (no Decode needed, value = label)
@@ -259,7 +266,11 @@ def _parse_codelists(mdv, ns: dict[str, str]) -> dict[str, Codelist]:
                     value = float(value)
                 except (ValueError, TypeError):
                     pass
-            entries.append(CodelistEntry(value=value, label=str(value)))
+            entries.append(
+                CodelistEntry(
+                    value=value, label=str(value), is_extended=_is_extended_value(item, ns)
+                )
+            )
 
         codelists[oid] = Codelist(
             name=name,

pointblank/metadata/_types.py

@@ -34,13 +34,17 @@ class CodelistEntry:
         Alternative terms for this entry.
     is_deprecated
         Whether this entry is deprecated.
+    is_extended
+        Whether this entry is a sponsor extension of an extensible codelist (declared in
+        Define-XML with `def:ExtendedValue="Yes"`).
     """
 
     value: Any
     label: str
     description: str | None = None
     synonyms: list[str] | None = None
     is_deprecated: bool = False
+    is_extended: bool = False
 
 
 @dataclass
@@ -83,6 +87,17 @@ def to_set(self) -> list:
         """
         return [entry.value for entry in self.codes if not entry.is_deprecated]
 
+    def extended_values(self) -> list:
+        """Get the values that extend an extensible codelist (sponsor extensions).
+
+        Returns
+        -------
+        list
+            The values of the entries marked as extended (e.g., with `def:ExtendedValue="Yes"` in
+            Define-XML).
+        """
+        return [entry.value for entry in self.codes if entry.is_extended]
+
     def to_dict(self) -> dict:
         """Get a value → label mapping.
 

tests/test_sdtm_conformant_package.py

@@ -317,3 +317,115 @@ def test_ct_bundle_has_no_variable_named_codelists():
     ct = ControlledTerminology.load_default()
     for name in ("AEACN", "AEOUT", "AEREL", "AETOXGR", "LBSTRESU"):
         assert name not in ct
+
+
+# ── Sponsor extensions of extensible codelists (declared in Define-XML) ────────────────────────
+
+
+def _define_xml(dataset: str, df: pl.DataFrame, column: str, codelist: str, values, extended):
+    """A minimal Define-XML 2.0 for one dataset, whose `column` uses the codelist `values`
+    (those in `extended` are marked `def:ExtendedValue="Yes"`)."""
+    item_refs, item_defs = [], []
+    for i, col in enumerate(df.columns, start=1):
+        oid = f"IT.{dataset}.{col}"
+        item_refs.append(f'<ItemRef ItemOID="{oid}" OrderNumber="{i}" Mandatory="No"/>')
+        dtype = df[col].dtype
+        data_type = "integer" if dtype.is_integer() else "float" if dtype.is_float() else "text"
+        cl_ref = f'<CodeListRef CodeListOID="CL.{codelist}"/>' if col == column else ""
+        item_defs.append(
+            f'<ItemDef OID="{oid}" Name="{col}" DataType="{data_type}" Length="40">{cl_ref}</ItemDef>'
+        )
+    extended_attr = ' def:ExtendedValue="Yes"'
+    items = "".join(
+        f'<EnumeratedItem CodedValue="{v}"{extended_attr if v in extended else ""}/>'
+        for v in values
+    )
+    return f"""<?xml version="1.0" encoding="UTF-8"?>
+<ODM xmlns="http://www.cdisc.org/ns/odm/v1.3" xmlns:def="http://www.cdisc.org/ns/def/v2.0"
+     FileOID="DEF.S001" FileType="Snapshot" ODMVersion="1.3.2">
+  <Study OID="S001">
+    <GlobalVariables>
+      <StudyName>S001</StudyName><StudyDescription>S001</StudyDescription>
+      <ProtocolName>S001</ProtocolName>
+    </GlobalVariables>
+    <MetaDataVersion OID="MDV.S001" Name="S001" def:DefineVersion="2.0.0"
+                     def:StandardName="SDTM-IG" def:StandardVersion="3.4">
+      <ItemGroupDef OID="IG.{dataset}" Name="{dataset}" Repeating="No" Domain="{dataset}"
+                    def:Structure="One record per row">{"".join(item_refs)}</ItemGroupDef>
+      {"".join(item_defs)}
+      <CodeList OID="CL.{codelist}" Name="{codelist}" DataType="text">{items}</CodeList>
+    </MetaDataVersion>
+  </Study>
+</ODM>
+"""
+
+
+# A two-period crossover, as in the SDTMIG 3.4 TA example: EPOCH is extensible, and the sponsor
+# adds TREATMENT 1 and TREATMENT 2
+CROSSOVER_TA = pl.DataFrame(
+    {
+        "STUDYID": ["S001"] * 5,
+        "DOMAIN": ["TA"] * 5,
+        "ARMCD": ["AB"] * 5,
+        "ARM": ["Drug A then Drug B"] * 5,
+        "TAETORD": [1, 2, 3, 4, 5],
+        "ETCD": ["SCRN", "A", "WO", "B", "FU"],
+        "ELEMENT": ["Screening", "Drug A", "Washout", "Drug B", "Follow-up"],
+        "EPOCH": ["SCREENING", "TREATMENT 1", "WASHOUT", "TREATMENT 2", "FOLLOW-UP"],
+    }
+)
+CROSSOVER_EPOCHS = ["SCREENING", "TREATMENT 1", "WASHOUT", "TREATMENT 2", "FOLLOW-UP"]
+
+
+@pytest.mark.parametrize(
+    "extended, n_issues",
+    [
+        (None, 2),  # no Define-XML: the sponsor terms aren't CDISC CT
+        ({"TREATMENT 1", "TREATMENT 2"}, 0),  # declared as extensions
+        (set(), 2),  # listed in Define-XML, but not marked as extensions
+    ],
+)
+def test_define_xml_extensions_of_extensible_codelist(tmp_path, extended, n_issues):
+    kwargs = {}
+    if extended is not None:
+        path = tmp_path / "define.xml"
+        path.write_text(
+            _define_xml("TA", CROSSOVER_TA, "EPOCH", "EPOCH", CROSSOVER_EPOCHS, extended)
+        )
+        kwargs["define_xml"] = path
+
+    report = pb.validate_sdtmig({"TA": CROSSOVER_TA}, **kwargs)
+
+    assert _rule(report, "SDTM-434").n_issues == n_issues
+
+
+def test_define_xml_cannot_extend_non_extensible_codelist(tmp_path):
+    # SEX (C66731) isn't extensible, so a value declared as an extension is still flagged
+    dm = conformant_sdtm_package()["DM"].with_columns(pl.Series("SEX", ["X", "F"]))
+    path = tmp_path / "define.xml"
+    path.write_text(_define_xml("DM", dm, "SEX", "SEX", ["M", "F", "X"], {"X"}))
+
+    report = pb.validate_sdtmig({"DM": dm}, define_xml=path)
+
+    assert _rule(report, "SDTM-007").n_issues == 1
+
+
+def test_ct_extensible_codelists():
+    from pointblank.metadata._conformance import ControlledTerminology
+
+    ct = ControlledTerminology.load_default()
+    assert ct.is_extensible("EPOCH") and ct.is_extensible("epoch")
+    assert not ct.is_extensible("SEX")
+
+
+def test_define_xml_extended_values_are_parsed(tmp_path):
+    from pointblank.metadata._readers_cdisc import _read_define_xml_metadata
+
+    path = tmp_path / "define.xml"
+    path.write_text(
+        _define_xml("TA", CROSSOVER_TA, "EPOCH", "EPOCH", CROSSOVER_EPOCHS, {"TREATMENT 1"})
+    )
+    # A Define-XML with one dataset gives that dataset's metadata
+    codelist = _read_define_xml_metadata(str(path)).codelists["EPOCH"]
+
+    assert codelist.extended_values() == ["TREATMENT 1"]

user_guide/11-metadata-import/04-cdisc-submission-conformance.qmd

@@ -258,6 +258,13 @@ marked `not_applicable`. Supplying the file allows the engine to verify that eve
 in your datasets is declared in Define-XML with the correct type and, where applicable, a valid
 codelist.
 
+Define-XML also tells the CT checks about sponsor extensions. Many CDISC codelists are extensible
+(EPOCH, FREQ, ROUTE, and UNIT among them): a sponsor may add terms, provided Define-XML declares
+them with `def:ExtendedValue="Yes"` on the codelist item. When `define_xml` is supplied, such
+declared terms pass the codelist checks for extensible codelists. For example, a crossover study can
+use the epochs `TREATMENT 1` and `TREATMENT 2`. Without Define-XML, or for a codelist that isn't
+extensible (such as SEX), terms outside the codelist are still reported.
+
 ### Loading XPT Files
 
 For datasets stored as SAS Transport (XPT) files, use `pyreadstat` to load them before passing to