queries/process_adjl.py

import pandas as pd
from geopy.distance import geodesic  # Imported but not used — consider removing

from queries.process_gsm import process_gsm_data
from queries.process_lte import process_lte_data
from queries.process_wcdma import process_wcdma_data
from utils.config_band import adjl_band
from utils.convert_to_excel import convert_dfs, save_dataframe
from utils.utils_vars import UtilsVars

# -------------------------------
# Constants
# -------------------------------
ADJL_GSM_COLUMNS = ["BSC", "BCF", "BTS", "ADJL", "earfcn", "lteAdjCellTac"]

ADJL_WCDMA_COLUMNS = ["RNC", "WBTS", "WCEL", "ADJL", "AdjLEARFCN"]

BTS_COLUMNS = ["ID_BTS", "name", "Code_Sector"]

WCEL_COLUMNS = ["ID_WCEL", "name", "Code_Sector"]

LTE_COLUMNS_CONFIG = ["Code_Sector", "site_config_band"]

LTE_COLUMNS_TAC = ["Code_Sector", "tac", "band"]

LTE_COLUMNS_ADJL = ["Code_Sector", "site_config_band", "tac", "band"]


# -------------------------------
# Helper functions
# -------------------------------
def check_bands(row: pd.Series) -> bool:
    """
    Verify whether all configured site bands exist in ADJL created bands.
    """
    site_bands = (
        set(str(row["site_config_band"]).split("/"))
        if pd.notna(row["site_config_band"])
        else set()
    )
    adjl_bands = (
        set(str(row["adjl_created_band"]).split("/"))
        if pd.notna(row["adjl_created_band"])
        else set()
    )
    return site_bands.issubset(adjl_bands)


def missing_bands(row: pd.Series) -> str | None:
    """
    Return missing bands from ADJL compared to site configuration.
    """
    site_bands = (
        set(str(row["site_config_band"]).split("/"))
        if pd.notna(row["site_config_band"])
        else set()
    )
    adjl_bands = (
        set(str(row["adjl_created_band"]).split("/"))
        if pd.notna(row["adjl_created_band"])
        else set()
    )
    diff = site_bands - adjl_bands
    return ",".join(diff) if diff else None


