Add CIQ 3G Generator with WBTS/WCEL sheet generation, refactor CIQ 2G to extract shared site parsing logic with MCC/MNC parameters, implement TRX sheet builder with BCCH/TRX frequency parsing and MAIO assignment, add BTS sheet builder with template name detection and sector ID mapping, and create MAL sheet builder with mobile allocation frequency extraction from CIQ brut Excel

app.py

@@ -119,6 +119,7 @@ if check_password():
                 "apps/parameters_distribution.py", title="📊Parameters distribution"
             ),
             st.Page("apps/ciq_2g_generator.py", title="🧾 CIQ 2G Generator"),
+            st.Page("apps/ciq_3g_generator.py", title="🧾 CIQ 3G Generator"),
             st.Page("apps/core_dump_page.py", title="📠Parse dump core"),
             st.Page("apps/gps_converter.py", title="🧭GPS Converter"),
             st.Page("apps/distance.py", title="🛰Distance Calculator"),

apps/ciq_2g_generator.py

@@ -13,6 +13,12 @@ with col2:
         "Upload CIQ brut 2G (Excel)", type=["xlsx", "xls"], key="ciq2g_ciq"
     )
 
+col3, col4 = st.columns(2)
+with col3:
+    mcc = st.number_input("MCC", value=610, step=1, min_value=0, key="ciq2g_mcc")
+with col4:
+    mnc = st.number_input("MNC", value=2, step=1, min_value=0, key="ciq2g_mnc")
+
 if dump_file is None or ciq_file is None:
     st.info("Upload dump xlsb + CIQ brut Excel to generate CIQ 2G.")
     st.stop()
@@ -20,7 +26,9 @@ if dump_file is None or ciq_file is None:
 if st.button("Generate", type="primary"):
     try:
         with st.spinner("Generating CIQ 2G... (dump is heavy)"):
-            sheets, excel_bytes = generate_ciq_2g_excel(dump_file, ciq_file)
+            sheets, excel_bytes = generate_ciq_2g_excel(
+                dump_file, ciq_file, mcc=int(mcc), mnc=int(mnc)
+            )
             st.session_state["ciq2g_sheets"] = sheets
             st.session_state["ciq2g_excel_bytes"] = excel_bytes
         st.success("CIQ 2G generated")

apps/ciq_3g_generator.py

@@ -0,0 +1,56 @@
+import pandas as pd
+import streamlit as st
+
+from queries.process_ciq_3g import generate_ciq_3g_excel
+
+st.title("CIQ 3G Generator")
+
+ciq_file = st.file_uploader(
+    "Upload CIQ brut 3G (Excel)", type=["xlsx", "xls"], key="ciq3g_ciq"
+)
+
+col1, col2 = st.columns(2)
+with col1:
+    year_suffix = st.text_input("Year suffix", value="25", key="ciq3g_year")
+with col2:
+    bands = st.text_input(
+        "Bands string",
+        value="G9G18U9U21L8L18L26",
+        key="ciq3g_bands",
+    )
+
+if ciq_file is None:
+    st.info("Upload CIQ brut 3G Excel to generate CIQ 3G (WBTS + WCEL).")
+    st.stop()
+
+if st.button("Generate", type="primary"):
+    try:
+        with st.spinner("Generating CIQ 3G..."):
+            sheets, excel_bytes = generate_ciq_3g_excel(
+                ciq_file, year_suffix=year_suffix.strip(), bands=bands.strip()
+            )
+            st.session_state["ciq3g_sheets"] = sheets
+            st.session_state["ciq3g_excel_bytes"] = excel_bytes
+        st.success("CIQ 3G generated")
+    except Exception as e:
+        st.error(f"Error: {e}")
+
+sheets = st.session_state.get("ciq3g_sheets")
+excel_bytes = st.session_state.get("ciq3g_excel_bytes")
+
+if sheets:
+    tab_names = list(sheets.keys())
+    tabs = st.tabs(tab_names)
+    for t, name in zip(tabs, tab_names):
+        with t:
+            df: pd.DataFrame = sheets[name]
+            st.dataframe(df, use_container_width=True)
+
+if excel_bytes:
+    st.download_button(
+        label="Download CIQ 3G Excel",
+        data=excel_bytes,
+        file_name="CIQ_3G.xlsx",
+        mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        type="primary",
+    )

queries/process_ciq_2g.py

@@ -7,6 +7,25 @@ import pandas as pd
 
 REQUIRED_DUMP_BTS_COLS = ["BSC", "BCF", "BTS", "usedMobileAllocation"]
 
