#!/usr/bin/env python3 import math import re from collections import OrderedDict from typing import Dict, Iterable, List, Optional, Sequence, Tuple import pandas as pd CELL_ITEM_SEPARATOR = "\n" def normalize_numeric(value: object) -> object: if value is None or (isinstance(value, float) and math.isnan(value)): return "" if isinstance(value, str): text = value.strip() if not text or text.upper() == "NA": return "" return text if pd.isna(value): return "" if isinstance(value, float): return round(value, 3) return value def clean_compact_text(value: object) -> str: text = str(normalize_numeric(value)).strip() return "" if text == "nan" else text def format_search_scan_entry(name: str, fields: Sequence[Tuple[str, object]]) -> str: parts = [] for key, value in fields: text = clean_compact_text(value) if text: parts.append(f"{key}={text}") if not parts: return "" return f"{name}({','.join(parts)})" def clean_event_text(value: object, max_length: int = 180) -> str: text = re.sub(r"\s+", " ", clean_compact_text(value)).strip() if len(text) <= max_length: return text return text[: max_length - 3].rstrip() + "..." def choose_first_event_text(values: Iterable[object]) -> str: for value in values: text = clean_event_text(value) if text: return text return "" def append_search_scan_event( rows: List[Dict[str, object]], item: pd.Series, name: str, fields: Sequence[Tuple[str, object]], ) -> None: summary = format_search_scan_entry(name, fields) if summary: rows.append({"ts": item.get("ts"), "summary": summary}) def parse_int_text(value: object) -> Optional[int]: text = clean_compact_text(value) if not text: return None try: return int(float(text)) except ValueError: return None def build_range_text(values: Sequence[int]) -> str: if not values: return "" if len(values) == 1: return str(values[0]) return f"{values[0]}-{values[-1]}" def append_lte_rach_attempt_events( rows: List[Dict[str, object]], lte_rach_attempt: pd.DataFrame, ai_summary: bool, ) -> None: if lte_rach_attempt.empty: return if not ai_summary: for _, item in lte_rach_attempt.iterrows(): append_search_scan_event( rows, item, "LTE_RACH_ATTEMPT", [ ("retx", item.get("Retx_Counter")), ("result", item.get("Rach_Result")), ("contention", item.get("Contention_Procedure")), ("earfcn", item.get("Earfcn")), ("pmax", item.get("P_Max")), ("rsrp", item.get("Max_Serv_RSRP_Cal")), ("msg2", item.get("Msg2_Result")), ("ta", item.get("Msg2_TA_Value")), ], ) return typed_rows = [] for _, item in lte_rach_attempt.sort_values("ts").iterrows(): if pd.isna(item.get("ts")): continue retx = parse_int_text(item.get("Retx_Counter")) stable_fields = ( clean_compact_text(item.get("Rach_Result")), clean_compact_text(item.get("Contention_Procedure")), clean_compact_text(item.get("Earfcn")), clean_compact_text(item.get("P_Max")), clean_compact_text(item.get("Max_Serv_RSRP_Cal")), clean_compact_text(item.get("Msg2_Result")), clean_compact_text(item.get("Msg2_TA_Value")), ) typed_rows.append({ "ts": item.get("ts"), "second": item.get("ts").floor("s"), "retx": retx, "stable_fields": stable_fields, "item": item, }) current_segment = [] for record in typed_rows: if not current_segment: current_segment = [record] continue previous = current_segment[-1] is_same_segment = ( record["second"] == previous["second"] and record["stable_fields"] == previous["stable_fields"] and ( record["retx"] is None or previous["retx"] is None or record["retx"] == previous["retx"] + 1 ) ) if is_same_segment: current_segment.append(record) continue append_lte_rach_attempt_segment(rows, current_segment) current_segment = [record] append_lte_rach_attempt_segment(rows, current_segment) def append_lte_rach_attempt_segment(rows: List[Dict[str, object]], segment: Sequence[Dict[str, object]]) -> None: if not segment: return first = segment[0]["item"] retx_values = [item["retx"] for item in segment if item["retx"] is not None] entry_name = "LTE_RACH_ATTEMPT_SUMMARY" if len(segment) > 1 else "LTE_RACH_ATTEMPT" fields: List[Tuple[str, object]] = [] if len(segment) > 1: fields.extend([ ("retx", build_range_text(retx_values)), ("count", len(segment)), ]) else: fields.append(("retx", first.get("Retx_Counter"))) fields.extend([ ("result", first.get("Rach_Result")), ("contention", first.get("Contention_Procedure")), ("earfcn", first.get("Earfcn")), ("pmax", first.get("P_Max")), ("rsrp", first.get("Max_Serv_RSRP_Cal")), ("msg2", first.get("Msg2_Result")), ("ta", first.get("Msg2_TA_Value")), ]) if len(segment) > 1: fields.append(("summary", "stable_fields_no_change")) append_search_scan_event(rows, first, entry_name, fields) def split_plmn_req(value: object) -> Tuple[str, str]: text = clean_compact_text(value) match = re.match(r"^(\d+)-(\d+)$", text) if not match: return "", "" return match.group(1), match.group(2) def map_nas_rat(value: object) -> str: text = clean_compact_text(value) mapping = { "0": "GSM", "1": "WCDMA", "2": "LTE", "3": "TDS", "4": "NR5G", "5": "NBNTN", } return mapping.get(text, "") def map_nas_scan_scope(value: object) -> str: text = clean_compact_text(value) mapping = { "0": "full_scan", "1": "acq_db_scan", } return mapping.get(text, "") def map_sdss_script_meaning(value: object) -> str: script = clean_compact_text(value) mapping = { "ssscr_gw_opr_srv_info": "srv_info_continue", "ssscr_called_clr_acq_sched_lsts": "clear_acq_sched_lists", "ssscr_gw_opr_sys_lost": "service_lost", "ssscr_called_srv_lost_norm_slnt": "new_acq_int_srv_lost", "ssscr_int_srv_lost_rlf": "rlf_service_lost", "ssscr_called_srv_lost_rlf_scan": "rlf_scan", "oem_ssscr_misc_elev_opti_start_not_in_nr5g": "elevator_eoos_acq", "oem_ssscr_misc_elev_opti_start_srv_lost": "elevator_eoos_acq", "ssscr_misc_elev_opti_start_srv_lost": "elevator_eoos_acq", } return mapping.get(script, "") def map_sd_event_action(value: object) -> str: action = clean_compact_text(value) mapping = { "SDLOG_ACT_CONTINUE": "continue", "SDLOG_ACT_ACQ_GW": "acq_gw", "0": "continue", "11": "acq_gw", } return mapping.get(action, "") def build_search_scan_summary( nr5g_acq: pd.DataFrame, lte_system_scan: pd.DataFrame, lte_init_acq: pd.DataFrame, lte_band_scan: pd.DataFrame, nas_req_plmn: pd.DataFrame, rrc_reg_summary: pd.DataFrame, nr5g_rrc_config: pd.DataFrame, lte_rach_trigger: pd.DataFrame, lte_rach_attempt: pd.DataFrame, mmode_sdss_activate: Optional[pd.DataFrame] = None, sd_event_action: Optional[pd.DataFrame] = None, max_entries_per_second: int = 30, ai_summary: bool = True, ) -> pd.DataFrame: rows: List[Dict[str, object]] = [] mmode_sdss_activate = mmode_sdss_activate if mmode_sdss_activate is not None else pd.DataFrame() sd_event_action = sd_event_action if sd_event_action is not None else pd.DataFrame() for _, item in mmode_sdss_activate.iterrows(): append_search_scan_event( rows, item, "SDSS_ACT", [ ("script", item.get("Activate_Script")), ("meaning", map_sdss_script_meaning(item.get("Activate_Script"))), ("sub", item.get("Sub_ID")), ("stk", item.get("stk")), ("scr", item.get("scr")), ("line", item.get("Line")), ], ) for _, item in sd_event_action.iterrows(): append_search_scan_event( rows, item, "SD_EVENT", [ ("event", item.get("ss_event")), ("act", item.get("ss_act")), ("act_type", map_sd_event_action(item.get("ss_act"))), ("state", item.get("sd_state")), ], ) for _, item in nas_req_plmn.iterrows(): plmn_mcc, plmn_mnc = split_plmn_req(item.get("PLMN_REQ")) rows.append({ "ts": item.get("ts"), "summary": format_search_scan_entry( "NAS_REQ_PLMN", [ ("plmn", item.get("PLMN_REQ")), ("mcc", plmn_mcc), ("mnc", plmn_mnc), ("rat", map_nas_rat(item.get("RAT"))), ("rat_value", item.get("RAT")), ("trans_id", item.get("trans_id")), ("scan_scope", item.get("scan_scope")), ("scan_scope_type", map_nas_scan_scope(item.get("scan_scope"))), ], ), }) for _, item in nr5g_acq.iterrows(): rows.append({ "ts": item.get("ts"), "summary": format_search_scan_entry( "NR5G_ACQ", [ ("band", item.get("Raster_Band")), ("arfcn", item.get("Raster_ARFCN")), ("scs", item.get("Raster_SCS")), ("status", item.get("Raster_Status")), ("cells", item.get("Raster_NumCell")), ("acq", item.get("Raster_ACQ")), ("nb", item.get("Raster_MaxNBEnergy")), ("pci", item.get("Cell_PCI")), ("rsrp", item.get("Cell_RSRP")), ], ), }) for _, item in lte_system_scan.iterrows(): rows.append({ "ts": item.get("ts"), "summary": format_search_scan_entry( "LTE_SYSTEM_SCAN", [ ("n", item.get("Num_Candidates")), ("idx", item.get("Candidate_Index")), ("earfcn", item.get("EARFCN")), ("band", item.get("Band")), ("bw", item.get("Bandwidth")), ("energy", item.get("Energy")), ("nb", item.get("NB_Energy")), ("pruned", item.get("Pruned")), ], ), }) for _, item in lte_init_acq.iterrows(): rows.append({ "ts": item.get("ts"), "summary": format_search_scan_entry( "LTE_INIT_ACQ", [ ("earfcn", item.get("EARFCN")), ("band", item.get("Band")), ("result", item.get("Result")), ("search_results", item.get("Num_Search_Results")), ("blocked", item.get("Num_Blocked_Cells")), ("pci", item.get("Search_Physical_Cell_ID")), ("blocked_pci", item.get("Blocked_Physical_Cell_ID")), ("pbch_pci", item.get("PBCH_Physical_Cell_ID")), ("pbch", item.get("PBCH_Decode_Result")), ], ), }) for _, item in lte_band_scan.iterrows(): rows.append({ "ts": item.get("ts"), "summary": format_search_scan_entry( "LTE_BAND_SCAN", [ ("packet_band", item.get("Packet_Band")), ("n", item.get("Num_Candidates")), ("idx", item.get("Candidate_Index")), ("earfcn", item.get("EARFCN")), ("band", item.get("Candidate_Band")), ("bw", item.get("Bandwidth")), ("wb", item.get("WB_Energy")), ("pruned", item.get("Pruned")), ("low_energy", item.get("Low_Energy_System")), ], ), }) has_lte_rach_detail = not lte_rach_trigger.empty or not lte_rach_attempt.empty for _, item in rrc_reg_summary.iterrows(): if has_lte_rach_detail and clean_compact_text(item.get("LogId")) in {"0xB061", "0xB062"}: continue event_summary = choose_first_event_text([item.get("Summary"), item.get("REG"), item.get("Comment")]) append_search_scan_event( rows, item, "RRC_REG", [ ("log", item.get("LogId")), ("name", item.get("LogName")), ("summary", event_summary), ("freq", item.get("Freq")), ("pci", item.get("PCI")), ], ) for _, item in nr5g_rrc_config.iterrows(): state = clean_event_text(item.get("State")) if not state: continue append_search_scan_event( rows, item, "NR5G_RRC_STATE", [ ("state", state), ("config", item.get("Config_Status")), ("mode", item.get("Connectivity_Mode")), ("active_cc", item.get("Num_Active_CC")), ("band", item.get("Serving_Cell_Band")), ("cell", item.get("Serving_Cell_Id")), ("dl_arfcn", item.get("Serving_Cell_DL_Arfcn")), ], ) for _, item in lte_rach_trigger.iterrows(): append_search_scan_event( rows, item, "LTE_RACH_TRIGGER", [ ("reason", item.get("Rach_Reason")), ("contention", item.get("RACH_Contention")), ("preamble", item.get("Preamble")), ("init_power", item.get("Preamble_Initial_Power")), ("ramp", item.get("Power_Ramping_Step")), ("prach_cfg", item.get("PRACH_Config")), ("cc", choose_first_event_text([item.get("CC_Id"), item.get("Reason_CC_Id")])), ("summary", item.get("Summary")), ], ) append_lte_rach_attempt_events(rows, lte_rach_attempt, ai_summary=ai_summary) compact_rows = [ row for row in rows if row.get("summary") and not pd.isna(row.get("ts")) ] if not compact_rows: return pd.DataFrame() working = pd.DataFrame(compact_rows).sort_values("ts").reset_index(drop=True) working["second"] = working["ts"].dt.floor("s") output_rows: List[Dict[str, object]] = [] for second, group in working.groupby("second", sort=True): summaries = list(OrderedDict.fromkeys(group["summary"].tolist())) overflow = max(0, len(summaries) - max_entries_per_second) summaries = summaries[:max_entries_per_second] if overflow: summaries.append(f"more={overflow}") output_rows.append({"second": second, "Search_Scan": CELL_ITEM_SEPARATOR.join(summaries)}) return pd.DataFrame(output_rows).sort_values("second").reset_index(drop=True)