"""Project the confirmed case records onto the canonical concept list and render the cohort export bundle (#101, docs/prd.md sections 3, 8.1). The column spec is the canonical concept list (`mapping.concept_list()`, #91), not the built-in CAP checklist: the export speaks concepts, so a second laboratory targeting the same concepts re-derives the same table. `mapping.cap_checklist_coverage()` (#95) is the bridge that says which confirmed CAP field fills which concept column. A concept no CAP field covers, and no derivation reaches, is absent, and an absent value is never a blank: it carries a coded reason in a parallel `__absent_reason` column, from the field-state vocabulary docs/prd.md section 4.3 defines (`not_applicable`, `not_stated`, `indeterminate`, `determined_by_join`, `constrained`). The two invariants this projection serves: 1. Numbers stay numbers. The stored quantity survives into its own column, and a derived category sits beside it in a separate column, recomputed from that quantity by the registry's named rule (`taxonomy.recompute`, #35), never stored in its place. A confirmed `myometrial_invasion_percent` of 60.0 and the `half_or_more` category it derives to are two columns, so the export can still answer the other convention's question. 2. No value without its evidence. The field-level provenance carries each value's verbatim evidence quote, its page, and the pass that read it, and the bundle carries a checksum over every cited span per case, so a second centre auditing the row reads the same source material. The bundle carries the schema the second centre re-derives against (the concept list and its checksum), the version of the rule set and the extraction model that touched the values, the staging edition each stage speaks (docs/prd.md section 4.2), and a per-case source-document reference with an evidence checksum. The rendered CSV, XLSX, and JSON come from one projection, so the three files and the on-screen preview never disagree. ## Relates to - #40 owns the target-schema selector and the `derive_stage` set projection: choosing an arbitrary confirmed dictionary as the target, and rendering a `constrained` stage as the candidate set with the variable whose absence blocks narrowing. This module names the staging edition it emitted and reserves the `constrained` code for that projection; it does not itself enumerate the candidate set. - #39 owns the deeper export provenance: the byte-level checksum of each scanned report and the positional region of every evidence span. This module carries the source-document reference and a checksum over the cited spans, the reproducible provenance available from the confirmed record alone. - #35 owns the per-value coding-system resolution beyond the FIGO edition; this module cites each concept to the registry via its `cited_to` in the column spec. """ from __future__ import annotations import base64 import csv import hashlib import io import json from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Optional from sqlalchemy import Connection, text from endopath import llm_extraction, longitudinal, mapping, projection, taxonomy from endopath.mapping import AssertionType, CompiledConcept from endopath.schema import ChecklistField, FieldStatus from endopath.validation import FieldState # The coded-absence reason applied to a concept the confirmed record does not carry. A molecular finding # absent from a report of this era arrives from the clinical record rather than the report, so it is # `determined_by_join` (docs/prd.md sections 4.3, 8.3); every other uncovered concept, and a derived # concept whose input primitives are absent, is `not_stated`, the searched-and-absent code. _JOIN_SOURCED_CATEGORY = "molecular" ABSENT_REASON_SUFFIX = "__absent_reason" @dataclass(frozen=True) class _Projected: """One concept's value on one case: a concrete value with no reason, or a coded absence with no value. `origin` records how the value was reached, for the column-spec descriptor.""" value: Optional[Any] absent_reason: Optional[str] origin: str # 'confirmed' | 'derived' | 'absent' def _primitive(value: object) -> object: """An enum to its value, everything else as-is, so the export carries JSON primitives.""" return value.value if hasattr(value, "value") else value def _concept_labels() -> dict[str, str]: """Concept id to human label, read from the dependency-graph nodes the concept ids key on.""" data = json.loads(mapping.DEPENDENCY_GRAPH_PATH.read_text(encoding="utf-8")) return {node["id"]: node["label"] for node in data["nodes"]} def _field_value_and_state(field: ChecklistField) -> tuple[Optional[Any], FieldState]: """Reduce one confirmed CAP field to a value and a field state. A gated field is `not_applicable`; a flagged field was searched for and confirmed absent (`not_stated`); an applicable field with no value is `not_stated`; a field carrying a value is present. Silence is never the response to a miss.""" if not field.applicable: return None, FieldState.NOT_APPLICABLE if field.status is FieldStatus.FLAGGED: return None, FieldState.NOT_STATED value = _primitive(field.value) if value is None: return None, FieldState.NOT_STATED return value, FieldState.PRESENT def _project_case( checklist, *, concepts: list[CompiledConcept], coverage: dict[str, str], ) -> dict[str, _Projected]: """Project one confirmed checklist onto the concept list: seed the covered concepts from their CAP fields, recompute a derived concept whose inputs are all present, and code every remaining concept as a coded absence (invariant: no value without a reason).""" concept_to_field = {concept_id: field_name for field_name, concept_id in coverage.items()} projected: dict[str, _Projected] = {} # Pass 1: every covered concept, from its confirmed CAP field, so a derivation's inputs are available # before any derived concept is filled. for concept in concepts: field_name = concept_to_field.get(concept.concept_id) if field_name is None: continue value, state = _field_value_and_state(getattr(checklist, field_name)) if state is FieldState.PRESENT: projected[concept.concept_id] = _Projected(value, None, "confirmed") else: projected[concept.concept_id] = _Projected(None, state.value, "absent") # Pass 2: the concepts no CAP field covers. for concept in concepts: cid = concept.concept_id if cid in projected: continue if concept.assertion_type is AssertionType.DERIVED and concept.derived is not None: projected[cid] = _derive(concept, projected) else: reason = ( FieldState.DETERMINED_BY_JOIN if concept.category == _JOIN_SOURCED_CATEGORY else FieldState.NOT_STATED ) projected[cid] = _Projected(None, reason.value, "absent") return projected def _derive(concept: CompiledConcept, projected: dict[str, _Projected]) -> _Projected: """Recompute a derived concept from its inputs by its registry-owned rule (`taxonomy.recompute`, #35), the same discipline the B1 validation pass runs: recompute only when every graph input is present, so a derived category is computed from the primitive rather than trusted as stored. A derivation whose input a gate removed inherits `not_applicable`; one whose input is merely absent, or that the per-case pass does not recompute, is `not_stated`.""" assert concept.derived is not None inputs = [projected.get(input_id) for input_id in concept.derived.inputs] if any(item is not None and item.absent_reason == FieldState.NOT_APPLICABLE.value for item in inputs): return _Projected(None, FieldState.NOT_APPLICABLE.value, "absent") if all(item is not None and item.value is not None for item in inputs): recomputed = taxonomy.recompute(concept.concept_id, [item.value for item in inputs]) # type: ignore[union-attr] if recomputed is not None: return _Projected(recomputed, None, "derived") return _Projected(None, FieldState.NOT_STATED.value, "absent") def _evidence_checksum(checklist) -> str: """A SHA-256 over the case's cited evidence: the field, verbatim quote, page, and pass of every value that carries a span, in a canonical order. A second centre confirming the same values against the same reports reads the same spans and computes the same checksum (invariant 2).""" spans = [] for field_name, field in sorted(checklist.all_fields().items()): if field.evidence is not None: spans.append( { "field": field_name, "quote": field.evidence.quote, "page_number": field.evidence.page_number, "source": field.evidence.source, } ) canonical = json.dumps(spans, sort_keys=True, ensure_ascii=False) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() def _columns(concepts: list[CompiledConcept], labels: dict[str, str]) -> list[dict]: """The column spec: the case-level identifiers, then, per concept in concept-list order, a value column and its parallel absent-reason column. The descriptor carries the concept's category, assertion type, cardinality, and registry citation so the bundle and the screen render the spec without re-deriving it.""" columns: list[dict] = [ {"key": "case_barcode", "label": "Case barcode", "kind": "identifier"}, {"key": "report_date", "label": "Report date", "kind": "identifier"}, {"key": "staging_edition", "label": "Staging edition", "kind": "identifier"}, {"key": "status", "label": "Case status", "kind": "identifier"}, ] for concept in concepts: cid = concept.concept_id label = labels.get(cid, cid) columns.append( { "key": cid, "label": label, "kind": "value", "concept_id": cid, "category": concept.category, "assertion_type": concept.assertion_type.value, "cardinality": concept.cardinality.value, "cited_to": concept.cited_to, "derived": concept.derived.model_dump(mode="json") if concept.derived else None, "absent_reason_key": f"{cid}{ABSENT_REASON_SUFFIX}", } ) columns.append( { "key": f"{cid}{ABSENT_REASON_SUFFIX}", "label": f"{label} absent reason", "kind": "absent_reason", "of": cid, } ) columns.extend(_outcome_columns()) return columns def _outcome_columns() -> list[dict]: """The join-sourced outcome columns, appended after the concept columns: the disease outcome joined from the GDC clinical record rather than read from the pathology report (`longitudinal.py`). Each value column is marked `sourced_via: "join"` and paired with its parallel absent-reason column, so the same renderer and the same coded-absence discipline apply as for a concept column. The block keeps the report-concept schema and its checksum meaning "what the report answers".""" columns: list[dict] = [] for spec in longitudinal.OUTCOME_COLUMNS: key = spec["key"] columns.append( { "key": key, "label": spec["label"], "kind": "value", "category": longitudinal.OUTCOME_CATEGORY, "assertion_type": spec["assertion_type"], "sourced_via": "join", "join_source": longitudinal.JOIN_SOURCE, "derived": spec.get("derived"), "note": spec.get("note"), "absent_reason_key": f"{key}{ABSENT_REASON_SUFFIX}", } ) columns.append( { "key": f"{key}{ABSENT_REASON_SUFFIX}", "label": f"{spec['label']} absent reason", "kind": "absent_reason", "of": key, } ) return columns def _load_longitudinal(conn: Connection) -> dict[str, list[longitudinal.Record]]: """Every longitudinal record grouped by participant, in one read (issue #131's batched discipline). The outcome projection joins by participant barcode, and a per-case query would reintroduce the N+1 the cohort export was restructured to avoid. The table holds a few thousand small rows, so one unfiltered read is cheaper than a query per case.""" by_participant: dict[str, list[longitudinal.Record]] = {} rows = conn.execute( text( "SELECT participant_id, record_id, record_type, structured_data FROM longitudinal_records" ) ).mappings().fetchall() for row in rows: data: dict = {} if row["structured_data"]: try: data = json.loads(row["structured_data"]) except (json.JSONDecodeError, TypeError): data = {} by_participant.setdefault(row["participant_id"], []).append( longitudinal.Record( record_type=row["record_type"], record_id=row["record_id"], data=data ) ) return by_participant def _load_longitudinal_joins(conn: Connection) -> dict[str, list[dict]]: """Every reviewer join decision grouped by case barcode, in one read (issue #131's batched discipline). #180 writes these from the Case Review panel: a per-record disposition (`confirmed` / `rejected`) with the confirmer's identity and role. The export and the cohort endpoint read them to drop rejected records and to attribute the confirmed linkages. A record with no row here has not been reviewed, not rejected.""" by_case: dict[str, list[dict]] = {} rows = conn.execute( text( "SELECT case_barcode, record_id, disposition, confirmed_by, role, confirmed_at " "FROM longitudinal_joins" ) ).mappings().fetchall() for row in rows: by_case.setdefault(row["case_barcode"], []).append( { "record_id": row["record_id"], "disposition": row["disposition"], "confirmed_by": row["confirmed_by"], "role": row["role"], "confirmed_at": row["confirmed_at"], } ) return by_case def _rejected_record_ids(case_joins: list[dict]) -> set[str]: """The record ids a reviewer explicitly rejected for this case. The barcode join is deterministic, so a record with no join row is unreviewed, not rejected: only an explicit `rejected` disposition excludes a record from the projection, leaving the default (no decision) projecting exactly as before.""" return {j["record_id"] for j in case_joins if j.get("disposition") == "rejected"} def _bundle( concepts: list[CompiledConcept], sources: list[dict], *, staging_edition: Optional[str], per_case_editions: list[str], ) -> dict: """The provenance bundle (docs/prd.md section 3): the schema the second centre re-derives against and its checksum, the rule-set and extraction-model versions that touched the values, the staging edition each stage speaks, the coding systems, and the per-case source references.""" schema_checksum = hashlib.sha256(mapping.concept_list_json().encode("utf-8")).hexdigest() rule_set_checksum = hashlib.sha256( mapping.DEPENDENCY_GRAPH_PATH.read_bytes() ).hexdigest() named_rules = sorted( { concept.derived.rule for concept in concepts if concept.derived is not None } ) return { "target_schema": { "name": "CAP/ICCR endometrial carcinoma synoptic, projected onto the canonical concept list", "concept_count": len(concepts), "schema_checksum": schema_checksum, }, "staging_edition": { "selected": staging_edition or "as_recorded", "per_case_editions": sorted(set(per_case_editions)), }, "generated_at": datetime.now(timezone.utc).isoformat(), "rule_versions": { "dependency_graph_checksum": rule_set_checksum, "named_rules": named_rules, }, "model_versions": {"extraction": llm_extraction.MODEL}, "coding_systems": { "stage": "FIGO; the edition each stage speaks is the per-case staging_edition column", "value_sets": ( "CAP, ICCR, WHO, and AJCC reference standards; each concept cites its registry source " "in the column spec's cited_to" ), }, "sources": sources, "reproduction": ( "A laboratory holding the same reports and confirming the same values against this concept " "list re-derives an identical table. Every absent value carries a coded reason from " "docs/prd.md section 4.3, so a missing value is never a bare blank." ), } def _provenance(case_barcode: str, checklist) -> list[dict]: """One field-level provenance record per checklist field: the value, its confidence and status, the coded-absence markers, and the verbatim evidence span (docs/prd.md section 3 case record).""" records: list[dict] = [] for field_name, field in checklist.all_fields().items(): evidence = field.evidence records.append( { "case_barcode": case_barcode, "field_name": field_name, "value": _primitive(field.value), "confidence": field.confidence, "status": field.status.value, "applicable": field.applicable, "not_applicable_reason": field.not_applicable_reason, "evidence_quote": evidence.quote if evidence else None, "evidence_char_start": evidence.char_start if evidence else None, "evidence_char_end": evidence.char_end if evidence else None, "evidence_page": evidence.page_number if evidence else None, "evidence_source": evidence.source if evidence else None, } ) return records # The export is the confirmed dataset (#231, decision D1): the API exports only cases a reviewer has # confirmed (or already exported), so the screen's "Confirmed records" title is truthful. build_export # still projects any set when `statuses` is None, for internal callers and the projection tests. EXPORTABLE_STATUSES: tuple[str, ...] = ("confirmed", "exported") def build_export( conn: Connection, *, staging_edition: Optional[str] = None, project: Optional[str] = None, statuses: Optional[tuple[str, ...]] = None, ) -> dict: """Project every case in the store onto the concept list and assemble the cohort export. Returns the column spec, one projected row per case, the field-level provenance, and the reproduction bundle. `staging_edition` names the edition the researcher chose at export (docs/prd.md section 4.2); the base projection carries the edition each case recorded, and #40 owns re-projecting a stage under a chosen edition as a candidate set. `project` scopes the export to one project's cases (issue #149); omitted, every case in the store is projected. `statuses`, when given, keeps only cases in those statuses; the API passes EXPORTABLE_STATUSES so the served export is the confirmed dataset (#231).""" from endopath import storage concepts = mapping.compile_concepts() coverage = mapping.cap_checklist_coverage() labels = _concept_labels() rows: list[dict] = [] provenance: list[dict] = [] sources: list[dict] = [] join_sources: list[dict] = [] per_case_editions: list[str] = [] # The joined GDC outcome records, grouped by participant, in one read so the per-case outcome projection # stays free of the N+1 the cohort export was restructured to avoid (issue #131). longitudinal_by_participant = _load_longitudinal(conn) # The reviewer join decisions (#180), so a rejected record is excluded from the projection and the # confirmed linkages are attributed in the bundle (#217). longitudinal_joins_by_case = _load_longitudinal_joins(conn) # One batched read of the whole cohort rather than a get_case per barcode (issue #131): a per-case # N+1 is ~100s over remote Postgres and 500s the export; this is a few queries. Scoped to the project # when one is named (issue #149). for case, patient_filename in storage.load_all_cases(conn, project=project): if statuses is not None and case.status.value not in statuses: continue # the export is the confirmed dataset (#231): an unreviewed case is not a record projected = _project_case(case.checklist, concepts=concepts, coverage=coverage) row: dict[str, Any] = { "case_barcode": case.case_barcode, "report_date": case.report_date.isoformat() if case.report_date else None, "staging_edition": case.staging_edition.value, "status": case.status.value, } for concept in concepts: item = projected[concept.concept_id] row[concept.concept_id] = item.value row[f"{concept.concept_id}{ABSENT_REASON_SUFFIX}"] = item.absent_reason # Re-project the FIGO stage under the chosen edition (#40). derive_stage returns a determined # code, a constrained candidate set, or an indeterminate reason, each rendered distinctly rather # than collapsed to one confident stage. The raw quantities stay in their own columns (invariant 1). target_edition = staging_edition or case.staging_edition.value stage_verdict = projection.project_stage(case.checklist, target_edition) stage_value, stage_reason = projection.stage_cell(stage_verdict) row["figo_stage"] = stage_value row[f"figo_stage{ABSENT_REASON_SUFFIX}"] = stage_reason row["figo_stage__projection"] = stage_verdict # The join-sourced outcome block: the disease outcome joined from the GDC clinical record by # participant barcode, projected onto the outcome columns (`longitudinal.py`). Every cell is a value # or a coded absence, never a bare blank; a participant the join did not reach reads determined_by_join. participant_id = ( case.case_barcode[:12] if len(case.case_barcode) >= 12 else case.case_barcode ) outcome_records = longitudinal_by_participant.get(participant_id, []) # Drop records a reviewer rejected (#217): a rejection removes the record from the outcome columns # and from the bundle's contributing ids, so a linkage the reviewer disowned has no effect. A record # with no decision is unreviewed, not rejected, and still projects (see `_rejected_record_ids`). case_joins = longitudinal_joins_by_case.get(case.case_barcode, []) rejected = _rejected_record_ids(case_joins) if rejected: outcome_records = [r for r in outcome_records if r.record_id not in rejected] outcomes = longitudinal.project_outcomes(outcome_records) for key in longitudinal.OUTCOME_COLUMN_KEYS: cell = outcomes[key] row[key] = cell.value row[f"{key}{ABSENT_REASON_SUFFIX}"] = cell.absent_reason source_entry = { "case_barcode": case.case_barcode, "participant_id": participant_id, "record_ids": longitudinal.contributing_record_ids(outcome_records), } # The reviewer-attributed provenance behind the join-sourced columns (#176, #217): who decided each # linkage, in what role, with what disposition. Present only when the case carries a decision. if case_joins: source_entry["linkage_decisions"] = case_joins join_sources.append(source_entry) rows.append(row) provenance.extend(_provenance(case.case_barcode, case.checklist)) sources.append( { "case_barcode": case.case_barcode, "source_document": patient_filename, "evidence_checksum": _evidence_checksum(case.checklist), } ) per_case_editions.append(case.staging_edition.value) from endopath import provenance as provenance_mod bundle = _bundle( concepts, sources, staging_edition=staging_edition, per_case_editions=per_case_editions, ) # The per-call LLM ledger (#116): how many model calls the cohort cost, in tokens and mean latency, # so a second centre sees what produced the values, not only the values. bundle["llm_calls"] = provenance_mod.summary(conn) # The join provenance for the outcome columns (invariant 2 for a join-sourced value): the source of the # join, the outcome columns it fills, the named derivation rule, the survival gap this pull carries, and # the GDC record ids that fed each case's outcomes. bundle["join_sources"] = { "source": longitudinal.JOIN_SOURCE, "outcome_columns": longitudinal.OUTCOME_COLUMN_KEYS, "derivation_rules": [longitudinal.RECURRENCE_RULE, longitudinal.OVERALL_SURVIVAL_RULE], "data_quality": ( "therapeutic_agents reads Tamoxifen for nearly every case. This is faithful to the GDC/TCGA " "harmonized clinical record, a known source artifact, not an error in this pipeline; it is " "surfaced with its provenance rather than silently dropped. The received_chemotherapy, " "received_radiation, and received_hormone_therapy flags read administration, not record " 'presence: GDC emits a treatment_ template record marked treatment_or_therapy = "no" on ' "cases where the therapy was considered but not given (most often Radiation Therapy, NOS), so a " 'flag is True only where a matching record is marked "yes", False on an explicit "no", and ' 'not_stated where the matching records carry only "unknown". Reading presence instead reports ' "radiation on 543/546 cases (99%); administration reports 264 (48%)." ), "per_case": join_sources, } return { "columns": _columns(concepts, labels), "rows": rows, "provenance": provenance, "bundle": bundle, } # --- rendering the three files from one projection ------------------------------------------------ def _cell(value: Any) -> Any: """A projected value rendered for a spreadsheet cell: None to an empty string, a bool kept as a bool, everything else as-is. An absent value's cell is empty; its reason rides the parallel column.""" if value is None: return "" return value def to_csv(export: dict) -> str: """The projected table as CSV: the column keys as the header, one row per case, an empty cell for an absent value with its reason in the parallel column.""" keys = [column["key"] for column in export["columns"]] buffer = io.StringIO() writer = csv.DictWriter(buffer, fieldnames=keys, extrasaction="ignore") writer.writeheader() for row in export["rows"]: writer.writerow({key: _cell(row.get(key)) for key in keys}) return buffer.getvalue() def to_json(export: dict) -> str: """The whole export as self-describing JSON: the bundle, the column spec, the rows, and the field-level provenance, so the file alone re-derives the table.""" return json.dumps( { "bundle": export["bundle"], "columns": export["columns"], "rows": export["rows"], "provenance": export["provenance"], }, indent=2, default=str, ) def to_xlsx_bytes(export: dict) -> bytes: """The projected table as an XLSX workbook: a `cases` sheet of the projected rows, a `provenance` sheet of the field-level evidence, and a `bundle` sheet of the reproduction metadata as key/value rows.""" from openpyxl import Workbook workbook = Workbook() cases = workbook.active cases.title = "cases" keys = [column["key"] for column in export["columns"]] cases.append([column["label"] for column in export["columns"]]) for row in export["rows"]: cases.append([_cell(row.get(key)) for key in keys]) provenance_sheet = workbook.create_sheet("provenance") provenance_headers = [ "case_barcode", "field_name", "value", "confidence", "status", "applicable", "not_applicable_reason", "evidence_quote", "evidence_page", "evidence_source", ] provenance_sheet.append(provenance_headers) for record in export["provenance"]: provenance_sheet.append([_cell(record.get(header)) for header in provenance_headers]) bundle_sheet = workbook.create_sheet("bundle") bundle_sheet.append(["key", "value"]) for key, value in _flatten(export["bundle"]): bundle_sheet.append([key, value]) buffer = io.BytesIO() workbook.save(buffer) return buffer.getvalue() def _flatten(value: Any, prefix: str = "") -> list[tuple[str, str]]: """Flatten the nested bundle into dotted key/value rows for the XLSX bundle sheet.""" rows: list[tuple[str, str]] = [] if isinstance(value, dict): for key, inner in value.items(): rows.extend(_flatten(inner, f"{prefix}.{key}" if prefix else str(key))) elif isinstance(value, list): if all(not isinstance(item, (dict, list)) for item in value): rows.append((prefix, ", ".join(str(item) for item in value))) else: for index, item in enumerate(value): rows.extend(_flatten(item, f"{prefix}[{index}]")) else: rows.append((prefix, "" if value is None else str(value))) return rows def render_files(export: dict) -> dict: """The three download files from one projection: CSV and JSON as text, XLSX as base64, each with its filename and media type. One projection feeds the preview and all three files, so they never diverge.""" return { "csv": { "filename": "endometrial_cohort.csv", "media_type": "text/csv", "content": to_csv(export), }, "json": { "filename": "endometrial_cohort.json", "media_type": "application/json", "content": to_json(export), }, "xlsx": { "filename": "endometrial_cohort.xlsx", "media_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "content_base64": base64.b64encode(to_xlsx_bytes(export)).decode("ascii"), }, }