+BTS_EXPORT_COLUMNS = [
+    "site",
+    "bscid",
+    "cellId",
+    "bcfId",
+    "btsId",
+    "Check",
+    "bsIdentityCodeNCC",
+    "bsIdentityCodeBCC",
+    "locationAreaIdLAC",
+    "locationAreaIdMCC",
+    "locationAreaIdMNC",
+    "usedMobileAllocation",
+    "malId",
+    "name",
+    "template_name",
+    "sectorId",
+]
+
 
 def _normalize_col(col: object) -> str:
     return re.sub(r"[^0-9A-Za-z]", "", str(col))
@@ -84,6 +103,36 @@ def _parse_site_number(site: object) -> int:
     return int(m.group(1)) if m else 0
 
 
+def _read_ciq_df(ciq_file) -> pd.DataFrame:
+    if hasattr(ciq_file, "seek"):
+        ciq_file.seek(0)
+
+    df = pd.read_excel(ciq_file, engine="calamine")
+    df.columns = df.columns.astype(str).str.strip()
+
+    if "Sites" not in df.columns:
+        raise ValueError("CIQ brut is missing required column: Sites")
+
+    df["Sites"] = df["Sites"].where(df["Sites"].notna(), pd.NA)
+    df["Sites"] = df["Sites"].astype("string").str.strip()
+    df["site_number"] = df["Sites"].apply(_parse_site_number)
+
+    if "BSC ID" in df.columns:
+        df["BSC ID"] = pd.to_numeric(df["BSC ID"], errors="coerce")
+    if "Nbre_TRE_DR" in df.columns:
+        df["Nbre_TRE_DR"] = pd.to_numeric(df["Nbre_TRE_DR"], errors="coerce")
+
+    if "NOM_CELLULE" in df.columns:
+        bands_sectors = df["NOM_CELLULE"].apply(_extract_band_and_sector)
+        df["band"] = bands_sectors.apply(lambda x: x[0])
+        df["sector"] = bands_sectors.apply(lambda x: x[1])
+    else:
+        df["band"] = None
+        df["sector"] = None
+
+    return df
+
+
 def _extract_band_and_sector(cell_name: object) -> tuple[Optional[str], Optional[int]]:
     if not isinstance(cell_name, str):
         return None, None
