Upload app.py with huggingface_hub

app.py

@@ -0,0 +1,276 @@
+"""
+Facade — architectural style identification.
+
+Photograph a building; get the closest styles from a synthetic reference
+corpus, a reading of what you are looking at, and real examples nearby.
+
+Design decisions that matter for a free Space:
+  * The corpus index is PRECOMPUTED and loaded from the Hub. Re-embedding a
+    thousand plates on every cold start would make the app unusable.
+  * The vision model loads lazily on first query, not at import. A Space that
+    times out during a live demo is worse than one that is slow once.
+  * OpenStreetMap is queried only on demand and failures degrade silently —
+    a rate-limited third party must never take the app down mid-demo.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import urllib.parse
+import urllib.request
+
+import gradio as gr
+import numpy as np
+
+# ZeroGPU: free Gradio hosting requires dynamic GPU allocation. The `spaces`
+# module is only present on a Space, so import defensively — the same file
+# must still run locally and in a notebook.
+try:
+    import spaces
+    ZERO_GPU = True
+except ImportError:  # local / Colab
+    ZERO_GPU = False
+
+    class _Shim:
+        @staticmethod
+        def GPU(*a, **k):
+            def deco(fn):
+                return fn
+            return deco
+
+    spaces = _Shim()
+import pandas as pd
+import torch
+from huggingface_hub import hf_hub_download, snapshot_download
+from PIL import Image
+
+DATASET_REPO = os.environ.get("FACADE_DATASET", "USERNAME/facade-styles")
+MODEL_ID = os.environ.get(
+    "FACADE_MODEL", "laion/CLIP-ViT-B-32-laion2B-s34B-b79K")
+TOP_K = 3
+
+_model = None
+_proc = None
+_state: dict = {}
+
+
+# --------------------------------------------------------------------------
+# Loading
+# --------------------------------------------------------------------------
+
+def load_index():
+    """Fetch the precomputed index and manifest from the Hub."""
+    if _state:
+        return _state
+    emb_path = hf_hub_download(DATASET_REPO, "index_embeddings.npy",
+                               repo_type="dataset")
+    ids_path = hf_hub_download(DATASET_REPO, "index_plate_ids.csv",
+                               repo_type="dataset")
+    man_path = hf_hub_download(DATASET_REPO, "plate_manifest.parquet",
+                               repo_type="dataset")
+    styles_path = hf_hub_download(DATASET_REPO, "style_seed.csv",
+                                  repo_type="dataset")
+
+    _state["E"] = np.load(emb_path)
+    _state["plate_ids"] = pd.read_csv(ids_path)["plate_id"].tolist()
+    _state["manifest"] = pd.read_parquet(man_path).set_index("plate_id")
+    _state["styles"] = pd.read_csv(styles_path).set_index("style_id")
+    _state["style_of"] = np.array([p.split("-")[0] for p in _state["plate_ids"]])
+    return _state
+
+
+def get_model():
+    """Load lazily and on CPU.
+
+    ZeroGPU allocates a device only inside an @spaces.GPU function, so the
+    model must not touch CUDA at import time — doing so breaks the Space at
+    startup rather than at first query.
+    """
+    global _model, _proc
+    if _model is None:
+        from transformers import AutoModel, AutoProcessor
+        _model = AutoModel.from_pretrained(MODEL_ID).eval()
+        _proc = AutoProcessor.from_pretrained(MODEL_ID)
+    return _model, _proc
+
+
+def _as_tensor(x):
+    if torch.is_tensor(x):
+        return x
+    for a in ("image_embeds", "pooler_output", "last_hidden_state"):
+        v = getattr(x, a, None)
+        if torch.is_tensor(v):
+            return v.mean(1) if v.dim() == 3 else v
+    raise TypeError(type(x))
+
+
+@spaces.GPU(duration=30)
+def embed_image(img: Image.Image) -> np.ndarray:
+    model, proc = get_model()
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    model = model.to(device)
+    with torch.no_grad():
+        px = proc(images=[img.convert("RGB")],
+                  return_tensors="pt")["pixel_values"].to(device)
+        v = _as_tensor(model.get_image_features(pixel_values=px)).float()
+        v = v / v.norm(dim=-1, keepdim=True)
+    return v[0].cpu().numpy()
+
+
+# --------------------------------------------------------------------------
+# Geographic prior (OpenStreetMap)
+# --------------------------------------------------------------------------
+
+OVERPASS = "https://overpass-api.de/api/interpreter"
+
+
+def nearby_eras(lat: float, lon: float, radius_m: int = 1500) -> dict:
+    """Construction dates of real buildings near a point.
+
+    Returns an empty dict on any failure. A rate-limited third party must not
+    be able to break the app during a demo.
+    """
+    q = (f"[out:json][timeout:25];"
+         f'(way["building"]["start_date"](around:{radius_m},{lat},{lon});'
+         f'relation["building"]["start_date"](around:{radius_m},{lat},{lon}););'
+         f"out tags 300;")
+    try:
+        req = urllib.request.Request(
+            OVERPASS, data=urllib.parse.urlencode({"data": q}).encode(),
+            headers={"User-Agent": "facade-app/1.0"})
+        with urllib.request.urlopen(req, timeout=25) as r:
+            data = json.load(r)
+    except Exception:
+        return {}
+
+    counts: dict[int, int] = {}
+    for el in data.get("elements", []):
+        d = str(el.get("tags", {}).get("start_date", ""))[:4]
+        if d.isdigit():
+            counts[int(d)] = counts.get(int(d), 0) + 1
+    return counts
+
+
+def period_prior(style_ids, era_counts, tolerance: int = 40) -> np.ndarray:
+    """Soft prior over styles, from how many nearby buildings share their era.
+
+    Soft on purpose: a genuinely unusual building should still be findable, so
+    this reranks rather than filters.
+    """
+    if not era_counts:
+        return np.zeros(len(style_ids))
+    styles = load_index()["styles"]
+    out = []
+    for sid in style_ids:
+        try:
+            start = int(str(styles.loc[sid, "period"]).split("-")[0])
+        except (ValueError, KeyError):
+            out.append(0.0)
+            continue
+        out.append(sum(c for yr, c in era_counts.items()
+                       if abs(yr - start) <= tolerance))
+    arr = np.array(out, dtype=float)
+    return arr / (arr.max() or 1.0)
+
+
+# --------------------------------------------------------------------------
+# Core query
+# --------------------------------------------------------------------------
+
+def identify(image, use_location: bool, lat: float, lon: float,
+             rerank_weight: float):
+    if image is None:
+        return "Upload a photograph of a building facade to begin.", None, ""
+
+    s = load_index()
+    q = embed_image(image)
+    scores = s["E"] @ q
+
+    note = ""
+    if use_location:
+        eras = nearby_eras(lat, lon)
+        if eras:
+            prior = period_prior(s["style_of"], eras)
+            scores = scores + rerank_weight * prior
+            note = (f"\n\n*Reranked using {sum(eras.values())} dated buildings "
+                    f"within 1.5 km.*")
+        else:
+            note = "\n\n*No dated OpenStreetMap buildings nearby; ranking is visual only.*"
+
+    # Best plate per style, then the top styles.
+    best: dict[str, tuple[float, int]] = {}
+    for i, sid in enumerate(s["style_of"]):
+        if sid not in best or scores[i] > best[sid][0]:
+            best[sid] = (float(scores[i]), i)
+    ranked = sorted(best.items(), key=lambda kv: -kv[1][0])[:TOP_K]
+
+    total = sum(np.exp(np.array([r[1][0] for r in ranked]) * 12))
+    lines, gallery = [], []
+    for rank, (sid, (score, idx)) in enumerate(ranked, 1):
+        row = s["styles"].loc[sid]
+        conf = float(np.exp(score * 12) / total)
+        lines.append(
+            f"### {rank}. {row['style_name']}  ·  {conf:.0%}\n"
+            f"**{row['period']}** — {row['key_features']}\n\n"
+            f"{row['massing']}; {row['primary_material']}; "
+            f"{row['window_rhythm']}."
+        )
+        pid = s["plate_ids"][idx]
+        gallery.append((plate_url(pid), f"{row['style_name']} (reference)"))
+
+    reading = ""
+    top_pid = s["plate_ids"][ranked[0][1][1]]
+    man = s["manifest"]
+    if "reading" in man.columns and pd.notna(man.loc[top_pid].get("reading")):
+        reading = f"**What you are looking at**\n\n{man.loc[top_pid]['reading']}"
+
+    caveat = (
+        "\n\n---\n*Visual-similarity search over a synthetic reference corpus. "
+        "Suggestions are stylistic, not an authoritative attribution, and carry "
+        "no claim about a building's architect, date, or heritage status.*"
+    )
+    return "\n\n".join(lines) + note + caveat, gallery, reading
+
+
+def plate_url(plate_id: str) -> str:
+    return (f"https://huggingface.co/datasets/{DATASET_REPO}/resolve/main/"
+            f"plates/{plate_id}.png")
+
+
+# --------------------------------------------------------------------------
+# Interface
+# --------------------------------------------------------------------------
+
+EXAMPLE_NOTE = """
+**Tip — include the whole building.** Retrieval is measurably weaker on facade
+close-ups: architectural style lives in massing, roofline and silhouette, and
+a cropped window grid discards all three. Step back if you can.
+"""
+
+with gr.Blocks(title="Facade — architectural style finder") as demo:
+    gr.Markdown("# Facade\n### Point a camera at a building. Find out what you are looking at.")
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            img = gr.Image(type="pil", label="Building photograph", height=340)
+            gr.Markdown(EXAMPLE_NOTE)
+            use_loc = gr.Checkbox(label="Use my location to rerank", value=False)
+            with gr.Row():
+                lat = gr.Number(label="Latitude", value=32.0771, precision=4)
+                lon = gr.Number(label="Longitude", value=34.7745, precision=4)
+            weight = gr.Slider(0.0, 0.6, value=0.25, step=0.05,
+                               label="Geographic prior weight")
+            go = gr.Button("Identify", variant="primary")
+
+        with gr.Column(scale=1):
+            out = gr.Markdown()
+            reading = gr.Markdown()
+            gallery = gr.Gallery(label="Closest reference plates", columns=3,
+                                 height=220)
+
+    go.click(identify, [img, use_loc, lat, lon, weight], [out, gallery, reading])
+
+if __name__ == "__main__":
+    demo.launch()