Upload 2 files

app.py

@@ -56,7 +56,30 @@ FILTER_DOMINANT_COUNTY = False  # process ALL counties (multi-county / full-stat
 #   "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 = "csv"
+MASTER_OUTPUT = "excel"
+
+
+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:
@@ -141,7 +164,7 @@ def _norm_fips(x):
 # 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", "DocumentRecordingCountyFIPS"]
+FALLBACK_FIPS_HEADERS = ["** REC FIPS-COUNTY NAME", "** REC FIPS-COUNTY", "County"]
 
 
 def detect_fips_county_column(df, sample_size=1000, min_score=0.5):
@@ -562,16 +585,18 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
         name = f.name
         try:
             if str(name).lower().endswith(".csv"):
-                dfs.append(pd.read_csv(name))
+                dfs.append(pd.read_csv(name, low_memory=False))
             else:
                 dfs.append(pd.read_excel(name))
-            log.append(f"Loaded {Path(name).name}")
+            _step(log, f"Loaded {Path(name).name}  ({len(dfs[-1]):,} rows)")
         except Exception as e:
-            log.append(f"ERROR loading {Path(name).name}: {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)
-    log.append(f"Merged shape: {df.shape}")
+    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)
@@ -579,13 +604,15 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
         return ("\n".join(log) +
                 "\n\nCOLUMN ERROR: could not find a FIPS-county column. Expected "
                 "values like '05045-FAULKNER' in one of the columns."), []
-    log.append(f"FIPS-county column: '{fips_col}' ({fips_reason})")
+    _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:
@@ -596,6 +623,7 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
     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",
@@ -623,12 +651,15 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
     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)
-    log.append(f"Clusters: {cluster_df['Cluster_ID'].nunique():,}")
+    _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"})
@@ -638,8 +669,10 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
     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"])
@@ -673,7 +706,10 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
     other = [c for c in existing if c not in lead]
     original_final = original_final[lead + other]
 
-    original_final_output = original_final.copy()   # full output snapshot
+    # 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",
@@ -735,12 +771,15 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
         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
@@ -807,13 +846,14 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
                     "Total Blank Zips", "Total Zips Assignment", "Total Blank Census",
                     "Total Census Assignment"]
     summary_df = pd.DataFrame(summary_rows, columns=summary_cols)
-    log.append(f"Summary: {len(summary_df)} county row(s).")
+    _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"
@@ -842,6 +882,7 @@ def run_pipeline(uploaded_files, reference_file, blank_cities_not_in_list,
 
     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

requirements.txt

@@ -1,5 +1,7 @@
 pandas>=2.0
 numpy>=1.24
 openpyxl>=3.1
+xlsxwriter>=3.1
 rapidfuzz>=3.0
+psutil>=5.9
 huggingface_hub>=0.23