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", "G1800 TRX Per BCF", "G900 TRX Per BCF", ] TRX_BTS_COLUMNS = [ "BSC", "BCF", "BTS", "TRX", "ID_BTS", "ID_BCF", "number_trx_per_cell", "number_trx_per_bcf", "number_trx_per_site", "G1800 TRX Per BCF", "G900 TRX Per BCF", "code", "name", "SectorId2", "band", "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") # Avoir les TRX par bande et par secteur et BCF sous forme concaténée comme 5/3/3 trx_band_bcfid = ( df_trx_bts_name.groupby(["ID_BCF", "band", "SectorId2"]) .size() .reset_index(name="count") ) # Étape 2 : pivot avec valeurs manquantes = 0 pivot_df = trx_band_bcfid.pivot_table( index=["ID_BCF", "band"], columns="SectorId2", values="count", fill_value=0 ) # Étape 3 : s’assurer que toutes les colonnes de secteur (1, 2, 3) existent for s in [1, 2, 3]: if s not in pivot_df.columns: pivot_df[s] = 0 # 🟢 Conversion en entier AVANT concaténation pivot_df[[1, 2, 3]] = pivot_df[[1, 2, 3]].astype(int) pivot_df["concat"] = pivot_df[[1, 2, 3]].astype(str).agg("/".join, axis=1) final = pivot_df["concat"].unstack().reset_index().rename_axis(None, axis=1) final.columns = ["ID_BCF", "G1800 TRX Per BCF", "G900 TRX Per BCF"] df_trx_bts_name = pd.merge(df_trx_bts_name, final, on="ID_BCF", how="left") df_trx_bts_name = df_trx_bts_name[TRX_BTS_COLUMNS] 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"])