# -------------------------------
# Main Processing
# -------------------------------
def process_adjl_data(file_path: str) -> list[pd.DataFrame]:
    """
    Process ADJL data from an Excel file and return structured DataFrames.

    Args:
        file_path (str): Path to the input Excel file.

    Returns:
        list[pd.DataFrame]: [GSM_ADJL, WCDMA_ADJL, BTS, WCEL, LTE]
    """
    # Read Excel sheets
    dfs = pd.read_excel(
        file_path,
        sheet_name=["ADJL", "BTS", "WCEL"],
        engine="calamine",
        skiprows=[0],
    )

    # ------------------- BTS -------------------
    df_bts = process_gsm_data(file_path)[BTS_COLUMNS]

    # ------------------- WCEL -------------------
    df_wcel = process_wcdma_data(file_path)[WCEL_COLUMNS]
    # df_wcel["ID_WCEL"] = (
    #     df_wcel[["RNC", "WBTS", "WCEL"]].astype(str).agg("_".join, axis=1)
    # )

    # ------------------- LTE -------------------
    lte_fdd_df, lte_tdd_df = process_lte_data(file_path)
    lte_tdd_df = lte_tdd_df.rename(columns={"earfcn": "earfcnDL"})
    lte_df = pd.concat([lte_fdd_df, lte_tdd_df], ignore_index=True)[LTE_COLUMNS_ADJL]

    # Config & TAC references
    lte_df_config = lte_df[LTE_COLUMNS_CONFIG]
    lte_df_global_tac = (
        lte_df[["Code_Sector", "tac"]]
        .drop_duplicates(subset=["Code_Sector"], keep="first")
        .rename(columns={"tac": "global_tac"})
    )

    lte_df_band_tac = lte_df[LTE_COLUMNS_TAC].copy()
    lte_df_band_tac["Code_Sector_band"] = (
        lte_df_band_tac[["Code_Sector", "band"]].astype(str).agg("_".join, axis=1)
    )
    lte_df_band_tac = lte_df_band_tac.drop(columns=["Code_Sector"])

    # ------------------- ADJL -------------------
    df_adjl = dfs["ADJL"]
    df_adjl.columns = df_adjl.columns.str.replace(r"[ ]", "", regex=True)

    gsm_adjl_df = df_adjl[ADJL_GSM_COLUMNS]
    wcdma_adjl_df = df_adjl[ADJL_WCDMA_COLUMNS]

    # --- GSM ADJL ---
    # Filter invalid rows
    gsm_adjl_df = gsm_adjl_df[
        gsm_adjl_df["BSC"].notna()
        & gsm_adjl_df["BCF"].notna()
        & gsm_adjl_df["BTS"].notna()
    ].reset_index(drop=True)

    # Build IDs and bands
    gsm_adjl_df["ID_BTS"] = (
        gsm_adjl_df[["BSC", "BCF", "BTS"]].astype(str).agg("_".join, axis=1)
    )
    gsm_adjl_df["ID_BTS"] = gsm_adjl_df["ID_BTS"].str.replace(".0", "", regex=False)
    gsm_adjl_df["adjl_band"] = gsm_adjl_df["earfcn"].map(UtilsVars.lte_band)

    # Merge BTS info
    gsm_adjl_df = pd.merge(gsm_adjl_df, df_bts, on="ID_BTS", how="left")

    # Aggregate ADJL band info
    gsm_adjl_df_band = adjl_band(gsm_adjl_df, "ID_BTS", "adjl_band")
    gsm_adjl_df = pd.merge(gsm_adjl_df, gsm_adjl_df_band, on="ID_BTS", how="left")

    # Build Code_Sector_band
    gsm_adjl_df["Code_Sector_band"] = (
        gsm_adjl_df[["Code_Sector", "adjl_band"]].astype(str).agg("_".join, axis=1)
    )

    # Merge LTE references
    gsm_adjl_df = gsm_adjl_df.merge(lte_df_config, on="Code_Sector", how="left")
    gsm_adjl_df = gsm_adjl_df.merge(lte_df_band_tac, on="Code_Sector_band", how="left")
    gsm_adjl_df = gsm_adjl_df.merge(lte_df_global_tac, on="Code_Sector", how="left")

    # Final TAC
    gsm_adjl_df["final_tac"] = gsm_adjl_df["tac"].fillna(gsm_adjl_df["global_tac"])

    # Validations
    gsm_adjl_df["check_bands"] = gsm_adjl_df.apply(check_bands, axis=1)
    gsm_adjl_df["missing_bands"] = gsm_adjl_df.apply(missing_bands, axis=1)
    gsm_adjl_df["check_tac"] = gsm_adjl_df["lteAdjCellTac"] == gsm_adjl_df["final_tac"]

    # Drop intermediate columns
    gsm_adjl_df = gsm_adjl_df.drop(
        columns=["Code_Sector_band", "tac", "band", "global_tac"]
    )

    # Mark existing BTS
    df_bts["adjl_exists"] = df_bts["ID_BTS"].isin(gsm_adjl_df["ID_BTS"])

    # --- WCDMA ADJL ---
    # Filter invalid rows
    wcdma_adjl_df = wcdma_adjl_df[
        wcdma_adjl_df["RNC"].notna()
        & wcdma_adjl_df["WBTS"].notna()
        & wcdma_adjl_df["WCEL"].notna()
    ].reset_index(drop=True)

    # Build IDs and bands
    wcdma_adjl_df["ID_WCEL"] = (
        wcdma_adjl_df[["RNC", "WBTS", "WCEL"]].astype(str).agg("_".join, axis=1)
    )
    wcdma_adjl_df["ID_WCEL"] = wcdma_adjl_df["ID_WCEL"].str.replace(
        ".0", "", regex=False
    )
    wcdma_adjl_df["adjl_band"] = wcdma_adjl_df["AdjLEARFCN"].map(UtilsVars.lte_band)

    # Merge WCEL info
    wcdma_adjl_df = pd.merge(wcdma_adjl_df, df_wcel, on="ID_WCEL", how="left")

    # Aggregate ADJL band info
    wcdma_adjl_df_band = adjl_band(wcdma_adjl_df, "ID_WCEL", "adjl_band")
    wcdma_adjl_df = pd.merge(
        wcdma_adjl_df, wcdma_adjl_df_band, on="ID_WCEL", how="left"
    )

    # Build Code_Sector_band
    wcdma_adjl_df["Code_Sector_band"] = (
        wcdma_adjl_df[["Code_Sector", "adjl_band"]].astype(str).agg("_".join, axis=1)
    )

    # Merge LTE references
    wcdma_adjl_df = wcdma_adjl_df.merge(lte_df_config, on="Code_Sector", how="left")

    # Validations
    wcdma_adjl_df["check_bands"] = wcdma_adjl_df.apply(check_bands, axis=1)
    wcdma_adjl_df["missing_bands"] = wcdma_adjl_df.apply(missing_bands, axis=1)

    # Mark existing WCEL
    df_wcel["adjl_exists"] = df_wcel["ID_WCEL"].isin(wcdma_adjl_df["ID_WCEL"])

    return [gsm_adjl_df, wcdma_adjl_df, df_bts, df_wcel, lte_df]


def process_adjl_data_to_excel(file_path: str) -> None:
    """
    Process ADJL data and save the result into an Excel-like format via UtilsVars.
    """
    adjl_dfs = process_adjl_data(file_path)
    UtilsVars.adjl_database = convert_dfs(
        adjl_dfs, ["GSM_ADJL", "WCDMA_ADJL", "BTS", "WCEL", "LTE"]
    )