Source code for util.cpf

# THIS util records all post-challenge configurations for each state to avoid confusion in the future.
from config import Config




class CPFUtil:
    """
    Util class for CPF pre/post-challenge configurations.
    """

    # NOTE: we are missing alabama_sabcd and michigan_ditroit tenants from this list
    state_profiles = {
        "MS": {
            "table_pre_challenge": "c_cpf_merge_tile_ms",
            "table_trueup": "c_cpf_merge_tile_ms",
            "org_id": 0,
            "ffa_version": 31,
            "fabric_version": 4,
            "true_up": False,
            "dsl_mod": True,
            "fw_mod": True,
            "fabric_version_pre_challenge": 4,
            "ffa_version_pre_challenge": 31,
            "tenant": "mississippi",
        },
        "OK": {
            "table_pre_challenge": "c_cpf_merge_tile_ok",
            "table_trueup": "c_cpf_merge_tile_ok",
            "org_id": 0,
            "ffa_version": 33,
            "fabric_version": 4,
            "true_up": True,
            "dsl_mod": True,
            "fw_mod": False,
            "notes": "multiple programs at the same time",
            "fabric_version_pre_challenge": 4,
            "ffa_version_pre_challenge": 33,
            "tenant": "oklahoma",
        },
    }



    @staticmethod
    def cpf_profile(state_abbr: str, check_pre: bool = True) -> dict:
        """
        table_pre_challenge: the table name before the challenge
        table_trueup: the table name after the true-up. Before the challenge results applied. This table is only for data processing. Not for actual frontend functionality
        org_id: the organization id for the state
        ffa_version: the post challenge version of the fabric fixed availability data
        fabric_version: the post challenge version of the fabric data
        true_up: whether the state has true-up or not
        dsl_mod: whether the state has DSL modification or not
        fw_mod: whether the state has Cellular fixed wireless modification or not
        tenant: the name of the tenant in the grapi database

        :param str state_abbr: the state abbreviation in upper case
        :param str, optional check_pre: whether to check the pre-challenge tables. Default is True.
        :return: the post challenge configuration for the state

        NOTE: this function does call the AWS Athena and AWS Glue APIs to check the existence of the tables.
        Do not use outside of a DAG unless `check_pre` is set to False.
        """

        profile = CPFUtil.state_profiles.get(state_abbr.upper(), dict())

        if profile and check_pre:
            import awswrangler as wr

            table_pre_challenge = profile["table_pre_challenge"]
            assert wr.catalog.does_table_exist(
                table=table_pre_challenge, database=Config.DATA_LAKE_DB
            ), f"Table {table_pre_challenge=} does not exist in {Config.DATA_LAKE_DB}!"
            MIN_SIZE_PRE_CHALLENGE_MERGE_TILE_TABLE = 5_000
            assert (
                wr.athena.read_sql_query(
                    f"SELECT count(*) AS cnt FROM {table_pre_challenge}",
                    database=Config.DATA_LAKE_DB,
                ).iloc[0, 0]
                > MIN_SIZE_PRE_CHALLENGE_MERGE_TILE_TABLE
            ), f"Table {table_pre_challenge=} has fewer than {MIN_SIZE_PRE_CHALLENGE_MERGE_TILE_TABLE} rows!"

        return profile




    @staticmethod
    def get_cpf_postchallenge_states() -> list:
        """
        Get the list of states that have post-challenge configurations recorded.
        This does not check the pre-challenge tables, so you would need
        to call `bead_profile` to check individual states.
        """
        return list(CPFUtil.state_profiles.keys())