queries/process_neighbors.py

import pandas as pd

from utils.convert_to_excel import convert_dfs, save_dataframe
from utils.utils_vars import UtilsVars

ADCE_INITIAL_COLUMNS = [
    "ID_BTS",
    "lac_id",
]

ADJS_INITIAL_COLUMNS = [
    "ID_WCEL",
    "lac_id",
]

BTS_SOURCE = [
    "ID_BTS",
    "name",
]
BTS_TARGET = [
    "lac_id",
    "name",
]

WCEL_SOURCE = [
    "ID_WCEL",
    "name",
]

WCEL_TARGET = [
    "lac_id",
    "name",
]


def process_neighbors_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=["ADCE", "ADJS", "ADJI", "ADJG", "ADJW", "BTS", "WCEL"],
        engine="calamine",
        skiprows=[0],
    )

    # # Process ADCE data
    df_adce = dfs["ADCE"]
    df_adce.columns = df_adce.columns.str.replace(r"[ ]", "", regex=True)
    df_adce["ID_BTS"] = (
        df_adce[["BSC", "BCF", "BTS"]].astype(str).apply("_".join, axis=1)
    )
    df_adce["lac_id"] = (
        df_adce[["adjacentCellIdLac", "adjacentCellIdCI"]]
        .astype(str)
        .apply("_".join, axis=1)
    )
    df_adce["lac_id"] = df_adce["lac_id"].str.replace(".0", "")
    df_adce = df_adce[ADCE_INITIAL_COLUMNS]

    # Process BTS data
    df_bts = dfs["BTS"]
    df_bts.columns = df_bts.columns.str.replace(r"[ ]", "", regex=True)
    df_bts["ID_BTS"] = df_bts[["BSC", "BCF", "BTS"]].astype(str).apply("_".join, axis=1)
    df_bts["lac_id"] = (
        df_bts[["locationAreaIdLAC", "cellId"]].astype(str).apply("_".join, axis=1)
    )

    df_bts_source = df_bts[BTS_SOURCE]
    df_bts_source.rename(columns={"name": "SOURCE_NAME"}, inplace=True)

    df_bts_target = df_bts[BTS_TARGET]
    df_bts_target.rename(columns={"name": "TARGET_NAME"}, inplace=True)

    # #create final adce
    df_adce_final = pd.merge(df_adce, df_bts_source, on="ID_BTS", how="left")
    df_adce_final = pd.merge(
        df_adce_final, df_bts_target, on="lac_id", how="left"
    ).dropna()
    df_adce_final.rename(
        columns={"ID_BTS": "SOURCE_ID", "lac_id": "TARGET_LAC_ID"}, inplace=True
    )

    # process ADJS data
    df_adjs = dfs["ADJS"]
    df_adjs.columns = df_adjs.columns.str.replace(r"[ ]", "", regex=True)

    df_adjs["ID_WCEL"] = (
        df_adjs[["RNC", "WBTS", "WCEL"]].astype(str).apply("_".join, axis=1)
    )
    df_adjs["lac_id"] = (
        df_adjs[["AdjsLAC", "AdjsCI"]].astype(str).apply("_".join, axis=1)
    )
    df_adjs = df_adjs[ADJS_INITIAL_COLUMNS]

    # process WCEL DATA
    df_wcel = dfs["WCEL"]
    df_wcel.columns = df_wcel.columns.str.replace(r"[ ]", "", regex=True)
    df_wcel["ID_WCEL"] = (
        df_wcel[["RNC", "WBTS", "WCEL"]].astype(str).apply("_".join, axis=1)
    )
    df_wcel["lac_id"] = df_wcel[["LAC", "CId"]].astype(str).apply("_".join, axis=1)
    df_wcel = df_wcel[["ID_WCEL", "lac_id", "name"]]

    df_wcel_source = df_wcel[WCEL_SOURCE]
    df_wcel_source.rename(columns={"name": "SOURCE_NAME"}, inplace=True)

    df_wcel_target = df_wcel[WCEL_TARGET]
    df_wcel_target.rename(columns={"name": "TARGET_NAME"}, inplace=True)

    # create final adjs
    df_adjs_final = pd.merge(df_adjs, df_wcel_source, on="ID_WCEL", how="left")
    df_adjs_final = pd.merge(
        df_adjs_final, df_wcel_target, on="lac_id", how="left"
    ).dropna()
    df_adjs_final.rename(
        columns={"ID_WCEL": "SOURCE_ID", "lac_id": "TARGET_LAC_ID"}, inplace=True
    )

    # process ADJI DATA
    df_adji = dfs["ADJI"]
    df_adji.columns = df_adji.columns.str.replace(r"[ ]", "", regex=True)

    df_adji["ID_WCEL"] = (
        df_adji[["RNC", "WBTS", "WCEL"]].astype(str).apply("_".join, axis=1)
    )
    df_adji["lac_id"] = (
        df_adji[["AdjiLAC", "AdjiCI"]].astype(str).apply("_".join, axis=1)
    )
    df_adji = df_adji[["ID_WCEL", "lac_id"]]

    df_adji_final = pd.merge(df_adji, df_wcel_source, on="ID_WCEL", how="left")
    df_adji_final = pd.merge(
        df_adji_final, df_wcel_target, on="lac_id", how="left"
    ).dropna()
    df_adji_final.rename(
        columns={"ID_WCEL": "SOURCE_ID", "lac_id": "TARGET_LAC_ID"}, inplace=True
    )

    # process ADJG DATA
    df_adjg = dfs["ADJG"]
    df_adjg.columns = df_adjg.columns.str.replace(r"[ ]", "", regex=True)

    df_adjg["ID_WCEL"] = (
        df_adjg[["RNC", "WBTS", "WCEL"]].astype(str).apply("_".join, axis=1)
    )
    df_adjg["lac_id"] = (
        df_adjg[["AdjgLAC", "AdjgCI"]].astype(str).apply("_".join, axis=1)
    )
    df_adjg = df_adjg[["ID_WCEL", "lac_id"]]

    df_adjg_final = pd.merge(df_adjg, df_wcel_source, on="ID_WCEL", how="left")
    df_adjg_final = pd.merge(
        df_adjg_final, df_bts_target, on="lac_id", how="left"
    ).dropna()
    df_adjg_final.rename(
        columns={"ID_WCEL": "SOURCE_ID", "lac_id": "TARGET_LAC_ID"}, inplace=True
    )

    # process ADJW DATA
    df_adjw = dfs["ADJW"]
    df_adjw.columns = df_adjw.columns.str.replace(r"[ ]", "", regex=True)

    df_adjw["ID_BTS"] = (
        df_adjw[["BSC", "BCF", "BTS"]].astype(str).apply("_".join, axis=1)
    )
    df_adjw["lac_id"] = df_adjw[["lac", "AdjwCId"]].astype(str).apply("_".join, axis=1)
    df_adjw = df_adjw[["ID_BTS", "lac_id"]]

    df_adjw_final = pd.merge(df_adjw, df_bts_source, on="ID_BTS", how="left")
    df_adjw_final = pd.merge(
        df_adjw_final, df_wcel_target, on="lac_id", how="left"
    ).dropna()
    df_adjw_final.rename(
        columns={"ID_BTS": "SOURCE_ID", "lac_id": "TARGET_LAC_ID"}, inplace=True
    )

    # save_dataframe(df_adjw_final, "ADJW")

    return [df_adjw_final, df_adjg_final, df_adji_final, df_adjs_final, df_adce_final]


def process_neighbors_data_to_excel(file_path: str):
    neighbors_dfs = process_neighbors_data(file_path)
    UtilsVars.neighbors_database = convert_dfs(
        neighbors_dfs, ["ADJW", "ADJG", "ADJI", "ADJS", "ADCE"]
    )