GSM CAPACITY initial commit

apps/kpi_analysis/gsm_capacity.py

@@ -0,0 +1,118 @@
+import pandas as pd
+import plotly.express as px
+import streamlit as st
+
+from process_kpi.process_gsm_capacity import GsmCapacity, analyze_gsm_data
+from utils.convert_to_excel import (  # Import convert_dfs from the appropriate module
+    convert_dfs,
+    convert_gsm_dfs,
+)
+
+st.title(" 📊 GSM Capacity Analysis")
+doc_col, image_col = st.columns(2)
+
+with doc_col:
+    st.write(
+        """
+        The report should be run with a minimum of 3 days of data.
+        - Daily Aggregated
+        - Site level
+        - Exported in CSV format.
+        """
+    )
+
+with image_col:
+    st.image("./assets/gsm_capacity.png", width=250)
+
+file1, file2, file3 = st.columns(3)
+
+with file1:
+    uploaded_dump = st.file_uploader("Upload Dump file in xlsb format", type="xlsb")
+with file2:
+    uploaded_daily_report = st.file_uploader(
+        "Upload Daily Report in CSV format", type="csv"
+    )
+with file3:
+    uploaded_bh_report = st.file_uploader(
+        "Upload Busy Hour Report in CSV format", type="csv"
+    )
+
+
+col1, col2 = st.columns(2)
+
+threshold_col1, threshold_col2 = st.columns(2)
+threshold_col3, threshold_col4 = st.columns(2)
+
+if (
+    uploaded_dump is not None
+    and uploaded_daily_report is not None
+    and uploaded_bh_report is not None
+):
+    # WbtsCapacity.final_results = None
+    with col1:
+        number_of_kpi_days = st.number_input(
+            "Number of days for analysis",
+            min_value=3,
+            max_value=30,
+            value=7,
+        )
+    with col2:
+        number_of_threshold_days = st.number_input(
+            "Number of days for threshold",
+            min_value=1,
+            max_value=30,
+            value=3,
+        )
+
+    with threshold_col1:
+        availability_threshold = st.number_input(
+            "Availability Threshold", min_value=1, max_value=100, value=95
+        )
+    with threshold_col2:
+        tch_abis_fails_threshold = st.number_input(
+            "TCH ABIS Fails Threshold", min_value=0, value=10
+        )
+    with threshold_col3:
+        sddch_blocking_threshold = st.number_input(
+            "SDDCH Blocking Threshold", min_value=0.1, value=0.5
+        )
+    with threshold_col4:
+        tch_blocking_threshold = st.number_input(
+            "TCH Blocking Threshold", min_value=0.1, value=0.5
+        )
+
+    if st.button("Analyze Data", type="primary"):
+        dfs = analyze_gsm_data(
+            dump_path=uploaded_dump,
+            daily_report_path=uploaded_daily_report,
+            bh_report_path=uploaded_bh_report,
+            number_of_kpi_days=number_of_kpi_days,
+            number_of_threshold_days=number_of_threshold_days,
+            availability_threshold=availability_threshold,
+            tch_abis_fails_threshold=tch_abis_fails_threshold,
+            sddch_blocking_threshold=sddch_blocking_threshold,
+            tch_blocking_threshold=tch_blocking_threshold,
+        )
+
+        if dfs is not None:
+            gsm_analysis_df = dfs[0]
+            bh_kpi_df = dfs[1]
+            GsmCapacity.final_results = convert_gsm_dfs(
+                [gsm_analysis_df, bh_kpi_df], ["GSM_Analysis", "BH_KPI_Analysis"]
+            )
+
+            # GsmCapacity.final_results = convert_gsm_dfs(
+            #     [gsm_analysis_df], ["GSM_Analysis"]
+            # )
+
+            if GsmCapacity.final_results is not None:
+                st.download_button(
+                    on_click="ignore",
+                    type="primary",
+                    label="Download the Analysis Report",
+                    data=GsmCapacity.final_results,
+                    file_name="GSM_Analysis_Report.xlsx",
+                    mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+                )
+
+        st.write(gsm_analysis_df)

