queries/process_neighbors.py

import pandas as pd
from geopy.distance import geodesic

from queries.process_gsm import process_gsm_data
from queries.process_wcdma import process_wcdma_data
from utils.convert_to_excel import convert_dfs, save_dataframe
from utils.utils_vars import UtilsVars

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

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

BTS_SOURCE = [
    "ID_BTS",
    "ID_BCF",
    "name",
    "BCCH",
    "BSIC",
    "Longitude",
    "Latitude",
]
BTS_TARGET = [
    "lac_id",
    "ID_BCF",
    "name",
    "BCCH",
    "BSIC",
    "Longitude",
    "Latitude",
]

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

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


def check_symmetry(df: pd.DataFrame):
    """
    Check for symmetric relationships in a dataframe of network neighbors.
    For each source-target pair, checks if the reverse target-source pair exists.

    Args:
        df: pandas DataFrame with columns 'SOURCE_NAME' and 'TARGET_NAME'

    Returns:
        DataFrame with added 'SYMETRIQUE' column ('YES' if symmetric, 'NO' otherwise)
    """
    # Create a set of all (source, target) pairs for fast lookup
    pairs = set(zip(df["SOURCE_NAME"], df["TARGET_NAME"]))

    # Check for each row if the reverse relationship exists
    df["SYMETRIQUE"] = df.apply(
        lambda row: (
            "YES" if (row["TARGET_NAME"], row["SOURCE_NAME"]) in pairs else "NO"
        ),
        axis=1,
    )

    return df


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 = process_gsm_data(file_path)
    df_bts["lac_id"] = (
        df_bts[["locationAreaIdLAC", "cellId"]]
        .astype(str)
        .apply("_".join, axis=1)
        .str.replace(".0", "")
    )

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

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

    # #create final adce
    df_adce_final = pd.merge(df_adce, df_bts_source, on="ID_BTS", how="left")

    # Rename SOURCELongitude and Latitude columns
    df_adce_final.rename(
        columns={
            "Longitude": "SOURCE_Longitude",
            "Latitude": "SOURCE_Latitude",
            "ID_BCF": "SOURCE_ID_BCF",
            "BCCH": "SOURCE_BCCH",
            "BSIC": "SOURCE_BSIC",
        },
        inplace=True,
    )

    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",
            "Longitude": "TARGET_Longitude",
            "Latitude": "TARGET_Latitude",
            "ID_BCF": "TARGET_ID_BCF",
            "BCCH": "TARGET_BCCH",
            "BSIC": "TARGET_BSIC",
        },
        inplace=True,
    )
    df_adce_final = check_symmetry(df_adce_final)
    # Add column "Sync Comment"
    # if SOURCE_ID_BCF = TARGET_ID_BCF and synchronized = 0 the "Need synchronized" else ""
    df_adce_final["Sync Comment"] = df_adce_final.apply(
        lambda row: (
            "Need synchronized"
            if row["SOURCE_ID_BCF"] == row["TARGET_ID_BCF"] and row["synchronized"] == 0
            else ""
        ),
        axis=1,
    )
    # Add column "Same BSIC" if SOURCE_BSIC = TARGET_BSIC THEN "Yes" else ""
    df_adce_final["Same BSIC"] = df_adce_final.apply(
        lambda row: "Yes" if row["SOURCE_BSIC"] == row["TARGET_BSIC"] else "",
        axis=1,
    )
    # Add column "Same BCCH" if SOURCE_BCCH = TARGET_BCCH THEN "Yes" else ""
    df_adce_final["Same BCCH"] = df_adce_final.apply(
        lambda row: "Yes" if row["SOURCE_BCCH"] == row["TARGET_BCCH"] else "",
        axis=1,
    )

    # create distance column
    df_adce_final["distance_km"] = df_adce_final.apply(
        lambda row: geodesic(
            (row["SOURCE_Latitude"], row["SOURCE_Longitude"]),
            (row["TARGET_Latitude"], row["TARGET_Longitude"]),
        ).kilometers,
        axis=1,
    )

    # create final adce
    df_adce_final = df_adce_final[
        [
            "SOURCE_ID",
            "SOURCE_NAME",
            "SOURCE_Longitude",
            "SOURCE_Latitude",
            "TARGET_LAC_ID",
            "TARGET_NAME",
            "TARGET_Longitude",
            "TARGET_Latitude",
            "SYMETRIQUE",
            "synchronized",
            "Sync Comment",
            "Same BSIC",
            "Same BCCH",
            "distance_km",
        ]
    ]

    # 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 = process_wcdma_data(file_path)

    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", "Longitude", "Latitude"]]

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

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

    # 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
    )
    df_adjs_final = check_symmetry(df_adjs_final)
    # create distance column
    df_adjs_final["distance_km"] = df_adjs_final.apply(
        lambda row: geodesic(
            (row["Latitude_x"], row["Longitude_x"]),
            (row["Latitude_y"], row["Longitude_y"]),
        ).kilometers,
        axis=1,
    )

    # 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
    )
    df_adji_final = check_symmetry(df_adji_final)
    # create distance column
    df_adji_final["distance_km"] = df_adji_final.apply(
        lambda row: geodesic(
            (row["Latitude_x"], row["Longitude_x"]),
            (row["Latitude_y"], row["Longitude_y"]),
        ).kilometers,
        axis=1,
    )

    # 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
    )
    # create distance column
    df_adjg_final["distance_km"] = df_adjg_final.apply(
        lambda row: geodesic(
            (row["Latitude_x"], row["Longitude_x"]),
            (row["Latitude_y"], row["Longitude_y"]),
        ).kilometers,
        axis=1,
    )

    # 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
    )
    df_adjw_final = check_symmetry(df_adjw_final)
    # create distance column
    df_adjw_final["distance_km"] = df_adjw_final.apply(
        lambda row: geodesic(
            (row["Latitude_x"], row["Longitude_x"]),
            (row["Latitude_y"], row["Longitude_y"]),
        ).kilometers,
        axis=1,
    )

    # 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"]
    )