forbiden BCF

queries/process_ciq_2g.py

@@ -265,8 +265,46 @@ def _parse_ciq_sites(ciq_df: pd.DataFrame) -> list[_PlannedSite]:
     return sorted(sites, key=lambda s: (s.bsc, s.site_number, s.site_name))
 
 
+def _read_forbidden_bcfs(
+    forbidden_file,
+) -> tuple[dict[int, set[int]], pd.DataFrame]:
+    if forbidden_file is None:
+        return {}, pd.DataFrame()
+
+    if hasattr(forbidden_file, "seek"):
+        forbidden_file.seek(0)
+
+    df = pd.read_excel(forbidden_file, engine="calamine")
+    df = _clean_columns(df)
+
+    if "BSCID" in df.columns and "BSC" not in df.columns:
+        df = df.rename(columns={"BSCID": "BSC"})
+    if "BCFID" in df.columns and "BCF" not in df.columns:
+        df = df.rename(columns={"BCFID": "BCF"})
+
+    required = ["BSC", "BCF"]
+    missing = [c for c in required if c not in df.columns]
+    if missing:
+        raise ValueError(f"Forbidden BCF list is missing required columns: {missing}")
+
+    df["BSC"] = pd.to_numeric(df["BSC"], errors="coerce")
+    df["BCF"] = pd.to_numeric(df["BCF"], errors="coerce")
+    df = df.dropna(subset=["BSC", "BCF"]).copy()
+    df["BSC"] = df["BSC"].astype(int)
+    df["BCF"] = df["BCF"].astype(int)
+
+    forbidden_by_bsc: dict[int, set[int]] = {}
+    for bsc, group in df.groupby("BSC"):
+        forbidden_by_bsc[int(bsc)] = set(group["BCF"].tolist())
+
+    return forbidden_by_bsc, df[["BSC", "BCF"]]
+
+
 def _assign_bcfs(
-    dump_bts: pd.DataFrame, planned_sites: list[_PlannedSite], ciq_df: pd.DataFrame
+    dump_bts: pd.DataFrame,
+    planned_sites: list[_PlannedSite],
+    ciq_df: pd.DataFrame,
+    forbidden_by_bsc: dict[int, set[int]] | None = None,
 ) -> list[_PlannedSite]:
 
     dump_bts = dump_bts.dropna(subset=["BSC"])
