scripts/dogis_export_probe.py

"""DOGIS 2025 imagery export probe — Exp 4 Phase 2, step 1.

Validates the Douglas County 2025 imagery MapServer as the training-crop source
BEFORE anything scales: fetches tile mosaics over known parcels at zoom levels
19-22, measures effective GSD / sharpness / detail gain per level (to find the
native resolution ceiling), greenness (leaf-off check), and JPEG quality.

Self-contained: no repo imports. Runs identically on the Windows dev box and
the owner's Mac. Everything the analysis needs is printed between the REPORT
BEGIN/END banners — copy that whole block back. Each stage catches and records
its own error, so partial failures still produce a report.

Setup on a fresh Mac (Terminal):

    python3 -m venv ~/lawn-train && source ~/lawn-train/bin/activate
    pip install requests pillow numpy
    python dogis_export_probe.py [--out ./dogis_probe_out]

Idempotent: chips already on disk are not re-fetched.
"""

from __future__ import annotations

import argparse
import io
import json
import math
import os
import platform
import time
import traceback

import numpy as np
import requests
from PIL import Image

SERVICE = "https://dcgis.org/server/rest/services/2025_Douglas_County_NE_Imagery/MapServer"
GEOCODER = "https://dcgis.org/server/rest/services/vector/Address_Points/FeatureServer/0/query"
TILE_SIZE = 256
LEVELS = [19, 20, 21, 22]
CHIP_M = 80.0  # ~80 m square around the address point — covers a residential parcel + ROW

# Known-difficult parcels from the accuracy experiments (all Douglas County).
# Fallback coords baked in so a geocoder outage doesn't kill the probe.
ADDRESSES = {
    "8571 Young St":      (41.332665, -96.046617),  # exp5 new-construction turf
    "14052 Hartman Ave":  (41.310482, -96.134419),  # canonical QA address
    "7617 Grover St":     (41.227234, -96.030871),  # shadowed turf strips (Phase 1 regression)
    "17531 Madison St":   (41.193864, -96.189410),  # street-edge band turf
    "1623 N 75th Ave":    (41.274847, -96.028857),  # deep setback / ROW-to-curb case
}

EARTH = 20037508.342787  # Web-Mercator half-circumference (m)


def geocode(session: requests.Session, address: str) -> tuple[float, float]:
    r = session.get(GEOCODER, params={
        "where": f"FULLADDR LIKE '{address.upper()}%'",
        "outFields": "FULLADDR", "returnGeometry": "true", "outSR": "4326", "f": "json",
    }, timeout=30)
    r.raise_for_status()
    feats = r.json().get("features") or []
    if not feats:
        raise ValueError(f"no address point for {address!r}")
    g = feats[0]["geometry"]
    return g["y"], g["x"]


def tile_xy(lat: float, lon: float, z: int) -> tuple[float, float]:
    """Fractional (col, row) at zoom z."""
    n = 2.0 ** z
    x = (lon + 180.0) / 360.0 * n
    siny = math.sin(math.radians(lat))
    y = (0.5 - math.log((1 + siny) / (1 - siny)) / (4 * math.pi)) * n
    return x, y


def meters_per_pixel(lat: float, z: int) -> float:
    return (2 * EARTH / (TILE_SIZE * 2.0 ** z)) * math.cos(math.radians(lat))


def fetch_tile(session: requests.Session, z: int, row: int, col: int,
               retries: int = 3) -> Image.Image:
    url = f"{SERVICE}/tile/{z}/{row}/{col}"
    last = None
    for attempt in range(retries):
        try:
            r = session.get(url, timeout=30)
            if r.status_code == 200 and r.headers.get("Content-Type", "").startswith("image"):
                return Image.open(io.BytesIO(r.content)).convert("RGB")
            last = f"HTTP {r.status_code} {r.headers.get('Content-Type')}"
        except requests.RequestException as e:  # transient gov-server 5xx / resets
            last = repr(e)
        time.sleep(1.5 * (attempt + 1))
    raise RuntimeError(f"tile {z}/{row}/{col}: {last}")


def fetch_chip(session: requests.Session, lat: float, lon: float, z: int,
               chip_m: float = CHIP_M) -> Image.Image:
    """Mosaic of cached tiles covering chip_m meters square centered on (lat, lon)."""
    mpp = meters_per_pixel(lat, z)
    half_px = chip_m / 2 / mpp
    cx, cy = tile_xy(lat, lon, z)
    px_c, py_c = cx * TILE_SIZE, cy * TILE_SIZE  # global pixel coords
    x0, y0 = int(px_c - half_px), int(py_c - half_px)
    x1, y1 = int(px_c + half_px), int(py_c + half_px)
    c0, c1 = x0 // TILE_SIZE, x1 // TILE_SIZE
    r0, r1 = y0 // TILE_SIZE, y1 // TILE_SIZE
    mosaic = Image.new("RGB", ((c1 - c0 + 1) * TILE_SIZE, (r1 - r0 + 1) * TILE_SIZE))
    for row in range(r0, r1 + 1):
        for col in range(c0, c1 + 1):
            mosaic.paste(fetch_tile(session, z, row, col),
                         ((col - c0) * TILE_SIZE, (row - r0) * TILE_SIZE))
    return mosaic.crop((x0 - c0 * TILE_SIZE, y0 - r0 * TILE_SIZE,
                        x1 - c0 * TILE_SIZE, y1 - r0 * TILE_SIZE))


