Upload 2 files

app.py

@@ -58,6 +58,17 @@ FILTER_DOMINANT_COUNTY = False  # process ALL counties (multi-county / full-stat
 # 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."""
@@ -690,6 +701,9 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
     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))
@@ -944,8 +958,7 @@ def run_from_dataset(input_repo, output_repo):
         except Exception as e:
             return f"Failed to download '{fn}' from '{input_repo}': {e}", []
 
-    log, outputs = run_pipeline(local, None, STRICT_BLANK_CITIES, FILTER_DOMINANT_COUNTY,
-                                CLUSTER_THRESHOLD, REPLACE_THRESHOLD)
+    log, outputs = process(local, None)
 
     if output_repo and outputs:
         try:
@@ -962,6 +975,419 @@ def run_from_dataset(input_repo, output_repo):
     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
 # =====================================================================
@@ -1032,8 +1458,7 @@ with gr.Blocks() as demo:
 
     def _run(files, reference):
         # All settings are fixed (see constants at top). No user-facing options.
-        return run_pipeline(files, reference, STRICT_BLANK_CITIES, FILTER_DOMINANT_COUNTY,
-                            CLUSTER_THRESHOLD, REPLACE_THRESHOLD)
+        return process(files, reference)
 
     run_btn.click(
         _run,