|
import pandas as pd |
|
|
|
from queries.process_small_bts import process_small_bts_data |
|
from utils.convert_to_excel import convert_dfs |
|
from utils.utils_vars import UtilsVars |
|
|
|
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): |
|
""" |
|
Process data from the specified file path. |
|
|
|
Args: |
|
file_path (str): The path to the file. |
|
""" |
|
|
|
df_mal = pd.read_excel( |
|
file_path, |
|
sheet_name="MAL", |
|
engine="calamine", |
|
skiprows=[0], |
|
) |
|
|
|
df_mal.columns = df_mal.columns.str.replace(r"[ ]", "", regex=True) |
|
|
|
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] |
|
|
|
|
|
|
|
return df_mal |
|
|
|
|
|
def process_mal_with_bts_name(file_path: str): |
|
""" |
|
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. |
|
""" |
|
mal_df = process_mal_data(file_path=file_path) |
|
df_bts = process_small_bts_data(file_path=file_path) |
|
|
|
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] |
|
|
|
|
|
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"]) |
|
|