def laplacian_var(img: Image.Image) -> float:
    """Sharpness proxy: variance of a 4-neighbor Laplacian on the gray channel."""
    a = np.asarray(img.convert("L"), dtype=np.float64)
    lap = (-4 * a[1:-1, 1:-1] + a[:-2, 1:-1] + a[2:, 1:-1] + a[1:-1, :-2] + a[1:-1, 2:])
    return float(lap.var())


def stats(img: Image.Image) -> dict:
    a = np.asarray(img, dtype=np.float64)
    r, g, b = a[..., 0], a[..., 1], a[..., 2]
    exg = 2 * g - r - b  # excess-green
    return {
        "size": list(img.size),
        "mean_rgb": [round(float(c.mean()), 1) for c in (r, g, b)],
        "brightness_std": round(float(a.mean(axis=2).std()), 1),
        "green_frac_exg20": round(float((exg > 20).mean()), 3),
        "laplacian_var": round(laplacian_var(img), 1),
        "blank_frac": round(float((a.mean(axis=2) < 5).mean()), 4),
    }


def detail_gain(fine: Image.Image, coarse: Image.Image) -> float:
    """Sharpness of the real fine chip vs the coarse chip bicubic-upsampled to the
    same size. Directional only — JPEG artifacts inflate it, so a high ratio is
    necessary but not sufficient; the settled verdict came from visual inspection
    (2026-07-17: real detail through L22, see docs/phase2-findings.md)."""
    up = coarse.resize(fine.size, Image.BICUBIC)
    lv_up = laplacian_var(up)
    return round(laplacian_var(fine) / lv_up, 2) if lv_up > 0 else float("nan")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default="./dogis_probe_out")
    args = ap.parse_args()
    os.makedirs(args.out, exist_ok=True)

    print("=" * 66)
    print("DOGIS PROBE REPORT BEGIN — copy everything down to REPORT END")
    print("=" * 66)
    print(f"machine: {platform.machine()} | {platform.platform()}")
    print(f"python: {platform.python_version()} | numpy: {np.__version__}")
    print(f"service: {SERVICE}")
    print(f"levels: {LEVELS} | chip: {CHIP_M:.0f} m | out: {os.path.abspath(args.out)}")

    session = requests.Session()
    session.headers["User-Agent"] = "lawn-estimator-exp4-probe/1.0"
    report: dict = {}

    try:
        meta = session.get(SERVICE, params={"f": "json"}, timeout=30).json()
        print(f"service ok: caps={meta.get('capabilities')} "
              f"maxLOD={max(l['level'] for l in meta['tileInfo']['lods'])} "
              f"format={meta['tileInfo'].get('format')}")
    except Exception:
        print("SERVICE METADATA FAILED (continuing — tiles may still work):")
        print(traceback.format_exc())

    for address, fallback in ADDRESSES.items():
        entry: dict = {}
        report[address] = entry
        print(f"\n--- {address} ---")
        try:
            try:
                lat, lon = geocode(session, address)
                src = "geocoded"
            except Exception as e:
                if fallback is None:
                    raise
                lat, lon = fallback
                src = f"fallback coords (geocode failed: {e})"
            entry["latlon"] = [round(lat, 6), round(lon, 6)]
            print(f"location: {lat:.6f}, {lon:.6f} ({src})")

            chips: dict[int, Image.Image] = {}
            for z in LEVELS:
                slug = address.lower().replace(" ", "_")
                path = os.path.join(args.out, f"{slug}_L{z}.png")
                t0 = time.perf_counter()
                if os.path.exists(path):
                    chips[z] = Image.open(path).convert("RGB")
                    fetched = "cached"
                else:
                    chips[z] = fetch_chip(session, lat, lon, z)
                    chips[z].save(path)
                    fetched = f"{time.perf_counter() - t0:.1f}s"
                s = stats(chips[z])
                s["mpp_cm"] = round(meters_per_pixel(lat, z) * 100, 1)
                s["fetch"] = fetched
                entry[f"L{z}"] = s
                print(f"L{z}: {json.dumps(s)}")

            gains = {}
            for zc, zf in zip(LEVELS, LEVELS[1:]):
                gains[f"L{zf}_vs_L{zc}"] = detail_gain(chips[zf], chips[zc])
            entry["detail_gain"] = gains
            print(f"detail gain (>=1.3 => real detail at finer level): {json.dumps(gains)}")
        except Exception:
            entry["error"] = "FAILED"
            print(f"{address}: FAILED")
            print(traceback.format_exc())

    ok = [a for a, e in report.items() if "error" not in e]
    print(f"\nsummary: {len(ok)}/{len(ADDRESSES)} addresses ok")
    greens = [e[f"L{LEVELS[-2]}"]["green_frac_exg20"] for e in report.values()
              if f"L{LEVELS[-2]}" in e]
    if greens:
        print(f"green fraction (ExG>20) across chips: min {min(greens):.2f} "
              f"max {max(greens):.2f} — low values = leaf-off/dormant (expected for 2025 flight)")
    print("=" * 66)
    print("REPORT END")
    print("=" * 66)


if __name__ == "__main__":
    main()