""" 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" # How records are processed: # "stream" -> ONE COUNTY AT A TIME. The input is split into per-county temp # files (chunked read, never holding the whole state), each county # is processed and its results appended. Peak RAM ~ the largest # single county, so full states run on a small/cheap tier. # "memory" -> the whole dataset is loaded and processed at once (faster on # small files, but needs enough RAM to hold everything). # Results are identical either way. Use "stream" for state-sized files. PROCESS_MODE = "stream" STREAM_CHUNK_ROWS = 400_000 # rows per chunk while splitting the input by county def _mem_mb(): """Current process memory (RSS) in MB. psutil if available, else peak RSS.""" try: import psutil return psutil.Process().memory_info().rss / (1024 * 1024) except Exception: try: import resource return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 # KB -> MB except Exception: return float("nan") def _step(log, msg): """Record a phase line with a timestamp + current RAM, and PRINT it live (flushed) so it reaches the Space's container log immediately. If the process is killed (e.g. out of memory), the last printed line pinpoints the phase.""" line = f"[{time.strftime('%H:%M:%S')}] {msg} | RAM {_mem_mb():.0f} MB" print(line, flush=True) log.append(line) return line 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, low_memory=False)) else: dfs.append(pd.read_excel(name)) _step(log, f"Loaded {Path(name).name} ({len(dfs[-1]):,} rows)") except Exception as e: _step(log, 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) del dfs gc.collect() _step(log, 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."), [] _step(log, f"FIPS-county column: '{fips_col}' ({fips_reason})") # ---- FIPS validation (uses the reference list) ------------------- _step(log, "Validating FIPS/city pairs...") 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}", [] _step(log, "Validation done") # ---- 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)) _step(log, "Wrote RawFile_FIPSValidated.csv") # ---- 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() _step(log, f"Source frames built (subdivisions: {len(original):,})") # ---- Cluster ------------------------------------------------------ _step(log, "Clustering LegalSubDivision per county...") cluster_df = build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold) _step(log, f"Clusters: {cluster_df['Cluster_ID'].nunique():,}") # ---- Replace-with (majority vote) -------------------------------- _step(log, "Building majority-vote tables...") 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"}) _step(log, "Majority-vote tables done") # ---- Map back onto original -------------------------------------- _step(log, "Mapping clusters + fills back onto records...") # 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"]))) # Clean integer Cluster_ID for output (no "2.0"); consistent with streaming. original_final["Cluster_ID"] = original_final["Cluster_ID"].astype("Int64") # 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 is only READ from here on (never mutated), so we can alias # instead of copying -- this avoids a second full copy of the 7M-row frame. original_final_output = original_final _step(log, "Full output frame ready") # ---- 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)) _step(log, "Wrote final_cities / final_zips / final_census_tracts / cluster_df") # ---- 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". _step(log, "Writing full output CSV...") 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)) _step(log, "Full output CSV written") # ---- 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) _step(log, 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) - _step(log, f"Building master ({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.") _step(log, f"DONE — {len(outputs)} files ready in {elapsed_min:.1f} min") log.append(f"Total processing time: {elapsed_min:.1f} minutes.") gc.collect() return "\n".join(log), outputs # ===================================================================== # DATASET I/O (for large inputs that time out on browser upload -> HTTP 408) # --------------------------------------------------------------------- # Instead of uploading a multi-GB file through the browser, put it in a private # HF *dataset* repo and let the Space read it over HF's internal network. Outputs # are written back to a dataset the same way (large downloads also 408 otherwise). # Configure as Space Variables/Secrets: # INPUT_DATASET_REPO e.g. "DDPLLeads/subdivision-inputs" # OUTPUT_DATASET_REPO e.g. "DDPLLeads/subdivision-outputs" (optional) # HF_TOKEN a token with read on input + write on output # ===================================================================== INPUT_DATASET_REPO = os.environ.get("INPUT_DATASET_REPO", "") OUTPUT_DATASET_REPO = os.environ.get("OUTPUT_DATASET_REPO", "") class _LocalFile: """Adapter so downloaded dataset files look like uploaded files to run_pipeline.""" def __init__(self, path): self.name = str(path) def list_dataset_inputs(repo_id, token=None): from huggingface_hub import list_repo_files token = token or os.environ.get("HF_TOKEN") files = list_repo_files(repo_id, repo_type="dataset", token=token) return sorted(f for f in files if f.lower().endswith((".csv", ".xlsx", ".xls"))) def run_from_dataset(input_repo, output_repo): """Download every supported file from input_repo, run the pipeline, and (if output_repo is set) upload the results back to that dataset. Returns (log_text, output_paths) just like run_pipeline, so it wires to the same UI.""" token = os.environ.get("HF_TOKEN") input_repo = (input_repo or "").strip() output_repo = (output_repo or "").strip() if not input_repo: return "Set an input dataset repo (or the INPUT_DATASET_REPO variable).", [] if not token: return "An HF_TOKEN secret is required to read/write datasets.", [] from huggingface_hub import hf_hub_download, HfApi try: filenames = list_dataset_inputs(input_repo, token) except Exception as e: return f"Could not list dataset '{input_repo}': {e}", [] if not filenames: return f"No .csv/.xlsx/.xls files found in dataset '{input_repo}'.", [] local = [] for fn in filenames: try: local.append(_LocalFile( hf_hub_download(repo_id=input_repo, filename=fn, repo_type="dataset", token=token))) except Exception as e: return f"Failed to download '{fn}' from '{input_repo}': {e}", [] log, outputs = process(local, None) if output_repo and outputs: try: api = HfApi(token=token) stamp = time.strftime("%Y%m%d_%H%M%S") for p in outputs: api.upload_file(path_or_fileobj=p, path_in_repo=f"runs/{stamp}/{Path(p).name}", repo_id=output_repo, repo_type="dataset") log += (f"\nUploaded {len(outputs)} file(s) to dataset '{output_repo}' " f"under runs/{stamp}/. Retrieve them there:\n" f"https://huggingface.co/datasets/{output_repo}/tree/main/runs/{stamp}") except Exception as e: log += f"\nWARNING: processed OK but upload to '{output_repo}' failed: {e}" return log, outputs # ===================================================================== # PER-COUNTY STREAMING (bounded memory for state-sized files) # ===================================================================== _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", ] def process_county(cdf, wl, fips_col, cluster_offset, blank_cities_not_in_list, cluster_threshold, replace_threshold): """Full per-county transform on ONE county's raw rows. Mirrors run_pipeline exactly, scoped to a single county. Returns a dict of that county's outputs plus how many clusters it produced (to continue global Cluster_ID numbering). Cluster_IDs are offset by `cluster_offset` so, processing counties in sorted order, the numbering matches the whole-dataset ("memory") path.""" ignore = [] cdf, audit = fips_validate(cdf, blank_cities_not_in_list, ignore, wl=wl, fips_col=fips_col) for col in ["DDPPropertyCityFinal", "DDPPropertyZIPFinal", "DDPRevisedCensusTract", "LegalSubDivision"]: if col in cdf.columns: cdf[col] = cdf[col].replace(["UNDETERMINED CITY", "UNDETERMINED ZIP", "NULL"], np.nan) if fips_col != "County": if "County" in cdf.columns: cdf = cdf.drop(columns=["County"]) cdf = cdf.rename(columns={fips_col: "County"}) city_df = cdf.loc[cdf["DDPPropertyCityFinal"].notna()].copy() zip_df = cdf.loc[cdf["DDPPropertyZIPFinal"].notna()].copy() census_tracts_df = cdf.loc[cdf["DDPRevisedCensusTract"].notna()].copy() original = cdf.loc[cdf["LegalSubDivision"].notna()].copy() cluster_df = build_clusters(city_df, zip_df, census_tracts_df, cluster_threshold) if len(cluster_df): cluster_df["Cluster_ID"] = cluster_df["Cluster_ID"] + cluster_offset n_clusters = int(cluster_df["Cluster_ID"].nunique()) if len(cluster_df) else 0 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"}) 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"]))) original_final["Cluster_ID"] = original_final["Cluster_ID"].astype("Int64") original_final["Replace with Zip Codes"] = original_final["Replace with Zip Codes"].apply(clean_zip) 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] 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() 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"]).copy() final_zips = final_zips.drop(columns=["DDPPropertyZIPFinalToRemove"]).copy() final_ct = final_ct.drop(columns=["DDPRevisedCensusTractToRemove"]).copy() 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() for d in (final_cities, final_zips, final_ct): for bc in ("ComesFromMtg2", "IsFlagged"): if bc in d.columns: d[bc] = boolify(d[bc]) # ---- summary row for this county (totals from full county frame) ---- 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) label = cdf["County"].dropna().iloc[0] if ("County" in cdf.columns and cdf["County"].notna().any()) else "" raw = str(label) parts = raw.split("-") fips = parts[0].strip() name = parts[1].strip().title() if len(parts) > 1 else raw.strip().title() g = original_final city_blank, zip_blank, tract_blank = (_blank(g, "DDPPropertyCityFinal"), _blank(g, "DDPPropertyZIPFinal"), _blank(g, "DDPRevisedCensusTract")) city_fill, zip_fill, tract_fill = (_fill(g, "Replace with Cities"), _fill(g, "Replace with Zip Codes"), _fill(g, "Replace with Census Tracts")) summary_row = { "County": name, "FIPS": fips, "Total Records in County": len(cdf), "Total Subdivisions": int(cdf["LegalSubDivision"].count()) if "LegalSubDivision" in cdf.columns else 0, "Total Clusters Made": n_clusters, "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()), } return { "validated": cdf, "audit": audit, "cluster_df": cluster_df, "output_df": original_final, "final_cities": final_cities, "final_zips": final_zips, "final_ct": final_ct, "summary_row": summary_row, "n_clusters": n_clusters, } class _RollingCsv: """Append rows to CSV part files, rolling to a new part every row_limit rows.""" def __init__(self, out_dir, base, row_limit=1_000_000): self.out_dir, self.base, self.row_limit = out_dir, base, row_limit self.part, self.count, self.header = 0, 0, False self.paths = [] self._new() def _new(self): self.part += 1 self.count, self.header = 0, False self.path = self.out_dir / f"{self.base}_part_{self.part}.csv" self.paths.append(self.path) def write(self, dfx): i, n = 0, len(dfx) while i < n: room = self.row_limit - self.count take = dfx.iloc[i:i + room] take.to_csv(self.path, mode="a", header=not self.header, index=False) self.header = True self.count += len(take) i += len(take) if self.count >= self.row_limit and i < n: self._new() def finalize(self): # If everything fit in one part, drop the "_part_1" suffix to match memory mode. if self.part == 1 and self.paths[0].exists(): final = self.out_dir / f"{self.base}.csv" os.replace(self.paths[0], final) return [str(final)] return [str(p) for p in self.paths if p.exists()] def _split_by_county(uploaded_files, out_dir, log, chunksize=STREAM_CHUNK_ROWS): """Stream the input(s) in chunks and route rows to one temp CSV per county. Never holds more than one chunk in memory. Returns (fips_col, reason, [(county_label, path), ...] sorted by county).""" cdir = out_dir / "_by_county" cdir.mkdir(parents=True, exist_ok=True) # detect the FIPS-county column from a small sample of the first file first = uploaded_files[0].name if str(first).lower().endswith(".csv"): sample = pd.read_csv(first, nrows=50_000, low_memory=False) else: sample = pd.read_excel(first) fips_col, reason = detect_fips_county_column(sample) del sample if fips_col is None: return None, reason, [] county_paths = {} # county label -> temp path counter = [0] def path_for(county): if county not in county_paths: counter[0] += 1 county_paths[county] = cdir / f"county_{counter[0]:05d}.csv" return county_paths[county] def route(chunk): keys = chunk[fips_col].astype(str).str.strip() for county, idx in chunk.groupby(keys).groups.items(): p = path_for(county) chunk.loc[idx].to_csv(p, mode="a", header=not p.exists(), index=False) total = 0 for f in uploaded_files: nm = f.name if str(nm).lower().endswith(".csv"): for chunk in pd.read_csv(nm, chunksize=chunksize, low_memory=False): route(chunk) total += len(chunk) _step(log, f"Split: routed {total:,} rows into {len(county_paths)} counties") else: chunk = pd.read_excel(nm) route(chunk) total += len(chunk) _step(log, f"Split: routed {total:,} rows into {len(county_paths)} counties") ordered = sorted(county_paths.items(), key=lambda kv: str(kv[0])) return fips_col, reason, ordered def run_streaming(uploaded_files, reference_file, blank_cities_not_in_list, filter_dominant_county, cluster_threshold, replace_threshold): """Per-county streaming pipeline: bounded memory, identical results.""" t_start = time.time() log = [] if not uploaded_files: return "Please upload at least one file.", [] out_dir = Path(tempfile.mkdtemp(prefix="subdiv_")) outputs = [] try: wl = build_whitelist_from_path(reference_file.name) if reference_file is not None else load_whitelist() _step(log, "Loaded safety list" + (" (uploaded override)" if reference_file is not None else "")) 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}", [] _step(log, "Splitting input by county (streaming)...") fips_col, reason, counties = _split_by_county(uploaded_files, out_dir, log) if fips_col is None: return ("\n".join(log) + "\n\nCOLUMN ERROR: could not find a FIPS-county column. " "Expected values like '05045-FAULKNER'."), [] _step(log, f"FIPS-county column: '{fips_col}' ({reason}) | {len(counties)} counties") # combined output writers audit_path = out_dir / "FIPS_Validation_Audit_ChangedRecords.csv" raw_path = out_dir / "RawFile_FIPSValidated.csv" fc_path = out_dir / "final_cities.csv" fz_path = out_dir / "final_zips.csv" fct_path = out_dir / "final_census_tracts.csv" clu_path = out_dir / "cluster_df.csv" out_roll = _RollingCsv(out_dir, f"OutputFile_{replace_threshold}Threshold_{cluster_threshold}ClustersRapidFuzz", row_limit=1_000_000) def _append(dfx, path): if len(dfx) == 0 and Path(path).exists(): return dfx.to_csv(path, mode="a", header=not Path(path).exists(), index=False) summary_rows = [] cluster_offset = 0 any_audit = False for i, (label, cpath) in enumerate(counties, start=1): cdf = pd.read_csv(cpath, low_memory=False) res = process_county(cdf, wl, fips_col, cluster_offset, blank_cities_not_in_list, cluster_threshold, replace_threshold) cluster_offset += res["n_clusters"] _append(res["validated"], raw_path) if len(res["audit"]) > 0: _append(res["audit"], audit_path); any_audit = True _append(res["final_cities"], fc_path) _append(res["final_zips"], fz_path) _append(res["final_ct"], fct_path) if len(res["cluster_df"]): _append(res["cluster_df"], clu_path) out_roll.write(res["output_df"]) summary_rows.append(res["summary_row"]) del cdf, res gc.collect() _step(log, f"County {i}/{len(counties)} done ({label})") # finalize combined files for p in [audit_path, raw_path, fc_path, fz_path, fct_path, clu_path]: if p.exists(): outputs.append(str(p)) output_parts = out_roll.finalize() outputs.extend(output_parts) 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) summary_path = out_dir / "Summary.csv" summary_df.to_csv(summary_path, index=False) outputs.append(str(summary_path)) _step(log, f"Summary: {len(summary_df)} county row(s)") # ---- Master ------------------------------------------------------ _step(log, f"Building master ({MASTER_OUTPUT})...") if MASTER_OUTPUT == "csv": 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 += [Path(p) for p in output_parts] with zipfile.ZipFile(master_path, "w", zipfile.ZIP_DEFLATED) as zf: for p in members: zf.write(p, arcname=Path(p).name) else: # Stream the big Output sheet from the combined CSV; small sheets load fully. master_path = out_dir / "Master_Dataset.xlsx" _write_master_streaming(master_path, output_parts, audit_path, clu_path, fc_path, fz_path, fct_path, summary_df) outputs.append(str(master_path)) elapsed_min = (time.time() - t_start) / 60 log.append(f"Done. {len(outputs)} files ready.") _step(log, f"DONE — {len(outputs)} files ready in {elapsed_min:.1f} min") log.append(f"Total processing time: {elapsed_min:.1f} minutes.") gc.collect() return "\n".join(log), outputs def _write_master_streaming(path, output_parts, audit_path, clu_path, fc_path, fz_path, fct_path, summary_df, row_limit=1_048_575): """Build Master_Dataset.xlsx in low memory: stream the Output sheet from the combined output CSV(s) in chunks; load the (small) side sheets fully.""" try: import xlsxwriter except Exception: xlsxwriter = None def _read(p): return pd.read_csv(p, low_memory=False) if Path(p).exists() else pd.DataFrame() if xlsxwriter is None: # openpyxl fallback (higher memory): load Output fully. frames = [] for p in output_parts: frames.append(pd.read_csv(p, low_memory=False)) output_df = pd.concat(frames, ignore_index=True) if frames else pd.DataFrame() write_master_workbook(path, [ ("Output_70_95_RapidFuzz", output_df), ("ValidationAuditResults", _read(audit_path)), ("cluster_df", _read(clu_path)), ("final_cities", _read(fc_path)), ("final_zips", _read(fz_path)), ("final_census_tracts", _read(fct_path)), ("Summary", summary_df), ], row_limit=row_limit) return wb = xlsxwriter.Workbook(str(path), {"constant_memory": True, "nan_inf_to_errors": True}) def _write_df(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) for r, rowvals in enumerate(clean.itertuples(index=False, name=None), start=1): ws.write_row(r, 0, rowvals) # Output sheet(s): stream from the combined CSV, split at row_limit sheet_idx, row_in_sheet, ws, header = 0, 0, None, None for p in output_parts: for chunk in pd.read_csv(p, chunksize=200_000, low_memory=False): if header is None: header = [str(c) for c in chunk.columns] clean = chunk.where(pd.notna(chunk), None) for rowvals in clean.itertuples(index=False, name=None): if ws is None or row_in_sheet >= row_limit: sheet_idx += 1 ws = wb.add_worksheet(f"Output_70_95_RapidFuzz_{sheet_idx}"[:31]) ws.write_row(0, 0, header) row_in_sheet = 0 row_in_sheet += 1 ws.write_row(row_in_sheet, 0, rowvals) if ws is None: # no output rows at all wb.add_worksheet("Output_70_95_RapidFuzz").write_row(0, 0, header or ["(empty)"]) _write_df("ValidationAuditResults", _read(audit_path)) _write_df("cluster_df", _read(clu_path)) _write_df("final_cities", _read(fc_path)) _write_df("final_zips", _read(fz_path)) _write_df("final_census_tracts", _read(fct_path)) _write_df("Summary", summary_df) wb.close() def process(uploaded_files, reference_file): """Dispatcher used by the UI: streaming or in-memory per PROCESS_MODE.""" if PROCESS_MODE == "stream": return run_streaming(uploaded_files, reference_file, STRICT_BLANK_CITIES, FILTER_DOMINANT_COUNTY, CLUSTER_THRESHOLD, REPLACE_THRESHOLD) return run_pipeline(uploaded_files, reference_file, STRICT_BLANK_CITIES, FILTER_DOMINANT_COUNTY, CLUSTER_THRESHOLD, REPLACE_THRESHOLD) # ===================================================================== # 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 (sidebar "Upload Files" layout) # ===================================================================== CSS = """ #sidebar {background: var(--block-background-fill); border-right: 1px solid var(--border-color-primary); padding: 18px 16px; min-width: 270px;} #sidebar h3 {margin: 0 0 14px 0; font-size: 1.15rem; font-weight: 700;} .step-label {font-size: .82rem; color: var(--body-text-color-subdued); margin: 14px 0 6px 0;} #main-title {font-size: 2.2rem; font-weight: 800; margin: 6px 0 18px 0;} .info-banner {background: #e8f1fe; color: #1d4ed8; border-radius: 10px; padding: 16px 18px; font-size: 1rem; margin-bottom: 18px;} """ with gr.Blocks() as demo: gr.HTML(f"") with gr.Row(): # -------- LEFT SIDEBAR: uploads --------------------------------- with gr.Column(scale=1, elem_id="sidebar", min_width=270): gr.Markdown("### Upload Files") gr.HTML('
1. Upload Files
') files_in = gr.File(file_count="multiple", file_types=[".xlsx", ".xls", ".csv"], label="XLSX, XLS, or CSV — upload several and they're merged", show_label=True) gr.HTML('
2. Upload Reference File (Optional)
') reference_in = gr.File(file_count="single", file_types=[".xlsx", ".csv"], label="XLSX or CSV", show_label=True) run_btn = gr.Button("Run pipeline", variant="primary") gr.HTML('
Large files: process from a HF Dataset ' '(no upload — avoids the upload timeout)
') ds_in = gr.Textbox(value=INPUT_DATASET_REPO, label="Input dataset repo", placeholder="user/dataset-name") ds_out = gr.Textbox(value=OUTPUT_DATASET_REPO, label="Output dataset repo (optional)", placeholder="user/dataset-name") ds_btn = gr.Button("Run from dataset") # -------- MAIN AREA: title, info, results ----------------------- with gr.Column(scale=4): gr.HTML('
Subdivision Cluster & FIPS Validation
') gr.HTML( '
Upload your file(s) on the sidebar and click ' 'Run pipeline. For files too large to upload (multi-GB / full-state), ' 'put them in a private HF Dataset and use Run from dataset instead — ' 'the results are written back to your output dataset.
' ) status = gr.Textbox(label="Log", lines=14) files_out = gr.File(label="Output files", file_count="multiple") def _run(files, reference): # All settings are fixed (see constants at top). No user-facing options. return process(files, reference) run_btn.click( _run, inputs=[files_in, reference_in], outputs=[status, files_out], ) ds_btn.click( run_from_dataset, inputs=[ds_in, ds_out], outputs=[status, files_out], ) if __name__ == "__main__": # max_file_size=None -> no per-file upload cap imposed by the app itself. # (Very large uploads are still limited by HF's proxy/RAM -- prefer splitting # the input into several files, which this app merges, or run big jobs in Colab.) demo.launch(max_file_size=None)