Spaces:
Sleeping
Sleeping
| """ | |
| 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 <name>_1, <name>_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"<style>{CSS}</style>") | |
| with gr.Row(): | |
| # -------- LEFT SIDEBAR: uploads --------------------------------- | |
| with gr.Column(scale=1, elem_id="sidebar", min_width=270): | |
| gr.Markdown("### Upload Files") | |
| gr.HTML('<div class="step-label">1. Upload Files</div>') | |
| 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('<div class="step-label">2. Upload Reference File (Optional)</div>') | |
| 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('<div class="step-label">Large files: process from a HF Dataset ' | |
| '(no upload — avoids the upload timeout)</div>') | |
| 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('<div id="main-title">Subdivision Cluster & FIPS Validation</div>') | |
| gr.HTML( | |
| '<div class="info-banner">Upload your file(s) on the sidebar and click ' | |
| '<b>Run pipeline</b>. For files too large to upload (multi-GB / full-state), ' | |
| 'put them in a private HF Dataset and use <b>Run from dataset</b> instead — ' | |
| 'the results are written back to your output dataset.</div>' | |
| ) | |
| 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) | |