db_query / queries /process_mrbts.py
DavMelchi's picture
Remove process MRBTS and introduce ADJL
2e86e92
raw
history blame
4.11 kB
import pandas as pd
from utils.convert_to_excel import convert_dfs
from utils.extract_code import extract_code_from_mrbts
from utils.utils_vars import UtilsVars, clean_bands
def process_mrbts_data(
file_path: str,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
"""
Process data from the specified file path.
Args:
file_path (str): The path to the file.
"""
dfs = pd.read_excel(
file_path,
sheet_name=["MRBTS", "GNBCF", "WNBTS", "ALD"],
engine="calamine",
skiprows=[0],
)
# MRBTS
df_mrbts = dfs["MRBTS"]
df_mrbts.columns = df_mrbts.columns.str.replace(r"[ ]", "", regex=True)
# Create an explicit copy of the filtered DataFrame to avoid SettingWithCopyWarning
df_mrbts = df_mrbts[df_mrbts["MRBTS"].apply(lambda x: str(x).isnumeric())].copy()
# Now use .loc to set values in the DataFrame
df_mrbts.loc[:, "code"] = df_mrbts["MRBTS"].apply(extract_code_from_mrbts)
df_mrbts = df_mrbts[["MRBTS", "code", "name", "btsName"]]
# GNBCF
df_gnbcf = dfs["GNBCF"]
df_gnbcf.columns = df_gnbcf.columns.str.replace(r"[ ]", "", regex=True)
df_gnbcf = df_gnbcf[["MRBTS", "bscId", "bcfId"]]
df_gnbcf["ID_BCF"] = (
(df_gnbcf[["bscId", "bcfId"]]).astype(str).apply("_".join, axis=1)
)
df_gnbcf = df_gnbcf[["MRBTS", "ID_BCF"]]
# WNBTS
df_wnbts = dfs["WNBTS"]
df_wnbts.columns = df_wnbts.columns.str.replace(r"[ ]", "", regex=True)
df_wnbts = df_wnbts[["MRBTS", "wbtsId"]]
df_wnbts = df_wnbts.rename(columns={"wbtsId": "WBTS"})
# ALD
df_ald = dfs["ALD"]
df_ald.columns = df_ald.columns.str.replace(r"[ ]", "", regex=True)
df_ald = df_ald[["MRBTS", "productCode"]]
df_ald = df_ald.drop_duplicates(subset=["MRBTS"], keep="first")
df_mrbts = pd.merge(df_mrbts, df_gnbcf, on="MRBTS", how="left")
df_mrbts = pd.merge(df_mrbts, df_wnbts, on="MRBTS", how="left")
df_mrbts = pd.merge(df_mrbts, df_ald, on="MRBTS", how="left")
##################################################################################"""
###################################################################################
gsm_df: pd.DataFrame = UtilsVars.all_db_dfs[0]
wcdma_df: pd.DataFrame = UtilsVars.all_db_dfs[3]
lte_fdd_df: pd.DataFrame = UtilsVars.all_db_dfs[4]
lte_tdd_df: pd.DataFrame = UtilsVars.all_db_dfs[5]
gsm_df = gsm_df[["ID_BCF", "site_name", "number_trx_per_bcf", "bcf_config_band"]]
gsm_df = gsm_df.drop_duplicates(subset=["ID_BCF"], keep="first")
gsm_df = gsm_df.rename(columns={"site_name": "gsm_name"})
wcdma_df = wcdma_df[["WBTS", "site_name", "wbts_config_band"]]
wcdma_df = wcdma_df.drop_duplicates(subset=["WBTS"], keep="first")
wcdma_df = wcdma_df.rename(columns={"site_name": "wcdma_name"})
lte_fdd_df = lte_fdd_df[["MRBTS", "lnbts_name", "lte_config_band"]]
lte_tdd_df = lte_tdd_df[["MRBTS", "lnbts_name", "lte_config_band"]]
lte_df = pd.concat([lte_fdd_df, lte_tdd_df], ignore_index=True)
lte_df = lte_df.drop_duplicates(subset=["MRBTS"], keep="first")
df_mrbts = pd.merge(df_mrbts, gsm_df, on="ID_BCF", how="left")
df_mrbts = pd.merge(df_mrbts, wcdma_df, on="WBTS", how="left")
df_mrbts = pd.merge(df_mrbts, lte_df, on="MRBTS", how="left")
df_mrbts["mrbts_config_band"] = (
df_mrbts[["bcf_config_band", "wbts_config_band", "lte_config_band"]]
.astype(str)
.apply("/".join, axis=1)
)
df_mrbts["mrbts_config_band"] = df_mrbts["mrbts_config_band"].apply(clean_bands)
UtilsVars.all_db_dfs.append(df_mrbts)
UtilsVars.all_db_dfs_names.append("MRBTS")
return df_mrbts, df_gnbcf, df_wnbts
# def process_mrbts_data_to_excel(file_path: str) -> None:
# """
# Process data from the specified file path and save it to a excel file.
# Args:
# file_path (str): The path to the file.
# """
# mrbts_df, gnbcf_df, wnbts_df = process_mrbts_data(file_path)
# UtilsVars.final_mrbts_database = convert_dfs(
# [mrbts_df, gnbcf_df, wnbts_df], ["MRBTS", "GNBCF", "WNBTS"]
# )