src/workflows/lineage.py

"""Lineage filter — select the pancreatic subset of a pan-cancer source.

Spike caveat 2: CCLE (`ccle_broad_2019`) is 1,739 cell lines across every lineage, but
Carl's intent is *pancreatic* lines. cBioPortal exposes lineage via sample clinical
attributes (`CANCER_TYPE` / `ONCOTREE_CODE`), so the pancreatic subset is a clinical-data
filter, NOT a separate study. Registered PDAC tumor cohorts (`paad_tcga`) are already
pancreatic end-to-end and need no filter — `needs_lineage_filter()` says which is which.
"""
from __future__ import annotations

from . import cbioportal_io, curated_store

# Pan-cancer sources that must be narrowed to their pancreatic subset before use.
PAN_CANCER_STUDIES = {"ccle_broad_2019"}

# Clinical-attribute values (any-case substring / exact) that mark a pancreatic sample.
_CANCER_TYPE_MATCH = "pancreatic"  # CANCER_TYPE == "Pancreatic Cancer"
_ONCOTREE_PANC = {"PAAD", "PAAC", "PANET", "UCP", "SPN", "PB", "PAASC", "ACCA"}  # OncoTree pancreas codes


def needs_lineage_filter(study_id: str) -> bool:
    """True for pan-cancer sources (CCLE); False for registered PDAC cohorts."""
    return study_id in PAN_CANCER_STUDIES


def _attribute(study_id: str, attribute_id: str) -> dict[str, str]:
    """`{sampleId: value}` for a clinical attribute — from the curated artifact if there is one.

    ADR-0005 C4: the request path must not call the API. A curated study carries its clinical
    attributes in the artifact, so the lineage filter resolves offline. The live call remains
    only for curation time (and for tests that patch the client).
    """
    cached = curated_store.clinical_values(study_id, attribute_id)
    if cached is not None:
        return dict(cached)
    return {
        r["sampleId"]: (r.get("value") or "")
        for r in cbioportal_io.clinical_data(study_id, [attribute_id])
    }


def pancreatic_sample_ids(study_id: str) -> list[str]:
    """Sample ids in `study_id` whose clinical lineage is pancreatic.

    Prefers `CANCER_TYPE`; falls back to `ONCOTREE_CODE`. Returns them sorted so the
    curated subset is deterministic (fixture/CI stability). Raises if neither attribute
    resolves any pancreatic sample — better to refuse than silently return the whole
    pan-cancer panel.
    """
    cancer_type = _attribute(study_id, "CANCER_TYPE")
    selected = {s for s, v in cancer_type.items() if _CANCER_TYPE_MATCH in (v or "").lower()}

    if not selected:  # fall back to OncoTree code
        for s, v in _attribute(study_id, "ONCOTREE_CODE").items():
            if (v or "").upper() in _ONCOTREE_PANC:
                selected.add(s)

    if not selected:
        raise ValueError(
            f"No pancreatic samples resolved for {study_id!r} via CANCER_TYPE/ONCOTREE_CODE — "
            "refusing to return the full pan-cancer cohort."
        )
    return sorted(selected)