""" Subdivision Cluster / FIPS-Validation pipeline — Hugging Face Space (Gradio). Faithful port of the Colab notebook `LatestUndeterminedZipCityCensusTract70_Threshold_95_Clusters.ipynb`. Colab-specific I/O replaced: * Drive input folder -> file upload (multiple .xlsx/.xls) * google.colab.files -> Gradio file outputs (written to a temp dir) * Drive safety-list path -> get_safety_list_path() (repo file OR private dataset) All processing logic (FIPS validation, RawFile snapshot, RapidFuzz clustering, wide tables, majority-vote replacement, column reorder, OnlyBlanks warnings, chunked CSV export, self-test) matches the notebook. """ import os import re import gc import math import time import tempfile import zipfile from pathlib import Path from collections import defaultdict import numpy as np import pandas as pd import gradio as gr from rapidfuzz import fuzz # ===================================================================== # REFERENCE FILE LOADER (the Subdivision Program Safety List) # --------------------------------------------------------------------- # 1. A file committed in the Space repo (override name via SAFETY_LIST_FILE). # Use ONLY if the Space is PRIVATE. # 2. A private HF *dataset* repo, downloaded with the HF_TOKEN secret. Set: # SAFETY_LIST_REPO e.g. "your-username/subdivision-safety-list" # SAFETY_LIST_REPO_FILE filename inside that dataset (optional) # HF_TOKEN read token, stored as a Space Secret # ===================================================================== DEFAULT_SAFETY_LIST_NAME = "Subdivision_Program_Safety_List.xlsx" EXCEL_ROW_LIMIT = 1_048_575 # matches the notebook # Fixed pipeline settings (notebook values). Not exposed in the UI on purpose: # users should not be able to deviate from the notebook's behavior. STRICT_BLANK_CITIES = True # blank cities that are not in the safety list CLUSTER_THRESHOLD = 95 # RapidFuzz similarity for LegalSubDivision clusters REPLACE_THRESHOLD = 70 # majority-vote % required to fill a blank FILTER_DOMINANT_COUNTY = False # process ALL counties (multi-county / full-state files) # Master bundle format: # "excel" -> Master_Dataset.xlsx, every result on its own sheet (full output # split across Output_..._1/_2/... if it exceeds Excel's row cap) # "csv" -> Master_Dataset.zip, every result bundled as a CSV (faster: no Excel # writing -- use this for very large / state-sized runs) # The individual per-category files are CSV in both modes. MASTER_OUTPUT = "excel" def get_safety_list_path() -> str: local_name = os.environ.get("SAFETY_LIST_FILE", DEFAULT_SAFETY_LIST_NAME) if Path(local_name).exists(): return local_name repo_id = os.environ.get("SAFETY_LIST_REPO") token = os.environ.get("HF_TOKEN") if repo_id and token: from huggingface_hub import hf_hub_download filename = os.environ.get("SAFETY_LIST_REPO_FILE", local_name) return hf_hub_download( repo_id=repo_id, filename=filename, repo_type="dataset", token=token ) raise FileNotFoundError( "Safety list not found.\n" f" - Commit '{local_name}' into this Space (keep the Space private), OR\n" " - Set SAFETY_LIST_REPO + HF_TOKEN to pull it from a private dataset." ) _SAFETY_CACHE: dict | None = None def build_whitelist_from_path(path): """Parse a safety-list file (xlsx or csv) into {pairs, cities, rows}.""" if str(path).lower().endswith(".csv"): safety_df = pd.read_csv(path) else: safety_df = pd.read_excel(path) acceptable_cols = [c for c in safety_df.columns if str(c).startswith("Acceptable FIPS")] valid_pairs, valid_cities = set(), set() for _, r in safety_df.iterrows(): c = _norm_city(r["City Name"]) if c is None: continue valid_cities.add(c) for col in acceptable_cols: f = _norm_fips(r[col]) if f: valid_pairs.add(c + "||" + f) return {"pairs": valid_pairs, "cities": valid_cities, "rows": len(safety_df)} def load_whitelist(): """Server-side safety list (repo file or private dataset). Cached per process.""" global _SAFETY_CACHE if _SAFETY_CACHE is None: _SAFETY_CACHE = build_whitelist_from_path(get_safety_list_path()) return _SAFETY_CACHE # ===================================================================== # NORMALIZATION HELPERS (identical to the notebook) # ===================================================================== PLACEHOLDERS = {"UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL", "NAN", "NONE", ""} def _norm_city(x): if pd.isna(x): return None return str(x).strip().upper() def _norm_fips(x): """5-digit FIPS string, or None. Handles 28071, 28071.0, '01067', '28071 - LAFAYETTE'.""" if pd.isna(x): return None m = re.search(r"\d+", str(x)) return str(int(m.group(0))).zfill(5) if m else None # --------------------------------------------------------------------- # FIPS-COUNTY COLUMN DETECTION (by CONTENT, not header name) # --------------------------------------------------------------------- # The source header has changed before ("** REC FIPS-COUNTY" -> # "** REC FIPS-COUNTY NAME"), so we identify the column by what's IN it: # values shaped like "05045-FAULKNER" (FIPS digits, a dash, a county name). # Header names are only used as a last-resort fallback. FIPS_COUNTY_PATTERN = r"^\s*\d{1,5}\s*-\s*[A-Za-z]" FALLBACK_FIPS_HEADERS = ["** REC FIPS-COUNTY NAME", "** REC FIPS-COUNTY", "County"] def detect_fips_county_column(df, sample_size=1000, min_score=0.5): """Return (column_name, reason). Picks the column whose values best match the 'FIPS-COUNTYNAME' shape; falls back to known headers, then to any header mentioning FIPS.""" best_col, best_score = None, 0.0 for col in df.columns: s = df[col].dropna() if s.empty: continue sample = s.head(sample_size).astype(str) try: score = float(sample.str.match(FIPS_COUNTY_PATTERN).mean()) except Exception: continue if score > best_score: best_col, best_score = col, score if best_col is not None and best_score >= min_score: return best_col, f"content match {best_score:.0%} of sampled rows" for name in FALLBACK_FIPS_HEADERS: if name in df.columns: return name, "fallback: known header name" for col in df.columns: if "FIPS" in str(col).upper(): return col, "fallback: header mentions FIPS" return None, "not found" # ===================================================================== # SELF-TEST (startup health check — same logic as the notebook's cell) # ===================================================================== def run_self_test() -> str: path = get_safety_list_path() st = pd.read_excel(path) fc = [c for c in st.columns if str(c).startswith("Acceptable FIPS")] pairs = set() for _, r in st.iterrows(): c = _norm_city(r["City Name"]) if c is None: continue for col in fc: f = _norm_fips(r[col]) if f: pairs.add(c + "||" + f) good_city = st.iloc[0]["City Name"] good_fips = st.iloc[0]["Acceptable FIPS 1"] test = pd.DataFrame({ "DDPPropertyCityFinal": [good_city, good_city], "** REC FIPS-COUNTY": [good_fips, 99999], # row 2 = wrong FIPS on purpose }) cn = test["DDPPropertyCityFinal"].map(_norm_city) fn = test["** REC FIPS-COUNTY"].map(_norm_fips) key = cn.fillna("") + "||" + fn.fillna("") ok = key.isin(pairs) assert bool(ok.iloc[0]) and not bool(ok.iloc[1]), \ "SELF-TEST FAILED -- matching logic is not working as expected." return f"SELF-TEST PASSED: known-good ({good_city}) kept, wrong FIPS flagged." # ===================================================================== # STEP 1 — FIPS SAFETY-LIST VALIDATION # ===================================================================== def fips_validate(df: pd.DataFrame, blank_cities_not_in_list: bool, log, wl=None, fips_col=None): if wl is None: wl = load_whitelist() valid_pairs, valid_cities = wl["pairs"], wl["cities"] if fips_col is None: fips_col, _reason = detect_fips_county_column(df) if fips_col is None: raise ValueError( "Could not find a FIPS-county column. Expected values like " "'05045-FAULKNER' in one of the columns." ) FIPS_COL = fips_col CITY_COL, ZIP_COL, TRACT_COL = ( "DDPPropertyCityFinal", "DDPPropertyZIPFinal", "DDPRevisedCensusTract", ) city_norm = df[CITY_COL].map(_norm_city) fips_norm = df[FIPS_COL].map(_norm_fips) row_key = city_norm.fillna("") + "||" + fips_norm.fillna("") is_real_city = df[CITY_COL].notna() & (~city_norm.fillna("").isin(PLACEHOLDERS)) city_in_list = city_norm.isin(valid_cities) pair_ok = row_key.isin(valid_pairs) if blank_cities_not_in_list: fails_mask = is_real_city & (~pair_ok) else: fails_mask = is_real_city & city_in_list & (~pair_ok) n_real = int(is_real_city.sum()) n_inlist = int((is_real_city & city_in_list).sum()) n_valid = int((is_real_city & pair_ok).sum()) n_fail = int(fails_mask.sum()) log.append( f"FIPS validation — list: {wl['rows']:,} rows | {len(valid_cities):,} cities | " f"{len(valid_pairs):,} pairs" ) log.append( f" real:{n_real:,} in-list:{n_inlist:,} valid-kept:{n_valid:,} blanked:{n_fail:,}" ) if n_real > 0 and n_inlist == 0: log.append(" !! 0 cities matched the list — likely a format/spelling mismatch.") # Audit built BEFORE overwrite (insert position matches the notebook). key_col = "RecordKey" if "RecordKey" in df.columns else None base_cols = [c for c in [key_col, CITY_COL, FIPS_COL, ZIP_COL, TRACT_COL] if c] audit = df.loc[fails_mask, base_cols].copy() audit.insert( len(base_cols) if key_col is None else 2, "NormalizedFIPS", fips_norm[fails_mask].values, ) audit["Reason"] = np.where( city_in_list[fails_mask].values, "FIPS does not match city", "City not in safety list" ) audit = audit.rename( columns={CITY_COL: "OLD_City", ZIP_COL: "OLD_ZIP", TRACT_COL: "OLD_CensusTract"} ) # Recent pandas reads all-numeric Excel columns as float64 and refuses to # accept a string into them; Colab's older pandas silently upcast to object. # Cast first so the placeholder write reproduces the notebook's result. df[CITY_COL] = df[CITY_COL].astype(object) df[ZIP_COL] = df[ZIP_COL].astype(object) df.loc[fails_mask, CITY_COL] = "UNDETERMINED CITY" df.loc[fails_mask, ZIP_COL] = "UNDETERMINED ZIP" df.loc[fails_mask, TRACT_COL] = np.nan return df, audit # ===================================================================== # STEP 2 — RAPIDFUZZ CLUSTERING (identical to the notebook) # ===================================================================== def group_similar_strings_optimized(strings, threshold=95, prefix_len=3, max_len_diff=3): cleaned, seen = [], set() for s in strings: if pd.isna(s): continue s = str(s).strip() if s and s not in seen: cleaned.append(s) seen.add(s) # Canonical order so the SAME set of names always clusters the same way, # regardless of the order rows arrived in (single-county vs whole-state, # app vs app_simple_ui, or any file-upload order). The greedy grouping below # is order-sensitive, so we sort first to make it deterministic. cleaned.sort(key=lambda x: (x.upper(), x)) visited, groups = set(), [] buckets = defaultdict(list) for s in cleaned: buckets[s.upper()[:prefix_len]].append(s) for key in sorted(buckets.keys()): bucket = buckets[key] for i, base in enumerate(bucket): if base in visited: continue group = [base] visited.add(base) base_upper = base.upper() base_len = len(base_upper) for j in range(i + 1, len(bucket)): compare = bucket[j] if compare in visited: continue comp_upper = compare.upper() if abs(base_len - len(comp_upper)) > max_len_diff: continue if fuzz.ratio(base_upper, comp_upper) >= threshold: group.append(compare) visited.add(compare) groups.append(group) return groups def build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold): all_legal = pd.concat( [ city_df[["County", "LegalSubDivision"]], zip_df[["County", "LegalSubDivision"]], census_tracts_df[["County", "LegalSubDivision"]], ], ignore_index=True, ).dropna(subset=["County", "LegalSubDivision"]) all_legal["County"] = all_legal["County"].astype(str).str.strip() all_legal["LegalSubDivision"] = all_legal["LegalSubDivision"].astype(str).str.strip() all_legal = all_legal[(all_legal["County"] != "") & (all_legal["LegalSubDivision"] != "")] all_legal = all_legal[["County", "LegalSubDivision"]].drop_duplicates() groups = [] # (county, [members]) so each cluster stays tied to its county for county in sorted(all_legal["County"].unique()): strings = all_legal[all_legal["County"] == county]["LegalSubDivision"].tolist() for grp in group_similar_strings_optimized( strings, threshold=cluster_threshold, prefix_len=3, max_len_diff=3): groups.append((county, grp)) rows = [] for cid, (county, group) in enumerate(groups, start=1): for sub in group: rows.append({"Cluster_ID": cid, "County": county, "LegalSubDivision": sub}) return pd.DataFrame(rows, columns=["Cluster_ID", "County", "LegalSubDivision"]) # ===================================================================== # STEP 3 — WIDE TABLES + MAJORITY-VOTE "REPLACE WITH" # ===================================================================== def build_cluster_wide_table(source_df, cluster_df, value_col, prefix): # Map by (County, LegalSubDivision) -- NOT LegalSubDivision alone. The same # subdivision name occurs in many counties, each with its own cluster; keying # on the name only would collapse them and pool values across counties. lk = cluster_df[["County", "LegalSubDivision", "Cluster_ID"]].drop_duplicates( ["County", "LegalSubDivision"]) cluster_map = {(c, s): cid for c, s, cid in zip(lk["County"].astype(str).str.strip(), lk["LegalSubDivision"].astype(str).str.strip(), lk["Cluster_ID"])} w = source_df[["County", "LegalSubDivision", value_col]].copy() w["County"] = w["County"].astype(str).str.strip() w["LegalSubDivision"] = w["LegalSubDivision"].astype(str).str.strip() w[value_col] = w[value_col].replace("", np.nan) w["Cluster_ID"] = [cluster_map.get((c, s)) for c, s in zip(w["County"], w["LegalSubDivision"])] w = w.dropna(subset=["Cluster_ID", value_col]) if w.empty: return pd.DataFrame(columns=["Cluster_ID"]) w["Cluster_ID"] = w["Cluster_ID"].astype(int) w[value_col] = w[value_col].astype(str).str.strip() w = w[w[value_col] != ""] counts = w.groupby(["Cluster_ID", value_col], observed=True).size().unstack(fill_value=0) totals = counts.sum(axis=1) pct = counts.div(totals.replace(0, np.nan), axis=0).mul(100).fillna(0) counts.columns = [f"{prefix}{c}_Count" for c in counts.columns] pct.columns = [f"{prefix}{c}_Percentage" for c in pct.columns] wide = pd.concat([counts, pct], axis=1).reset_index() # Match the notebook: Cluster_ID first, all other columns sorted. other = sorted([c for c in wide.columns if c != "Cluster_ID"]) return wide[["Cluster_ID"] + other] def build_replace_with(wide_df, prefix_to_remove, threshold): pct_cols = [c for c in wide_df.columns if c.endswith("_Percentage")] if not pct_cols: res = wide_df[["Cluster_ID"]].copy() res["Replace with"] = None return res.drop_duplicates(subset=["Cluster_ID"]) labels = np.array( [c.replace(prefix_to_remove, "").replace("_Percentage", "") for c in pct_cols], dtype=object, ) values = wide_df[pct_cols].to_numpy(copy=False) out = [] for row in values: matched = labels[row >= threshold] out.append(", ".join(matched) if len(matched) else None) res = wide_df[["Cluster_ID"]].copy() res["Replace with"] = out return res.drop_duplicates(subset=["Cluster_ID"]) def clean_census_tract_suffix(value): if pd.isna(value): return np.nan try: f = float(value) return str(int(f)) if f == int(f) else str(f) except ValueError: return str(value) def clean_zip(value): """Normalize ZIPs read from Excel as numbers: drop the float '.0' artifact and zero-pad to 5 digits (recovering leading zeros lost on numeric import, e.g. 7221.0 -> 07221). Handles comma-joined majority-vote results. ZIP+4 left as-is.""" if pd.isna(value): return np.nan out = [] for part in str(value).split(","): p = part.strip() if p.endswith(".0"): p = p[:-2] digits = re.sub(r"\D", "", p) if not digits: continue out.append(digits.zfill(5) if len(digits) <= 5 else digits) return ", ".join(out) if out else np.nan _BOOL_MAP = { True: "TRUE", False: "FALSE", 1: "TRUE", 0: "FALSE", "1": "TRUE", "0": "FALSE", "true": "TRUE", "false": "FALSE", "t": "TRUE", "f": "FALSE", "yes": "TRUE", "no": "FALSE", } def boolify(series): """Render a 0/1 (or True/False) column as literal 'TRUE'/'FALSE' text. Recognized values are mapped; anything unrecognized is left as-is; NaN kept.""" def conv(v): if pd.isna(v): return v if v in _BOOL_MAP: return _BOOL_MAP[v] return _BOOL_MAP.get(str(v).strip().lower(), v) return series.map(conv) # ===================================================================== # CHUNKED CSV WRITER (faithful port of save_csv_in_parts) # ===================================================================== def save_csv_in_parts(df, base_filename, out_dir, row_limit=1_000_000): base_name = base_filename.replace(".csv", "") created = [] total = len(df) num_parts = math.ceil(total / row_limit) for i in range(num_parts): start = i * row_limit end = min((i + 1) * row_limit, total) part = f"{base_name}_part_{i + 1}.csv" if num_parts > 1 else base_filename p = out_dir / part df.iloc[start:end].to_csv(p, index=False) created.append(str(p)) gc.collect() return created def write_master_workbook(path, sheets_in_order, row_limit=1_048_575): """Write a multi-sheet .xlsx from [(sheet_name, df), ...]. Any sheet larger than Excel's 1,048,576-row cap is split into _1, _2, ... so nothing is dropped. Uses xlsxwriter in constant_memory mode, driven row-by-row (low RAM -- important for the full-output sheet on state-sized files). We write the rows ourselves because the pandas + constant_memory path corrupts rows that contain NaN cells. Falls back to openpyxl if xlsxwriter isn't installed.""" try: import xlsxwriter except Exception: xlsxwriter = None if xlsxwriter is not None: wb = xlsxwriter.Workbook(str(path), {"constant_memory": True, "nan_inf_to_errors": True}) def _write(name, dfx): ws = wb.add_worksheet(str(name)[:31]) ws.write_row(0, 0, [str(c) for c in dfx.columns]) clean = dfx.where(pd.notna(dfx), None) # NaN -> blank cell for r, rowvals in enumerate(clean.itertuples(index=False, name=None), start=1): ws.write_row(r, 0, rowvals) for name, dfx in sheets_in_order: if dfx is None: continue n = len(dfx) if n <= row_limit: _write(name, dfx) else: for i in range(math.ceil(n / row_limit)): _write(f"{name}_{i + 1}", dfx.iloc[i * row_limit:(i + 1) * row_limit]) gc.collect() wb.close() else: with pd.ExcelWriter(path, engine="openpyxl") as writer: for name, dfx in sheets_in_order: if dfx is None: continue n = len(dfx) if n <= row_limit: dfx.to_excel(writer, sheet_name=str(name)[:31], index=False) else: for i in range(math.ceil(n / row_limit)): dfx.iloc[i * row_limit:(i + 1) * row_limit].to_excel( writer, sheet_name=f"{name}_{i + 1}"[:31], index=False) return str(path) # ===================================================================== # MAIN PIPELINE # ===================================================================== def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list, filter_dominant_county, cluster_threshold, replace_threshold): t_start = time.time() log = [] if not uploaded_files: return "Please upload at least one Excel file.", [] out_dir = Path(tempfile.mkdtemp(prefix="subdiv_")) outputs = [] # ---- Resolve the reference (safety) list ------------------------- # Default: server-side list (repo file or private dataset). If the user # uploads an override on the sidebar, use it for THIS run only. try: if reference_file is not None: wl = build_whitelist_from_path(reference_file.name) log.append(f"Using UPLOADED reference list: {Path(reference_file.name).name}") else: wl = load_whitelist() log.append("Using server-side safety list.") except FileNotFoundError as e: return "\n".join(log) + f"\n\nREFERENCE FILE ERROR:\n{e}", [] except Exception as e: return "\n".join(log) + f"\n\nCould not parse reference list: {e}", [] # ---- Load & merge uploads (Excel or CSV) ------------------------- dfs = [] for f in uploaded_files: name = f.name try: if str(name).lower().endswith(".csv"): dfs.append(pd.read_csv(name)) else: dfs.append(pd.read_excel(name)) log.append(f"Loaded {Path(name).name}") except Exception as e: log.append(f"ERROR loading {Path(name).name}: {e}") if not dfs: return "\n".join(log) + "\nNo files could be read.", [] df = pd.concat(dfs, ignore_index=True) log.append(f"Merged shape: {df.shape}") # ---- Locate the FIPS-county column by CONTENT -------------------- fips_col, fips_reason = detect_fips_county_column(df) if fips_col is None: return ("\n".join(log) + "\n\nCOLUMN ERROR: could not find a FIPS-county column. Expected " "values like '05045-FAULKNER' in one of the columns."), [] log.append(f"FIPS-county column: '{fips_col}' ({fips_reason})") # ---- FIPS validation (uses the reference list) ------------------- try: df, audit = fips_validate(df, blank_cities_not_in_list, log, wl=wl, fips_col=fips_col) except FileNotFoundError as e: return "\n".join(log) + f"\n\nREFERENCE FILE ERROR:\n{e}", [] # ---- Verification files (audit + RawFile_FIPSValidated), CSV ----- if len(audit) > 0: ap = out_dir / "FIPS_Validation_Audit_ChangedRecords.csv" audit.to_csv(ap, index=False) outputs.append(str(ap)) rp = out_dir / "RawFile_FIPSValidated.csv" df.to_csv(rp, index=False) outputs.append(str(rp)) # ---- UNDETERMINED/NULL -> NaN ------------------------------------ for col in ["DDPPropertyCityFinal", "DDPPropertyZIPFinal", "DDPRevisedCensusTract", "LegalSubDivision"]: if col in df.columns: df[col] = df[col].replace(["UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL"], np.nan) # ---- Rename detected FIPS col -> County -------------------------- if fips_col != "County": if "County" in df.columns: df = df.drop(columns=["County"]) df = df.rename(columns={fips_col: "County"}) # ---- Filter to single most-dominant county ----------------------- # DISABLED: we now process big multi-county files (up to a full state) in # one run, so we keep every county. Clustering already runs per-county. # To re-enable single-county mode, uncomment the block below. # if filter_dominant_county: # dominant = df["County"].value_counts().idxmax() # df = df[df["County"] == dominant].copy() # log.append(f"Filtered to dominant county: {dominant} -> {df.shape}") # ---- Source frames ----------------------------------------------- city_df = df.loc[df["DDPPropertyCityFinal"].notna()].copy() zip_df = df.loc[df["DDPPropertyZIPFinal"].notna()].copy() census_tracts_df = df.loc[df["DDPRevisedCensusTract"].notna()].copy() original = df.loc[df["LegalSubDivision"].notna()].copy() # ---- Cluster ------------------------------------------------------ cluster_df = build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold) log.append(f"Clusters: {cluster_df['Cluster_ID'].nunique():,}") # ---- Replace-with (majority vote) -------------------------------- zip_w = build_replace_with( build_cluster_wide_table(zip_df, cluster_df, "DDPPropertyZIPFinal", "ZIP_"), "ZIP_", replace_threshold).rename(columns={"Replace with": "Replace with Zip Codes"}) city_w = build_replace_with( build_cluster_wide_table(city_df, cluster_df, "DDPPropertyCityFinal", "City_"), "City_", replace_threshold).rename(columns={"Replace with": "Replace with Cities"}) cen_w = build_replace_with( build_cluster_wide_table(census_tracts_df, cluster_df, "DDPRevisedCensusTract", "CensusTract_"), "CensusTract_", replace_threshold).rename(columns={"Replace with": "Replace with Census Tracts"}) # ---- Map back onto original -------------------------------------- # Same fix: assign each row its county's cluster, keyed on (County, name). lk = cluster_df[["County", "LegalSubDivision", "Cluster_ID"]].drop_duplicates( ["County", "LegalSubDivision"]) cmap = {(c, s): cid for c, s, cid in zip(lk["County"].astype(str).str.strip(), lk["LegalSubDivision"].astype(str).str.strip(), lk["Cluster_ID"])} original_final = original.copy() _oc = original_final["County"].astype(str).str.strip() _os = original_final["LegalSubDivision"].astype(str).str.strip() original_final["Cluster_ID"] = [cmap.get((c, s)) for c, s in zip(_oc, _os)] original_final["Replace with Zip Codes"] = original_final["Cluster_ID"].map( dict(zip(zip_w["Cluster_ID"], zip_w["Replace with Zip Codes"]))) original_final["Replace with Cities"] = original_final["Cluster_ID"].map( dict(zip(city_w["Cluster_ID"], city_w["Replace with Cities"]))) original_final["Replace with Census Tracts"] = original_final["Cluster_ID"].map( dict(zip(cen_w["Cluster_ID"], cen_w["Replace with Census Tracts"]))) # Normalize ZIPs (drop float '.0', restore 5-digit zero-padding). original_final["Replace with Zip Codes"] = ( original_final["Replace with Zip Codes"].apply(clean_zip)) # ---- Reorder columns (desired_leading_columns) ------------------- desired_leading = [ "RecordKey", "ComesFromMtg2", "IsFlagged", "Cluster_ID", "LegalSubDivision", "County", "Replace with Zip Codes", "Replace with Cities", "Replace with Census Tracts", "DDPPropertyZIPFinal", "DDPPropertyCityFinal", "DDPRevisedCensusTract", "LotBlockSubdivisionTract", "DDPRevisedLatitude", "DDPRevisedLongitude", ] existing = original_final.columns.tolist() lead = [c for c in desired_leading if c in existing] other = [c for c in existing if c not in lead] original_final = original_final[lead + other] original_final_output = original_final.copy() # full output snapshot # ---- Split into final cities / zips / census tracts -------------- keep = ["RecordKey", "ComesFromMtg2", "IsFlagged", "Replace with Zip Codes", "Replace with Cities", "Replace with Census Tracts", "DDPPropertyZIPFinal", "DDPPropertyCityFinal", "DDPRevisedCensusTract"] keep = [c for c in keep if c in original_final.columns] of = original_final[keep].rename(columns={ "DDPPropertyZIPFinal": "DDPPropertyZIPFinalToRemove", "DDPPropertyCityFinal": "DDPPropertyCityFinalToRemove", "DDPRevisedCensusTract": "DDPRevisedCensusTractToRemove", "Replace with Zip Codes": "DDPPropertyZIPFinal", "Replace with Cities": "DDPPropertyCityFinal", "Replace with Census Tracts": "DDPRevisedCensusTract", }) of["DDPRevisedCensusTract"] = of["DDPRevisedCensusTract"].apply(clean_census_tract_suffix) idcols = [c for c in ["RecordKey", "ComesFromMtg2", "IsFlagged"] if c in of.columns] final_cities = of[idcols + ["DDPPropertyCityFinalToRemove", "DDPPropertyCityFinal"]].copy() final_zips = of[idcols + ["DDPPropertyZIPFinalToRemove", "DDPPropertyZIPFinal"]].copy() final_ct = of[idcols + ["DDPRevisedCensusTractToRemove", "DDPRevisedCensusTract"]].copy() # keep only rows whose ORIGINAL value was blank (the ones we filled) final_cities = final_cities[final_cities["DDPPropertyCityFinalToRemove"].isna()].copy() final_zips = final_zips[final_zips["DDPPropertyZIPFinalToRemove"].isna()].copy() final_ct = final_ct[final_ct["DDPRevisedCensusTractToRemove"].isna()].copy() final_cities = final_cities.drop(columns=["DDPPropertyCityFinalToRemove"]) final_zips = final_zips.drop(columns=["DDPPropertyZIPFinalToRemove"]) final_ct = final_ct.drop(columns=["DDPRevisedCensusTractToRemove"]) # ---- OnlyBlanks warnings (column entirely null) ------------------ def only_blanks_warning(d, col, prefix): if col in d.columns and d[col].isnull().all(): p = out_dir / f"OnlyBlanks_{prefix}.csv" d.to_csv(p, index=False) outputs.append(str(p)) log.append(f"WARNING: {prefix} has only blanks in '{col}' -> {p.name}") only_blanks_warning(final_cities, "DDPPropertyCityFinal", "final_cities") only_blanks_warning(final_zips, "DDPPropertyZIPFinal", "final_zips") only_blanks_warning(final_ct, "DDPRevisedCensusTract", "final_census_tracts") # ---- Drop NaNs so only program assignments remain ---------------- final_cities = final_cities[final_cities["DDPPropertyCityFinal"].notna()].copy() final_zips = final_zips[final_zips["DDPPropertyZIPFinal"].notna()].copy() final_ct = final_ct[final_ct["DDPRevisedCensusTract"].notna()].copy() # ---- Boolean columns as TRUE/FALSE (platform can't read 0/1) ----- for d in (final_cities, final_zips, final_ct): for bc in ("ComesFromMtg2", "IsFlagged"): if bc in d.columns: d[bc] = boolify(d[bc]) # ---- Write the per-category + cluster outputs (CSV) -------------- for d, name in [(final_cities, "final_cities.csv"), (final_zips, "final_zips.csv"), (final_ct, "final_census_tracts.csv"), (cluster_df, "cluster_df.csv")]: p = out_dir / name d.to_csv(p, index=False) outputs.append(str(p)) # ---- Full output CSV (chunked at 1,000,000 rows) ----------------- # NB: no '%' in the filename -- '%' is the URL-encoding character and makes # the browser download fail with "unknown server error". csv_name = f"OutputFile_{replace_threshold}Threshold_{cluster_threshold}ClustersRapidFuzz.csv" outputs.extend(save_csv_in_parts(original_final_output, csv_name, out_dir, row_limit=1_000_000)) # ---- Summary sheet: ONE ROW PER COUNTY --------------------------- # Totals (records / subdivisions) come from the FULL post-validation frame # (df -- every row, incl. those with no LegalSubDivision). The blank/fill/ # cluster metrics come from the clustered subset (output_df), which is the # only set the pipeline can actually fill. Other output files are unchanged. output_df = original_final_output # rows WITH a LegalSubDivision + fills full_df = df # ALL post-validation rows, has County city_df, zips_df, census_df = final_cities, final_zips, final_ct def _fips_name(label): raw = str(label) parts = raw.split("-") fips = parts[0].strip() name = parts[1].strip().title() if len(parts) > 1 else raw.strip().title() return fips, name RW_CITY, RW_ZIP, RW_TRACT = ("Replace with Cities", "Replace with Zip Codes", "Replace with Census Tracts") ORIG_CITY, ORIG_ZIP, ORIG_TRACT = ("DDPPropertyCityFinal", "DDPPropertyZIPFinal", "DDPRevisedCensusTract") def _blank(g, col): return g[col].isna() if col in g.columns else pd.Series(False, index=g.index) def _fill(g, col): return g[col].notna() if col in g.columns else pd.Series(False, index=g.index) def _out_subset(label): if "County" not in output_df.columns: return output_df.iloc[0:0] m = output_df["County"].isna() if pd.isna(label) else (output_df["County"] == label) return output_df[m] summary_rows = [] if "County" in full_df.columns and len(full_df): for county_label, gfull in full_df.groupby("County", dropna=False): fips, name = _fips_name(county_label) g = _out_subset(county_label) # clustered rows for this county city_blank, zip_blank, tract_blank = (_blank(g, ORIG_CITY), _blank(g, ORIG_ZIP), _blank(g, ORIG_TRACT)) city_fill, zip_fill, tract_fill = (_fill(g, RW_CITY), _fill(g, RW_ZIP), _fill(g, RW_TRACT)) summary_rows.append({ "County": name, "FIPS": fips, # true totals from the full frame "Total Records in County": len(gfull), # how many of those records carry a LegalSubDivision (non-unique) "Total Subdivisions": int(gfull["LegalSubDivision"].count()) if "LegalSubDivision" in gfull.columns else 0, "Total Clusters Made": int(g["Cluster_ID"].nunique()) if "Cluster_ID" in g.columns else 0, "Total Blank Cities": int(city_blank.sum()), "Total Cities Assignment": int((city_blank & city_fill).sum()), "Total Blank Zips": int(zip_blank.sum()), "Total Zips Assignment": int((zip_blank & zip_fill).sum()), "Total Blank Census": int(tract_blank.sum()), "Total Census Assignment": int((tract_blank & tract_fill).sum()), }) summary_cols = ["County", "FIPS", "Total Records in County", "Total Subdivisions", "Total Clusters Made", "Total Blank Cities", "Total Cities Assignment", "Total Blank Zips", "Total Zips Assignment", "Total Blank Census", "Total Census Assignment"] summary_df = pd.DataFrame(summary_rows, columns=summary_cols) log.append(f"Summary: {len(summary_df)} county row(s).") summary_path = out_dir / "Summary.csv" summary_df.to_csv(summary_path, index=False) outputs.append(str(summary_path)) # ---- Master bundle: Excel workbook OR zip of CSVs (see MASTER_OUTPUT) - if MASTER_OUTPUT == "csv": # Bundle the already-written CSVs into one zip. No Excel writing -> fast. master_path = out_dir / "Master_Dataset.zip" member_names = ["FIPS_Validation_Audit_ChangedRecords.csv", "cluster_df.csv", "final_cities.csv", "final_zips.csv", "final_census_tracts.csv", "Summary.csv"] members = [out_dir / n for n in member_names if (out_dir / n).exists()] members += sorted(out_dir.glob("OutputFile_*.csv")) # full output (chunked) with zipfile.ZipFile(master_path, "w", zipfile.ZIP_DEFLATED) as zf: for p in members: zf.write(p, arcname=p.name) else: # Every result on its own sheet; the full output splits across # Output_70_95_RapidFuzz_1, _2, ... if it exceeds Excel's row cap. master_path = out_dir / "Master_Dataset.xlsx" write_master_workbook(master_path, [ ("Output_70_95_RapidFuzz", output_df), ("ValidationAuditResults", audit), ("cluster_df", cluster_df), ("final_cities", final_cities), ("final_zips", final_zips), ("final_census_tracts", final_ct), ("Summary", summary_df), ]) outputs.append(str(master_path)) elapsed_min = (time.time() - t_start) / 60 log.append(f"Done. {len(outputs)} files ready.") log.append(f"Total processing time: {elapsed_min:.1f} minutes.") gc.collect() return "\n".join(log), outputs # ===================================================================== # STARTUP HEALTH CHECK # ===================================================================== def _startup_health_check(): if os.environ.get("SELF_TEST_ON_STARTUP", "1") != "1": return try: print("[startup] " + run_self_test()) except FileNotFoundError as e: # Reference file not configured yet — let the UI load so the user can fix it. print(f"[startup] Self-test skipped — safety list unavailable:\n{e}") # AssertionError intentionally propagates: a failed self-test fails the boot. _startup_health_check() # ===================================================================== # GRADIO UI (simple vertical layout — previous design) # ===================================================================== with gr.Blocks() as demo: gr.Markdown( "## Subdivision Cluster + FIPS Validation\n" "Upload the Excel files to process. Records are validated against the " "Subdivision Program Safety List, then `LegalSubDivision` values are " "fuzzy-clustered and city/ZIP/census-tract are filled by majority vote.\n\n" "*Runs with the notebook's fixed settings.*" ) files_in = gr.File(file_count="multiple", file_types=[".xlsx", ".xls", ".csv"], label="Files to process (XLSX, XLS, or CSV)") run_btn = gr.Button("Run pipeline", variant="primary") status = gr.Textbox(label="Log", lines=14) files_out = gr.File(label="Output files", file_count="multiple") def _run_simple(files): # No reference upload in this layout -> always the server-side list. # All settings are fixed (see constants at top). No user-facing options. return run_pipeline(files, None, STRICT_BLANK_CITIES, FILTER_DOMINANT_COUNTY, CLUSTER_THRESHOLD, REPLACE_THRESHOLD) run_btn.click( _run_simple, inputs=[files_in], outputs=[status, files_out], ) if __name__ == "__main__": demo.launch()