process_kpi/gsm_kpi_requirements.md

@@ -0,0 +1,47 @@
+# Required Input
+
+- BH report
+- Daily Report
+- Dump file (2G dump)
+- Number of last day for the analysis
+- Number of days for blocking
+- Sddch blocking threshold
+- TCH blocking threshold
+- Availability threshold
+- TCH abis fails threshold
+
+Analyse
+
+DUMP
+
+- Check that mandatory sheet exists in the dump
+- Parse 2G databases
+- Get number of TRX,TCH,SDCCH,amrSegLoadDepTchRateLower,amrSegLoadDepTchRateUpper from databases
+- Add "GPRS" colomn equal to (dedicatedGPRScapacity * number_tch_per_cell)/100
+- Get "Coef HF rate" by mapping "amrSegLoadDepTchRateLower" to 2G analysis_utility "hf_rate_coef" dict
+- "TCH Actual HR%" equal to "number of TCH" multiplyed by "Coef HF rate"
+- Get "Offered Traffic" by mapping approximate "TCH Actual HR%" to 2G analysis_utility "erlangB" dict
+
+BH DATA
+
+- Pivot KPI in BH report
+- Calculate Average and Max of Traffic
+- Average of TCH blocking
+- Average of SDCCH blocking
+- Count number of Days with TCH blocking exceeded TCH blocking threshold
+- Count number of Days with SDCCH blocking exceeded Sddch blocking threshold
+- Count number of Days with Availability below Availability threshold
+- "TCH UTILIZATION (@Max Traffic)" equal to "Max_Trafic" divided by "offered Traffic"
+- Add "ErlabngB_value" =MAX TRAFFIC/(1-(MAX TCH call blocking/200))
+- Get "Target FR CHs" by mapping "ERLANG value" to 2G analysis_utility "erlangB" dict
+- "Target HR CHs" equal to  "Target FR CHs" * 2
+- Get "Signal" and "GPRS" value from databases
+- Target TCHs equal to Target HR CHs + Signal + GPRS + SDCCH
+- "Target TRXs" equal to roundup(Target TCHs/8)
+- "# of required TRXs" equal to difference between "Target TRXs" and "number of TRX"
+
+Daily DATA
+
+- Pivot KPI in Daily Report
+- Count number of Days with Availability below Availability threshold
+- Count number of Days with abis fails exceeded TCH abis fails threshold

process_kpi/process_gsm_capacity.py

