queries/process_lte.py

import numpy as np
import pandas as pd

from utils.config_band import config_band, lte_mrbts_band
from utils.convert_to_excel import convert_dfs, save_dataframe
from utils.dump_excel import read_dump_excel
from utils.kml_creator import generate_kml_from_df
from utils.utils_vars import (
    LteFddAnalysisData,
    LteTddAnalysisData,
    UtilsVars,
    get_band,
    get_physical_db,
)

LNCEL_COLUMNS = [
    "ID_LNBTS",
    "ID_LNCEL",
    "MRBTS",
    "LNBTS",
    "LNCEL",
    "final_name",
    "name",
    "cellName",
    "code",
    "SectorId",
    "Code_Sector",
    "actModulationSchemeDl",
    "actModulationSchemeUL",
    "administrativeState",
    "eutraCelId",
    "lcrId",
    "pMax",
    "phyCellId",
    "tac",
    "Region",
    "band",
    "band_type",
]


LNCEL_MOBILITY_COLUMNS = [
    "ID_LNBTS",
    "ID_LNCEL",
    "MRBTS",
    "LNBTS",
    "LNCEL",
    "final_name",
    "name",
    "cellName",
    "code",
    "SectorId",
    "Code_Sector",
    "administrativeState",
    "lcrId",
    "band",
    "band_type",
    "a3Offset",
    "enableBetterCellHo",
    "enableCovHo",
    "threshold3",
    "threshold3a",
    "threshold4",
    "threshold2InterFreq",
    "threshold2Wcdma",
    "threshold2a",
    "threshold1",
    "hysThreshold2InterFreq",
    "hysThreshold2Wcdma",
    "hysThreshold2a",
    "hysThreshold3",
    "hysThreshold4",
]


LNCEL_FDD_COLUMNS = [
    "ID_LNCEL",
    "dlChBw",
    "dlMimoMode",
    "dlRsBoost",
    "earfcnDL",
    "earfcnUL",
    "prachCS",
    "rootSeqIndex",
    "ulChBw",
]

LNCEL_TDD_COLUMNS = [
    "ID_LNCEL",
    "chBw",
    "dlMimoMode",
    "dlRsBoost",
    "earfcn",
    "prachCS",
    "rootSeqIndex",
]

LTE_KML_COLUMNS = [
    "code",
    "final_name",
    "Longitude",
    "Latitude",
    "Azimut",
    "Hauteur",
    "lcrId",
    "pMax",
    "phyCellId",
    "tac",
    "rootSeqIndex",
    "band",
]


def process_lncel(file_path: str):
    """
    Process data from the specified file path.

    Args:
        file_path (str): The path to the file.
    """
    # Read excel sheets into dataframes
    dfs = read_dump_excel(
        file_path,
        sheet_name=["LNCEL"],
        expected_columns=["MRBTS", "LNBTS", "LNCEL", "name", "cellName"],
    )

    # Process LNCEL data
    df_lncel = dfs["LNCEL"]
    df_lncel.columns = df_lncel.columns.str.replace(r"[ ]", "", regex=True)
    df_lncel["final_name"] = df_lncel["name"].fillna(df_lncel["cellName"])
    df_lncel["code"] = df_lncel["final_name"].str.split("_").str[0]
    df_lncel["code"] = (
        pd.to_numeric(df_lncel["code"], errors="coerce").fillna(0).astype(int)
    )
    df_lncel["SectorId"] = (
        df_lncel["lcrId"].map(UtilsVars.sector_mapping).fillna(df_lncel["lcrId"])
    )
    df_lncel["Code_Sector"] = (
        df_lncel[["code", "SectorId"]]
        .astype(str)
        .apply("_".join, axis=1)
        .str.replace(".0", "")
        .str.lstrip("0")
    )
    df_lncel["ID_LNCEL"] = (
        df_lncel[["MRBTS", "LNBTS", "LNCEL"]].astype(str).apply("_".join, axis=1)
    )
    df_lncel["ID_LNBTS"] = (
        df_lncel[["MRBTS", "LNBTS"]].astype(str).apply("_".join, axis=1)
    )
    df_lncel["Region"] = df_lncel["final_name"].str.split("_").str[1]
    df_lncel["band"] = df_lncel["final_name"].apply(get_band)
    df_lncel["band_type"] = np.where(df_lncel["band"] == "L2300", "TDD", "FDD")

    return df_lncel


