Add CIQ 2G Generator with BCF assignment logic, site parsing from CIQ brut Excel, dump BTS column extraction, band/sector detection, configuration building, and multi-sheet Excel export with placeholder sheets for BTS, GPRS, AMR, HOC, POC, MAL, PLMNPERMITTED, and TRX

app.py

@@ -118,6 +118,7 @@ if check_password():
             st.Page(
                 "apps/parameters_distribution.py", title="📊Parameters distribution"
             ),
+            st.Page("apps/ciq_2g_generator.py", title="🧾 CIQ 2G 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

@@ -0,0 +1,48 @@
+import pandas as pd
+import streamlit as st
+
+from queries.process_ciq_2g import generate_ciq_2g_excel
+
+st.title("CIQ 2G Generator")
+
+col1, col2 = st.columns(2)
+with col1:
+    dump_file = st.file_uploader("Upload dump file", type=["xlsb"], key="ciq2g_dump")
+with col2:
+    ciq_file = st.file_uploader(
+        "Upload CIQ brut 2G (Excel)", type=["xlsx", "xls"], key="ciq2g_ciq"
+    )
+
+if dump_file is None or ciq_file is None:
+    st.info("Upload dump xlsb + CIQ brut Excel to generate CIQ 2G.")
+    st.stop()
+
+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)
+            st.session_state["ciq2g_sheets"] = sheets
+            st.session_state["ciq2g_excel_bytes"] = excel_bytes
+        st.success("CIQ 2G generated")
+    except Exception as e:
+        st.error(f"Error: {e}")
+
+sheets = st.session_state.get("ciq2g_sheets")
+excel_bytes = st.session_state.get("ciq2g_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 2G Excel",
+        data=excel_bytes,
+        file_name="CIQ_2G.xlsx",
+        mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        type="primary",
+    )

queries/process_ciq_2g.py

