"""B1: validate a per-case extraction against the confirmed A2 schema (#97). The per-case extraction (ColPali + rules + LLM, #99) populates one value per concept. This pass reads that extraction and the confirmed concept list (#91/#95) and, for each concept, emits a validation block: - a **type** check (the value is the concept's primitive type), - an **enum** check (a categorical value is one of the concept's registered codes), - a **cardinality** check (a single-valued concept does not carry a list of values), and - for a derived concept, a **derivation recompute**: the concept is recomputed from its input concepts by its named rule (#35), and the result is compared to the extracted value. A field whose block passes is promoted from needs-review to `confirmed_eligible`. A person still confirms it: no value counts as data without human confirmation (docs/prd.md 8.3). A field whose block fails stays `needs_review` with the reason named, never silently. Two facts the compiled concept (`mapping.CompiledConcept`) does not carry are supplied here: - the **value shape** (`_VALUE_SHAPES`): the primitive type and, for a categorical concept, the closed set of codes. It is hand-transcribed from `docs/taxonomy.md` section 1 (asserted values) and section 2 (derived values), the same way `mapping._ENRICHMENT` transcribes the three facts the dependency graph omits, and `value_shapes()` validates it covers exactly the compiled concept set so a concept added to the registry fails until its shape is supplied. - the **recompute** of a derived concept (`_RECOMPUTE`): the concept-keyed map from `endopath.taxonomy.RECOMPUTE_BY_CONCEPT`. The rule logic lives in the taxonomy registry (#35), not here: this pass owns no formula. `ratio_percent` (the myometrial and cervical ratios), the myometrial-invasion-category threshold, `grade_binary`, and `histotype_aggressiveness` recompute from a record's dependency-graph inputs alone and so are in that map. `lvsi_extent` needs a coding system (the substantial cut-off is 3, 4, or 5 vessels by system) and `figo_stage` needs a FIGO edition (a bare stage code is edition-keyed); neither is carried by the editionless per-case record, so they are projected through `taxonomy.derive`/`taxonomy.derive_stage` with that context rather than recomputed here, and their block reports `rule_not_implemented` in this pass so a value that depends on them is not promoted on an unverified derivation. The two invariants this pass serves: 1. Numbers stay numbers. A derived category (`myometrial_invasion_category`) is recomputed from the primitive (`myometrial_invasion_percent`, itself recomputed from the two millimetre quantities), never trusted as extracted. A category that disagrees with its recomputed value fails validation. 2. No value without its evidence. The extracted evidence span is carried through the block for the confirmation card (#98) to render and for #49 to resolve to a page region. A value that is absent must name a coded reason (docs/prd.md 4.3); a missing value with no reason fails, because silence is never the response to a miss. """ from __future__ import annotations import math from dataclasses import dataclass from enum import Enum from typing import Any, Callable, Iterable, Optional from pydantic import BaseModel, ConfigDict, Field from endopath import taxonomy from endopath.mapping import ( AssertionType, Cardinality, CompiledConcept, compile_concepts, ) from endopath.schema import EvidenceSpan class ValidationConfigError(ValueError): """The validator is misconfigured against the registry: the value-shape table and the compiled concept list have drifted, or a record names a concept the schema does not carry.""" # ------------------------------------------------------------------ value shapes class ValueType(str, Enum): """The primitive type of a concept's value, mirroring the JSON-schema types `src/endopath/fields.py` already uses for the CAP checklist.""" NUMBER = "number" INTEGER = "integer" STRING = "string" BOOLEAN = "boolean" @dataclass(frozen=True) class ValueShape: """The type and, for a categorical concept, the closed code set a value must satisfy. `enum` is `None` for a quantity (a number carries no value set) and for the concepts whose value set `docs/taxonomy.md` evidences in prose rather than as an explicit closed list (the raw POLE assay, and the edition-keyed FIGO stage code); enum membership is then reported `not_applicable` rather than checked against a fabricated set. """ value_type: ValueType enum: Optional[frozenset[str]] = None minimum: Optional[float] = None maximum: Optional[float] = None # concept_id -> value shape, transcribed from docs/taxonomy.md sections 1 and 2. The comment on # each block cites the section. `value_shapes()` checks this covers exactly the compiled concept # set, so a concept added to the dependency graph fails the pass until its shape lands here. _VALUE_SHAPES: dict[str, ValueShape] = { # --- Histotype (section 1.1, 1.2; section 2) --------------------------------------- "histologic_type": ValueShape( ValueType.STRING, frozenset( { "endometrioid", "serous", "clear_cell", "undifferentiated", "dedifferentiated", "carcinosarcoma", "mesonephric_like", "squamous_cell", "gastric_type", "small_cell_neuroendocrine", "large_cell_neuroendocrine", "mixed", "other_nos", } ), ), "figo_grade": ValueShape(ValueType.STRING, frozenset({"1", "2", "3"})), "grade_binary": ValueShape(ValueType.STRING, frozenset({"low", "high"})), "histotype_aggressiveness": ValueShape( ValueType.STRING, frozenset({"non_aggressive", "aggressive", "indeterminate"}) ), # --- Myometrial and cervical invasion (section 1.3, 1.4; section 2) ----------------- # Numbers stay numbers: the millimetre quantities are numbers, the percentage and the # category derive from them. "invasion_depth": ValueShape(ValueType.NUMBER, minimum=0.0), "myometrial_thickness": ValueShape(ValueType.NUMBER, minimum=0.0), "myometrial_invasion_percent": ValueShape(ValueType.NUMBER, minimum=0.0, maximum=100.0), "myometrial_invasion_category": ValueShape( ValueType.STRING, frozenset({"none", "less_than_half", "half_or_more"}) ), "cervical_stromal_invasion": ValueShape( ValueType.STRING, frozenset({"present", "absent", "cannot_be_assessed"}) ), "cervical_invasion_depth": ValueShape(ValueType.NUMBER, minimum=0.0), "cervical_wall_thickness": ValueShape(ValueType.NUMBER, minimum=0.0), "cervical_wall_percent": ValueShape(ValueType.NUMBER, minimum=0.0, maximum=100.0), # --- Lymphovascular space invasion (section 1.5; section 2, 5) ---------------------- "lvsi_status": ValueShape( ValueType.STRING, frozenset({"present", "not_identified", "suspicious", "cannot_be_determined"}), ), "lvsi_foci_count": ValueShape(ValueType.INTEGER, minimum=0), "lvsi_extent": ValueShape(ValueType.STRING, frozenset({"focal", "substantial"})), # --- Regional lymph nodes (section 1.6) -------------------------------------------- "pn_category": ValueShape( ValueType.STRING, frozenset({"pN0", "pN0_i_plus", "pN1mi", "pN1a", "pN2mi", "pN2a"}), ), # --- Molecular (section 1.7) ------------------------------------------------------- # The raw POLE assay has no explicit closed value set in section 1.7 (the section states the # ProMisE precedence in prose): enum is left open rather than fabricated. MMR and p53 read # into the explicit sets the interpretation notes give. "pole_exonuclease_mutation": ValueShape(ValueType.STRING), "mmr_ihc": ValueShape(ValueType.STRING, frozenset({"intact", "lost", "subclonal_loss"})), "p53_ihc": ValueShape(ValueType.STRING, frozenset({"wild_type", "abnormal"})), "promise_class": ValueShape( ValueType.STRING, frozenset({"pole_mutated", "mmr_deficient", "p53_abnormal", "nsmp", "double_classifier"}), ), "tcga_class": ValueShape( ValueType.STRING, frozenset( {"pole_ultramutated", "msi_hypermutated", "copy_number_low", "copy_number_high"} ), ), # --- Stage (section 4) ------------------------------------------------------------- # A FIGO stage code is edition-keyed (`1988.IB` and `2009.IB` are different concepts, # section 4), so no single closed set covers it: enum is left open. "figo_stage": ValueShape(ValueType.STRING), } def value_shapes() -> dict[str, ValueShape]: """The value shape of every compiled concept, validated to cover exactly the concept set. Raises ValidationConfigError if the table and the compiled concept list have drifted, the same drift guard `mapping.compile_concepts` runs over its enrichment table. """ concept_ids = {concept.concept_id for concept in compile_concepts()} missing = sorted(concept_ids - set(_VALUE_SHAPES)) extra = sorted(set(_VALUE_SHAPES) - concept_ids) if missing or extra: raise ValidationConfigError( "value-shape table and the compiled concept list disagree on the concept set: " f"missing shape for {missing}; shape names unknown concepts {extra}" ) return dict(_VALUE_SHAPES) # ------------------------------------------------------------------ recompute of derived concepts # concept_id -> the recompute of its named rule, owned by the taxonomy registry (#35) and imported # rather than duplicated here. `ratio_percent` is bound twice (once per organ), so the map is keyed by # concept. `histotype_aggressiveness` recomputes from a record's inputs alone and is verified here now. # `lvsi_extent` (needs a coding system) and `figo_stage` (needs a FIGO edition) carry a context the # per-case record does not, so they are absent from this map and report `rule_not_implemented`; a caller # projects them through `taxonomy.derive`/`taxonomy.derive_stage`. _RECOMPUTE: dict[str, Callable[..., Optional[Any]]] = dict(taxonomy.RECOMPUTE_BY_CONCEPT) # ------------------------------------------------------------------ the extraction record contract class FieldState(str, Enum): """A field is never simply present or blank (CLAUDE.md, the field-state corollary; docs/prd.md 4.3). It carries a concrete value, or a coded reason it does not.""" PRESENT = "present" NOT_APPLICABLE = "not_applicable" NOT_STATED = "not_stated" INDETERMINATE = "indeterminate" DETERMINED_BY_JOIN = "determined_by_join" CONSTRAINED = "constrained" _ABSENCE_STATES = frozenset(state for state in FieldState if state is not FieldState.PRESENT) class ExtractedValue(BaseModel): """One concept's extraction: the value the pipeline drafted (or a coded absence), and the evidence span that justified it. This is the per-case extraction contract B1 consumes (#99 produces it) and the row the validation block is attached to.""" model_config = ConfigDict(frozen=True) concept_id: str value: Optional[Any] = None state: FieldState = FieldState.PRESENT evidence: Optional[EvidenceSpan] = None # For a CONSTRAINED field: the set the evidence narrowed the answer to, no further. candidates: Optional[list[str]] = None # ------------------------------------------------------------------ the validation block class CheckName(str, Enum): TYPE = "type" ENUM = "enum" CARDINALITY = "cardinality" STATE = "state" class CheckStatus(str, Enum): PASS = "pass" FAIL = "fail" NOT_APPLICABLE = "not_applicable" class DerivationStatus(str, Enum): """The result of recomputing a derived concept and comparing it to the extracted value.""" NOT_DERIVED = "not_derived" # the concept is asserted; nothing to recompute MATCHED = "matched" # recomputed value agrees with the extracted value MISMATCHED = "mismatched" # recomputed value disagrees: the extracted value is not trusted INPUTS_ABSENT = "inputs_absent" # a required input is missing, so the rule cannot recompute RULE_NOT_IMPLEMENTED = "rule_not_implemented" # the named rule is not recomputed here (#35) NOT_CHECKED = "not_checked" # the value is a coded absence; no derivation to recompute class Promotion(str, Enum): """Where the validation block leaves the field. `confirmed_eligible` is promoted from needs-review toward confirmation; a person still confirms it before it counts as data.""" CONFIRMED_ELIGIBLE = "confirmed_eligible" NEEDS_REVIEW = "needs_review" class Check(BaseModel): model_config = ConfigDict(frozen=True) name: CheckName status: CheckStatus detail: str = "" class Derivation(BaseModel): model_config = ConfigDict(frozen=True) status: DerivationStatus rule: Optional[str] = None inputs: tuple[str, ...] = () recomputed: Optional[Any] = None extracted: Optional[Any] = None detail: str = "" class FieldValidation(BaseModel): """The per-field validation block: the shape checks, the derivation recompute, and the promotion the block earns. This is the data the confirmation card (#98) renders beside the value.""" model_config = ConfigDict(frozen=True) concept_id: str assertion_type: AssertionType cardinality: Cardinality state: FieldState value: Optional[Any] = None evidence: Optional[EvidenceSpan] = None checks: list[Check] = Field(default_factory=list) derivation: Derivation promotion: Promotion reasons: list[str] = Field(default_factory=list) @property def passed(self) -> bool: return self.promotion is Promotion.CONFIRMED_ELIGIBLE class ValidationRecord(BaseModel): """The whole per-case extraction after validation: one block per validated concept.""" model_config = ConfigDict(frozen=True) fields: dict[str, FieldValidation] @property def all_confirmed_eligible(self) -> bool: return bool(self.fields) and all(block.passed for block in self.fields.values()) def promoted(self) -> list[str]: return [cid for cid, block in self.fields.items() if block.passed] def _is_multivalued(value: Any) -> bool: return isinstance(value, (list, tuple, set)) def _type_ok(value: Any, value_type: ValueType) -> bool: # bool is a subclass of int in Python; keep the two kinds distinct. if value_type is ValueType.BOOLEAN: return isinstance(value, bool) if value_type is ValueType.INTEGER: return isinstance(value, int) and not isinstance(value, bool) if value_type is ValueType.NUMBER: return isinstance(value, (int, float)) and not isinstance(value, bool) return isinstance(value, str) def _values_agree(recomputed: Any, extracted: Any, shape: ValueShape) -> bool: if shape.value_type in (ValueType.NUMBER, ValueType.INTEGER): try: return math.isclose(float(recomputed), float(extracted), rel_tol=1e-6, abs_tol=0.5) except (TypeError, ValueError): return False return recomputed == extracted def _shape_checks(value: Any, concept: CompiledConcept, shape: ValueShape) -> list[Check]: """The type, enum, and cardinality checks for a present value.""" checks: list[Check] = [] multivalued = _is_multivalued(value) # Cardinality: a single-valued concept must not carry a list of values. if concept.cardinality is Cardinality.ZERO_OR_MORE: checks.append(Check(name=CheckName.CARDINALITY, status=CheckStatus.PASS)) elif multivalued: checks.append( Check( name=CheckName.CARDINALITY, status=CheckStatus.FAIL, detail=( f"cardinality {concept.cardinality.value} allows one value, " f"but {len(list(value))} were extracted" ), ) ) else: checks.append(Check(name=CheckName.CARDINALITY, status=CheckStatus.PASS)) # Type: checked on the scalar. When cardinality already failed on a list, type and enum are # not applicable rather than run against the collection. if multivalued and concept.cardinality is not Cardinality.ZERO_OR_MORE: checks.append(Check(name=CheckName.TYPE, status=CheckStatus.NOT_APPLICABLE)) checks.append(Check(name=CheckName.ENUM, status=CheckStatus.NOT_APPLICABLE)) return checks scalar = value type_ok = _type_ok(scalar, shape.value_type) if not type_ok: checks.append( Check( name=CheckName.TYPE, status=CheckStatus.FAIL, detail=f"expected {shape.value_type.value}, got {type(scalar).__name__}", ) ) else: detail = "" if shape.minimum is not None and scalar < shape.minimum: type_ok = False detail = f"below minimum {shape.minimum}" elif shape.maximum is not None and scalar > shape.maximum: type_ok = False detail = f"above maximum {shape.maximum}" checks.append( Check( name=CheckName.TYPE, status=CheckStatus.PASS if type_ok else CheckStatus.FAIL, detail=detail, ) ) # Enum: only when the concept registers a closed set and the type held. if shape.enum is None: checks.append( Check( name=CheckName.ENUM, status=CheckStatus.NOT_APPLICABLE, detail="concept registers no closed value set", ) ) elif not type_ok: checks.append(Check(name=CheckName.ENUM, status=CheckStatus.NOT_APPLICABLE)) elif scalar in shape.enum: checks.append(Check(name=CheckName.ENUM, status=CheckStatus.PASS)) else: checks.append( Check( name=CheckName.ENUM, status=CheckStatus.FAIL, detail=f"{scalar!r} is not one of {sorted(shape.enum)}", ) ) return checks def _recompute_derivation( concept: CompiledConcept, extracted_value: Any, record: dict[str, ExtractedValue], shape: ValueShape, ) -> Derivation: """Recompute a derived concept from its inputs and compare to the extracted value.""" rule = concept.derived.rule if concept.derived else None inputs = concept.derived.inputs if concept.derived else () fn = _RECOMPUTE.get(concept.concept_id) if fn is None: return Derivation( status=DerivationStatus.RULE_NOT_IMPLEMENTED, rule=rule, inputs=inputs, extracted=extracted_value, detail=( f"rule {rule!r} is implemented in endopath.taxonomy but needs a projection context " "this per-case pass does not carry (a coding system for lvsi_extent, a FIGO edition " "for figo_stage); project it through taxonomy.derive/derive_stage instead" ), ) input_values: list[Any] = [] for input_id in inputs: entry = record.get(input_id) if entry is None or entry.state is not FieldState.PRESENT or entry.value is None: return Derivation( status=DerivationStatus.INPUTS_ABSENT, rule=rule, inputs=inputs, extracted=extracted_value, detail=f"input {input_id!r} is absent, so the derivation cannot be recomputed", ) input_values.append(entry.value) recomputed = fn(*input_values) if recomputed is None: return Derivation( status=DerivationStatus.INPUTS_ABSENT, rule=rule, inputs=inputs, extracted=extracted_value, detail="inputs present but the rule could not compute a value (e.g. zero denominator)", ) agrees = _values_agree(recomputed, extracted_value, shape) return Derivation( status=DerivationStatus.MATCHED if agrees else DerivationStatus.MISMATCHED, rule=rule, inputs=inputs, recomputed=recomputed, extracted=extracted_value, detail=( "" if agrees else f"extracted {extracted_value!r} disagrees with recomputed {recomputed!r}" ), ) def validate_field( extracted: ExtractedValue, concept: CompiledConcept, shape: ValueShape, record: dict[str, ExtractedValue], ) -> FieldValidation: """Validate one extracted concept against its confirmed shape and, if derived, its recompute.""" checks: list[Check] = [] reasons: list[str] = [] is_derived = concept.assertion_type is AssertionType.DERIVED # Coded absence: a value is either present or names a reason it is not (silence is never a # miss). Shape and derivation checks do not apply to a coded absence. if extracted.state in _ABSENCE_STATES: if extracted.value is not None: checks.append( Check( name=CheckName.STATE, status=CheckStatus.FAIL, detail=f"state {extracted.state.value} carries a value {extracted.value!r}", ) ) reasons.append("a coded absence must not carry a value") else: checks.append(Check(name=CheckName.STATE, status=CheckStatus.PASS)) derivation = Derivation( status=DerivationStatus.NOT_CHECKED, rule=concept.derived.rule if concept.derived else None, inputs=concept.derived.inputs if concept.derived else (), detail="value is a coded absence; no value to recompute", ) promotion = Promotion.CONFIRMED_ELIGIBLE if not reasons else Promotion.NEEDS_REVIEW return FieldValidation( concept_id=concept.concept_id, assertion_type=concept.assertion_type, cardinality=concept.cardinality, state=extracted.state, value=None, evidence=extracted.evidence, checks=checks, derivation=derivation, promotion=promotion, reasons=reasons, ) # State PRESENT but no value: silence with no coded reason. if extracted.value is None: checks.append( Check( name=CheckName.STATE, status=CheckStatus.FAIL, detail="value is absent but names no coded reason (section 4.3)", ) ) reasons.append( "This value is blank and no reason is recorded. Enter a value, or mark it not stated, " "not applicable, or indeterminate." ) derivation = Derivation( status=DerivationStatus.NOT_DERIVED if not is_derived else DerivationStatus.INPUTS_ABSENT, rule=concept.derived.rule if concept.derived else None, inputs=concept.derived.inputs if concept.derived else (), ) return FieldValidation( concept_id=concept.concept_id, assertion_type=concept.assertion_type, cardinality=concept.cardinality, state=extracted.state, value=None, evidence=extracted.evidence, checks=checks, derivation=derivation, promotion=Promotion.NEEDS_REVIEW, reasons=reasons, ) # A present, concrete value: run the shape checks. checks.extend(_shape_checks(extracted.value, concept, shape)) shape_ok = all(check.status is not CheckStatus.FAIL for check in checks) for check in checks: if check.status is CheckStatus.FAIL: reasons.append(f"{check.name.value} check failed: {check.detail}") # The derivation recompute for a derived concept. if is_derived: derivation = _recompute_derivation(concept, extracted.value, record, shape) if derivation.status is DerivationStatus.MISMATCHED: reasons.append(f"derivation disagrees: {derivation.detail}") elif derivation.status is DerivationStatus.INPUTS_ABSENT: reasons.append(f"derivation unverified: {derivation.detail}") elif derivation.status is DerivationStatus.RULE_NOT_IMPLEMENTED: reasons.append(f"derivation unverified: {derivation.detail}") derivation_ok = derivation.status is DerivationStatus.MATCHED else: derivation = Derivation(status=DerivationStatus.NOT_DERIVED) derivation_ok = True promotion = ( Promotion.CONFIRMED_ELIGIBLE if (shape_ok and derivation_ok) else Promotion.NEEDS_REVIEW ) return FieldValidation( concept_id=concept.concept_id, assertion_type=concept.assertion_type, cardinality=concept.cardinality, state=extracted.state, value=extracted.value, evidence=extracted.evidence, checks=checks, derivation=derivation, promotion=promotion, reasons=reasons, ) def _as_record(record: Iterable[ExtractedValue] | dict[str, ExtractedValue]) -> dict[str, ExtractedValue]: if isinstance(record, dict): return dict(record) by_id: dict[str, ExtractedValue] = {} for extracted in record: by_id[extracted.concept_id] = extracted return by_id def validate_extraction( record: Iterable[ExtractedValue] | dict[str, ExtractedValue], *, concepts: Optional[list[CompiledConcept]] = None, shapes: Optional[dict[str, ValueShape]] = None, ) -> ValidationRecord: """Validate a per-case extraction against the confirmed A2 schema. `concepts` defaults to the whole compiled concept list (`mapping.compile_concepts`); pass the confirmed in-scope subset to restrict validation to it (see `confirmed_concept_ids`). Every concept the record names must be in the schema, or the record is invalid. """ by_id = _as_record(record) concepts = concepts if concepts is not None else compile_concepts() shapes = shapes if shapes is not None else value_shapes() concept_by_id = {concept.concept_id: concept for concept in concepts} unknown = sorted(set(by_id) - set(concept_by_id)) if unknown: raise ValidationConfigError( f"extraction names concepts not in the confirmed schema: {unknown}" ) fields: dict[str, FieldValidation] = {} for concept_id, extracted in by_id.items(): concept = concept_by_id[concept_id] shape = shapes.get(concept_id) if shape is None: raise ValidationConfigError(f"no value shape registered for concept {concept_id!r}") fields[concept_id] = validate_field(extracted, concept, shape, by_id) return ValidationRecord(fields=fields) def confirmed_concept_ids(field_set: Iterable[dict]) -> frozenset[str]: """The distinct concepts a confirmed mapping (`mapping_review.mapping_review(...)['field_set']`, #95) resolves its source fields to. A caller passes the compiled concepts filtered to these, plus their derivation inputs, as the confirmed in-scope schema for `validate_extraction`.""" return frozenset( row["concept_id"] for row in field_set if row.get("concept_id") is not None )