"""County-wide parcel ingest — replaces the Westerville-only parcels + parcel_geoms. Pulls every parcel from DelawareCountyData/MapServer/0 with rich fields (valuation, tax, year built, bd/hb/fb, sqft, subdivision, owner, address, township/muni) plus polygon geometry. Writes two tables: - parcel_geoms : polygon geometry + basic attrs (owner, addr, class) - parcels : centroid POINT + rich attrs (same schema as before, plus n_voters / n_r / n_d / dom_party / dom_lean columns kept at 0/'' — build_home_lean.py refills them). """ from __future__ import annotations import concurrent.futures as cf from pathlib import Path import geopandas as gpd import pandas as pd import requests from shapely.geometry import shape from sqlalchemy import create_engine, text DATA_DIR = Path("/home/m3ac/genoa-entwuerfe.com") LAYER = "https://maps.delco-gis.org/arcgiswebadaptor/rest/services/DelawareCountyData/MapServer/0" PAGE = 1000 FIELDS = ",".join([ "PARCEL_NO", "OWNER1", "ADDR1", "CLASS", "SUB_NAME", "ACRES", "MARKET_TOT", "MARKET_LAN", "MARKET_IMP", "TAXABLE_TO", "ANNUALTAX", "YRBUILT", "BDROOMS", "HAFBATHS", "FULBATHS", "SQFT", "SALE_AMNT", "TWP_NAME", "MUN_NAME", ]) 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 _fetch_page(offset: int) -> list[dict]: r = requests.get( f"{LAYER}/query", params={ "where": "1=1", "outFields": FIELDS, "returnGeometry": "true", "outSR": "4326", "f": "geojson", "resultOffset": offset, "resultRecordCount": PAGE, "geometryPrecision": 7, "orderByFields": "PARCEL_NO", }, timeout=90, ) r.raise_for_status() return r.json().get("features", []) def _clean_str(v) -> str: return (v or "").strip() if isinstance(v, str) else "" def main() -> None: count = requests.get( f"{LAYER}/query", params={"where": "1=1", "returnCountOnly": "true", "f": "json"}, timeout=30, ).json()["count"] print(f"[arcgis] {count} parcels to fetch in pages of {PAGE}") offsets = list(range(0, count, PAGE)) all_features: list[dict] = [] with cf.ThreadPoolExecutor(max_workers=8) as ex: for i, batch in enumerate(ex.map(_fetch_page, offsets)): all_features.extend(batch) if (i + 1) % 10 == 0 or i + 1 == len(offsets): print(f" page {i+1}/{len(offsets)} — total {len(all_features)}") print(f"[fetch] got {len(all_features)} features") geom_rows = [] rich_rows = [] bad = 0 seen_ids: set[str] = set() for f in all_features: try: g = shape(f["geometry"]) if not g.is_valid: g = g.buffer(0) except Exception: bad += 1 continue p = f.get("properties", {}) or {} parcel_no = _clean_str(p.get("PARCEL_NO")) if not parcel_no: continue owner = _clean_str(p.get("OWNER1")) addr = _clean_str(p.get("ADDR1")) cls = _clean_str(p.get("CLASS")) sub = _clean_str(p.get("SUB_NAME")) geom_rows.append({ "parcel_no": parcel_no, "owner": owner, "addr": addr, "class": cls, "geometry": g, }) # One rich row per unique parcel_no (some parcels split into multiple # polygons; we keep attrs of the first, centroid of the union). if parcel_no in seen_ids: continue seen_ids.add(parcel_no) c = g.centroid rich_rows.append({ "id": parcel_no, "own": owner, "adr": addr, "cls": cls, "sub": sub, "mkt": int(p.get("MARKET_TOT") or 0) or None, "lan": int(p.get("MARKET_LAN") or 0) or None, "imp": int(p.get("MARKET_IMP") or 0) or None, "txb": int(p.get("TAXABLE_TO") or 0) or None, "tax": float(p.get("ANNUALTAX") or 0) or None, "yr": int(p.get("YRBUILT") or 0) or None, "ac": float(p.get("ACRES") or 0) or None, "bd": int(p.get("BDROOMS") or 0) or None, "fb": int(p.get("FULBATHS") or 0) or None, "hb": int(p.get("HAFBATHS") or 0) or None, "sqft": int(p.get("SQFT") or 0) or None, "sale_amnt": int(p.get("SALE_AMNT") or 0) or None, "twp": _clean_str(p.get("TWP_NAME")), "muni": _clean_str(p.get("MUN_NAME")), "lat": c.y, "lng": c.x, "geometry": c, }) if bad: print(f"[warn] {bad} features had invalid geometry (dropped)") engine = create_engine( f"postgresql+psycopg2://gisapp:{_read_gispass()}@127.0.0.1:5432/westerville" ) # parcels_full mat view depends on both parcel_geoms and parcels; drop it # so pandas' to_postgis(if_exists="replace") can DROP+CREATE the tables. # It gets rebuilt at the end. with engine.begin() as conn: conn.execute(text("DROP MATERIALIZED VIEW IF EXISTS parcels_full;")) # ---- parcel_geoms (polygons) ---- ggdf = gpd.GeoDataFrame(geom_rows, crs="EPSG:4326") ggdf.to_postgis("parcel_geoms", engine, if_exists="replace", index=False) # ---- parcels (centroids + rich attrs) ---- pgdf = gpd.GeoDataFrame(rich_rows, crs="EPSG:4326") pgdf["parcel_pk"] = range(1, len(pgdf) + 1) pgdf["n_voters"] = 0 pgdf["n_r"] = 0 pgdf["n_d"] = 0 pgdf["n_u"] = 0 pgdf["n_super"] = 0 pgdf["dom_party"] = "" pgdf["dom_lean"] = "" pgdf.to_postgis("parcels", engine, if_exists="replace", index=False) with engine.begin() as conn: conn.execute(text("CREATE INDEX parcel_geoms_geom_idx ON parcel_geoms USING GIST (geometry);")) conn.execute(text("CREATE INDEX parcel_geoms_pn_idx ON parcel_geoms (parcel_no);")) conn.execute(text("ALTER TABLE parcels ADD PRIMARY KEY (parcel_pk);")) conn.execute(text("CREATE INDEX parcels_geom_idx ON parcels USING GIST (geometry);")) conn.execute(text("CREATE INDEX parcels_id_idx ON parcels (id);")) conn.execute(text('CREATE INDEX parcels_addr_idx ON parcels (LOWER("adr"));')) (ng,) = conn.execute(text("SELECT count(*) FROM parcel_geoms")).one() (np_,) = conn.execute(text("SELECT count(*) FROM parcels")).one() print(f"[postgis] parcel_geoms={ng} parcels={np_}") # Recreate parcels_full materialized view (definition from prior deploy). print("[postgis] rebuilding parcels_full mat view …") with engine.begin() as conn: conn.execute(text(""" CREATE MATERIALIZED VIEW parcels_full AS SELECT pg.parcel_no, COALESCE(p.own, pg.owner) AS owner, COALESCE(p.adr, pg.addr) AS addr, COALESCE(p.cls, pg.class) AS cls, p.mkt, p.lan, p.imp, p.tax, p.yr, p.ac, p.bd, p.sub, COALESCE(p.n_voters, 0) AS n_voters, COALESCE(p.n_r, 0) AS n_r, COALESCE(p.n_d, 0) AS n_d, COALESCE(p.n_u, 0) AS n_u, COALESCE(p.n_super, 0) AS n_super, COALESCE(p.dom_party, '') AS dom_party, COALESCE(p.dom_lean, '') AS dom_lean, (p.parcel_pk IS NOT NULL) AS has_attrs, pg.geometry FROM parcel_geoms pg LEFT JOIN parcels p ON pg.parcel_no = p.id; """)) conn.execute(text( "CREATE INDEX parcels_full_geom_gix ON parcels_full USING GIST (geometry);" )) conn.execute(text("CREATE INDEX parcels_full_cls_idx ON parcels_full (cls);")) conn.execute(text("CREATE INDEX parcels_full_party_idx ON parcels_full (dom_party);")) # parcel_no can appear on multiple polygon rows (split parcels), so this # is a plain index — REFRESH MATERIALIZED VIEW must not be CONCURRENTLY. conn.execute(text( "CREATE INDEX parcels_full_pn_idx ON parcels_full (parcel_no);" )) (npf,) = conn.execute(text("SELECT count(*) FROM parcels_full")).one() print(f"[postgis] parcels_full rebuilt: {npf} rows") if __name__ == "__main__": main()