queries/process_invunit.py

import pandas as pd

from utils.convert_to_excel import convert_invunit_dfs, save_dataframe
from utils.extract_code import extract_code_from_mrbts
from utils.utils_vars import UtilsVars

RF_UNIT = [
    "AHDA",
    "AHEGB",
    "AHEGC",
    "AHEGHA",
    "AHGA",
    "AHMA",
    "AHPMDA",
    "AHPMDG",
    "AHPMDI",
    "ARDA",
    "AREA",
    "ARGA",
    "ARMA",
    "AZNA",
    "FHDB",
    "FHEB",
    "FHEL",
    "FRGU",
    "FRGY",
    "FRMB",
    "FRMF",
    "FXDB",
    "FXED",
    "FZNI",
]

BB_UNIT_LIST = [
    "FBBA",
    "FBBC",
    "FSMF",
    "ABIA",
    "total_number_of_subunit",
]

BB_UNIT_VALUES = {
    "FBBA": 6,
    "FBBC": 6,
    "FSMF": 5.5,
    "ABIA": 8,
}


def create_invunit_summary(df: pd.DataFrame) -> pd.DataFrame:
    """
    Creates a summary string column in the given DataFrame by concatenating non-NaN values of all columns except the first one (MRBTS) into a single string with '/' as separator.

    Args:
        df (pd.DataFrame): The DataFrame to process.

    Returns:
        pd.DataFrame: The DataFrame with the added "invunit_summary" column.
    """

    def process_row(row):
        values = []
        for col in df.columns[1:]:  # Exclude 'MRBTS'
            if pd.notna(row[col]):  # Check if value is not NaN
                values.append(f"{int(row[col])} {col}")  # Format as 'count column_name'
        return "/".join(values) if values else ""

    df["invunit_summary"] = df.apply(process_row, axis=1)
    return df


def process_invunit_data(file_path: str) -> pd.DataFrame:
    """
    Process data from the specified file path.

    Args:
        file_path (str): The path to the file.
    """
    dfs = pd.read_excel(
        file_path,
        sheet_name=["INVUNIT"],
        engine="calamine",
        skiprows=[0],
    )
    # Parse INVUNIT
    df_invunit = dfs["INVUNIT"]
    df_invunit.columns = df_invunit.columns.str.replace(r"[ ]", "", regex=True)

    df_invunit = df_invunit[df_invunit["MRBTS"].apply(lambda x: str(x).isnumeric())]
    df_invunit["code"] = df_invunit["MRBTS"].apply(extract_code_from_mrbts)
    df_invunit = df_invunit[["MRBTS", "inventoryUnitType"]]

    df_invunit = (
        df_invunit.groupby(["MRBTS", "inventoryUnitType"])
        .size()
        .unstack(fill_value=None)
        .reset_index()
    )
    # Rename columns
    df_invunit = df_invunit.rename(
        columns={
            "ABIA AirScale Capacity": "ABIA",
            "AMIA AirScale Indoor Subrack": "AMIA",
            "AMOB AirScale Outdoor Subrack": "AMOB",
            "ASIA AirScale Common": "ASIA",
            "ASIB AirScale Common": "ASIB",
            "BB Extension Outdoor Sub-Module FBBA": "FBBA",
            "CORE_ASIA AirScale Common": "CORE_ASIA",
            "CORE_ASIB AirScale Common": "CORE_ASIB",
            "CORE_Flexi System Module Outdoor FSMF": "CORE_FSMF",
            "CORE_SMOD": "CORE_SMOD",
            "Flexi Baseband Sub-Module FBBC": "FBBC",
            "Flexi System Module Outdoor FSMF": "FSMF",
            "Not available": "NOT_AVAILABLE",
            "SingleAntennaDevice": "SAD",
        }
    )
    df_invunit = create_invunit_summary(df_invunit)
    df_invunit["code"] = df_invunit["MRBTS"].apply(extract_code_from_mrbts)

    # Compute total_number_of_subunit
    df_invunit["total_number_of_subunit"] = sum(
        df_invunit[col].fillna(0) * val for col, val in BB_UNIT_VALUES.items()
    )

    # Start order with "MRBTS", "code", "invunit_summary" follow by bb_unit  , rf_unit and then all other columns
    df_invunit = df_invunit[
        ["MRBTS", "code", "invunit_summary"]
        + BB_UNIT_LIST
        + RF_UNIT
        + df_invunit.columns.difference(
            ["MRBTS", "code", "invunit_summary"] + BB_UNIT_LIST + RF_UNIT
        ).tolist()
    ]

    UtilsVars.all_db_dfs.append(df_invunit)
    UtilsVars.all_db_dfs_names.append("INVUNIT")
    return df_invunit


def process_invunit_data_to_excel(file_path: str) -> None:
    """
    Process data from the specified file path and save it to a excel file.

    Args:
        file_path (str): The path to the file.
    """
    invunit_df = process_invunit_data(file_path)
    UtilsVars.final_invunit_database = convert_invunit_dfs([invunit_df], ["INVUNIT"])