@@ -0,0 +1,408 @@
+import numpy as np
+import pandas as pd
+
+from queries.process_gsm import combined_gsm_database
+from utils.check_sheet_exist import execute_checks_sheets_exist
+from utils.convert_to_excel import convert_dfs, save_dataframe
+from utils.kpi_analysis_utils import (
+    GsmAnalysis,
+    create_daily_date,
+    create_dfs_per_kpi,
+    create_hourly_date,
+    kpi_naming_cleaning,
+)
+
+
+class GsmCapacity:
+    final_results = None
+
+
+GSM_COLUMNS = [
+    "ID_BTS",
+    "site_name",
+    "name",
+    "BSC",
+    "BCF",
+    "BTS",
+    "code",
+    "Region",
+    "adminState",
+    "frequencyBandInUse",
+    "amrSegLoadDepTchRateLower",
+    "amrSegLoadDepTchRateUpper",
+    "dedicatedGPRScapacity",
+    "defaultGPRScapacity",
+    "cellId",
+    "band",
+    "site_config_band",
+    "trxRfPower",
+    "BCCH",
+    "number_trx_per_cell",
+    "number_trx_per_bcf",
+    "TRX_TCH",
+    "MAL_TCH",
+]
+
+TRX_COLUMNS = [
+    "ID_BTS",
+    "number_tch_per_cell",
+    "number_sd_per_cell",
+    "number_bcch_per_cell",
+    "number_ccch_per_cell",
+    "number_cbc_per_cell",
+    "number_total_channels_per_cell",
+    "number_signals_per_cell",
+]
+
+KPI_COLUMNS = [
+    "date",
+    "BTS_name",
+    "TCH_availability_ratio",
+    "2G_Carried_Traffic",
+    "TCH_call_blocking",
+    "TCH_ABIS_FAIL_CALL_c001084",
+    "SDCCH_real_blocking",
+]
+BH_COLUMNS_FOR_CAPACITY = [
+    "Max_Traffic BH",
+    "Avg_Traffic BH",
+    "Max_tch_call_blocking BH",
+    "Avg_tch_call_blocking BH",
+    "number_of_days_with_tch_blocking_exceeded",
+    "Max_sdcch_real_blocking BH",
+    "Avg_sdcch_real_blocking BH",
+    "number_of_days_with_sdcch_blocking_exceeded",
+]
+
+
+def bh_tch_call_blocking_analysis(
+    df: pd.DataFrame,
+    number_of_kpi_days: int,
+    tch_blocking_threshold: int,
+    number_of_threshold_days: int,
+) -> pd.DataFrame:
+
+    result_df = df.copy()
+    last_days_df = result_df.iloc[:, -number_of_kpi_days:]
+    # last_days_df = last_days_df.fillna(0)
+
+    result_df["Avg_tch_call_blocking BH"] = last_days_df.mean(axis=1).round(2)
+    result_df["Max_tch_call_blocking BH"] = last_days_df.max(axis=1)
+    # Count the number of days above threshold
+    result_df["number_of_days_with_tch_blocking_exceeded"] = last_days_df.apply(
+        lambda row: sum(1 for x in row if x >= tch_blocking_threshold), axis=1
+    )
+    return result_df
+
+
+def bh_sdcch_call_blocking_analysis(
+    df: pd.DataFrame,
+    number_of_kpi_days: int,
+    sdcch_blocking_threshold: int,
+    number_of_threshold_days: int,
+) -> pd.DataFrame:
+
+    result_df = df.copy()
+    last_days_df = result_df.iloc[:, -number_of_kpi_days:]
+    # last_days_df = last_days_df.fillna(0)
+
+    result_df["Avg_sdcch_real_blocking BH"] = last_days_df.mean(axis=1).round(2)
+    result_df["Max_sdcch_real_blocking BH"] = last_days_df.max(axis=1)
+    # Count the number of days above threshold
+    result_df["number_of_days_with_sdcch_blocking_exceeded"] = last_days_df.apply(
+        lambda row: sum(1 for x in row if x >= sdcch_blocking_threshold), axis=1
+    )
+    return result_df
+
+
+def bh_traffic_analysis(
+    df: pd.DataFrame,
+    number_of_kpi_days: int,
+) -> pd.DataFrame:
+
+    result_df = df.copy()
+    last_days_df = result_df.iloc[:, -number_of_kpi_days:]
+    # last_days_df = last_days_df.fillna(0)
+
+    result_df["Avg_Traffic BH"] = last_days_df.mean(axis=1).round(2)
+    result_df["Max_Traffic BH"] = last_days_df.max(axis=1)
+    return result_df
+
+
+def bh_dfs_per_kpi(
+    df: pd.DataFrame,
+    number_of_kpi_days: int = 7,
+    tch_blocking_threshold: int = 0.50,
+    sdcch_blocking_threshold: int = 0.50,
+    number_of_threshold_days: int = 3,
+) -> pd.DataFrame:
+    """
+    Create pivoted DataFrames for each KPI and perform analysis.
+
+    Args:
+        df: DataFrame containing KPI data
+        number_of_kpi_days: Number of days to analyze
+        threshold: Utilization threshold percentage for flagging
+        number_of_threshold_days: Minimum days above threshold to flag for upgrade
+
+    Returns:
+        DataFrame with combined analysis results
+    """
+    pivoted_kpi_dfs = {}
+
+    pivoted_kpi_dfs = create_dfs_per_kpi(
+        df=df,
+        pivot_date_column="date",
+        pivot_name_column="BTS_name",
+        kpi_columns_from=2,
+    )
+
+    tch_call_blocking_df: pd.DataFrame = pivoted_kpi_dfs["TCH_call_blocking"]
+    sdcch_real_blocking_df: pd.DataFrame = pivoted_kpi_dfs["SDCCH_real_blocking"]
+    Carried_Traffic_df: pd.DataFrame = pivoted_kpi_dfs["2G_Carried_Traffic"]
+    tch_availability_ratio_df: pd.DataFrame = pivoted_kpi_dfs["TCH_availability_ratio"]
+
+    # ANALISYS
+
+    tch_call_blocking_df = bh_tch_call_blocking_analysis(
+        df=tch_call_blocking_df,
+        number_of_kpi_days=number_of_kpi_days,
+        tch_blocking_threshold=tch_blocking_threshold,
+        number_of_threshold_days=number_of_threshold_days,
+    )
+
+    sdcch_real_blocking_df = bh_sdcch_call_blocking_analysis(
+        df=sdcch_real_blocking_df,
+        number_of_kpi_days=number_of_kpi_days,
+        sdcch_blocking_threshold=sdcch_blocking_threshold,
+        number_of_threshold_days=number_of_threshold_days,
+    )
+
+    Carried_Traffic_df = bh_traffic_analysis(
+        df=Carried_Traffic_df,
+        number_of_kpi_days=number_of_kpi_days,
+    )
+
+    # Carried_Traffic_df["Max_Traffic BH"] = Carried_Traffic_df.max(axis=1)
+    # Carried_Traffic_df["Avg_Traffic BH"] = Carried_Traffic_df.mean(axis=1)
+
+    bh_kpi_df = pd.concat(
+        [
+            tch_availability_ratio_df,
+            Carried_Traffic_df,
+            tch_call_blocking_df,
+            sdcch_real_blocking_df,
+        ],
+        axis=1,
+    )
+    # print(Carried_Traffic_df)
+
+    return bh_kpi_df
+
+
+def analyse_bh_data(
+    bh_report_path: str,
+    number_of_kpi_days: int,
+    tch_blocking_threshold: int,
+    sdcch_blocking_threshold: int,
+    number_of_threshold_days: int,
+) -> pd.DataFrame:
+    df = pd.read_csv(bh_report_path, delimiter=";")
+    df = kpi_naming_cleaning(df)
+    df = create_hourly_date(df)
+    df = df[KPI_COLUMNS]
+    df = bh_dfs_per_kpi(
+        df=df,
+        number_of_kpi_days=number_of_kpi_days,
+        tch_blocking_threshold=tch_blocking_threshold,
+        sdcch_blocking_threshold=sdcch_blocking_threshold,
+        number_of_threshold_days=number_of_threshold_days,
+    )
+
+    bh_df_for_capacity = df.copy()
+    bh_df_for_capacity = bh_df_for_capacity[BH_COLUMNS_FOR_CAPACITY]
+    bh_df_for_capacity = bh_df_for_capacity.reset_index()
+
+    # If columns have multiple levels (MultiIndex), flatten them
+    if isinstance(bh_df_for_capacity.columns, pd.MultiIndex):
+        bh_df_for_capacity.columns = [
+            "_".join([str(el) for el in col if el])
+            for col in bh_df_for_capacity.columns.values
+        ]
+    # bh_df_for_capacity = bh_df_for_capacity.reset_index()
+
+    # rename Bts_name to name
+    bh_df_for_capacity = bh_df_for_capacity.rename(columns={"BTS_name": "name"})
+
+    return [bh_df_for_capacity, df]
+
+
+def daily_dfs_per_kpi(
+    df: pd.DataFrame,
+    number_of_kpi_days: int = 7,
+    availability_threshold: int = 95,
+    number_of_threshold_days: int = 3,
+    tch_abis_fails_threshold: int = 10,
+) -> pd.DataFrame:
+    """
+    Create pivoted DataFrames for each KPI and perform analysis.
+
+    Args:
+        df: DataFrame containing KPI data
+        number_of_kpi_days: Number of days to analyze
+        threshold: Utilization threshold percentage for flagging
+        number_of_threshold_days: Minimum days above threshold to flag for upgrade
+
+    Returns:
+        DataFrame with combined analysis results
+    """
+    pivoted_kpi_dfs = {}
+
+    pivoted_kpi_dfs = create_dfs_per_kpi(
+        df=df,
+        pivot_date_column="date",
+        pivot_name_column="BTS_name",
+        kpi_columns_from=2,
+    )
+
+    tch_call_blocking_df: pd.DataFrame = pivoted_kpi_dfs["TCH_call_blocking"]
+    sdcch_real_blocking_df: pd.DataFrame = pivoted_kpi_dfs["SDCCH_real_blocking"]
+    Carried_Traffic_df: pd.DataFrame = pivoted_kpi_dfs["2G_Carried_Traffic"]
+    tch_availability_ratio_df: pd.DataFrame = pivoted_kpi_dfs["TCH_availability_ratio"]
+    tch_abis_fails_df: pd.DataFrame = pivoted_kpi_dfs["TCH_ABIS_FAIL_CALL_c001084"]
+
+
+def analyse_daily_data(
+    daily_report_path: str,
+    number_of_kpi_days: int,
+    tch_abis_fails_threshold: int,
+    availability_threshold: int,
+    number_of_threshold_days: int,
+) -> pd.DataFrame:
+    df = pd.read_csv(daily_report_path, delimiter=";")
+    df = kpi_naming_cleaning(df)
+    df = create_daily_date(df)
+    df = df[KPI_COLUMNS]
+    df = daily_dfs_per_kpi(
+        df=df,
+        number_of_kpi_days=number_of_kpi_days,
+        availability_threshold=availability_threshold,
+        tch_abis_fails_threshold=tch_abis_fails_threshold,
+        number_of_threshold_days=number_of_threshold_days,
+    )
+    # print(df)
+
+
+def get_gsm_databases(dump_path: str) -> pd.DataFrame:
+
+    dfs = combined_gsm_database(dump_path)
+    bts_df: pd.DataFrame = dfs[0]
+    trx_df: pd.DataFrame = dfs[2]
+
+    # Clean GSM df
+    bts_df = bts_df[GSM_COLUMNS]
+    trx_df = trx_df[TRX_COLUMNS]
+
+    # Remove duplicate in TRX df
+    trx_df = trx_df.drop_duplicates(subset=["ID_BTS"])
+
+    gsm_df = pd.merge(bts_df, trx_df, on="ID_BTS", how="left")
+
+    # add hf_rate_coef
+    gsm_df["hf_rate_coef"] = gsm_df["amrSegLoadDepTchRateLower"].map(
+        GsmAnalysis.hf_rate_coef
+    )
+    # Add "GPRS" colomn equal to (dedicatedGPRScapacity * number_tch_per_cell)/100
+    gsm_df["GPRS"] = (
+        gsm_df["dedicatedGPRScapacity"] * gsm_df["number_tch_per_cell"]
+    ) / 100
+
+    # "TCH Actual HR%" equal to "number of TCH" multiplyed by "Coef HF rate"
+    gsm_df["TCH Actual HR%"] = gsm_df["number_tch_per_cell"] * gsm_df["hf_rate_coef"]
+
+    # Remove empty rows
+    gsm_df = gsm_df.dropna(subset=["TCH Actual HR%"])
+
+    # Get "Offered Traffic BH" by mapping approximate "TCH Actual HR%" to 2G analysis_utility "erlangB" dict
+    gsm_df["Offered Traffic BH"] = gsm_df["TCH Actual HR%"].apply(
+        lambda x: GsmAnalysis.erlangB_table.get(int(x), 0)
+    )
+
+    # save_dataframe(gsm_df, "GSM")
+    return gsm_df
+
+
+def analyze_gsm_data(
+    dump_path: str,
+    daily_report_path: str,
+    bh_report_path: str,
+    number_of_kpi_days: int,
+    number_of_threshold_days: int,
+    availability_threshold: int,
+    tch_abis_fails_threshold: int,
+    sddch_blocking_threshold: float,
+    tch_blocking_threshold: float,
+):
+    # print("Analyzing data...")
+    # print(f"Number of days: {number_of_kpi_days}")
+    # print(f"availability_threshold: {availability_threshold}")
+
+    analyse_daily_data(
+        daily_report_path=daily_report_path,
+        number_of_kpi_days=number_of_kpi_days,
+        availability_threshold=availability_threshold,
+        tch_abis_fails_threshold=tch_abis_fails_threshold,
+        number_of_threshold_days=number_of_threshold_days,
+    )
+
+    gsm_database_df: pd.DataFrame = get_gsm_databases(dump_path)
+
+    bh_kpi_dfs = analyse_bh_data(
+        bh_report_path=bh_report_path,
+        number_of_kpi_days=number_of_kpi_days,
+        tch_blocking_threshold=tch_blocking_threshold,
+        sdcch_blocking_threshold=sddch_blocking_threshold,
+        number_of_threshold_days=number_of_threshold_days,
+    )
+
+    bh_kpi_df = bh_kpi_dfs[0]
+    bh_kpi_full_df = bh_kpi_dfs[1]
+
+    gsm_analysis_df = gsm_database_df.merge(bh_kpi_df, on="name", how="left")
+
+    # "TCH UTILIZATION (@Max Traffic)" equal to "(Max_Trafic" divided by "Offered Traffic BH)*100"
+    gsm_analysis_df["TCH UTILIZATION (@Max Traffic)"] = (
+        gsm_analysis_df["Max_Traffic BH"] / gsm_analysis_df["Offered Traffic BH"]
+    ) * 100
+
+    # Add "ERLANGB value" =MAX TRAFFIC/(1-(MAX TCH call blocking/200))
+    gsm_analysis_df["ErlabngB_value"] = gsm_analysis_df["Max_Traffic BH"] / (
+        1 - (gsm_analysis_df["Max_tch_call_blocking BH"] / 200)
+    )
+
+    # - Get "Target FR CHs" by mapping "ERLANG value" to 2G analysis_utility "erlangB" dict
+    gsm_analysis_df["Target FR CHs"] = gsm_analysis_df["ErlabngB_value"].apply(
+        lambda x: GsmAnalysis.erlangB_table.get(int(x) if pd.notnull(x) else 0, 0)
+    )
+
+    # "Target HR CHs" equal to  "Target FR CHs" * 2
+    gsm_analysis_df["Target HR CHs"] = gsm_analysis_df["Target FR CHs"] * 2
+
+    # - Target TCHs equal to Target HR CHs + Signal + GPRS + SDCCH
+    gsm_analysis_df["Target TCHs"] = (
+        gsm_analysis_df["Target HR CHs"]
+        + gsm_analysis_df["number_signals_per_cell"]
+        + gsm_analysis_df["GPRS"]
+        + gsm_analysis_df["number_sd_per_cell"]
+    )
+    # "Target TRXs" equal to roundup(Target TCHs/8)
+    gsm_analysis_df["Target TRXs"] = np.ceil(
+        gsm_analysis_df["Target TCHs"] / 8
+    )  # df["Target TCHs"] / 8
+
+    # "Numberof required TRXs" equal to difference between "Target TRXs" and "number_trx_per_cell"
+    gsm_analysis_df["Numberof required TRXs"] = (
+        gsm_analysis_df["Target TRXs"] - gsm_analysis_df["number_trx_per_cell"]
+    )
+
+    return [gsm_analysis_df, bh_kpi_full_df]