queries/process_lte.py

import numpy as np
import pandas as pd

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

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_FDD_COLUMNS = [
    "ID_LNCEL",
    "dlChBw",
    "dlMimoMode",
    "dlRsBoost",
    "earfcnDL",
    "earfcnUL",
    "prachCS",
    "rootSeqIndex",
    "ulChBw",
]

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


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 = pd.read_excel(
        file_path,
        sheet_name=["LNCEL", "LNBTS", "LNCEL_FDD", "LNCEL_TDD"],
        engine="calamine",
        skiprows=[0],
    )

    # 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["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")
    df_lncel = df_lncel[LNCEL_COLUMNS]

    # create band dataframe
    df_band = config_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_physical_db = UtilsVars.physisal_db
    df_physical_db = df_physical_db[
        ["Code_Sector", "Azimut", "Longitude", "Latitude", "Hauteur"]
    ]
    df_lncel_lnbts = pd.merge(
        df_lncel_lnbts, df_physical_db, on="Code_Sector", 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])

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


# process_lte_data(r"data2\20240805_5810_05082024_Dump.xml.gz.xlsb")