File size: 9,880 Bytes
13a5750 a64569a bd3da99 a64569a 13a5750 c005a67 13a5750 c005a67 13a5750 a64569a 13a5750 a64569a 13a5750 a64569a 13a5750 a64569a 13a5750 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 |
import pandas as pd
from utils.kpi_analysis_utils import (
cell_availability_analysis,
combine_comments,
create_daily_date,
create_dfs_per_kpi,
kpi_naming_cleaning,
)
class WbtsCapacity:
final_results: pd.DataFrame = None
def check_deviation(row: pd.Series, max_diff: float = 3.0, type: str = "") -> str:
"""
Check if any value in the row deviates more than max_diff from the most common value.
Args:
row: Series of values to check for deviation
max_diff: Maximum allowed difference from the most common value
type: Type identifier for the deviation message
Returns:
A message indicating deviation if found, otherwise an empty string
"""
numeric_row = row.astype(float) # Ensure numeric
mode_series = numeric_row.mode()
# Safe fallback in case mode is empty
most_common = mode_series.iloc[0] if not mode_series.empty else numeric_row.iloc[0]
diffs = abs(numeric_row - most_common)
if (diffs > max_diff).any():
return f"{type} Deviation > {max_diff} detected"
else:
return ""
def max_used_bb_subunits_analysis(
df: pd.DataFrame,
days: int = 7,
threshold: int = 80,
number_of_threshold_days: int = 3,
) -> pd.DataFrame:
"""
Analyze maximum used baseband subunits and identify sites needing upgrades.
Args:
df: DataFrame containing baseband utilization data
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 analysis results and upgrade recommendations
"""
result_df = df.copy()
last_days_df = result_df.iloc[:, -days:]
last_days_df = last_days_df.fillna(0)
result_df["Average_used_bb_ratio"] = last_days_df.mean(axis=1).round(2)
# Count the number of days above threshold
result_df["bb_number_of_days_exceeding_threshold"] = last_days_df.apply(
lambda row: sum(1 for x in row if x >= threshold), axis=1
)
# Initialize comment column
result_df["Average_used_bb_ratio_comment"] = ""
# Apply condition for upgrade recommendation
result_df.loc[
(result_df["bb_number_of_days_exceeding_threshold"] >= number_of_threshold_days)
& (result_df["Average_used_bb_ratio"] >= threshold),
"Average_used_bb_ratio_comment",
] = "need BB upgrade"
return result_df
def max_used_ce_analysis(
df: pd.DataFrame,
days: int = 7,
threshold: int = 80,
number_of_threshold_days: int = 3,
) -> pd.DataFrame:
"""
Analyze maximum used channel elements and identify sites needing upgrades.
Args:
df: DataFrame containing channel element utilization data
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 analysis results and upgrade recommendations
"""
result_df = df.copy().fillna(0)
last_days_df = result_df.iloc[:, -days:]
result_df["Average_used_ce_ratio"] = last_days_df.mean(axis=1).round(2)
# Count the number of days above threshold
result_df["ce_number_of_days_exceeding_threshold"] = last_days_df.apply(
lambda row: sum(1 for x in row if x >= threshold), axis=1
)
# Initialize comment column
result_df["Average_used_ce_ratio_comment"] = ""
# Apply condition for upgrade recommendation
result_df.loc[
(result_df["ce_number_of_days_exceeding_threshold"] >= number_of_threshold_days)
& (result_df["Average_used_ce_ratio"] >= threshold),
"Average_used_ce_ratio_comment",
] = "need CE upgrade"
return result_df
def num_bb_subunits_analysis(df: pd.DataFrame, days: int = 3) -> pd.DataFrame:
"""
Analyze baseband subunit count for deviations.
Args:
df: DataFrame containing baseband subunit count data
days: Number of days to analyze
Returns:
DataFrame with deviation analysis comments
"""
result_df = df.copy()
last_days_df = result_df.iloc[:, -days:]
result_df["num_bb_subunits_comment"] = last_days_df.apply(
lambda row: check_deviation(row, type="bb"), axis=1
)
return result_df
def avail_ce_analysis(df: pd.DataFrame, days: int = 7) -> pd.DataFrame:
"""
Analyze available channel elements for deviations.
Args:
df: DataFrame containing available channel element data
days: Number of days to analyze
Returns:
DataFrame with deviation analysis comments
"""
result_df = df.copy()
last_days_df = result_df.iloc[:, -days:]
result_df["avail_ce_comment"] = last_days_df.apply(
lambda row: check_deviation(row, max_diff=96, type="ce"), axis=1
)
return result_df
def bb_comments_analysis(df: pd.DataFrame) -> pd.DataFrame:
"""
Combine baseband related comments into a single column.
Args:
df: DataFrame containing baseband comment columns
Returns:
DataFrame with combined baseband comments
"""
return combine_comments(
df,
"num_bb_subunits_comment",
"Average_used_bb_ratio_comment",
"availability_comment_daily",
new_column="bb_comments",
)
def ce_comments_analysis(df: pd.DataFrame) -> pd.DataFrame:
"""
Combine channel element related comments into a single column.
Args:
df: DataFrame containing channel element comment columns
Returns:
DataFrame with combined channel element comments
"""
return combine_comments(
df,
"avail_ce_comment",
"Average_used_ce_ratio_comment",
"availability_comment_daily",
new_column="ce_comments",
)
def wbts_kpi_analysis(
df: pd.DataFrame,
num_days: int = 7,
threshold: int = 80,
number_of_threshold_days: int = 3,
) -> pd.DataFrame:
"""
Create pivoted DataFrames for each KPI and perform analysis.
Args:
df: DataFrame containing KPI data
num_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
"""
# kpi_columns = df.columns[5:]
pivoted_kpi_dfs = {}
pivoted_kpi_dfs = create_dfs_per_kpi(
df=df, pivot_date_column="date", pivot_name_column="DN", kpi_columns_from=5
)
# Extract individual KPI DataFrames
wbts_name_df = pivoted_kpi_dfs["WBTS_name"].iloc[:, 0]
licensed_ce_df = pivoted_kpi_dfs["LICENSED_R99CE_WBTS_M5008C48"]
max_used_ce_dl_df = pivoted_kpi_dfs["MAX_USED_CE_R99_DL_M5008C12"]
max_used_ce_ul_df = pivoted_kpi_dfs["MAX_USED_CE_R99_UL_M5008C15"]
max_avail_ce_df = pivoted_kpi_dfs["MAX_AVAIL_R99_CE_M5006C0"]
max_used_bb_subunits_df = pivoted_kpi_dfs["MAX_USED_BB_SUBUNITS_M5008C38"]
num_bb_subunits_df = pivoted_kpi_dfs["NUM_BB_SUBUNITS_M5008C39"]
max_bb_sus_util_ratio_df = pivoted_kpi_dfs["Max_BB_SUs_Util_ratio"]
cell_availability_df = pivoted_kpi_dfs[
"Cell_Availability_excluding_blocked_by_user_state_BLU"
]
total_cs_traffic_df = pivoted_kpi_dfs["Total_CS_traffic_Erl"]
total_data_traffic_df = pivoted_kpi_dfs["Total_Data_Traffic"]
max_used_ce_ratio_flexi_df = pivoted_kpi_dfs["Max_Used_CE_s_ratio_Flexi_R2"]
# Perform analysis on each KPI DataFrame
max_bb_sus_util_ratio_df = max_used_bb_subunits_analysis(
max_bb_sus_util_ratio_df, num_days, threshold, number_of_threshold_days
)
cell_availability_df = cell_availability_analysis(cell_availability_df, num_days)
max_used_ce_ratio_flexi_df = max_used_ce_analysis(
max_used_ce_ratio_flexi_df, num_days, threshold, number_of_threshold_days
)
num_bb_subunits_df = num_bb_subunits_analysis(num_bb_subunits_df, num_days)
licensed_ce_df = avail_ce_analysis(licensed_ce_df, num_days)
# Concatenate all DataFrames
result_df = pd.concat(
[
wbts_name_df,
licensed_ce_df,
max_used_ce_dl_df,
max_used_ce_ul_df,
max_avail_ce_df,
max_used_bb_subunits_df,
num_bb_subunits_df,
max_bb_sus_util_ratio_df,
cell_availability_df,
total_cs_traffic_df,
total_data_traffic_df,
max_used_ce_ratio_flexi_df,
],
axis=1,
)
# Add combined comments analysis
result_df = bb_comments_analysis(result_df)
result_df = ce_comments_analysis(result_df)
return result_df
def load_data(
filepath: str,
num_days: int,
threshold: int,
number_of_threshold_days: int,
) -> pd.DataFrame:
"""
Load data from CSV file and perform preprocessing and analysis.
Args:
filepath: Path to CSV file or uploaded file object
num_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 processed and analyzed data
"""
df = pd.read_csv(filepath, delimiter=";")
# Preprocess data
df = create_daily_date(df)
df = kpi_naming_cleaning(df)
# Reorder columns for better organization
df = df[["date"] + [col for col in df.columns if col not in ["date"]]]
df = df[[col for col in df.columns if col != "WBTS_name"] + ["WBTS_name"]]
# Perform KPI analysis
df = wbts_kpi_analysis(df, num_days, threshold, number_of_threshold_days)
# for col, col_index in zip(df.columns, df.columns.get_indexer(df.columns)):
# print(f"Column: {col}, Index: {col_index}")
return df
|