queries/process_gsm.py

import pandas as pd

from queries.process_mal import process_mal_data, process_mal_with_bts_name
from queries.process_trx import process_trx_data, process_trx_with_bts_name
from utils.config_band import config_band
from utils.convert_to_excel import convert_dfs, save_dataframe
from utils.utils_vars import UtilsVars

BTS_COLUMNS = [
    "ID_BCF",
    "ID_BTS",
    "ID_MAL",
    "BSC",
    "BCF",
    "BTS",
    "usedMobileAllocation",
    "code",
    "plmnPermitted",
    "frequencyBandInUse",
    "name",
    "Region",
    "adminState",
    "allowIMSIAttachDetach",
    "amrSegLoadDepTchRateLower",
    "amrSegLoadDepTchRateUpper",
    "antennaHopping",
    "bcchTrxPower",
    "bsIdentityCodeBCC",
    "bsIdentityCodeNCC",
    "BSIC",
    "cellId",
    "dedicatedGPRScapacity",
    "defaultGPRScapacity",
    "fddQMin",
    "fddQOffset",
    "fddRscpMin",
    "gprsEnabled",
    "locationAreaIdLAC",
    "rac",
    "rachDropRxLevelThreshold",
    "sectorId",
    "SectorId2",
    "segmentId",
    "fastReturnToLTE",
    "gsmPriority",
    "segmentName",
    "Code_Sector",
    "band_frequence",
    "type_cellule",
    "configuration_schema",
    "band",
]

BCF_COLUMNS = [
    "ID_BCF",
    "site_name",
]


def compare_trx_tch_versus_mal(tch1, tch2):
    # Split the strings by commas, convert to sets, and compare
    set1 = set(str(tch1).split(",")) if isinstance(tch1, str) else set()
    set2 = set(str(tch2).split(",")) if isinstance(tch2, str) else set()
    return set1 == set2


def process_gsm_data(file_path: str):
    """
    Process data from the specified file path.

    Args:
        file_path (str): The path to the file.
    """
    # Read the specific sheet into a DataFrame
    dfs = pd.read_excel(
        file_path,
        sheet_name=["BTS", "BCF"],
        engine="calamine",
        skiprows=[0],
    )

    # Process BTS data
    df_bts = dfs["BTS"]
    df_bts.columns = df_bts.columns.str.replace(r"[ ]", "", regex=True)
    df_bts["code"] = df_bts["name"].str.split("_").str[0].astype(int)
    df_bts["Region"] = df_bts["name"].str.split("_").str[1]
    df_bts["ID_BTS"] = df_bts[["BSC", "BCF", "BTS"]].astype(str).apply("_".join, axis=1)
    df_bts["ID_MAL"] = (
        df_bts[["BSC", "usedMobileAllocation"]].astype(str).apply("_".join, axis=1)
    )
    df_bts["BSIC"] = (
        df_bts[["bsIdentityCodeNCC", "bsIdentityCodeBCC"]]
        .astype(str)
        .apply("".join, axis=1)
    )
    df_bts["SectorId2"] = (
        df_bts["sectorId"].map(UtilsVars.sector_mapping).fillna(df_bts["sectorId"])
    )
    df_bts["band_frequence"] = (
        df_bts["frequencyBandInUse"]
        .map(UtilsVars.oml_band_frequence)
        .fillna("not found")
    )
    df_bts["type_cellule"] = (
        df_bts["frequencyBandInUse"].map(UtilsVars.type_cellule).fillna("not found")
    )
    df_bts["band"] = (
        df_bts["frequencyBandInUse"].map(UtilsVars.gsm_band).fillna("not found")
    )
    df_bts["configuration_schema"] = (
        df_bts["frequencyBandInUse"]
        .map(UtilsVars.configuration_schema)
        .fillna("not found")
    )

    df_bts["ID_BCF"] = df_bts[["BSC", "BCF"]].astype(str).apply("_".join, axis=1)
    df_bts["Code_Sector"] = (
        df_bts[["code", "SectorId2"]].astype(str).apply("_".join, axis=1)
    )
    df_bts["Code_Sector"] = df_bts["Code_Sector"].str.replace(".0", "")
    df_bts = df_bts[BTS_COLUMNS]

    # Process BCF data
    df_bcf = dfs["BCF"]
    df_bcf.columns = df_bcf.columns.str.replace(r"[ ]", "", regex=True)
    df_bcf["ID_BCF"] = df_bcf[["BSC", "BCF"]].astype(str).apply("_".join, axis=1)
    df_bcf.rename(columns={"name": "site_name"}, inplace=True)
    df_bcf = df_bcf[BCF_COLUMNS]

    # Process TRX data
    df_trx = process_trx_data(file_path)

    # Process MAL data
    df_mal = process_mal_data(file_path)

    # create band dataframe
    df_band = config_band(df_bts)

    # Merge dataframes
    df_bts_bcf = pd.merge(df_bcf, df_bts, on="ID_BCF", how="left")
    df_2g = pd.merge(df_bts_bcf, df_trx, on="ID_BTS", how="left")
    df_2g = pd.merge(df_2g, df_band, on="code", how="left")
    df_2g = pd.merge(df_2g, df_mal, on="ID_MAL", how="left")
    df_2g["TRX_TCH_VS_MAL"] = df_2g.apply(
        lambda row: compare_trx_tch_versus_mal(row["TRX_TCH"], row["MAL_TCH"]), axis=1
    )

    df_physical_db = UtilsVars.physisal_db
    df_2g = pd.merge(df_2g, df_physical_db, on="Code_Sector", how="left")

    # Save dataframes
    # save_dataframe(df_band, "BAND")
    # save_dataframe(df_bcf, "bcf")
    # save_dataframe(df_trx, "trx")
    # df_2g2 = save_dataframe(df_2g, "2g")

    # UtilsVars.all_db_dfs.append(df_2g)
    # UtilsVars.final_gsm_database = convert_dfs([df_2g], ["GSM"])
    # UtilsVars.final_gsm_database = [df_2g]
    return df_2g


def combined_gsm_database(file_path: str):
    gsm_df = process_gsm_data(file_path)
    mal_df = process_mal_with_bts_name(file_path)
    trx_df = process_trx_with_bts_name(file_path)

    UtilsVars.all_db_dfs.extend([gsm_df, mal_df, trx_df])
    return [gsm_df, mal_df, trx_df]


def process_gsm_data_to_excel(file_path: str):
    """
    Process data from the specified file path and save it to a excel file.

    Args:
        file_path (str): The path to the file.
    """
    gsm_dfs = combined_gsm_database(file_path)
    UtilsVars.final_gsm_database = convert_dfs(gsm_dfs, ["GSM", "MAL", "TRX"])