queries/process_wcdma.py

import pandas as pd

from utils.convert_to_excel import convert_dfs, save_dataframe
from utils.extract_code import extract_code_from_mrbts
from utils.utils_vars import UtilsVars

WCEL_COLUMNS = [
    "ID_WBTS",
    "ID_WCEL",
    "RNC",
    "WBTS",
    "WCEL",
    "site_name",
    "name",
    "code",
    "AdminCellState",
    "CId",
    "LAC",
    "UARFCN",
    "PriScrCode",
    "SAC",
    "maxCarrierPower",
    "PtxPrimaryCPICH",
    "CellRange",
    "CodeTreeOptTimer",
    "CodeTreeOptimisation",
    "CodeTreeUsage",
    "PRACHDelayRange",
    "PrxOffset",
    "PrxTarget",
    "PrxTargetMax",
    "PrxTargetPSMax",
    "PrxTargetPSMaxtHSRACH",
    "PtxCellMax",
    "PtxOffset",
    "PtxTarget",
    "SmartLTELayeringEnabled",
    "SectorID",
    "Code_Sector",
    "code_wcel",
]

WBTS_COLUMNS = [
    "ID_WBTS",
    "site_name",
]

WNCEL_COLUMNS = [
    "code_wcel",
    "maxCarrierPower",
]


def process_wcdma_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
    # df_wcel = pd.read_excel(
    #     file_path, sheet_name="WCEL", engine="calamine", skiprows=[0]
    # )
    # df_wbts = pd.read_excel(
    #     file_path, sheet_name="WBTS", engine="calamine", skiprows=[0]
    # )
    # df_wncel = pd.read_excel(
    #     file_path, sheet_name="WNCEL", engine="calamine", skiprows=[0]
    # )

    dfs = pd.read_excel(
        file_path,
        sheet_name=["WCEL", "WBTS", "WNCEL"],
        engine="calamine",
        skiprows=[0],
    )

    # Process BTS data
    df_wcel = dfs["WCEL"]
    df_wcel.columns = df_wcel.columns.str.replace(r"[ ]", "", regex=True)
    df_wcel["code"] = df_wcel["name"].str.split("_").str[0].astype(int)
    df_wcel["ID_WCEL"] = (
        df_wcel[["RNC", "WBTS", "WCEL"]].astype(str).apply("_".join, axis=1)
    )

    df_wcel["ID_WBTS"] = df_wcel[["RNC", "WBTS"]].astype(str).apply("_".join, axis=1)
    df_wcel["Code_Sector"] = (
        df_wcel[["code", "SectorID"]].astype(str).apply("_".join, axis=1)
    )
    df_wcel["code_wcel"] = df_wcel[["code", "WCEL"]].astype(str).apply("_".join, axis=1)

    df_wcel["Code_Sector"] = df_wcel["Code_Sector"].str.replace(".0", "")

    # Process WBTS data
    df_wbts = dfs["WBTS"]
    df_wbts.columns = df_wbts.columns.str.replace(r"[ ]", "", regex=True)
    df_wbts["ID_WBTS"] = df_wbts[["RNC", "WBTS"]].astype(str).apply("_".join, axis=1)
    df_wbts.rename(columns={"name": "site_name"}, inplace=True)
    df_wbts = df_wbts[WBTS_COLUMNS]

    # Process WNCEL data
    df_wncel = dfs["WNCEL"]
    df_wncel.columns = df_wncel.columns.str.replace(r"[ ]", "", regex=True)
    df_wncel["CODE"] = df_wncel["MRBTS"].apply(extract_code_from_mrbts)
    df_wncel["code_wcel"] = (
        df_wncel[["CODE", "WNCEL"]].astype(str).apply("_".join, axis=1)
    )
    df_wncel = df_wncel[WNCEL_COLUMNS]

    # Merge dataframes
    df_wcel_bcf = pd.merge(df_wcel, df_wbts, on="ID_WBTS", how="left")

    df_3g = pd.merge(df_wcel_bcf, df_wncel, on="code_wcel", how="left")

    df_3g = df_3g[WCEL_COLUMNS]

    df_physical_db = UtilsVars.physisal_db
    df_3g = pd.merge(df_3g, df_physical_db, on="Code_Sector", how="left")
    # Save dataframes
    # save_dataframe(df_wcel, "wcel")
    # save_dataframe(df_wcel_bcf, "wbts")
    # save_dataframe(df_wncel, "wncel")
    # df_3g = save_dataframe(df_3g, "3G")

    UtilsVars.final_wcdma_database = convert_dfs([df_3g], ["WCDMA"])

    # BTS.process_ok = "Done"