queries/process_trx.py

import pandas as pd

from queries.process_small_bts import process_small_bts_data
from utils.convert_to_excel import convert_dfs, save_dataframe
from utils.utils_vars import UtilsVars

TRX_COLUMNS = [
    "ID_BTS",
    "trxRfPower",
    "BCCH",
    "TRX_TCH",
    "number_trx_per_cell",
    "number_trx_per_bcf",
    "number_trx_per_site",
]


TRX_BTS_COLUMNS = [
    "BSC",
    "BCF",
    "BTS",
    "TRX",
    "ID_BTS",
    "number_trx_per_cell",
    "number_trx_per_bcf",
    "number_trx_per_site",
    "code",
    "name",
    "adminState",
    "bbUnitSupportsEdge",
    "channel0Maio",
    "channel0Type",
    "channel1Maio",
    "channel1Type",
    "channel2Maio",
    "channel2Type",
    "channel3Maio",
    "channel3Type",
    "channel4Maio",
    "channel4Type",
    "channel5Maio",
    "channel5Type",
    "channel6Maio",
    "channel6Type",
    "channel7Maio",
    "channel7Type",
    "initialFrequency",
    "lapdLinkName",
    "lapdLinkNumber",
    "mcpaTrxNumber",
    "mcpaTrxPortId",
    "mcpaTrxPosition",
    "numberOfTrxRfPowerLevels",
    "optimumRxLevDL",
    "optimumRxLevUL",
    "preferredBcchMark",
    "trxAbilities",
    "trxFrequencyType",
    "trxRfPower",
    "tsc",
    "TCHs",
    "SDs",
    "BCCHs",
    "CCCHs",
    "CBCs",
    "TotalChannels",
    "Signal",
    "number_tch_per_cell",
    "number_sd_per_cell",
    "number_bcch_per_cell",
    "number_ccch_per_cell",
    "number_cbc_per_cell",
    "number_total_channels_per_cell",
    "number_signals_per_cell",
]


def process_brute_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=["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_bcf"] = df_trx.groupby("ID_BCF")["ID_BCF"].transform("count")

    return df_trx


def process_trx_with_bts_name(file_path: str):

    df_gsm_trx = process_brute_trx_data(file_path=file_path).copy()
    df_gsm_trx.drop(columns=["name"], axis=1, inplace=True, errors="ignore")

    df_bts = process_small_bts_data(file_path=file_path)

    df_trx_bts_name: pd.DataFrame = pd.merge(
        df_gsm_trx, df_bts, on="ID_BTS", how="left"
    )
    df_trx_bts_name["number_trx_per_site"] = df_trx_bts_name.groupby("code")[
        "code"
    ].transform("count")
    # Filter columns strictly by names like "channelXType"
    channel_columns = [
        col
        for col in df_trx_bts_name.columns
        if col.startswith("channel") and col.endswith("Type")
    ]

    # TCHs	SDs	BCCH	CCCH	CBC	Total	Signal

    # Calculate "count of channels per TRX" for each row
    df_trx_bts_name["TCHs"] = df_trx_bts_name[channel_columns].apply(
        lambda row: (row == 2).sum(), axis=1
    )
    df_trx_bts_name["SDs"] = df_trx_bts_name[channel_columns].apply(
        lambda row: (row == 3).sum(), axis=1
    )

    df_trx_bts_name["BCCHs"] = df_trx_bts_name[channel_columns].apply(
        lambda row: (row == 4).sum(), axis=1
    )

    df_trx_bts_name["CCCHs"] = df_trx_bts_name[channel_columns].apply(
        lambda row: (row == 6).sum(), axis=1
    )

    df_trx_bts_name["CBCs"] = df_trx_bts_name[channel_columns].apply(
        lambda row: (row == 8).sum(), axis=1
    )

    # Total Channels =  TCHs + SDs + BCCHs + CCCHs + CBCs

    df_trx_bts_name["TotalChannels"] = (
        df_trx_bts_name["TCHs"]
        + df_trx_bts_name["SDs"]
        + df_trx_bts_name["BCCHs"]
        + df_trx_bts_name["CCCHs"]
        + df_trx_bts_name["CBCs"]
    )

    # Signal =  BCCHs + CCCHs + CBCs

    df_trx_bts_name["Signal"] = (
        df_trx_bts_name["BCCHs"] + df_trx_bts_name["CCCHs"] + df_trx_bts_name["CBCs"]
    )
    df_trx_bts_name["number_tch_per_cell"] = df_trx_bts_name.groupby("ID_BTS")[
        "TCHs"
    ].transform("sum")
    df_trx_bts_name["number_sd_per_cell"] = df_trx_bts_name.groupby("ID_BTS")[
        "SDs"
    ].transform("sum")
    df_trx_bts_name["number_bcch_per_cell"] = df_trx_bts_name.groupby("ID_BTS")[
        "BCCHs"
    ].transform("sum")
    df_trx_bts_name["number_ccch_per_cell"] = df_trx_bts_name.groupby("ID_BTS")[
        "CCCHs"
    ].transform("sum")
    df_trx_bts_name["number_cbc_per_cell"] = df_trx_bts_name.groupby("ID_BTS")[
        "CBCs"
    ].transform("sum")
    df_trx_bts_name["number_total_channels_per_cell"] = df_trx_bts_name.groupby(
        "ID_BTS"
    )["TotalChannels"].transform("sum")
    df_trx_bts_name["number_signals_per_cell"] = df_trx_bts_name.groupby("ID_BTS")[
        "Signal"
    ].transform("sum")

    df_trx_bts_name = df_trx_bts_name[TRX_BTS_COLUMNS]

    # UtilsVars.all_db_dfs.append(df_trx_bts_name)

    return df_trx_bts_name


def process_trx_data(file_path: str):

    df_gsm_trx = process_trx_with_bts_name(file_path=file_path).copy()

    bcch = df_gsm_trx[df_gsm_trx["channel0Type"] == 4]
    tch = df_gsm_trx[df_gsm_trx["channel0Type"] != 4][["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", "TRX_TCH"]

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

    return df_gsm_trx


def process_trx_with_bts_name_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.
    """
    trx_bts_name = process_trx_with_bts_name(file_path)
    UtilsVars.final_trx_database = convert_dfs([trx_bts_name], ["TRX"])