def process_lte_data(file_path: str):
    """
    Process data from the specified file path.

    Args:
        file_path (str): The path to the file.
    """
    # Read excel sheets into dataframes
    dfs = read_dump_excel(
        file_path,
        sheet_name=["LNBTS", "LNCEL_FDD", "LNCEL_TDD"],
        expected_columns=["MRBTS", "LNBTS", "name"],
    )

    # Get LNCEL data
    df_lncel = process_lncel(file_path)
    df_lncel = df_lncel[LNCEL_COLUMNS]

    # create band dataframe
    df_band = config_band(df_lncel)
    df_mrbts_band = lte_mrbts_band(df_lncel)

    # Process LNBTS data
    df_lnbts = dfs["LNBTS"]
    df_lnbts.columns = df_lnbts.columns.str.replace(r"[ ]", "", regex=True)
    df_lnbts["ID_LNBTS"] = (
        df_lnbts[["MRBTS", "LNBTS"]].astype(str).apply("_".join, axis=1)
    )
    df_lnbts.rename(columns={"name": "lnbts_name"}, inplace=True)
    df_lnbts = df_lnbts[["ID_LNBTS", "lnbts_name"]]

    # Merge dataframes
    df_lncel_lnbts = pd.merge(df_lncel, df_lnbts, on="ID_LNBTS", how="left")
    df_lncel_lnbts = pd.merge(df_lncel_lnbts, df_band, on="code", how="left")
    df_lncel_lnbts = pd.merge(df_lncel_lnbts, df_mrbts_band, on="MRBTS", how="left")

    df_physical_db = get_physical_db()
    df_lncel_lnbts = pd.merge(
        df_lncel_lnbts, df_physical_db, on="Code_Sector", how="left"
    )

    # Add Number of cells per band on MRBTS
    df_id_mrbts_band = df_lncel_lnbts[["MRBTS", "band"]]
    df_id_mrbts_band_grouped = (
        df_id_mrbts_band.groupby(["MRBTS", "band"])
        .size()
        .unstack(fill_value=0)
        .rename(columns=lambda x: f"Number of {x} cells on MRBTS")
        .reset_index()
    )

    df_lncel_lnbts = pd.merge(
        df_lncel_lnbts, df_id_mrbts_band_grouped, on="MRBTS", how="left"
    )

    # Process LNCEL_FDD and LNCEL_TDD data
    df_lncel_fdd = dfs["LNCEL_FDD"]
    df_lncel_fdd.columns = df_lncel_fdd.columns.str.replace(r"[ ]", "", regex=True)
    df_lncel_fdd["ID_LNCEL"] = (
        df_lncel_fdd[["MRBTS", "LNBTS", "LNCEL"]].astype(str).apply("_".join, axis=1)
    )

    df_lncel_fdd = df_lncel_fdd[LNCEL_FDD_COLUMNS]

    df_lncel_tdd = dfs["LNCEL_TDD"]
    df_lncel_tdd.columns = df_lncel_tdd.columns.str.replace(r"[ ]", "", regex=True)
    df_lncel_tdd["ID_LNCEL"] = (
        df_lncel_tdd[["MRBTS", "LNBTS", "LNCEL"]].astype(str).apply("_".join, axis=1)
    )
    df_lncel_tdd = df_lncel_tdd[LNCEL_TDD_COLUMNS]

    # Create df_fdd and df_tdd base on "band"
    df_fdd = df_lncel_lnbts[df_lncel_lnbts["band"] != "L2300"]
    df_tdd = df_lncel_lnbts[df_lncel_lnbts["band"] == "L2300"]

    df_fdd_final = pd.merge(df_fdd, df_lncel_fdd, on="ID_LNCEL", how="left")
    df_tdd_final = pd.merge(df_tdd, df_lncel_tdd, on="ID_LNCEL", how="left")

    # Save dataframes
    # save_dataframe(df_fdd_final, "fdd")
    # save_dataframe(df_tdd_final, "tdd")
    UtilsVars.all_db_dfs.extend([df_fdd_final, df_tdd_final])
    UtilsVars.lte_dfs.extend([df_fdd_final, df_tdd_final])
    UtilsVars.all_db_dfs_names.extend(["LTE_FDD", "LTE_TDD"])

    return [df_fdd_final, df_tdd_final]
    # add the fdd and tdd to the list

    # UtilsVars.final_lte_database = [df_fdd_final, df_tdd_final]


