DavMelchi commited on
Commit
53c4407
·
1 Parent(s): 1d869c5

adding CIQ verification app

Browse files
Files changed (3) hide show
  1. app.py +1 -0
  2. apps/ciq_verification.py +229 -0
  3. queries/verify_ciq.py +514 -0
app.py CHANGED
@@ -121,6 +121,7 @@ if check_password():
121
  st.Page("apps/ciq_2g_generator.py", title="🧾 CIQ 2G Generator"),
122
  st.Page("apps/ciq_3g_generator.py", title="🧾 CIQ 3G Generator"),
123
  st.Page("apps/ciq_4g_generator.py", title="🧾 CIQ 4G Generator"),
 
124
  st.Page("apps/core_dump_page.py", title="📠Parse dump core"),
125
  st.Page("apps/gps_converter.py", title="🧭GPS Converter"),
126
  st.Page("apps/distance.py", title="🛰Distance Calculator"),
 
121
  st.Page("apps/ciq_2g_generator.py", title="🧾 CIQ 2G Generator"),
122
  st.Page("apps/ciq_3g_generator.py", title="🧾 CIQ 3G Generator"),
123
  st.Page("apps/ciq_4g_generator.py", title="🧾 CIQ 4G Generator"),
124
+ st.Page("apps/ciq_verification.py", title="🔍 CIQ Verification"),
125
  st.Page("apps/core_dump_page.py", title="📠Parse dump core"),
126
  st.Page("apps/gps_converter.py", title="🧭GPS Converter"),
127
  st.Page("apps/distance.py", title="🛰Distance Calculator"),
