Source code for util.sbo_cai

from config import Config

# from .dag import DagUtil  # TODO: find ROOT CAUSE: local import works for all other util/module
# from util.dag import DagUtil  # this module REQUIRES absolute import!
# from util.rad import RADUtil  # this module REQUIRES absolute import!




class CAIUtil:
    """All sbos' CAI files and their location in datalake"""

    EXCEL_FILE_EXTENSIONS = (
        ".xls",
        ".xlsx",
        ".xlsm",
        ".xlsb",
        ".odf",
        ".ods",
        ".odt",
    )
    DICT_PTYPE_RAW_TO_MID = {
        # 7 specific CAI types: H SL GF CP (dont include generic cai_type 'CAI')
        "H": "HEALTH",
        "F": "GOVERNMENT/SAFETY",
        "S": "SCHOOL",
        "L": "LIBRARY",
        "G": "GOVERNMENT/BUILDING",
        "C": "SUPPORT",
        "P": "HOUSING",
    }  # NOTE: all keys are uppercased
    DICT_PTYPE_MID_TO_RAW = {v: k for k, v in DICT_PTYPE_RAW_TO_MID.items()}
    CAI_RAW_REQUIRED_COLS = {
        "raw_cai_type",
        "raw_entity_name",
        "raw_street",
        "raw_city",
        "raw_zipfull",
        "raw_state",
        "erate_entity_number",
        "cms_number",
        "frn",
        "explanation",
        "location_id",
        "download",
        "need1000",
        "lat",
        "lng",
    }

    # NOTE: col "stage" is not part of md5 formula
    # table r_cai_sbo (partitioned by state/volume/fa_ver/stage) might have duplicate md5
    # in different partitions/stages (same state, volume, fabric_address_version)
    CAI_DEDUP_IDCOLS = "state volume full_address_str entity_name cai_type".split()
    CAI_UNIQUE_MD5_COLS = CAI_DEDUP_IDCOLS + ["datasource_idx"]



    @staticmethod
    def pprint(print_msg, marker="NHAT_LOGGING"):
        print(
            f""" ******************************************************************************
                **************************************************** START LOGGING // {marker}
                {print_msg}
                **************************************************** END LOGGING
        """
        )




    @staticmethod
    def get_latlng_fractional_len(coord_df, coord_col):
        import pandas as pd

        series = coord_df[coord_col].astype(str).str.split(".", expand=True)
        series = (
            # Entire column (including string vals=empty, nan,Null,None,...) has 0decimal
            pd.Series([0] * len(coord_df))
            if series.shape[1] == 1
            # Num of decimals for values in this col; empty/None = has 0decimal
            else series[1].fillna("").str.len()
        )
        return series