"""County-wide voter ingest — replaces the Westerville-only voters table. Source: /home/m3ac/private_voter/DELAWARE.txt (Ohio SoS quarterly export, ~167k rows). Strategy (hybrid geocoding, per user preference): 1) Parse the tab/CSV file, filter to "Active" voter status. 2) Normalize each residential address; join against parcels.adr for lat/lng. 3) For the remainder, submit to Census batch geocoder in chunks of 10000. 4) Skip any voter still lacking coords. Derived fields (matching existing schema): fn, ln first / last name p PARTY_AFFILIATION (R/D/L/G/N/etc. or blank) pl derived primary lean (R/D/'') from last 4 primary votes gv general-election count in the last 4 generals gh 4-bit string of last-4 general vote history ph 4-bit string of last-4 primary vote history a residential address (upper-cased, single-spaced) yr birth year (str, from DATE_OF_BIRTH) reg registration year (str) pct PRECINCT_NAME lat, lng, geometry from parcel match or Census geocode """ from __future__ import annotations import csv import io import re import time from pathlib import Path import geopandas as gpd import pandas as pd import requests from shapely.geometry import Point from sqlalchemy import create_engine, text DATA_DIR = Path("/home/m3ac/genoa-entwuerfe.com") VOTER_FILE = Path("/home/m3ac/private_voter/DELAWARE.txt") # Recent generals + primaries, newest → oldest. GEN_COLS = [ "GENERAL-11/04/2025", "GENERAL-11/05/2024", "GENERAL-11/07/2023", "GENERAL-11/08/2022", ] PRI_COLS = [ "PRIMARY-05/06/2025", "PRIMARY-03/19/2024", "PRIMARY-05/02/2023", "PRIMARY-05/03/2022", ] ACTIVE_STATUSES = {"ACTIVE", "CONFIRMATION"} CENSUS_BATCH_URL = "https://geocoding.geo.census.gov/geocoder/locations/addressbatch" CENSUS_BENCHMARK = "Public_AR_Current" CENSUS_CHUNK = 5000 # smaller than Census max (10k) to avoid timeouts _STREET_SUFFIX = { "STREET": "ST", "AVENUE": "AVE", "AVE.": "AVE", "ROAD": "RD", "DRIVE": "DR", "COURT": "CT", "CIRCLE": "CIR", "BOULEVARD": "BLVD", "LANE": "LN", "PLACE": "PL", "PARKWAY": "PKWY", "TERRACE": "TER", "TRAIL": "TRL", "HIGHWAY": "HWY", "POINT": "PT", "SQUARE": "SQ", "CROSSING": "XING", } def _read_gispass() -> str: for line in (DATA_DIR / ".env.gis").read_text().splitlines(): if line.startswith("GISPASS="): return line.split("=", 1)[1].strip() raise SystemExit("GISPASS missing") def normalize_addr(s: str) -> str: if not s: return "" s = s.upper().strip() # collapse multiple spaces, drop punctuation except '/' and '-' s = re.sub(r"[^\w/\-\s]", " ", s) s = re.sub(r"\s+", " ", s).strip() tokens = s.split(" ") tokens = [_STREET_SUFFIX.get(t, t) for t in tokens] return " ".join(tokens) def year_of(datestr: str) -> str: if not datestr: return "" m = re.match(r"(\d{4})", datestr) return m.group(1) if m else "" def bits(row: dict, cols: list[str]) -> str: """"X" (or "V"/"N"/"A"…) = voted; blank = did not.""" return "".join("1" if (row.get(c) or "").strip() else "0" for c in cols) def lean_from_primaries(row: dict) -> str: """Count R/D primary participation across recent primaries. Return dominant if 2+.""" rc = dc = 0 for c in PRI_COLS: v = (row.get(c) or "").strip().upper() if v == "R": rc += 1 elif v == "D": dc += 1 if rc >= 2 and rc > dc: return "R" if dc >= 2 and dc > rc: return "D" return "" def load_voters_csv(path: Path) -> list[dict]: """Stream-parse the SoS voter export.""" print(f"[read] {path}") voters = [] with path.open("r", encoding="utf-8", errors="replace") as fh: reader = csv.DictReader(fh) for i, row in enumerate(reader): status = (row.get("VOTER_STATUS") or "").strip().upper() if status not in ACTIVE_STATUSES: continue addr_raw = (row.get("RESIDENTIAL_ADDRESS1") or "").strip() if not addr_raw: continue voters.append({ "fn": (row.get("FIRST_NAME") or "").strip().title(), "ln": (row.get("LAST_NAME") or "").strip().title(), "p": ((row.get("PARTY_AFFILIATION") or "").strip().upper()[:1] or "U"), "pl": lean_from_primaries(row), "gv": sum(1 for c in GEN_COLS if (row.get(c) or "").strip()), "gh": bits(row, GEN_COLS), "ph": bits(row, PRI_COLS), "a": normalize_addr(addr_raw), "yr": year_of(row.get("DATE_OF_BIRTH") or ""), "reg": year_of(row.get("REGISTRATION_DATE") or ""), "pct": (row.get("PRECINCT_NAME") or "").strip(), "city": (row.get("RESIDENTIAL_CITY") or "").strip(), "zip": (row.get("RESIDENTIAL_ZIP") or "").strip()[:5], }) if (i + 1) % 25000 == 0: print(f" scanned {i+1} rows, kept {len(voters)}") print(f"[read] active voters: {len(voters)}") return voters def match_via_parcels(voters: list[dict], engine) -> tuple[list[dict], list[dict]]: """Join voters to parcels.adr on normalized address. Returns (matched, unmatched).""" with engine.begin() as conn: rows = conn.execute(text('SELECT "adr", lat, lng FROM parcels WHERE lat IS NOT NULL')).all() addr_to_coord: dict[str, tuple[float, float]] = {} for adr, lat, lng in rows: key = normalize_addr(adr) if key and key not in addr_to_coord: addr_to_coord[key] = (lat, lng) print(f"[match] {len(addr_to_coord)} unique parcel addresses indexed") matched, unmatched = [], [] for v in voters: c = addr_to_coord.get(v["a"]) if c: v["lat"], v["lng"] = c matched.append(v) else: unmatched.append(v) print(f"[match] {len(matched)}/{len(voters)} matched by parcel address " f"({len(matched)/len(voters):.1%})") return matched, unmatched def geocode_batch(voters: list[dict]) -> None: """Fill lat/lng on voters in place via Census batch geocoder (10k rows/req).""" if not voters: return print(f"[geocode] Census batch: {len(voters)} addresses in chunks of {CENSUS_CHUNK}") for start in range(0, len(voters), CENSUS_CHUNK): chunk = voters[start:start + CENSUS_CHUNK] buf = io.StringIO() w = csv.writer(buf) for i, v in enumerate(chunk): w.writerow([i, v["a"], v.get("city") or "", "OH", v.get("zip") or ""]) payload = buf.getvalue().encode() for attempt in range(3): try: r = requests.post( CENSUS_BATCH_URL, files={"addressFile": ("addrs.csv", payload, "text/csv")}, data={"benchmark": CENSUS_BENCHMARK}, timeout=600, ) r.raise_for_status() break except Exception as e: print(f" attempt {attempt+1} failed: {e}; retrying in 30s") time.sleep(30) else: print(f" chunk {start} — all attempts failed, skipping") continue # Response format: id, input, match/no_match, match_type, matched_addr, coords, ... n_hit = 0 reader = csv.reader(io.StringIO(r.text)) for row in reader: if len(row) < 6: continue try: idx = int(row[0]) except ValueError: continue if row[2] != "Match": continue coord = row[5] if not coord or "," not in coord: continue lng_s, lat_s = coord.split(",", 1) try: chunk[idx]["lat"] = float(lat_s) chunk[idx]["lng"] = float(lng_s) n_hit += 1 except (ValueError, IndexError): continue print(f" chunk {start}: geocoded {n_hit}/{len(chunk)}") def main() -> None: engine = create_engine( f"postgresql+psycopg2://gisapp:{_read_gispass()}@127.0.0.1:5432/westerville" ) voters = load_voters_csv(VOTER_FILE) matched, unmatched = match_via_parcels(voters, engine) geocode_batch(unmatched) all_voters = matched + [v for v in unmatched if "lat" in v and "lng" in v] print(f"[final] {len(all_voters)}/{len(voters)} voters have coords " f"({len(all_voters)/len(voters):.1%})") df = pd.DataFrame(all_voters) # keep only the columns the existing table expects keep_cols = ["fn", "ln", "p", "pl", "gv", "gh", "ph", "lat", "lng", "a", "yr", "reg", "pct"] for c in keep_cols: if c not in df.columns: df[c] = None df = df[keep_cols].reset_index(drop=True) df["voter_id"] = range(1, len(df) + 1) gdf = gpd.GeoDataFrame( df, geometry=[Point(xy) for xy in zip(df["lng"], df["lat"])], crs="EPSG:4326", ) gdf.to_postgis("voters", engine, if_exists="replace", index=False) with engine.begin() as conn: conn.execute(text("ALTER TABLE voters ADD PRIMARY KEY (voter_id);")) conn.execute(text("CREATE INDEX voters_geom_idx ON voters USING GIST (geometry);")) conn.execute(text('CREATE INDEX voters_addr_idx ON voters (LOWER("a"));')) (n,) = conn.execute(text("SELECT count(*) FROM voters")).one() print(f"[postgis] voters loaded: {n}") if __name__ == "__main__": main()