src/endopath/projection.py

"""Project one confirmed extraction into a chosen target coding system (#40, T14).

The payoff of the taxonomy layer: one extraction, many target schemas, each with a stated confidence and
an auditable derivation. This module builds the canonical assertions from a confirmed checklist and
projects the FIGO stage under a chosen edition through `taxonomy.derive_stage` (#35), returning the
verdict in serializable form: a determined code, a `constrained` candidate set the answer narrows to, or
an `indeterminate` reason. Collapsing a constrained answer to one stage in the export would be the exact
silent-confident-error the product exists to prevent, so the three render distinctly.
"""

from __future__ import annotations

from typing import Any, Optional

from endopath import taxonomy

# The staging editions the selector offers and the projection targets. FIGO 2009 and 2023 only: FIGO 1988
# is a superseded edition the product does not present as an output (the reporting pathologist's direction,
# #132). 2023 adds the molecular modifiers. Recognizing that a legacy report was originally staged under
# FIGO 1988 is a separate concern (edition inference and the corpus provenance), not an offered target.
STAGING_EDITIONS: list[dict[str, str]] = [
    {"id": "figo_endo/2009", "label": "FIGO 2009"},
    {"id": "figo_endo/2023", "label": "FIGO 2023"},
]
DEFAULT_EDITION = "figo_endo/2009"

# A case's stored FigoStagingEdition value ("figo_2009") named as the registry coding-system id the
# projection reads ("figo_endo/2009"). The selector already sends the coding-system id, and "2009" is
# accepted directly, so those pass through unchanged.
_STAGING_EDITION_ALIASES = {
    "figo_2009": "figo_endo/2009",
    "figo_2023": "figo_endo/2023",
}


def edition_id(edition: Any) -> str:
    """The registry coding-system id for a stored staging edition, an alias, or an id passed through."""
    return _STAGING_EDITION_ALIASES.get(str(edition), edition)


def is_offered_edition(edition: Any) -> bool:
    """Whether a staging edition is one the product offers as an output target (`STAGING_EDITIONS`). The
    request path checks this so a superseded edition removed from the offered set (FIGO 1988, #132) is
    rejected rather than silently projected: `derive_stage` still routes FIGO 2009 through the shared
    1988/2009 algorithm, so without this gate a hand-crafted `figo_endo/1988` request would derive a stage."""
    return edition_id(edition) in {e["id"] for e in STAGING_EDITIONS}

# The checklist fields whose confirmed values feed the stage. The projection confidence is the least
# confident of the ones that contributed, so a stage derived from a low-confidence grade reads low.
_STAGE_INPUT_FIELDS = (
    "histologic_type",
    "histologic_grade",
    "myometrial_invasion_percent",
    "cervical_stromal_invasion",
    "lymphovascular_space_invasion",
    "molecular_classification",
)


def _primitive(value: object) -> object:
    return value.value if hasattr(value, "value") else value


def assertions_from_checklist(checklist) -> dict[str, Any]:
    """Canonical assertions for a FIGO projection, from the confirmed checklist. Only a stated value is
    passed, so an unstated finding stays unstated rather than defaulting to a value the report never made
    (an unstated cervical or nodal finding is read as not involved inside `derive_stage`, not here)."""
    ck = checklist
    assertions: dict[str, Any] = {}

    histotype = _primitive(ck.histologic_type.value)
    if histotype is not None:
        assertions["histologic_type"] = histotype
    grade = _primitive(ck.histologic_grade.value)
    if grade is not None:
        assertions["figo_grade"] = grade
    percent = _primitive(ck.myometrial_invasion_percent.value)
    if percent is not None:
        assertions["myometrial_invasion_percent"] = percent

    cervical = _primitive(ck.cervical_stromal_invasion.value)
    if cervical is not None:
        assertions["cervical_stromal_invasion"] = "present" if cervical else "absent"

    lvsi = _primitive(ck.lymphovascular_space_invasion.value)
    if lvsi is not None:
        assertions["lvsi_extent"] = {"not_identified": "negative"}.get(lvsi, lvsi)

    molecular = _primitive(ck.molecular_classification.value)
    if molecular is not None:
        assertions["promise_class"] = {"pole_ultramutated": "pole_mutated"}.get(molecular, molecular)

    return assertions


def _min_confidence(checklist) -> Optional[float]:
    confidences = []
    for name in _STAGE_INPUT_FIELDS:
        field = getattr(checklist, name, None)
        if field is not None and field.value is not None and field.confidence is not None:
            confidences.append(field.confidence)
    return min(confidences) if confidences else None


def project_stage(checklist, edition: Any) -> dict:
    """Project the FIGO stage under `edition`, as a serializable verdict. `kind` is `determined`,
    `constrained`, or `indeterminate`; the payload carries the code or the candidate set or the reason,
    the projection confidence, and the derivation reference (the rule, the edition, and the assertions it
    read), so every projected value is auditable back to what produced it."""
    edition = edition_id(edition)
    assertions = assertions_from_checklist(checklist)
    system_id = taxonomy._system_id(edition)
    verdict = taxonomy.derive_stage(assertions, edition)
    derivation = {"rule": "derive_stage", "edition": system_id, "inputs": sorted(assertions.keys())}
    confidence = _min_confidence(checklist)

    if isinstance(verdict, taxonomy.Stage):
        return {
            "kind": "determined",
            "code": verdict.code,
            "modifier": verdict.modifier,
            "system_id": verdict.system_id,
            "confidence": confidence,
            "derivation": derivation,
        }
    if isinstance(verdict, taxonomy.Constrained):
        return {
            "kind": "constrained",
            "candidates": list(verdict.candidates),
            "reason": verdict.reason,
            "system_id": verdict.system_id or system_id,
            "confidence": confidence,
            "derivation": derivation,
        }
    return {
        "kind": "indeterminate",
        "reason": verdict.reason,
        "system_id": verdict.system_id or system_id,
        "confidence": confidence,
        "derivation": derivation,
    }


def stage_cell(verdict: dict) -> tuple[Optional[str], Optional[str]]:
    """The (value, absent_reason) a projected stage occupies in the flat export: a determined code (with
    its molecular modifier appended), the candidate set joined for a constrained answer under a
    `constrained` reason, or an empty value under an `indeterminate` reason. The three read distinctly in
    the CSV, never collapsed to one confident stage."""
    if verdict["kind"] == "determined":
        code = verdict["code"]
        if verdict.get("modifier"):
            code = f"{code}m{verdict['modifier']}"
        return code, None
    if verdict["kind"] == "constrained":
        return "|".join(verdict["candidates"]), "constrained"
    return None, "indeterminate"