@@ -277,8 +315,11 @@ def _assign_bcfs(
     for s in planned_sites:
         sites_by_bsc.setdefault(s.bsc, []).append(s)
 
+    forbidden_by_bsc = forbidden_by_bsc or {}
+
     for bsc, sites_in_bsc in sites_by_bsc.items():
         sub_dump = dump_bts[dump_bts["BSC"].fillna(-1).astype(int) == int(bsc)]
+        forbidden = forbidden_by_bsc.get(int(bsc), set())
 
         used_bcfs: set[int] = set(
             pd.to_numeric(sub_dump["BCF"], errors="coerce")
@@ -312,7 +353,7 @@ def _assign_bcfs(
             assigned_needed_ids: Optional[tuple[int, ...]] = None
 
             for cand in range(10, 4401, 10):
-                if cand in used_bcfs:
+                if cand in used_bcfs or cand in forbidden:
                     continue
 
                 site_needed_ids = _needed_bts_ids_from_site_rows(cand, site_rows)
@@ -354,11 +395,14 @@ def _assign_bcfs(
     return sorted(assigned, key=lambda s: (s.bsc, s.site_number, s.site_name))
 
 
-def build_bcf_sheet(dump_file, ciq_file) -> pd.DataFrame:
+def build_bcf_sheet(dump_file, ciq_file, forbidden_file=None) -> 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)
+    forbidden_by_bsc, _ = _read_forbidden_bcfs(forbidden_file)
+    assigned_sites = _assign_bcfs(
+        dump_bts, planned_sites, ciq_df, forbidden_by_bsc=forbidden_by_bsc
+    )
     return _build_bcf_sheet_from_assigned_sites(assigned_sites)
 
 
@@ -381,6 +425,50 @@ def _build_bcf_sheet_from_assigned_sites(
     return pd.DataFrame(rows)
 
 
+def _build_free_bcf_sheet(
+    dump_bts: pd.DataFrame, forbidden_by_bsc: dict[int, set[int]] | None = None
+) -> pd.DataFrame:
+    rows = []
+    forbidden_by_bsc = forbidden_by_bsc or {}
+    bsc_values = (
+        pd.to_numeric(dump_bts["BSC"], errors="coerce").dropna().astype(int).unique()
+    )
+    for bsc in sorted(bsc_values):
+        sub_dump = dump_bts[dump_bts["BSC"].fillna(-1).astype(int) == int(bsc)]
+        forbidden = forbidden_by_bsc.get(int(bsc), set())
+
+        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()
+        )
+
+        for cand in range(10, 4401, 10):
+            if cand in used_bcfs or cand in forbidden:
+                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
+
+            rows.append({"BSC": int(bsc), "BCF": int(cand)})
+
+    return pd.DataFrame(rows)
+
+
 def _sector_id_from_band_sector(band: object, sector: object) -> int:
     if band not in {"G9", "G18"}:
         raise ValueError(f"Invalid band '{band}'")
@@ -530,12 +618,17 @@ def _build_trx_sheet_from_assigned_sites(
     return df_trx
 
 
-def build_bts_sheet(dump_file, ciq_file, mcc: int = 610, mnc: int = 2) -> pd.DataFrame:
+def build_bts_sheet(
+    dump_file, ciq_file, mcc: int = 610, mnc: int = 2, forbidden_file=None
+) -> 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)
+    forbidden_by_bsc, _ = _read_forbidden_bcfs(forbidden_file)
+    assigned_sites = _assign_bcfs(
+        dump_bts, planned_sites, ciq_df, forbidden_by_bsc=forbidden_by_bsc
+    )
     return _build_bts_sheet_from_assigned_sites(
         ciq_df, assigned_sites, mcc=mcc, mnc=mnc
     )
@@ -686,14 +779,18 @@ def _build_mal_sheet_from_assigned_sites(
 
 
 def generate_ciq_2g_excel(
-    dump_file, ciq_file, mcc: int = 610, mnc: int = 2
+    dump_file, ciq_file, mcc: int = 610, mnc: int = 2, forbidden_file=None
 ) -> 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)
+    forbidden_by_bsc, forbidden_df = _read_forbidden_bcfs(forbidden_file)
+    assigned_sites = _assign_bcfs(
+        dump_bts, planned_sites, ciq_df, forbidden_by_bsc=forbidden_by_bsc
+    )
 
     df_bcf = _build_bcf_sheet_from_assigned_sites(assigned_sites)
+    df_bcf_free = _build_free_bcf_sheet(dump_bts, forbidden_by_bsc=forbidden_by_bsc)
     df_bts = _build_bts_sheet_from_assigned_sites(
         ciq_df, assigned_sites, mcc=mcc, mnc=mnc
     )
@@ -741,6 +838,7 @@ def generate_ciq_2g_excel(
 
     sheets: dict[str, pd.DataFrame] = {
         "BCF": df_bcf,
+        "BCF_LIBRE": df_bcf_free,
         "BTS": df_bts,
         "BTS_GPRS": df_bts_min,
         "BTS_AMR": df_bts_min,
@@ -751,6 +849,9 @@ def generate_ciq_2g_excel(
         "TRX": df_trx,
     }
 
+    if not forbidden_df.empty:
+        sheets["BCF_FORBIDDEN"] = forbidden_df
+
     bytes_io = io.BytesIO()
     with pd.ExcelWriter(bytes_io, engine="xlsxwriter") as writer:
         for sheet_name, df in sheets.items():