import pandas as pd from utils.config_band import config_band from utils.convert_to_excel import convert_dfs, save_dataframe from utils.extract_code import extract_code_from_mrbts from utils.kml_creator import generate_kml_from_df from utils.utils_vars import UtilsVars, WcdmaAnalysisData, get_physical_db WCEL_COLUMNS = [ "ID_WBTS", "ID_WCEL", "RNC", "WBTS", "WCEL", "site_name", "name", "code", "Region", "AdminCellState", "CId", "LAC", "RAC", "UARFCN", "PriScrCode", "SAC", "maxCarrierPower", "PtxPrimaryCPICH", "CellRange", "CodeTreeOptTimer", "CodeTreeOptimisation", "CodeTreeUsage", "PRACHDelayRange", "PrxOffset", "PrxTarget", "PrxTargetMax", "PrxTargetPSMax", "PrxTargetPSMaxtHSRACH", "PtxCellMax", "PtxOffset", "PtxTarget", "SmartLTELayeringEnabled", "HSDPAFmcgIdentifier", "NrtFmcgIdentifier", "RtFmcgIdentifier", "RTWithHSDPAFmcgIdentifier", "HSDPAFmciIdentifier", "NrtFmciIdentifier", "RtFmciIdentifier", "RTWithHSDPAFmciIdentifier", "HSDPAFmcsIdentifier", "HSPAFmcsIdentifier", "NrtFmcsIdentifier", "RtFmcsIdentifier", "RTWithHSDPAFmcsIdentifier", "RTWithHSPAFmcsIdentifier", "Sintersearch", "SintersearchConn", "Sintrasearch", "SintrasearchConn", "Ssearch_RATConn", "TreselectionFACH", "TreselectionPCH", "SectorID", "Code_Sector", "code_wcel", "porteuse", "band", ] WBTS_COLUMNS = [ "ID_WBTS", "site_name", ] WNCEL_COLUMNS = [ "code_wcel", "maxCarrierPower", ] WCDMA_KML_COLUMNS = [ "code", "name", "Longitude", "Latitude", "Azimut", "Hauteur", "LAC", "CId", "LAC", "UARFCN", "PriScrCode", "band", ] 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] df_wcel["code"] = ( pd.to_numeric(df_wcel["code"], errors="coerce").fillna(0).astype(int) ) df_wcel["Region"] = df_wcel["name"].str.split("_").str[1] 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", "") df_wcel["porteuse"] = ( df_wcel["UARFCN"].map(UtilsVars.porteuse_mapping).fillna("not found") ) df_wcel["band"] = df_wcel["UARFCN"].map(UtilsVars.wcdma_band).fillna("not found") # create config_band dataframe df_band = config_band(df_wcel) # 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 = get_physical_db() df_3g = pd.merge(df_3g, df_band, on="code", how="left") 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.all_db_dfs.append(df_3g) UtilsVars.wcdma_dfs.append(df_3g) UtilsVars.all_db_dfs_names.append("WCDMA") # UtilsVars.final_wcdma_database = convert_dfs([df_3g], ["WCDMA"]) return df_3g # UtilsVars.final_wcdma_database = [df_3g] # BTS.process_ok = "Done" def process_wcdma_data_to_excel(file_path: str): """ Process WCDMA data from the specified file path and convert it to Excel format Args: file_path (str): The path to the file. """ wcdma_dfs = process_wcdma_data(file_path) UtilsVars.final_wcdma_database = convert_dfs([wcdma_dfs], ["WCDMA"]) ############################# KML CREATION ################################# def process_wcdma_data_to_kml(file_path: str): """ Process WCDMA data from the specified file path and convert it to KML format Args: file_path (str): The path to the file. """ wcdma_kml_df = process_wcdma_data(file_path) wcdma_kml_df = wcdma_kml_df[WCDMA_KML_COLUMNS] # Add colors column base on "band" column wcdma_kml_df["color"] = wcdma_kml_df["band"].map(UtilsVars.color_mapping) # Add size column base on "band" column wcdma_kml_df["size"] = wcdma_kml_df["band"].map(UtilsVars.size_mapping) # Remove empty rows wcdma_kml_df = wcdma_kml_df.dropna(subset=["Longitude", "Latitude", "Azimut"]) # Generate kml UtilsVars.wcdma_kml_file = generate_kml_from_df(wcdma_kml_df) ############################ANALYTICSS AND STATISTICS############################ def wcdma_analaysis( filepath: str, # region_list: list ): """ Process WCDMA data from the specified file path and convert it to Excel format Args: filepath (str): The path to the file. """ # wcdma_df = process_wcdma_data(filepath) wcdma_df: pd.DataFrame = UtilsVars.wcdma_dfs[0] # filter per list of regions # wcdma_df = wcdma_df.loc[wcdma_df["Region"].isin(region_list)] # df to count number of site per rnc df_site_per_rnc = wcdma_df[["RNC", "code"]] df_site_per_rnc = df_site_per_rnc.drop_duplicates(subset=["code"], keep="first") df_site_per_lac = wcdma_df.loc[:, ["RNC", "LAC", "code"]].copy() df_site_per_lac.loc[:, "code_lac"] = ( df_site_per_lac["code"].astype(str) + "_" + df_site_per_lac["LAC"].astype(str) ) df_site_per_lac = df_site_per_lac.drop_duplicates(subset=["code_lac"], keep="first") WcdmaAnalysisData.total_number_of_rnc = wcdma_df["RNC"].nunique() WcdmaAnalysisData.total_number_of_wcel = wcdma_df["ID_WCEL"].nunique() WcdmaAnalysisData.number_of_site = len(wcdma_df["site_name"].dropna().unique()) WcdmaAnalysisData.number_of_site_per_rnc = df_site_per_rnc["RNC"].value_counts() WcdmaAnalysisData.number_of_cell_per_rnc = wcdma_df["RNC"].value_counts() WcdmaAnalysisData.number_of_empty_wbts_name = wcdma_df["site_name"].isnull().sum() WcdmaAnalysisData.number_of_empty_wcel_name = wcdma_df["name"].isnull().sum() WcdmaAnalysisData.wcel_administate_distribution = wcdma_df[ "AdminCellState" ].value_counts() WcdmaAnalysisData.psc_distribution = wcdma_df["PriScrCode"].value_counts() # Manage Cells count per LAC and RNC # Pivot RNC and LAC WcdmaAnalysisData.number_of_cell_per_lac = ( wcdma_df.groupby(["RNC", "LAC"]).size().reset_index(name="count") ) # Rename columns WcdmaAnalysisData.number_of_cell_per_lac = ( WcdmaAnalysisData.number_of_cell_per_lac.rename( columns={"RNC": "RNC", "LAC": "LAC", "count": "LAC_Count"} ) ) # Add "RNC_" and "LAC_" prefix WcdmaAnalysisData.number_of_cell_per_lac["RNC"] = ( "RNC_" + WcdmaAnalysisData.number_of_cell_per_lac["RNC"].astype(str) ) WcdmaAnalysisData.number_of_cell_per_lac["LAC"] = ( "LAC_" + WcdmaAnalysisData.number_of_cell_per_lac["LAC"].astype(str) ) ##################### Number of site per LAC WcdmaAnalysisData.number_of_site_per_lac = ( df_site_per_lac.groupby(["RNC", "LAC"]).size().reset_index(name="count") ) # Rename columns WcdmaAnalysisData.number_of_site_per_lac = ( WcdmaAnalysisData.number_of_site_per_lac.rename( columns={"RNC": "RNC", "LAC": "LAC", "count": "Site_Count"} ) ) # Add "RNC_" and "LAC_" prefix WcdmaAnalysisData.number_of_site_per_lac["RNC"] = ( "RNC_" + WcdmaAnalysisData.number_of_site_per_lac["RNC"].astype(str) ) WcdmaAnalysisData.number_of_site_per_lac["LAC"] = ( "LAC_" + WcdmaAnalysisData.number_of_site_per_lac["LAC"].astype(str) )