apps/ciq_verification.py ADDED
@@ -0,0 +1,229 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """
2
+ CIQ Verification App
3
+
4
+ Streamlit interface to verify CIQ parameters against dump database.
5
+ Supports 2G, 3G, and LTE verification with optional file uploads.
6
+ """
7
+
8
+ import pandas as pd
9
+ import streamlit as st
10
+
11
+ from queries.verify_ciq import (
12
+ generate_verification_report,
13
+ process_dump_gsm,
14
+ process_dump_lte,
15
+ process_dump_wcdma,
16
+ read_ciq_file,
17
+ verify_2g,
18
+ verify_3g,
19
+ verify_lte,
20
+ )
21
+
22
+ st.title("🔍 CIQ Verification")
23
+ st.markdown(
24
+ """
25
+ Vérifiez que les paramètres CIQ correspondent aux valeurs du dump OML.
26
+ - **Dump** : Obligatoire (format .xlsb)
27
+ - **CIQ** : Au moins un fichier CIQ (2G, 3G ou LTE) est requis
28
+ """
29
+ )
30
+
31
+ # File uploaders
32
+ st.subheader("📁 Fichiers d'entrée")
33
+
34
+ dump_file = st.file_uploader(
35
+ "Upload Dump (xlsb)", type=["xlsb"], key="verify_dump", help="Fichier dump obligatoire"
36
+ )
37
+
38
+ col1, col2, col3 = st.columns(3)
39
+
40
+ with col1:
41
+ ciq_2g_file = st.file_uploader(
42
+ "CIQ 2G (optionnel)", type=["xlsx", "xls"], key="verify_ciq_2g"
43
+ )
44
+
45
+ with col2:
46
+ ciq_3g_file = st.file_uploader(
47
+ "CIQ 3G (optionnel)", type=["xlsx", "xls"], key="verify_ciq_3g"
48
+ )
49
+
50
+ with col3:
51
+ ciq_lte_file = st.file_uploader(
52
+ "CIQ LTE (optionnel)", type=["xlsx", "xls"], key="verify_ciq_lte"
53
+ )
54
+
55
+ # Validation
56
+ if dump_file is None:
57
+ st.info("⬆️ Veuillez uploader le fichier dump (xlsb).")
58
+ st.stop()
59
+
60
+ if ciq_2g_file is None and ciq_3g_file is None and ciq_lte_file is None:
61
+ st.warning("⚠️ Au moins un fichier CIQ (2G, 3G ou LTE) est requis.")
62
+ st.stop()
63
+
64
+ # Verify button
65
+ if st.button("🔎 Vérifier", type="primary"):
66
+ try:
67
+ results_2g = None
68
+ results_3g = None
69
+ results_lte = None
70
+
71
+ with st.spinner("Traitement en cours..."):
72
+ # Process 2G if provided
73
+ if ciq_2g_file is not None:
74
+ st.text("📶 Traitement 2G...")
75
+ dump_gsm = process_dump_gsm(dump_file)
76
+ dump_file.seek(0) # Reset file pointer
77
+ ciq_2g_df = read_ciq_file(ciq_2g_file)
78
+ results_2g = verify_2g(ciq_2g_df, dump_gsm)
79
+
80
+ # Process 3G if provided
81
+ if ciq_3g_file is not None:
82
+ st.text("📶 Traitement 3G...")
83
+ dump_wcdma = process_dump_wcdma(dump_file)
84
+ dump_file.seek(0) # Reset file pointer
85
+ ciq_3g_df = read_ciq_file(ciq_3g_file)
86
+ results_3g = verify_3g(ciq_3g_df, dump_wcdma)
87
+
88
+ # Process LTE if provided
89
+ if ciq_lte_file is not None:
90
+ st.text("📶 Traitement LTE...")
91
+ dump_lte = process_dump_lte(dump_file)
92
+ dump_file.seek(0) # Reset file pointer
93
+ ciq_lte_df = read_ciq_file(ciq_lte_file)
94
+ results_lte = verify_lte(ciq_lte_df, dump_lte)
95
+
96
+ # Generate report
97
+ sheets, excel_bytes = generate_verification_report(
98
+ results_2g=results_2g,
99
+ results_3g=results_3g,
100
+ results_lte=results_lte,
101
+ )
102
+
103
+ st.session_state["verify_results_2g"] = results_2g
104
+ st.session_state["verify_results_3g"] = results_3g
105
+ st.session_state["verify_results_lte"] = results_lte
106
+ st.session_state["verify_sheets"] = sheets
107
+ st.session_state["verify_excel_bytes"] = excel_bytes
108
+
109
+ st.success("✅ Vérification terminée!")
110
+
111
+ except Exception as e:
112
+ st.error(f"❌ Erreur: {e}")
113
+ import traceback
114
+ st.code(traceback.format_exc())
115
+
116
+ # Display results
117
+ results_2g = st.session_state.get("verify_results_2g")
118
+ results_3g = st.session_state.get("verify_results_3g")
119
+ results_lte = st.session_state.get("verify_results_lte")
120
+ sheets = st.session_state.get("verify_sheets")
121
+ excel_bytes = st.session_state.get("verify_excel_bytes")
122
+
123
+
124
+ def display_stats(stats: dict, tech: str):
125
+ """Display verification statistics."""
126
+ col1, col2, col3, col4 = st.columns(4)
127
+ with col1:
128
+ st.metric("Total Cells", stats["total_cells"])
129
+ with col2:
130
+ st.metric("✅ OK", stats["ok_count"])
131
+ with col3:
132
+ st.metric("⚠️ Mismatch", stats["mismatch_count"])
133
+ with col4:
134
+ st.metric("❓ Not Found", stats["not_found_count"])
135
+
136
+
137
+ def style_results(df: pd.DataFrame) -> pd.DataFrame:
138
+ """Apply styling to highlight mismatches."""
139
+ def highlight_status(val):
140
+ if val == "OK":
141
+ return "background-color: #d4edda; color: #155724;"
142
+ elif val == "MISMATCH":
143
+ return "background-color: #f8d7da; color: #721c24;"
144
+ elif val == "NOT_FOUND":
145
+ return "background-color: #fff3cd; color: #856404;"
146
+ return ""
147
+
148
+ def highlight_match(val):
149
+ if val is True:
150
+ return "background-color: #d4edda;"
151
+ elif val is False:
152
+ return "background-color: #f8d7da;"
153
+ return ""
154
+
155
+ # Get status column name
156
+ status_col = "Status"
157
+
158
+ # Apply styling (using map instead of deprecated applymap)
159
+ styled = df.style.map(
160
+ highlight_status, subset=[status_col]
161
+ )
162
+
163
+ # Highlight match columns
164
+ match_cols = [c for c in df.columns if c.endswith("_Match")]
165
+ if match_cols:
166
+ styled = styled.map(highlight_match, subset=match_cols)
167
+
168
+ return styled
169
+
170
+
171
+ if sheets:
172
+ st.divider()
173
+ st.subheader("📊 Résultats de la vérification")
174
+
175
+ tabs = []
176
+ tab_names = []
177
+
178
+ if results_2g:
179
+ tab_names.append("2G")
180
+ if results_3g:
181
+ tab_names.append("3G")
182
+ if results_lte:
183
+ tab_names.append("LTE")
184
+
185
+ if tab_names:
186
+ tabs = st.tabs(tab_names)
187
+ tab_idx = 0
188
+
189
+ if results_2g:
190
+ with tabs[tab_idx]:
191
+ st.markdown("### 📶 Vérification 2G")
192
+ display_stats(results_2g[1], "2G")
193
+ st.dataframe(
194
+ style_results(results_2g[0]),
195
+ use_container_width=True,
196
+ hide_index=True,
197
+ )
198
+ tab_idx += 1
199
+
200
+ if results_3g:
201
+ with tabs[tab_idx]:
202
+ st.markdown("### 📶 Vérification 3G")
203
+ display_stats(results_3g[1], "3G")
204
+ st.dataframe(
205
+ style_results(results_3g[0]),
206
+ use_container_width=True,
207
+ hide_index=True,
208
+ )
209
+ tab_idx += 1
210
+
211
+ if results_lte:
212
+ with tabs[tab_idx]:
213
+ st.markdown("### 📶 Vérification LTE")
214
+ display_stats(results_lte[1], "LTE")
215
+ st.dataframe(
216
+ style_results(results_lte[0]),
217
+ use_container_width=True,
218
+ hide_index=True,
219
+ )
220
+
221
+ if excel_bytes:
222
+ st.divider()
223
+ st.download_button(
224
+ label="📥 Télécharger le rapport de vérification (Excel)",
225
+ data=excel_bytes,
226
+ file_name="CIQ_Verification_Report.xlsx",
227
+ mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
228
+ type="primary",
229
+ )
queries/verify_ciq.py ADDED
@@ -0,0 +1,514 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """
2
+ CIQ Verification Module
3
+
4
+ Compares CIQ parameters (2G/3G/LTE) against dump database values.
5
+ Identifies discrepancies between configured and actual network parameters.
6
+ """
7
+
8
+ from io import BytesIO
9
+ from typing import Optional, Tuple
10
+
11
+ import pandas as pd
12
+
13
+
14
+ # Column mappings: CIQ column name -> Dump column name
15
+ CIQ_2G_MAPPING = {
16
+ "NOM_CELLULE": "name",
17
+ "LAC": "locationAreaIdLAC",
18
+ "CI": "cellId",
19
+ }
20
+
21
+ CIQ_3G_MAPPING = {
22
+ "NOM_CELLULE": "name",
23
+ "CELLID": "CId",
24
+ "SAC": "SAC",
25
+ "LAC": "LAC",
26
+ "PSCRAMBCODE": "PriScrCode",
27
+ }
28
+
29
+ CIQ_LTE_MAPPING = {
30
+ "CellName": "cellName",
31
+ "TAC": "tac",
32
+ "Physical cell ID": "phyCellId",
33
+ "Root sequence index": "rootSeqIndex",
34
+ }
35
+
36
+
37
+ def normalize_cell_name(name: str) -> str:
38
+ """
39
+ Normalize cell name by removing _NA suffix if present.
40
+ Handles both CIQ and dump cell names for comparison.
41
+
42
+ Args:
43
+ name: Cell name to normalize
44
+
45
+ Returns:
46
+ Normalized cell name without _NA suffix
47
+ """
48
+ if pd.isna(name):
49
+ return ""
50
+ name_str = str(name).strip()
51
+ if name_str.endswith("_NA"):
52
+ return name_str[:-3]
53
+ return name_str
54
+
55
+
56
+ def process_dump_gsm(dump_file) -> pd.DataFrame:
57
+ """
58
+ Process GSM data from dump file for verification.
59
+
60
+ Args:
61
+ dump_file: Uploaded dump file (xlsb format)
62
+
63
+ Returns:
64
+ DataFrame with GSM cells and parameters
65
+ """
66
+ dfs = pd.read_excel(
67
+ dump_file,
68
+ sheet_name=["BTS"],
69
+ engine="calamine",
70
+ skiprows=[0],
71
+ )
72
+
73
+ df_bts = dfs["BTS"]
74
+ df_bts.columns = df_bts.columns.str.replace(r"[ ]", "", regex=True)
75
+
76
+ # Select only needed columns for verification
77
+ columns_needed = ["name", "locationAreaIdLAC", "cellId"]
78
+ df_gsm = df_bts[columns_needed].copy()
79
+
80
+ # Normalize cell names
81
+ df_gsm["name_normalized"] = df_gsm["name"].apply(normalize_cell_name)
82
+
83
+ return df_gsm
84
+
85
+
86
+ def process_dump_wcdma(dump_file) -> pd.DataFrame:
87
+ """
88
+ Process WCDMA data from dump file for verification.
89
+
90
+ Args:
91
+ dump_file: Uploaded dump file (xlsb format)
92
+
93
+ Returns:
94
+ DataFrame with WCDMA cells and parameters
95
+ """
96
+ dfs = pd.read_excel(
97
+ dump_file,
98
+ sheet_name=["WCEL"],
99
+ engine="calamine",
100
+ skiprows=[0],
101
+ )
102
+
103
+ df_wcel = dfs["WCEL"]
104
+ df_wcel.columns = df_wcel.columns.str.replace(r"[ ]", "", regex=True)
105
+
106
+ # Select only needed columns for verification
107
+ columns_needed = ["name", "CId", "SAC", "LAC", "PriScrCode"]
108
+ df_wcdma = df_wcel[columns_needed].copy()
109
+
110
+ # Normalize cell names
111
+ df_wcdma["name_normalized"] = df_wcdma["name"].apply(normalize_cell_name)
112
+
113
+ return df_wcdma
114
+
115
+
116
+ def process_dump_lte(dump_file) -> pd.DataFrame:
117
+ """
118
+ Process LTE data from dump file for verification.
119
+ Uses the existing process_lte module to correctly merge FDD/TDD data.
120
+ rootSeqIndex is in LNCEL_FDD/LNCEL_TDD, not LNCEL.
121
+
122
+ Args:
123
+ dump_file: Uploaded dump file (xlsb format)
124
+
125
+ Returns:
126
+ DataFrame with LTE cells and parameters
127
+ """
128
+ from queries.process_lte import process_lte_data
129
+
130
+ # Use existing function which correctly handles FDD/TDD merge
131
+ lte_dfs = process_lte_data(dump_file)
132
+ df_fdd = lte_dfs[0] # FDD cells with rootSeqIndex
133
+ df_tdd = lte_dfs[1] # TDD cells with rootSeqIndex
134
+
135
+ # Combine FDD and TDD
136
+ df_lte = pd.concat([df_fdd, df_tdd], ignore_index=True)
137
+
138
+ # Select only needed columns for verification
139
+ # Use cellName or final_name depending on what's available
140
+ if "cellName" in df_lte.columns:
141
+ name_col = "cellName"
142
+ elif "final_name" in df_lte.columns:
143
+ name_col = "final_name"
144
+ else:
145
+ name_col = "name"
146
+
147
+ columns_mapping = {
148
+ name_col: "cellName",
149
+ "tac": "tac",
150
+ "phyCellId": "phyCellId",
151
+ "rootSeqIndex": "rootSeqIndex",
152
+ }
153
+
154
+ # Check each column exists and build result
155
+ result_data = {}
156
+ for target_col, source_col in [("cellName", name_col), ("tac", "tac"),
157
+ ("phyCellId", "phyCellId"), ("rootSeqIndex", "rootSeqIndex")]:
158
+ if source_col in df_lte.columns:
159
+ result_data[target_col] = df_lte[source_col]
160
+ else:
161
+ result_data[target_col] = None
162
+
163
+ df_result = pd.DataFrame(result_data)
164
+
165
+ # Normalize cell names
166
+ df_result["name_normalized"] = df_result["cellName"].apply(normalize_cell_name)
167
+
168
+ return df_result
169
+
170
+
171
+ def read_ciq_file(ciq_file) -> pd.DataFrame:
172
+ """
173
+ Read CIQ Excel file and return as DataFrame.
174
+
175
+ Args:
176
+ ciq_file: Uploaded CIQ Excel file
177
+
178
+ Returns:
179
+ DataFrame with CIQ data
180
+ """
181
+ df = pd.read_excel(ciq_file, engine="calamine")
182
+ return df
183
+
184
+
185
+ def verify_2g(
186
+ ciq_df: pd.DataFrame, dump_df: pd.DataFrame
187
+ ) -> Tuple[pd.DataFrame, dict]:
188
+ """
189
+ Verify 2G CIQ parameters against dump.
190
+
191
+ Args:
192
+ ciq_df: CIQ 2G DataFrame
193
+ dump_df: Dump GSM DataFrame
194
+
195
+ Returns:
196
+ Tuple of (comparison DataFrame, summary stats)
197
+ """
198
+ # Normalize CIQ cell names
199
+ ciq_df = ciq_df.copy()
200
+ ciq_df["name_normalized"] = ciq_df["NOM_CELLULE"].apply(normalize_cell_name)
201
+
202
+ # Merge on normalized cell name
203
+ merged = ciq_df.merge(
204
+ dump_df,
205
+ on="name_normalized",
206
+ how="left",
207
+ suffixes=("_ciq", "_dump"),
208
+ )
209
+
210
+ # Compare parameters
211
+ results = []
212
+ for _, row in merged.iterrows():
213
+ cell_name = row["NOM_CELLULE"]
214
+ found_in_dump = not pd.isna(row.get("name"))
215
+
216
+ if not found_in_dump:
217
+ results.append(
218
+ {
219
+ "NOM_CELLULE": cell_name,
220
+ "Status": "NOT_FOUND",
221
+ "LAC_CIQ": row.get("LAC"),
222
+ "LAC_DUMP": None,
223
+ "LAC_Match": False,
224
+ "CI_CIQ": row.get("CI"),
225
+ "CI_DUMP": None,
226
+ "CI_Match": False,
227
+ }
228
+ )
229
+ else:
230
+ lac_match = _compare_values(row.get("LAC"), row.get("locationAreaIdLAC"))
231
+ ci_match = _compare_values(row.get("CI"), row.get("cellId"))
232
+ status = "OK" if (lac_match and ci_match) else "MISMATCH"
233
+
234
+ results.append(
235
+ {
236
+ "NOM_CELLULE": cell_name,
237
+ "Status": status,
238
+ "LAC_CIQ": row.get("LAC"),
239
+ "LAC_DUMP": row.get("locationAreaIdLAC"),
240
+ "LAC_Match": lac_match,
241
+ "CI_CIQ": row.get("CI"),
242
+ "CI_DUMP": row.get("cellId"),
243
+ "CI_Match": ci_match,
244
+ }
245
+ )
246
+
247
+ result_df = pd.DataFrame(results)
248
+
249
+ # Summary stats
250
+ stats = {
251
+ "total_cells": len(result_df),
252
+ "ok_count": len(result_df[result_df["Status"] == "OK"]),
253
+ "mismatch_count": len(result_df[result_df["Status"] == "MISMATCH"]),
254
+ "not_found_count": len(result_df[result_df["Status"] == "NOT_FOUND"]),
255
+ }
256
+
257
+ return result_df, stats
258
+
259
+
260
+ def verify_3g(
261
+ ciq_df: pd.DataFrame, dump_df: pd.DataFrame
262
+ ) -> Tuple[pd.DataFrame, dict]:
263
+ """
264
+ Verify 3G CIQ parameters against dump.
265
+
266
+ Args:
267
+ ciq_df: CIQ 3G DataFrame
268
+ dump_df: Dump WCDMA DataFrame
269
+
270
+ Returns:
271
+ Tuple of (comparison DataFrame, summary stats)
272
+ """
273
+ # Normalize CIQ cell names
274
+ ciq_df = ciq_df.copy()
275
+ ciq_df["name_normalized"] = ciq_df["NOM_CELLULE"].apply(normalize_cell_name)
276
+
277
+ # Merge on normalized cell name
278
+ merged = ciq_df.merge(
279
+ dump_df,
280
+ on="name_normalized",
281
+ how="left",
282
+ suffixes=("_ciq", "_dump"),
283
+ )
284
+
285
+ # Compare parameters
286
+ results = []
287
+ for _, row in merged.iterrows():
288
+ cell_name = row["NOM_CELLULE"]
289
+ found_in_dump = not pd.isna(row.get("name"))
290
+
291
+ if not found_in_dump:
292
+ results.append(
293
+ {
294
+ "NOM_CELLULE": cell_name,
295
+ "Status": "NOT_FOUND",
296
+ "CELLID_CIQ": row.get("CELLID"),
297
+ "CELLID_DUMP": None,
298
+ "CELLID_Match": False,
299
+ "SAC_CIQ": row.get("SAC"),
300
+ "SAC_DUMP": None,
301
+ "SAC_Match": False,
302
+ "LAC_CIQ": row.get("LAC"),
303
+ "LAC_DUMP": None,
304
+ "LAC_Match": False,
305
+ "PSC_CIQ": row.get("PSCRAMBCODE"),
306
+ "PSC_DUMP": None,
307
+ "PSC_Match": False,
308
+ }
309
+ )
310
+ else:
311
+ cellid_match = _compare_values(row.get("CELLID"), row.get("CId"))
312
+ sac_match = _compare_values(row.get("SAC_ciq", row.get("SAC")), row.get("SAC_dump", row.get("SAC")))
313
+ lac_match = _compare_values(row.get("LAC_ciq", row.get("LAC")), row.get("LAC_dump"))
314
+ psc_match = _compare_values(row.get("PSCRAMBCODE"), row.get("PriScrCode"))
315
+
316
+ # Handle potential column name conflicts from merge
317
+ sac_ciq = row.get("SAC_ciq") if "SAC_ciq" in row.index else row.get("SAC")
318
+ sac_dump = row.get("SAC_dump") if "SAC_dump" in row.index else None
319
+ lac_ciq = row.get("LAC_ciq") if "LAC_ciq" in row.index else row.get("LAC")
320
+ lac_dump = row.get("LAC_dump") if "LAC_dump" in row.index else None
321
+
322
+ sac_match = _compare_values(sac_ciq, sac_dump)
323
+ lac_match = _compare_values(lac_ciq, lac_dump)
324
+
325
+ status = (
326
+ "OK"
327
+ if (cellid_match and sac_match and lac_match and psc_match)
328
+ else "MISMATCH"
329
+ )
330
+
331
+ results.append(
332
+ {
333
+ "NOM_CELLULE": cell_name,
334
+ "Status": status,
335
+ "CELLID_CIQ": row.get("CELLID"),
336
+ "CELLID_DUMP": row.get("CId"),
337
+ "CELLID_Match": cellid_match,
338
+ "SAC_CIQ": sac_ciq,
339
+ "SAC_DUMP": sac_dump,
340
+ "SAC_Match": sac_match,
341
+ "LAC_CIQ": lac_ciq,
342
+ "LAC_DUMP": lac_dump,
343
+ "LAC_Match": lac_match,
344
+ "PSC_CIQ": row.get("PSCRAMBCODE"),
345
+ "PSC_DUMP": row.get("PriScrCode"),
346
+ "PSC_Match": psc_match,
347
+ }
348
+ )
349
+
350
+ result_df = pd.DataFrame(results)
351
+
352
+ # Summary stats
353
+ stats = {
354
+ "total_cells": len(result_df),
355
+ "ok_count": len(result_df[result_df["Status"] == "OK"]),
356
+ "mismatch_count": len(result_df[result_df["Status"] == "MISMATCH"]),
357
+ "not_found_count": len(result_df[result_df["Status"] == "NOT_FOUND"]),
358
+ }
359
+
360
+ return result_df, stats
361
+
362
+
363
+ def verify_lte(
364
+ ciq_df: pd.DataFrame, dump_df: pd.DataFrame
365
+ ) -> Tuple[pd.DataFrame, dict]:
366
+ """
367
+ Verify LTE CIQ parameters against dump.
368
+
369
+ Args:
370
+ ciq_df: CIQ LTE DataFrame
371
+ dump_df: Dump LTE DataFrame
372
+
373
+ Returns:
374
+ Tuple of (comparison DataFrame, summary stats)
375
+ """
376
+ # Normalize CIQ cell names
377
+ ciq_df = ciq_df.copy()
378
+ ciq_df["name_normalized"] = ciq_df["CellName"].apply(normalize_cell_name)
379
+
380
+ # Merge on normalized cell name
381
+ merged = ciq_df.merge(
382
+ dump_df,
383
+ on="name_normalized",
384
+ how="left",
385
+ suffixes=("_ciq", "_dump"),
386
+ )
387
+
388
+ # Compare parameters
389
+ results = []
390
+ for _, row in merged.iterrows():
391
+ cell_name = row["CellName"]
392
+ found_in_dump = not pd.isna(row.get("cellName"))
393
+
394
+ if not found_in_dump:
395
+ results.append(
396
+ {
397
+ "CellName": cell_name,
398
+ "Status": "NOT_FOUND",
399
+ "TAC_CIQ": row.get("TAC"),
400
+ "TAC_DUMP": None,
401
+ "TAC_Match": False,
402
+ "PCI_CIQ": row.get("Physical cell ID"),
403
+ "PCI_DUMP": None,
404
+ "PCI_Match": False,
405
+ "RSI_CIQ": row.get("Root sequence index"),
406
+ "RSI_DUMP": None,
407
+ "RSI_Match": False,
408
+ }
409
+ )
410
+ else:
411
+ # Handle potential column name conflicts from merge
412
+ tac_ciq = row.get("TAC_ciq") if "TAC_ciq" in row.index else row.get("TAC")
413
+ tac_dump = row.get("tac_dump") if "tac_dump" in row.index else row.get("tac")
414
+
415
+ tac_match = _compare_values(tac_ciq, tac_dump)
416
+ pci_match = _compare_values(row.get("Physical cell ID"), row.get("phyCellId"))
417
+ rsi_match = _compare_values(
418
+ row.get("Root sequence index"), row.get("rootSeqIndex")
419
+ )
420
+
421
+ status = "OK" if (tac_match and pci_match and rsi_match) else "MISMATCH"
422
+
423
+ results.append(
424
+ {
425
+ "CellName": cell_name,
426
+ "Status": status,
427
+ "TAC_CIQ": tac_ciq,
428
+ "TAC_DUMP": tac_dump,
429
+ "TAC_Match": tac_match,
430
+ "PCI_CIQ": row.get("Physical cell ID"),
431
+ "PCI_DUMP": row.get("phyCellId"),
432
+ "PCI_Match": pci_match,
433
+ "RSI_CIQ": row.get("Root sequence index"),
434
+ "RSI_DUMP": row.get("rootSeqIndex"),
435
+ "RSI_Match": rsi_match,
436
+ }
437
+ )
438
+
439
+ result_df = pd.DataFrame(results)
440
+
441
+ # Summary stats
442
+ stats = {
443
+ "total_cells": len(result_df),
444
+ "ok_count": len(result_df[result_df["Status"] == "OK"]),
445
+ "mismatch_count": len(result_df[result_df["Status"] == "MISMATCH"]),
446
+ "not_found_count": len(result_df[result_df["Status"] == "NOT_FOUND"]),
447
+ }
448
+
449
+ return result_df, stats
450
+
451
+
452
+ def _compare_values(val1, val2) -> bool:
453
+ """
454
+ Compare two values for equality, handling NaN and type differences.
455
+
456
+ Args:
457
+ val1: First value
458
+ val2: Second value
459
+
460
+ Returns:
461
+ True if values are equal, False otherwise
462
+ """
463
+ # Handle NaN cases
464
+ if pd.isna(val1) and pd.isna(val2):
465
+ return True
466
+ if pd.isna(val1) or pd.isna(val2):
467
+ return False
468
+
469
+ # Convert to comparable types
470
+ try:
471
+ # Try numeric comparison first
472
+ num1 = float(val1)
473
+ num2 = float(val2)
474
+ return num1 == num2
475
+ except (ValueError, TypeError):
476
+ # Fall back to string comparison
477
+ return str(val1).strip() == str(val2).strip()
478
+
479
+
480
+ def generate_verification_report(
481
+ results_2g: Optional[Tuple[pd.DataFrame, dict]] = None,
482
+ results_3g: Optional[Tuple[pd.DataFrame, dict]] = None,
483
+ results_lte: Optional[Tuple[pd.DataFrame, dict]] = None,
484
+ ) -> Tuple[dict, bytes]:
485
+ """
486
+ Generate verification report as Excel file.
487
+
488
+ Args:
489
+ results_2g: Tuple of (DataFrame, stats) for 2G verification
490
+ results_3g: Tuple of (DataFrame, stats) for 3G verification
491
+ results_lte: Tuple of (DataFrame, stats) for LTE verification
492
+
493
+ Returns:
494
+ Tuple of (sheets dict, Excel bytes)
495
+ """
496
+ sheets = {}
497
+
498
+ if results_2g is not None:
499
+ sheets["2G_Verification"] = results_2g[0]
500
+
501
+ if results_3g is not None:
502
+ sheets["3G_Verification"] = results_3g[0]
503
+
504
+ if results_lte is not None:
505
+ sheets["LTE_Verification"] = results_lte[0]
506
+
507
+ # Create Excel file
508
+ output = BytesIO()
509
+ with pd.ExcelWriter(output, engine="openpyxl") as writer:
510
+ for sheet_name, df in sheets.items():
511
+ df.to_excel(writer, sheet_name=sheet_name, index=False)
512
+
513
+ output.seek(0)
514
+ return sheets, output.getvalue()