queries/process_invunit.py

import pandas as pd

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


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)
    df_invunit = df_invunit[["MRBTS", "code", "invunit_summary"]]
    # df_invunit = df_invunit.sort_values(
    #     by=["MRBTS", "code"],
    # )
    # df_invunit["MRBTS_NUM"] = df_invunit.groupby("code").cumcount() + 1
    # df_invunit["MRBTS_NUM"] = "MRBTS_" + df_invunit["MRBTS_NUM"].astype(str)

    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_dfs([invunit_df], ["INVUNIT"])