`pull.py`

67 lines
#!/usr/bin/env python3
"""Pull all 2-4 unit residential-income parcels in LA County (2025 roll) from the
public LA County Assessor FeatureServer. Splits into duplex/triplex/fourplex lists
plus an investor-owned subset (no homeowners exemption). Free, keyless, read-only."""
import csv, json, time, urllib.parse, urllib.request, os

BASE = ("https://services.arcgis.com/RmCCgQtiZLDCtblq/arcgis/rest/services/"
        "Parcel_Data_2021_Table/FeatureServer/0/query")
WHERE = ("RollYear='2025' AND (UseCode LIKE '020%' OR UseCode LIKE '030%' "
         "OR UseCode LIKE '040%')")
FIELDS = ("AIN,AssessorID,PropertyLocation,SitusHouseNo,SitusDirection,SitusStreet,"
          "SitusUnit,SitusCity,SitusZIP5,TaxRateArea_CITY,UseCode,UseCodeDescChar2,"
          "Units,YearBuilt,SQFTmain,Bedrooms,Bathrooms,Roll_LandValue,Roll_ImpValue,"
          "Roll_TotalValue,Roll_HomeOwnersExemp,RecordingDate,CENTER_LAT,CENTER_LON")
OUT = os.path.join(os.path.dirname(__file__), "lists")

def fetch(offset):
    q = {"where": WHERE, "outFields": FIELDS, "returnGeometry": "false",
         "resultRecordCount": 2000, "resultOffset": offset, "f": "json"}
    url = BASE + "?" + urllib.parse.urlencode(q)
    with urllib.request.urlopen(url, timeout=60) as r:
        return json.load(r)

rows, offset = [], 0
while True:
    data = fetch(offset)
    feats = data.get("features", [])
    if not feats:
        break
    rows += [f["attributes"] for f in feats]
    print(f"  fetched {len(rows)} ...", flush=True)
    if len(feats) < 2000:
        break
    offset += 2000
    time.sleep(0.4)

cols = FIELDS.split(",")
def tier(uc):
    if uc.startswith("02"): return "2-unit"
    if uc.startswith("03"): return "3-unit"
    return "4-unit"

buckets = {"2-unit": [], "3-unit": [], "4-unit": []}
for r in rows:
    buckets[tier(r.get("UseCode", ""))].append(r)

def write(path, recs):
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
        w.writeheader()
        for r in recs:
            w.writerow(r)

os.makedirs(OUT, exist_ok=True)
total = len(rows)
for k, recs in buckets.items():
    write(os.path.join(OUT, f"la-county-{k}.csv"), recs)
# investor-owned subset (no homeowners exemption claimed)
inv = [r for r in rows if not r.get("Roll_HomeOwnersExemp")]
write(os.path.join(OUT, "la-county-2to4-investor-owned.csv"), inv)
write(os.path.join(OUT, "la-county-2to4-all.csv"), rows)

print(f"\nDONE total={total}")
for k in ("2-unit", "3-unit", "4-unit"):
    print(f"  {k}: {len(buckets[k])}")
print(f"  investor-owned (no HOX): {len(inv)}")