#!/usr/bin/env python3 import argparse import csv import math import re from collections import OrderedDict from pathlib import Path from typing import Dict, Iterable, List, Optional, Sequence, Tuple import numpy as np import pandas as pd from compact_rule_utils import ( build_horizontal_rules, load_transposed_rules, write_horizontal_rules, ) from radio_band_utils import derive_band_from_arfcn from search_scan_utils import build_search_scan_summary CELL_ITEM_SEPARATOR = "\n" AP_CELLINFO_PATTERN = re.compile(r"CellInfo\{(.+)\}") AP_KEY_VALUE_PATTERN = re.compile(r"(\w+)=([^,}]+)") AP_ENTER_EXIT_PATTERN = re.compile(r"EVENT_ELEVATOR_(ENTER|EXIT)\b") AP_PREDICT_FAIL_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+predict fail\b", re.IGNORECASE) AP_PREDICT_SUCCESS_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+predict success\b", re.IGNORECASE) AP_BAD_CELL_PATTERN = re.compile(r"EVENT_ELEVATOR_BAD_CELL_AVOIDANCE,\s*cause\s+([A-Z0-9_]+)") AP_TRIGGER_NO_NR_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+trigger no-NR return after elevator exit", re.IGNORECASE) AP_SET_DEPRI_NR5G_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+setDepriNr5g\b") AP_SET_L2NR_SELECTION_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+setL2NrSelection\b") AP_SET_DUBIOUS_CELL_LIST_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+setDubiousCellList\b") AP_SET_ELEVATOR_STATUS_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+setElevatorStatus,\s+elevatorStatus\s*=\s*(\d+)") AP_SET_ELEVATOR_STATUS_FAILED_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+setElevatorStatus failed\b", re.IGNORECASE) AP_OUT_OPT_PATTERN = re.compile(r"EVENT_ELEVATOR_OUT_OPT\b") AP_RESULT_PATTERN = re.compile(r"EVENT_ELEVATOR_RESULT\b") AP_DUBIOUS_RECOVER_PATTERN = re.compile(r"EVENT_ELEVATOR_DUBIOUS_RECOVER\b") AP_TRUE_SRV_STATUS_PATTERN = re.compile(r"\[EVENT_ELEVATOR\]\s+trueSrvStatus\s*=\s*([-\d]+)") AP_ELEVATOR_IDENTITY_PATTERN = re.compile( r"elevatorIdentity\s*=\s*([^,]+).*?kickInCount\s*=\s*(\d+).*?elevatorType\s*=\s*(\d+)" ) MD_ELEVATOR_TRANSITION_PATTERN = re.compile( r"RRC:\s*elevator state:\[Old:\s*(\d+)\s*New:\s*(\d+)\]", re.IGNORECASE, ) MD_ELEVATOR_MM_IND_PATTERN = re.compile( r"Elevator Enter mode indication received In-state\s*=\s*(\d+)", re.IGNORECASE, ) MD_ELEVATOR_NR_IND_PATTERN = re.compile( r"NR5G_RRC_ELEVATOR_INDI,\s*in_elevator_state\s*(\d+)", re.IGNORECASE, ) MD_ELEVATOR_LTE_STATE_PATTERN = re.compile( r"RRC:\s*in elevator state\s*\[(\d+)\]", re.IGNORECASE, ) MD_SET_ELEVATOR_STATUS_PATTERN = re.compile( r"set_elevator_status:\s*sub_id=\d+,\s*g_elevator_status=(\d+)", re.IGNORECASE, ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Generate 1-second elevator compact CSVs from existing AP and MD filter outputs.", ) parser.add_argument("--transposed-rules", required=True, help="Path to the 2-column CSV/TSV rule sheet or Markdown rule document.") parser.add_argument("--ap-dir", required=True, help="Path to the aplog_filter directory.") parser.add_argument("--md-dir", required=True, help="Path to the mdlog auto_analysis directory.") parser.add_argument("--date", required=True, help="Beijing date, for example 2026-06-24.") parser.add_argument("--windows", required=True, help="Comma-separated window labels, for example 1601_1602,1604_1605.") parser.add_argument("--output-dir", help="Directory used to write outputs.") parser.add_argument( "--window-output-dir", help="Direct output directory for a single window, for example an existing 1601_1602 folder.", ) parser.add_argument( "--ai-summary", dest="ai_summary", action="store_true", default=True, help="Enable deterministic Search_Scan summaries for repetitive typed events. Enabled by default.", ) parser.add_argument( "--no-ai-summary", dest="ai_summary", action="store_false", help="Disable Search_Scan summaries and keep raw per-event entries.", ) parser.add_argument( "--performance-csv", help="Optional Douyin live stutter summary CSV, for example mp4_SmoothedTestSummary.csv.", ) args = parser.parse_args() if not args.output_dir and not args.window_output_dir: parser.error("one of --output-dir or --window-output-dir is required") return args def parse_window_label(base_date: pd.Timestamp, label: str) -> Tuple[pd.Timestamp, pd.Timestamp]: parts = [item for item in label.split("_") if item] if len(parts) < 2: raise ValueError(f"Bad window label: {label}") start_label, end_label = parts[-2], parts[-1] if not (re.fullmatch(r"\d{4}|\d{6}", start_label) and re.fullmatch(r"\d{4}|\d{6}", end_label)): raise ValueError(f"Bad window label time token: {label}") start_second = int(start_label[4:6]) if len(start_label) == 6 else 0 end_second = int(end_label[4:6]) if len(end_label) == 6 else 59 start_ts = base_date.replace( hour=int(start_label[0:2]), minute=int(start_label[2:4]), second=start_second, ) end_ts = base_date.replace( hour=int(end_label[0:2]), minute=int(end_label[2:4]), second=end_second, ) return start_ts, end_ts def parse_ap_timestamp(line: str, base_date: pd.Timestamp) -> Optional[pd.Timestamp]: if len(line) < 18: return None prefix = line[:18] try: month = int(prefix[0:2]) day = int(prefix[3:5]) hour = int(prefix[6:8]) minute = int(prefix[9:11]) second = int(prefix[12:14]) millisecond = int(prefix[15:18]) return pd.Timestamp( year=base_date.year, month=month, day=day, hour=hour, minute=minute, second=second, microsecond=millisecond * 1000, ) except ValueError: return None 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 sanitize_serving_value(value: object) -> object: text = str(value).strip() if text in {"2147483647", "9223372036854775807", "NA", ""}: return "" return value def detect_datetime_columns(df: pd.DataFrame) -> Tuple[str, str]: date_col = "Date" if "Date" in df.columns else "date" time_col = "Time" if "Time" in df.columns else "time" return date_col, time_col def load_md_csv(path: Path, time_is_utc: bool = True) -> pd.DataFrame: if not path.exists(): return pd.DataFrame() df = pd.read_csv(path) if df.empty: return df date_col, time_col = detect_datetime_columns(df) df["ts"] = pd.to_datetime( df[date_col].astype(str).str.strip() + " " + df[time_col].astype(str).str.strip(), errors="coerce", ) if time_is_utc: df["ts"] = df["ts"] + pd.Timedelta(hours=8) df = df.dropna(subset=["ts"]).sort_values("ts").reset_index(drop=True) return df def load_md_csv_by_patterns(md_dir: Path, patterns: Sequence[str], time_is_utc: bool = True) -> pd.DataFrame: for pattern in patterns: matches = sorted(md_dir.glob(pattern)) if matches: return load_md_csv(matches[0], time_is_utc=time_is_utc) return pd.DataFrame() def load_md_csvs_by_patterns(md_dir: Path, patterns: Sequence[str], time_is_utc: bool = True) -> pd.DataFrame: frames: List[pd.DataFrame] = [] seen: set = set() for pattern in patterns: for match in sorted(md_dir.glob(pattern)): if match in seen: continue seen.add(match) frame = load_md_csv(match, time_is_utc=time_is_utc) if not frame.empty: frames.append(frame) if not frames: return pd.DataFrame() return pd.concat(frames, ignore_index=True).sort_values("ts").reset_index(drop=True) def load_md_csvs_with_fallback( md_dir: Path, preferred_patterns: Sequence[str], fallback_patterns: Sequence[str], time_is_utc: bool = True, ) -> pd.DataFrame: preferred = load_md_csvs_by_patterns(md_dir, preferred_patterns, time_is_utc=time_is_utc) if not preferred.empty: return preferred return load_md_csvs_by_patterns(md_dir, fallback_patterns, time_is_utc=time_is_utc) def resolve_first_existing_path(base_dir: Path, candidates: Sequence[str]) -> Path: for candidate in candidates: path = base_dir / candidate if path.exists(): return path return base_dir / candidates[0] def extract_ap_event_tags(line: str) -> List[str]: tags: List[str] = [] match = AP_ENTER_EXIT_PATTERN.search(line) if match: direction = match.group(1) tags.append("enter_elevator" if direction == "ENTER" else "exit_elevator") if AP_PREDICT_FAIL_PATTERN.search(line): tags.append("predict_fail") if AP_PREDICT_SUCCESS_PATTERN.search(line): tags.append("predict_success") match = AP_BAD_CELL_PATTERN.search(line) if match: tags.append(f"bad_cell_cause:{match.group(1)}") if AP_TRIGGER_NO_NR_PATTERN.search(line): tags.append("trigger_no_nr_return") if AP_SET_DEPRI_NR5G_PATTERN.search(line): tags.append("set_depri_nr5g") if AP_SET_L2NR_SELECTION_PATTERN.search(line): tags.append("set_l2nr_selection") if AP_SET_DUBIOUS_CELL_LIST_PATTERN.search(line): tags.append("set_dubious_cell_list") match = AP_SET_ELEVATOR_STATUS_PATTERN.search(line) if match: tags.append(f"set_elevator_status:{match.group(1)}") if AP_SET_ELEVATOR_STATUS_FAILED_PATTERN.search(line): tags.append("set_elevator_status_failed") if AP_OUT_OPT_PATTERN.search(line): tags.append("out_opt") if AP_RESULT_PATTERN.search(line): tags.append("elevator_result") if AP_DUBIOUS_RECOVER_PATTERN.search(line): tags.append("dubious_recover") match = AP_TRUE_SRV_STATUS_PATTERN.search(line) if match: tags.append(f"true_srv_status:{match.group(1)}") match = AP_ELEVATOR_IDENTITY_PATTERN.search(line) if match: tags.append( "elevator_identity:" f"{match.group(1)},kickInCount={match.group(2)},elevatorType={match.group(3)}" ) deduped = list(OrderedDict.fromkeys(tags)) return deduped def load_ap_key_events(path: Path, base_date: pd.Timestamp) -> pd.DataFrame: rows: List[Dict[str, object]] = [] if not path.exists(): return pd.DataFrame(columns=["ts", "event_tag", "event_source"]) with path.open("r", encoding="utf-8") as fp: for line in fp: ts = parse_ap_timestamp(line, base_date) if ts is None: continue for tag in extract_ap_event_tags(line): rows.append({"ts": ts, "event_tag": tag, "event_source": "ap"}) if not rows: return pd.DataFrame(columns=["ts", "event_tag", "event_source"]) return pd.DataFrame(rows).sort_values("ts").reset_index(drop=True) def _append_elevator_enter_exit_tag(tags: List[str], state_value: str, exit_state_values: Sequence[str]) -> None: state_text = str(state_value).strip() if state_text == "1": tags.append("enter_elevator") elif state_text in exit_state_values: tags.append("exit_elevator") def extract_md_elevator_event_tags(summary: object) -> List[str]: text = str(summary or "").strip().strip('"') tags: List[str] = [] if not text: return tags match = MD_ELEVATOR_TRANSITION_PATTERN.search(text) if match: _append_elevator_enter_exit_tag(tags, match.group(2), ("0",)) match = MD_SET_ELEVATOR_STATUS_PATTERN.search(text) if match: status = match.group(1) tags.append(f"set_elevator_status:{status}") _append_elevator_enter_exit_tag(tags, status, ("0", "2")) match = MD_ELEVATOR_MM_IND_PATTERN.search(text) if match: _append_elevator_enter_exit_tag(tags, match.group(1), ("0",)) match = MD_ELEVATOR_NR_IND_PATTERN.search(text) if match: _append_elevator_enter_exit_tag(tags, match.group(1), ("0",)) match = MD_ELEVATOR_LTE_STATE_PATTERN.search(text) if match and match.group(1).strip() == "1": tags.append("inner_elevator") return list(OrderedDict.fromkeys(tags)) def load_md_elevator_events(md_dir: Path) -> pd.DataFrame: event_df = load_md_csvs_by_patterns( md_dir, [ "ELEVATOR_STATUS_*.csv", "ELEVATOR_STATUS.csv", "sim1_ELEVATOR_STATUS_*.csv", "sim1_ELEVATOR_STATUS.csv", "sim_unknown_ELEVATOR_STATUS_*.csv", "sim_unknown_ELEVATOR_STATUS.csv", "_primary_csv/ELEVATOR_STATUS_*.csv", "_primary_csv/ELEVATOR_STATUS.csv", "backup/ELEVATOR_STATUS_*.csv", "backup/ELEVATOR_STATUS.csv", ], ) if event_df.empty or "Summary" not in event_df.columns: return pd.DataFrame(columns=["ts", "event_tag", "event_source"]) rows: List[Dict[str, object]] = [] for _, item in event_df.iterrows(): ts = item.get("ts") if pd.isna(ts): continue for tag in extract_md_elevator_event_tags(item.get("Summary", "")): rows.append({"ts": ts, "event_tag": tag, "event_source": "md"}) if not rows: return pd.DataFrame(columns=["ts", "event_tag", "event_source"]) return pd.DataFrame(rows).sort_values("ts").reset_index(drop=True) def load_ap_serving_cell(path: Path, base_date: pd.Timestamp) -> pd.DataFrame: rows: List[Dict[str, object]] = [] if not path.exists(): return pd.DataFrame(columns=["ts"]) with path.open("r", encoding="utf-8") as fp: for line in fp: ts = parse_ap_timestamp(line, base_date) if ts is None: continue match = AP_CELLINFO_PATTERN.search(line) if not match: continue payload = match.group(1) values: Dict[str, object] = {"ts": ts} for key, raw_value in AP_KEY_VALUE_PATTERN.findall(payload): values[key] = raw_value.strip() rows.append(values) if not rows: return pd.DataFrame(columns=["ts"]) df = pd.DataFrame(rows).sort_values("ts").reset_index(drop=True) return df def parse_performance_time(value: object, base_date: pd.Timestamp) -> Optional[pd.Timestamp]: text = str(value).strip() if not text: return None ts = pd.to_datetime(f"{base_date.date()} {text}", errors="coerce") if pd.isna(ts): return None return pd.Timestamp(ts) def map_performance_label(value: object) -> Tuple[str, int]: label = pd.to_numeric(value, errors="coerce") if pd.isna(label): return "", -1 label_int = int(label) if label_int == 1: return "douyin_live_stutter", 100 if label_int == 0: return "douyin_live_smooth", 0 return f"douyin_live_label_{label_int}", 50 def load_performance_summary(path: Optional[str], base_date: pd.Timestamp) -> pd.DataFrame: if not path: return pd.DataFrame(columns=["second", "Performance"]) csv_path = Path(path) if not csv_path.exists(): return pd.DataFrame(columns=["second", "Performance"]) df = pd.read_csv(csv_path) required_columns = {"start_time", "end_time", "label"} if df.empty or not required_columns.issubset(df.columns): return pd.DataFrame(columns=["second", "Performance"]) labels_by_second: Dict[pd.Timestamp, Tuple[str, int]] = {} for _, item in df.iterrows(): start_ts = parse_performance_time(item.get("start_time"), base_date) end_ts = parse_performance_time(item.get("end_time"), base_date) label_text, label_priority = map_performance_label(item.get("label")) if start_ts is None or end_ts is None or not label_text or end_ts <= start_ts: continue cursor = start_ts.floor("s") last_second = (end_ts - pd.Timedelta(microseconds=1)).floor("s") while cursor <= last_second: second_start = cursor second_end = cursor + pd.Timedelta(seconds=1) if second_start < end_ts and second_end > start_ts: previous = labels_by_second.get(cursor) if previous is None or label_priority > previous[1]: labels_by_second[cursor] = (label_text, label_priority) cursor += pd.Timedelta(seconds=1) if not labels_by_second: return pd.DataFrame(columns=["second", "Performance"]) return pd.DataFrame( { "second": sorted(labels_by_second.keys()), "Performance": [labels_by_second[second][0] for second in sorted(labels_by_second.keys())], } ) def group_md_mean(df: pd.DataFrame, numeric_columns: Sequence[str], filters: Optional[Dict[str, object]] = None) -> pd.DataFrame: if df.empty: return pd.DataFrame() working = df.copy() if filters: for column, expected in filters.items(): if column not in working.columns: return pd.DataFrame() working = working[working[column].astype(str) == str(expected)] if working.empty: return pd.DataFrame() valid_columns = [column for column in numeric_columns if column in working.columns] if not valid_columns: return pd.DataFrame() for column in valid_columns: working[column] = pd.to_numeric(working[column], errors="coerce") grouped = working.groupby(working["ts"].dt.floor("s"))[valid_columns].mean().reset_index() grouped = grouped.rename(columns={"ts": "second"}) return grouped def group_md_sum(df: pd.DataFrame, numeric_columns: Sequence[str], filters: Optional[Dict[str, object]] = None) -> pd.DataFrame: if df.empty: return pd.DataFrame() working = df.copy() if filters: for column, expected in filters.items(): if column not in working.columns: return pd.DataFrame() working = working[working[column].astype(str) == str(expected)] if working.empty: return pd.DataFrame() valid_columns = [column for column in numeric_columns if column in working.columns] if not valid_columns: return pd.DataFrame() for column in valid_columns: working[column] = pd.to_numeric(working[column], errors="coerce") grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second")[valid_columns] .sum(min_count=1) .reset_index() .sort_values("second") ) return grouped def group_md_last(df: pd.DataFrame, columns: Sequence[str], filters: Optional[Dict[str, object]] = None) -> pd.DataFrame: if df.empty: return pd.DataFrame() working = df.copy() if filters: for column, expected in filters.items(): if column not in working.columns: return pd.DataFrame() working = working[working[column].astype(str) == str(expected)] if working.empty: return pd.DataFrame() valid_columns = [column for column in columns if column in working.columns] if not valid_columns: return pd.DataFrame() grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second")[valid_columns] .last() .reset_index() .sort_values("second") ) return grouped def group_md_count( df: pd.DataFrame, output_column: str, filters: Optional[Dict[str, object]] = None, ) -> pd.DataFrame: if df.empty: return pd.DataFrame() working = df.copy() if filters: for column, expected in filters.items(): if column not in working.columns: return pd.DataFrame() working = working[working[column].astype(str) == str(expected)] if working.empty: return pd.DataFrame() grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second") .size() .reset_index(name=output_column) .sort_values("second") ) return grouped def group_md_distinct_count( df: pd.DataFrame, source_column: str, output_column: str, filters: Optional[Dict[str, object]] = None, ) -> pd.DataFrame: if df.empty or source_column not in df.columns: return pd.DataFrame() working = df.copy() if filters: for column, expected in filters.items(): if column not in working.columns: return pd.DataFrame() working = working[working[column].astype(str) == str(expected)] if working.empty: return pd.DataFrame() working[source_column] = working[source_column].astype(str).str.strip() working = working[working[source_column] != ""] if working.empty: return pd.DataFrame() grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second")[source_column] .nunique() .reset_index(name=output_column) .sort_values("second") ) return grouped def rename_prefixed(df: pd.DataFrame, prefix: str) -> pd.DataFrame: if df.empty: return df renamed = {} for column in df.columns: if column != "second": renamed[column] = f"{prefix}{column}" return df.rename(columns=renamed) def merge_exact_second(timeline: pd.DataFrame, df: pd.DataFrame) -> pd.DataFrame: if df.empty: return timeline return timeline.merge(df, on="second", how="left") def merge_asof_second(timeline: pd.DataFrame, df: pd.DataFrame) -> pd.DataFrame: if df.empty: return timeline return pd.merge_asof( timeline.sort_values("second"), df.sort_values("second"), on="second", direction="backward", ) def map_ap_rat(value: object) -> str: text = str(value).strip() if text == "NGRAN": return "NR" if text == "EUTRAN": return "LTE" return "" def map_ap_network_type(value: object) -> str: text = normalize_numeric(value) if str(text).strip() == "20": return "NR" if str(text).strip() == "13": return "LTE" return "" def map_ap_scs(value: object) -> str: text = str(value).strip() if text == "0": return "15kHz" if text == "1": return "30kHz" return "" def map_md_scs(value: object) -> str: text = str(value).strip().upper().replace(" ", "") if text == "15KHZ": return "15kHz" if text == "30KHZ": return "30kHz" return "" def parse_scs_to_khz(value: object) -> float: text = str(value).strip().upper().replace(" ", "") if text in {"15KHZ", "15"}: return 15.0 if text in {"30KHZ", "30"}: return 30.0 if text == "0": return 15.0 if text == "1": return 30.0 return float("nan") def map_modulation_order(value: object) -> object: text = str(value).strip().upper() if not text or text == "NA": return "" mapping = { "QPSK": 2, "16QAM": 4, "64QAM": 6, "256QAM": 8, } if text in mapping: return mapping[text] numeric = pd.to_numeric(text, errors="coerce") if pd.notna(numeric): return float(numeric) return "" def map_ul_layers_pusch(value: object) -> object: text = str(value).strip().upper() if not text or text == "NA": return "" mapping = { "SISO": 1, "SMSL": 1, "SMDL": 2, "1TX": 1, "2TX": 2, } if text in mapping: return mapping[text] numeric = pd.to_numeric(text, errors="coerce") if pd.notna(numeric): return float(numeric) return "" def map_dl_rx_num(value: object) -> object: text = str(value).strip().upper() if not text or text == "NA": return "" mapping = { "2X2_MIMO": 2, "4X4_MIMO": 4, } if text in mapping: return mapping[text] numeric = pd.to_numeric(text, errors="coerce") if pd.notna(numeric): return float(numeric) return "" def _cdrx_slots_per_frame_from_scs(value: object) -> float: scs_khz = parse_scs_to_khz(value) if not np.isfinite(scs_khz) or scs_khz <= 0: return float("nan") return scs_khz / 15.0 def merge_frames_on_second(frames: Sequence[pd.DataFrame]) -> pd.DataFrame: merged: Optional[pd.DataFrame] = None for frame in frames: if frame is None or frame.empty: continue if merged is None: merged = frame.copy() else: merged = merged.merge(frame, on="second", how="outer", suffixes=("", "__dup")) duplicate_columns = [column for column in merged.columns if column.endswith("__dup")] for duplicate_column in duplicate_columns: base_column = duplicate_column[:-5] if base_column in merged.columns: merged[base_column] = merged[base_column].where(merged[base_column].notna(), merged[duplicate_column]) else: merged[base_column] = merged[duplicate_column] merged = merged.drop(columns=[duplicate_column]) return merged if merged is not None else pd.DataFrame() def build_nr_ul_metrics(nr_mac_ul: pd.DataFrame) -> pd.DataFrame: if nr_mac_ul.empty: return pd.DataFrame() working = nr_mac_ul[nr_mac_ul["Channel"].astype(str).str.strip() == "PUSCH"].copy() if working.empty: return pd.DataFrame() working["Num_RBs"] = pd.to_numeric(working["Num_RBs"], errors="coerce") working["PUSCH_TB_Size_bytes"] = pd.to_numeric(working["PUSCH_TB_Size_bytes"], errors="coerce") working["UL_TX_NUM"] = pd.to_numeric(working["UL_TX_NUM"], errors="coerce") working["UL_LAYERS_PUSCH"] = working["PUSCH_TX_Mode"].apply(map_ul_layers_pusch) working["UL_ModType"] = working["PUSCH_Modulation_Order"].apply(map_modulation_order) working["UL_Bandwidth"] = working["Num_RBs"] * working["SCS"].apply(parse_scs_to_khz) * 12.0 / 1000.0 working["UL_TPUT_PUSCH"] = working["PUSCH_TB_Size_bytes"] * 8.0 / 1_000_000.0 grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second") .agg( UL_Bandwidth=("UL_Bandwidth", "mean"), UL_TPUT_PUSCH=("UL_TPUT_PUSCH", "sum"), UL_LAYERS_PUSCH=("UL_LAYERS_PUSCH", "mean"), UL_ModType=("UL_ModType", "mean"), PUSCH_NUM_RB_AVG=("Num_RBs", "mean"), UL_TX_NUM=("UL_TX_NUM", "mean"), ) .reset_index() .sort_values("second") ) return grouped def build_lte_ul_metrics(lte_mac_tx: pd.DataFrame) -> pd.DataFrame: if lte_mac_tx.empty: return pd.DataFrame() working = lte_mac_tx[lte_mac_tx["Channel"].astype(str).str.strip() == "PUSCH"].copy() if working.empty: return pd.DataFrame() working["Num_RBs"] = pd.to_numeric(working["Num_RBs"], errors="coerce") working["TB_size_bytes"] = pd.to_numeric(working["TB_size_bytes"], errors="coerce") working["UL_Bandwidth"] = working["Num_RBs"] * 12.0 * 15.0 / 1000.0 working["UL_TPUT_PUSCH"] = working["TB_size_bytes"] * 8.0 / 1_000_000.0 grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second") .agg( UL_Bandwidth=("UL_Bandwidth", "mean"), UL_TPUT_PUSCH=("UL_TPUT_PUSCH", "sum"), PUSCH_NUM_RB_AVG=("Num_RBs", "mean"), ) .reset_index() .sort_values("second") ) return grouped def build_nr_dl_metrics(nr_mac_pdsch: pd.DataFrame) -> pd.DataFrame: if nr_mac_pdsch.empty: return pd.DataFrame() working = nr_mac_pdsch.copy() working["Rbs"] = pd.to_numeric(working["Rbs"], errors="coerce") working["TB_size"] = pd.to_numeric(working["TB_size"], errors="coerce") working["Layers"] = pd.to_numeric(working["Layers"], errors="coerce") working["MCS"] = pd.to_numeric(working["MCS"], errors="coerce") working["Num_RX"] = working["Num_RX"].apply(map_dl_rx_num) working["DL_ModType"] = working["Mod_Type"].apply(map_modulation_order) working["DL_Bandwidth"] = working["Rbs"] * working["SCS_MU"].apply(parse_scs_to_khz) * 12.0 / 1000.0 working["DL_TPUT_PDSCH"] = working["TB_size"] * 8.0 / 1_000_000.0 grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second") .agg( DL_Bandwidth=("DL_Bandwidth", "mean"), DL_TPUT_PDSCH=("DL_TPUT_PDSCH", "sum"), DL_NUM_RB_AVG_PDSCH=("Rbs", "mean"), DL_Layers=("Layers", "mean"), DL_MCS=("MCS", "mean"), DL_ModType=("DL_ModType", "mean"), DL_RX_NUM=("Num_RX", "mean"), ) .reset_index() .sort_values("second") ) return grouped def build_lte_dl_metrics(lte_mac_pdsch: pd.DataFrame) -> pd.DataFrame: if lte_mac_pdsch.empty: return pd.DataFrame() working = lte_mac_pdsch.copy() working["RBs"] = pd.to_numeric(working["RBs"], errors="coerce") working["TB_size_bytes"] = pd.to_numeric(working["TB_size_bytes"], errors="coerce") working["Layers"] = pd.to_numeric(working["Layers"], errors="coerce") working["MCS"] = pd.to_numeric(working["MCS"], errors="coerce") working["DL_ModType"] = working["Mod_type"].apply(map_modulation_order) working["DL_Bandwidth"] = working["RBs"] * 12.0 * 15.0 / 1000.0 working["DL_TPUT_PDSCH"] = working["TB_size_bytes"] * 8.0 / 1_000_000.0 grouped = ( working.assign(second=working["ts"].dt.floor("s")) .groupby("second") .agg( DL_Bandwidth=("DL_Bandwidth", "mean"), DL_TPUT_PDSCH=("DL_TPUT_PDSCH", "sum"), DL_NUM_RB_AVG_PDSCH=("RBs", "mean"), DL_Layers=("Layers", "mean"), DL_MCS=("MCS", "mean"), DL_ModType=("DL_ModType", "mean"), ) .reset_index() .sort_values("second") ) return grouped def build_cdrx_inactive_metrics(cdrx_df: pd.DataFrame, offset_hours: float = 8.0) -> pd.DataFrame: if cdrx_df.empty: return pd.DataFrame() required_columns = {"Timestamp", "SCS", "SFN", "Slot", "CurState"} if not required_columns.issubset(cdrx_df.columns): return pd.DataFrame() working = cdrx_df[list(required_columns)].copy() working["Timestamp"] = pd.to_numeric(working["Timestamp"], errors="coerce") working["SFN"] = pd.to_numeric(working["SFN"], errors="coerce") working["Slot"] = pd.to_numeric(working["Slot"], errors="coerce") working["__slots_per_frame__"] = working["SCS"].apply(_cdrx_slots_per_frame_from_scs) working["__slot_duration_ms__"] = 10.0 / working["__slots_per_frame__"] valid = working[ working["Timestamp"].notna() & working["SFN"].notna() & working["Slot"].notna() & working["__slots_per_frame__"].notna() & (working["Slot"] >= 0) & (working["Slot"] < working["__slots_per_frame__"]) ].copy() if valid.empty: return pd.DataFrame() cycle_ms = 1024.0 * 10.0 wraps = 0 prev_cycle_pos_ms: Optional[float] = None abs_cycle_pos_ms: List[float] = [] for sfn_value, slot_value, slot_duration_ms in zip( valid["SFN"].to_numpy(dtype=float), valid["Slot"].to_numpy(dtype=float), valid["__slot_duration_ms__"].to_numpy(dtype=float), ): cycle_pos_ms = float(sfn_value) * 10.0 + float(slot_value) * float(slot_duration_ms) if prev_cycle_pos_ms is not None and cycle_pos_ms < prev_cycle_pos_ms - 1e-9: wraps += 1 abs_cycle_pos_ms.append(wraps * cycle_ms + cycle_pos_ms) prev_cycle_pos_ms = cycle_pos_ms valid["__abs_cycle_pos_ms__"] = abs_cycle_pos_ms packet_endpoints = valid.groupby("Timestamp", sort=False)["__abs_cycle_pos_ms__"].max().reset_index() intercept_sec = (packet_endpoints["Timestamp"] - packet_endpoints["__abs_cycle_pos_ms__"] / 1000.0).mean() if pd.isna(intercept_sec): return pd.DataFrame() valid["__wall_clock_sec__"] = intercept_sec + valid["__abs_cycle_pos_ms__"] / 1000.0 upper_states = valid["CurState"].astype(str).str.strip().str.upper().to_numpy() wall_clock_sec = valid["__wall_clock_sec__"].to_numpy(dtype=float) bin_totals_ms: Dict[float, float] = {} for index, state in enumerate(upper_states[:-1]): if state != "INACTIVE": continue start_sec = float(wall_clock_sec[index]) + offset_hours * 3600.0 end_sec = float(wall_clock_sec[index + 1]) + offset_hours * 3600.0 if not np.isfinite(start_sec) or not np.isfinite(end_sec) or end_sec <= start_sec: continue cursor = start_sec while cursor < end_sec - 1e-12: bin_start = math.floor(cursor) bin_end = bin_start + 1.0 overlap_sec = min(end_sec, bin_end) - cursor if overlap_sec > 0: bin_totals_ms[bin_start] = bin_totals_ms.get(bin_start, 0.0) + overlap_sec * 1000.0 cursor = min(end_sec, bin_end) if not bin_totals_ms: return pd.DataFrame() return pd.DataFrame( { "second": [pd.to_datetime(epoch, unit="s") for epoch in sorted(bin_totals_ms.keys())], "CDRX_Inactive_ms": [bin_totals_ms[epoch] for epoch in sorted(bin_totals_ms.keys())], } ) def map_srv_status(ap_is_oos: object, md_srv_status: object) -> object: md_status = normalize_numeric(md_srv_status) if md_status != "": return md_status text = str(ap_is_oos).strip().lower() if text == "true": return "isOos" if text == "false": return "InService" return "" def choose_first_non_empty(values: Iterable[object]) -> object: for value in values: normalized = normalize_numeric(value) if normalized != "": return normalized return "" def has_compact_value(value: object) -> bool: return normalize_numeric(value) != "" def has_any_compact_value(item: pd.Series, columns: Sequence[str]) -> bool: return any(has_compact_value(item.get(column)) for column in columns) def choose_serving_tuple(item: pd.Series) -> Dict[str, object]: nr_has_exact_sample = has_any_compact_value( item, [ "md_nr_mean_RSRP_0", "md_nr_mean_RSRP_1", "md_nr_mean_RSRP_2", "md_nr_mean_RSRP_3", "md_nr_snr_SNR_0", "md_nr_snr_SNR_1", "md_nr_snr_SNR_2", "md_nr_snr_SNR_3", ], ) lte_has_exact_sample = has_any_compact_value( item, [ "md_lte_mean_RSRP_0", "md_lte_mean_RSRP_1", "md_lte_mean_RSRP_2", "md_lte_mean_RSRP_3", "md_lte_mean_SNR_0", "md_lte_mean_SNR_1", "md_lte_mean_SNR_2", "md_lte_mean_SNR_3", ], ) nr_freq = sanitize_serving_value(item.get("md_nr_state_RARFCN")) lte_freq = sanitize_serving_value(item.get("md_lte_state_EARFCN")) nr_pci = sanitize_serving_value(item.get("md_nr_state_PCI")) lte_pci = sanitize_serving_value(item.get("md_lte_state_PCI")) ap_rat = choose_first_non_empty([map_ap_network_type(item.get("networkType")), map_ap_rat(item.get("ratType"))]) md_rat = "" if has_compact_value(nr_freq) and not has_compact_value(lte_freq): md_rat = "NR" elif has_compact_value(lte_freq) and not has_compact_value(nr_freq): md_rat = "LTE" elif has_compact_value(nr_freq) and has_compact_value(lte_freq): if nr_has_exact_sample and not lte_has_exact_sample: md_rat = "NR" elif lte_has_exact_sample and not nr_has_exact_sample: md_rat = "LTE" elif nr_has_exact_sample and lte_has_exact_sample: md_rat = "NR" else: md_rat = "NR" selected_rat = "" if md_rat: selected_rat = md_rat else: selected_rat = ap_rat if selected_rat == "LTE": freq = lte_freq pci = lte_pci scs = "15kHz" band = choose_first_non_empty([derive_band_from_arfcn("LTE", freq), item.get("band")]) elif selected_rat == "NR": freq = nr_freq pci = nr_pci scs = map_md_scs(item.get("md_nr_state_SCS")) band = choose_first_non_empty([item.get("md_rrc_Serving_Cell_Band"), derive_band_from_arfcn("NR", freq)]) else: freq = sanitize_serving_value(item.get("freq")) pci = sanitize_serving_value(item.get("pci")) scs = map_ap_scs(item.get("scs")) band = choose_first_non_empty([item.get("band"), derive_band_from_arfcn(ap_rat, freq)]) if not has_compact_value(freq): freq = sanitize_serving_value(item.get("freq")) if not has_compact_value(pci): pci = sanitize_serving_value(item.get("pci")) if not has_compact_value(band): band = choose_first_non_empty([item.get("band"), derive_band_from_arfcn(ap_rat, freq)]) if not has_compact_value(scs): scs = map_ap_scs(item.get("scs")) if not has_compact_value(scs) and selected_rat == "LTE": scs = "15kHz" return { "rat": selected_rat, "freq": freq, "pci": pci, "scs": scs, "band": band, } def derive_key_phase( ts: pd.Timestamp, enter_ts: Optional[pd.Timestamp], exit_ts: Optional[pd.Timestamp], ) -> str: if enter_ts is None: return "" second_ts = ts.floor("s") if second_ts < enter_ts.floor("s"): return "before_elevator" if second_ts == enter_ts.floor("s"): return "enter_elevator" if exit_ts is not None and second_ts >= exit_ts.floor("s"): return "exit_elevator" return "inner_elevator" def build_key_event_value( base_phase: str, manual_tags: Sequence[str], md_tags: Sequence[str], ap_tags: Sequence[str], ) -> str: ordered: List[str] = [] for tag in [base_phase] + list(manual_tags) + list(md_tags) + list(ap_tags): if tag and tag not in ordered: ordered.append(tag) return CELL_ITEM_SEPARATOR.join(ordered) def build_manual_elevator_tags( ts: pd.Timestamp, manual_enter_ts: pd.Timestamp, manual_exit_ts: pd.Timestamp, ) -> List[str]: second_ts = ts.floor("s") tags: List[str] = [] if second_ts == manual_enter_ts.floor("s"): tags.append("manual_enter_elevator") if second_ts == manual_exit_ts.floor("s"): tags.append("manual_exit_elevator") return tags def format_output_value(value: object) -> object: normalized = normalize_numeric(value) if normalized == "": return "" return normalized def write_csv(path: Path, headers: Sequence[str], rows: Sequence[Dict[str, object]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8", newline="") as fp: writer = csv.DictWriter(fp, fieldnames=list(headers), extrasaction="ignore") writer.writeheader() for row in rows: writer.writerow(row) def coverage_for_rows(headers: Sequence[str], rows: Sequence[Dict[str, object]]) -> Tuple[List[str], List[str]]: filled: List[str] = [] missing: List[str] = [] for header in headers: if any(str(row.get(header, "")).strip() for row in rows): filled.append(header) else: missing.append(header) return filled, missing def build_window_rows( label: str, start_ts: pd.Timestamp, end_ts: pd.Timestamp, headers: Sequence[str], ap_key_events: pd.DataFrame, md_elevator_events: pd.DataFrame, ap_serving_cell: pd.DataFrame, nr_ml1_mean: pd.DataFrame, nr_ml1_state: pd.DataFrame, nr_snr_mean: pd.DataFrame, lte_ml1_mean: pd.DataFrame, lte_ml1_state: pd.DataFrame, srv_ue_status: pd.DataFrame, nr5g_rrc_config: pd.DataFrame, exact_frames: Sequence[pd.DataFrame], state_frames: Sequence[pd.DataFrame], search_scan: pd.DataFrame, performance_summary: pd.DataFrame, ) -> Tuple[List[Dict[str, object]], Dict[str, object]]: seconds = pd.date_range(start_ts, end_ts, freq="1s") timeline = pd.DataFrame({"second": seconds}) event_frames: List[pd.DataFrame] = [] for event_df in (ap_key_events, md_elevator_events): if event_df.empty: continue window_df = event_df[(event_df["ts"] >= start_ts) & (event_df["ts"] <= end_ts)].copy() if not window_df.empty: event_frames.append(window_df) if event_frames: window_events = pd.concat(event_frames, ignore_index=True).sort_values("ts").reset_index(drop=True) else: window_events = pd.DataFrame(columns=["ts", "event_tag", "event_source"]) md_window_events = window_events[window_events.get("event_source", "") == "md"] if not window_events.empty else window_events phase_events = md_window_events if not md_window_events.empty else window_events enter_rows = phase_events[phase_events["event_tag"] == "enter_elevator"] enter_ts = enter_rows.iloc[0]["ts"] if not enter_rows.empty else None exit_rows = phase_events[phase_events["event_tag"] == "exit_elevator"] if enter_ts is not None and not exit_rows.empty: exit_rows = exit_rows[exit_rows["ts"] >= enter_ts] exit_ts = exit_rows.iloc[0]["ts"] if not exit_rows.empty else None md_tags_by_second: Dict[pd.Timestamp, List[str]] = {} ap_tags_by_second: Dict[pd.Timestamp, List[str]] = {} if not window_events.empty: window_events["second"] = window_events["ts"].dt.floor("s") for second, group in window_events.groupby("second", sort=True): for source, target in (("md", md_tags_by_second), ("ap", ap_tags_by_second)): source_group = group[group["event_source"] == source].sort_values("ts") if not source_group.empty: target[second] = list(OrderedDict.fromkeys(source_group["event_tag"].tolist())) ap_state = ap_serving_cell[ (ap_serving_cell["ts"] >= start_ts - pd.Timedelta(minutes=1)) & (ap_serving_cell["ts"] <= end_ts) ].copy() if not ap_state.empty: ap_state["second"] = ap_state["ts"].dt.floor("s") keep_columns = [ "second", "ratType", "networkType", "band", "freq", "scs", "pci", "isOos", "rsrp", "snr", "rsrq", "dataRegState", ] ap_state = ap_state[[column for column in keep_columns if column in ap_state.columns]] ap_state = ap_state.groupby("second").last().reset_index().sort_values("second") timeline = merge_asof_second(timeline, ap_state) timeline = merge_asof_second(timeline, rename_prefixed(nr_ml1_state, "md_nr_state_")) timeline = merge_exact_second(timeline, rename_prefixed(nr_ml1_mean, "md_nr_mean_")) timeline = merge_exact_second(timeline, rename_prefixed(nr_snr_mean, "md_nr_snr_")) timeline = merge_asof_second(timeline, rename_prefixed(lte_ml1_state, "md_lte_state_")) timeline = merge_exact_second(timeline, rename_prefixed(lte_ml1_mean, "md_lte_mean_")) timeline = merge_asof_second(timeline, rename_prefixed(srv_ue_status, "md_srv_")) timeline = merge_asof_second(timeline, rename_prefixed(nr5g_rrc_config, "md_rrc_")) for frame in exact_frames: timeline = merge_exact_second(timeline, frame) for frame in state_frames: timeline = merge_asof_second(timeline, frame) timeline = merge_exact_second(timeline, search_scan) timeline = merge_exact_second(timeline, performance_summary) rows: List[Dict[str, object]] = [] for _, item in timeline.iterrows(): serving = choose_serving_tuple(item) row: Dict[str, object] = {header: "" for header in headers} row["Date"] = item["second"].strftime("%Y-%m-%d") row["Time"] = item["second"].strftime("%H:%M:%S") row["sub"] = choose_first_non_empty([item.get("md_nr_state_SubID"), item.get("md_lte_state_Sub_ID"), 1]) row["KEY_EVENT"] = build_key_event_value( derive_key_phase(item["second"], enter_ts, exit_ts), build_manual_elevator_tags(item["second"], start_ts, end_ts), md_tags_by_second.get(item["second"], []), ap_tags_by_second.get(item["second"], []), ) row["Performance"] = choose_first_non_empty([item.get("Performance")]) row["Search_Scan"] = choose_first_non_empty([item.get("Search_Scan")]) row["RAT"] = serving["rat"] row["BAND"] = serving["band"] row["Freq_DL"] = serving["freq"] row["SCS"] = serving["scs"] row["Pci"] = serving["pci"] row["DL_CC_num"] = choose_first_non_empty([item.get("md_rrc_Num_Active_CC"), item.get("DL_CC_num")]) row["UL_CC_num"] = choose_first_non_empty([item.get("UL_CC_num")]) row["SRV_STATUS"] = map_srv_status(item.get("isOos"), item.get("md_srv_SrvStatus")) row["RSRP_0"] = choose_first_non_empty([item.get("md_nr_mean_RSRP_0"), item.get("md_lte_mean_RSRP_0"), item.get("rsrp")]) row["RSRP_1"] = choose_first_non_empty([item.get("md_nr_mean_RSRP_1"), item.get("md_lte_mean_RSRP_1")]) row["RSRP_2"] = choose_first_non_empty([item.get("md_nr_mean_RSRP_2"), item.get("md_lte_mean_RSRP_2")]) row["RSRP_3"] = choose_first_non_empty([item.get("md_nr_mean_RSRP_3"), item.get("md_lte_mean_RSRP_3")]) row["SSB_SNR_0"] = choose_first_non_empty([item.get("md_nr_snr_SNR_0"), item.get("md_lte_mean_SNR_0"), item.get("snr")]) row["SSB_SNR_1"] = choose_first_non_empty([item.get("md_nr_snr_SNR_1"), item.get("md_lte_mean_SNR_1")]) row["SSB_SNR_2"] = choose_first_non_empty([item.get("md_nr_snr_SNR_2"), item.get("md_lte_mean_SNR_2")]) row["SSB_SNR_3"] = choose_first_non_empty([item.get("md_nr_snr_SNR_3"), item.get("md_lte_mean_SNR_3")]) row["UL_Bandwidth"] = choose_first_non_empty([item.get("UL_Bandwidth")]) row["DL_Bandwidth"] = choose_first_non_empty([item.get("DL_Bandwidth")]) row["PathLoss"] = choose_first_non_empty([item.get("PathLoss")]) row["UL_TPUT_QSH"] = choose_first_non_empty([item.get("UL_TPUT_QSH")]) row["UL_TPUT_PUSCH"] = choose_first_non_empty([item.get("UL_TPUT_PUSCH")]) row["DL_TPUT_QSH"] = choose_first_non_empty([item.get("DL_TPUT_QSH")]) row["DL_TPUT_PDSCH"] = choose_first_non_empty([item.get("DL_TPUT_PDSCH")]) row["UL_BLER_QSH"] = choose_first_non_empty([item.get("UL_BLER_QSH")]) row["DL_BLER_QSH"] = choose_first_non_empty([item.get("DL_BLER_QSH")]) row["UL_Layers_QSH"] = choose_first_non_empty([item.get("UL_Layers_QSH")]) row["UL_LAYERS_PUSCH"] = choose_first_non_empty([item.get("UL_LAYERS_PUSCH")]) row["UL_NUM_RB_AVG_PUSCH"] = choose_first_non_empty([item.get("PUSCH_NUM_RB_AVG")]) row["DL_NUM_RB_AVG_PDSCH"] = choose_first_non_empty([item.get("DL_NUM_RB_AVG_PDSCH")]) row["DL_Layers"] = choose_first_non_empty([item.get("DL_Layers")]) row["DL_MCS"] = choose_first_non_empty([item.get("DL_MCS")]) row["UL_ModType"] = choose_first_non_empty([item.get("UL_ModType")]) row["DL_ModType"] = choose_first_non_empty([item.get("DL_ModType")]) row["PRACH_TxPower"] = choose_first_non_empty([item.get("PRACH_TxPower")]) row["PUSCH_TxPower"] = choose_first_non_empty([item.get("PUSCH_TxPower")]) row["PUSCH_MTPL"] = choose_first_non_empty([item.get("PUSCH_MTPL")]) row["MTPL"] = choose_first_non_empty([item.get("PUSCH_MTPL")]) row["PUCCH_TxPower"] = choose_first_non_empty([item.get("PUCCH_TxPower")]) row["PUCCH_slot_num"] = choose_first_non_empty([item.get("PUCCH_slot_num")]) row["SRS_TxPower"] = choose_first_non_empty([item.get("SRS_TxPower")]) row["SRS_slot_num"] = choose_first_non_empty([item.get("SRS_slot_num")]) row["UL_TX_NUM"] = choose_first_non_empty([item.get("UL_TX_NUM")]) row["DL_RX_NUM"] = choose_first_non_empty([item.get("DL_RX_NUM")]) row["Tx_Ant_Number"] = choose_first_non_empty([item.get("Tx_Ant_Number")]) row["Pusch_slot_num"] = choose_first_non_empty([item.get("Pusch_slot_num")]) row["CQI"] = choose_first_non_empty([item.get("CQI")]) row["RI"] = choose_first_non_empty([item.get("RI")]) row["TMD_Level"] = choose_first_non_empty([item.get("TMD_Level")]) row["drxEnable"] = choose_first_non_empty([item.get("drxEnable")]) row["OnDuration"] = choose_first_non_empty([item.get("OnDuration")]) row["Inactive"] = choose_first_non_empty([item.get("Inactive")]) row["LongCycle"] = choose_first_non_empty([item.get("LongCycle")]) row["CDRX_Inactive_ms"] = choose_first_non_empty([item.get("CDRX_Inactive_ms")]) for header in headers: row[header] = format_output_value(row.get(header, "")) rows.append(row) filled_columns, missing_columns = coverage_for_rows(headers, rows) metadata = { "label": label, "row_count": len(rows), "enter_ts": "" if enter_ts is None else str(enter_ts), "exit_ts": "" if exit_ts is None else str(exit_ts), "filled_columns": filled_columns, "missing_columns": missing_columns, } return rows, metadata def write_summary(summary_path: Path, metadata_list: Sequence[Dict[str, object]]) -> None: summary_path.parent.mkdir(parents=True, exist_ok=True) lines: List[str] = ["# Elevator compact csv test summary", ""] for metadata in metadata_list: lines.append(f"## {metadata['label']}") lines.append(f"- row_count: {metadata['row_count']}") lines.append(f"- enter_ts: {metadata['enter_ts']}") lines.append(f"- exit_ts: {metadata['exit_ts']}") lines.append(f"- filled_columns: {len(metadata['filled_columns'])}") lines.append(f"- missing_columns: {len(metadata['missing_columns'])}") lines.append(f"- filled_column_names: {', '.join(metadata['filled_columns'])}") lines.append(f"- missing_column_names: {', '.join(metadata['missing_columns'])}") lines.append("") summary_path.write_text("\n".join(lines), encoding="utf-8") def write_source_manifest( manifest_path: Path, args: argparse.Namespace, ap_key_event_path: Path, ap_serving_cell_path: Path, performance_csv_path: Optional[Path], md_dir: Path, window_labels: Sequence[str], ) -> None: manifest_path.parent.mkdir(parents=True, exist_ok=True) lines = [ "compact_policy=mdlog_filter_auto_analysis_first_aplog_filter_fallback", "key_event_policy=manual_test_record_tags_plus_md_elevator_status_first_ap_optimization_event_fallback", "window_policy=caller_provided_windows", f"date={args.date}", f"windows={','.join(window_labels)}", f"transposed_rules={Path(args.transposed_rules).as_posix()}", f"md_dir={md_dir.as_posix()}", f"ap_dir={Path(args.ap_dir).as_posix()}", f"ap_key_event_path={ap_key_event_path.as_posix()}", f"ap_serving_cell_path={ap_serving_cell_path.as_posix()}", f"performance_csv_path={'' if performance_csv_path is None else performance_csv_path.as_posix()}", f"ai_summary={'Y' if args.ai_summary else 'N'}", ] manifest_path.write_text("\n".join(lines) + "\n", encoding="utf-8") def resolve_output_dir( all_labels: Sequence[str], output_dir: Optional[str], window_output_dir: Optional[str], ) -> Path: if window_output_dir: if len(all_labels) != 1: raise ValueError("--window-output-dir only supports a single window") return Path(window_output_dir) if output_dir is None: raise ValueError("output_dir is required when window_output_dir is not set") return Path(output_dir) def main() -> None: args = parse_args() window_labels = [item.strip() for item in args.windows.split(",") if item.strip()] base_output_dir = Path(args.output_dir) if args.output_dir else None if base_output_dir is not None: base_output_dir.mkdir(parents=True, exist_ok=True) rows = load_transposed_rules(args.transposed_rules) headers, primary_row, secondary_row = build_horizontal_rules(rows) base_date = pd.Timestamp(args.date) ap_dir = Path(args.ap_dir) md_dir = Path(args.md_dir) ap_key_event_path = resolve_first_existing_path( ap_dir, [ "opt_elevator_event.txt", "opt_airport_key_event.txt", ], ) ap_serving_cell_path = resolve_first_existing_path( ap_dir, [ "reg_data_sub_serv_cell.txt", ], ) ap_key_events = load_ap_key_events(ap_key_event_path, base_date) md_elevator_events = load_md_elevator_events(md_dir) ap_serving_cell = load_ap_serving_cell(ap_serving_cell_path, base_date) performance_csv_path = Path(args.performance_csv) if args.performance_csv else None performance_summary = load_performance_summary(args.performance_csv, base_date) nr_ml1 = load_md_csv_by_patterns(md_dir, ["sim1_NR_ML1_RX_*.csv", "sim1_NR_ML1_RX.csv"]) nr_snr = load_md_csv_by_patterns(md_dir, ["sim1_NR_SVC_SNR_*.csv", "sim1_NR_SVC_SNR.csv"]) lte_ml1 = load_md_csv_by_patterns(md_dir, ["sim1_LTE_ML1_RX_*.csv", "sim1_LTE_ML1_RX.csv"]) srv_ue_status = load_md_csv_by_patterns(md_dir, ["sim1_SRV_UE_STATUS.csv", "sim1_SRV_UE_STATUS_*.csv"]) nr5g_rrc_config = load_md_csv_by_patterns( md_dir, [ "sim1_NR5G_RRC_CONFIG_*.csv", "sim1_NR5G_RRC_CONFIG.csv", "sim_unknown_NR5G_RRC_CONFIG_*.csv", "sim_unknown_NR5G_RRC_CONFIG.csv", "_primary_csv/NR5G_RRC_CONFIG_*.csv", "_primary_csv/NR5G_RRC_CONFIG.csv", ], ) com_data_phy = load_md_csv_by_patterns(md_dir, ["sim1_COM_DATA_PHY_*.csv", "sim1_COM_DATA_PHY.csv"]) nr_mac_ul = load_md_csv_by_patterns(md_dir, ["sim1_NR_MAC_UL_SCHEDULE_*.csv", "sim1_NR_MAC_UL_SCHEDULE.csv"]) nr_mac_pdsch = load_md_csv_by_patterns(md_dir, ["sim1_NR_MAC_PDSCH_*.csv", "sim1_NR_MAC_PDSCH.csv"]) nr_mac_csf = load_md_csv_by_patterns(md_dir, ["sim1_NR_MAC_CSF_*.csv", "sim1_NR_MAC_CSF.csv"], time_is_utc=False) nr_mac_tx = load_md_csv_by_patterns(md_dir, ["sim1_NR_MAC_TX_*.csv", "sim1_NR_MAC_TX.csv"]) nr5g_cdrx = load_md_csv_by_patterns(md_dir, ["sim1_NR5G_CDRX_Config_and_State_*.csv", "sim1_NR5G_CDRX_Config_and_State.csv"]) nr5g_ml1_fc = load_md_csv_by_patterns(md_dir, ["sim1_NR5G_ML1_FC_Information_*.csv", "sim1_NR5G_ML1_FC_Information.csv"], time_is_utc=False) lte_mac_tx = load_md_csv_by_patterns(md_dir, ["sim1_LTE_MAC_TX_*.csv", "sim1_LTE_MAC_TX.csv"]) lte_mac_pdsch = load_md_csv_by_patterns(md_dir, ["sim1_LTE_MAC_PDSCH_*.csv", "sim1_LTE_MAC_PDSCH.csv"]) lte_cdrx = load_md_csv_by_patterns(md_dir, ["sim1_LTE_CDRX_Config_*.csv", "sim1_LTE_CDRX_Config.csv"]) com_rf_asdiv = load_md_csv_by_patterns( md_dir, [ "sim1_COM_RF_ASDIV_EVENT_*.csv", "sim1_COM_RF_ASDIV_EVENT.csv", "sim_unknown_COM_RF_ASDIV_EVENT_*.csv", "sim_unknown_COM_RF_ASDIV_EVENT.csv", "_primary_csv/COM_RF_ASDIV_EVENT_*.csv", "_primary_csv/COM_RF_ASDIV_EVENT.csv", ], ) nr5g_acq = load_md_csvs_with_fallback( md_dir, [ "sim1_NR5G_ACQ_*.csv", "sim1_NR5G_ACQ.csv", "sim_unknown_NR5G_ACQ_*.csv", "sim_unknown_NR5G_ACQ.csv", ], [ "_primary_csv/NR5G_ACQ_*.csv", "_primary_csv/NR5G_ACQ.csv", "NR5G_ACQ_*.csv", "NR5G_ACQ.csv", ], ) lte_system_scan = load_md_csvs_with_fallback( md_dir, [ "sim1_LTE_SYSTEM_SCAN_*.csv", "sim1_LTE_SYSTEM_SCAN.csv", "sim_unknown_LTE_SYSTEM_SCAN_*.csv", "sim_unknown_LTE_SYSTEM_SCAN.csv", ], [ "_primary_csv/LTE_SYSTEM_SCAN_*.csv", "_primary_csv/LTE_SYSTEM_SCAN.csv", "LTE_SYSTEM_SCAN_*.csv", "LTE_SYSTEM_SCAN.csv", ], ) lte_init_acq = load_md_csvs_with_fallback( md_dir, [ "sim1_LTE_INIT_ACQ_*.csv", "sim1_LTE_INIT_ACQ.csv", "sim_unknown_LTE_INIT_ACQ_*.csv", "sim_unknown_LTE_INIT_ACQ.csv", ], [ "_primary_csv/LTE_INIT_ACQ_*.csv", "_primary_csv/LTE_INIT_ACQ.csv", "LTE_INIT_ACQ_*.csv", "LTE_INIT_ACQ.csv", ], ) lte_band_scan = load_md_csvs_with_fallback( md_dir, [ "sim1_LTE_BAND_SCAN_*.csv", "sim1_LTE_BAND_SCAN.csv", "sim_unknown_LTE_BAND_SCAN_*.csv", "sim_unknown_LTE_BAND_SCAN.csv", ], [ "_primary_csv/LTE_BAND_SCAN_*.csv", "_primary_csv/LTE_BAND_SCAN.csv", "LTE_BAND_SCAN_*.csv", "LTE_BAND_SCAN.csv", ], ) nas_req_plmn = load_md_csvs_with_fallback( md_dir, [ "sim1_NAS_REQ_PLMN_*.csv", "sim1_NAS_REQ_PLMN.csv", "sim1_NW_NAS_REQ_PLMN_*.csv", "sim1_NW_NAS_REQ_PLMN.csv", "sim_unknown_NAS_REQ_PLMN_*.csv", "sim_unknown_NAS_REQ_PLMN.csv", "sim_unknown_NW_NAS_REQ_PLMN_*.csv", "sim_unknown_NW_NAS_REQ_PLMN.csv", ], [ "_primary_csv/NAS_REQ_PLMN_*.csv", "_primary_csv/NAS_REQ_PLMN.csv", "_primary_csv/NW_NAS_REQ_PLMN_*.csv", "_primary_csv/NW_NAS_REQ_PLMN.csv", "NAS_REQ_PLMN_*.csv", "NAS_REQ_PLMN.csv", "NW_NAS_REQ_PLMN_*.csv", "NW_NAS_REQ_PLMN.csv", ], ) mmode_sdss_activate = load_md_csvs_with_fallback( md_dir, [ "sim1_MMODE_SDSS_ACTIVATE_*.csv", "sim1_MMODE_SDSS_ACTIVATE.csv", "sim_unknown_MMODE_SDSS_ACTIVATE_*.csv", "sim_unknown_MMODE_SDSS_ACTIVATE.csv", ], [ "_primary_csv/MMODE_SDSS_ACTIVATE_*.csv", "_primary_csv/MMODE_SDSS_ACTIVATE.csv", "MMODE_SDSS_ACTIVATE_*.csv", "MMODE_SDSS_ACTIVATE.csv", ], ) sd_event_action = load_md_csvs_with_fallback( md_dir, [ "sim1_SD_EVENT_ACTION_*.csv", "sim1_SD_EVENT_ACTION.csv", "sim1_EVENT_SD_EVENT_ACTION_*.csv", "sim1_EVENT_SD_EVENT_ACTION.csv", "sim_unknown_SD_EVENT_ACTION_*.csv", "sim_unknown_SD_EVENT_ACTION.csv", "sim_unknown_EVENT_SD_EVENT_ACTION_*.csv", "sim_unknown_EVENT_SD_EVENT_ACTION.csv", ], [ "_primary_csv/SD_EVENT_ACTION_*.csv", "_primary_csv/SD_EVENT_ACTION.csv", "_primary_csv/EVENT_SD_EVENT_ACTION_*.csv", "_primary_csv/EVENT_SD_EVENT_ACTION.csv", "SD_EVENT_ACTION_*.csv", "SD_EVENT_ACTION.csv", "EVENT_SD_EVENT_ACTION_*.csv", "EVENT_SD_EVENT_ACTION.csv", ], ) rrc_reg_summary = load_md_csvs_with_fallback( md_dir, [ "sim1_RRC_REG_Summary_*.csv", "sim1_RRC_REG_Summary.csv", "sim_unknown_RRC_REG_Summary_*.csv", "sim_unknown_RRC_REG_Summary.csv", ], [ "_primary_csv/RRC_REG_Summary_*.csv", "_primary_csv/RRC_REG_Summary.csv", "RRC_REG_Summary_*.csv", "RRC_REG_Summary.csv", ], ) lte_rach_trigger = load_md_csvs_with_fallback( md_dir, [ "sim1_LTE_RACH_TRIGGER_*.csv", "sim1_LTE_RACH_TRIGGER.csv", "sim_unknown_LTE_RACH_TRIGGER_*.csv", "sim_unknown_LTE_RACH_TRIGGER.csv", ], [ "_primary_csv/LTE_RACH_TRIGGER_*.csv", "_primary_csv/LTE_RACH_TRIGGER.csv", "LTE_RACH_TRIGGER_*.csv", "LTE_RACH_TRIGGER.csv", ], ) lte_rach_attempt = load_md_csvs_with_fallback( md_dir, [ "sim1_LTE_RACH_ATTEMPT_*.csv", "sim1_LTE_RACH_ATTEMPT.csv", "sim_unknown_LTE_RACH_ATTEMPT_*.csv", "sim_unknown_LTE_RACH_ATTEMPT.csv", ], [ "_primary_csv/LTE_RACH_ATTEMPT_*.csv", "_primary_csv/LTE_RACH_ATTEMPT.csv", "LTE_RACH_ATTEMPT_*.csv", "LTE_RACH_ATTEMPT.csv", ], ) nr_ml1_mean = group_md_mean(nr_ml1, ["RSRP_0", "RSRP_1", "RSRP_2", "RSRP_3"]) nr_ml1_state = group_md_last(nr_ml1, ["SubID", "SCS", "RARFCN", "PCI"]) nr_snr_mean = group_md_mean( nr_snr, ["SNR_0", "SNR_1", "SNR_2", "SNR_3"], filters={"RS": "SSB"}, ) lte_ml1_mean = group_md_mean(lte_ml1, ["RSRP_0", "RSRP_1", "RSRP_2", "RSRP_3", "SNR_0", "SNR_1", "SNR_2", "SNR_3"]) lte_ml1_state = group_md_last(lte_ml1, ["Sub_ID", "EARFCN", "PCI"]) srv_ue_status_grouped = group_md_last(srv_ue_status, ["Sub_ID", "RatMode", "SrvStatus"]) nr5g_rrc_grouped = group_md_last( nr5g_rrc_config, ["Num_Active_CC", "Serving_Cell_Band"], filters={"Detail_Type": "NR5G_Serving_Cell", "Detail_Index": 0}, ) ul_qsh = rename_prefixed( group_md_mean(com_data_phy, ["AvgPHY_Mbps", "BLER_%", "AvgLayers"], filters={"Direction": "UL"}), "", ).rename(columns={"AvgPHY_Mbps": "UL_TPUT_QSH", "BLER_%": "UL_BLER_QSH", "AvgLayers": "UL_Layers_QSH"}) dl_qsh = rename_prefixed( group_md_mean(com_data_phy, ["AvgPHY_Mbps", "BLER_%"], filters={"Direction": "DL"}), "", ).rename(columns={"AvgPHY_Mbps": "DL_TPUT_QSH", "BLER_%": "DL_BLER_QSH"}) ul_cc_num = group_md_distinct_count(com_data_phy, "CC", "UL_CC_num", filters={"Direction": "UL"}) dl_cc_num = group_md_distinct_count(com_data_phy, "CC", "DL_CC_num", filters={"Direction": "DL"}) nr_ul_metrics = build_nr_ul_metrics(nr_mac_ul) lte_ul_metrics = build_lte_ul_metrics(lte_mac_tx) ul_metrics = merge_frames_on_second([nr_ul_metrics, lte_ul_metrics]) nr_dl_metrics = build_nr_dl_metrics(nr_mac_pdsch) lte_dl_metrics = build_lte_dl_metrics(lte_mac_pdsch) dl_metrics = merge_frames_on_second([nr_dl_metrics, lte_dl_metrics]) nr_pusch_tx = group_md_mean(nr_mac_tx, ["Tx_power", "MTPL", "PathLoss"], filters={"Channel": "PUSCH"}).rename( columns={"Tx_power": "PUSCH_TxPower", "MTPL": "PUSCH_MTPL", "PathLoss": "PathLoss"} ) nr_pusch_slots = group_md_count(nr_mac_tx, "Pusch_slot_num", filters={"Channel": "PUSCH"}) nr_pucch_tx = group_md_mean(nr_mac_tx, ["Tx_power"], filters={"Channel": "PUCCH"}).rename( columns={"Tx_power": "PUCCH_TxPower"} ) nr_pucch_slots = group_md_count(nr_mac_tx, "PUCCH_slot_num", filters={"Channel": "PUCCH"}) nr_srs_tx = group_md_mean(nr_mac_tx, ["Tx_power"], filters={"Channel": "SRS"}).rename( columns={"Tx_power": "SRS_TxPower"} ) nr_srs_slots = group_md_count(nr_mac_tx, "SRS_slot_num", filters={"Channel": "SRS"}) lte_pusch_tx = group_md_mean(lte_mac_tx, ["Tx_power", "MTPL", "PathLoss"], filters={"Channel": "PUSCH"}).rename( columns={"Tx_power": "PUSCH_TxPower", "MTPL": "PUSCH_MTPL", "PathLoss": "PathLoss"} ) lte_pusch_slots = group_md_count(lte_mac_tx, "Pusch_slot_num", filters={"Channel": "PUSCH"}) lte_pucch_tx = group_md_mean(lte_mac_tx, ["Tx_power"], filters={"Channel": "PUCCH"}).rename( columns={"Tx_power": "PUCCH_TxPower"} ) lte_pucch_slots = group_md_count(lte_mac_tx, "PUCCH_slot_num", filters={"Channel": "PUCCH"}) lte_prach_tx = group_md_mean(lte_mac_tx, ["Tx_power"], filters={"Channel": "PRACH"}).rename( columns={"Tx_power": "PRACH_TxPower"} ) tx_metrics = merge_frames_on_second( [ nr_pusch_tx, nr_pusch_slots, nr_pucch_tx, nr_pucch_slots, nr_srs_tx, nr_srs_slots, lte_pusch_tx, lte_pusch_slots, lte_pucch_tx, lte_pucch_slots, lte_prach_tx, ] ) nr_csf_metrics = group_md_mean(nr_mac_csf, ["CQI", "RI"]).rename(columns={"CQI": "CQI", "RI": "RI"}) nr_cdrx_state = group_md_last(nr5g_cdrx, ["DrxEnable", "OnDuration", "InactivityTimer", "LongDrxCycle"]).rename( columns={ "DrxEnable": "drxEnable", "OnDuration": "OnDuration", "InactivityTimer": "Inactive", "LongDrxCycle": "LongCycle", } ) lte_cdrx_state = group_md_last(lte_cdrx, ["CdrxEnable", "OnDurationTimer", "InactivityTimer", "LongCycleLength"]).rename( columns={ "CdrxEnable": "drxEnable", "OnDurationTimer": "OnDuration", "InactivityTimer": "Inactive", "LongCycleLength": "LongCycle", } ) cdrx_inactive = build_cdrx_inactive_metrics(nr5g_cdrx) tmd_level = group_md_last(nr5g_ml1_fc, ["CurrentState"]).rename(columns={"CurrentState": "TMD_Level"}) tx_ant = group_md_last(com_rf_asdiv, ["Antenna_Number"], filters={"Event_Type": "TRM_ASDIV_EVENT_SWITCHING_CONN"}).rename( columns={"Antenna_Number": "Tx_Ant_Number"} ) search_scan = build_search_scan_summary( nr5g_acq, lte_system_scan, lte_init_acq, lte_band_scan, nas_req_plmn, rrc_reg_summary, nr5g_rrc_config, lte_rach_trigger, lte_rach_attempt, mmode_sdss_activate, sd_event_action, ai_summary=args.ai_summary, ) metadata_list: List[Dict[str, object]] = [] summary_output_dir: Optional[Path] = None for label in window_labels: current_output_dir = resolve_output_dir( all_labels=window_labels, output_dir=args.output_dir, window_output_dir=args.window_output_dir, ) current_output_dir.mkdir(parents=True, exist_ok=True) horizontal_rules_path = current_output_dir / "modem_last_1s_avg_elevator_horizontal_rules.csv" write_horizontal_rules(str(horizontal_rules_path), headers, primary_row, secondary_row) summary_output_dir = current_output_dir start_ts, end_ts = parse_window_label(base_date, label) window_rows, metadata = build_window_rows( label=label, start_ts=start_ts, end_ts=end_ts, headers=headers, ap_key_events=ap_key_events, md_elevator_events=md_elevator_events, ap_serving_cell=ap_serving_cell, nr_ml1_mean=nr_ml1_mean, nr_ml1_state=nr_ml1_state, nr_snr_mean=nr_snr_mean, lte_ml1_mean=lte_ml1_mean, lte_ml1_state=lte_ml1_state, srv_ue_status=srv_ue_status_grouped, nr5g_rrc_config=nr5g_rrc_grouped, exact_frames=[ul_qsh, dl_qsh, ul_cc_num, dl_cc_num, ul_metrics, dl_metrics, tx_metrics, nr_csf_metrics, cdrx_inactive], state_frames=[merge_frames_on_second([nr_cdrx_state, lte_cdrx_state]), tmd_level, tx_ant], search_scan=search_scan, performance_summary=performance_summary, ) csv_path = current_output_dir / f"{label}_compact.csv" write_csv(csv_path, headers, window_rows) metadata_list.append(metadata) print(f"Wrote compact csv: {csv_path}") if summary_output_dir is not None: write_summary(summary_output_dir / "summary.md", metadata_list) write_source_manifest( summary_output_dir / "source_manifest.txt", args=args, ap_key_event_path=ap_key_event_path, ap_serving_cell_path=ap_serving_cell_path, performance_csv_path=performance_csv_path, md_dir=md_dir, window_labels=window_labels, ) print(f"Wrote summary: {summary_output_dir / 'summary.md'}") if __name__ == "__main__": main()