diff --git a/flows/ETL_US_BirthData_to_OMOP.json b/flows/ETL_US_BirthData_to_OMOP.json new file mode 100644 index 0000000..b76bb1c --- /dev/null +++ b/flows/ETL_US_BirthData_to_OMOP.json @@ -0,0 +1 @@ +{"id":"59745eff-9bb9-486a-ab64-04433cb586d5","name":"ETL US BirthData to OMOP","nodes":[{"id":"node-load-bronze","data":{"name":"parse_nat2022_to_staging_csv","description":"Bronze: parse the fixed-width Nat2022 file (pruned columns) into a staging CSV on the shared trex volume -- no database access; node-facts consumes the CSV directly","python_code":"# Deployment-topology constants (deliberately NOT pipeline variables): these\n# paths are fixed by the platform's container mounts and images, not user-\n# tunable -- the flow container mounts the trex data volume at /app/duckdb_data\n# (trex sees that same directory as /usr/src/data), and the source-data folder\n# is part of the flow image at /app/data_load. Changing them in a pipeline\n# variable could never make a flow work; if the platform's mounts ever change,\n# update them here.\nNAT2022_DATA_DIR = \"/app/data_load\"\nSTAGING_DIR_FLOW = \"/app/duckdb_data/flow_staging\" # this container's view of the shared staging dir\nSTAGING_DIR_TREX = \"/usr/src/data/flow_staging\" # trex's view of the same directory\n\nFIELDS = [\n (\"DOB_YY\", 9, 12),\n (\"DOB_MM\", 13, 14),\n (\"DOB_TT\", 19, 22),\n (\"DOB_WK\", 23, 23),\n (\"OCTERR\", 24, 25),\n (\"OCNTYFIPS\", 28, 30),\n (\"OCNTYPOP\", 31, 31),\n (\"BFACIL\", 32, 32),\n (\"F_BFACIL\", 33, 33),\n (\"BFACIL3\", 50, 50),\n (\"MAGE_IMPFLG\", 73, 73),\n (\"MAGE_REPFLG\", 74, 74),\n (\"MAGER\", 75, 76),\n (\"MAGER14\", 77, 78),\n (\"MAGER9\", 79, 79),\n (\"MBCNTRY\", 80, 81),\n (\"MBSTATE_REC\", 84, 84),\n (\"MRCNTRY\", 85, 86),\n (\"MRTERR\", 89, 90),\n (\"RCNTY\", 91, 93),\n (\"RCNTY_POP\", 99, 99),\n (\"RCITY_POP\", 100, 100),\n (\"RECTYPE\", 103, 103),\n (\"RESTATUS\", 104, 104),\n (\"MRACE31\", 105, 106),\n (\"MRACE6\", 107, 107),\n (\"MRACE15\", 108, 109),\n (\"MRACEIMP\", 111, 111),\n (\"MHISPX\", 112, 112),\n (\"MHISP_R\", 115, 115),\n (\"F_MHISP\", 116, 116),\n (\"MRACEHISP\", 117, 117),\n (\"MAR_P\", 119, 119),\n (\"DMAR\", 120, 120),\n (\"MAR_IMP\", 121, 121),\n (\"F_MAR_P\", 123, 123),\n (\"MEDUC\", 124, 124),\n (\"F_MEDUC\", 126, 126),\n (\"FAGERPT_FLG\", 142, 142),\n (\"FAGECOMB\", 147, 148),\n (\"FAGEREC11\", 149, 150),\n (\"FRACE31\", 151, 152),\n (\"FRACE6\", 153, 153),\n (\"FRACE15\", 154, 155),\n (\"FHISPX\", 159, 159),\n (\"FHISP_R\", 160, 160),\n (\"F_FHISP\", 161, 161),\n (\"FRACEHISP\", 162, 162),\n (\"FEDUC\", 163, 163),\n (\"F_FEDUC\", 165, 165),\n (\"PRIORLIVE\", 171, 172),\n (\"PRIORDEAD\", 173, 174),\n (\"PRIORTERM\", 175, 176),\n (\"LBO_REC\", 179, 179),\n (\"TBO_REC\", 182, 182),\n (\"ILLB_R\", 198, 200),\n (\"ILLB_R11\", 201, 202),\n (\"ILOP_R\", 206, 208),\n (\"ILOP_R11\", 209, 210),\n (\"ILP_R\", 214, 216),\n (\"ILP_R11\", 217, 218),\n (\"PRECARE\", 224, 225),\n (\"F_MPCB\", 226, 226),\n (\"PRECARE5\", 227, 227),\n (\"PREVIS\", 238, 239),\n (\"PREVIS_REC\", 242, 243),\n (\"F_TPCV\", 244, 244),\n (\"WIC\", 251, 251),\n (\"F_WIC\", 252, 252),\n (\"CIG_0\", 253, 254),\n (\"CIG_1\", 255, 256),\n (\"CIG_2\", 257, 258),\n (\"CIG_3\", 259, 260),\n (\"CIG0_R\", 261, 261),\n (\"CIG1_R\", 262, 262),\n (\"CIG2_R\", 263, 263),\n (\"CIG3_R\", 264, 264),\n (\"F_CIGS_0\", 265, 265),\n (\"F_CIGS_1\", 266, 266),\n (\"F_CIGS_2\", 267, 267),\n (\"F_CIGS_3\", 268, 268),\n (\"CIG_REC\", 269, 269),\n (\"F_TOBACO\", 270, 270),\n (\"M_HT_IN\", 280, 281),\n (\"F_M_HT\", 282, 282),\n (\"BMI\", 283, 286),\n (\"BMI_R\", 287, 287),\n (\"PWGT_R\", 292, 294),\n (\"F_PWGT\", 295, 295),\n (\"DWGT_R\", 299, 301),\n (\"F_DWGT\", 303, 303),\n (\"WTGAIN\", 304, 305),\n (\"WTGAIN_REC\", 306, 306),\n (\"F_WTGAIN\", 307, 307),\n (\"RF_PDIAB\", 313, 313),\n (\"RF_GDIAB\", 314, 314),\n (\"RF_PHYPE\", 315, 315),\n (\"RF_GHYPE\", 316, 316),\n (\"RF_EHYPE\", 317, 317),\n (\"RF_PPTERM\", 318, 318),\n (\"F_RF_PDIAB\", 319, 319),\n (\"F_RF_GDIAB\", 320, 320),\n (\"F_RF_PHYPER\", 321, 321),\n (\"F_RF_GHYPER\", 322, 322),\n (\"F_RF_ECLAMP\", 323, 323),\n (\"F_RF_PPB\", 324, 324),\n (\"RF_INFTR\", 325, 325),\n (\"RF_FEDRG\", 326, 326),\n (\"RF_ARTEC\", 327, 327),\n (\"F_RF_INFT\", 328, 328),\n (\"F_RF_INF_DRG\", 329, 329),\n (\"F_RF_INF_ART\", 330, 330),\n (\"RF_CESAR\", 331, 331),\n (\"RF_CESARN\", 332, 333),\n (\"F_RF_CESAR\", 335, 335),\n (\"F_RF_NCESAR\", 336, 336),\n (\"NO_RISKS\", 337, 337),\n (\"IP_GON\", 343, 343),\n (\"IP_SYPH\", 344, 344),\n (\"IP_CHLAM\", 345, 345),\n (\"IP_HEPB\", 346, 346),\n (\"IP_HEPC\", 347, 347),\n (\"F_IP_GONOR\", 348, 348),\n (\"F_IP_SYPH\", 349, 349),\n (\"F_IP_CHLAM\", 350, 350),\n (\"F_IP_HEPATB\", 351, 351),\n (\"F_IP_HEPATC\", 352, 352),\n (\"NO_INFEC\", 353, 353),\n (\"OB_ECVS\", 360, 360),\n (\"OB_ECVF\", 361, 361),\n (\"F_OB_SUCC\", 363, 363),\n (\"F_OB_FAIL\", 364, 364),\n (\"LD_INDL\", 383, 383),\n (\"LD_AUGM\", 384, 384),\n (\"LD_STER\", 385, 385),\n (\"LD_ANTB\", 386, 386),\n (\"LD_CHOR\", 387, 387),\n (\"LD_ANES\", 388, 388),\n (\"F_LD_INDL\", 389, 389),\n (\"F_LD_AUGM\", 390, 390),\n (\"F_LD_STER\", 391, 391),\n (\"F_LD_ANTB\", 392, 392),\n (\"F_LD_CHOR\", 393, 393),\n (\"F_LD_ANES\", 394, 394),\n (\"NO_LBRDLV\", 395, 395),\n (\"ME_PRES\", 401, 401),\n (\"ME_ROUT\", 402, 402),\n (\"ME_TRIAL\", 403, 403),\n (\"F_ME_PRES\", 404, 404),\n (\"F_ME_ROUT\", 405, 405),\n (\"F_ME_TRIAL\", 406, 406),\n (\"RDMETH_REC\", 407, 407),\n (\"DMETH_REC\", 408, 408),\n (\"F_DMETH_REC\", 409, 409),\n (\"MM_MTR\", 415, 415),\n (\"MM_PLAC\", 416, 416),\n (\"MM_RUPT\", 417, 417),\n (\"MM_UHYST\", 418, 418),\n (\"MM_AICU\", 419, 419),\n (\"F_MM_MTR\", 421, 421),\n (\"F_MM_PLAC\", 422, 422),\n (\"F_MM_RUPT\", 423, 423),\n (\"F_MM_UHYST\", 424, 424),\n (\"F_MM_AICU\", 425, 425),\n (\"NO_MMORB\", 427, 427),\n (\"ATTEND\", 433, 433),\n (\"MTRAN\", 434, 434),\n (\"PAY\", 435, 435),\n (\"PAY_REC\", 436, 436),\n (\"F_PAY\", 437, 437),\n (\"F_PAY_REC\", 438, 438),\n (\"APGAR5\", 444, 445),\n (\"APGAR5R\", 446, 446),\n (\"F_APGAR5\", 447, 447),\n (\"APGAR10\", 448, 449),\n (\"APGAR10R\", 450, 450),\n (\"DPLURAL\", 454, 454),\n (\"IMP_PLUR\", 456, 456),\n (\"SETORDER_R\", 459, 459),\n (\"SEX\", 475, 475),\n (\"IMP_SEX\", 476, 476),\n (\"DLMP_MM\", 477, 478),\n (\"DLMP_YY\", 481, 484),\n (\"COMPGST_IMP\", 488, 488),\n (\"OBGEST_FLG\", 489, 489),\n (\"COMBGEST\", 490, 491),\n (\"GESTREC10\", 492, 493),\n (\"GESTREC3\", 494, 494),\n (\"LMPUSED\", 498, 498),\n (\"OEGEST_COMB\", 499, 500),\n (\"OEGEST_R10\", 501, 502),\n (\"OEGEST_R3\", 503, 503),\n (\"DBWT\", 504, 507),\n (\"BWTR12\", 509, 510),\n (\"BWTR4\", 511, 511),\n (\"AB_AVEN1\", 517, 517),\n (\"AB_AVEN6\", 518, 518),\n (\"AB_NICU\", 519, 519),\n (\"AB_SURF\", 520, 520),\n (\"AB_ANTI\", 521, 521),\n (\"AB_SEIZ\", 522, 522),\n (\"F_AB_VENT\", 524, 524),\n (\"F_AB_VENT6\", 525, 525),\n (\"F_AB_NIUC\", 526, 526),\n (\"F_AB_SURFAC\", 527, 527),\n (\"F_AB_ANTIBIO\", 528, 528),\n (\"F_AB_SEIZ\", 529, 529),\n (\"NO_ABNORM\", 531, 531),\n (\"CA_ANEN\", 537, 537),\n (\"CA_MNSB\", 538, 538),\n (\"CA_CCHD\", 539, 539),\n (\"CA_CDH\", 540, 540),\n (\"CA_OMPH\", 541, 541),\n (\"CA_GAST\", 542, 542),\n (\"F_CA_ANEN\", 543, 543),\n (\"F_CA_MENIN\", 544, 544),\n (\"F_CA_HEART\", 545, 545),\n (\"F_CA_HERNIA\", 546, 546),\n (\"F_CA_OMPHA\", 547, 547),\n (\"F_CA_GASTRO\", 548, 548),\n (\"CA_LIMB\", 549, 549),\n (\"CA_CLEFT\", 550, 550),\n (\"CA_CLPAL\", 551, 551),\n (\"CA_DOWN\", 552, 552),\n (\"CA_DISOR\", 553, 553),\n (\"CA_HYPO\", 554, 554),\n (\"F_CA_LIMB\", 555, 555),\n (\"F_CA_CLEFTLP\", 556, 556),\n (\"F_CA_CLEFT\", 557, 557),\n (\"F_CA_DOWNS\", 558, 558),\n (\"F_CA_CHROM\", 559, 559),\n (\"F_CA_HYPOS\", 560, 560),\n (\"NO_CONGEN\", 561, 561),\n (\"ITRAN\", 567, 567),\n (\"ILIVE\", 568, 568),\n (\"BFED\", 569, 569),\n (\"F_BFED\", 570, 570),\n]\n\n# Column pruning: the raw file has 237 fields, but node-facts only reads the\n# ones below (derived from its field lists, alias table, and every direct\n# rec[...] access). Parsing and staging only these keeps the CSV and the\n# transform's input small. If node-facts starts reading a new raw column, it\n# must be added here.\nRAW_KEEP_COLUMNS = {\n \"AB_ANTI\", \"AB_AVEN1\", \"AB_AVEN6\", \"AB_NICU\", \"AB_SEIZ\", \"AB_SURF\",\n \"APGAR10\", \"APGAR5\", \"ATTEND\", \"BFACIL\", \"BFED\", \"BMI\",\n \"CA_ANEN\", \"CA_CCHD\", \"CA_CDH\", \"CA_CLEFT\", \"CA_CLPAL\", \"CA_DISOR\",\n \"CA_DOWN\", \"CA_GAST\", \"CA_HYPO\", \"CA_LIMB\", \"CA_MNSB\", \"CA_OMPH\",\n \"CIG0_R\", \"CIG1_R\", \"CIG2_R\", \"CIG3_R\", \"COMBGEST\", \"DBWT\",\n \"DLMP_MM\", \"DLMP_YY\", \"DMAR\", \"DMETH_REC\", \"DOB_MM\", \"DOB_TT\",\n \"DOB_WK\", \"DOB_YY\", \"DPLURAL\", \"FAGECOMB\", \"FEDUC\", \"FHISPX\",\n \"FRACE15\", \"FRACE6\", \"FRACEHISP\", \"ILIVE\", \"ILLB_R\", \"ILOP_R\",\n \"IP_CHLAM\", \"IP_GON\", \"IP_HEPB\", \"IP_HEPC\", \"IP_SYPH\", \"ITRAN\",\n \"LD_ANES\", \"LD_ANTB\", \"LD_AUGM\", \"LD_CHOR\", \"LD_INDL\", \"LD_STER\",\n \"MAGER\", \"MAR_P\", \"MEDUC\", \"ME_PRES\", \"ME_ROUT\", \"ME_TRIAL\",\n \"MHISPX\", \"MM_AICU\", \"MM_MTR\", \"MM_PLAC\", \"MM_RUPT\", \"MM_UHYST\",\n \"MRACE15\", \"MRACE6\", \"MRACEHISP\", \"MTRAN\", \"M_HT_IN\", \"NO_ABNORM\",\n \"NO_CONGEN\", \"NO_LBRDLV\", \"NO_MMORB\", \"NO_RISKS\", \"OB_ECVF\", \"OB_ECVS\",\n \"OEGEST_COMB\", \"PAY\", \"PRECARE\", \"PREVIS\", \"PRIORDEAD\", \"PRIORLIVE\",\n \"PRIORTERM\", \"PWGT_R\", \"RDMETH_REC\", \"RF_ARTEC\", \"RF_CESAR\", \"RF_CESARN\",\n \"RF_EHYPE\", \"RF_FEDRG\", \"RF_GDIAB\", \"RF_GHYPE\", \"RF_INFTR\", \"RF_PDIAB\",\n \"RF_PHYPE\", \"RF_PPTERM\", \"SETORDER_R\", \"SEX\", \"WIC\", \"WTGAIN\",\n}\n\nFIELDS = [f for f in FIELDS if f[0] in RAW_KEEP_COLUMNS]\n\nCOLSPECS = [(start - 1, end) for (_, start, end) in FIELDS]\nNAMES = [name for (name, _, _) in FIELDS]\n\n\ndef load_nat2022(filepath, chunksize=None):\n read_kwargs = dict(colspecs=COLSPECS, names=NAMES, dtype=str, encoding=\"utf-8\")\n if chunksize:\n return pd.read_fwf(filepath, chunksize=chunksize, **read_kwargs)\n return pd.read_fwf(filepath, **read_kwargs)\n\n\ndef _sanitize_cache_id(raw_id: str) -> str:\n \"\"\"\n Mirrors the platform's own Dataset.cacheId derivation\n (portal/src/dataset/entity/dataset.entity.ts: sanitizeIdForCacheId) --\n hyphens aren't valid in a bare SQL/DuckDB identifier, so the Dataset's\n UUID gets converted the same way here before use as a trex cache_id.\n \"\"\"\n cleaned = raw_id.replace(\"-\", \"_\")\n return f\"_{cleaned}\" if cleaned[:1].isdigit() else cleaned\n\n\ndef exec(myinput):\n \"\"\"\n Bronze: parse the fixed-width source file (columns pruned to\n RAW_KEEP_COLUMNS) into one staging CSV on the shared trex-volume staging\n dir, plus a `.complete` marker holding the row count. This node needs no\n database access -- node-facts reads the CSV directly, and node-cleanup\n deletes it after a successful transform.\n\n Crash-safety: the marker is deleted BEFORE writing and recreated only\n after the last row -- an interrupted run leaves csv-without-marker, which\n node-facts refuses to consume.\n\n ROW_IDX is assigned here, once, directly from the file's read order (which\n pandas read_fwf guarantees is sequential) -- downstream uses this stored\n value as record_idx rather than re-deriving positions.\n \"\"\"\n import time\n import uuid\n\n logger = get_run_logger()\n nat2022_file_path = os.path.join(NAT2022_DATA_DIR, nat2022_filename)\n cache_catalog = _sanitize_cache_id(dataset_id)\n fname = f\"{cache_catalog}_nat2022_raw.csv\"\n csv_path = os.path.join(STAGING_DIR_FLOW, fname)\n complete_path = csv_path + \".complete\"\n\n try:\n probe = os.path.join(STAGING_DIR_FLOW, f\".probe_{uuid.uuid4().hex}\")\n with open(probe, \"w\") as f:\n f.write(\"x\")\n os.remove(probe)\n except Exception as e:\n raise RuntimeError(\n f\"Shared staging dir {STAGING_DIR_FLOW} is not writable ({e}). It is required \"\n f\"infrastructure for this pipeline: on the trex side run once \"\n f\"`mkdir -p /usr/src/data/flow_staging && chmod 1777 /usr/src/data/flow_staging` \"\n f\"(new trex images and the helm chart provision it automatically).\"\n ) from e\n\n # The marker must be removed before writing begins: a crash mid-write can\n # then never leave a fresh marker next to a stale or partial CSV.\n try:\n os.remove(complete_path)\n except OSError:\n pass\n\n total = 0\n t_parse = time.time()\n for chunk_idx, chunk in enumerate(load_nat2022(nat2022_file_path, chunksize=int(chunk_size))):\n chunk = chunk.copy()\n chunk.insert(0, \"ROW_IDX\", range(total, total + len(chunk)))\n chunk.to_csv(csv_path, mode=\"w\" if chunk_idx == 0 else \"a\", header=(chunk_idx == 0), index=False, na_rep=\"\")\n total += len(chunk)\n logger.info(f\"Parsed {total:,} records -> staging CSV ({time.time() - t_parse:.0f}s elapsed)\")\n\n with open(complete_path, \"w\") as f:\n f.write(str(total))\n logger.info(f\"Staged {total:,} records at {csv_path} (+ .complete marker)\")\n return f\"Parsed {total} records into {csv_path}\""},"type":"python_node","width":350,"height":210,"position":{"x":130,"y":110},"selected":false,"dragHandle":"","sourcePosition":"right","targetPosition":"left"},{"id":"node-facts","data":{"name":"transform_facts","description":"Gold: build PERSON, VISIT_OCCURRENCE, OBSERVATION, MEASUREMENT, CONDITION_OCCURRENCE, PROCEDURE_OCCURRENCE, and PAYER_PLAN_PERIOD in a single per-record pass (merged from 7 originally separate nodes; see node_backups/ for the originals).","python_code":"# Deployment-topology constants (deliberately NOT pipeline variables) -- see\n# node-load-bronze for the full rationale: fixed by the platform's container\n# mounts; the two paths are the flow container's and trex's views of the same\n# shared staging directory on the trex data volume.\nNAT2022_DATA_DIR = \"/app/data_load\"\nSTAGING_DIR_FLOW = \"/app/duckdb_data/flow_staging\"\nSTAGING_DIR_TREX = \"/usr/src/data/flow_staging\"\n\nMAPPING_CSV_NODE = \"csv_node_0\"\n\n_PREFIX_GROUP_RE = re.compile(r'^(all )?variables beginning with\\s+(\\w+?)_?$', re.I)\n\n_REAL_CDM_COLUMNS = {\n \"PERSON\": {\"person_id\", \"gender_concept_id\", \"year_of_birth\", \"month_of_birth\", \"day_of_birth\",\n \"birth_datetime\", \"race_concept_id\", \"ethnicity_concept_id\", \"location_id\", \"provider_id\",\n \"care_site_id\", \"person_source_value\", \"gender_source_value\", \"gender_source_concept_id\",\n \"race_source_value\", \"race_source_concept_id\", \"ethnicity_source_value\",\n \"ethnicity_source_concept_id\"},\n \"VISIT_OCCURRENCE\": {\"visit_occurrence_id\", \"person_id\", \"visit_concept_id\", \"visit_start_date\",\n \"visit_start_datetime\", \"visit_end_date\", \"visit_end_datetime\",\n \"visit_type_concept_id\", \"provider_id\", \"care_site_id\", \"visit_source_value\",\n \"visit_source_concept_id\", \"admitted_from_concept_id\", \"admitted_from_source_value\",\n \"discharged_to_concept_id\", \"discharged_to_source_value\",\n \"preceding_visit_occurrence_id\"},\n \"OBSERVATION\": {\"observation_id\", \"person_id\", \"observation_concept_id\", \"observation_date\",\n \"observation_datetime\", \"observation_type_concept_id\", \"value_as_number\",\n \"value_as_string\", \"value_as_concept_id\", \"qualifier_concept_id\", \"unit_concept_id\",\n \"provider_id\", \"visit_occurrence_id\", \"visit_detail_id\", \"observation_source_value\",\n \"observation_source_concept_id\", \"unit_source_value\", \"qualifier_source_value\",\n \"value_source_value\", \"observation_event_id\", \"obs_event_field_concept_id\"},\n \"MEASUREMENT\": {\"measurement_id\", \"person_id\", \"measurement_concept_id\", \"measurement_date\",\n \"measurement_datetime\", \"measurement_time\", \"measurement_type_concept_id\",\n \"operator_concept_id\", \"value_as_number\", \"value_as_concept_id\", \"unit_concept_id\",\n \"range_low\", \"range_high\", \"provider_id\", \"visit_occurrence_id\", \"visit_detail_id\",\n \"measurement_source_value\", \"measurement_source_concept_id\", \"unit_source_value\",\n \"unit_source_concept_id\", \"value_source_value\", \"measurement_event_id\",\n \"meas_event_field_concept_id\"},\n \"CONDITION\": {\"condition_occurrence_id\", \"person_id\", \"condition_concept_id\", \"condition_start_date\",\n \"condition_start_datetime\", \"condition_end_date\", \"condition_end_datetime\",\n \"condition_type_concept_id\", \"condition_status_concept_id\", \"stop_reason\", \"provider_id\",\n \"visit_occurrence_id\", \"visit_detail_id\", \"condition_source_value\",\n \"condition_source_concept_id\", \"condition_status_source_value\"},\n \"PROCEDURE\": {\"procedure_occurrence_id\", \"person_id\", \"procedure_concept_id\", \"procedure_date\",\n \"procedure_datetime\", \"procedure_end_date\", \"procedure_end_datetime\",\n \"procedure_type_concept_id\", \"modifier_concept_id\", \"quantity\", \"provider_id\",\n \"visit_occurrence_id\", \"visit_detail_id\", \"procedure_source_value\",\n \"procedure_source_concept_id\", \"modifier_source_value\"},\n \"PAYER_PLAN_PERIOD\": {\"payer_plan_period_id\", \"person_id\", \"payer_plan_period_start_date\",\n \"payer_plan_period_end_date\", \"payer_concept_id\", \"payer_source_value\",\n \"payer_source_concept_id\", \"plan_concept_id\", \"plan_source_value\",\n \"plan_source_concept_id\", \"sponsor_concept_id\", \"sponsor_source_value\",\n \"sponsor_source_concept_id\", \"family_source_value\", \"stop_reason_concept_id\",\n \"stop_reason_source_value\", \"stop_reason_source_concept_id\"},\n \"PROVIDER\": {\"provider_id\", \"provider_name\", \"npi\", \"dea\", \"specialty_concept_id\", \"care_site_id\",\n \"year_of_birth\", \"gender_concept_id\", \"provider_source_value\", \"specialty_source_value\",\n \"specialty_source_concept_id\", \"gender_source_value\", \"gender_source_concept_id\"},\n}\n\ndef _validate_mapping_against_cdm(rows: \"tuple[MappingRow, ...]\", logger) -> None:\n \"\"\"Reject mapping rows whose target table or field is not part of OMOP CDM 5.4.\"\"\"\n unexpected = []\n for row in rows:\n valid_columns = _REAL_CDM_COLUMNS.get(row.target_table)\n if valid_columns is None:\n unexpected.append(\n f\"{row.source_field}/{row.person_role}: unrecognized Target_Table {row.target_table!r}\"\n )\n continue\n if row.field_name.lower() not in valid_columns:\n unexpected.append(\n f\"{row.source_field}/{row.person_role}: Field_Name {row.field_name!r} is not a real \"\n f\"column on {row.target_table} (spreadsheet concept: {row.concept_name!r})\"\n )\n if unexpected:\n raise ValueError(\n \"Mapping spreadsheet has invalid OMOP CDM 5.4 Target_Table/Field_Name combinations:\\n\"\n + \"\\n\".join(f\" - {e}\" for e in unexpected)\n )\n\n\n\n@dataclass(frozen=True)\nclass MappingRow:\n source_field: str # lowercase NBER field name, e.g. \"rf_pdiab\"\n person_role: str # \"Child\", \"Mother\", or \"Father\"\n target_table: str # normalized OMOP table name, e.g. \"OBSERVATION\"\n field_name: str # OMOP column this row supplies, e.g. \"value_as_concept_id\"\n concept_id: int | None\n source_value_type: str | None # \"integer\"/\"float\" means copy the matched source value\n concept_name: str\n matcher: \"ValueMatcher\"\n\n\ndef _is_number(s: str) -> bool:\n \"\"\"True for anything float() accepts, e.g. \"9\", \"-5\", \"69.9\" -- used\n instead of str.isdigit() so decimal Values (e.g. bmi's \"13-69.9\") parse\n as ranges/exact values instead of falling through to literal-string\n matching, which no real decimal value would ever hit.\"\"\"\n try:\n float(s)\n return True\n except ValueError:\n return False\n\n\nclass ValueMatcher:\n \"\"\"\n Parses the spreadsheet's free-text Values column into something\n callable against an actual field value from the data.\n\n Recognized forms: blank/\"any\" (always matches -- used for field-level\n \"header\" rows that apply regardless of the specific code), exact\n tokens separated by \"or\"/\",\", and numeric ranges (\"0-30\", \"1 to 98\",\n \"13-69.9\" -- bounds/values may be decimal, e.g. bmi).\n \"\"\"\n\n def __init__(self, raw: str):\n raw = (raw or \"\").strip()\n self.raw = raw\n self.kind: str\n if raw == \"\" or raw.lower() == \"any\":\n self.kind = \"always\"\n return\n if raw.startswith(\">\") or raw.startswith(\"<\"):\n bound = raw[1:].strip()\n if _is_number(bound):\n self.kind = \"inequality\"\n self.op = raw[0]\n self.bound = float(bound)\n return\n low = raw.lower().replace(\" to \", \"-\")\n parts = [p.strip() for p in low.split(\"-\")]\n if len(parts) == 2 and all(_is_number(p) for p in parts):\n self.kind = \"range\"\n self.lo = float(parts[0])\n self.hi = float(parts[1])\n return\n self.kind = \"exact\"\n self.tokens = {t.strip().lower() for t in raw.replace(\",\", \" or \").split(\" or \")}\n\n def matches(self, value) -> bool:\n if value is None:\n return False\n value_s = str(value).strip()\n if value_s == \"\":\n return False\n if self.kind == \"always\":\n return True\n if self.kind == \"range\":\n try:\n v = float(value_s)\n except ValueError:\n return False\n return self.lo <= v <= self.hi\n if self.kind == \"inequality\":\n try:\n v = float(value_s)\n except ValueError:\n return False\n return v > self.bound if self.op == \">\" else v < self.bound\n if value_s.lower() in self.tokens:\n return True\n # Raw fields are read with dtype=str (pd.read_fwf), so numeric codes\n # keep their fixed-width zero-padding (e.g. \"09\") -- compare\n # numerically too so an exact token like \"9\" still matches \"09\",\n # and a decimal token like \"99.9\" still matches a decimal value.\n if _is_number(value_s):\n return any(_is_number(t) and float(t) == float(value_s) for t in self.tokens)\n return False\n\n\ndef _normalize_table(raw: str) -> str:\n return raw.strip().upper().replace(\" \", \"_\")\n\n\n_MAPPING_ROWS: tuple[MappingRow, ...] | None = None\n_PREFIX_GROUP_DEFAULTS: dict[tuple[str, str], int] | None = None\n_FIELD_INDEX: dict[tuple[str, str], list[MappingRow]] | None = None\n_MAPPING_DF_FROM_NODE: \"pd.DataFrame | None\" = None\n\n\ndef _load_mapping_df() -> \"pd.DataFrame\":\n if _MAPPING_DF_FROM_NODE is None:\n raise RuntimeError(\n f\"Expected mapping spreadsheet input from upstream csv node {MAPPING_CSV_NODE!r}. \"\n \"Connect that node to transform_facts (node-facts).\"\n )\n return _MAPPING_DF_FROM_NODE.copy()\n\n\ndef load_mapping_rows() -> tuple[MappingRow, ...]:\n global _MAPPING_ROWS\n if _MAPPING_ROWS is not None:\n return _MAPPING_ROWS\n\n out = []\n for r in _load_mapping_df().itertuples(index=False, name=None):\n if not r or not r[0]:\n continue\n nber_field, values, _nber_name, field_name, target_table, vocab_id, concept_name = r[:7]\n person_role = (r[11] or \"\").strip()\n if not person_role or not target_table or not field_name:\n continue\n if _PREFIX_GROUP_RE.match(str(nber_field).strip()):\n continue # prefix-group default row, handled by _prefix_group_defaults below\n field_name_s = str(field_name).strip()\n vocab_id_s = str(vocab_id).strip() if vocab_id is not None else \"\"\n concept_id = None\n source_value_type = None\n if vocab_id_s.lstrip(\"-\").isdigit():\n concept_id = int(vocab_id_s)\n elif field_name_s == \"value_as_number\" and vocab_id_s.lower() == \"integer\":\n source_value_type = \"integer\"\n elif field_name_s == \"value_as_number\" and vocab_id_s.lower() in {\"float\", \"number\"}:\n source_value_type = \"float\"\n else:\n continue # unsupported computed/imputed row\n source_field_s = str(nber_field).strip().lower()\n values_s = \"\" if values is None else str(values)\n out.append(MappingRow(\n source_field=source_field_s,\n person_role=person_role,\n target_table=_normalize_table(str(target_table)),\n field_name=field_name_s,\n concept_id=concept_id,\n source_value_type=source_value_type,\n concept_name=str(concept_name).strip() if concept_name else \"\",\n matcher=ValueMatcher(values_s),\n ))\n _MAPPING_ROWS = tuple(out)\n return _MAPPING_ROWS\n\n\ndef _prefix_group_defaults() -> dict[tuple[str, str], int]:\n \"\"\"(person_role, prefix) -> default observation_concept_id for that field family.\"\"\"\n global _PREFIX_GROUP_DEFAULTS\n if _PREFIX_GROUP_DEFAULTS is not None:\n return _PREFIX_GROUP_DEFAULTS\n\n out = {}\n for r in _load_mapping_df().itertuples(index=False, name=None):\n if not r or not r[0]:\n continue\n m = _PREFIX_GROUP_RE.match(str(r[0]).strip())\n if not m:\n continue\n vocab_id_s = str(r[5]).strip() if r[5] is not None else \"\"\n if not vocab_id_s.lstrip(\"-\").isdigit():\n continue\n person_role = (r[11] or \"\").strip()\n prefix = m.group(2).lower()\n out[(person_role, prefix)] = int(vocab_id_s)\n _PREFIX_GROUP_DEFAULTS = out\n return _PREFIX_GROUP_DEFAULTS\n\n\ndef _index_by_field() -> dict[tuple[str, str], list[MappingRow]]:\n global _FIELD_INDEX\n if _FIELD_INDEX is not None:\n return _FIELD_INDEX\n\n index: dict[tuple[str, str], list[MappingRow]] = {}\n for row in load_mapping_rows():\n index.setdefault((row.person_role, row.source_field), []).append(row)\n _FIELD_INDEX = index\n return _FIELD_INDEX\n\n\n# Fields whose \"none of the above\" answer belongs to a question family\n# with a different source-field prefix.\n_OBSERVATION_PREFIX_ALIAS = {\n (\"Mother\", \"no_mmorb\"): \"mm\",\n (\"Child\", \"no_abnorm\"): \"ab\",\n (\"Child\", \"no_congen\"): \"ca\",\n (\"Mother\", \"no_lbrdlv\"): \"ld\",\n}\n\ndef get_records(person_role: str, source_field: str, value):\n \"\"\"\n Returns {target_table: {field_name: concept_id, ...}, ...} for every\n target table that has at least one matching mapping row for this\n (person_role, source_field, value). Rows with no value match (e.g. the\n field is blank/not reported) contribute nothing.\n \"\"\"\n source_field = source_field.lower()\n candidates = _index_by_field().get((person_role, source_field), [])\n out: dict[str, dict[str, int | float]] = {}\n for row in candidates:\n if row.matcher.matches(value):\n if row.source_value_type == \"integer\":\n try:\n mapped_value = int(str(value).strip())\n except ValueError:\n continue\n out.setdefault(row.target_table, {})[row.field_name] = mapped_value\n continue\n if row.source_value_type == \"float\":\n try:\n mapped_value = float(str(value).strip())\n except ValueError:\n continue\n out.setdefault(row.target_table, {})[row.field_name] = mapped_value\n continue\n out.setdefault(row.target_table, {})[row.field_name] = row.concept_id\n\n if \"OBSERVATION\" in out and \"observation_concept_id\" not in out[\"OBSERVATION\"]:\n prefix = _OBSERVATION_PREFIX_ALIAS.get(\n (person_role, source_field), source_field.split(\"_\")[0]\n )\n default = _prefix_group_defaults().get((person_role, prefix))\n if default is not None:\n out[\"OBSERVATION\"][\"observation_concept_id\"] = default\n\n return out\n\n\n\n\nDEFAULT_TYPE_CONCEPT_ID = 32879 # Registry\n\nWEEKDAY_CODE_TO_PY = {1: 6, 2: 0, 3: 1, 4: 2, 5: 3, 6: 4, 7: 5} # 1=Sunday..7=Saturday -> Python Monday=0..Sunday=6\n\n\ndef _int_or_none(value):\n s = \"\" if value is None else str(value).strip()\n if s == \"\":\n return None\n try:\n return int(s)\n except ValueError:\n return None\n\n\ndef _float_or_none(value):\n s = \"\" if value is None else str(value).strip()\n if s == \"\":\n return None\n try:\n return float(s)\n except ValueError:\n return None\n\n\ndef _impute_child_dob(dob_yy, dob_mm, dob_wk):\n year, month, wk = _int_or_none(dob_yy), _int_or_none(dob_mm), _int_or_none(dob_wk)\n if year is None or month is None:\n return None, None, None\n if wk is None or wk not in WEEKDAY_CODE_TO_PY:\n return year, month, None\n target_py_weekday = WEEKDAY_CODE_TO_PY[wk]\n first_of_month = dt.date(year, month, 1)\n offset = (target_py_weekday - first_of_month.weekday()) % 7\n return year, month, 1 + offset\n\n\ndef _parse_dob_time(dob_tt) -> tuple[int, int] | None:\n \"\"\"Parse DOB_TT as zero-padded HHMM; 9999 means not stated.\"\"\"\n raw = \"\" if dob_tt is None else str(dob_tt).strip()\n if not raw.isdigit() or len(raw) > 4:\n return None\n s = raw.zfill(4)\n if s == \"9999\":\n return None\n hour, minute = int(s[:2]), int(s[2:])\n if not (0 <= hour <= 23 and 0 <= minute <= 59):\n return None\n return hour, minute\n\n\ndef dob_date_for(rec):\n year, month, day = _impute_child_dob(rec.get(\"DOB_YY\"), rec.get(\"DOB_MM\"), rec.get(\"DOB_WK\"))\n if year and month and day:\n return dt.date(year, month, day)\n return None\n\n\ndef _subtract_months(base_date, months):\n total = base_date.year * 12 + (base_date.month - 1) - months\n year, month = divmod(total, 12)\n day = min(base_date.day, 28)\n return dt.date(year, month + 1, day)\n\n\nclass IdGenerator:\n # Dense, chunk-safe id allocation: exactly 3 roles per record, so\n # record_idx*3 + offset + 1 produces 1,2,3,4,5,6,... with no gaps,\n # while still being computable independently per record (no shared\n # counter needed across chunks/workers).\n PERSON_ROLE_OFFSET = {\"Child\": 0, \"Mother\": 1, \"Father\": 2}\n\n @staticmethod\n def person_id(record_idx, role):\n return record_idx * 3 + IdGenerator.PERSON_ROLE_OFFSET[role] + 1\n\n @staticmethod\n def visit_occurrence_id(record_idx, role):\n return record_idx * 3 + IdGenerator.PERSON_ROLE_OFFSET[role] + 1\n\n @staticmethod\n def fact_id(record_idx, local_counter):\n # 500-id block per record: the destination CDM's fact id columns\n # (observation_id etc.) are 32-bit INTEGER (max 2,147,483,647) and the\n # full Nat2022 file has 3,676,029 records, so the block size must keep\n # the largest id under 2^31 (500 tops out at ~1.84e9) while exceeding\n # the most facts one record produces in a table (~60 observations).\n return (record_idx + 1) * 500 + local_counter\n\n\nPROVIDER_DIMENSION = [\n {\"specialty_concept_id\": 38004446, \"provider_source_value\": \"Physician\"},\n {\"specialty_concept_id\": 38003822, \"provider_source_value\": \"Osteopathic Practitioner\"},\n {\"specialty_concept_id\": 38004482, \"provider_source_value\": \"CNM/CM\"},\n {\"specialty_concept_id\": 38003807, \"provider_source_value\": \"Other midwife\"},\n {\"specialty_concept_id\": 0, \"provider_source_value\": \"Other/Unknown\"},\n]\nPROVIDER_ID_BY_CONCEPT = {row[\"specialty_concept_id\"]: i + 1 for i, row in enumerate(PROVIDER_DIMENSION)}\n\n\ndef provider_id_for(rec: dict) -> int | None:\n provider = get_records(\"Mother\", \"attend\", rec.get(\"ATTEND\")).get(\"PROVIDER\", {})\n return PROVIDER_ID_BY_CONCEPT.get(provider.get(\"specialty_concept_id\"))\n\n\n\n_CHILD_FIELDS = [\n # dmeth_rec deliberately excluded -- spreadsheet owner confirmed it's a collapsed\n # duplicate of rdmeth_rec (dmeth_rec=1 combines rdmeth_rec 1/2/5, dmeth_rec=2\n # combines rdmeth_rec 3/4/6) and to prefer rdmeth_rec, which has more granularity.\n # See questions_for_spreadsheet_owner.md #9.\n # mtran deliberately excluded -- spreadsheet owner confirmed mtran (\"Mother\n # Transferred\") belongs on the mother's own record only, not the child's; itran\n # (\"Infant Transferred\") is the child's own, separate fact. See new_findings.md.\n \"ab_anti\", \"ab_aven1\", \"ab_aven6\", \"ab_nicu\", \"ab_seiz\", \"ab_surf\", \"attend\",\n \"bfed\", \"ca_anen\", \"ca_cchd\", \"ca_cdh\", \"ca_cleft\", \"ca_clpal\",\n \"ca_disor\", \"ca_downs\", \"ca_gast\", \"ca_hypo\", \"ca_limb\", \"ca_mnsb\", \"ca_omph\",\n \"dmar\", \"dob_tt\", \"dplural\", \"fagecomb\", \"ilive\", \"itran\", \"ld_indl\",\n \"mar_p\", \"me_pres\", \"me_rout\", \"meduc\", \"no_abnorm\",\n \"no_congen\", \"ob_ecvs\", \"oegest_comb\", \"pay\", \"precare\",\n \"previs\", \"rdmeth_rec\", \"setorder_r\", \"apgar5\", \"apgar10\",\n \"combgest\", \"dbwt\",\n]\n_MOTHER_FIELDS = [\n # dlmp_mm/dlmp_yy deliberately excluded -- combined into one hand-computed\n # OBSERVATION row below instead of two independent per-field rows.\n # dmeth_rec deliberately excluded -- see _CHILD_FIELDS comment above; same\n # rdmeth_rec-preferred resolution applies to Mother's own copy.\n \"attend\", \"cig0_r\", \"cig1_r\", \"cig2_r\", \"cig3_r\",\n \"dmar\", \"illb_r\", \"ilop_r\", \"ip_chlam\", \"ip_gon\", \"ip_hepatb\", \"ip_hepatc\",\n \"ip_syph\", \"ld_anes\", \"ld_antb\", \"ld_augm\", \"ld_chor\", \"ld_indl\", \"ld_ster\",\n \"m_ht_in\", \"mager\", \"mar_p\", \"me_pres\", \"me_rout\", \"me_trial\", \"meduc\",\n \"mm_aicu\", \"mm_mtr\", \"mm_plac\", \"mm_rupt\", \"mm_uhyst\", \"mrace15\", \"mtran\",\n \"no_lbrdlv\", \"no_mmorb\", \"no_risks\", \"ob_ecvs\", \"oegest_comb\",\n \"pay\", \"precare\", \"previs\", \"rdmeth_rec\", \"rf_artec\", \"rf_cesar\",\n \"rf_ehype\", \"rf_fedrg\", \"rf_gdiab\", \"rf_ghype\", \"rf_inftr\", \"rf_pdiab\",\n \"rf_phype\", \"rf_ppterm\", \"sex\", \"setorder_r\", \"wic\", \"wtgain\", \"bmi\",\n \"combgest\", \"dbwt\", \"pwgt_r\", \"dplural\",\n \"priorlive\", \"priordead\", \"rf_cesarn\", \"priorterm\",\n]\n_FATHER_FIELDS = [\"fagecomb\", \"feduc\", \"frace15\"]\n\n\ndef _observation_date_for(field: str, value, dob_date: dt.date | None) -> dt.date | None:\n # OBSERVATION.observation_date is when the information was collected.\n # For this birth-certificate ETL, that is the child's birth date.\n return dob_date\n\n\n# Spreadsheet NBER Field names that don't match the raw bronze column name for the\n# same field (source_field.upper() -> real raw column). Confirmed against both\n# node-load-bronze's FIELDS list and UserGuide2022.pdf: the real fields are\n# CA_DOWN/IP_HEPB/IP_HEPC, not CA_DOWNS/IP_HEPATB/IP_HEPATC. Without this, rec.get()\n# always misses and these 3 fields silently never produce any record, on any real\n# data -- verified via _iter_field_records() returning [] for all three. See\n# questions_for_spreadsheet_owner.md.\n_RAW_FIELD_ALIAS = {\n \"ca_downs\": \"CA_DOWN\",\n \"ip_hepatb\": \"IP_HEPB\",\n \"ip_hepatc\": \"IP_HEPC\",\n}\n\n\ndef _iter_field_records(rec: dict, dob_date: dt.date | None):\n \"\"\"\n Yields (target_table, role, fact_dict) for every field/value/role match\n found via the spreadsheet-driven mapping, across all of OBSERVATION,\n MEASUREMENT, CONDITION, PROCEDURE, PAYER_PLAN_PERIOD, and PROVIDER.\n\n Each of the 8 build_() functions below filters this same stream\n down to its own table, rather than duplicating the field-matching logic\n once per table -- the per-table split is purely about which node emits\n which rows, not a second independent implementation of the mapping.\n \"\"\"\n for role, fields in ((\"Child\", _CHILD_FIELDS), (\"Mother\", _MOTHER_FIELDS), (\"Father\", _FATHER_FIELDS)):\n for field in fields:\n value = rec.get(_RAW_FIELD_ALIAS.get(field, field.upper()))\n if value is None or str(value).strip() == \"\":\n continue\n if field == \"dob_tt\" and _parse_dob_time(value) is None:\n continue\n if field in (\"illb_r\", \"ilop_r\"):\n interval = _int_or_none(value)\n if interval is not None and 0 <= interval <= 3:\n continue # plurality is mapped from DPLURAL, not these interval fields\n matches = get_records(role, field, value)\n for table, cols in matches.items():\n if table == \"OBSERVATION\":\n if field == \"mager\" and \"value_as_number\" not in cols:\n cols = {**cols, \"value_as_number\": _int_or_none(value)}\n if field in (\"illb_r\", \"ilop_r\") and dob_date is not None:\n interval = _int_or_none(value)\n if interval is not None and 4 <= interval <= 300:\n # The observation was collected at delivery; the calculated\n # historical date is its answer. CDM 5.4 has no value_as_date.\n cols = {**cols, \"value_as_string\": _subtract_months(dob_date, interval).replace(day=1).isoformat()}\n if field in (\"illb_r\", \"ilop_r\", \"precare\"):\n # Preserve the verbatim source answer separately from its\n # standardized numeric, concept, or calculated-date value.\n cols = {**cols, \"value_source_value\": str(value).strip()}\n yield \"OBSERVATION\", role, {\n \"observation_date\": _observation_date_for(field, value, dob_date),\n \"observation_type_concept_id\": DEFAULT_TYPE_CONCEPT_ID,\n **cols,\n }\n elif table == \"MEASUREMENT\":\n yield \"MEASUREMENT\", role, {\n \"measurement_date\": dob_date,\n \"measurement_type_concept_id\": DEFAULT_TYPE_CONCEPT_ID,\n **cols,\n }\n elif table == \"CONDITION\":\n yield \"CONDITION\", role, {\n \"condition_start_date\": dob_date,\n \"condition_type_concept_id\": DEFAULT_TYPE_CONCEPT_ID,\n **cols,\n }\n elif table == \"PROCEDURE\":\n yield \"PROCEDURE\", role, {\n \"procedure_date\": dob_date,\n \"procedure_type_concept_id\": DEFAULT_TYPE_CONCEPT_ID,\n **cols,\n }\n elif table == \"PAYER_PLAN_PERIOD\":\n # end_date left null rather than also set to dob_date -- the source\n # data has no real coverage-period dates at all (only \"which payer\"),\n # so forcing both dates to the birth date would assert a false\n # one-day coverage period. start_date must stay non-null (OMOP\n # requires it) and dob_date is the only date this data has to offer;\n # end_date left open is more honest than a fabricated end. Flagged\n # for confirmation in questions_for_spreadsheet_owner.md -- not a\n # settled fact.\n yield \"PAYER_PLAN_PERIOD\", role, {\n \"payer_plan_period_start_date\": dob_date,\n \"payer_plan_period_end_date\": None,\n **cols,\n }\n\n # dlmp_mm/dlmp_yy -> one combined OBSERVATION row (spreadsheet's own Notes on\n # dlmp_mm: \"Combine dlmp_yy and dlmp_mm and assume 1st day of month\"), instead of\n # two independent per-field rows sharing the same concept (3002314). dlmp_yy's own\n # Values (\"2020\") is a single literal example, not a real range, so the concept is\n # looked up via dlmp_mm's row (Values \"1 to 12\") instead. DLMP_MM=99/DLMP_YY=9999\n # are the official \"unknown or not stated\" sentinels -- no date computed for those.\n dlmp_month = _int_or_none(rec.get(\"DLMP_MM\"))\n dlmp_year = _int_or_none(rec.get(\"DLMP_YY\"))\n if dlmp_month is not None and 1 <= dlmp_month <= 12 and dlmp_year is not None and dlmp_year != 9999:\n fact = {\n \"observation_date\": dob_date,\n \"observation_type_concept_id\": DEFAULT_TYPE_CONCEPT_ID,\n \"value_as_string\": dt.date(dlmp_year, dlmp_month, 1).isoformat(),\n }\n fact.update(get_records(\"Mother\", \"dlmp_mm\", dlmp_month).get(\"OBSERVATION\", {}))\n yield \"OBSERVATION\", \"Mother\", fact\n\n\ndef build_persons(rec: dict, record_idx: int) -> list[dict]:\n persons = []\n\n dob_year, dob_month, dob_day = _impute_child_dob(rec.get(\"DOB_YY\"), rec.get(\"DOB_MM\"), rec.get(\"DOB_WK\"))\n birth_datetime = None\n if dob_year is not None and dob_month is not None and dob_day is not None:\n # DOB_TT unknown/unparseable (incl. NCHS's \"9999\" sentinel) leaves\n # birth_datetime null rather than assuming midnight -- birth_datetime is\n # nullable in the destination schema, so no time is a truer statement than\n # an assumed one. See questions_for_spreadsheet_owner.md #15.\n dob_time = _parse_dob_time(rec.get(\"DOB_TT\"))\n if dob_time is not None:\n birth_datetime = dt.datetime(dob_year, dob_month, dob_day, dob_time[0], dob_time[1])\n child = {\n \"person_id\": IdGenerator.person_id(record_idx, \"Child\"),\n \"year_of_birth\": dob_year,\n \"month_of_birth\": dob_month,\n \"day_of_birth\": dob_day,\n \"birth_datetime\": birth_datetime,\n }\n child.update(get_records(\"Child\", \"sex\", rec.get(\"SEX\")).get(\"PERSON\", {}))\n mother_race6 = _int_or_none(rec.get(\"MRACE6\"))\n father_race6 = _int_or_none(rec.get(\"FRACE6\"))\n if mother_race6 in range(1, 7) and father_race6 in range(1, 7):\n if mother_race6 == father_race6:\n child.update(get_records(\"Mother\", \"mrace6\", mother_race6).get(\"PERSON\", {}))\n else:\n child[\"race_concept_id\"] = 1546847 # More than one race\n persons.append(child)\n\n mager = _int_or_none(rec.get(\"MAGER\"))\n mother = {\n \"person_id\": IdGenerator.person_id(record_idx, \"Mother\"),\n \"year_of_birth\": (dob_year - mager) if (dob_year is not None and mager is not None) else None,\n }\n mother.update(get_records(\"Mother\", \"sex\", \"F\").get(\"PERSON\", {}))\n for field in (\"mrace6\", \"mhispx\", \"mracehisp\"):\n mother.update(get_records(\"Mother\", field, rec.get(field.upper())).get(\"PERSON\", {}))\n persons.append(mother)\n\n fagecomb = _int_or_none(rec.get(\"FAGECOMB\"))\n if fagecomb == 99:\n # NCHS's own \"unknown or not stated\" sentinel for FAGECOMB, not a real\n # age -- confirmed via the spreadsheet's own fagecomb=99 rows (\"Unknown\n # or Not Stated\" / \"Age unknown\"). Excluded here so the father's\n # year_of_birth isn't computed as if 99 were a real age.\n fagecomb = None\n father = {\n \"person_id\": IdGenerator.person_id(record_idx, \"Father\"),\n \"year_of_birth\": (dob_year - fagecomb) if (dob_year is not None and fagecomb is not None) else None,\n }\n father.update(get_records(\"Father\", \"sex\", \"M\").get(\"PERSON\", {}))\n for field in (\"frace6\", \"fhispx\", \"fracehisp\"):\n father.update(get_records(\"Father\", field, rec.get(field.upper())).get(\"PERSON\", {}))\n persons.append(father)\n\n return persons\n\n\ndef build_visits(rec: dict, record_idx: int, dob_date: dt.date | None) -> list[dict]:\n visits = []\n provider_id = provider_id_for(rec)\n for role in (\"Child\", \"Mother\"):\n match = get_records(role, \"bfacil\", rec.get(\"BFACIL\"))\n if \"VISIT_OCCURRENCE\" not in match:\n continue\n visit = {\n \"visit_occurrence_id\": IdGenerator.visit_occurrence_id(record_idx, role),\n \"person_id\": IdGenerator.person_id(record_idx, role),\n \"visit_start_date\": dob_date,\n \"visit_end_date\": dob_date,\n \"visit_type_concept_id\": DEFAULT_TYPE_CONCEPT_ID,\n \"provider_id\": provider_id,\n }\n # spreadsheet field name is \"Visit_Concept_id\" (case-as-published); normalize here\n for k, v in match[\"VISIT_OCCURRENCE\"].items():\n visit[k.lower()] = v\n visits.append(visit)\n return visits\n\n\n_ID_COLUMN = {\n \"OBSERVATION\": \"observation_id\",\n \"MEASUREMENT\": \"measurement_id\",\n \"CONDITION\": \"condition_occurrence_id\",\n \"PROCEDURE\": \"procedure_occurrence_id\",\n \"PAYER_PLAN_PERIOD\": \"payer_plan_period_id\",\n}\n\nSQL_TABLE_NAME = {\n \"PERSON\": \"person\",\n \"VISIT_OCCURRENCE\": \"visit_occurrence\",\n \"OBSERVATION\": \"observation\",\n \"MEASUREMENT\": \"measurement\",\n \"CONDITION\": \"condition_occurrence\",\n \"PROCEDURE\": \"procedure_occurrence\",\n \"PAYER_PLAN_PERIOD\": \"payer_plan_period\",\n}\n\nTABLE_COLUMNS = {\n \"PERSON\": [\"person_id\", \"year_of_birth\", \"month_of_birth\", \"day_of_birth\", \"birth_datetime\",\n \"gender_concept_id\", \"race_concept_id\", \"ethnicity_concept_id\"],\n \"VISIT_OCCURRENCE\": [\"visit_occurrence_id\", \"person_id\", \"visit_start_date\", \"visit_end_date\",\n \"visit_type_concept_id\", \"visit_concept_id\", \"provider_id\"],\n \"OBSERVATION\": ['person_id', 'observation_id', 'observation_date', 'observation_type_concept_id',\n 'observation_concept_id', 'qualifier_concept_id', 'unit_concept_id',\n 'value_as_concept_id', 'value_as_number', 'value_as_string',\n 'value_source_value', 'provider_id'],\n \"MEASUREMENT\": ['person_id', 'measurement_id', 'measurement_date', 'measurement_type_concept_id',\n 'measurement_concept_id', 'unit_concept_id', 'value_as_concept_id', 'value_as_number',\n 'provider_id'],\n \"CONDITION\": ['person_id', 'condition_occurrence_id', 'condition_start_date', 'condition_type_concept_id',\n 'condition_concept_id', 'condition_status_concept_id', 'provider_id'],\n \"PROCEDURE\": ['person_id', 'procedure_occurrence_id', 'procedure_date', 'procedure_type_concept_id',\n 'procedure_concept_id', 'provider_id'],\n \"PAYER_PLAN_PERIOD\": ['person_id', 'payer_plan_period_id', 'payer_plan_period_start_date',\n 'payer_plan_period_end_date', 'payer_concept_id'],\n}\n\n\ndef build_facts(rec: dict, record_idx: int, dob_date: dt.date | None) -> dict[str, list[dict]]:\n \"\"\"\n One pass over _iter_field_records per record, bucketed by target table --\n _iter_field_records computes matches for every table in a single\n generator pass. Covers OBSERVATION/MEASUREMENT/CONDITION/\n PROCEDURE/PAYER_PLAN_PERIOD only -- PERSON and VISIT_OCCURRENCE are\n built separately (build_persons/build_visits) since they're driven by\n a handful of targeted get_records() calls, not this field-family scan.\n \"\"\"\n ob_ecvs = str(rec.get(\"OB_ECVS\") or \"\").strip().upper()\n ob_ecvf = str(rec.get(\"OB_ECVF\") or \"\").strip().upper()\n if ob_ecvs and ob_ecvf:\n if ob_ecvs not in {\"Y\", \"N\"} or ob_ecvf not in {\"Y\", \"N\"} or ob_ecvs == ob_ecvf:\n raise ValueError(\n f\"OB_ECVS/OB_ECVF consistency check failed at ROW_IDX={record_idx}: \"\n f\"OB_ECVS={ob_ecvs!r}, OB_ECVF={ob_ecvf!r}; expected Y/N or N/Y\"\n )\n\n provider_id = provider_id_for(rec)\n counters: dict[str, int] = {}\n out: dict[str, list[dict]] = {table: [] for table in _ID_COLUMN}\n for table, role, fact in _iter_field_records(rec, dob_date):\n if table not in _ID_COLUMN:\n continue # e.g. PROVIDER_USAGE -- not one of this node's tables\n if table in {\"OBSERVATION\", \"MEASUREMENT\", \"CONDITION\", \"PROCEDURE\"}:\n fact = {\"provider_id\": provider_id, **fact}\n counters[table] = counters.get(table, 0) + 1\n out[table].append({\n \"person_id\": IdGenerator.person_id(record_idx, role),\n _ID_COLUMN[table]: IdGenerator.fact_id(record_idx, counters[table]),\n **fact,\n })\n return out\n\n\ndef _sanitize_cache_id(raw_id: str) -> str:\n \"\"\"\n Mirrors the platform's own Dataset.cacheId derivation\n (portal/src/dataset/entity/dataset.entity.ts: sanitizeIdForCacheId) --\n hyphens aren't valid in a bare SQL/DuckDB identifier, so the Dataset's\n UUID gets converted the same way here before use as a trex cache_id.\n \"\"\"\n cleaned = raw_id.replace(\"-\", \"_\")\n return f\"_{cleaned}\" if cleaned[:1].isdigit() else cleaned\n\n\ndef exec(myinput):\n \"\"\"\n Gold transform: reads the staging CSV written by node-load-bronze in\n chunks, builds the OMOP rows in Python, writes each chunk's rows\n per-table to a CSV in the shared trex-volume staging dir, and loads each\n with one `INSERT INTO
(cols) SELECT * FROM read_csv('')` -- trex's DuckDB reads the file straight from disk at native\n speed. Each load statement is atomic, and row counts are verified per\n table at the end; a mismatch raises.\n \"\"\"\n import csv\n import time\n import uuid\n\n logger = get_run_logger()\n\n global _MAPPING_DF_FROM_NODE\n mapping_input = myinput.get(MAPPING_CSV_NODE)\n if mapping_input is None or mapping_input.result is None:\n raise RuntimeError(\n f\"Expected mapping spreadsheet input from upstream csv node {MAPPING_CSV_NODE!r}. \"\n \"Connect that node to transform_facts (node-facts).\"\n )\n if not isinstance(mapping_input.result, pd.DataFrame):\n raise RuntimeError(\n f\"Upstream mapping input {MAPPING_CSV_NODE!r} must be a pandas DataFrame, got \"\n f\"{type(mapping_input.result).__name__}\"\n )\n _MAPPING_DF_FROM_NODE = mapping_input.result.copy()\n\n _validate_mapping_against_cdm(load_mapping_rows(), logger)\n cache_catalog = _sanitize_cache_id(dataset_id)\n dbdao = DBDao(database_code=destination_database_code, cache_id=cache_catalog, dialect=SupportedDatabaseDialects.TREX)\n dest_schema = f\"{cache_catalog}.{destination_schema_name}\"\n\n # Input: the staging CSV written by node-load-bronze. The .complete marker\n # holds the row count and only exists if bronze finished; csv-without-\n # marker means an interrupted bronze run and is refused.\n src_fname = f\"{cache_catalog}_nat2022_raw.csv\"\n src_csv_path = os.path.join(STAGING_DIR_FLOW, src_fname)\n src_complete_path = src_csv_path + \".complete\"\n if not (os.path.exists(src_csv_path) and os.path.exists(src_complete_path)):\n raise RuntimeError(\n f\"Staging CSV from node-load-bronze not found or incomplete \"\n f\"({src_csv_path}; marker present: {os.path.exists(src_complete_path)}). \"\n f\"Run node-load-bronze first -- a CSV without its .complete marker means \"\n f\"the bronze run was interrupted mid-write.\"\n )\n with open(src_complete_path) as f:\n total = int(f.read().strip())\n\n for sql_table in SQL_TABLE_NAME.values():\n # destination_schema_name.
is created by the OMOP CDM plugin, not here --\n # truncate rather than drop so we never touch its schema/table definition.\n dbdao.truncate_table(dest_schema, sql_table)\n\n size = int(chunk_size)\n\n # Self-healing sweep: staging files are deleted right after each bulk\n # load, but a hard-killed run orphans its in-flight file. Current-run\n # filenames get a fresh run_token below, so anything matching these\n # prefixes now is stale by construction; other datasets' files\n # (different prefix) are never touched.\n table_prefixes = tuple(f\"{cache_catalog}_{t}_\" for t in SQL_TABLE_NAME.values())\n removed = 0\n for f in os.listdir(STAGING_DIR_FLOW):\n if f.startswith(table_prefixes) and f.endswith(\".csv\"):\n try:\n os.remove(os.path.join(STAGING_DIR_FLOW, f))\n removed += 1\n except OSError:\n pass\n if removed:\n logger.info(f\"Removed {removed} stale staging file(s) left by a previous interrupted run\")\n\n run_token = uuid.uuid4().hex\n\n def _bulk_load(sql_table, columns, values, tag):\n \"\"\"One CSV in the shared staging dir + one read_csv INSERT for this table-chunk.\"\"\"\n fname = f\"{cache_catalog}_{sql_table}_{tag}_{run_token}.csv\"\n fpath = os.path.join(STAGING_DIR_FLOW, fname)\n with open(fpath, \"w\", newline=\"\") as f:\n w = csv.writer(f)\n w.writerow(columns)\n for row in values:\n w.writerow([\"\" if v is None else v for v in row])\n col_list = \", \".join(f'\"{c}\"' for c in columns)\n try:\n dbdao.execute_sql(\n f\"INSERT INTO {dest_schema}.{sql_table} ({col_list}) \"\n f\"SELECT * FROM read_csv('{STAGING_DIR_TREX}/{fname}', header=true, all_varchar=true)\"\n )\n finally:\n os.remove(fpath)\n\n expected_counts = {table: 0 for table in SQL_TABLE_NAME}\n processed = 0\n # Sequential single-pass read of the staging CSV -- flat cost per chunk,\n # order preserved (ROW_IDX rides along as a column and record_idx comes\n # from its stored value, not from position). dtype=str + keep_default_na\n # =False gives '' for missing values, which the transform treats the same\n # as NULL.\n for chunk_idx, src_chunk in enumerate(pd.read_csv(src_csv_path, dtype=str, keep_default_na=False, chunksize=size)):\n t_chunk = time.time()\n rows_by_table = {table: [] for table in SQL_TABLE_NAME}\n for rec in src_chunk.to_dict(\"records\"):\n record_idx = int(rec[\"ROW_IDX\"])\n dob_date = dob_date_for(rec)\n rows_by_table[\"PERSON\"].extend(build_persons(rec, record_idx))\n rows_by_table[\"VISIT_OCCURRENCE\"].extend(build_visits(rec, record_idx, dob_date))\n built = build_facts(rec, record_idx, dob_date)\n for table, rows in built.items():\n rows_by_table[table].extend(rows)\n t_transform = time.time() - t_chunk\n\n chunk_rows = 0\n for table, sql_table in SQL_TABLE_NAME.items():\n columns = TABLE_COLUMNS[table]\n values = [tuple(row.get(col) for col in columns) for row in rows_by_table[table]]\n if not values:\n continue\n expected_counts[table] += len(values)\n chunk_rows += len(values)\n t_tbl = time.time()\n _bulk_load(sql_table, columns, values, str(chunk_idx))\n logger.info(f\" {sql_table}: {len(values):,} rows bulk-loaded in {time.time() - t_tbl:.1f}s\")\n\n processed += len(src_chunk)\n done = processed\n logger.info(\n f\"{done:,}/{total:,} records -> {chunk_rows:,} OMOP rows \"\n f\"(transform {t_transform:.0f}s, chunk total {time.time() - t_chunk:.0f}s)\"\n )\n\n if processed != total:\n raise RuntimeError(f\"staging CSV row count mismatch: marker says {total:,}, read {processed:,}\")\n\n mismatches = []\n for table, sql_table in SQL_TABLE_NAME.items():\n actual = int(dbdao.execute_sql(f\"SELECT COUNT(*) FROM {dest_schema}.{sql_table}\", fetch=True)[0][0])\n if actual != expected_counts[table]:\n mismatches.append(f\"{sql_table}: expected {expected_counts[table]:,}, table has {actual:,}\")\n if mismatches:\n raise RuntimeError(\"row-count verification failed -- \" + \"; \".join(mismatches))\n logger.info(\"Row-count verification passed for all tables: \"\n + \", \".join(f\"{SQL_TABLE_NAME[t]}={expected_counts[t]:,}\" for t in SQL_TABLE_NAME))\n\n return (f\"Loaded {total} records into person, visit_occurrence, observation, measurement, \"\n f\"condition_occurrence, procedure_occurrence, payer_plan_period\")\n"},"type":"python_node","width":350,"height":210,"position":{"x":630,"y":110},"selected":false,"dragHandle":"","sourcePosition":"right","targetPosition":"left"},{"id":"node-cleanup","data":{"name":"cleanup_raw_staging","description":"Cleanup: delete the staging CSV and marker for this dataset from the shared volume","python_code":"# Deployment-topology constant -- see node-load-bronze for the rationale.\nSTAGING_DIR_FLOW = \"/app/duckdb_data/flow_staging\"\n\n\ndef _sanitize_cache_id(raw_id: str) -> str:\n \"\"\"\n Mirrors the platform's own Dataset.cacheId derivation\n (portal/src/dataset/entity/dataset.entity.ts: sanitizeIdForCacheId) --\n hyphens aren't valid in a bare SQL/DuckDB identifier, so the Dataset's\n UUID gets converted the same way here before use as a trex cache_id.\n \"\"\"\n cleaned = raw_id.replace(\"-\", \"_\")\n return f\"_{cleaned}\" if cleaned[:1].isdigit() else cleaned\n\n\ndef exec(myinput):\n logger = get_run_logger()\n cache_catalog = _sanitize_cache_id(dataset_id)\n\n # The staging CSV is scratch for transform_facts only; delete it after a\n # successful transform so ~5GB doesn't sit on the shared trex volume\n # between runs. Every outcome is logged individually -- a failed delete\n # must never look like a clean run.\n removed, failed = [], []\n targets = []\n csv_path = os.path.join(STAGING_DIR_FLOW, f\"{cache_catalog}_nat2022_raw.csv\")\n targets.extend([csv_path, csv_path + \".complete\"])\n # stragglers from this dataset (orphaned files from interrupted runs)\n try:\n for f in os.listdir(STAGING_DIR_FLOW):\n p = os.path.join(STAGING_DIR_FLOW, f)\n if f.startswith(f\"{cache_catalog}_nat2022_raw\") and p not in targets:\n targets.append(p)\n except OSError as e:\n logger.warning(f\"Could not list staging dir {STAGING_DIR_FLOW}: {e}\")\n for path in targets:\n if not os.path.exists(path):\n continue\n try:\n os.remove(path)\n removed.append(os.path.basename(path))\n except OSError as e:\n failed.append(f\"{os.path.basename(path)} ({e})\")\n if failed:\n logger.warning(f\"Failed to remove staging file(s): {'; '.join(failed)} -- \"\n f\"check ownership/permissions on {STAGING_DIR_FLOW}\")\n\n logger.info(f\"Removed staging artifacts: {', '.join(removed) if removed else 'none found'}\"\n + (f\"; FAILED: {len(failed)}\" if failed else \"\"))\n return f\"Cleaned up staging CSV ({len(removed)} removed, {len(failed)} failed)\""},"type":"python_node","width":350,"height":210,"position":{"x":1130,"y":110},"selected":false,"dragHandle":"","sourcePosition":"right","targetPosition":"left"},{"id":"node-provider","data":{"name":"transform_provider","description":"Gold: build the OMOP PROVIDER table (static dimension, 5 attendant types)","python_code":"PROVIDER_DIMENSION = [\n {\"specialty_concept_id\": 38004446, \"provider_source_value\": \"Physician\"},\n {\"specialty_concept_id\": 38003822, \"provider_source_value\": \"Osteopathic Practitioner\"},\n {\"specialty_concept_id\": 38004482, \"provider_source_value\": \"CNM/CM\"},\n {\"specialty_concept_id\": 38003807, \"provider_source_value\": \"Other midwife\"},\n {\"specialty_concept_id\": 0, \"provider_source_value\": \"Other/Unknown\"},\n]\nPROVIDER_ID_BY_CONCEPT = {row[\"specialty_concept_id\"]: i + 1 for i, row in enumerate(PROVIDER_DIMENSION)}\n\n\n\nTABLE_COLUMNS = [\"provider_id\", \"specialty_concept_id\", \"provider_source_value\"]\n\n\ndef build_provider_table() -> pd.DataFrame:\n rows = []\n for i, row in enumerate(PROVIDER_DIMENSION):\n rows.append({\"provider_id\": i + 1, **row})\n return pd.DataFrame(rows, columns=TABLE_COLUMNS)\n\n\n\ndef _sanitize_cache_id(raw_id: str) -> str:\n \"\"\"\n Mirrors the platform's own Dataset.cacheId derivation\n (portal/src/dataset/entity/dataset.entity.ts: sanitizeIdForCacheId) --\n hyphens aren't valid in a bare SQL/DuckDB identifier, so the Dataset's\n UUID gets converted the same way here before use as a trex cache_id.\n \"\"\"\n cleaned = raw_id.replace(\"-\", \"_\")\n return f\"_{cleaned}\" if cleaned[:1].isdigit() else cleaned\n\n\ndef exec(myinput):\n logger = get_run_logger()\n cache_catalog = _sanitize_cache_id(dataset_id)\n dbdao = DBDao(database_code=destination_database_code, cache_id=cache_catalog, dialect=SupportedDatabaseDialects.TREX)\n dest_schema = f\"{cache_catalog}.{destination_schema_name}\"\n\n # destination_schema_name.provider is created by the OMOP CDM plugin, not here --\n # truncate rather than drop so we never touch its schema/table definition.\n dbdao.truncate_table(dest_schema, \"provider\")\n provider_df = build_provider_table()\n values = [tuple(row.get(col) for col in TABLE_COLUMNS) for row in provider_df.to_dict(\"records\")]\n dbdao.batch_insert_values(dest_schema, \"provider\", TABLE_COLUMNS, values)\n\n logger.info(f\"Loaded {len(provider_df)} rows into {destination_schema_name}.provider\")\n return f\"Loaded {len(provider_df)} rows into {destination_schema_name}.provider\"\n"},"type":"python_node","width":350,"height":210,"position":{"x":1630,"y":110},"selected":false,"dragHandle":"","sourcePosition":"right","targetPosition":"left"},{"id":"1c6cc31b-fb15-4628-88f5-5c37abe4bec0","data":{"name":"csv_node_0","columns":[],"encoding":"utf-8","delimiter":",","hasheader":true,"description":"Describe the task of node csv_node_0"},"type":"csv_node","width":353,"height":213,"dragging":false,"position":{"x":120,"y":-180},"selected":true,"dragHandle":"","sourcePosition":"right","targetPosition":"left","positionAbsolute":{"x":120,"y":-180}}],"edges":[{"id":"edge-node-load-bronze-node-facts","source":"node-load-bronze","target":"node-facts","sourceHandle":"source_node-load-bronze_object","targetHandle":"target_node-facts_any"},{"id":"edge-node-facts-node-cleanup","source":"node-facts","target":"node-cleanup","sourceHandle":"source_node-facts_object","targetHandle":"target_node-cleanup_any"},{"id":"edge-node-cleanup-node-provider","source":"node-cleanup","target":"node-provider","sourceHandle":"source_node-cleanup_object","targetHandle":"target_node-provider_any"},{"id":"reactflow__edge-1c6cc31b-fb15-4628-88f5-5c37abe4bec0source_1c6cc31b-fb15-4628-88f5-5c37abe4bec0_any-node-factstarget_node-facts_any","source":"1c6cc31b-fb15-4628-88f5-5c37abe4bec0","target":"node-facts","sourceHandle":"source_1c6cc31b-fb15-4628-88f5-5c37abe4bec0_any","targetHandle":"target_node-facts_any"}],"variables":[{"key":"nat2022_filename","value":""},{"key":"dataset_id","value":""},{"key":"chunk_size","value":"200000"}],"importLibs":["import os","import re","import pandas as pd","import datetime as dt","from dataclasses import dataclass","from prefect.logging import get_run_logger","from _shared_flow_utils.types import SupportedDatabaseDialects","from _shared_flow_utils.dao.DBDao import DBDao"],"databases":[{"code":"","name":"destination_database_code"}],"schemas":[{"name":"destination_schema_name","schema":""}]} \ No newline at end of file diff --git a/flows/ETL_US_BirthData_to_OMOP.md b/flows/ETL_US_BirthData_to_OMOP.md new file mode 100644 index 0000000..4fb9365 --- /dev/null +++ b/flows/ETL_US_BirthData_to_OMOP.md @@ -0,0 +1,219 @@ +# US Birth Data to OMOP ETL Template + +This runbook explains how to import and run +[`ETL_US_BirthData_to_OMOP.json`](./ETL_US_BirthData_to_OMOP.json). + +## Data source + +The input is the 2022 US birth public-use data from the CDC/NCHS +[Vital Statistics Online Data Portal](https://www.cdc.gov/nchs/data_access/vitalstatsonline.htm). + +The template currently retains `nat2022` in several variable names and staging-file names. These are implementation names; when loading 2022 data, verify that the fixed-width positions used by the parser match the 2022 User Guide before running the complete dataset. + +This guide uses: + +```text +Unziped_2022_US_birth_data.txt +``` + +The mapping input is: + +```text +Merged_US_Birth_Data.csv # necessary to be uploaded locally +``` + +## Prerequisites + +- A running D2E environment. +- An OMOP CDM 5.4 dataset created before running the ETL. +- The dataset ID copied from the Datasets page. +- The destination database code and schema name. +- The fixed-width birth-data file available on the host. +- The `Merged_US_Birth_Data.csv` mapping file. + +## Step 1: Create the dataset and set `dataset_id` + +Create the destination OMOP CDM 5.4 dataset before running the flow. Creating it through the Dataset page, copy the dataset ID which will be used in later step 4. + + +## Step 2: Mount the source-data directory + +Mount the host directory containing the fixed-width source file into the flow container at `/app/data_load`: + +- Open a terminal in the d2e directory. +- Run the following commands to define directories: + +```sh +export BIRTH_DATA_DIR="/absolute/path/to/birth_data" +yq -i '.services.alp-dataflow-gen-worker.volumes = ((.services.alp-dataflow-gen-worker.volumes // []) + [strenv(BIRTH_DATA_DIR) + ":/app/data_load"] | unique)' docker-compose.yml +``` + +Restart D2E to apply the updated container mount: + +```sh +d2e stop +d2e start +``` + +After the restart, verify that the worker can see the mounted files: + +```sh +docker exec alp-dataflow-gen-worker ls -l /app/data_load +``` + +## Step 3: Import and save the template + +1. Import `ETL_US_BirthData_to_OMOP.json` in the ETL page within admin portal. +2. Save the imported flow immediately. + +## Step 4: Configure the source filename and destination +1. Open variables setting drawer +alt text + +2. Then set individual variables as showen below: +alt text + +- Set dataset_id from step 1 +- Set "nat2022_filename" to the filename inside `/app/data_load`: +- Set the correct destination database and schema in the flow configuration: +```text +dataset_id= +nat2022_filename= +destination_database_code= +destination_schema_name= +``` + +3. Save the configuration of flow immediately. + +## Step 5: Upload local mapping table via CSV node + +- In CSV node, upload the mapping table using exact name of `Merged_US_Birth_Data.csv`. +- Ensure that the CSV node name is `csv_node_0` + +*The Python transform expects its mapping DataFrame from `csv_node_0`. A differently named or disconnected node will cause the flow to fail. + +## Step 6: Run the flow + +The ETL writes the following tables: + +- `person` +- `visit_occurrence` +- `observation` +- `measurement` +- `condition_occurrence` +- `procedure_occurrence` +- `payer_plan_period` +- `provider` + +## Step 7: Verify the cache tables + +### Get the cache catalog ID + +Run below command to query catalog_id using the dataset UUID configured in the ETL: + +```sh +export DATASET_ID = +SCHEMA_NAME = +CACHE_ID="$( + docker exec d2e-minerva-postgres-1 \ + psql -U postgres -d alp -tA \ + -c "SELECT cache_id FROM portal.dataset WHERE id = '${DATASET_ID}'::uuid;" +)" +``` + +If the deployment uses a different `PROJECT_NAME`, replace `d2e-minerva-postgres-1` and `d2e-trex` with the actual Minerva PostgreSQL container name shown by: + +```sh +docker ps --filter 'name=minerva-postgres' --format '{{.Names}}' +docker ps --filter 'name=trex' --format '{{.Names}}' +``` + +### 1. First list the destination tables +Run below command to get the list of all tables in target schema + +```sql +TREX_SQL_PASSWORD="$( + docker exec d2e-trex printenv TREX__SQL__PASSWORD +)" + +docker exec -i \ + -e PGPASSWORD="$TREX_SQL_PASSWORD" \ + d2e-minerva-postgres-1 \ + psql \ + -h d2e-trex \ + -p 5433 \ + -U postgres \ + -d "$CACHE_ID" \ + -v table_catalog="$CACHE_ID" \ + -v table_schema="$SCHEMA_NAME" \ + -tA <<'SQL' +SELECT + table_catalog, + table_schema, + table_name +FROM information_schema.tables +WHERE table_catalog = :'table_catalog' + AND table_schema = :'table_schema' + AND table_type = 'BASE TABLE' +ORDER BY table_name; +SQL +``` + +### 2. List the row counts of each table in target schema +- Generate exact `COUNT(*)` queries for all tables in the destination schema + +```sh +COUNT_QUERY="$( +docker exec -i \ + -e PGPASSWORD="$TREX_SQL_PASSWORD" \ + d2e-minerva-postgres-1 \ + psql \ + -h d2e-trex \ + -p 5433 \ + -U postgres \ + -d "$CACHE_ID" \ + -v table_catalog="$CACHE_ID" \ + -v table_schema="$SCHEMA_NAME" \ + -tA <<'SQL' +SELECT string_agg( + 'SELECT ''' + || replace(table_name, '''', '''''') + || ''' AS table_name, COUNT(*) AS row_count FROM "' + || replace(table_catalog, '"', '""') + || '"."' + || replace(table_schema, '"', '""') + || '"."' + || replace(table_name, '"', '""') + || '"', + ' UNION ALL ' +) || ' ORDER BY row_count DESC' +FROM information_schema.tables +WHERE table_catalog = :'table_catalog' + AND table_schema = :'table_schema' + AND table_type = 'BASE TABLE'; +SQL +)" +``` + +- Display the row counts of each table in target schema + +```sh +docker exec \ + -e PGPASSWORD="$TREX_SQL_PASSWORD" \ + d2e-minerva-postgres-1 \ + psql \ + -h d2e-trex \ + -p 5433 \ + -U postgres \ + -d "$CACHE_ID" \ + -c "$COUNT_QUERY" +``` +A sample dataset with 10 rows will result in 30 rows in person table: +alt text + +## Common failures + +- **Import libraries disappear:** Save the flow immediately after importing the JSON. +- **Source file is not found:** Confirm the host-directory mount and `nat2022_filename` value. +- **Wrong cache receives data:** Confirm that `dataset_id` is the ID returned for the intended Web API dataset. +- **Wrong destination:** Confirm both `destination_database_code` and `destination_schema_name`. diff --git a/flows/image-1.png b/flows/image-1.png new file mode 100644 index 0000000..787dad1 Binary files /dev/null and b/flows/image-1.png differ diff --git a/flows/image-2.png b/flows/image-2.png new file mode 100644 index 0000000..59a6a58 Binary files /dev/null and b/flows/image-2.png differ diff --git a/flows/image.png b/flows/image.png new file mode 100644 index 0000000..07a7359 Binary files /dev/null and b/flows/image.png differ