|
|
import pandas as pd |
|
|
|
|
|
from queries.process_trx import process_trx_data |
|
|
from utils.convert_to_excel import convert_dfs, save_dataframe |
|
|
from utils.utils_vars import UtilsVars |
|
|
|
|
|
BTS_COLUMNS = [ |
|
|
"ID_BCF", |
|
|
"ID_BTS", |
|
|
"BSC", |
|
|
"BCF", |
|
|
"BTS", |
|
|
"code", |
|
|
"plmnPermitted", |
|
|
"frequencyBandInUse", |
|
|
"name", |
|
|
"adminState", |
|
|
"allowIMSIAttachDetach", |
|
|
"amrSegLoadDepTchRateLower", |
|
|
"amrSegLoadDepTchRateUpper", |
|
|
"antennaHopping", |
|
|
"bcchTrxPower", |
|
|
"bsIdentityCodeBCC", |
|
|
"bsIdentityCodeNCC", |
|
|
"BSIC", |
|
|
"cellId", |
|
|
"dedicatedGPRScapacity", |
|
|
"defaultGPRScapacity", |
|
|
"fddQMin", |
|
|
"fddQOffset", |
|
|
"fddRscpMin", |
|
|
"gprsEnabled", |
|
|
"locationAreaIdLAC", |
|
|
"locationAreaIdMCC", |
|
|
"locationAreaIdMNC", |
|
|
"rac", |
|
|
"rachDropRxLevelThreshold", |
|
|
"sectorId", |
|
|
"SectorId2", |
|
|
"segmentId", |
|
|
"fastReturnToLTE", |
|
|
"gsmPriority", |
|
|
"segmentName", |
|
|
"Code_Sector", |
|
|
] |
|
|
|
|
|
BCF_COLUMNS = [ |
|
|
"ID_BCF", |
|
|
"site_name", |
|
|
] |
|
|
|
|
|
|
|
|
def process_gsm_data(file_path: str): |
|
|
""" |
|
|
Process data from the specified file path. |
|
|
|
|
|
Args: |
|
|
file_path (str): The path to the file. |
|
|
""" |
|
|
|
|
|
dfs = pd.read_excel( |
|
|
file_path, |
|
|
sheet_name=["BTS", "BCF", "TRX"], |
|
|
engine="calamine", |
|
|
skiprows=[0], |
|
|
) |
|
|
|
|
|
|
|
|
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["ID_BTS"] = df_bts[["BSC", "BCF", "BTS"]].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["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] |
|
|
|
|
|
|
|
|
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] |
|
|
|
|
|
df_trx = process_trx_data(file_path) |
|
|
|
|
|
|
|
|
df_bts_bcf = pd.merge(df_bts, df_bcf, on="ID_BCF", how="left") |
|
|
df_2g = pd.merge(df_bts_bcf, df_trx, on="ID_BTS", how="left") |
|
|
|
|
|
df_physical_db = UtilsVars.physisal_db |
|
|
df_2g = pd.merge(df_2g, df_physical_db, on="Code_Sector", how="left") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
save_dataframe(df_trx, "trx") |
|
|
|
|
|
|
|
|
UtilsVars.final_gsm_database = convert_dfs([df_2g], ["GSM"]) |
|
|
|
