| """One case put against the whole target dictionary, with the source of every answer. |
| |
| The case-review surface shows the checklist concepts the pathology report answers. The study's dictionary |
| asks more than that: 204 fields over 19 sections, drawn from five record types, of which the report answers |
| one. This module merges the dictionary with the case's answers so a reviewer reads every field against the |
| record that fills it, and sees which fields nothing fills yet. |
| |
| Three sources answer a field, and each names itself on the field it fills: |
| |
| 1. The pathology report. A dictionary field whose `fill_via` is `checklist` maps to a checklist concept by |
| `fill_key`, so it holds the extracted value with the page, quote, and bounding box behind it. |
| 2. The GDC clinical join. `join_projection` answers the 82 fields that name the clinical record as their |
| source. The records a reviewer rejected are dropped first, the same rule `export.py` applies, so a |
| rejected linkage stops filling the dictionary. |
| 3. A sampled record type. `record_type_extractions` holds the registry and local-research samples, which |
| the open TCGA corpus has no data for. |
| |
| A field no rule maps has `mapped: false`. That states something different from a coded absence: a |
| coded absence means a source was read and gave no value, while an unmapped field means no rule connects |
| this dictionary question to a source yet. Rendering them the same way would claim a search that never ran. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| from typing import Any, Optional |
|
|
| from sqlalchemy import Connection, text |
|
|
| from endopath import coverage_data, join_projection, longitudinal, mapping, storage, taxonomy |
| from endopath.schema import Case |
| from endopath.validation import FieldState |
|
|
|
|
| def edition_dependent_dimensions() -> frozenset[str]: |
| """The dimensions whose stored value means nothing until a coding system is named. |
| |
| Read from the registry rather than listed here: a dimension decided by `algorithm` rows under more |
| than one system means a different thing per system. `figo_stage` qualifies because FIGO 1988, 2009, |
| and 2023 each decide it, and `IB` denotes invasion of less than half the myometrium under 1988 and |
| half or more under 2009. A registry that gains a second edition for another dimension gains this |
| treatment with no edit here.""" |
| systems_by_dimension: dict[str, set[str]] = {} |
| for rule in taxonomy.load_registry().derivations: |
| if rule.rule_kind is taxonomy.RuleKind.ALGORITHM and rule.under_system: |
| systems_by_dimension.setdefault(rule.target_dimension, set()).add(rule.under_system) |
| return frozenset(dim for dim, systems in systems_by_dimension.items() if len(systems) > 1) |
|
|
|
|
|
|
| |
| _ANSWERED_ABSENCES = frozenset({FieldState.CONSTRAINED.value}) |
|
|
|
|
| def _primitive(value: Any) -> Any: |
| """A checklist value as JSON, matching what the review surface serializes.""" |
| if value is None or isinstance(value, (str, int, float, bool)): |
| return value |
| return str(getattr(value, "value", value)) |
|
|
|
|
| def load_case_records(conn: Connection, case_barcode: str) -> list[longitudinal.Record]: |
| """The participant's GDC records, minus the ones a reviewer rejected for this case. |
| |
| The join matches on the participant barcode, the first 12 characters of the case barcode, which is what |
| `longitudinal_records.participant_id` holds. A record with no `longitudinal_joins` row is unreviewed |
| rather than rejected, so only an explicit `rejected` disposition removes it. |
| """ |
| participant_id = case_barcode[:12] if len(case_barcode) >= 12 else case_barcode |
| rejected = { |
| row[0] |
| for row in conn.execute( |
| text( |
| "SELECT record_id FROM longitudinal_joins " |
| "WHERE case_barcode = :barcode AND disposition = 'rejected'" |
| ), |
| {"barcode": case_barcode}, |
| ).fetchall() |
| } |
| rows = conn.execute( |
| text( |
| "SELECT record_id, record_type, structured_data FROM longitudinal_records " |
| "WHERE participant_id = :pid ORDER BY record_type, record_index" |
| ), |
| {"pid": participant_id}, |
| ).mappings().fetchall() |
|
|
| records: list[longitudinal.Record] = [] |
| for row in rows: |
| if row["record_id"] in rejected: |
| continue |
| data: dict = {} |
| if row["structured_data"]: |
| try: |
| data = json.loads(row["structured_data"]) |
| except (json.JSONDecodeError, TypeError): |
| data = {} |
| records.append( |
| longitudinal.Record( |
| record_type=row["record_type"], record_id=row["record_id"], data=data |
| ) |
| ) |
| return records |
|
|
|
|
| def _sampled_extractions(conn: Connection, project: str, case_barcode: str) -> dict[str, dict]: |
| """This case's rows from the sampled record-type extractions, keyed on dictionary field name.""" |
| sampled: dict[str, dict] = {} |
| for record_type in coverage_data.RECORD_TYPES: |
| record_type_id = record_type[0] |
| for row in storage.list_record_type_extractions(conn, project, record_type_id): |
| if row["case_barcode"] == case_barcode: |
| sampled[row["field"]] = { |
| "value": row["value"], |
| "evidence": row["evidence"], |
| "record_type_id": record_type_id, |
| } |
| return sampled |
|
|
|
|
| def _unmapped(name: str, label: str, section: str, source: str) -> dict: |
| return { |
| "name": name, |
| "label": label, |
| "section": section, |
| "source": source, |
| "mapped": False, |
| "value": None, |
| "absence_code": None, |
| "constrained_to": None, |
| "evidence": None, |
| "join": None, |
| "checklist_field": None, |
| } |
|
|
|
|
| def _from_checklist( |
| base: dict, checklist_name: str, field, *, edition: str = "", edition_dependent: bool = False |
| ) -> dict: |
| """A dictionary field answered by the report, with the evidence span the extraction recorded, so the |
| page and its bounding box reach the screen with the value. |
| |
| A value on an edition-dependent dimension carries the coding system it speaks, because the same code |
| denotes different findings across editions. Where no source stated the edition, the entry says so |
| rather than presenting the bare code as though it read the same everywhere.""" |
| entry = dict(base) |
| entry.update( |
| { |
| "mapped": True, |
| "value": _primitive(field.value), |
| "absence_code": field.absence_code.value if field.absence_code else None, |
| "evidence": field.evidence.model_dump() if field.evidence else None, |
| "checklist_field": checklist_name, |
| } |
| ) |
| if edition_dependent: |
| entry["edition_dependent"] = True |
| entry["edition"] = edition or None |
| |
| entry["edition_label"] = taxonomy.coding_system_labels().get(edition) if edition else None |
| if entry["value"] is not None and not edition: |
| entry["edition_unstated"] = True |
| if entry["value"] is None and entry["absence_code"] is None: |
| |
| entry["absence_code"] = FieldState.NOT_STATED.value |
| return entry |
|
|
|
|
| def _from_join(base: dict, cell: join_projection.JoinCell) -> dict: |
| entry = dict(base) |
| entry.update( |
| { |
| "mapped": True, |
| "value": cell.value, |
| "absence_code": cell.absent_reason, |
| "constrained_to": cell.constrained_to, |
| "join": {"evidence": cell.evidence, "record_ids": list(cell.record_ids)} |
| if cell.evidence or cell.record_ids |
| else None, |
| } |
| ) |
| return entry |
|
|
|
|
| def _from_sample(base: dict, sample: dict) -> dict: |
| """A field answered by a sampled record type. The seed stores a coded absence in the value column when |
| the sample recorded one, so a value matching an absence code is read back as that code.""" |
| entry = dict(base) |
| value = sample["value"] |
| codes = {state.value for state in FieldState} |
| entry.update( |
| { |
| "mapped": True, |
| "value": None if value in codes else value, |
| "absence_code": value if value in codes else None, |
| "join": {"evidence": sample["evidence"], "record_ids": []} |
| if sample["evidence"] |
| else None, |
| } |
| ) |
| return entry |
|
|
|
|
| def build_case_dictionary( |
| conn: Connection, |
| case: Case, |
| *, |
| project: str = coverage_data.PROJECT_ID, |
| page_count: int = 0, |
| patient_filename: Optional[str] = None, |
| ) -> dict: |
| """Merge the project's target dictionary with one case's answers and their sources. |
| |
| Args: |
| conn: an open store connection. |
| case: the case whose checklist supplies the report-sourced answers. |
| project: the project owning the target dictionary. |
| page_count: the case's scanned page count, so the caller can open an evidence page. |
| patient_filename: the report file behind the case, for display. |
| |
| Returns: |
| The dictionary payload: the record types, the fields grouped into their dictionary sections, and a |
| per-source summary of how many fields are mapped and how many this case answers. |
| """ |
| dictionary_fields = storage.list_coverage_fields(conn, project) |
| record_types = storage.list_record_types(conn, project) |
| checklist = case.checklist.all_fields() |
| join_cells = join_projection.project_dictionary( |
| case.case_barcode[:12], load_case_records(conn, case.case_barcode) |
| ) |
| sampled = _sampled_extractions(conn, project, case.case_barcode) |
|
|
| |
| |
| |
| |
| edition_dimensions = edition_dependent_dimensions() |
| edition_dependent = { |
| checklist_field |
| for checklist_field, concept in mapping.cap_checklist_coverage().items() |
| if concept in edition_dimensions |
| } |
| case_edition = case.staging_edition or "" |
|
|
| sections: dict[str, list[dict]] = {} |
| order: list[str] = [] |
| summary_by_source: dict[str, dict[str, int]] = {} |
|
|
| for row in dictionary_fields: |
| name, section, source = row["name"], row["section"], row["record_type_id"] |
| base = _unmapped(name, row["label"], section, source) |
|
|
| checklist_name = row["fill_key"] if row["fill_via"] == "checklist" else None |
| if checklist_name and checklist_name in checklist: |
| entry = _from_checklist( |
| base, |
| checklist_name, |
| checklist[checklist_name], |
| edition=case_edition, |
| edition_dependent=checklist_name in edition_dependent, |
| ) |
| elif name in join_cells: |
| entry = _from_join(base, join_cells[name]) |
| elif name in sampled: |
| entry = _from_sample(base, sampled[name]) |
| else: |
| entry = base |
|
|
| if section not in sections: |
| sections[section] = [] |
| order.append(section) |
| sections[section].append(entry) |
|
|
| counts = summary_by_source.setdefault(source, {"fields": 0, "mapped": 0, "answered": 0}) |
| counts["fields"] += 1 |
| counts["mapped"] += int(entry["mapped"]) |
| counts["answered"] += int(_is_answered(entry)) |
|
|
| fields_out = [{"title": title, "fields": sections[title]} for title in order] |
| total = sum(c["fields"] for c in summary_by_source.values()) |
| return { |
| "case_barcode": case.case_barcode, |
| "patient_filename": patient_filename, |
| "dictionary_id": coverage_data.DICTIONARY_ID, |
| "page_count": page_count, |
| "record_types": record_types, |
| "sections": fields_out, |
| "summary": { |
| "total": total, |
| "mapped": sum(c["mapped"] for c in summary_by_source.values()), |
| "answered": sum(c["answered"] for c in summary_by_source.values()), |
| "by_source": summary_by_source, |
| }, |
| } |
|
|
|
|
| def _is_answered(entry: dict) -> bool: |
| """A field this case answers: a stated value, or a constrained field whose primitive the join knows.""" |
| return entry["value"] is not None or entry["absence_code"] in _ANSWERED_ABSENCES |
|
|