@@ -0,0 +1,362 @@
+import io
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+import pandas as pd
+
+REQUIRED_DUMP_BTS_COLS = ["BSC", "BCF", "BTS", "usedMobileAllocation"]
+
+
+def _normalize_col(col: object) -> str:
+    return re.sub(r"[^0-9A-Za-z]", "", str(col))
+
+
+def _clean_columns(df: pd.DataFrame) -> pd.DataFrame:
+    df = df.copy()
+    df.columns = [_normalize_col(c) for c in df.columns]
+    return df
+
+
+def _read_dump_bts_required_columns(dump_file) -> pd.DataFrame:
+    if hasattr(dump_file, "seek"):
+        dump_file.seek(0)
+
+    hdr = pd.read_excel(
+        dump_file,
+        sheet_name="BTS",
+        engine="calamine",
+        skiprows=[0],
+        nrows=0,
+    )
+
+    original_cols = list(hdr.columns)
+    normalized_to_original: dict[str, str] = {}
+    for c in original_cols:
+        n = _normalize_col(c)
+        if n and n not in normalized_to_original:
+            normalized_to_original[n] = c
+
+    missing = [c for c in REQUIRED_DUMP_BTS_COLS if c not in normalized_to_original]
+    if missing:
+        raise ValueError(
+            f"Dump sheet 'BTS' is missing required columns after cleanup: {missing}. "
+            f"Found columns (normalized): {sorted(normalized_to_original.keys())[:50]}"
+        )
+
+    usecols = [normalized_to_original[c] for c in REQUIRED_DUMP_BTS_COLS]
+
+    if hasattr(dump_file, "seek"):
+        dump_file.seek(0)
+
+    df = pd.read_excel(
+        dump_file,
+        sheet_name="BTS",
+        engine="calamine",
+        skiprows=[0],
+        usecols=usecols,
+    )
+    df = _clean_columns(df)
+
+    df = df[REQUIRED_DUMP_BTS_COLS]
+    for c in ["BSC", "BCF", "BTS", "usedMobileAllocation"]:
+        df[c] = pd.to_numeric(df[c], errors="coerce")
+
+    return df
+
+
+@dataclass(frozen=True)
+class _PlannedSite:
+    site_name: str
+    site_number: int
+    bsc: int
+    bsc_name: str
+    name: str
+    configuration: str
+    assigned_bcf: Optional[int]
+    needed_bts_ids: tuple[int, ...]
+
+
+def _parse_site_number(site: object) -> int:
+    if not isinstance(site, str):
+        return 0
+    m = re.match(r"^(\d+)", site.strip())
+    return int(m.group(1)) if m else 0
+
+
+def _extract_band_and_sector(cell_name: object) -> tuple[Optional[str], Optional[int]]:
+    if not isinstance(cell_name, str):
+        return None, None
+
+    parts = cell_name.strip().split("_")
+    for i in range(len(parts) - 1):
+        if parts[i].isdigit() and parts[i + 1] in {"900", "1800"}:
+            sector = int(parts[i])
+            band = "G9" if parts[i + 1] == "900" else "G18"
+            return band, sector
+
+    if cell_name.endswith("_900"):
+        return "G9", None
+    if cell_name.endswith("_1800"):
+        return "G18", None
+
+    return None, None
+
+
+def _build_configuration(site_rows: pd.DataFrame) -> str:
+    rows = site_rows.copy()
+
+    rows["sector"] = pd.to_numeric(rows.get("sector"), errors="coerce")
+    rows["Nbre_TRE_DR"] = pd.to_numeric(rows.get("Nbre_TRE_DR"), errors="coerce")
+
+    configs: list[str] = []
+    for band in ["G9", "G18"]:
+        sub = rows[rows["band"] == band]
+        if sub.empty:
+            continue
+
+        sub = (
+            sub.dropna(subset=["Nbre_TRE_DR"])
+            .drop_duplicates(subset=["sector"], keep="first")
+            .sort_values(by=["sector"], na_position="last")
+        )
+        digits = "".join(str(int(v)) for v in sub["Nbre_TRE_DR"].tolist())
+        if digits:
+            configs.append(f"{band}-{digits}")
+
+    return ", ".join(configs)
+
+
+def _needed_bts_ids_from_site_rows(
+    bcf: int, site_rows: pd.DataFrame
+) -> tuple[int, ...]:
+    ids: set[int] = set()
+
+    offset_map = {
+        ("G9", 1): 1,
+        ("G9", 2): 2,
+        ("G9", 3): 3,
+        ("G18", 1): 4,
+        ("G18", 2): 5,
+        ("G18", 3): 6,
+    }
+
+    for _, r in site_rows.iterrows():
+        band = r.get("band")
+        sector = r.get("sector")
+        if (
+            band in {"G9", "G18"}
+            and isinstance(sector, (int, float))
+            and not pd.isna(sector)
+        ):
+            sector_int = int(sector)
+            off = offset_map.get((band, sector_int))
+            if off is not None:
+                ids.add(bcf + off)
+
+    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]
+    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])
+
+    sites: list[_PlannedSite] = []
+
+    for site_name, site_rows in df.groupby("Sites", dropna=False):
+        if not isinstance(site_name, str) or not site_name.strip():
+            continue
+
+        bsc_series = site_rows["BSC ID"].dropna()
+        if bsc_series.empty:
+            raise ValueError(f"Missing BSC ID for site '{site_name}'")
+        bsc = int(bsc_series.iloc[0])
+
+        bsc_name_series = site_rows["Nom BSC"].dropna()
+        bsc_name = str(bsc_name_series.iloc[0]) if not bsc_name_series.empty else ""
+
+        site_number = int(site_rows["site_number"].dropna().iloc[0])
+
+        configuration = _build_configuration(site_rows)
+
+        sites.append(
+            _PlannedSite(
+                site_name=site_name,
+                site_number=site_number,
+                bsc=bsc,
+                bsc_name=bsc_name,
+                name=f"{site_name}_NA",
+                configuration=configuration,
+                assigned_bcf=None,
+                needed_bts_ids=(),
+            )
+        )
+
+    return sorted(sites, key=lambda s: (s.bsc, s.site_number, s.site_name))
+
+
+def _assign_bcfs(
+    dump_bts: pd.DataFrame, planned_sites: list[_PlannedSite], ciq_file
+) -> 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"])
+
+    assigned: list[_PlannedSite] = []
+
+    sites_by_bsc: dict[int, list[_PlannedSite]] = {}
+    for s in planned_sites:
+        sites_by_bsc.setdefault(s.bsc, []).append(s)
+
+    for bsc, sites_in_bsc in sites_by_bsc.items():
+        sub_dump = dump_bts[dump_bts["BSC"].fillna(-1).astype(int) == int(bsc)]
+
+        used_bcfs: set[int] = set(
+            pd.to_numeric(sub_dump["BCF"], errors="coerce")
+            .dropna()
+            .astype(int)
+            .tolist()
+        )
+        used_bts: set[int] = set(
+            pd.to_numeric(sub_dump["BTS"], errors="coerce")
+            .dropna()
+            .astype(int)
+            .tolist()
+        )
+        used_mal: set[int] = set(
+            pd.to_numeric(sub_dump["usedMobileAllocation"], errors="coerce")
+            .dropna()
+            .astype(int)
+            .tolist()
+        )
+
+        sites_in_bsc_sorted = sorted(
+            sites_in_bsc, key=lambda s: (s.site_number, s.site_name)
+        )
+
+        for site in sites_in_bsc_sorted:
+            site_rows = ciq_df[ciq_df["Sites"] == site.site_name]
+            if site_rows.empty:
+                raise ValueError(f"No CIQ rows found for site '{site.site_name}'")
+
+            assigned_bcf = None
+            assigned_needed_ids: Optional[tuple[int, ...]] = None
+
+            for cand in range(10, 4401, 10):
+                if cand in used_bcfs:
+                    continue
+
+                site_needed_ids = _needed_bts_ids_from_site_rows(cand, site_rows)
+                if not site_needed_ids:
+                    continue
+
+                required_ids = tuple(cand + i for i in range(1, 7))
+
+                if any((i in used_bts) or (i in used_mal) for i in required_ids):
+                    continue
+
+                assigned_bcf = cand
+                assigned_needed_ids = site_needed_ids
+                break
+
+            if assigned_bcf is None or assigned_needed_ids is None:
+                raise ValueError(
+                    f"No available BCF found for site '{site.site_name}' on BSC {bsc}"
+                )
+
+            used_bcfs.add(assigned_bcf)
+            reserved_ids = [assigned_bcf + i for i in range(1, 7)]
+            used_bts.update(reserved_ids)
+            used_mal.update(reserved_ids)
+
+            assigned.append(
+                _PlannedSite(
+                    site_name=site.site_name,
+                    site_number=site.site_number,
+                    bsc=site.bsc,
+                    bsc_name=site.bsc_name,
+                    name=site.name,
+                    configuration=site.configuration,
+                    assigned_bcf=int(assigned_bcf),
+                    needed_bts_ids=assigned_needed_ids,
+                )
+            )
+
+    return sorted(assigned, key=lambda s: (s.bsc, s.site_number, s.site_name))
+
+
+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)
+
+    rows = []
+    for i, s in enumerate(assigned_sites, start=1):
+        rows.append(
+            {
+                "S. No.": i,
+                "Site Number": s.site_number,
+                "BSC": s.bsc,
+                "BSC Name": s.bsc_name,
+                "BCF": s.assigned_bcf,
+                "name": s.name,
+                "Configuration": s.configuration,
+            }
+        )
+
+    df_bcf = pd.DataFrame(rows)
+    return df_bcf
+
+
+def generate_ciq_2g_excel(dump_file, ciq_file) -> tuple[dict[str, pd.DataFrame], bytes]:
+    df_bcf = build_bcf_sheet(dump_file, ciq_file)
+
+    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(),
+    }
+
+    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()