import pandas as pd from utils.convert_to_excel import convert_invunit_dfs, save_dataframe from utils.dump_excel import read_dump_excel 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, } INVENTORY_UNIT_TYPE_RENAME_MAP = { "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", } INVUNIT_NUMBER_COLUMNS = [ "MRBTS", "inventoryUnitType", "vendorUnitTypeNumber", "serialNumber", ] INVUNIT_NUMBER_OUTPUT_COLUMNS = [ "MRBTS", "name", "inventoryUnitType", "vendorUnitTypeNumber", "serialNumber", ] def ensure_columns(df: pd.DataFrame, columns: list[str], default_value=0) -> pd.DataFrame: """ Ensure requested columns exist in the DataFrame. Missing columns are created with a default value. """ for col in columns: if col not in df.columns: df[col] = default_value return df 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 build_invunit_number_dataframe(file_path: str) -> pd.DataFrame: """ Build detailed INVUNIT_NUMBER dataframe from dump INVUNIT sheet. """ dfs = read_dump_excel( file_path, sheet_name=["INVUNIT", "MRBTS"], expected_columns=[ "MRBTS", "inventoryUnitType", "vendorUnitTypeNumber", "serialNumber", ], ) df_invunit = dfs["INVUNIT"].copy() df_invunit.columns = df_invunit.columns.str.replace(r"[ ]", "", regex=True) missing_columns = [col for col in INVUNIT_NUMBER_COLUMNS if col not in df_invunit] if missing_columns: raise ValueError( "INVUNIT sheet is missing required columns for INVUNIT_NUMBER: " f"{missing_columns}" ) df_invunit_number = df_invunit[INVUNIT_NUMBER_COLUMNS].copy() for col in INVUNIT_NUMBER_COLUMNS: df_invunit_number[col] = df_invunit_number[col].astype("string").str.strip() df_invunit_number["inventoryUnitType"] = df_invunit_number[ "inventoryUnitType" ].replace(INVENTORY_UNIT_TYPE_RENAME_MAP) text_cols = ["inventoryUnitType", "vendorUnitTypeNumber", "serialNumber"] for col in text_cols: df_invunit_number[col] = df_invunit_number[col].replace("", pd.NA) df_invunit_number["MRBTS"] = df_invunit_number["MRBTS"].replace("", pd.NA) df_invunit_number = df_invunit_number[ df_invunit_number["MRBTS"].str.fullmatch(r"\d+", na=False) ].copy() df_invunit_number = df_invunit_number[ df_invunit_number[["vendorUnitTypeNumber", "serialNumber"]] .notna() .any(axis=1) ].copy() df_mrbts = dfs["MRBTS"].copy() df_mrbts.columns = df_mrbts.columns.str.replace(r"[ ]", "", regex=True) if "MRBTS" not in df_mrbts.columns: raise ValueError("MRBTS sheet is missing required column: MRBTS") if "name" not in df_mrbts.columns: df_mrbts["name"] = pd.NA df_mrbts_lookup = df_mrbts[["MRBTS", "name"]].copy() df_mrbts_lookup["MRBTS"] = df_mrbts_lookup["MRBTS"].astype("string").str.strip() df_mrbts_lookup["name"] = df_mrbts_lookup["name"].astype("string").str.strip() df_mrbts_lookup = df_mrbts_lookup[ df_mrbts_lookup["MRBTS"].str.fullmatch(r"\d+", na=False) ].copy() df_mrbts_lookup = df_mrbts_lookup.drop_duplicates(subset=["MRBTS"], keep="first") df_invunit_number = df_invunit_number.merge( df_mrbts_lookup, on="MRBTS", how="left" ) df_invunit_number = df_invunit_number[INVUNIT_NUMBER_OUTPUT_COLUMNS] df_invunit_number = df_invunit_number.sort_values( by=INVUNIT_NUMBER_COLUMNS, kind="stable", na_position="last" ).reset_index(drop=True) return df_invunit_number def process_invunit_number_data(file_path: str) -> pd.DataFrame: """ Process and append INVUNIT_NUMBER dataframe to All DB buffers. """ df_invunit_number = build_invunit_number_dataframe(file_path) UtilsVars.all_db_dfs.append(df_invunit_number) UtilsVars.all_db_dfs_names.append("INVUNIT_NUMBER") return df_invunit_number 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 = read_dump_excel( file_path, sheet_name=["INVUNIT"], expected_columns=["MRBTS", "inventoryUnitType"], ) # 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=INVENTORY_UNIT_TYPE_RENAME_MAP) df_invunit = create_invunit_summary(df_invunit) df_invunit["code"] = df_invunit["MRBTS"].apply(extract_code_from_mrbts) # Some inventory unit types may be absent in a file; create missing columns to avoid KeyError. df_invunit = ensure_columns(df_invunit, list(BB_UNIT_VALUES.keys()) + RF_UNIT, 0) # 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() ) # Ensure final ordered columns always exist. df_invunit = ensure_columns(df_invunit, BB_UNIT_LIST + RF_UNIT, 0) # 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"])