queries/process_trx.py

import pandas as pd

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

TRX_COLUMNS = [
    "ID_BTS",
    "trxRfPower",
    "BCCH",
    "TCH",
    "number_trx_per_cell",
    "number_trx_per_site",
]


def process_trx_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", "TRX"],
        engine="calamine",
        skiprows=[0],
    )

    # Process TRX data
    df_trx = dfs["TRX"]
    df_trx.columns = df_trx.columns.str.replace(r"[ ]", "", regex=True)
    df_trx["ID_BTS"] = df_trx[["BSC", "BCF", "BTS"]].astype(str).apply("_".join, axis=1)
    df_trx["ID_BCF"] = df_trx[["BSC", "BCF"]].astype(str).apply("_".join, axis=1)
    df_trx["number_trx_per_cell"] = df_trx.groupby("ID_BTS")["ID_BTS"].transform(
        "count"
    )
    df_trx["number_trx_per_site"] = df_trx.groupby("ID_BCF")["ID_BCF"].transform(
        "count"
    )

    bcch = df_trx[df_trx["channel0Type"] == 4]
    tch = df_trx[df_trx["channel0Type"] == 3][["ID_BTS", "initialFrequency"]]

    tch = tch.pivot_table(
        index="ID_BTS",
        values="initialFrequency",
        aggfunc=lambda x: " ".join(map(str, x)),
    )

    tch = tch.reset_index()

    # rename the columns
    tch.columns = ["ID_BTS", "TCH"]

    # Merge dataframes

    df_trx = pd.merge(bcch, tch, on="ID_BTS", how="left")
    # rename "initialFrequency" to "BCCH"
    df_trx = df_trx.rename(columns={"initialFrequency": "BCCH"})
    df_trx = df_trx[TRX_COLUMNS]

    # Save dataframes
    # save_dataframe(df_trx, "trx")
    # df_2g2 = save_dataframe(df_2g, "2g")

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


# process_trx_data(r"data2\20240805_5810_05082024_Dump.xml.gz.xlsb")