@@ -157,28 +206,22 @@ def _needed_bts_ids_from_site_rows(
     return tuple(sorted(ids))
 
 
-def _parse_ciq_sites(ciq_file) -> list[_PlannedSite]:
-    if hasattr(ciq_file, "seek"):
-        ciq_file.seek(0)
-
-    df = pd.read_excel(ciq_file, engine="calamine")
-
-    df.columns = df.columns.astype(str).str.strip()
-
-    required = ["Sites", "NOM_CELLULE", "Nbre_TRE_DR", "Nom BSC", "BSC ID"]
-    missing = [c for c in required if c not in df.columns]
+def _parse_ciq_sites(ciq_df: pd.DataFrame) -> list[_PlannedSite]:
+    required = [
+        "Sites",
+        "NOM_CELLULE",
+        "Nbre_TRE_DR",
+        "Nom BSC",
+        "BSC ID",
+        "band",
+        "sector",
+        "site_number",
+    ]
+    missing = [c for c in required if c not in ciq_df.columns]
     if missing:
         raise ValueError(f"CIQ brut is missing required columns: {missing}")
 
-    df = df[required].copy()
-
-    df["site_number"] = df["Sites"].apply(_parse_site_number)
-    df["BSC ID"] = pd.to_numeric(df["BSC ID"], errors="coerce")
-    df["Nbre_TRE_DR"] = pd.to_numeric(df["Nbre_TRE_DR"], errors="coerce")
-
-    bands_sectors = df["NOM_CELLULE"].apply(_extract_band_and_sector)
-    df["band"] = bands_sectors.apply(lambda x: x[0])
-    df["sector"] = bands_sectors.apply(lambda x: x[1])
+    df = ciq_df[required].copy()
 
     sites: list[_PlannedSite] = []
 
@@ -215,21 +258,8 @@ def _parse_ciq_sites(ciq_file) -> list[_PlannedSite]:
 
 
 def _assign_bcfs(
-    dump_bts: pd.DataFrame, planned_sites: list[_PlannedSite], ciq_file
+    dump_bts: pd.DataFrame, planned_sites: list[_PlannedSite], ciq_df: pd.DataFrame
 ) -> list[_PlannedSite]:
-    if hasattr(ciq_file, "seek"):
-        ciq_file.seek(0)
-
-    ciq_df = pd.read_excel(ciq_file, engine="calamine")
-    ciq_df.columns = ciq_df.columns.astype(str).str.strip()
-    ciq_df = ciq_df[["Sites", "NOM_CELLULE", "Nbre_TRE_DR", "Nom BSC", "BSC ID"]].copy()
-
-    ciq_df["BSC ID"] = pd.to_numeric(ciq_df["BSC ID"], errors="coerce")
-    ciq_df["Nbre_TRE_DR"] = pd.to_numeric(ciq_df["Nbre_TRE_DR"], errors="coerce")
-
-    bands_sectors = ciq_df["NOM_CELLULE"].apply(_extract_band_and_sector)
-    ciq_df["band"] = bands_sectors.apply(lambda x: x[0])
-    ciq_df["sector"] = bands_sectors.apply(lambda x: x[1])
 
     dump_bts = dump_bts.dropna(subset=["BSC"])
 
@@ -318,9 +348,15 @@ def _assign_bcfs(
 
 def build_bcf_sheet(dump_file, ciq_file) -> pd.DataFrame:
     dump_bts = _read_dump_bts_required_columns(dump_file)
-    planned_sites = _parse_ciq_sites(ciq_file)
-    assigned_sites = _assign_bcfs(dump_bts, planned_sites, ciq_file)
+    ciq_df = _read_ciq_df(ciq_file)
+    planned_sites = _parse_ciq_sites(ciq_df)
+    assigned_sites = _assign_bcfs(dump_bts, planned_sites, ciq_df)
+    return _build_bcf_sheet_from_assigned_sites(assigned_sites)
 
+
+def _build_bcf_sheet_from_assigned_sites(
+    assigned_sites: list[_PlannedSite],
+) -> pd.DataFrame:
     rows = []
     for i, s in enumerate(assigned_sites, start=1):
         rows.append(
@@ -334,24 +370,377 @@ def build_bcf_sheet(dump_file, ciq_file) -> pd.DataFrame:
                 "Configuration": s.configuration,
             }
         )
+    return pd.DataFrame(rows)
+
 
-    df_bcf = pd.DataFrame(rows)
-    return df_bcf
+def _sector_id_from_band_sector(band: object, sector: object) -> int:
+    if band not in {"G9", "G18"}:
+        raise ValueError(f"Invalid band '{band}'")
+    if sector is None or (isinstance(sector, float) and pd.isna(sector)):
+        raise ValueError("Missing sector")
+    sec = int(sector)
+    if sec not in {1, 2, 3}:
+        raise ValueError(f"Invalid sector '{sec}'")
+    return sec if band == "G9" else sec + 3
 
 
-def generate_ciq_2g_excel(dump_file, ciq_file) -> tuple[dict[str, pd.DataFrame], bytes]:
-    df_bcf = build_bcf_sheet(dump_file, ciq_file)
+def _template_name_from_freq(freq: object) -> str:
+    s = str(freq) if freq is not None else ""
+    s_u = s.upper()
+    if "1800" in s_u:
+        return "GSM1800"
+    if "900" in s_u:
+        return "GSM900"
+    return s
+
+
+def _template_name_from_band(band: object) -> str:
+    if band == "G9":
+        return "GSM900"
+    if band == "G18":
+        return "GSM1800"
+    return _template_name_from_freq(band)
+
+
+def _frequency_band_in_use_from_band(band: object) -> int:
+    if band == "G9":
+        return 0
+    if band == "G18":
+        return 1
+    raise ValueError(f"Invalid band '{band}'")
+
+
+def _parse_trx_frequencies(value: object) -> list[str]:
+    if value is None or (isinstance(value, float) and pd.isna(value)):
+        return []
+
+    s = str(value)
+    nums = re.findall(r"\d+", s)
+    return nums
+
+
+def _build_trx_sheet_from_assigned_sites(
+    ciq_df: pd.DataFrame, assigned_sites: list[_PlannedSite]
+) -> pd.DataFrame:
+    assigned_by_site = {s.site_name: s for s in assigned_sites}
+
+    required = [
+        "Sites",
+        "CI",
+        "band",
+        "sector",
+        "BCCH",
+        "TRX",
+        "BCC",
+    ]
+    missing = [c for c in required if c not in ciq_df.columns]
+    if missing:
+        raise ValueError(
+            f"CIQ brut is missing required columns for TRX sheet: {missing}"
+        )
+
+    rows = []
+
+    bcch_types = [4, 8, 6, 2, 2, 2, 2, 2]
+
+    for _, r in ciq_df[ciq_df["Sites"].isin(assigned_by_site.keys())].iterrows():
+        site_name = r["Sites"]
+        site = assigned_by_site.get(site_name)
+        if site is None or site.assigned_bcf is None:
+            continue
+
+        sector_id = _sector_id_from_band_sector(r.get("band"), r.get("sector"))
+        bts_id = int(site.assigned_bcf) + int(sector_id)
+
+        cell_id = pd.to_numeric(r.get("CI"), errors="coerce")
+        bcch = pd.to_numeric(r.get("BCCH"), errors="coerce")
+        bcc = pd.to_numeric(r.get("BCC"), errors="coerce")
+
+        freq_band = _frequency_band_in_use_from_band(r.get("band"))
+
+        base = {
+            "site": int(site.site_number),
+            "bscid": int(site.bsc),
+            "cellId": int(cell_id) if not pd.isna(cell_id) else None,
+            "bcfId": int(site.assigned_bcf),
+            "btsId": int(bts_id),
+            "tsc": int(bcc) if not pd.isna(bcc) else None,
+            "FrequencyBandinUse": int(freq_band),
+        }
+
+        bcch_row = dict(base)
+        bcch_row["TRX"] = None
+        bcch_row["initialFrequency"] = int(bcch) if not pd.isna(bcch) else None
+        bcch_row["_sort_type"] = 0
+        bcch_row["_sort_maio"] = -1
+
+        for i in range(8):
+            bcch_row[f"channel{i}Maio"] = None
+            bcch_row[f"channel{i}Type"] = bcch_types[i]
+        rows.append(bcch_row)
+
+        trx_list = _parse_trx_frequencies(r.get("TRX"))
+        if not pd.isna(bcch):
+            bcch_str = str(int(bcch))
+            trx_list = [x for x in trx_list if x != bcch_str]
+
+        for maio, f in enumerate(trx_list):
+            tr_row = dict(base)
+            tr_row["TRX"] = None
+            tr_row["initialFrequency"] = int(f)
+            tr_row["_sort_type"] = 1
+            tr_row["_sort_maio"] = int(maio)
+            for i in range(8):
+                tr_row[f"channel{i}Maio"] = maio
+                tr_row[f"channel{i}Type"] = 3 if i == 0 else 2
+            rows.append(tr_row)
+
+    df_trx = pd.DataFrame(rows)
+    if df_trx.empty:
+        return df_trx
+
+    ordered_cols = [
+        "site",
+        "bscid",
+        "cellId",
+        "bcfId",
+        "btsId",
+        "TRX",
+        "tsc",
+        "FrequencyBandinUse",
+        "initialFrequency",
+    ]
+    for i in range(8):
+        ordered_cols.append(f"channel{i}Maio")
+        ordered_cols.append(f"channel{i}Type")
+
+    df_trx = df_trx.sort_values(
+        by=["site", "btsId", "_sort_type", "_sort_maio"], kind="stable"
+    )
+    df_trx["TRX"] = range(1, len(df_trx) + 1)
+    df_trx = df_trx[ordered_cols]
+    return df_trx
+
+
+def build_bts_sheet(dump_file, ciq_file, mcc: int = 610, mnc: int = 2) -> pd.DataFrame:
+    dump_bts = _read_dump_bts_required_columns(dump_file)
+    ciq_df = _read_ciq_df(ciq_file)
+
+    planned_sites = _parse_ciq_sites(ciq_df)
+    assigned_sites = _assign_bcfs(dump_bts, planned_sites, ciq_df)
+    return _build_bts_sheet_from_assigned_sites(
+        ciq_df, assigned_sites, mcc=mcc, mnc=mnc
+    )
+
+
+def _build_bts_sheet_from_assigned_sites(
+    ciq_df: pd.DataFrame, assigned_sites: list[_PlannedSite], mcc: int, mnc: int
+) -> pd.DataFrame:
+    assigned_by_site = {s.site_name: s for s in assigned_sites}
+
+    required = [
+        "Sites",
+        "NOM_CELLULE",
+        "CI",
+        "LAC",
+        "Frequence",
+        "NCC",
+        "BCC",
+        "band",
+        "sector",
+    ]
+    missing = [c for c in required if c not in ciq_df.columns]
+    if missing:
+        raise ValueError(
+            f"CIQ brut is missing required columns for BTS sheet: {missing}"
+        )
+
+    rows = []
+    for _, r in ciq_df[ciq_df["Sites"].isin(assigned_by_site.keys())].iterrows():
+        site_name = r["Sites"]
+        site = assigned_by_site.get(site_name)
+        if site is None or site.assigned_bcf is None:
+            continue
+
+        sector_id = _sector_id_from_band_sector(r.get("band"), r.get("sector"))
+        bts_id = int(site.assigned_bcf) + int(sector_id)
+        cell_id = pd.to_numeric(r.get("CI"), errors="coerce")
+        lac = pd.to_numeric(r.get("LAC"), errors="coerce")
+        ncc = pd.to_numeric(r.get("NCC"), errors="coerce")
+        bcc = pd.to_numeric(r.get("BCC"), errors="coerce")
+
+        rows.append(
+            {
+                "site": int(site.site_number),
+                "bscid": int(site.bsc),
+                "cellId": int(cell_id) if not pd.isna(cell_id) else None,
+                "bcfId": int(site.assigned_bcf),
+                "btsId": int(bts_id),
+                "Check": int(sector_id),
+                "bsIdentityCodeNCC": int(ncc) if not pd.isna(ncc) else None,
+                "bsIdentityCodeBCC": int(bcc) if not pd.isna(bcc) else None,
+                "locationAreaIdLAC": int(lac) if not pd.isna(lac) else None,
+                "locationAreaIdMCC": int(mcc),
+                "locationAreaIdMNC": int(mnc),
+                "usedMobileAllocation": int(bts_id),
+                "malId": int(bts_id),
+                "name": f"{str(r.get('NOM_CELLULE'))}_NA",
+                "template_name": _template_name_from_band(r.get("band")),
+                "sectorId": int(sector_id),
+            }
+        )
+
+    df_bts = pd.DataFrame(rows)
+    if not df_bts.empty:
+        df_bts = df_bts[BTS_EXPORT_COLUMNS].sort_values(
+            by=["site", "sectorId"], kind="stable"
+        )
+    return df_bts
+
+
+def _build_mal_sheet_from_assigned_sites(
+    ciq_df: pd.DataFrame, assigned_sites: list[_PlannedSite]
+) -> pd.DataFrame:
+    assigned_by_site = {s.site_name: s for s in assigned_sites}
+
+    required = [
+        "Sites",
+        "CI",
+        "band",
+        "sector",
+        "BCCH",
+        "TRX",
+    ]
+    missing = [c for c in required if c not in ciq_df.columns]
+    if missing:
+        raise ValueError(
+            f"CIQ brut is missing required columns for MAL sheet: {missing}"
+        )
+
+    rows = []
+    for _, r in ciq_df[ciq_df["Sites"].isin(assigned_by_site.keys())].iterrows():
+        site_name = r["Sites"]
+        site = assigned_by_site.get(site_name)
+        if site is None or site.assigned_bcf is None:
+            continue
+
+        sector_id = _sector_id_from_band_sector(r.get("band"), r.get("sector"))
+        bts_id = int(site.assigned_bcf) + int(sector_id)
+
+        cell_id = pd.to_numeric(r.get("CI"), errors="coerce")
+        bcch = pd.to_numeric(r.get("BCCH"), errors="coerce")
+
+        trx_list = _parse_trx_frequencies(r.get("TRX"))
+        freq_str = ", ".join(trx_list)
+
+        row = {
+            "site": int(site.site_number),
+            "siteId": int(site.site_number),
+            "bscid": int(site.bsc),
+            "cellId": int(cell_id) if not pd.isna(cell_id) else None,
+            "bcfId": int(site.assigned_bcf),
+            "btsId": int(bts_id),
+            "frequencyBandInUse": _frequency_band_in_use_from_band(r.get("band")),
+            "malId": int(bts_id),
+            "initial frequency": int(bcch) if not pd.isna(bcch) else None,
+            "frequency": freq_str if freq_str else None,
+        }
+
+        for i in range(1, 7):
+            row[f"frequency{i}"] = trx_list[i - 1] if len(trx_list) >= i else None
+
+        rows.append(row)
+
+    df_mal = pd.DataFrame(rows)
+    if df_mal.empty:
+        return df_mal
+
+    ordered_cols = [
+        "site",
+        "siteId",
+        "bscid",
+        "cellId",
+        "bcfId",
+        "btsId",
+        "frequencyBandInUse",
+        "malId",
+        "initial frequency",
+        "frequency",
+        "frequency1",
+        "frequency2",
+        "frequency3",
+        "frequency4",
+        "frequency5",
+        "frequency6",
+    ]
+    df_mal = df_mal[ordered_cols].sort_values(by=["site", "btsId"], kind="stable")
+    return df_mal
+
+
+def generate_ciq_2g_excel(
+    dump_file, ciq_file, mcc: int = 610, mnc: int = 2
+) -> tuple[dict[str, pd.DataFrame], bytes]:
+    dump_bts = _read_dump_bts_required_columns(dump_file)
+    ciq_df = _read_ciq_df(ciq_file)
+    planned_sites = _parse_ciq_sites(ciq_df)
+    assigned_sites = _assign_bcfs(dump_bts, planned_sites, ciq_df)
+
+    df_bcf = _build_bcf_sheet_from_assigned_sites(assigned_sites)
+    df_bts = _build_bts_sheet_from_assigned_sites(
+        ciq_df, assigned_sites, mcc=mcc, mnc=mnc
+    )
+    df_mal = _build_mal_sheet_from_assigned_sites(ciq_df, assigned_sites)
+    df_trx = _build_trx_sheet_from_assigned_sites(ciq_df, assigned_sites)
+
+    df_bts_min = pd.DataFrame()
+    if not df_bts.empty:
+        df_bts_min = df_bts[["site", "bscid", "cellId", "bcfId", "btsId"]].rename(
+            columns={"site": "Site"}
+        )
+
+    df_hoc = pd.DataFrame()
+    df_poc = pd.DataFrame()
+    if not df_bts.empty:
+        base = df_bts[
+            ["site", "bscid", "cellId", "bcfId", "btsId", "template_name"]
+        ].rename(columns={"site": "Site"})
+
+        df_hoc = base.copy()
+        df_hoc.insert(5, "hocId", 1)
+        df_hoc = df_hoc[
+            ["Site", "bscid", "cellId", "bcfId", "btsId", "hocId", "template_name"]
+        ]
+
+        df_poc = base.copy()
+        df_poc.insert(5, "pocId", 1)
+        df_poc = df_poc[
+            ["Site", "bscid", "cellId", "bcfId", "btsId", "pocId", "template_name"]
+        ]
+
+    df_plmn_permitted = pd.DataFrame()
+    if not df_bts.empty:
+        base_plmn = df_bts[["bscid", "cellId", "bcfId", "btsId"]].rename(
+            columns={"bscid": "BSCId"}
+        )
+        df_plmn_permitted = base_plmn.loc[base_plmn.index.repeat(8)].reset_index(
+            drop=True
+        )
+        df_plmn_permitted["template_name"] = list(range(1, 9)) * len(base_plmn)
+        df_plmn_permitted["plmnPermitted"] = "List;1;1;1;1;1;1;1;1"
+        df_plmn_permitted = df_plmn_permitted[
+            ["BSCId", "cellId", "bcfId", "btsId", "template_name", "plmnPermitted"]
+        ]
 
     sheets: dict[str, pd.DataFrame] = {
         "BCF": df_bcf,
-        "BTS": pd.DataFrame(),
-        "BTS_GPRS": pd.DataFrame(),
-        "BTS_AMR": pd.DataFrame(),
-        "HOC": pd.DataFrame(),
-        "POC": pd.DataFrame(),
-        "MAL": pd.DataFrame(),
-        "BTS_PLMNPERMITTED": pd.DataFrame(),
-        "TRX": pd.DataFrame(),
+        "BTS": df_bts,
+        "BTS_GPRS": df_bts_min,
+        "BTS_AMR": df_bts_min,
+        "HOC": df_hoc,
+        "POC": df_poc,
+        "MAL": df_mal,
+        "BTS_PLMNPERMITTED": df_plmn_permitted,
+        "TRX": df_trx,
     }
 
     bytes_io = io.BytesIO()

queries/process_ciq_3g.py

@@ -0,0 +1,276 @@
+import io
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+import pandas as pd
+
+
+def _parse_int(value: object) -> Optional[int]:
+    v = pd.to_numeric(value, errors="coerce")
+    if pd.isna(v):
+        return None
+    return int(v)
+
+
+def _base_site_name_from_sites(sites: object) -> str:
+    if not isinstance(sites, str):
+        return ""
+
+    s = sites.strip()
+
+    for suffix in ["_3G", "_3g"]:
+        if s.endswith(suffix):
+            return s[: -len(suffix)]
+
+    return s
+
+
+def read_ciq_3g_brut(ciq_file) -> pd.DataFrame:
+    if hasattr(ciq_file, "seek"):
+        ciq_file.seek(0)
+
+    df = pd.read_excel(ciq_file, engine="calamine")
+    df.columns = df.columns.astype(str).str.strip()
+
+    if "Sites" not in df.columns:
+        raise ValueError("CIQ 3G brut is missing required column: Sites")
+
+    df["Sites"] = df["Sites"].where(df["Sites"].notna(), pd.NA)
+    df["Sites"] = df["Sites"].astype("string").str.strip()
+
+    return df
+
+
+def _band_from_cell_name(cell_name: object) -> str:
+    if not isinstance(cell_name, str):
+        return ""
+    s = cell_name.upper()
+    if "_U900" in s or s.endswith("U900"):
+        return "U900"
+    if "_U2100" in s or s.endswith("U2100"):
+        return "U2100"
+    return ""
+
+
+def _band_label(band: str) -> str:
+    if band == "U900":
+        return "U900 (U9)"
+    if band == "U2100":
+        return "U2100 (U21)"
+    return band
+
+
+def _cell_number_from_cell_name(cell_name: object) -> Optional[int]:
+    if not isinstance(cell_name, str):
+        return None
+    m = re.search(r"_(\d+)_U(?:900|2100)\b", cell_name.upper())
+    if not m:
+        return None
+    try:
+        v = int(m.group(1))
+    except ValueError:
+        return None
+    return v if v > 0 else None
+
+
+def _sector_id_from_cell_name(cell_name: object) -> int:
+    cell_no = _cell_number_from_cell_name(cell_name)
+    if cell_no is None:
+        raise ValueError(f"Cannot derive SectorID from NOM_CELLULE='{cell_name}'")
+    return ((int(cell_no) - 1) % 3) + 1
+
+
+def _tcell_from_band_and_sector(band: str, sector_id: int) -> int:
+    if band == "U900":
+        return sector_id + 2  # 1->3, 2->4, 3->5
+    # U2100
+    tcell_map = {1: 0, 2: 1, 3: 3}
+    if sector_id not in tcell_map:
+        raise ValueError(f"Invalid SectorID '{sector_id}' for Tcell")
+    return tcell_map[sector_id]
+
+
+def build_wcel_sheet(ciq_df: pd.DataFrame) -> pd.DataFrame:
+    required = [
+        "Sites",
+        "NodeB_ID",
+        "NOM_CELLULE",
+        "CELLID",
+        "SAC",
+        "LAC",
+        "RAC",
+        "FREQUENCE",
+        "PSCRAMBCODE",
+        "RNC_id",
+    ]
+    missing = [c for c in required if c not in ciq_df.columns]
+    if missing:
+        raise ValueError(f"CIQ 3G brut is missing required columns for WCEL: {missing}")
+
+    rows = []
+    for site_key, group in ciq_df.groupby(["NodeB_ID", "RNC_id"], dropna=False):
+        nodeb_id_raw, rnc_id_raw = site_key
+        nodeb_id = _parse_int(nodeb_id_raw)
+        rnc_id = _parse_int(rnc_id_raw)
+        if nodeb_id is None or rnc_id is None:
+            continue
+
+        tmp = group.copy()
+        tmp["_band"] = tmp["NOM_CELLULE"].apply(_band_from_cell_name)
+
+        # U2100 LcrId grouping by UARFCN (FREQUENCE)
+        u2100 = tmp[tmp["_band"] == "U2100"].copy()
+        u2100_uarfcns = sorted(
+            pd.to_numeric(u2100["FREQUENCE"], errors="coerce")
+            .dropna()
+            .astype(int)
+            .unique()
+        )
+        u2100_base_by_uarfcn = {
+            uarfcn: 1 + 3 * idx for idx, uarfcn in enumerate(u2100_uarfcns)
+        }
+
+        for _, r in tmp.iterrows():
+            band = r.get("_band")
+            if band not in {"U900", "U2100"}:
+                continue
+
+            uarfcn = _parse_int(r.get("FREQUENCE"))
+            if uarfcn is None:
+                continue
+
+            sector_id = _sector_id_from_cell_name(r.get("NOM_CELLULE"))
+
+            if band == "U900":
+                lcr_id = 9 + sector_id  # 10..12
+            else:
+                base = u2100_base_by_uarfcn.get(uarfcn)
+                if base is None:
+                    # Should not happen, but keep safe
+                    base = 1
+                lcr_id = base + (sector_id - 1)
+
+            cid = _parse_int(r.get("CELLID"))
+            lac = _parse_int(r.get("LAC"))
+            rac = _parse_int(r.get("RAC"))
+            sac = _parse_int(r.get("SAC"))
+
+            name = f"{str(r.get('NOM_CELLULE'))}_NA"
+
+            rows.append(
+                {
+                    "Site": nodeb_id,
+                    "RncId": rnc_id,
+                    "WBTSId": nodeb_id,
+                    "LcrId": int(lcr_id),
+                    "Band": _band_label(band),
+                    "CId": cid,
+                    "LAC": lac,
+                    "name": name,
+                    "PriScrCode": _parse_int(r.get("PSCRAMBCODE")),
+                    "PWSMCellGroup": int(sector_id),
+                    "RAC": rac,
+                    "SAC": sac,
+                    "Tcell": _tcell_from_band_and_sector(band, int(sector_id)),
+                    "UARFCN": int(uarfcn),
+                    "SectorID": int(sector_id),
+                }
+            )
+
+    df_wcel = pd.DataFrame(rows)
+    if df_wcel.empty:
+        return df_wcel
+
+    ordered = [
+        "Site",
+        "RncId",
+        "WBTSId",
+        "LcrId",
+        "Band",
+        "CId",
+        "LAC",
+        "name",
+        "PriScrCode",
+        "PWSMCellGroup",
+        "RAC",
+        "SAC",
+        "Tcell",
+        "UARFCN",
+        "SectorID",
+    ]
+    df_wcel = df_wcel[ordered].sort_values(by=["Site", "LcrId"], kind="stable")
+    return df_wcel
+
+
+def build_wbts_sheet(
+    ciq_df: pd.DataFrame, year_suffix: str, bands: str
+) -> pd.DataFrame:
+    required = ["Sites", "NodeB_ID", "RNC_id"]
+    missing = [c for c in required if c not in ciq_df.columns]
+    if missing:
+        raise ValueError(f"CIQ 3G brut is missing required columns for WBTS: {missing}")
+
+    rows = []
+    for sites, group in ciq_df.groupby("Sites", dropna=False):
+        if sites is None or (isinstance(sites, float) and pd.isna(sites)):
+            continue
+        sites_str = str(sites).strip()
+        if not sites_str:
+            continue
+
+        nodeb_ids = pd.to_numeric(group["NodeB_ID"], errors="coerce").dropna().unique()
+        if len(nodeb_ids) == 0:
+            raise ValueError(f"Missing NodeB_ID for site '{sites_str}'")
+        nodeb_id = int(nodeb_ids[0])
+
+        rnc_ids = pd.to_numeric(group["RNC_id"], errors="coerce").dropna().unique()
+        if len(rnc_ids) == 0:
+            raise ValueError(f"Missing RNC_id for site '{sites_str}'")
+        rnc_id = int(rnc_ids[0])
+
+        base_name = _base_site_name_from_sites(sites_str)
+        name = f"{base_name}_{year_suffix}_{bands}_NA"
+        wbts_name = f"{sites_str}_NA"
+
+        rows.append(
+            {
+                "S": nodeb_id,
+                "Name": name,
+                "RncId": rnc_id,
+                "WBTSId": nodeb_id,
+                "name": wbts_name,
+                "WBTSName": wbts_name,
+            }
+        )
+
+    df_wbts = pd.DataFrame(rows)
+    if not df_wbts.empty:
+        df_wbts = df_wbts[
+            ["S", "Name", "RncId", "WBTSId", "name", "WBTSName"]
+        ].sort_values(by=["S"], kind="stable")
+
+    return df_wbts
+
+
+def generate_ciq_3g_excel(
+    ciq_file,
+    year_suffix: str = "25",
+    bands: str = "G9G18U9U21L8L18L26",
+) -> tuple[dict[str, pd.DataFrame], bytes]:
+    ciq_df = read_ciq_3g_brut(ciq_file)
+
+    df_wbts = build_wbts_sheet(ciq_df, year_suffix=year_suffix, bands=bands)
+    df_wcel = build_wcel_sheet(ciq_df)
+
+    sheets: dict[str, pd.DataFrame] = {
+        "WBTS": df_wbts,
+        "WCEL": df_wcel,
+    }
+
+    bytes_io = io.BytesIO()
+    with pd.ExcelWriter(bytes_io, engine="xlsxwriter") as writer:
+        for sheet_name, df in sheets.items():
+            df.to_excel(writer, sheet_name=sheet_name, index=False)
+
+    return sheets, bytes_io.getvalue()