gods-eye-view/intro-sequence/tools/flurry/tiles.py

"""Polite cached Esri World Imagery fetch + 512px centred capture + quality metrics."""
import hashlib
import io
import math
import os
import threading
import time

import numpy as np
import requests
from PIL import Image

HERE = os.path.dirname(os.path.abspath(__file__))
CACHE = os.path.join(HERE, "cache")
ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
OUT = os.path.join(ROOT, "imagery", "flurry")

ESRI = "https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}"
UA = "gods-eye-view-intro/1.0 (flurry montage prep)"
MAX_WORKERS = 8
PLACEHOLDER_MD5 = {"f27d9de7f80c13501f470595e327aa6d"}  # "Map data not yet available"

_local = threading.local()


def _session():
    s = getattr(_local, "s", None)
    if s is None:
        s = requests.Session()
        s.headers["User-Agent"] = UA
        _local.s = s
    return s


def fetch_tile(z, x, y, retries=6):
    n = 2 ** z
    x %= n
    path = os.path.join(CACHE, str(z), str(x), f"{y}.jpg")
    if os.path.exists(path) and os.path.getsize(path) > 0:
        with open(path, "rb") as f:
            return f.read()
    os.makedirs(os.path.dirname(path), exist_ok=True)
    url = ESRI.format(z=z, x=x, y=y)
    last = None
    for attempt in range(retries):
        try:
            r = _session().get(url, timeout=45)
            if r.status_code == 200 and r.content:
                tmp = path + f".{threading.get_ident()}.part"
                with open(tmp, "wb") as f:
                    f.write(r.content)
                os.replace(tmp, path)
                return r.content
            last = f"HTTP {r.status_code}"
            if r.status_code in (400, 404):
                break
        except Exception as e:
            last = repr(e)
        time.sleep(min(20, 1.0 * 2 ** attempt))
    raise RuntimeError(f"failed {url}: {last}")


def is_placeholder(data):
    if hashlib.md5(data).hexdigest() in PLACEHOLDER_MD5:
        return True
    a = np.asarray(Image.open(io.BytesIO(data)).convert("RGB")).astype(np.float32)
    m = a.mean(axis=(0, 1))
    return bool(abs(m - 204.7).max() < 4 and a.std() < 9)


def pixel_xy(lat, lon, z):
    n = 2 ** z * 256
    px = (lon + 180.0) / 360.0 * n
    s = math.sin(math.radians(lat))
    py = (0.5 - math.log((1 + s) / (1 - s)) / (4 * math.pi)) * n
    return px, py


def capture(lat, lon, z, size=512):
    """Return (image, info). info has placeholder flag and quality metrics."""
    px, py = pixel_xy(lat, lon, z)
    tx, ty = int(px // 256), int(py // 256)
    mosaic = Image.new("RGB", (768, 768))
    ox, oy = (tx - 1) * 256, (ty - 1) * 256
    left, top = int(round(px - ox - size / 2)), int(round(py - oy - size / 2))
    ph = False
    for j in range(3):
        for i in range(3):
            data = fetch_tile(z, tx - 1 + i, ty - 1 + j)
            # does this tile overlap the crop window?
            tl, tt = i * 256, j * 256
            overlaps = tl < left + size and tl + 256 > left and tt < top + size and tt + 256 > top
            if overlaps and is_placeholder(data):
                ph = True
            mosaic.paste(Image.open(io.BytesIO(data)).convert("RGB"), (tl, tt))
    img = mosaic.crop((left, top, left + size, top + size))
    info = metrics(img)
    info["placeholder"] = ph
    return img, info


def metrics(img):
    a = np.asarray(img).astype(np.float32)
    lum = a @ np.array([0.299, 0.587, 0.114], np.float32)
    small = np.asarray(img.convert("L").resize((128, 128), Image.BILINEAR)).astype(np.float32)
    gx = np.abs(np.diff(lum, axis=1)).mean()
    gy = np.abs(np.diff(lum, axis=0)).mean()
    # coarse-scale structure: std of 32px block means (catches flat ocean/cloud/ice)
    blocks = lum.reshape(16, 32, 16, 32).mean(axis=(1, 3))
    sat = (a.max(axis=2) - a.min(axis=2)).mean()
    return {
        "std": float(lum.std()),
        "grad": float((gx + gy) / 2),
        "block_std": float(blocks.std()),
        "small_std": float(small.std()),
        "sat": float(sat),
        "mean": float(lum.mean()),
    }


def quality_ok(info):
    if info["placeholder"]:
        return False, "placeholder"
    if info["std"] < 12:
        return False, f"low variance (std {info['std']:.1f})"
    if info["grad"] < 1.3:
        return False, f"featureless (grad {info['grad']:.1f})"
    if info["mean"] > 225 and info["std"] < 25:
        return False, "cloud/snow white-out"
    if info["mean"] < 22:
        return False, "too dark"
    return True, ""