import pandas as pd from queries.process_small_bts import process_small_bts_data from utils.convert_to_excel import convert_dfs from utils.dump_excel import read_dump_excel from utils.session_state import get_2g_bsc_options from utils.utils_vars import UtilsVars, filter_excluded_2g_bsc MAL_COLUMNS = [ "ID_MAL", "MAL_TCH", "number_mal_tch", ] MAL_BTS_COLUMNS = [ "ID_MAL", "code", "name", "MAL_TCH", "number_mal_tch", ] def process_mal_data( file_path: str, *, exclude_decommissioned_2g_bsc: bool | None = None, decommissioned_2g_bsc_ids: set[int] | frozenset[int] | tuple[int, ...] | None = None, ): """ Process data from the specified file path. Args: file_path (str): The path to the file. """ exclude_2g_bsc, excluded_bsc_ids = get_2g_bsc_options( exclude_decommissioned_2g_bsc=exclude_decommissioned_2g_bsc, decommissioned_2g_bsc_ids=decommissioned_2g_bsc_ids, ) # Read the specific sheet into a DataFrame df_mal = read_dump_excel( file_path, sheet_name="MAL", expected_columns=["BSC", "MAL", "frequency"], ) df_mal.columns = df_mal.columns.str.replace(r"[ ]", "", regex=True) df_mal = filter_excluded_2g_bsc( df_mal, "BSC", exclude_decommissioned_2g_bsc=exclude_2g_bsc, decommissioned_2g_bsc_ids=excluded_bsc_ids, ) df_mal["ID_MAL"] = df_mal[["BSC", "MAL"]].astype(str).apply("_".join, axis=1) df_mal["frequency"] = df_mal["frequency"].str.replace("List;", "") df_mal["MAL_TCH"] = df_mal["frequency"].str.replace(";", ",") df_mal["number_mal_tch"] = df_mal["MAL_TCH"].apply( lambda x: len(str(x).split(",")) if isinstance(x, str) else 0 ) df_mal = df_mal[MAL_COLUMNS] # UtilsVars.all_db_dfs.append(df_mal) # save_dataframe(df_mal, "MAL") return df_mal def process_mal_with_bts_name( file_path: str, mal_df: pd.DataFrame | None = None, df_bts: pd.DataFrame | None = None, *, exclude_decommissioned_2g_bsc: bool | None = None, decommissioned_2g_bsc_ids: set[int] | frozenset[int] | tuple[int, ...] | None = None, ) -> pd.DataFrame: """ Process data from the specified file path and merge it with the BTS data to get the BTS name associated with each MAL. Args: file_path (str): The path to the file. Returns: pd.DataFrame: A DataFrame with the MAL data and the BTS name associated with each MAL. """ exclude_2g_bsc, excluded_bsc_ids = get_2g_bsc_options( exclude_decommissioned_2g_bsc=exclude_decommissioned_2g_bsc, decommissioned_2g_bsc_ids=decommissioned_2g_bsc_ids, ) if mal_df is None: mal_df = process_mal_data( file_path=file_path, exclude_decommissioned_2g_bsc=exclude_2g_bsc, decommissioned_2g_bsc_ids=excluded_bsc_ids, ) if df_bts is None: df_bts = process_small_bts_data( file_path=file_path, exclude_decommissioned_2g_bsc=exclude_2g_bsc, decommissioned_2g_bsc_ids=excluded_bsc_ids, ) df_mal_bts_name = pd.merge(mal_df, df_bts, on="ID_MAL", how="left") df_mal_bts_name = df_mal_bts_name[MAL_BTS_COLUMNS] # UtilsVars.all_db_dfs.append(df_mal_bts_name) return df_mal_bts_name def process_mal_data_to_excel(file_path: str): """ Process data from the specified file path and save it to a excel file. Args: file_path (str): The path to the file. """ mal_df = process_mal_with_bts_name(file_path) UtilsVars.final_mal_database = convert_dfs([mal_df], ["MAL"])