diff --git a/core/censorship.py b/core/censorship.py new file mode 100644 index 0000000..74cbf7f --- /dev/null +++ b/core/censorship.py @@ -0,0 +1,391 @@ +"""Utilities for applying censorship rules and building censorship summaries.""" + +import json +from typing import Any +from datetime import datetime, timezone + +from google.cloud import bigquery, storage + +from core import constants, utils, destination_table_builder + +def get_first_cid(column_name: str) -> str | None: + cids = utils.extract_ordered_concept_ids(column_name) + return cids[0] if cids else None + +def is_unrecognized_censorship_cid(col_name: str) -> bool: + """ + Returns True if col_name's first CID does not match any known pattern + (i.e., it is not in MODULE_CENSOR_RULES), is not in + NON_CENSORED_SECONDARY_SOURCE_CIDS, and is not an ALWAYS_INCLUDE_NON_CID + column. These columns fall through to the fail-closed NULL branch in + render_censored_column_expression, which is safe (no data leaks) but + usually signals a naming mistake, a missing MODULE_CENSOR_RULES entry, + or a column that does not belong in this table at all. + + Columns with no CID at all (e.g. "token") are NOT flagged here since + the absence of a CID is a different, already-handled case + (ALWAYS_INCLUDE_NON_CID_COLUMNS) rather than an unrecognized one. + + Args: + col_name: The column name to check. + + Returns: + bool: True if this column's first CID is unrecognized and would + be set to NULL by render_censored_column_expression's + fail-closed path. + """ + first_cid = get_first_cid(col_name) + + if first_cid is None: + # No CID present at all - handled separately by + # ALWAYS_INCLUDE_NON_CID_COLUMNS, not a "censorship pattern" issue. + return False + + if first_cid in constants.NON_CENSORED_SECONDARY_SOURCE_CIDS: + return False + + if first_cid in constants.MODULE_CENSOR_RULES: + return False + + return True + +def render_censored_column_expression( + source_alias: str, + col_name: str, + classification_alias: str = "c", + has_classification: bool = True +) -> str: + """ + Renders the SELECT expression for a single column, applying + censorship rules only when classification is actually in effect + for this destination table. + + Args: + source_alias: Alias of the table this column is selected from. + col_name: The column name to render. + classification_alias: Alias of the classification table, only + meaningful when has_classification=True. + has_classification: Whether this destination table's query + actually joined a classification table + (i.e. allowed_consent_groups was non-empty). + When False, module-censored columns are passed + through as plain SELECT expressions instead + of being wrapped in an eligibility IF(...), + since no {status_col}_eligible column exists + to reference in that case. + + Returns: + str: A SQL SELECT expression for this column, aliased to col_name. + """ + first_cid = get_first_cid(col_name) + + if first_cid is None: + if col_name in constants.ALWAYS_INCLUDE_NON_CID_COLUMNS: + return f"{source_alias}.{col_name}" + return f"CAST(NULL AS STRING) AS {col_name}" + + if first_cid in constants.NON_CENSORED_SECONDARY_SOURCE_CIDS: + return f"{source_alias}.{col_name}" + + # Known module columns: keep only if module status is "Submitted" + # and for Consent Groups 3A/3B before cutoff. + rule = constants.MODULE_CENSOR_RULES.get(first_cid) + if rule is not None: + if not has_classification: + # No classification join exists in this table's query, so + # there is no {status_col}_eligible column to reference. + # Pass the column through unfiltered rather than referencing + # a column that does not exist. + return f"{source_alias}.{col_name}" + flag_name = f"{rule['status_col']}_eligible" + return f"IF({classification_alias}.{flag_name}, {source_alias}.{col_name}, NULL) AS {col_name}" + + # Unknown first CID: fail closed + return f"CAST(NULL AS STRING) AS {col_name}" + +def build_column_rule_map(output_columns: list[str]) -> dict[str, dict]: + """ + Maps each output column to its MODULE_CENSOR_RULES entry based on the + column's first concept ID. Columns whose first CID has no entry in + MODULE_CENSOR_RULES are excluded (they are not module_censor columns). + + Args: + output_columns: List of column names from the destination table + (e.g. via utils.get_column_names). + + Returns: + dict: {column_name: rule_dict} for every column governed by a + module censorship rule. + """ + column_rule_map = {} + for col in output_columns: + first_cid = get_first_cid(col) + if first_cid is None: + continue + rule = constants.MODULE_CENSOR_RULES.get(first_cid) + if rule is not None: + column_rule_map[col] = rule + return column_rule_map + + +def _build_reason_cte_block( + status_col: str, + rule: dict, + classification_table: str, + allowed_consent_groups: list[str], + classification_alias: str = "c" +) -> str: + """ + Builds a single SELECT block reporting whether column_name was + censored for each Connect_ID and why. + + Args: + status_col: The deduped status column name (e.g. "module2_status"). + rule: The rule dict for this status_col — one value from the + {status_col: rule} mapping built by build_censorship_summary_sql's + dedup step (keys: status_col, completion_ts, cutoff). + classification_table: Fully qualified classification table. + allowed_consent_groups: Consent groups that pass output_table's + classification_filter (e.g. ["1", "2", "3A", "3B"]). + classification_alias: Alias of the classification table. + + Returns: + str: A SELECT statement returning Connect_ID, consent_group, status_col, + and reason. + """ + completion_ts = rule["completion_ts"] + cutoff = rule["cutoff"] + eligible_expr = destination_table_builder.render_eligibility_condition( + status_col, completion_ts, cutoff, alias=classification_alias + ) + + allowed_consent_groups_sql = ", ".join(f"'{c}'" for c in allowed_consent_groups) + + return f""" + SELECT + {classification_alias}.Connect_ID, + {classification_alias}.consent_group, + '{status_col}' AS status_col, + CASE + -- The eligibility decision itself is delegated to + -- render_eligibility_condition — the same expression used to + -- build {status_col}_eligible in build_destination_table_query. + -- Everything below only decides which human-readable message to + -- show once it is known the participant is NOT eligible. + WHEN {eligible_expr} IS NOT TRUE + THEN CASE + -- Reason 1: Status is not "Submitted" (NULL, empty, + -- or any other value). This alone is disqualifying + -- regardless of consent_group. + WHEN {classification_alias}.{status_col} IS DISTINCT FROM 'Submitted' + THEN CASE + WHEN {classification_alias}.{status_col} IS NULL + THEN '{status_col} = NULL' + ELSE CONCAT( + '{status_col} = "', {classification_alias}.{status_col}, '"', + IF({classification_alias}.{status_col} = '', ' (empty string)', '') + ) + END + -- Reason 2: Status was "Submitted", but for Consent Groups 3A/3B + -- the completion has to land strictly before the cutoff. + -- Missing timestamps or a completion on/after cutoff both count + -- as censored. Treat NULL and "too late" as distinct + -- sub-reasons for a clearer message. + ELSE CASE + WHEN {classification_alias}.{completion_ts} IS NULL + THEN '{completion_ts} = NULL' + WHEN {classification_alias}.{cutoff} IS NULL + THEN '{cutoff} = NULL' + ELSE CONCAT( + '{completion_ts} = ', + CAST({classification_alias}.{completion_ts} AS STRING), + ' is on or after ', '{cutoff}', ' = ', + CAST({classification_alias}.{cutoff} AS STRING) + ) + END + END + -- Eligible: Not censored for this column. + ELSE NULL + END AS reason + FROM `{classification_table}` {classification_alias} + WHERE {classification_alias}.consent_group IN ({allowed_consent_groups_sql}) + """.strip() + +def build_censorship_summary_sql( + output_table: str, + destination_table: str, + column_rule_map: dict[str, dict], + classification_table: str, + allowed_consent_groups: list[str], + classification_alias: str = "c" +) -> str: + """ + Builds the full UNION ALL query across all censorable columns. + + Args: + output_table: Fully qualified final destination table that + column_rule_map's columns were derived from. Used + only for the descriptive header comment in the + generated SQL. + destination_table: Fully qualified table to write the censorship summary for. + column_rule_map: {column_name: rule_dict}, as produced by + build_column_rule_map. + classification_table: Fully qualified table containing consent_group, status, + completion_ts, and cutoff columns for every + participant. + allowed_consent_groups: Consent groups that pass output_table's classification_filter + (e.g. ["1", "2", "3A", "3B"]). + classification_alias: Alias used for that table in the generated SQL. + + Returns: + str: A complete SQL query returning Connect_ID, consent_group, column_name, + and reason for every censorship event, filtered to reason IS NOT NULL. + + Raises: + ValueError: If column_rule_map is empty. + """ + # Multiple output columns can point at the same status_col — e.g., both + # menstrual-survey CIDs (912367929, 232438133) drive menstrual_status. + # Collapse to one rule per status_col, so _build_reason_cte_block is not + # generating identical CASE logic twice for the same underlying column. + # + # Note: Unlike build_unique_eligibility_rules (used for the *_eligible + # flags in build_destination_table_query), this dedup does NOT check for + # conflicting completion_ts/cutoff values across CIDs sharing a + # status_col; it assumes column_rule_map is already consistent since + # every entry in it traces back to the same MODULE_CENSOR_RULES that + # build_unique_eligibility_rules validates elsewhere in the pipeline. + status_col_to_rule: dict[str, dict] = {} + for rule in column_rule_map.values(): + # Values are guaranteed identical for a shared status_col; see note above + status_col_to_rule[rule["status_col"]] = rule + + # Build one reason-block per unique status_col then UNION them together + reason_blocks = [ + _build_reason_cte_block( + status_col=status_col, + rule=rule, + classification_table=classification_table, + classification_alias=classification_alias, + allowed_consent_groups=allowed_consent_groups, + ) + for status_col, rule in status_col_to_rule.items() + ] + reasons_cte = "\nUNION ALL\n".join(reason_blocks) + + # Map every output column back to its status_col as a VALUES/STRUCT + # literal, so each column (including duplicates like the two menstrual + # CIDs) joins onto its single shared reasons row below. + column_map_rows = ",\n ".join( + f"('{col}', '{rule['status_col']}')" + for col, rule in column_rule_map.items() + ) + + final_sql = f""" +/* Censorship summary query for {output_table} -> {destination_table} */ + +CREATE OR REPLACE TABLE `{destination_table}` AS ( +WITH reasons AS ( + {reasons_cte} +), +column_status_map AS ( + SELECT * FROM UNNEST([ + STRUCT + {column_map_rows} + ]) +) +SELECT + reasons.Connect_ID, + reasons.consent_group, + column_status_map.column_name, + reasons.reason +FROM reasons +JOIN column_status_map + ON reasons.status_col = column_status_map.status_col +WHERE reasons.reason IS NOT NULL +ORDER BY reasons.Connect_ID, column_status_map.column_name +) +""".strip() + + return final_sql + +def get_censorship_rollup(destination_table: str) -> dict: + """ + Queries the censorship summary table and returns two rollups: + 1. Per-column counts (how many participants were censored, per column) + 2. Per-consent-group counts (how many censorship events occurred, per consent group) + + Args: + destination_table: The table created by create_censorship_summary_table. + + Returns: + dict: {"by_column": [...], "by_consent_group": [...]} + """ + client = bigquery.Client() + + by_column_sql = f""" + SELECT column_name, COUNT(*) AS censored_count + FROM `{destination_table}` + GROUP BY column_name + ORDER BY censored_count DESC + """ + by_consent_group_sql = f""" + SELECT consent_group, COUNT(*) AS censorship_events + FROM `{destination_table}` + GROUP BY consent_group + ORDER BY consent_group + """ + + by_column = [dict(row) for row in client.query(by_column_sql).result()] + by_consent_group = [dict(row) for row in client.query(by_consent_group_sql).result()] + + return {"by_column": by_column, "by_consent_group": by_consent_group} + +def create_censorship_rollup_json( + client: storage.Client, + output_path: str, + destination_table: str, + by_column: list[dict[str, Any]], + by_consent_group: list[dict[str, Any]], +) -> None: + """ + Write a JSON report of censorship rollup counts to a GCS location. + + Args: + client (storage.Client): GCS client. + output_path (str): GCS path (gs://...) to write the report. + destination_table (str): Censorship summary table name (the table + get_censorship_rollup was queried against). + by_column (list): Per-column censored counts, as returned by + get_censorship_rollup()["by_column"]. + by_consent_group (list): Per-consent-group censorship event counts, as returned by + get_censorship_rollup()["by_consent_group"]. + """ + # Build report structure + report = { + "_metadata": { + "generated_at": datetime.now(timezone.utc).isoformat(), + "source": "pr2-transformation pipeline", + "destination_table": destination_table, + "description": "Rollup of censorship events from the censorship summary table, broken down by column and by consent group", + "structure": { + "by_column": "Per-column counts of how many participants were censored for that column, ordered by censored_count descending", + "by_consent_group": "Per-consent-group counts of how many censorship events occurred, ordered by consent_group" + } + }, + "by_column": by_column, + "by_consent_group": by_consent_group, + } + + # Parse GCS path + path = output_path.removeprefix("gs://") + bucket_name, blob_path = path.split("/", 1) + + bucket = client.bucket(bucket_name) + blob = bucket.blob(blob_path) + + # Upload JSON report + blob.upload_from_string( + json.dumps(report, indent=2, ensure_ascii=False), + content_type="application/json", + ) \ No newline at end of file diff --git a/core/classification.py b/core/classification.py new file mode 100644 index 0000000..bb4fba7 --- /dev/null +++ b/core/classification.py @@ -0,0 +1,418 @@ +"""Participant classification and cutoff calculation utilities.""" + +import pandas as pd +import numpy as np + +from google.cloud import bigquery + +def read_parquet_to_dataframe(parquet_url: str) -> pd.DataFrame: + """ + Read participant status data from a parquet file and prepare it for + classification. + + The parquet file is filtered to include only verified participants. + Timestamp fields used during participant classification are normalized + to timezone-aware UTC datetimes with invalid or missing values coerced + to NaT. + + Args: + parquet_url (str): GCS or local path to the participant status + parquet file. + + Returns: + pd.DataFrame: DataFrame containing verified participants with + normalized timestamp columns. + """ + # Only keep verified participants + df = pd.read_parquet(parquet_url, + engine="pyarrow", + filters=[("verified_status", "==", "Verified")]) + + # Normalize timestamp columns. + # Blank strings, nulls, and invalid timestamps become NaT. + ts_cols = ["consent_withdrawn_ts", "hipaa_revoked_ts", "data_destruction_ts", + "module1_complete_ts", "module2_complete_ts", "module3_complete_ts", + # Blood/Urine/Mouthwash, Blood/Urine, and Mouthwash are commented out + # because it is unclear how the follow-up specimens will be handled. + # For now, these modules with thier timestamps are excluded. + "module4_complete_ts", #"bio_complete_ts", "clinicalbio_complete_ts", "mouthwash_complete_ts", + "menstrual_complete_ts", "covid19_complete_ts", + "experience2024_complete_ts"] + for col in ts_cols: + df[col] = pd.to_datetime(df[col], errors="coerce", utc=True).astype("datetime64[us, UTC]") + + return df + +def classify_participants(df: pd.DataFrame) -> pd.DataFrame: + """ + Classify participants into consent and data-retention groups. + + Evaluates each participant's HIPAA revocation, consent withdrawn, + and data destruction status to assign one of the consent groups + (1, 2, 3A, 3B, or 4). Participants with invalid flag or timestamp + combinations are marked as data-quality exclusions or undefined. + + The function also determines the appropriate EHR and survey cutoff + timestamps along with any anomaly or exclusion reason needed for + downstream processing. + + Args: + df (pd.DataFrame): Participant status DataFrame. + + Returns: + pd.DataFrame: The input DataFrame with classification, cutoff, + anomaly, exclusion, and helper columns added. + """ + # ========================================================================= + # Boolean masks for conditions + # Each variable becomes a True/False Series across the dataframe + # ========================================================================= + + # Boolean masks for "Yes" flag values + destroy_yes = df["data_destruction_requested"].eq("Yes") + withdraw_yes = df["consent_withdrawn"].eq("Yes") + revoke_yes = df["hipaa_revoked"].eq("Yes") + + # Boolean masks for "No" flag values + destroy_no = df["data_destruction_requested"].eq("No") + withdraw_no = df["consent_withdrawn"].eq("No") + revoke_no = df["hipaa_revoked"].eq("No") + + # Boolean masks for missing timestamps + destroy_ts_missing = df["data_destruction_ts"].isna() + withdraw_ts_missing = df["consent_withdrawn_ts"].isna() + revoke_ts_missing = df["hipaa_revoked_ts"].isna() + + # ========================================================================= + # Timestamp validation for Consent Group 3 + # ========================================================================= + + # revoke_after_withdraw: True when revoke timestamp exists AND is later than withdraw timestamp + # This is considered an anomaly because EHR cutoff cannot be reliably determined + revoke_after_withdraw = ( + df["hipaa_revoked_ts"].notna() + & df["consent_withdrawn_ts"].notna() + & (df["hipaa_revoked_ts"] > df["consent_withdrawn_ts"]) + ) + + # revoke_equals_withdraw: True when revoke timestamp exists AND is equal to withdraw timestamp + # Treated as its own, separate 3A anomaly from revoke_after_withdraw. Since there is no + # ambiguity window, both events happened at the same instant, so withdraw_ts can safely be used + # for ehr_cutoff too. + revoke_equals_withdraw = ( + df["hipaa_revoked_ts"].notna() + & df["consent_withdrawn_ts"].notna() + & (df["hipaa_revoked_ts"] == df["consent_withdrawn_ts"]) + ) + + # revoke_before_withdraw: True when revoke timestamp exists AND occurs before withdraw timestamp + # This is the expected ordering (valid) Consent Group 3B scenario + revoke_before_withdraw = ( + df["hipaa_revoked_ts"].notna() + & df["consent_withdrawn_ts"].notna() + & (df["hipaa_revoked_ts"] < df["consent_withdrawn_ts"]) + ) + + # ========================================================================= + # Validation / exclusion masks + # These identify records that violate expected data rules, such as having a + # "Yes" flag but missing the corresponding timestamp. + # These participants are excluded from downstream processing and flagged + # for data quality review. + # ========================================================================= + + # Rule 1: Withdraw Consent = Yes but withdraw timestamp is NULL + # Rule 2: Revoke HIPAA = Yes but revoke timestamp is NULL AND participant did NOT withdraw consent + data_quality_mask = ( + (destroy_no & withdraw_yes & revoke_yes & withdraw_ts_missing) + | (destroy_no & withdraw_no & revoke_yes & revoke_ts_missing) + ) + # ========================================================================= + # Anomaly validation rules + # These participants are still processed downstream but are flagged because + # their data is internally inconsistent or violates expected temporal logic. + # ========================================================================= + + # Rule: Destory Data = Yes but destroy timestamp is NULL + # Participant is still classified as Consent Group 4 and excluded downstream, + # but the missing timestamp is recorded as an anomaly. + anomaly_destroy_missing_ts = destroy_yes & withdraw_yes & revoke_yes & destroy_ts_missing + + # Rule: Revoke timestamp occurs AFTER withdraw timestamp + # This creates ambiguity because EHR cutoff cannot be reliably determined. + # Participant is still processed as Consent Group 3A using withdraw timestamp for + # survey cutoff, but anomaly is recorded. + anomaly_revoke_after_withdraw = destroy_no & withdraw_yes & revoke_yes & revoke_after_withdraw + + # Rule: Revoke timestamp is EQUAL to withdraw timestamp + # A separate, distinct anomaly from anomaly_revoke_after_withdraw. The + # ehr_cutoff CAN be reliably set since revoke and withdraw happened at + # the same instant (withdraw_ts = revoke_ts, so using either is + # equivalent). Still flagged as an anomaly for visibility/review but + # processed with a real ehr_cutoff rather than leaving it blank. + anomaly_revoke_equals_withdraw = destroy_no & withdraw_yes & revoke_yes & revoke_equals_withdraw + + # ============================================================================ + # Consent group classification masks + # These masks determine which consent group each participant belongs to. + # ============================================================================ + + # ============================================================================ + # Consent Group 1 + # - Destroy Data = No + # - Withdraw Consent = No + # - Revoke HIPAA = No + # + # No restrictions apply. + # Participant and all their data can be included without any cutoffs. + # ============================================================================ + consent_group_1 = ( + destroy_no + & withdraw_no + & revoke_no + ) + + # ============================================================================ + # Consent Group 2 + # - Destroy Data = No + # - Withdraw Consent = No + # - Revoke HIPAA = Yes (timestamp present) + # + # EHR cutoff = Revoke timestamp + # Survey cutoff = None + # ============================================================================ + consent_group_2 = ( + destroy_no + & withdraw_no + & revoke_yes + & df["hipaa_revoked_ts"].notna() + ) + + # ============================================================================ + # Consent Group 3A + # - Destroy Data = No + # - Withdraw Consent = Yes (timestamp present) + # - Revoke HIPAA = Yes (timestamp is either missing, occurs strictly + # after withdraw timestamp, or occurs at exactly the same time as + # withdraw timestamp) + # + # EHR cutoff = Withdraw timestamp for normal Consent Group 3A + # EHR cutoff = NULL for the revoke-after-withdraw anomaly + # EHR cutoff = Withdraw timestamp for the revoke-equals-withdraw anomaly + # Survey cutoff = Withdraw timestamp + # ============================================================================ + consent_group_3a = ( + destroy_no + & withdraw_yes + & revoke_yes + & df["consent_withdrawn_ts"].notna() + & ( + df["hipaa_revoked_ts"].isna() + | revoke_after_withdraw + | revoke_equals_withdraw + ) + ) + + # ============================================================================ + # Consent Group 3B + # - Destroy Data = No + # - Withdraw Consent = Yes (timestamp present and occurs AFTER revoke timestamp) + # - Revoke HIPAA = Yes (timestamp present and occurs BEFORE withdraw timestamp) + # + # EHR cutoff = Revoke timestamp + # Survey cutoff = Withdraw timestamp + # ============================================================================ + consent_group_3b = ( + destroy_no + & withdraw_yes + & revoke_yes + & df["consent_withdrawn_ts"].notna() + & revoke_before_withdraw + ) + + # ============================================================================ + # Consent Group 4 + # - Destroy Data = Yes + # - Withdraw Consent = Yes + # - Revoke HIPAA = Yes + # + # The participant and all their data is excluded from downstream processing + # ============================================================================ + consent_group_4 = ( + destroy_yes + & withdraw_yes + & revoke_yes + ) + + # Build output classification column based on the above consent groups and rules + df["consent_group"] = np.select( + [consent_group_1, consent_group_2, consent_group_3a, consent_group_3b, consent_group_4, data_quality_mask], + ["1", "2", "3A", "3B", "4", "DATA_QUALITY_EXCLUSION"], + default="UNDEFINED" + ) + + # Build exclusion reason column to specify which rule was violated for + # participants classified as "DATA_QUALITY_EXCLUSION" + df["exclusion_reason"] = np.select( + [ + destroy_no & withdraw_yes & withdraw_ts_missing, + destroy_no & revoke_yes & revoke_ts_missing & withdraw_no, + ], + [ + "Withdraw Consent = 'Yes' but timestamp is NULL", + "Revoke HIPAA = 'Yes' but timestamp is NULL", + ], + default=pd.NA + ) + + # Assign reason for unmatched participants + df.loc[ + df["consent_group"] == "UNDEFINED", + "exclusion_reason" + ] = "Participant flag combination did not match any defined consent group" + + # Build anomaly column to specify which rule was violated for + # participants classified as Consent Groups 4 or 3A but have internal + # inconsistencies in their data + df["anomaly"] = np.select( + [ + anomaly_destroy_missing_ts, + anomaly_revoke_after_withdraw, + anomaly_revoke_equals_withdraw + ], + [ + "Destroy Data = 'Yes' but timestamp is NULL", + "Revoke HIPAA timestamp occurs AFTER Withdraw Consent timestamp - EHR cutoff could not be determined reliably", + "Revoke HIPAA timestamp is EQUAL to Withdraw Consent timestamp - EHR cutoff set to Withdraw Consent timestamp" + ], + default=pd.NA + ) + + # Cutoff logic + df["ehr_cutoff"] = pd.Series(pd.NaT, index=df.index, dtype="datetime64[us, UTC]") + df["survey_cutoff"] = pd.Series(pd.NaT, index=df.index, dtype="datetime64[us, UTC]") + + # Consent Group 1: No cutoffs, so leave all columns as NaT (as initialized previously). + # Participant is included downstream without restrictions. No further + # code is needed here for this consent group. + + # Consent Group 2: EHR cutoff = Revoke timestamp, Survey cutoff = None + df.loc[consent_group_2, "ehr_cutoff"] = df.loc[consent_group_2, "hipaa_revoked_ts"] + + # Consent Group 3A: EHR/Survey cutoff = Withdraw timestamp + normal_3a = consent_group_3a & df["hipaa_revoked_ts"].isna() + + df.loc[normal_3a, "ehr_cutoff"] = df.loc[normal_3a, "consent_withdrawn_ts"] + df.loc[normal_3a, "survey_cutoff"] = df.loc[normal_3a, "consent_withdrawn_ts"] + + # Consent Group 3A anomaly - revoke AFTER withdraw: ehr_cutoff stays NULL. + # If revoke_ts is later than withdraw_ts, keep as anomaly and leave + # ehr_cutoff blank. This will still be processed as Consent Group 3A. + anomalous_3a_revoke_after = consent_group_3a & revoke_after_withdraw + + df.loc[anomalous_3a_revoke_after, "survey_cutoff"] = df.loc[anomalous_3a_revoke_after, "consent_withdrawn_ts"] + + # Consent Group 3A anomaly - revoke EQUALS withdraw: ehr_cutoff CAN be reliably + # set since revoke and withdraw happened at the same instant. Use + # withdraw_ts for all cutoffs. This will be processed as Consent Group 3A. + anomalous_3a_revoke_equals = consent_group_3a & revoke_equals_withdraw + + df.loc[anomalous_3a_revoke_equals, "ehr_cutoff"] = df.loc[anomalous_3a_revoke_equals, "consent_withdrawn_ts"] + df.loc[anomalous_3a_revoke_equals, "survey_cutoff"] = df.loc[anomalous_3a_revoke_equals, "consent_withdrawn_ts"] + + # Consent Group 3B: EHR cutoff = Revoke timestamp, Survey cutoff = Withdraw timestamp + df.loc[consent_group_3b, "ehr_cutoff"] = df.loc[consent_group_3b, "hipaa_revoked_ts"] + df.loc[consent_group_3b, "survey_cutoff"] = df.loc[consent_group_3b, "consent_withdrawn_ts"] + + # Consent Group 4: No cutoffs, so leave all columns as NaT (as initialized previously). + # Participant is excluded downstream. No further code is needed here for this consent group. + + # Sort the dataframe for code development and debugging purposes. This can be removed in the future. + df = df.sort_values(by=["exclusion_reason", "anomaly", "consent_group", "Connect_ID"]) + + # Sort dataframe in order of: + # 1. Undefined + # 2. Consent groups with data quality exclusion + # 3. Consent groups with anomalies + # 4. Normal consent groups + df["sort_order"] = np.select( + [df["consent_group"].eq("UNDEFINED"), df["consent_group"].eq("DATA_QUALITY_EXCLUSION"), df["anomaly"].notna()], + [1, 2, 3], default=4 + ) + + return df + +def write_classification_to_bq( + classification_df: pd.DataFrame, + bq_table_name: str, + client: bigquery.Client, +) -> None: + """ + Write participant classification results to a BigQuery table. + + Creates or replaces the destination table using the expected schema + for participant classification output. + + Args: + classification_df (pd.DataFrame): DataFrame containing participant + classifications and derived cutoff fields. + bq_table_name (str): Fully qualified BigQuery table name + (project.dataset.table). + client (bigquery.Client): BigQuery client used to load the table. + + Returns: + None + """ + job_config = bigquery.LoadJobConfig( + schema=[ + # String columns + bigquery.SchemaField("Connect_ID", "STRING"), + bigquery.SchemaField("verified_status", "STRING"), + bigquery.SchemaField("verified_status_concept_id", "STRING"), + bigquery.SchemaField("consent_withdrawn", "STRING"), + bigquery.SchemaField("consent_withdrawn_concept_id", "STRING"), + bigquery.SchemaField("hipaa_revoked", "STRING"), + bigquery.SchemaField("hipaa_revoked_concept_id", "STRING"), + bigquery.SchemaField("data_destruction_requested", "STRING"), + bigquery.SchemaField("data_destruction_requested_concept_id", "STRING"), + + # Timestamp columns + bigquery.SchemaField("consent_withdrawn_ts", "TIMESTAMP"), + bigquery.SchemaField("hipaa_revoked_ts", "TIMESTAMP"), + bigquery.SchemaField("data_destruction_ts", "TIMESTAMP"), + bigquery.SchemaField("module1_complete_ts", "TIMESTAMP"), + bigquery.SchemaField("module2_complete_ts", "TIMESTAMP"), + bigquery.SchemaField("module3_complete_ts", "TIMESTAMP"), + bigquery.SchemaField("module4_complete_ts", "TIMESTAMP"), + # Blood/Urine/Mouthwash, Blood/Urine, and Mouthwash are commented out + # because it is unclear how the follow-up specimens will be handled. + # For now, these modules with thier timestamps are excluded. + #bigquery.SchemaField("bio_complete_ts", "TIMESTAMP"), + #bigquery.SchemaField("clinicalbio_complete_ts", "TIMESTAMP"), + #bigquery.SchemaField("mouthwash_complete_ts", "TIMESTAMP"), + bigquery.SchemaField("menstrual_complete_ts", "TIMESTAMP"), + bigquery.SchemaField("covid19_complete_ts", "TIMESTAMP"), + bigquery.SchemaField("experience2024_complete_ts", "TIMESTAMP"), + + # Classification fields + bigquery.SchemaField("consent_group", "STRING"), + bigquery.SchemaField("exclusion_reason", "STRING"), + bigquery.SchemaField("anomaly", "STRING"), + + # Derived cutoffs + bigquery.SchemaField("ehr_cutoff", "TIMESTAMP"), + bigquery.SchemaField("survey_cutoff", "TIMESTAMP"), + + # Helper field + bigquery.SchemaField("sort_order", "INTEGER"), + ], + write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE + ) + + # Load DataFrame to BigQuery + job = client.load_table_from_dataframe( + classification_df, + bq_table_name, + job_config=job_config + ) + job.result() # Wait for the job to complete \ No newline at end of file diff --git a/core/constants.py b/core/constants.py index 7ff823a..7b55943 100644 --- a/core/constants.py +++ b/core/constants.py @@ -185,3 +185,99 @@ } ] } + +# Secondary source concept IDs for each module/survey +MODULE_CENSOR_RULES = { + # Module 1: Background and Overall Health + "726699695": { + "source_table": "module1", + "status_col": "module1_status", + "completion_ts": "module1_complete_ts", + "cutoff": "survey_cutoff", + }, + # Module 2: Medications, Reproductive Health, Exercise, and Sleep + "745268907": { + "source_table": "module2", + "status_col": "module2_status", + "completion_ts": "module2_complete_ts", + "cutoff": "survey_cutoff", + }, + # Module 3: Smoking, Alcohol, and Sun Exposure + "965707586": { + "source_table": "module3", + "status_col": "module3_status", + "completion_ts": "module3_complete_ts", + "cutoff": "survey_cutoff", + }, + # Module 4: Where You Live and Work + "716117817": { + "source_table": "module4", + "status_col": "module4_status", + "completion_ts": "module4_complete_ts", + "cutoff": "survey_cutoff", + }, + # Blood/Urine/Mouthwash, Blood/Urine, and Mouthwash are commented out + # because it is unclear how the follow-up specimens will be handled. + # For now, these modules and their timestamps are excluded. + # + # Blood/Urine/Mouthwash + #"299215535": { + # "source_table": "bioSurvey", + # "status_col": "bio_status", + # "completion_ts": "bio_complete_ts", + # "cutoff": "survey_cutoff", + #}, + # Blood/Urine + #"826163434": { + # "source_table": "clinicalBioSurvey", + # "status_col": "clinicalbio_status", + # "completion_ts": "clinicalbio_complete_ts", + # "cutoff": "survey_cutoff", + #}, + # Mouthwash + #"390351864": { + # "source_table": "mouthwash", + # "status_col": "mouthwash_status", + # "completion_ts": "mouthwash_complete_ts", + # "cutoff": "survey_cutoff", + #}, + # Menstrual Cycle (contains two secondary source concept IDs) + "912367929": { + "source_table": "menstrualSurvey", + "status_col": "menstrual_status", + "completion_ts": "menstrual_complete_ts", + "cutoff": "survey_cutoff", + }, + # Menstrual Cycle (contains two secondary source concept IDs) + "232438133": { + "source_table": "menstrualSurvey", + "status_col": "menstrual_status", + "completion_ts": "menstrual_complete_ts", + "cutoff": "survey_cutoff", + }, + # COVID-19 + "793330426": { + "source_table": "covid19Survey", + "status_col": "covid19_status", + "completion_ts": "covid19_complete_ts", + "cutoff": "survey_cutoff", + }, + # 2024 Connect Experience Survey + "506648060": { + "source_table": "experience2024", + "status_col": "experience2024_status", + "completion_ts": "experience2024_complete_ts", + "cutoff": "survey_cutoff", + }, +} + +NON_CENSORED_SECONDARY_SOURCE_CIDS = { + "214456996": {"source_table": "participants"}, # Eligibility Screener + "332759827": {"source_table": "participants"}, # User Profile + "218595434": {"source_table": "participants"}, # Verification + "883203566": {"source_table": "participants"}, # Sign in + "273437590": {"source_table": "participants"}, # Consent + "104913069": {"source_table": "participants"}, # Research- Finalization and shipping +} + +ALWAYS_INCLUDE_NON_CID_COLUMNS = {} \ No newline at end of file diff --git a/core/destination_table_builder.py b/core/destination_table_builder.py new file mode 100644 index 0000000..6bbe957 --- /dev/null +++ b/core/destination_table_builder.py @@ -0,0 +1,711 @@ +"""Utilities for building destination tables from configuration.""" + +import json +from typing import Any, Optional +from datetime import datetime, timezone + +from google.cloud import bigquery, storage + +from core import constants, utils, censorship + +def load_destination_config(config_file_path: str) -> dict: + """ + Load the destination configuration JSON file. + + This configuration defines destination table mappings and related settings + used to generate SQL queries. + + Args: + config_file_path (str): Path to the destination configuration JSON file. + + Returns: + dict: Parsed configuration dictionary. Returns an empty dict if loading fails. + """ + try: + with open(config_file_path, 'r') as f: + return json.load(f) + except Exception as e: + utils.logger.error(f"Error loading destination configuration from {config_file_path}: {e}") + return {} + +def get_destination_table_config( + destination_config: dict[str, Any], + destination_dataset: str, + destination_table: str, +) -> dict[str, Any] | None: + """ + Retrieve the configuration for a specific destination table. + + Searches the configuration for a destination table matching the given dataset + and table name. + + Args: + destination_config (dict): Full destination table configuration. + destination_dataset (str): Target dataset name. + destination_table (str): Target table name. + + Returns: + dict | None: Matching destination table configuration or None if not found. + """ + # Loop through each destination table defined in the config + for destination in destination_config.get("destination_tables", []): + dataset_name = destination.get("dataset") + table_name = destination.get("table") + + # Check if both dataset and table match the requested ones + if dataset_name == destination_dataset and table_name == destination_table: + # Return the matching configuration immediately + return destination + + # If no match was found after checking all entries, return None + return None + +def get_fq_base_table( + client: bigquery.Client, + destination_table_config: dict[str, Any] +) -> str: + """ + Construct the fully qualified BigQuery base table name. + + Uses the base_table configuration and the current project to build + a fully qualified table identifier. + + Args: + client (bigquery.Client): BigQuery client with project context. + destination_table_config (dict): Destination table configuration. + + Returns: + str: Fully qualified table name in the format project.dataset.table. + + Raises: + ValueError: If base_table config is missing or malformed. + """ + project = client.project + + # Base table config must exist and be a dict + base_table_config = destination_table_config["base_table"] + + if not isinstance(base_table_config, dict): + raise ValueError( + f"'base_table' must be an object for " + f"{destination_table_config['dataset']}.{destination_table_config['table']}" + ) + + # Validate required keys exist + if "dataset" not in base_table_config or "table" not in base_table_config: + raise ValueError( + f"'base_table' must include 'dataset' and 'table' for " + f"{destination_table_config['dataset']}.{destination_table_config['table']}" + ) + + # Extract dataset and table + dataset = base_table_config["dataset"] + table = base_table_config["table"] + + # Return fully qualified table name + return f"{project}.{dataset}.{table}" + +def build_table_schemas( + client: bigquery.Client, + destination_table_config: dict[str, Any], +) -> tuple[dict[tuple[str, str], list[str]], list[dict[str, Any]]]: + """ + Fetch column schemas for all tables used in a destination configuration. + + Retrieves column names for the base table and all join tables, storing them + in a lookup dictionary for later validation and SQL generation. + + Args: + client (bigquery.Client): BigQuery client. + destination_table_config (dict): Destination table configuration. + + Returns: + tuple: + - dict[(dataset, table), list[str]]: Table schemas. + - list[dict]: Schema retrieval issues (if any). + """ + # Stores schemas keyed by (dataset, table) + table_schemas: dict[tuple[str, str], list[str]] = {} + + # Collect schema fetch failures for reporting + schema_issues: list[dict[str, Any]] = [] + + def _fetch_schema(dataset: str, table: str): + key = (dataset, table) + # Skip if already fetched (avoid duplicate queries) + if key in table_schemas: + return + + fq_table = f"{client.project}.{dataset}.{table}" + try: + # Retrieve column names from BigQuery + table_schemas[key] = utils.get_column_names(client, fq_table) + except Exception as e: + utils.logger.error(f"Error fetching schema for {fq_table}: {e}") + # Store empty schema so downstream logic does not break + table_schemas[key] = [] + # Track issue for reporting + schema_issues.append({ + "dataset": dataset, + "table": table, + "error": str(e), + }) + + # Always fetch base table schema first + base_table_config = destination_table_config["base_table"] + _fetch_schema(base_table_config["dataset"], base_table_config["table"]) + + # Fetch schemas for all join tables + for join_table in destination_table_config.get("join_tables", []): + _fetch_schema(join_table["dataset"], join_table["table"]) + + return table_schemas, schema_issues + +def find_missing_columns( + destination_table_config: dict[str, Any], + table_schemas: dict[tuple[str, str], list[str]] +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """ + Identify columns specified in config that do not exist in source tables. + + Checks both base table and join tables against fetched schemas. + + Args: + destination_table_config (dict): Destination table configuration. + table_schemas (dict): Mapping of (dataset, table) to column names. + + Returns: + tuple: + - list[dict]: Missing standard columns. + - list[dict]: Missing loop variable columns. + """ + missing_cleaned_cols = {} + missing_cleaned_loop_vars = {} + + def _check_cols(config): + dataset = config["dataset"] + table = config["table"] + schema = set(table_schemas.get((dataset, table), [])) + cols = config.get("columns", []) + + # Skip wildcard or empty (means select all or nothing specified) + if cols == "*" or not cols: + return + + if not isinstance(cols, list): + raise ValueError(f"Invalid columns for {dataset}.{table}") + + for col in cols: + # Skip valid columns + if col in schema: + continue + + key = (dataset, table, col) + + # Separate loop variables versus standard columns + if utils.is_cleaned_loop_variable(col): + missing_cleaned_loop_vars[key] = missing_cleaned_loop_vars.get(key, 0) + 1 + else: + missing_cleaned_cols[key] = missing_cleaned_cols.get(key, 0) + 1 + + # Check base table + _check_cols(destination_table_config["base_table"]) + + # Check join tables + for join_table in destination_table_config.get("join_tables", []): + _check_cols(join_table) + + def _format_output(missing_columns): + return [ + { + "dataset": dataset, + "table": table, + "source_col": source_col, + "occurrence_count": occurrence_count + } + for (dataset, table, source_col), occurrence_count in missing_columns.items() + ] + + return _format_output(missing_cleaned_cols), _format_output(missing_cleaned_loop_vars) + +def build_row_filter_sql( + filter_profiles: dict[str, Any], + destination_table_config: dict[str, Any] +) -> str: + """ + Generate a SQL WHERE clause from the destination table's filter_profile. + + Looks up the filter_profile named in destination_table_config, resolves + it against filter_profiles, and builds a WHERE clause from its column/ + value conditions. This is only one of two ways a table's rows can end + up restricted. build_destination_table_query separately adds a + join-key-not-null condition and (if allowed_consent_groups is set) a + classification filter. A table using the "no_filters" profile can + still be row-restricted by those other conditions; this function only + reports on the conditions it itself contributes. + + Args: + filter_profiles (dict): Mapping of filter profile names to filter rules. + destination_table_config (dict): Destination table configuration. + + Returns: + str: SQL WHERE clause string or an empty string if the resolved + filter_profile has no conditions. + + Raises: + ValueError: If filter_profile is missing from destination_table_config + or if the named profile is not found in filter_profiles. + """ + # Get selected filter profile + filter_profile = destination_table_config.get("filter_profile") + if not filter_profile: + raise ValueError( + f"Missing filter_profile for " + f"{destination_table_config['dataset']}.{destination_table_config['table']}" + ) + # Identify filters for the selected table + filters = filter_profiles.get(filter_profile) + + # Catch invalid filter profile names + if filters is None: + raise ValueError(f"Unknown filter_profile: {filter_profile}") + + # Handle "no_filters" case (empty list) + if not filters: + utils.logger.info( + f"[{destination_table_config['table']}] " + f"filter_profile '{filter_profile}' has no conditions; " + f"no row filter SQL generated here" + ) + return "" + + clauses = [] + + # Build SQL conditions for each filter + for f in filters: + col = f["column"] + val = f["value"] + + # Default to "=" operator unless specified + op = f.get("operator", "=") + + clause = f"{col} {op} '{val}'" + + # Append optional comment + if "comment" in f: + clause += f" -- {f['comment']}" + + clauses.append(clause) + + # Join conditions with AND + return "WHERE " + "\n AND ".join(clauses) + +def build_unique_eligibility_rules(module_censor_rules: dict[str, dict]) -> dict[str, dict]: + """ + Deduplicates MODULE_CENSOR_RULES by status_col, so that CIDs which share + the same gating condition (e.g. the two menstrual survey CIDs, which both + gate on menstrual_status/menstrual_complete_ts/survey_cutoff) collapse to + a single rule instead of being treated as independent conditions. + + Raises if two different CIDs claim the same status_col but disagree on + completion_ts or cutoff — that would mean two columns silently computing + different "eligibility" under the same flag name, which is a data + integrity risk given this logic gates consent/censorship-sensitive data. + + Args: + module_censor_rules: constants.MODULE_CENSOR_RULES + + Returns: + dict: {status_col: rule} with one entry per unique gating condition. + + Raises: + ValueError: If two CIDs share a status_col but have conflicting + completion_ts or cutoff values. + """ + unique_rules: dict[str, dict] = {} + signatures: dict[str, tuple] = {} + + for cid, rule in module_censor_rules.items(): + status_col = rule["status_col"] + sig = (rule["status_col"], rule["completion_ts"], rule["cutoff"]) + + if status_col in signatures and signatures[status_col] != sig: + raise ValueError( + f"Conflicting MODULE_CENSOR_RULES entries share status_col=" + f"{status_col!r} but disagree on completion_ts/cutoff: " + f"{signatures[status_col]} vs {sig} (conflict introduced by CID {cid})" + ) + + signatures[status_col] = sig + unique_rules[status_col] = rule + + return unique_rules + +def render_eligibility_condition( + status_col: str, + completion_ts: str, + cutoff: str, + alias: str = "c", +) -> str: + """ + The single source of truth for "is {status_col} eligible" — status is + Submitted, and for Consent Groups 3A/3B, completion happened strictly before + cutoff. Used both to build the *_eligible flags in + build_destination_table_query's CTE and to determine the + not-censored/censored branch in _build_reason_cte_block, so the real + table and the censorship summary can never compute this condition + differently from each other. + + Args: + status_col: The status column name (e.g. "module2_status"). + completion_ts: The completion timestamp column name. + cutoff: The cutoff timestamp column name. + alias: Table alias to qualify all column references with. + + Returns: + str: A parenthesized boolean SQL expression (no AS alias). + """ + return f"""( + {alias}.{status_col} = 'Submitted' + AND ( + {alias}.consent_group NOT IN ('3A', '3B') + OR ( + {alias}.{completion_ts} IS NOT NULL + AND {alias}.{cutoff} IS NOT NULL + AND {alias}.{completion_ts} < {alias}.{cutoff} + ) + ) + )""" + +def build_destination_table_query( + client: bigquery.Client, + full_config: dict[str, Any], + destination_table_config: dict[str, Any], + table_schemas: dict[tuple[str, str], list[str]], + classification_table: Optional[str] = None +) -> dict[str, Any]: + """ + Build a CREATE OR REPLACE TABLE SQL query based on destination configuration. + + Combines base table, optional filters, optional classification filter, + and join tables into a single SQL query. + + Args: + client (bigquery.Client): BigQuery client. + full_config (dict): Full configuration including filter profiles. + destination_table_config (dict): Destination table configuration. + table_schemas (dict): Table schema lookup. + classification_table (str, optional): Fully qualified classification table name + + Returns: + dict: Contains destination table name and generated SQL string. + + Raises: + ValueError: If no columns are selected or if classification_table is missing + when allowed_consent_groups is populated. + """ + project = client.project + destination_dataset = destination_table_config["dataset"] + destination_table = destination_table_config["table"] + # Default join key if not specified + join_key = destination_table_config.get("join_key", "Connect_ID") + + # Check for classification filter + classification_filter = destination_table_config.get("classification_filter") + allowed_consent_groups = classification_filter.get("allowed_consent_groups", []) if classification_filter else [] + + has_classification = bool(allowed_consent_groups) + + # Build WHERE clause from filter profile + filter_sql = build_row_filter_sql( + full_config.get("filter_profiles", {}), + destination_table_config + ) + + # Base table + base_config = destination_table_config["base_table"] + fq_base_table = get_fq_base_table(client, destination_table_config) + base_table = base_config["table"] + + # CTE name and alias + cte_name = f"filtered_{base_table}" + base_alias = base_table + classification_alias = base_alias + + # Build CTE WHERE clause + # Combines base constraint, filter profile, and classification filter + + # Add classification filter condition if present + if allowed_consent_groups: + where_conditions = [f"p.{join_key} IS NOT NULL"] # Base constraint to ensure join key is not null + consent_groups_str = ", ".join([f"'{c}'" for c in allowed_consent_groups]) + where_conditions.append(f"c.consent_group IN ({consent_groups_str})") + else: + where_conditions = [f"{join_key} IS NOT NULL"] + + # Add raw column filter conditions if present + if filter_sql: + # Remove leading WHERE from filter_sql safely + filter_body = filter_sql.replace("WHERE", "", 1).strip() + where_conditions.append(filter_body) + + final_where = "WHERE " + "\n AND ".join(where_conditions) + + # Build eligibility flag columns for the CTE + # One boolean per unique gating rule (deduped by status_col) + eligibility_flags_sql = "" + if allowed_consent_groups: + unique_rules = build_unique_eligibility_rules(constants.MODULE_CENSOR_RULES) + if unique_rules: + flag_lines = [] + for status_col, rule in unique_rules.items(): + completion_ts = rule["completion_ts"] + cutoff = rule["cutoff"] + eligible_expr = render_eligibility_condition( + status_col, completion_ts, cutoff, alias="c" + ) + flag_lines.append(f" {eligible_expr} AS {status_col}_eligible") + eligibility_flags_sql = ",\n" + ",\n".join(flag_lines) + + # Build CTE with or without classification join + if allowed_consent_groups: + if not classification_table: + raise ValueError( + "classification_table must be provided when allowed_consent_groups is populated." + ) + cte = f""" +{cte_name} AS ( + SELECT + p.*, + c.* EXCEPT({join_key}){eligibility_flags_sql} + FROM `{fq_base_table}` p + INNER JOIN `{classification_table}` c + ON p.{join_key} = c.{join_key} + {final_where} +)""".strip() + else: + cte = f""" +{cte_name} AS ( + SELECT * + FROM `{fq_base_table}` + {final_where} +)""".strip() + + select_parts = [] + join_parts = [] + + selected_column_names = set() + duplicate_columns = [] + unrecognized_cid_columns = [] + + # Always include join key first from base table + select_parts.append(f"{base_alias}.{join_key}") + selected_column_names.add(join_key) + + # Base table columns + base_cols = base_config.get("columns", []) + + # Lookup schema for validation + base_schema = set(table_schemas.get((base_config["dataset"], base_config["table"]), [])) + + if base_cols == "*": + base_cols = sorted(base_schema) + + for col in base_cols: + if col in base_schema: + if col == join_key: + continue # Skip join key since it is already included + + # If column has already been selected (e.g. from a previous table), + # track it as a duplicate and skip to avoid ambiguity + if col in selected_column_names: + duplicate_columns.append(col) + continue + + # Flag unrecognized CIDs before/alongside rendering + if censorship.is_unrecognized_censorship_cid(col): + unrecognized_cid_columns.append({ + "dataset": base_config["dataset"], + "table": base_config["table"], + "column": col + }) + + select_parts.append( + censorship.render_censored_column_expression( + source_alias=base_alias, + col_name=col, + classification_alias=classification_alias, + has_classification=has_classification + ) + ) + selected_column_names.add(col) + + # Join tables + for join_table in destination_table_config.get("join_tables", []): + dataset = join_table["dataset"] + table = join_table["table"] + + join_schema = set(table_schemas.get((dataset, table), [])) + cols = join_table.get("columns", []) + + if cols == "*": + cols = sorted(join_schema) + + if not cols: + continue + + # Add LEFT JOIN + join_parts.append( + f"LEFT JOIN `{project}.{dataset}.{table}` {table} " + f"ON {table}.{join_key} = {base_alias}.{join_key}" + ) + + # Add selected columns + for col in cols: + if col in join_schema: + if col == join_key: + continue # Skip join key since it is already included + + # If column has already been selected (e.g. from a previous table), + # track it as a duplicate and skip to avoid ambiguity + if col in selected_column_names: + duplicate_columns.append(f"{dataset}.{table}.{col}") + continue + + # Flag unrecognized CIDs for join-table columns too + if censorship.is_unrecognized_censorship_cid(col): + unrecognized_cid_columns.append({ + "dataset": dataset, + "table": table, + "column": col + }) + + select_parts.append( + censorship.render_censored_column_expression( + source_alias=table, + col_name=col, + classification_alias=classification_alias, + has_classification=has_classification + ) +) + selected_column_names.add(col) + + # Log duplicate columns (if any) + if duplicate_columns: + unique_dupes = sorted(set(duplicate_columns)) + utils.logger.warning( + f"[{destination_table}] Duplicate columns detected and skipped: " + f"{', '.join(unique_dupes)}. Using first occurrence only." + ) + + # Log unrecognized CID columns (if any) - same pattern as duplicate_columns logging + if unrecognized_cid_columns: + unrecognized_names = sorted({c["column"] for c in unrecognized_cid_columns}) + utils.logger.warning( + f"[{destination_table}] Columns with unrecognized CIDs detected " + f"(will be set to NULL by fail-closed default): {', '.join(unrecognized_names)}. " + f"Please verify these CIDs are valid and update MODULE_CENSOR_RULES if necessary." + ) + + # Ensure at least one column is selected + if not select_parts: + raise ValueError(f"No columns selected for {destination_table}") + + select_sql = ",\n ".join(select_parts) + join_sql = "\n".join(join_parts) + + sql = f""" +/* Combined transformation query to build {project}.{destination_dataset}.{destination_table} from the destination configuration. + Applies participant classification, row filters, and censorship rules if configured. */ + +CREATE OR REPLACE TABLE `{project}.{destination_dataset}.{destination_table}` AS +WITH +{cte} +SELECT + {select_sql} +FROM {cte_name} {base_alias} +{join_sql} +""".strip() + + return { + "destination_table": f"{project}.{destination_dataset}.{destination_table}", + "sql": sql, + "duplicate_columns": sorted(set(duplicate_columns)), + "unrecognized_cid_columns": unrecognized_cid_columns + } + +def create_destination_missing_columns_json( + client: storage.Client, + output_path: str, + destination_table: str, + missing_cleaned_cols: list[dict[str, Any]], + missing_cleaned_loop_vars: list[dict[str, Any]], + duplicate_columns: list[str], + unrecognized_cid_columns: list[dict[str, Any]] +) -> None: + """ + Write a JSON report of missing columns to a GCS location. + + Args: + client (storage.Client): GCS client. + output_path (str): GCS path (gs://...) to write the report. + destination_table (str): Destination table name. + missing_cleaned_cols (list): Missing standard columns. + missing_cleaned_loop_vars (list): Missing loop variable columns. + duplicate_columns (list): List of duplicate column names. + unrecognized_cid_columns (list): List of columns with unrecognized CIDs. + """ + # Build report structure + report = { + "_metadata": { + "generated_at": datetime.now(timezone.utc).isoformat(), + "source": "pr2-transformation pipeline", + "destination_table": destination_table, + "description": "Report of columns requested by the destination config that were not found in their respective source tables", + "structure": { + "missing_cleaned_cols": "Columns that were not found and are NOT cleaned loop variables — these are unexpected and should be investigated", + "missing_cleaned_loop_vars": "Columns that were not found but ARE cleaned loop variables — these may be expected if the loop variable does not exist for a given table", + "duplicate_columns": "Columns skipped due to duplicate names in SELECT", + "unrecognized_cid_columns": "Columns whose first CID matched neither MODULE_CENSOR_RULES nor NON_CENSORED_SECONDARY_SOURCE_CIDS. These were set to NULL by the fail-closed default and likely indicate a naming mistake, a missing MODULE_CENSOR_RULES entry, or a column that does not belong in this table", + "occurrence_count": "Number of times this column appeared in the config AND was missing" + } + }, + "missing_cleaned_cols": missing_cleaned_cols, + "missing_cleaned_loop_vars": missing_cleaned_loop_vars, + "duplicate_columns": duplicate_columns, + "unrecognized_cid_columns": unrecognized_cid_columns + } + + # Parse GCS path + path = output_path.removeprefix("gs://") + bucket_name, blob_path = path.split("/", 1) + + bucket = client.bucket(bucket_name) + blob = bucket.blob(blob_path) + + # Upload JSON report + blob.upload_from_string( + json.dumps(report, indent=2, ensure_ascii=False), + content_type="application/json", + ) + +def table_has_classification_filter( + destination_config: dict[str, Any], + destination_dataset: str, + destination_table: str, +) -> bool: + """ + Checks whether a given destination table's config defines a + classification_filter with non-empty allowed_consent_groups. + Mirrors the exact same allowed_consent_groups resolution used + inside build_destination_table_query, so this can never disagree + with what that function actually does. + """ + destination_table_config = get_destination_table_config( + destination_config, destination_dataset, destination_table + ) + if not destination_table_config: + return False + classification_filter = destination_table_config.get("classification_filter") + allowed_consent_groups = classification_filter.get("allowed_consent_groups", []) if classification_filter else [] + return bool(allowed_consent_groups) \ No newline at end of file diff --git a/core/endpoints.py b/core/endpoints.py index 7a4135b..6228a49 100644 --- a/core/endpoints.py +++ b/core/endpoints.py @@ -4,7 +4,7 @@ from flask import Flask, jsonify, request # type: ignore -from core import constants, transformations, utils, request_helpers +from core import constants, transformations, utils, request_helpers, destination_table_builder app = Flask(__name__) @@ -112,4 +112,133 @@ def create_standardized_mapping_table(): 'message': str(e) }), 500 +@app.route('/create_classification_table', methods=['POST']) +def create_classification_table(): + mapping: dict[str, any] = request.get_json() or {} + parquet_url = mapping.get("parquet_url") + classification_table = mapping.get("classification_table") + try: + utils.logger.info(f"create_classification_table endpoint called. Building {classification_table} from {parquet_url}.") + status = transformations.create_classification_table( + parquet_url=parquet_url, + classification_table=classification_table + ) + return jsonify({ + 'status': status, + 'timestamp': datetime.utcnow().isoformat(), + 'service': constants.SERVICE_NAME + }), 200 + except Exception as e: + utils.logger.exception("An error occurred in create_classification_table endpoint.") + return jsonify({'error': 'Internal Server Error', 'message': str(e)}), 500 + + +@app.route('/create_destination_table', methods=['POST']) +def create_destination_table(): + mapping: dict[str, any] = request.get_json() or {} + config_path = mapping.get("config_path") + destination_dataset = mapping.get("destination_dataset") + destination_table = mapping.get("destination_table") + classification_table = mapping.get("classification_table") + + try: + utils.logger.info(f"create_destination_table endpoint called. Creating {destination_table} in {destination_dataset}.") + status = transformations.create_destination_table( + config_path=config_path, + destination_dataset=destination_dataset, + destination_table=destination_table, + classification_table=classification_table + ) + return jsonify({ + 'status': status, + 'timestamp': datetime.utcnow().isoformat(), + 'service': constants.SERVICE_NAME + }), 200 + except Exception as e: + utils.logger.exception("An error occurred in create_destination_table endpoint.") + return jsonify({'error': 'Internal Server Error', 'message': str(e)}), 500 + +@app.route('/check_has_classification_filter', methods=['POST']) +def check_has_classification_filter(): + mapping: dict[str, any] = request.get_json() or {} + config_path = mapping.get("config_path") + destination_dataset = mapping.get("destination_dataset") + destination_table = mapping.get("destination_table") + + try: + destination_config = destination_table_builder.load_destination_config(config_path) + has_filter = destination_table_builder.table_has_classification_filter( + destination_config, destination_dataset, destination_table + ) + + if has_filter: + utils.logger.info( + f"[{destination_table}] This table has classification_filter configured" + ) + else: + utils.logger.info( + f"[{destination_table}] No classification_filter configured; " + f"censorship summary will be skipped for this table" + ) + + return jsonify({ + 'has_classification_filter': has_filter, + 'timestamp': datetime.utcnow().isoformat(), + 'service': constants.SERVICE_NAME + }), 200 + except Exception as e: + utils.logger.exception("An error occurred in check_has_classification_filter endpoint.") + return jsonify({'error': 'Internal Server Error', 'message': str(e)}), 500 + +@app.route('/create_censorship_summary_table', methods=['POST']) +def create_censorship_summary_table(): + mapping: dict[str, any] = request.get_json() or {} + output_table = mapping.get("output_table") + classification_table = mapping.get("classification_table") + destination_table = mapping.get("destination_table") + config_path = mapping.get("config_path") + destination_dataset = mapping.get("destination_dataset") + censorship_table = mapping.get("censorship_table") + + try: + utils.logger.info(f"create_censorship_summary_table endpoint called. Creating {censorship_table}.") + destination_config = destination_table_builder.load_destination_config(config_path) + destination_table_config = destination_table_builder.get_destination_table_config( + destination_config, + destination_dataset, + censorship_table + ) + if not destination_table_config: + raise ValueError( + f"No config found for {destination_dataset}.{censorship_table}" + ) + allowed_consent_groups = destination_table_config.get("classification_filter", {}).get("allowed_consent_groups") + if not allowed_consent_groups: + msg = ( + f"[{censorship_table}] Skipped: no classification_filter.allowed_consent_groups " + f"configured. This table is not classification-gated, so no " + f"censorship summary applies" + ) + utils.logger.info(msg) + return jsonify({ + 'status': msg, + 'skipped': True, + 'timestamp': datetime.utcnow().isoformat(), + 'service': constants.SERVICE_NAME + }), 200 + + status = transformations.create_censorship_summary_table( + output_table=output_table, + classification_table=classification_table, + destination_table=destination_table, + allowed_consent_groups=allowed_consent_groups, + ) + return jsonify({ + 'status': status, + 'timestamp': datetime.utcnow().isoformat(), + 'service': constants.SERVICE_NAME + }), 200 + except Exception as e: + utils.logger.exception("An error occurred in create_censorship_summary_table endpoint.") + return jsonify({'error': 'Internal Server Error', 'message': str(e)}), 500 \ No newline at end of file diff --git a/core/generate_synthetic_bq_table.py b/core/generate_synthetic_bq_table.py new file mode 100644 index 0000000..3ecff52 --- /dev/null +++ b/core/generate_synthetic_bq_table.py @@ -0,0 +1,216 @@ +''' +Generates a synthetic "mvp" base table to be used in testing. This synthetic +table is designed to be 1:1 with the participant_status_synthetic.parquet +generated by the generate_synthetic_parquet.py. + +Strategy: + - Every column gets a distinct filler value specific to the Connect_ID + and column name. + - A handful of Connect_IDs get deliberate NULLs sprinkled into specific + columns to test that "originally NULL" and "censored down to NULL" + are distinguishable and that pass-through columns correctly preserve + NULLs rather than accidentally filling them. + - Column identity (which CID governs which column) is derived from the + constants.MODULE_CENSOR_RULES, constants.NON_CENSORED_SECONDARY_SOURCE_CIDS, + and the known "uncovered" CIDs (130371375, 111111111), so this file stays + correct if constants.py changes. +''' + +import os +import sys + +if __name__ == "__main__": + sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import pandas as pd +from core import constants, utils + +MVP_COLUMNS = [ + "d_130371375_d_303552867_d_297462035", # uncovered CID (unknown), no censoring rule + "d_111111111_d_222222222", # uncovered CID (unknown), no censoring rule + "d_883203566_d_253532712", # non-censored secondary source CID + "d_214456996_d_142654897_d_196856782", # non-censored secondary source CID + "d_273437590_d_811353546", # non-censored secondary source CID + "token", + "d_726699695_d_783167257", + "d_726699695_d_384191091_d_706998638", + "d_726699695_d_384191091_d_458435048", + "d_745268907_d_536735468", + "d_745268907_d_384191091", + "d_745268907_d_745268907", + "d_965707586_d_976570371", + "d_965707586_d_947205597_d_712653855", + "d_965707586_d_763164658", + "d_716117817_d_663265240", + "d_716117817_d_111111111", + "d_716117817_d_222222222", + "d_912367929_d_111111111", + "d_912367929_d_222222222", + "d_912367929_d_333333333", + "d_232438133_d_111111111", + "d_232438133_d_222222222", + "d_232438133_d_333333333", + "d_793330426_d_111111111", + "d_793330426_d_222222222", + "d_793330426_d_333333333", + "d_506648060_d_111111111", + "d_506648060_d_222222222", + "d_506648060_d_333333333" +] + +# Connect_IDs from generate_synthetic_parquet.py -- reused here so +# the base table's Connect_IDs line up 1:1 with the classification table. +TEST_CONNECT_IDS = [ + "TEST_UNVERIFIED", + "TEST_CASE1_ALL_SUBMITTED", + "TEST_CASE1_MOD1_NOT_SUBMITTED", + "TEST_CASE1_MOD_STATUS_NULL", + "TEST_CASE1_MOD_MENSTRUAL_NOT_SUBMITTED", + "TEST_CASE2_BASIC", + "TEST_CASE2_MOD2_IN_PROGRESS", + "TEST_CASE2_MOD3_STATUS_NULL", + "TEST_CASE3A_MOD1_BEFORE_CUTOFF", + "TEST_CASE3A_MOD1_AFTER_CUTOFF", + "TEST_CASE3A_MOD1_EXACT_CUTOFF", + "TEST_CASE3A_ANOMALY_REVOKE_AFTER_WITHDRAW", + "TEST_CASE3A_SUBMITTED_BUT_NO_TS", + "TEST_CASE3A_ALL_FOUR_MODULES_MIXED", + "TEST_CASE3B_MOD2_BEFORE_CUTOFF", + "TEST_CASE3B_MOD2_AFTER_CUTOFF", + "TEST_CASE4_BASIC", + "TEST_CASE4_ANOMALY_MISSING_DESTROY_TS", + "TEST_EXCLUSION_WITHDRAW_NO_TS", + "TEST_EXCLUSION_REVOKE_NO_TS", + "TEST_UNDEFINED_DESTROY_ONLY", + "TEST_BLANK_MODULE_STATUS", + "TEST_UNKNOWN_MODULE_STATUS", + "TEST_UNKNOWN_STATUS_CASE3A", + "TEST_REVOKE_EQUALS_WITHDRAW", +] + +# Connect_IDs that get deliberate NULLs sprinkled in. +# These are chosen to hit each category: a module_censor column, a +# non_censored column, and an uncovered-CID column, so "originally NULL" +# behaves correctly no matter which branch of render_subset_expression +# handles that column. +NULL_INJECTION_PLAN = { + "TEST_CASE1_ALL_SUBMITTED": [ + "d_726699695_d_783167257", # module_censor (module1) -- originally NULL, should STAY NULL (case 1, no cutoff issue, but source was empty) + "d_883203566_d_253532712", # non_censored -- originally NULL, should pass through as NULL + ], + "TEST_CASE2_BASIC": [ + "d_130371375_d_303552867_d_297462035", # uncovered CID -- irrelevant, always NULL regardless of source value + "d_965707586_d_976570371", # module_censor (module3) -- originally NULL + ], + "TEST_CASE3A_MOD1_BEFORE_CUTOFF": [ + "d_726699695_d_783167257", # module_censor (module1), before cutoff -- originally NULL, should STAY NULL (nothing to censor, but also nothing to show) + ] +} + +def build_mvp_df() -> pd.DataFrame: + """ + Returns a DataFrame matching the real "mvp" table schema, with every + Connect_ID from TEST_CONNECT_IDS present, every column filled with a + UNIQUE, traceable value by default, and deliberate NULLs injected per + NULL_INJECTION_PLAN. + + Each cell holds "{Connect_ID}_{column_name}" as a value, so misplaced + or leaked values are traceable on sight — no lookup table needed. + """ + rows = [] + for connect_id in TEST_CONNECT_IDS: + row = {"Connect_ID": connect_id} + for col in MVP_COLUMNS: + if col == "token": + row[col] = f"TOKEN_{connect_id}" + else: + row[col] = f"{connect_id}_{col}" + rows.append(row) + + df = pd.DataFrame(rows) + + # Apply deliberate NULL injections + for connect_id, cols_to_null in NULL_INJECTION_PLAN.items(): + for col in cols_to_null: + df.loc[df["Connect_ID"] == connect_id, col] = None + + return df + + +def classify_columns_by_rule() -> dict: + """ + Diagnostic helper: cross-references MVP_COLUMNS against constants.py + to show which category each column falls into. Useful for confirming + NULL_INJECTION_PLAN actually covers all three categories and for + catching any column whose first CID is not recognized anywhere. + """ + categories = { + "module_censor": [], + "non_censored": [], + "uncovered": [], + "no_cid": [], + } + + for col in MVP_COLUMNS: + first_cid = utils.get_first_cid(col) + if first_cid is None: + categories["no_cid"].append(col) + elif first_cid in constants.MODULE_CENSOR_RULES: + categories["module_censor"].append(col) + elif first_cid in constants.NON_CENSORED_SECONDARY_SOURCE_CIDS: + categories["non_censored"].append(col) + else: + categories["uncovered"].append(col) + + return categories + + +def write_mvp_to_bq(mvp_df: pd.DataFrame, bq_table_name: str, client) -> None: + """ + Writes the synthetic mvp DataFrame to a BigQuery table. + + All data columns are loaded as STRING, matching the real mvp table's + schema (these are flattened survey response columns, not typed data). + Uses WRITE_TRUNCATE so reruns overwrite cleanly rather than appending + duplicate test rows. + + Args: + mvp_df: DataFrame from build_mvp_df(). + bq_table_name: Fully qualified destination table + (e.g. "project.dataset.mvp_synthetic"). + client: bigquery.Client instance. + """ + schema = [bigquery.SchemaField("Connect_ID", "STRING")] + schema += [ + bigquery.SchemaField(col, "STRING") + for col in MVP_COLUMNS + ] + + job_config = bigquery.LoadJobConfig( + schema=schema, + write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE, + ) + + job = client.load_table_from_dataframe(mvp_df, bq_table_name, job_config=job_config) + job.result() # Wait for the job to complete + utils.logger.info(f"Synthetic data written to {bq_table_name} successfully.") + + +if __name__ == "__main__": + from google.cloud import bigquery + + df = build_mvp_df() + print(df.to_string(index=False)) + print(f"\nGenerated {len(df)} synthetic mvp rows across {len(MVP_COLUMNS)} columns.") + + categories = classify_columns_by_rule() + print("\nColumn classification:") + for category, cols in categories.items(): + print(f" {category}: {len(cols)} columns") + if category == "uncovered": + print(f" -> {cols}") + + # Write to BigQuery + client = bigquery.Client() + destination_table = "nih-nci-dceg-connect-stg-5519.pr2_mvp.mvp_synthetic" + write_mvp_to_bq(df, destination_table, client) \ No newline at end of file diff --git a/core/generate_synthetic_parquet.py b/core/generate_synthetic_parquet.py new file mode 100644 index 0000000..3bbcd43 --- /dev/null +++ b/core/generate_synthetic_parquet.py @@ -0,0 +1,342 @@ +''' +Generates a synthetic participant_status.parquet file matching the real +schema (Connect_ID, verified_status, consent/revoke/destroy flags + ts, +module1-4 + menstrual/covid19/experience2024 status + ts), with one row +per test case / edge case, for feeding into read_parquet_to_dataframe() +and classify_participants() end-to-end. + +''' + +import pandas as pd + +# Placeholder concept IDs -- arbitrary, only used to populate the +# *_status_concept_id companion columns so the schema shape matches. +# "UNKNOWN" and None are explicitly mapped to None: an UNKNOWN status +# means no concept ID was ever assigned, so its companion column should +# be NULL, same as an outright missing status. +STATUS_CID_MAP = { + "Submitted": "231311385", + "Started": "615768760", + "Not Started": "972455046", + "In Progress": "615768760", + "UNKNOWN": None, + None: None, +} +YES_CID = "353358909" +NO_CID = "104430631" + + +def _flag_cid(value): + if value == "Yes": + return YES_CID + if value == "No": + return NO_CID + return None + + +def _status_cid(value): + # Any status not explicitly mapped (including "UNKNOWN") returns None, + # matching real data where UNKNOWN/unrecognized statuses have no CID. + return STATUS_CID_MAP.get(value, None) + + +def build_participant_status_df() -> pd.DataFrame: + """ + Returns a DataFrame of synthetic participant_status rows, one per + test case / edge case, matching the real parquet schema. + + NOTE: verified_status is always "Verified" for every row except one + deliberate "unverified" row, since read_parquet_to_dataframe() filters + to verified_status == "Verified" at read time -- rows with any other + value should disappear before classify_participants() ever sees them. + """ + rows = [] + + def add( + connect_id, test_label, + verified_status="Verified", + destroy=None, withdraw=None, revoke=None, + destroy_ts=None, withdraw_ts=None, revoke_ts=None, + module1_status="Submitted", module1_ts=None, + module2_status="Submitted", module2_ts=None, + module3_status="Submitted", module3_ts=None, + module4_status="Submitted", module4_ts=None, + menstrual_status="Submitted", menstrual_ts=None, + covid19_status="Submitted", covid19_ts=None, + experience2024_status="Submitted", experience2024_ts=None, + ): + # Default flags to "No" unless explicitly overridden (mirrors + # realistic participant_status data -- flags are rarely NULL + # except in the deliberate UNDEFINED edge case below). + destroy = destroy if destroy is not None else "No" + withdraw = withdraw if withdraw is not None else "No" + revoke = revoke if revoke is not None else "No" + + rows.append({ + "Connect_ID": connect_id, + "test_label": test_label, # extra column for reference; drop before real use if needed + "verified_status": verified_status, + "verified_status_concept_id": "197316935" if verified_status == "Verified" else "875007964", + "consent_withdrawn": withdraw, + "consent_withdrawn_concept_id": _flag_cid(withdraw), + "hipaa_revoked": revoke, + "hipaa_revoked_concept_id": _flag_cid(revoke), + "data_destruction_requested": destroy, + "data_destruction_requested_concept_id": _flag_cid(destroy), + "module1_status": module1_status, + "module1_status_concept_id": _status_cid(module1_status), + "module2_status": module2_status, + "module2_status_concept_id": _status_cid(module2_status), + "module3_status": module3_status, + "module3_status_concept_id": _status_cid(module3_status), + "module4_status": module4_status, + "module4_status_concept_id": _status_cid(module4_status), + "menstrual_status": menstrual_status, + "menstrual_status_concept_id": _status_cid(menstrual_status), + "covid19_status": covid19_status, + "covid19_status_concept_id": _status_cid(covid19_status), + "experience2024_status": experience2024_status, + "experience2024_status_concept_id": _status_cid(experience2024_status), + "consent_withdrawn_ts": withdraw_ts, + "hipaa_revoked_ts": revoke_ts, + "data_destruction_ts": destroy_ts, + "module1_complete_ts": module1_ts, + "module2_complete_ts": module2_ts, + "module3_complete_ts": module3_ts, + "module4_complete_ts": module4_ts, + "menstrual_complete_ts": menstrual_ts, + "covid19_complete_ts": covid19_ts, + "experience2024_complete_ts": experience2024_ts, + }) + + # ------------------------------------------------------------------ + # UNVERIFIED: should be filtered out entirely by + # read_parquet_to_dataframe()'s pyarrow filter, before classification + # ever runs. If this Connect_ID shows up downstream, the filter broke. + # ------------------------------------------------------------------ + add( + "TEST_UNVERIFIED", "not verified -- must be dropped at parquet read time", + verified_status="Unverified", + ) + + # ------------------------------------------------------------------ + # CASE 1: No restrictions. + # ------------------------------------------------------------------ + add("TEST_CASE1_ALL_SUBMITTED", "case 1, everything submitted") + + add( + "TEST_CASE1_MOD1_NOT_SUBMITTED", "case 1, module1 not submitted", + module1_status="Not Started", + ) + + add( + "TEST_CASE1_MOD_STATUS_NULL", "case 1, module2 status is NULL", + module2_status=None, + ) + + add( + "TEST_CASE1_MOD_MENSTRUAL_NOT_SUBMITTED", "case 1, all modules submitted but menstrual", + menstrual_status="Not Started", + module1_ts="2025-03-04 09:18:42+00:00", + module2_ts="2025-03-11 14:37:05+00:00", + module3_ts="2025-03-19 16:52:18+00:00", + module4_ts="2025-03-28 11:09:56+00:00", + covid19_ts="2025-04-06 08:43:21+00:00", + experience2024_ts="2025-04-17 19:26:47+00:00", + ) + + # ------------------------------------------------------------------ + # CASE 2: Destroy=No, Withdraw=No, Revoke=Yes (ts present). + # ------------------------------------------------------------------ + add( + "TEST_CASE2_BASIC", "case 2, revoke present, all modules submitted", + revoke="Yes", revoke_ts="2025-03-01 00:00:00+00:00", + ) + + add( + "TEST_CASE2_MOD2_IN_PROGRESS", "case 2, module2 status = In Progress", + revoke="Yes", revoke_ts="2025-03-01 00:00:00+00:00", + module2_status="In Progress", + ) + + add( + "TEST_CASE2_MOD3_STATUS_NULL", "case 2, module3 status is NULL", + revoke="Yes", revoke_ts="2025-03-01 00:00:00+00:00", + module3_status=None, + ) + + # ------------------------------------------------------------------ + # CASE 3A: Withdraw=Yes (ts present), Revoke=Yes, + # revoke_ts missing OR occurs AFTER withdraw_ts. + # ------------------------------------------------------------------ + add( + "TEST_CASE3A_MOD1_BEFORE_CUTOFF", "case 3A, module1 completed before cutoff", + withdraw="Yes", withdraw_ts="2025-04-01 00:00:00+00:00", + revoke="Yes", revoke_ts=None, + module1_ts="2025-03-15 00:00:00+00:00", + ) + + add( + "TEST_CASE3A_MOD1_AFTER_CUTOFF", "case 3A, module1 completed after cutoff", + withdraw="Yes", withdraw_ts="2025-06-28 00:00:00+00:00", + revoke="Yes", revoke_ts=None, + module1_ts="2025-06-28 00:00:01+00:00", + ) + + add( + "TEST_CASE3A_MOD1_EXACT_CUTOFF", "case 3A, module1 completed exactly at cutoff (boundary test)", + withdraw="Yes", withdraw_ts="2025-12-31 23:59:59+00:00", + revoke="Yes", revoke_ts=None, + module1_ts="2025-12-31 23:59:59+00:00", + ) + + add( + "TEST_CASE3A_ANOMALY_REVOKE_AFTER_WITHDRAW", + "case 3A anomaly, revoke_ts after withdraw_ts -- ehr_cutoff should be NULL", + withdraw="Yes", withdraw_ts="2025-04-01 00:00:00+00:00", + revoke="Yes", revoke_ts="2025-05-01 00:00:00+00:00", + module1_ts="2025-03-15 00:00:00+00:00", + ) + + add( + "TEST_CASE3A_SUBMITTED_BUT_NO_TS", + "case 3A, module1 status Submitted but completion_ts NULL (data quality issue)", + withdraw="Yes", withdraw_ts="2025-04-01 00:00:00+00:00", + revoke="Yes", revoke_ts=None, + module1_status="Submitted", module1_ts=None, + ) + + add( + "TEST_CASE3A_ALL_FOUR_MODULES_MIXED", + "case 3A, all 4 modules -- 2 before cutoff, 2 after, to check independence", + withdraw="Yes", withdraw_ts="2025-04-01 00:00:00+00:00", + revoke="Yes", revoke_ts=None, + module1_ts="2025-03-01 00:00:00+00:00", # before -> not censored + module2_ts="2025-05-01 00:00:00+00:00", # after -> censored + module3_ts="2025-03-20 00:00:00+00:00", # before -> not censored + module4_ts="2025-06-01 00:00:00+00:00", # after -> censored + ) + + + # ------------------------------------------------------------------ + # CASE 3B: Withdraw=Yes (ts present), Revoke=Yes (ts present), + # revoke_ts occurs BEFORE withdraw_ts. + # ------------------------------------------------------------------ + add( + "TEST_CASE3B_MOD2_BEFORE_CUTOFF", "case 3B, module2 completed before withdraw cutoff", + withdraw="Yes", withdraw_ts="2025-05-01 00:00:00+00:00", + revoke="Yes", revoke_ts="2025-03-01 00:00:00+00:00", + module2_ts="2025-04-01 00:00:00+00:00", + ) + + add( + "TEST_CASE3B_MOD2_AFTER_CUTOFF", "case 3B, module2 completed after withdraw cutoff", + withdraw="Yes", withdraw_ts="2025-05-01 00:00:00+00:00", + revoke="Yes", revoke_ts="2025-03-01 00:00:00+00:00", + module2_ts="2025-06-01 00:00:00+00:00", + ) + + # ------------------------------------------------------------------ + # CASE 4: Destroy=Yes, Withdraw=Yes, Revoke=Yes -- fully excluded. + # ------------------------------------------------------------------ + add( + "TEST_CASE4_BASIC", "case 4, full destruction request", + destroy="Yes", withdraw="Yes", revoke="Yes", + destroy_ts="2025-06-01 00:00:00+00:00", + withdraw_ts="2025-03-01 00:00:00+00:00", + revoke_ts="2025-03-01 00:00:00+00:00", + ) + + add( + "TEST_CASE4_ANOMALY_MISSING_DESTROY_TS", "case 4, destroy=Yes but destroy_ts NULL", + destroy="Yes", withdraw="Yes", revoke="Yes", + destroy_ts=None, + withdraw_ts="2025-03-01 00:00:00+00:00", + revoke_ts="2025-03-01 00:00:00+00:00", + ) + + # ------------------------------------------------------------------ + # EXCLUSION: data quality violations + # ------------------------------------------------------------------ + add( + "TEST_EXCLUSION_WITHDRAW_NO_TS", "withdraw=Yes but withdraw_ts NULL", + withdraw="Yes", revoke="Yes", withdraw_ts=None, + ) + + add( + "TEST_EXCLUSION_REVOKE_NO_TS", "revoke=Yes but revoke_ts NULL, withdraw=No", + revoke="Yes", revoke_ts=None, withdraw="No", + ) + + # ------------------------------------------------------------------ + # UNDEFINED: flag combination matches nothing -- destroy=Yes but + # withdraw/revoke=No, which matches no defined case and is not an + # exclusion rule either) + # ------------------------------------------------------------------ + add( + "TEST_UNDEFINED_DESTROY_ONLY", + "destroy=Yes but withdraw=No, revoke=No -- matches no defined case", + destroy="Yes", withdraw="No", revoke="No", + ) + + # ------------------------------------------------------------------ + # BLANK STRING edge case -- empty string rather than NULL + # ------------------------------------------------------------------ + add( + "TEST_BLANK_MODULE_STATUS", "case 1, module4 status is blank string", + module4_status="", + ) + + # ------------------------------------------------------------------ + # UNKNOWN status edge case + # (e.g. module1_status = "UNKNOWN"). Since "Submitted" is the only + # value MODULE_CENSOR_RULES treats as passing, UNKNOWN should be + # treated the same as any other non-Submitted value: censored. + # Its companion concept_id column should be NULL (see STATUS_CID_MAP). + # ------------------------------------------------------------------ + add( + "TEST_UNKNOWN_MODULE_STATUS", "case 1, module3 status is UNKNOWN -- should be censored, CID NULL", + module3_status="UNKNOWN", + ) + + add( + "TEST_UNKNOWN_STATUS_CASE3A", + "case 3A, module1 status UNKNOWN -- censored regardless of cutoff since not Submitted", + withdraw="Yes", withdraw_ts="2025-04-01 00:00:00+00:00", + revoke="Yes", revoke_ts=None, + module1_status="UNKNOWN", module1_ts="2025-03-01 00:00:00+00:00", # ts before cutoff, but status is not Submitted + module2_status="Submitted", module2_ts="2025-03-01 00:00:00+00:00", # ts before cutoff, status Submitted + ) + + # ------------------------------------------------------------------ + # Case 3A anomaly: revoke_ts and withdraw_ts are equal. + # + # Resolves to case 3A, flagged via anomaly_revoke_equals_withdraw. + # Unlike the revoke-after-withdraw anomaly (where ehr_cutoff is left + # NULL because the true cutoff cannot be determined), the ehr_cutoff + # is set to withdraw_ts since revoke and withdraw happened at + # the same instant, so there is no ambiguity window between them. + # ------------------------------------------------------------------ + add( + "TEST_REVOKE_EQUALS_WITHDRAW", + "case 3A anomaly, revoke_ts == withdraw_ts -- ehr_cutoff set to withdraw_ts)", + withdraw="Yes", withdraw_ts="2025-04-01 22:22:22+00:00", + revoke="Yes", revoke_ts="2025-04-01 22:22:22+00:00", + module2_ts="2025-04-01 22:22:22+00:00", # same -> censored + module3_ts="2025-04-01 22:22:23+00:00", # after -> censored + covid19_ts="2025-03-30 08:12:15+00:00", # before -> not censored + experience2024_ts="2025-04-01 22:22:21+00:00", # before -> not censored + ) + + df = pd.DataFrame(rows) + + return df + +if __name__ == "__main__": + df = build_participant_status_df() + print(df[["Connect_ID", "test_label", "verified_status", + "data_destruction_requested", "consent_withdrawn", "hipaa_revoked"]].to_string(index=False)) + print(f"\nGenerated {len(df)} synthetic participant_status rows.") + output_path = "/Users/trivittge/Desktop/participant_status_synthetic.parquet" + df.to_parquet(output_path, engine="pyarrow", index=False) + print(f"Written to {output_path}") diff --git a/core/transformations.py b/core/transformations.py index 955d5a4..425af57 100644 --- a/core/transformations.py +++ b/core/transformations.py @@ -3,6 +3,7 @@ import re import os import sys +from typing import Optional from flask import json from google.cloud import bigquery from google.cloud import storage @@ -11,7 +12,7 @@ if __name__ == "__main__": sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from core import constants, utils, transform_renderer +from core import constants, utils, transform_renderer, classification, destination_table_builder, censorship ######################################################################## ############# Table-level Transformations ############################# @@ -922,10 +923,367 @@ def create_standardized_mapping_table( utils.logger.error(f"Exception args: {e.args}") raise e +######################################################################## +############# Subset Table Construction ############################### +######################################################################## +def create_classification_table( + parquet_url: str, + classification_table: str, + local_csv_path: Optional[str] = None, +) -> dict: + """ + Reads participant status data from parquet, classifies each participant + into a consent group (1, 2, 3A, 3B, 4, or an exclusion/anomaly), and + writes the result to a BigQuery classification table. + + This function: + 1. Reads and filters the parquet file to verified participants + 2. Classifies participants into consent groups based on + consent/revoke/destroy flags + 3. Optionally writes a local CSV for inspection/debugging + 4. Writes the classification DataFrame to a BigQuery table + + Args: + parquet_url: GCS path to the participant_status parquet file. + classification_table: Fully qualified BigQuery table to write the + classification results to. + local_csv_path: Optional local file path to save the classification + DataFrame as CSV for inspection. + + Returns: + dict: Contains "status" and "classification_table". + + Raises: + Exception: Propagates any errors encountered during read, classification, + CSV write, or BigQuery write. + """ + client = bigquery.Client() + _, _, table_short_name = utils.parse_fq_table(classification_table) + + # Read the parquet file into a DataFrame + try: + utils.logger.info(f"[{table_short_name}] Reading parquet from {parquet_url}") + classification_df = classification.read_parquet_to_dataframe(parquet_url) + utils.logger.info(f"[{table_short_name}] Read {len(classification_df)} verified participant rows") + except Exception as e: + utils.logger.exception(f"[{table_short_name}] Error reading parquet from {parquet_url}: {e}") + raise e + + # Classify participants into consent groups + try: + utils.logger.info(f"[{table_short_name}] Classifying participants...") + classification_df = classification.classify_participants(classification_df) + utils.logger.info(f"[{table_short_name}] Classification complete") + except Exception as e: + utils.logger.exception(f"[{table_short_name}] Error classifying participants: {e}") + raise e + + # Optionally write the classification DataFrame to a local CSV for inspection + if local_csv_path: + try: + classification_df.to_csv(local_csv_path, index=False) + utils.logger.info(f"[{table_short_name}] Local CSV saved to {local_csv_path}") + except Exception as e: + utils.logger.exception(f"[{table_short_name}] Error saving local CSV to {local_csv_path}: {e}") + raise e + + # Write the classification DataFrame to BigQuery + try: + utils.logger.info(f"[{table_short_name}] Writing classification to BigQuery...") + classification.write_classification_to_bq(classification_df, classification_table, client) + status = f"[{table_short_name}] Classification table successfully created at {classification_table}" + utils.logger.info(status) + except Exception as e: + utils.logger.exception(f"[{table_short_name}] Error writing classification to BigQuery: {e}") + raise e + + return { + "status": status, + "classification_table": classification_table, + } + +def create_destination_table( + config_path: str, + destination_dataset: str, + destination_table: str, + missing_report_base_path: str = constants.MISSING_COLUMNS_REPORT_PATH, + classification_table: Optional[str] = None +) -> dict: + """ + Create the destination table based on configuration. + + This function: + 1. Loads the destination table configuration + 2. Retrieves the target destination table config + 3. Builds table schemas + 4. Identifies missing columns + 5. Generates SQL + 6. Saves SQL to GCS + 7. Executes query + 8. Writes missing column report JSON + + Args: + config_path (str): Path to destination config JSON. + destination_dataset (str): Target dataset. + destination_table (str): Target table. + missing_report_base_path (str): Base GCS path for missing column reports. + classification_table (Optional[str]): Fully qualified BigQuery classification table name. + + Returns: + dict: + - status + - submitted_sql_path + - missing_columns_report_path + + Raises: + ValueError: If configuration is invalid. + Exception: If query execution fails. + """ + client = bigquery.Client() + + # Load destination table configuration + utils.logger.info(f"[{destination_table}] Loading destination table configuration...") + destination_config = destination_table_builder.load_destination_config(config_path) + + # Retrieve the target destination table config + utils.logger.info(f"[{destination_table}] Retrieving destination table configuration...") + destination_table_config = destination_table_builder.get_destination_table_config( + destination_config, + destination_dataset, + destination_table + ) + + if not destination_table_config: + raise ValueError( + f"[{destination_table}] No config found for {destination_dataset}.{destination_table}" + ) + + # Build table schemas + table_schemas, schema_issues = destination_table_builder.build_table_schemas( + client, + destination_table_config + ) + + if schema_issues: + utils.logger.warning(f"[{destination_table}] Schema issues detected: {schema_issues}") + + # Identify missing columns + missing_cleaned_cols, missing_cleaned_loop_vars = destination_table_builder.find_missing_columns( + destination_table_config, + table_schemas + ) + + # Generate SQL + result = destination_table_builder.build_destination_table_query( + client=client, + full_config=destination_config, + destination_table_config=destination_table_config, + table_schemas=table_schemas, + classification_table=classification_table + ) + + sql = result["sql"] + fq_destination_table = result["destination_table"] + duplicate_columns = result.get("duplicate_columns", []) + unrecognized_cid_columns = result.get("unrecognized_cid_columns", []) + + # Save the SQL to GCS for audit purposes + try: + utils.logger.info(f"[{destination_table}] Saving SQL to GCS...") + gcs_client = storage.Client() + gcs_path = f"{constants.OUTPUT_SQL_PATH}{fq_destination_table}.sql" + utils.save_sql_string(sql=sql, path=gcs_path, storage_client=gcs_client) + utils.logger.info(f"[{destination_table}] SQL saved to GCS at {gcs_path}") + except Exception as e: + utils.logger.exception(f"[{destination_table}] Error saving SQL to GCS: {e}") + raise e + + # Execute the SQL + try: + utils.logger.info(f"[{destination_table}] Executing SQL query...") + query_job = client.query(sql) + utils.logger.info(f"[{destination_table}] Query job created with ID: {query_job.job_id}") + query_job.result() + utils.logger.info(f"[{destination_table}] Query execution completed successfully") + status = f"[{destination_table}] Table successfully created at {fq_destination_table}" + except Exception as e: + utils.logger.exception(f"[{destination_table}] Error executing SQL: {e}") + # Log more details about the exception + utils.logger.error(f"[{destination_table}] Exception type: {type(e).__name__}") + utils.logger.error(f"[{destination_table}] Exception args: {e.args}") + raise e + + # Write missing column report JSON + try: + report_path = f"{missing_report_base_path}{fq_destination_table}_missing_columns.json" + + destination_table_builder.create_destination_missing_columns_json( + client=gcs_client, + output_path=report_path, + destination_table=fq_destination_table, + missing_cleaned_cols=missing_cleaned_cols, + missing_cleaned_loop_vars=missing_cleaned_loop_vars, + duplicate_columns=duplicate_columns, + unrecognized_cid_columns=unrecognized_cid_columns + ) + + utils.logger.info(f"[{destination_table}] Missing column report saved to {report_path}") + + except Exception as e: + utils.logger.exception(f"[{destination_table}] Error writing missing column report for {fq_destination_table}: {e}") + raise e + + # Return result (matches pattern) + return { + "status": status, + "submitted_sql_path": constants.OUTPUT_SQL_PATH, + "missing_columns_report_path": report_path + } + +def create_censorship_summary_table( + output_table: str, + classification_table: str, + destination_table: str, + allowed_consent_groups: list[str], + rollup_report_base_path: str = constants.MISSING_COLUMNS_REPORT_PATH +) -> dict: + """ + Generates the censorship summary query, saves it to GCS for audit + purposes (consistent with other transform functions in this codebase), + executes it, writes the result to destination_table, and creates a + JSON rollup report of censorship counts by column and by consent_group. + + Args: + output_table: Fully qualified final destination table. Used only + to determine which columns exist and are therefore + eligible for censorship classification. + classification_table: Fully qualified table with consent_group/status/timestamp + columns for every participant. + destination_table: Fully qualified table to create with the summary. + allowed_consent_groups: Consent groups that are allowed to be included + in the censorship summary. + rollup_report_base_path: Base GCS path for the censorship rollup report. + + Returns: + dict: Contains "status", "submitted_sql_path", and "rollup_report_path". + + Raises: + Exception: Propagates any errors encountered during query execution. + """ + client = bigquery.Client() + _, _, table_short_name = utils.parse_fq_table(destination_table) + + output_columns = utils.get_column_names(client, output_table) + column_rule_map = censorship.build_column_rule_map(output_columns) + + utils.logger.info( + f"[{table_short_name}] Found {len(column_rule_map)} censorable columns out of " + f"{len(output_columns)} total columns in {output_table}" + ) + + final_sql = censorship.build_censorship_summary_sql( + output_table=output_table, + destination_table=destination_table, + column_rule_map=column_rule_map, + classification_table=classification_table, + allowed_consent_groups=allowed_consent_groups + ) + + # Save the SQL to GCS for audit purposes, matching existing pipeline convention + try: + gcs_client = storage.Client() + gcs_path = f"{constants.OUTPUT_SQL_PATH}{destination_table}.sql" + utils.save_sql_string(sql=final_sql, path=gcs_path, storage_client=gcs_client) + except Exception as e: + utils.logger.exception(f"[{table_short_name}] Error saving censorship summary SQL to {gcs_path}") + raise e + + # Execute the SQL + try: + utils.logger.info(f"[{table_short_name}] Executing censorship summary query...") + query_job = client.query(final_sql) + utils.logger.info(f"[{table_short_name}] Query job created with ID: {query_job.job_id}") + query_job.result() + status = f"[{table_short_name}] Censorship summary table successfully created at {destination_table}" + utils.logger.info(status) + except Exception as e: + utils.logger.exception(f"[{table_short_name}] Error executing censorship summary SQL: {e}") + raise e + + # Compute and write the rollup report + try: + rollup = censorship.get_censorship_rollup(destination_table) + report_path = f"{rollup_report_base_path}{destination_table}_rollup.json" + + censorship.create_censorship_rollup_json( + client=gcs_client, + output_path=report_path, + destination_table=destination_table, + by_column=rollup["by_column"], + by_consent_group=rollup["by_consent_group"], + ) + + utils.logger.info(f"[{table_short_name}] Censorship rollup report saved to {report_path}") + except Exception as e: + utils.logger.exception(f"[{table_short_name}] Error writing censorship rollup report: {e}") + raise e + + return { + "status": status, + "submitted_sql_path": constants.OUTPUT_SQL_PATH, + "rollup_report_path": report_path, + } + if __name__ == "__main__": #source_table = "nih-nci-dceg-connect-prod-6d04.ForTestingOnly.module1_v1_with_cleaned_columns" #destination_table = "nih-nci-dceg-connect-prod-6d04.CleanConnect.module1_fixed_binary_and_false_arrays" #process_rows(source_table=source_table, destination_table=destination_table) #create_sensitive_tier(source_table=source_table, destination_table=destination_table) - create_standardized_mapping_table("CleanConnect", "nih-nci-dceg-connect-stg-5519.pr2_mvp.mvp") + #create_standardized_mapping_table("CleanConnect", "nih-nci-dceg-connect-stg-5519.pr2_mvp.mvp") + client = bigquery.Client() + config_path = "reference/destination_config.json" + + create_classification_table( + parquet_url="gs://ehr_pipeline_tmp_stg/participant_status.parquet", + classification_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.classification", + local_csv_path=r"/Users/trivittge/Desktop/classification_df.csv", + ) + + print("######################") + create_destination_table( + config_path=config_path, + destination_dataset="SensitiveTier", + destination_table="mvp_0_1", + classification_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.classification" + ) + print("######################") + create_censorship_summary_table( + output_table="nih-nci-dceg-connect-stg-5519.SensitiveTier.mvp_0_1", + classification_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.classification", + destination_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.mvp_0_1_censorship_summary", + allowed_cases=["1", "2", "3A", "3B"], + rollup_report_base_path="gs://pr2-pipeline-artifacts-stg/missing_columns_report/" + ) + + # Synthetic data +# create_classification_table( +# parquet_url="gs://ehr_pipeline_tmp_stg/participant_status_synthetic.parquet", +# classification_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.classification_synthetic_2026_07_15", +# local_csv_path=r"/Users/trivittge/Desktop/classification_df_synthetic_2026_07_15.csv", +# ) +# +# print("######################") +# create_destination_table( +# config_path=config_path, +# destination_dataset="pr2_mvp", +# destination_table="mvp_synthetic_2026_07_15_case_1_2_3A_3B", +# classification_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.classification_synthetic_2026_07_15" +# ) +# print("######################") +# create_censorship_summary_table( +# output_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.mvp_synthetic_2026_07_15_case_1_2_3A_3B", +# classification_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.classification_synthetic_2026_07_15", +# destination_table="nih-nci-dceg-connect-stg-5519.pr2_mvp.censorship_summary_synthetic_2026_07_15", +# rollup_report_base_path="gs://pr2-pipeline-artifacts-stg/missing_columns_report/" +# ) \ No newline at end of file diff --git a/core/utils.py b/core/utils.py index 3d9cefa..659e585 100644 --- a/core/utils.py +++ b/core/utils.py @@ -6,11 +6,13 @@ import json import logging from collections import defaultdict -from typing import Optional +from typing import Any, Optional from datetime import datetime, timezone +from xmlrpc import client from google.cloud import bigquery, storage import pandas as pd #TODO Try to avoid using pandas +import numpy as np if __name__ == "__main__": # Add parent directory to Python path when running as script diff --git a/reference/destination_config.json b/reference/destination_config.json new file mode 100644 index 0000000..47b2bc3 --- /dev/null +++ b/reference/destination_config.json @@ -0,0 +1,178 @@ +{ + "filter_profiles": { + "default_participants": [ + { "column": "d_821247024", "value": "197316935", "comment": "Verified = Yes" }, + { "column": "d_831041022", "value": "104430631", "comment": "Destroy data = No" }, + { "column": "d_747006172", "value": "104430631", "comment": "Withdraw consent = No" }, + { "column": "d_773707518", "value": "104430631", "comment": "Revoke HIPAA = No"} + ], + "no_filters": [] + }, + "destination_tables": [ + { + "dataset": "SensitiveTier", + "table": "mvp_test", + "join_key": "Connect_ID", + "filter_profile": "no_filters", + "classification_filter": { + "allowed_consent_groups": ["1", "2", "3A", "3B"] + }, + "base_table": { + "dataset": "CleanConnect", + "table": "participants", + "columns": [ + "d_130371375_d_303552867_d_297462035", + "fakecolumn", + "d_130371375_d_303552867_20", + "d_130371375_d_303552867_20", + "d_130371375_d_303552867_d_320023644", + "d_130371375_d_303552867_d_438636757", + "d_130371375_d_303552867_d_438636757" + ] + }, + "join_tables": [ + { + "dataset": "pr2_mvp", + "table": "mvp", + "columns": [ + "d_726699695_d_384191091_d_746038746", + "d_726699695_d_384191091_d_807835037", + "d_965707586_d_947205597_d_712653855", + "d_965707586_d_976570371" + ] + }, + { + "dataset": "CleanConnect", + "table": "module1", + "columns": [ + "d_103397024_d_206625031" + ] + } + ] + }, + { + "dataset": "SensitiveTier", + "table": "mvp_0_1", + "join_key": "Connect_ID", + "filter_profile": "no_filters", + "classification_filter": { + "allowed_consent_groups": ["1", "2", "3A", "3B"] + }, + "base_table": { + "dataset": "pr2_mvp", + "table": "mvp", + "columns": "*" + }, + "join_tables": [] + }, + { + "dataset": "SensitiveTier", + "table": "NO_CLASSIFICATION_EMPTY_LIST", + "join_key": "Connect_ID", + "filter_profile": "no_filters", + "classification_filter": { + "allowed_consent_groups": [] + }, + "base_table": { + "dataset": "pr2_mvp", + "table": "mvp", + "columns": "*" + }, + "join_tables": [] + }, + { + "dataset": "SensitiveTier", + "table": "NO_CLASSIFICATION", + "join_key": "Connect_ID", + "filter_profile": "no_filters", + "base_table": { + "dataset": "pr2_mvp", + "table": "mvp", + "columns": "*" + }, + "join_tables": [] + }, + { + "dataset": "SensitiveTier", + "table": "mvp_test_filters", + "join_key": "Connect_ID", + "filter_profile": "default_participants", + "classification_filter": { + "allowed_consent_groups": ["1", "2", "3A", "3B"] + }, + "base_table": { + "dataset": "CleanConnect", + "table": "participants", + "columns": [ + "d_130371375_d_303552867_d_297462035", + "fakecolumn", + "d_130371375_d_303552867_20", + "d_130371375_d_303552867_20", + "d_130371375_d_303552867_d_320023644", + "d_130371375_d_303552867_d_438636757", + "d_130371375_d_303552867_d_438636757" + ] + }, + "join_tables": [ + { + "dataset": "pr2_mvp", + "table": "mvp", + "columns": [ + "d_726699695_d_384191091_d_746038746", + "d_726699695_d_384191091_d_807835037", + "d_965707586_d_947205597_d_712653855", + "d_965707586_d_976570371" + ] + }, + { + "dataset": "CleanConnect", + "table": "module1", + "columns": [ + "d_103397024_d_206625031" + ] + } + ] + }, + { + "dataset": "SensitiveTier", + "table": "mvp_test_NO_FILTERS", + "join_key": "Connect_ID", + "filter_profile": "no_filters", + "classification_filter": { + "allowed_consent_groups": [] + }, + "base_table": { + "dataset": "CleanConnect", + "table": "participants", + "columns": [ + "d_130371375_d_303552867_d_297462035", + "fakecolumn", + "d_130371375_d_303552867_20", + "d_130371375_d_303552867_20", + "d_130371375_d_303552867_d_320023644", + "d_130371375_d_303552867_d_438636757", + "d_130371375_d_303552867_d_438636757" + ] + }, + "join_tables": [ + { + "dataset": "pr2_mvp", + "table": "mvp", + "columns": [ + "d_726699695_d_384191091_d_746038746", + "d_726699695_d_384191091_d_807835037", + "d_965707586_d_947205597_d_712653855", + "d_965707586_d_976570371" + ] + }, + { + "dataset": "CleanConnect", + "table": "module1", + "columns": [ + "d_103397024_d_206625031" + ] + } + ] + } + ] +} \ No newline at end of file diff --git a/reference/pr2_mvp_column_mapping.json b/reference/pr2_mvp_column_mapping.json index 74c2d5e..6814344 100644 --- a/reference/pr2_mvp_column_mapping.json +++ b/reference/pr2_mvp_column_mapping.json @@ -1,6 +1,6 @@ { "_metadata": { - "generated_at": "2026-04-03T18:16:06.301340+00:00", + "generated_at": "2026-06-30T14:31:31.102738+00:00", "source": "CSV conversion script", "description": "Column mapping of concept IDs grouped by table", "structure": { @@ -19,33 +19,10 @@ "cleanconnect_cid": "Concept ID from CleanConnect", "cleanconnect_cid_updated": "CleanConnect concept ID with secondary source concept ID prepended" }, - "total_tables": 4, + "total_tables": 3, "total_rows": 67, - "missing_loop_variables": { - "bioSurvey": [ - { - "flatconnect_cid": "", - "cleanconnect_cid": "d_715581797_10_v2", - "cleanconnect_cid_updated": "" - } - ] - }, - "missing_columns": { - "participants": [ - { - "flatconnect_cid": "", - "cleanconnect_cid": "test1", - "cleanconnect_cid_updated": "" - } - ], - "module1": [ - { - "flatconnect_cid": "", - "cleanconnect_cid": "test2", - "cleanconnect_cid_updated": "test2" - } - ] - }, + "missing_loop_variables": {}, + "missing_columns": {}, "duplicate_rows": {} }, "tables": { @@ -356,6 +333,11 @@ "flatconnect_cid": "D_746012894", "cleanconnect_cid": "d_746012894", "cleanconnect_cid_updated": "d_726699695_d_746012894" + }, + { + "flatconnect_cid": "D_367803647_D_367803647", + "cleanconnect_cid": "d_367803647_d_367803647", + "cleanconnect_cid_updated": "d_726699695_d_367803647_d_367803647" } ], "module3": [ @@ -384,13 +366,6 @@ "cleanconnect_cid": "d_304657762", "cleanconnect_cid_updated": "d_965707586_d_304657762" } - ], - "bioSurvey": [ - { - "flatconnect_cid": "", - "cleanconnect_cid": "d_715581797_10_v2", - "cleanconnect_cid_updated": "" - } ] } } \ No newline at end of file