def process_lte_data_to_excel(file_path: str):
    lte_dfs = process_lte_data(file_path)
    UtilsVars.final_lte_database = convert_dfs(lte_dfs, ["LTE_FDD", "LTE_TDD"])


############################# KML CREATION #################################
def process_lte_data_to_kml(file_path: str):
    lte_kml_dfs = process_lte_data(file_path)

    lte_fdd_klm_df = lte_kml_dfs[0]
    lte_fdd_klm_df = lte_fdd_klm_df[LTE_KML_COLUMNS]

    lte_tdd_klm_df = lte_kml_dfs[1]
    lte_tdd_klm_df = lte_tdd_klm_df[LTE_KML_COLUMNS]

    # Merge FDD and TDD dataframes
    lte_kml_df = pd.concat([lte_fdd_klm_df, lte_tdd_klm_df], ignore_index=True)

    # Rename "final_name" to "name"
    lte_kml_df.rename(columns={"final_name": "name"}, inplace=True)
    # Add colors column base on "band" column
    lte_kml_df["color"] = lte_kml_df["band"].map(UtilsVars.color_mapping)
    # Add size column base on "band" column
    lte_kml_df["size"] = lte_kml_df["band"].map(UtilsVars.size_mapping)
    # Remove empty rows
    lte_kml_df = lte_kml_df.dropna(subset=["Longitude", "Latitude", "Azimut"])
    # Generate kml
    UtilsVars.lte_kml_file = generate_kml_from_df(lte_kml_df)


#############################LTE ANALYSIS#################################


def lte_fdd_analaysis(file_path: str):
    # df_fdd = process_lte_data(file_path)[0]
    df_fdd: pd.DataFrame = UtilsVars.lte_dfs[0]

    LteFddAnalysisData.total_number_of_lncel = len(df_fdd["ID_LNCEL"].unique())
    LteFddAnalysisData.total_number_of_site = len(df_fdd["code"].unique())
    LteFddAnalysisData.number_of_empty_lncel_name = df_fdd["name"].isna().sum()
    LteFddAnalysisData.number_of_empty_lncel_cellname = df_fdd["cellName"].isna().sum()
    LteFddAnalysisData.number_of_empty_lnbts_name = df_fdd["lnbts_name"].isna().sum()
    LteFddAnalysisData.number_of_cell_per_band = df_fdd["band"].value_counts()
    LteFddAnalysisData.phycellid_distribution = df_fdd["phyCellId"].value_counts()
    LteFddAnalysisData.rootsequenceindex_distribution = df_fdd[
        "rootSeqIndex"
    ].value_counts()
    LteFddAnalysisData.lncel_administate_distribution = df_fdd[
        "administrativeState"
    ].value_counts()
    LteFddAnalysisData.number_of_cell_per_tac = df_fdd["tac"].value_counts()


def lte_tdd_analaysis(file_path: str):
    # df_tdd = process_lte_data(file_path)[1]
    df_tdd: pd.DataFrame = UtilsVars.lte_dfs[1]

    LteTddAnalysisData.total_number_of_lncel = len(df_tdd["ID_LNCEL"].unique())
    LteTddAnalysisData.total_number_of_site = len(df_tdd["code"].unique())
    LteTddAnalysisData.number_of_empty_lncel_name = df_tdd["name"].isna().sum()
    LteTddAnalysisData.number_of_empty_lncel_cellname = df_tdd["cellName"].isna().sum()
    LteTddAnalysisData.number_of_empty_lnbts_name = df_tdd["lnbts_name"].isna().sum()
    LteTddAnalysisData.number_of_cell_per_band = df_tdd["band"].value_counts()
    LteTddAnalysisData.phycellid_distribution = df_tdd["phyCellId"].value_counts()
    LteTddAnalysisData.rootsequenceindex_distribution = df_tdd[
        "rootSeqIndex"
    ].value_counts()
    LteTddAnalysisData.lncel_administate_distribution = df_tdd[
        "administrativeState"
    ].value_counts()
    LteTddAnalysisData.number_of_cell_per_tac = df_tdd["tac"].value_counts()