Process ADJL - 2G part 1st

queries/process_adjl.py

@@ -0,0 +1,179 @@
+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", "locationAreaIdLAC", "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)
+    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["exists"] = df_bts["ID_BTS"].isin(gsm_adjl_df["ID_BTS"])
+
+    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"]
+    )

utils/config_band.py

@@ -123,3 +123,34 @@ def lte_mrbts_band(df: pd.DataFrame) -> pd.DataFrame:
     df_band.rename(columns={"band": "lte_config_band"}, inplace=True)
 
     return df_band
+
+
+def adjl_band(df: pd.DataFrame, id_col: str, band_col: str) -> pd.DataFrame:
+    """
+    Create a dataframe that contains the adjl configuration band for each adjl ID.
+
+    Parameters
+    ----------
+    df : pd.DataFrame
+        The dataframe containing the adjl information, with columns "ID" and "band"
+
+    Returns
+    -------
+    pd.DataFrame
+        The dataframe containing the adjl configuration band for each adjl ID, with columns "ID" and "adjl_config_band"
+    """
+    df_band = df[[id_col, band_col]].copy()
+    df_band["ID"] = df_band[[id_col, band_col]].astype(str).apply("_".join, axis=1)
+    # remove duplicates ID
+    df_band = df_band.drop_duplicates(subset=["ID"])
+    df_band = df_band[[id_col, band_col]]
+    df_band[band_col] = df_band[band_col].fillna("empty")
+    df_band = (
+        df_band.groupby(id_col)[band_col]
+        .apply(lambda x: "/".join(sorted(x)))
+        .reset_index()
+    )
+    # rename band to config
+    df_band.rename(columns={band_col: "adjl_created_band"}, inplace=True)
+
+    return df_band

utils/utils_vars.py

@@ -71,6 +71,13 @@ class UtilsVars:
         "L2300": 90,
         "L2600": 80,
     }
+    lte_band = {
+        1786: "L1800",
+        6350: "L800",
+        3050: "L2600",
+        38750: "L2300",
+        1761: "L1800",
+    }
     wcdma_band = {
         3004: "U900",
         3006: "U900",
@@ -109,6 +116,7 @@ class UtilsVars:
     gsm_kml_file = None
     wcdma_kml_file = None
     lte_kml_file = None
+    adjl_database = None
     # physisal_db = get_physical_db()