"""Map a GDC case record to the `longitudinal_records` rows the app stores (#177). The longitudinal source is the GDC clinical record for each TCGA-UCEC participant: the follow-ups (with the disease response and the recurrence flag), the treatments, the observation time, and the demographic (which carries vital status and days to death). This module is the pure mapping from one GDC `/cases` hit to the rows the store holds; `scripts/ingest_longitudinal.py` does the HTTP pull and the write. Two properties this mapping fixes over the earlier CSV load: 1. Every record is keyed on its GDC-native UUID (`follow_up_id`, `treatment_id`, `demographic_id`), not a synthesized `{participant}_{type}_{index}` string. That string collided (a treatment index resets per diagnosis block), so the earlier load, keyed on a PRIMARY KEY, was lossy. A UUID key is unique and makes the load idempotent. 2. The demographic entity is pulled, so survival arrives: `vital_status` and `days_to_death`, which the earlier CSV omitted entirely. That turns the survival columns of the export from a coded-absence placeholder into real values (`docs/prd.md` sections 4.1, 4.3). The `structured_data` column keeps the raw GDC entity as JSON, for provenance and so the export projection (`longitudinal.py`) reads the same field names GDC uses. """ from __future__ import annotations from typing import Any, Optional # The GDC `/cases` fields to request, as explicit dotted paths. `expand` triggers a GDC 500 on this project, # so every leaf is named. Follow-up carries the recurrence signal; the demographic carries survival. GDC_FIELDS: list[str] = [ "submitter_id", "demographic.demographic_id", "demographic.vital_status", "demographic.days_to_death", "diagnoses.diagnosis_id", "diagnoses.days_to_last_follow_up", "diagnoses.treatments.treatment_id", "diagnoses.treatments.submitter_id", "diagnoses.treatments.treatment_type", "diagnoses.treatments.therapeutic_agents", "diagnoses.treatments.days_to_treatment_start", "diagnoses.treatments.days_to_treatment_end", "diagnoses.treatments.treatment_intent_type", "diagnoses.treatments.treatment_or_therapy", "follow_ups.follow_up_id", "follow_ups.days_to_follow_up", "follow_ups.disease_response", "follow_ups.progression_or_recurrence", "follow_ups.progression_or_recurrence_type", "follow_ups.progression_or_recurrence_anatomic_site", "follow_ups.days_to_recurrence", "follow_ups.days_to_progression", ] def normalize_treatment_type(treatment_type: str) -> str: """The `record_type` for a treatment: `treatment_` plus the GDC `treatment_type` lowercased with spaces as underscores, so "Radiation Therapy, NOS" becomes `treatment_radiation_therapy,_nos`, matching the record-type vocabulary the export and the completeness panel already read.""" return "treatment_" + treatment_type.strip().lower().replace(" ", "_") def _num(value: Any) -> Optional[float]: if value is None or value == "": return None try: return float(value) except (TypeError, ValueError): return None def _int(value: Any) -> Optional[int]: n = _num(value) return int(n) if n is not None else None def _row( *, record_id: str, participant_id: str, record_type: str, record_index: int, data: dict, ingested_at: str, days_to_followup: Optional[int] = None, days_to_treatment: Optional[int] = None, days_to_recurrence: Optional[int] = None, recurrence_type: Optional[str] = None, therapy_type: Optional[str] = None, ) -> dict: """One `longitudinal_records` row. The flat day/type columns mirror the earlier shape for the read endpoint; `structured_data` holds the raw GDC entity.""" return { "record_id": record_id, "participant_id": participant_id, "case_barcode": participant_id, "record_type": record_type, "record_index": record_index, "days_to_followup": days_to_followup, "days_to_treatment": days_to_treatment, "days_to_recurrence": days_to_recurrence, "days_to_lost_to_followup": None, "biopsy_result": None, "followup_treatment_success": None, "recurrence_type": recurrence_type, "therapy_type": therapy_type, "lost_to_followup": None, "structured_data": data, "ingested_at": ingested_at, } def records_from_case(case: dict, *, ingested_at: str) -> list[dict]: """Every `longitudinal_records` row for one GDC case hit. Emits one `follow_up` per GDC follow-up (keyed on `follow_up_id`), one `treatment_` per treatment (keyed on `treatment_id`), one derived per-participant `last_followup` (the observation time), and one `demographic` carrying survival. A follow-up or treatment GDC did not return is simply absent; the row is never fabricated. `structured_data` is left as the raw GDC dict for provenance. """ participant_id = case.get("submitter_id") if not participant_id: return [] rows: list[dict] = [] index = 0 follow_ups = case.get("follow_ups") or [] for follow_up in follow_ups: fid = follow_up.get("follow_up_id") if not fid: continue index += 1 rows.append( _row( record_id=fid, participant_id=participant_id, record_type="follow_up", record_index=index, data=follow_up, ingested_at=ingested_at, days_to_followup=_int(follow_up.get("days_to_follow_up")), days_to_recurrence=_int(follow_up.get("days_to_recurrence")), recurrence_type=follow_up.get("progression_or_recurrence_type"), ) ) for diagnosis in case.get("diagnoses") or []: for treatment in diagnosis.get("treatments") or []: tid = treatment.get("treatment_id") treatment_type = treatment.get("treatment_type") if not tid or not treatment_type: continue index += 1 rows.append( _row( record_id=tid, participant_id=participant_id, record_type=normalize_treatment_type(treatment_type), record_index=index, data=treatment, ingested_at=ingested_at, days_to_treatment=_int(treatment.get("days_to_treatment_start")), therapy_type=treatment_type, ) ) # The observation/censoring time: the largest days-to-last-follow-up any diagnosis carries, or the latest # follow-up day. One derived record per participant, keyed deterministically. last_day = _last_follow_up_day(case) if last_day is not None: index += 1 rows.append( _row( record_id=f"{participant_id}_last_followup", participant_id=participant_id, record_type="last_followup", record_index=index, data={"days_to_last_follow_up": last_day}, ingested_at=ingested_at, days_to_followup=last_day, ) ) # Survival: the demographic entity, pulled so the export's survival columns carry real values rather than # a coded-absence placeholder. Keyed on the GDC demographic UUID when present. demographic = case.get("demographic") or {} if demographic.get("vital_status") is not None or demographic.get("days_to_death") is not None: demographic_id = demographic.get("demographic_id") or f"{participant_id}_demographic" index += 1 rows.append( _row( record_id=demographic_id, participant_id=participant_id, record_type="demographic", record_index=index, data=demographic, ingested_at=ingested_at, ) ) return rows def _last_follow_up_day(case: dict) -> Optional[int]: candidates: list[int] = [] for diagnosis in case.get("diagnoses") or []: day = _int(diagnosis.get("days_to_last_follow_up")) if day is not None: candidates.append(day) for follow_up in case.get("follow_ups") or []: day = _int(follow_up.get("days_to_follow_up")) if day is not None: candidates.append(day) return max(candidates) if candidates else None