Spaces:

SharpMeow
/

paris4-ink-map

Running

Paris 4 Ink Map 1.2: 3D scroll, neighboring turns, VC3D project, saved views
d3405a2 verified 7 days ago
48 kB
"""Paris 4 Ink Map: the organizers' whole-scroll PHerc. Paris 4 ink map, streamed as deep-zoom tiles.
Public Space (SharpMeow/paris4-ink-map). Everything it shows comes from the Vesuvius Challenge open-data bucket at
start-up, read over HTTPS with no account: the ink map of segment 20260926044226-spiral-output-generation-19 (the
organizers' whole-scroll spiral fit), its winding map and mesh, and the 2.4 um 78 keV CT for the cross-section map.
Nothing is copied into this repository or the Space's files; the downloads link to the bucket itself.
  GET  /                        the page (static/)
  GET  /api/status              loading progress
  GET  /api/info                sizes, scale, windings, sources with their SHA-256
  GET  /tiles/{ink|wind}.dzi    Deep Zoom descriptors; tiles at /tiles/{name}_files/{level}/{col}_{row}.{jpg|png}
  GET  /api/overview.png        the whole map wrapped into rows (rows, width, cmap)
  GET  /api/profile             mean ink per 16-pixel column bin (for the density track)
  GET  /api/point               one map pixel: winding, distances, scan height and voxel position
  GET  /api/section             the fit's spiral at one scan height (points in CT level-4 pixels)
  GET  /api/ct.png              the CT slice at one scan height, level 4 (38.4 um), cropped to the scroll
  GET  /api/crop                a region at full or reduced resolution: PNG up to 60 Mpx, tiled TIFF of any size
  GET  /api/neighbors            the turns just outside and inside one pixel: where, how far, how alike
  GET  /api/vc3d.zip            a VC3D project: the 2.4 um CT streamed from the open data, the fit as a segment
  GET  /api/mesh3d.bin          the fit's mesh as a compact point cloud for the 3D view (scroll3d.py; .json: layout)
  GET  /zarr/                   the map as a streamable OME-Zarr 0.4 (zarr v2, 512 px zlib chunks, 9 levels)
  GET  /api                     this list as JSON; /docs and /openapi.json describe every endpoint
  POST /mcp                     the same as an MCP server for agents (p4_mcp.py)
An ink map is model output, never a reading.
"""
from __future__ import annotations

import hashlib
import http.client
import io
import json
import math
import os
import threading
import tempfile
import time
import zipfile
import zlib
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache
from pathlib import Path

import numpy as np
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles
from PIL import Image

Image.MAX_IMAGE_PIXELS = 400_000_000

HERE = Path(__file__).resolve().parent
CACHE = Path(os.environ.get("CACHE", "/tmp/p4cache"))
LOCAL = os.environ.get("DATA_DIR")            # a folder holding the same files, for tests and local runs
OPEN = "https://vesuvius-challenge-open-data.s3.us-east-1.amazonaws.com"
SEG = "PHercParis4/segments/20260926044226-spiral-output-generation-19"
SEG_ID = "20260926044226-spiral-output-generation-19"
SPACE_URL = "https://sharpmeow-paris4-ink-map.hf.space"
PNG_MAX = 60_000_000         # pixels in one PNG download; TIFF has no limit
ZCHUNK = 512                 # chunk side of the OME-Zarr view
INK_NAME = ("PHercParis4-20260926044226-2.4um-0.22m-78keV-volume-20260411134726-ink3d-20260926084000-student-v12A-"
            "25000-c7766be9-l128-accmax4-max-52-76-scale0.0625-nearest")
CT_ZARR = f"{OPEN}/PHercParis4/volumes/20260411134726-2.400um-0.2m-78keV-masked.zarr"
CT_LEVEL = 4                  # 38.4 um; the fit's coordinates are 9.6 um voxels (level 2), so level 4 = fit / 4
CT_SHAPE = (4737, 2044, 2044)  # level-4 shape (z, y, x), from its .zarray
CHUNK = 128

SOURCES = {
    "ink": {"title": "Ink Map, Full Resolution (BigTIFF, uint8, 64 px Tiles)", "path": f"{SEG}/ink-detection/{INK_NAME}.tif",
            "local": "ink/ink.tif", "load": True},
    "ink_jpg": {"title": "Ink Map, Downsampled 4x (JPEG)", "path": f"{SEG}/ink-detection/downsampled/{INK_NAME}-ds4.jpg"},
    "windings": {"title": "Winding Map (float32 Winding ID per Mesh Node)", "path": f"{SEG}/mesh/spiral-fit/windings.tif",
                 "local": "spiral/windings.tif", "load": True},
    "meta": {"title": "Spiral Fit meta.json (Winding Bounds, Grid)", "path": f"{SEG}/mesh/spiral-fit/meta.json",
             "local": "norm/meta.json", "load": True},
    "x": {"title": "Mesh X (9.6 um Voxels of Scan 20260411134726)", "path": f"{SEG}/mesh/intermediate/tifxyz_normalized/x.tif",
          "local": "norm/x.tif", "load": True},
    "y": {"title": "Mesh Y", "path": f"{SEG}/mesh/intermediate/tifxyz_normalized/y.tif", "local": "norm/y.tif", "load": True},
    "z": {"title": "Mesh Z", "path": f"{SEG}/mesh/intermediate/tifxyz_normalized/z.tif", "local": "norm/z.tif", "load": True},
    "notes": {"title": "The Fit Author's Notes (FIT_NOTES.md)", "path": f"{SEG}/mesh/spiral-fit/FIT_NOTES.md"},
    "windings_txt": {"title": "What windings.tif Holds (WINDINGS.txt)", "path": f"{SEG}/mesh/spiral-fit/WINDINGS.txt"},
    "mesh_791": {"title": "Mesh on the 7.91 um Scan 20230205180739 (tifxyz Folder)",
                 "path": f"{SEG}/mesh/20260926044226-on-20230205180739-7.91um.tifxyz/"},
    "mesh_455": {"title": "Mesh on the 45.532 um Scan 20260310170716 (tifxyz Folder)",
                 "path": f"{SEG}/mesh/20260926044226-on-20260310170716-45.532um.tifxyz/"},
    "ckpt": {"title": "Spiral Fitter Checkpoint (1.29 GB, PyTorch; Load Only a Trusted Copy)",
             "path": f"{SEG}/mesh/spiral-fit/checkpoint_autosave.ckpt"},
    "ct": {"title": "CT, 2.4 um 78 keV Scan 20260411134726 (OME-Zarr 0.4 on zarr v2, Uncompressed, 128 Cubed Chunks)",
           "path": "PHercParis4/volumes/20260411134726-2.400um-0.2m-78keV-masked.zarr/"},
}

MM_PER_GRID = 0.192          # one mesh grid step: 20 voxels of 9.6 um
INK_PER_GRID = 2.5           # ink pixels per grid step (142,283 / 56,913)
MM_PER_PX = MM_PER_GRID / INK_PER_GRID    # 0.0768
TILE = 256
XYZ_STRIDE = 2               # the mesh kept in memory every 2nd node (0.384 mm)

CMAPS = {
    "gray": [(0, (0, 0, 0)), (1, (255, 255, 255))],
    "ember": [(0, (0, 0, 4)), (0.25, (87, 16, 110)), (0.5, (188, 55, 84)), (0.75, (249, 142, 9)), (1, (252, 255, 164))],
    "ice": [(0, (2, 6, 14)), (0.35, (20, 60, 130)), (0.7, (70, 175, 220)), (1, (235, 250, 255))],
    "viridis": [(0, (68, 1, 84)), (0.25, (59, 82, 139)), (0.5, (33, 145, 140)), (0.75, (94, 201, 98)), (1, (253, 231, 37))],
}

state = {"phase": "starting", "done_bytes": 0, "total_bytes": 0, "error": None, "started": time.time(),
         "ready": False, "sha256": {}, "bytes": {}}
D: dict = {}                  # the loaded arrays
_lock = threading.Lock()


def url(key):
    return f"{OPEN}/{SOURCES[key]['path']}"


def lut(name: str) -> np.ndarray:
    """A 256 x 3 uint8 color table from the control points of CMAPS; unknown names give gray."""
    return _lut(name if name in CMAPS else "gray")


@lru_cache(maxsize=None)
def _lut(name: str) -> np.ndarray:
    pts = CMAPS[name]
    xs = np.array([p for p, _ in pts])
    out = np.zeros((256, 3), np.uint8)
    t = np.linspace(0, 1, 256)
    for ch in range(3):
        out[:, ch] = np.round(np.interp(t, xs, [c[ch] for _, c in pts])).astype(np.uint8)
    return out


def winding_rgba(w: np.ndarray, alpha: int = 110) -> np.ndarray:
    """RGBA for winding ids: neighbors get far-apart hues; -2 (ambiguous) gray, -1 (no mesh) clear."""
    out = np.zeros(w.shape + (4,), np.uint8)
    ok = w >= 0
    h = (w[ok].astype(np.float64) * 0.6180339887) % 1.0
    out[ok, :3] = hsv(h, 0.72, 1.0)
    out[ok, 3] = alpha
    amb = w == -2
    out[amb] = (150, 150, 150, alpha)
    return out


def hsv(h, s, v):
    i = np.floor(h * 6).astype(int) % 6
    f = h * 6 - np.floor(h * 6)
    p, q, t = v * (1 - s), v * (1 - s * f), v * (1 - s * (1 - f))
    r = np.choose(i, [v, q, p, p, t, v])
    g = np.choose(i, [t, v, v, q, p, p])
    b = np.choose(i, [p, p, t, v, v, q])
    return (np.stack([r, g, b], -1) * 255).round().astype(np.uint8)


# ---------------------------------------------------------------- loading

def fetch(key) -> Path:
    """A source file, downloaded once into CACHE (or read from DATA_DIR), with its SHA-256 recorded."""
    if LOCAL:
        p = Path(LOCAL) / SOURCES[key]["local"]
        state["bytes"][key] = p.stat().st_size
        return p
    p = CACHE / SOURCES[key]["local"]
    if p.exists() and (p.with_suffix(p.suffix + ".sha256")).exists():
        state["sha256"][key] = p.with_suffix(p.suffix + ".sha256").read_text().strip()
        state["bytes"][key] = p.stat().st_size
        state["done_bytes"] += p.stat().st_size
        return p
    p.parent.mkdir(parents=True, exist_ok=True)
    h = hashlib.sha256()
    tmp = p.with_suffix(p.suffix + ".part")
    for attempt in range(4):
        try:
            got = 0
            with urllib.request.urlopen(url(key), timeout=60) as r, open(tmp, "wb") as f:
                want = int(r.headers.get("Content-Length") or -1)
                while True:
                    b = r.read(1 << 22)
                    if not b:
                        break
                    f.write(b)
                    h.update(b)
                    got += len(b)
                    state["done_bytes"] += len(b)
            if want >= 0 and got != want:
                raise ConnectionError(f"{key}: {got} of {want} bytes")
            break
        except (urllib.error.URLError, TimeoutError, ConnectionError, http.client.HTTPException):
            state["done_bytes"] -= got
            if attempt == 3:
                raise
            h = hashlib.sha256()
            time.sleep(2 ** (attempt + 1))
    tmp.rename(p)
    state["sha256"][key] = h.hexdigest()
    state["bytes"][key] = p.stat().st_size
    p.with_suffix(p.suffix + ".sha256").write_text(h.hexdigest())
    return p


def head_size(key) -> int:
    try:
        req = urllib.request.Request(url(key), method="HEAD")
        with urllib.request.urlopen(req, timeout=20) as r:
            return int(r.headers.get("Content-Length") or 0)
    except Exception:  # noqa: BLE001
        return 0


def pyramid(full: np.ndarray, levels: int) -> list:
    """full, then 2x2 means down to one pixel; uint8 throughout."""
    out = [full]
    a = full
    for _ in range(levels):
        h, w = a.shape
        if h == 1 and w == 1:
            break
        a = np.pad(a, ((0, h % 2), (0, w % 2)), mode="edge").astype(np.uint16)
        a = ((a[0::2, 0::2] + a[1::2, 0::2] + a[0::2, 1::2] + a[1::2, 1::2] + 2) // 4).astype(np.uint8)
        out.append(a)
    return out


def load():
    import tifffile
    try:
        keys = [k for k, s in SOURCES.items() if s.get("load")]
        if not LOCAL:
            state["phase"] = "sizing"
            state["total_bytes"] = sum(head_size(k) for k in keys)
        state["phase"] = "downloading"
        paths = {k: fetch(k) for k in keys}
        state["phase"] = "reading the ink map"
        ink = tifffile.imread(str(paths["ink"]))
        state["phase"] = "building the tile pyramid"
        H, W = ink.shape
        top = math.ceil(math.log2(max(H, W)))
        pyr = pyramid(ink, top)
        state["phase"] = "reading the mesh"
        meta = json.loads(paths["meta"].read_text())
        wind = tifffile.imread(str(paths["windings"])).astype(np.int16)
        xyz = []
        for k in "xyz":
            a = tifffile.imread(str(paths[k]))
            xyz.append(np.ascontiguousarray(a[::XYZ_STRIDE, ::XYZ_STRIDE]).astype(np.float32))
            del a
        X, Y, Z = xyz
        bad = X == -1
        for a in (X, Y, Z):
            a[bad] = np.nan
        state["phase"] = "measuring"
        gh, gw = wind.shape
        # valid mask and ink density at level 4 (16 px)
        l4 = pyr[4].astype(np.float32)
        rr = np.minimum((np.arange(l4.shape[0]) * 16 + 8) / INK_PER_GRID, gh - 1).astype(int)
        cc = np.minimum((np.arange(l4.shape[1]) * 16 + 8) / INK_PER_GRID, gw - 1).astype(int)
        m4 = (wind[rr][:, cc] != -1).astype(np.float32)
        prof = (l4 * m4).sum(0) / np.maximum(m4.sum(0), 1)
        windings = []
        for b in meta["winding_bounds"]:
            windings.append({"winding": b["winding"],
                             "x0": round(b["column_begin"] * INK_PER_GRID), "x1": round(b["column_end"] * INK_PER_GRID),
                             "y0": round(b["row_begin"] * INK_PER_GRID), "y1": round(b["row_end"] * INK_PER_GRID)})
        zmm = Z[~np.isnan(Z)] * 9.6e-3
        with _lock:
            D.update(ink=ink, pyr=pyr, top=top, W=W, H=H, wind=wind, X=X, Y=Y, Z=Z, meta=meta, windings=windings,
                     profile=np.round(prof).astype(int).tolist(), z_mm=[float(zmm.min()), float(zmm.max())],
                     bbox=[float(np.nanmin(X)), float(np.nanmin(Y)), float(np.nanmax(X)), float(np.nanmax(Y))])
        state.update(phase="ready", ready=True)
    except Exception as exc:  # noqa: BLE001
        state.update(phase="failed", error=f"{type(exc).__name__}: {exc}")
        raise


def need():
    if not state["ready"]:
        raise HTTPException(503, f"loading: {state['phase']}")


# ---------------------------------------------------------------- core functions (shared with the MCP server)

def dzi_levels(W: int, H: int) -> int:
    return math.ceil(math.log2(max(W, H)))


def tile_array(name: str, level: int, col: int, row: int, cmap: str = "gray", gamma: float = 1.0):
    """One Deep Zoom tile: (array, is_rgba). Level `top` is full resolution."""
    top = D["top"]
    if not 0 <= level <= top:
        raise HTTPException(404, "level")
    k = top - level                      # pyramid index: each step halves
    s = 2 ** k
    lw, lh = math.ceil(D["W"] / s), math.ceil(D["H"] / s)
    x0, y0 = col * TILE, row * TILE
    if x0 >= lw or y0 >= lh or col < 0 or row < 0:
        raise HTTPException(404, "tile")
    x1, y1 = min(x0 + TILE, lw), min(y0 + TILE, lh)
    if name == "ink":
        a = D["pyr"][k][y0:y1, x0:x1]
        if gamma != 1.0:
            a = (255 * (a / 255.0) ** (1 / gamma)).round().astype(np.uint8)
        return (a if cmap == "gray" else lut(cmap)[a]), False
    if name == "wind":
        g = D["wind"]
        rr = np.minimum(((np.arange(y0, y1) + 0.5) * s / INK_PER_GRID).astype(int), g.shape[0] - 1)
        cc = np.minimum(((np.arange(x0, x1) + 0.5) * s / INK_PER_GRID).astype(int), g.shape[1] - 1)
        return winding_rgba(g[np.ix_(rr, cc)]), True
    raise HTTPException(404, "layer")


def encode(a: np.ndarray, rgba: bool, fmt: str = "jpg", quality: int = 88) -> bytes:
    im = Image.fromarray(a)
    b = io.BytesIO()
    if rgba or fmt == "png":
        im.save(b, "PNG", optimize=False, compress_level=3)
    else:
        im.save(b, "JPEG", quality=quality)
    return b.getvalue()


def _bilinear(A, r, c):
    """A sampled at fractional (r, c) of the stride grid; NaN when a corner is missing."""
    h, w = A.shape
    r = min(max(r, 0), h - 1.001)
    c = min(max(c, 0), w - 1.001)
    r0, c0 = int(r), int(c)
    fr, fc = r - r0, c - c0
    q = A[r0:r0 + 2, c0:c0 + 2]
    if q.shape != (2, 2) or np.isnan(q).any():
        v = A[int(round(r)), int(round(c))]
        return float(v)
    return float((q[0, 0] * (1 - fc) + q[0, 1] * fc) * (1 - fr) + (q[1, 0] * (1 - fc) + q[1, 1] * fc) * fr)


def point(x: float, y: float) -> dict:
    """Everything known at map pixel (x, y)."""
    W, H = D["W"], D["H"]
    if not (0 <= x < W and 0 <= y < H):
        raise HTTPException(400, "outside the map")
    g = D["wind"]
    gr, gc = y / INK_PER_GRID, x / INK_PER_GRID
    w = int(g[min(int(gr), g.shape[0] - 1), min(int(gc), g.shape[1] - 1)])
    out = {"x": round(x, 1), "y": round(y, 1), "ink": int(D["ink"][int(y), int(x)]),
           "winding": w if w >= 0 else None, "on_mesh": w != -1, "ambiguous_winding": w == -2,
           "mm_from_outer_end": round(x * MM_PER_PX, 2), "mm_from_inner_end": round((W - x) * MM_PER_PX, 2),
           "mm_down_map": round(y * MM_PER_PX, 2), "grid": [round(gr, 2), round(gc, 2)]}
    if w != -1:
        sr, sc = gr / XYZ_STRIDE, gc / XYZ_STRIDE
        x9, y9, z9 = (_bilinear(D[k], sr, sc) for k in "XYZ")
        if not any(math.isnan(v) for v in (x9, y9, z9)):
            out.update({"scan": "20260411134726 (2.4 um, 78 keV)",
                        "voxel_2_4um_xyz": [round(x9 * 4), round(y9 * 4), round(z9 * 4)],
                        "voxel_9_6um_xyz": [round(x9, 1), round(y9, 1), round(z9, 1)],
                        "scan_z_mm": round(z9 * 9.6e-3, 3)})
    return out


def winding_info(w: int) -> dict:
    for b in D["windings"]:
        if b["winding"] == w:
            return dict(b, length_mm=round((b["x1"] - b["x0"]) * MM_PER_PX, 1))
    raise HTTPException(404, f"no winding {w}")


def section(z_mm: float, step: int = 2) -> dict:
    """The fit's spiral at scan height z_mm: for every `step`-th kept column, the node whose z is z_mm (linear
    between rows), in CT level-4 pixels, with its winding and its map x."""
    Z, X, Y = D["Z"], D["X"], D["Y"]
    z0 = z_mm / 9.6e-3
    cols = np.arange(0, Z.shape[1], step)
    Zc = Z[:, cols]
    d = Zc - z0
    with np.errstate(invalid="ignore"):
        s = np.sign(d)
        cross = (s[:-1] * s[1:] <= 0) & ~np.isnan(d[:-1]) & ~np.isnan(d[1:])
    has = cross.any(0)
    r0 = np.argmax(cross, 0)
    ci = np.nonzero(has)[0]
    r0 = r0[ci]
    da, db = d[r0, ci], d[r0 + 1, ci]
    t = np.where(np.abs(da - db) > 1e-6, da / (da - db), 0.0)
    cc = cols[ci]
    xs = X[r0, cc] * (1 - t) + X[r0 + 1, cc] * t
    ys = Y[r0, cc] * (1 - t) + Y[r0 + 1, cc] * t
    rows_full = (r0 + t) * XYZ_STRIDE
    g = D["wind"]
    wv = g[np.minimum(np.round(rows_full).astype(int), g.shape[0] - 1), np.minimum(cc * XYZ_STRIDE, g.shape[1] - 1)]
    mx = cc * XYZ_STRIDE * INK_PER_GRID
    my = rows_full * INK_PER_GRID
    lv = 2 ** (CT_LEVEL - 2)
    return {"z_mm": z_mm, "level": CT_LEVEL, "um_per_px": 2.4 * 2 ** CT_LEVEL, "n": int(len(ci)),
            "x": np.round(xs / lv, 1).tolist(), "y": np.round(ys / lv, 1).tolist(),
            "winding": wv.astype(int).tolist(), "map_x": np.round(mx).astype(int).tolist(),
            "map_y": np.round(my).astype(int).tolist()}


def ct_crop_box():
    x0, y0, x1, y1 = D["bbox"]
    lv = 2 ** (CT_LEVEL - 2)
    pad = 24
    return (max(0, int(x0 / lv) - pad), max(0, int(y0 / lv) - pad),
            min(CT_SHAPE[2], int(math.ceil(x1 / lv)) + pad), min(CT_SHAPE[1], int(math.ceil(y1 / lv)) + pad))


def _plane(zi, cy, cx):
    """One z plane of one uncompressed 128^3 chunk by an HTTP range request (16 KB); zeros when masked (403, 404).
    None when the read failed three times, so the slice is not cached."""
    off = (zi % CHUNK) * CHUNK * CHUNK
    u = f"{CT_ZARR}/{CT_LEVEL}/{zi // CHUNK}/{cy}/{cx}"
    req = urllib.request.Request(u, headers={"Range": f"bytes={off}-{off + CHUNK * CHUNK - 1}"})
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                b = r.read()
            return np.frombuffer(b, np.uint8).reshape(CHUNK, CHUNK)
        except urllib.error.HTTPError as e:
            if e.code in (403, 404):
                return np.zeros((CHUNK, CHUNK), np.uint8)
            time.sleep(1 + attempt)
        except (urllib.error.URLError, TimeoutError, ConnectionError, http.client.HTTPException, ValueError):
            time.sleep(1 + attempt)
    return None


_pool = ThreadPoolExecutor(16)
_heavy = threading.BoundedSemaphore(2)   # crops, overviews, sections and CT reads: two at a time on cpu-basic


class busy:
    """Hold one of the two heavy-work slots; a request that waits over 30 s gets a 503 instead of piling up."""

    def __enter__(self):
        if not _heavy.acquire(timeout=30):
            raise HTTPException(503, "busy: try again in a moment")

    def __exit__(self, *exc):
        _heavy.release()


_ct_cache: dict = {}          # zi -> PNG, only slices whose every plane was read
_ct_lock = threading.Lock()
CT_CACHE_MAX = 48


def ct_slice(zi: int) -> bytes:
    """The level-4 CT slice zi, cropped to the scroll, as PNG; cached unless a plane could not be read."""
    with _ct_lock:
        if zi in _ct_cache:
            return _ct_cache[zi]
    with busy():
        with _ct_lock:              # another request may have read it while this one waited
            if zi in _ct_cache:
                return _ct_cache[zi]
        png, complete = _ct_slice(zi)
    if complete:
        with _ct_lock:
            while len(_ct_cache) >= CT_CACHE_MAX:
                _ct_cache.pop(next(iter(_ct_cache)))
            _ct_cache[zi] = png
    return png


def _ct_slice(zi: int):
    x0, y0, x1, y1 = ct_crop_box()
    cys = range(y0 // CHUNK, (y1 - 1) // CHUNK + 1)
    cxs = range(x0 // CHUNK, (x1 - 1) // CHUNK + 1)
    jobs = {(cy, cx): _pool.submit(_plane, zi, cy, cx) for cy in cys for cx in cxs}
    big = np.zeros((len(cys) * CHUNK, len(cxs) * CHUNK), np.uint8)
    complete = True
    for (cy, cx), f in jobs.items():
        plane = f.result()
        if plane is None:
            complete = False
            continue
        big[(cy - cys[0]) * CHUNK:(cy - cys[0] + 1) * CHUNK, (cx - cxs[0]) * CHUNK:(cx - cxs[0] + 1) * CHUNK] = plane
    oy, ox = y0 - cys[0] * CHUNK, x0 - cxs[0] * CHUNK
    a = big[oy:oy + (y1 - y0), ox:ox + (x1 - x0)]
    return encode(a, False, "png"), complete


def crop(x0: int, y0: int, x1: int, y1: int, scale: int = 1, cmap: str = "gray", limit: int | None = PNG_MAX):
    """A map region reduced by `scale` (a power of two), as an array; at most `limit` pixels after reduction (None:
    any size, up to the whole map). Coordinates may come in either order and are clamped to the map."""
    W, H = D["W"], D["H"]
    if not all(math.isfinite(v) for v in (x0, y0, x1, y1)):
        raise HTTPException(400, "coordinates must be numbers")
    x0, x1 = (min(max(0, int(v)), W) for v in sorted((x0, x1)))
    y0, y1 = (min(max(0, int(v)), H) for v in sorted((y0, y1)))
    if x1 - x0 < 1 or y1 - y0 < 1:
        raise HTTPException(400, "empty region")
    scale = int(scale)
    if scale < 1 or scale & (scale - 1) or scale > 2 ** D["top"]:
        raise HTTPException(400, "scale must be a power of two from 1 to the whole map")
    k = scale.bit_length() - 1
    if limit is not None and ((x1 - x0) >> k) * ((y1 - y0) >> k) > limit:
        raise HTTPException(413, f"a PNG holds at most {limit:,} pixels: raise scale, or ask for fmt=tif, which has no limit")
    a = D["pyr"][k][y0 >> k:max((y0 >> k) + 1, y1 >> k), x0 >> k:max((x0 >> k) + 1, x1 >> k)]
    return a if cmap == "gray" else lut(cmap)[a]


def overview(rows: int, width: int, cmap: str = "gray") -> bytes:
    """The whole map wrapped into `rows` strips stacked top to bottom, `width` px wide, 6 px gaps."""
    W, H = D["W"], D["H"]
    seg = math.ceil(W / rows)
    scale = width / seg
    k = max(0, int(math.floor(math.log2(1 / scale))))
    src = D["pyr"][k]
    rh = max(1, round(H * scale))
    out = np.full((rows * rh + (rows - 1) * 6, width), 18, np.uint8)
    for i in range(rows):
        a = src[:, (i * seg) >> k:((i + 1) * seg) >> k]
        im = Image.fromarray(a).resize((max(1, round(a.shape[1] * width / (seg >> k))), rh), Image.BILINEAR)
        b = np.asarray(im)
        out[i * (rh + 6):i * (rh + 6) + rh, :b.shape[1]] = b[:, :width]
    if cmap != "gray":
        return encode(lut(cmap)[out], False, "png")
    return encode(out, False, "png")


def info() -> dict:
    return {
        "segment": SEG_ID, "scroll": "PHerc. Paris 4", "width": D["W"], "height": D["H"], "mm_per_px": MM_PER_PX,
        "dzi_levels": D["top"] + 1, "tile": TILE, "grid": list(D["wind"].shape), "mm_per_grid": MM_PER_GRID,
        "scan_z_mm": [round(v, 2) for v in D["z_mm"]], "windings": D["windings"],
        "orientation": "left is the outer end (winding 135), right the inner end (winding 10); up is the top of the "
                       "scroll (high scan z). The map is the fit rotated 180 degrees, as the organizers published it.",
        "model": "student-v12A-25000 (the organizers' 3D ink model), layers 52 to 76, max projection",
        "ct": {"zarr": CT_ZARR, "level": CT_LEVEL, "um_per_px": 38.4, "crop": ct_crop_box()},
        "sources": [{"id": k, "title": s["title"], "url": url(k), "bytes": state["bytes"].get(k),
                     "sha256": state["sha256"].get(k)} for k, s in SOURCES.items()],
        "stream": stream_urls(),
        "license": "Vesuvius Challenge open data, CC BY-NC 4.0; the ink map and spiral fit are the organizers'",
        "note": "Model output, not a reading.",
    }


def stream_urls() -> dict:
    return {"ome_zarr": f"{SPACE_URL}/zarr/", "dzi": f"{SPACE_URL}/tiles/ink.dzi", "api": f"{SPACE_URL}/api",
            "api_docs": f"{SPACE_URL}/docs", "mcp": f"{SPACE_URL}/mcp",
            "original_tif": url("ink"), "vc3d_project": f"{SPACE_URL}/api/vc3d.zip", "ct_zarr": CT_ZARR}


# ---------------------------------------------------------------- neighboring turns

def _window(cx: float, cy: float, half: int, k: int) -> np.ndarray:
    """A square of the level-k pyramid around map pixel (cx, cy), half-size `half` level-k pixels, zero-padded."""
    a = D["pyr"][k]
    h, w = a.shape
    x, y = int(round(cx / 2 ** k)), int(round(cy / 2 ** k))
    out = np.zeros((2 * half, 2 * half), np.float32)
    x0, y0 = x - half, y - half
    sx0, sy0, sx1, sy1 = max(0, x0), max(0, y0), min(w, x0 + 2 * half), min(h, y0 + 2 * half)
    if sx1 > sx0 and sy1 > sy0:
        out[sy0 - y0:sy1 - y0, sx0 - x0:sx1 - x0] = a[sy0:sy1, sx0:sx1]
    return out


def _best_ncc(a: np.ndarray, b: np.ndarray, shift: int) -> float:
    """The highest normalized cross-correlation of a and b over shifts up to `shift` pixels in each axis."""
    a = a - a.mean()
    b = b - b.mean()
    na, nb = np.sqrt((a * a).sum()), np.sqrt((b * b).sum())
    if na == 0 or nb == 0:
        return 0.0
    fa, fb = np.fft.rfft2(a, s=(a.shape[0] * 2, a.shape[1] * 2)), np.fft.rfft2(b, s=(a.shape[0] * 2, a.shape[1] * 2))
    cc = np.fft.irfft2(fa * np.conj(fb))
    sl = np.r_[0:shift + 1, cc.shape[0] - shift:cc.shape[0]]
    sc = np.r_[0:shift + 1, cc.shape[1] - shift:cc.shape[1]]
    return float(cc[np.ix_(sl, sc)].max() / (na * nb))


def neighbors(x: float, y: float, half_mm: float = 12.0, shift_mm: float = 3.0, n_null: int = 24) -> dict:
    """The papyrus physically next to map pixel (x, y): on the turns just outside and inside it at the same scan
    height, the crossing nearest in the scan. For each, the map position, the distance through the scroll, and how
    alike its ink is to this place (best normalized cross-correlation within `shift_mm`). Neighboring turns are
    alike everywhere to some degree, so each score comes with a baseline: the same score for the same pair of turns
    at up to `n_null` other places along them at this height. A score above that baseline's 95th percentile marks a
    place where ink may come from the neighboring turn, or from where the fit meets it, rather than from this one."""
    p = point(x, y)
    w = p["winding"]
    if w is None or "voxel_9_6um_xyz" not in p:
        raise HTTPException(400, "no single winding at this pixel: pick a place on the mesh")
    z = p["scan_z_mm"]
    s = section(z, 1)
    lv = 2 ** (CT_LEVEL - 2)
    px, py = p["voxel_9_6um_xyz"][0] / lv, p["voxel_9_6um_xyz"][1] / lv
    cx, cy = np.array(s["x"]), np.array(s["y"])
    wv, mx, my = np.array(s["winding"]), np.array(s["map_x"]), np.array(s["map_y"])
    um = 2.4 * 2 ** CT_LEVEL
    k = 1                                           # compare at 0.154 mm per pixel
    half = int(round(half_mm / (MM_PER_PX * 2 ** k)))
    shift = int(round(shift_mm / (MM_PER_PX * 2 ** k)))

    def across(ax, ay, ww):
        """Index of the crossing on winding ww nearest to CT point (ax, ay), and its distance in CT pixels."""
        m = np.nonzero(wv == ww)[0]
        if not len(m):
            return None, None
        d = np.hypot(cx[m] - ax, cy[m] - ay)
        j = int(np.argmin(d))
        if d[j] * um / 1000 > NEAR_MM:              # a gap in the fit: the nearest crossing is not the next turn
            return None, float(d[j])
        return int(m[j]), float(d[j])

    here = _window(x, y, half, k)
    # places along this turn at this height, at least one window away from here, for the baselines
    own = np.nonzero(wv == w)[0]
    own = own[np.abs(mx[own] - x) > 2 * half * 2 ** k]
    if len(own) > n_null:
        own = own[np.linspace(0, len(own) - 1, n_null).astype(int)]
    out = {"x": p["x"], "y": p["y"], "winding": w, "scan_z_mm": z, "half_mm": half_mm, "shift_mm": shift_mm,
           "window_px": 2 * half * 2 ** k, "turns": []}
    for side, ww in (("outer", w + 1), ("inner", w - 1)):
        i, d = across(px, py, ww)
        if i is None:
            t = {"side": side, "winding": ww, "found": False}
            if d is not None:
                t["nearest_mm"] = round(d * um / 1000, 3)
            out["turns"].append(t)
            continue
        nx, ny = int(mx[i]), int(my[i])
        t = {"side": side, "winding": ww, "found": True, "x": nx, "y": ny, "distance_mm": round(d * um / 1000, 3),
             "similarity": round(_best_ncc(here, _window(nx, ny, half, k), shift), 3)}
        base = []
        for o in own:
            j, _ = across(cx[o], cy[o], ww)
            if j is not None:
                base.append(_best_ncc(_window(int(mx[o]), int(my[o]), half, k), _window(int(mx[j]), int(my[j]), half, k),
                                      shift))
        if base:
            t["baseline"] = {"n": len(base), "median": round(float(np.median(base)), 3),
                             "p95": round(float(np.percentile(base, 95)), 3)}
            t["above_baseline"] = t["similarity"] > t["baseline"]["p95"]
        out["turns"].append(t)
    return out


# ---------------------------------------------------------------- VC3D bundle

NEAR_MM = 3.0                # a neighboring turn farther than this through the scroll counts as not found
VC3D_NAME = "paris4-spiral-output-generation-19"
_vc3d_lock = threading.Lock()


def vc3d_bundle() -> Path:
    """A zip that VC3D opens as a project: a .volpkg.json naming the 2.4 um CT (read remotely from the open data)
    and the fit as a tifxyz segment in that scan's 2.4 um voxels (the published normalized mesh times 4, every
    second node, so 0.384 mm between nodes). Built once, kept in CACHE."""
    path = CACHE / f"{VC3D_NAME}-vc3d.zip"
    with _vc3d_lock:
        if path.exists():
            return path
        import tifffile
        seg = f"{VC3D_NAME}-2.4um"
        meta = {"uuid": seg, "name": seg, "type": "seg", "format": "tifxyz",
                "scale": [0.05 / XYZ_STRIDE / 4, 0.05 / XYZ_STRIDE / 4],
                "source": url("x").rsplit("/", 1)[0] + "/",
                "note": "The organizers' spiral fit of PHerc. Paris 4 (tifxyz_normalized, 9.6 um voxels of scan "
                        "20260411134726), every second node, coordinates times 4 for the 2.4 um scan. Invalid "
                        "nodes are -1. The ink map rendered on this fit is the organizers'; CC BY-NC 4.0."}
        grids = {}
        for k in "xyz":
            g = D[k.upper()] * 4.0
            grids[k] = np.where(np.isnan(g), -1.0, g).astype(np.float32)
        valid = grids["x"] >= 0
        lo = [float(grids[k][valid].min()) for k in "xyz"]
        hi = [float(grids[k][valid].max()) for k in "xyz"]
        meta["bbox"] = [lo, hi]
        project = {"name": "PHerc. Paris 4, spiral-output-generation-19", "version": 1,
                   "volumes": [CT_ZARR], "segments": ["segments"], "output_segments": "segments"}
        readme = (
            "PHerc. Paris 4, spiral-output-generation-19, for VC3D\n\n"
            "1. Unzip this folder anywhere.\n"
            f"2. In VC3D: File > Open Project..., pick {VC3D_NAME}.volpkg.json.\n"
            "   The CT (2.4 um, 78 keV, scan 20260411134726) streams from the Vesuvius Challenge open data;\n"
            f"   the segment {seg} is the organizers' spiral fit in that scan's voxels.\n\n"
            "Without the project file: File > New Project, then File > Attach Remote Zarr... with\n"
            f"{CT_ZARR}\n"
            "and File > Attach Segments... with the segments folder.\n\n"
            "Made by the Paris 4 Ink Map Space (https://huggingface.co/spaces/SharpMeow/paris4-ink-map).\n"
            "Data: Vesuvius Challenge open data, CC BY-NC 4.0. An ink map is model output, not a reading.\n")
        tmp = path.with_suffix(".part")
        CACHE.mkdir(parents=True, exist_ok=True)
        with zipfile.ZipFile(tmp, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as z:
            root = f"{VC3D_NAME}-vc3d"
            z.writestr(f"{root}/README.txt", readme)
            z.writestr(f"{root}/{VC3D_NAME}.volpkg.json", json.dumps(project, indent=2))
            z.writestr(f"{root}/segments/{seg}/meta.json", json.dumps(meta, indent=2))
            for k in "xyz":
                b = io.BytesIO()
                tifffile.imwrite(b, grids[k])
                z.writestr(f"{root}/segments/{seg}/{k}.tif", b.getvalue())
        tmp.replace(path)
        return path


# ---------------------------------------------------------------- OME-Zarr view

def zarr_levels() -> list:
    """Pyramid levels served as OME-Zarr datasets: full resolution down to the last one at least ZCHUNK wide."""
    return [k for k, a in enumerate(D["pyr"]) if k == 0 or max(a.shape) >= ZCHUNK]


def zarray(k: int) -> dict:
    h, w = D["pyr"][k].shape
    return {"zarr_format": 2, "shape": [h, w], "chunks": [ZCHUNK, ZCHUNK], "dtype": "|u1",
            "compressor": {"id": "zlib", "level": 1}, "fill_value": 0, "order": "C", "filters": None,
            "dimension_separator": "."}


def zattrs() -> dict:
    ds = []
    for k in zarr_levels():
        f = 2 ** k
        ds.append({"path": str(k), "coordinateTransformations": [
            {"type": "scale", "scale": [MM_PER_PX * f, MM_PER_PX * f]},
            {"type": "translation", "translation": [MM_PER_PX * (f - 1) / 2] * 2}]})
    return {"multiscales": [{"version": "0.4", "name": f"PHerc. Paris 4 ink map ({SEG_ID})", "type": "mean",
                             "axes": [{"name": "y", "type": "space", "unit": "millimeter"},
                                      {"name": "x", "type": "space", "unit": "millimeter"}],
                             "datasets": ds}],
            "license": "Vesuvius Challenge open data, CC BY-NC 4.0; the ink map is the organizers'",
            "source": url("ink"), "note": "Model output, not a reading. Level 0 is the published map, 0.0768 mm per "
                                          "pixel; each level halves it by the mean of 2 by 2 pixels."}


def zchunk(k: int, cy: int, cx: int) -> bytes:
    a = D["pyr"][k]
    h, w = a.shape
    if not (0 <= cy * ZCHUNK < h and 0 <= cx * ZCHUNK < w):
        raise HTTPException(404, "no such chunk")
    part = a[cy * ZCHUNK:(cy + 1) * ZCHUNK, cx * ZCHUNK:(cx + 1) * ZCHUNK]
    if part.shape != (ZCHUNK, ZCHUNK):
        full = np.zeros((ZCHUNK, ZCHUNK), np.uint8)
        full[:part.shape[0], :part.shape[1]] = part
        part = full
    return zlib.compress(np.ascontiguousarray(part).tobytes(), 1)


# ---------------------------------------------------------------- web

app = FastAPI(
    title="Paris 4 Ink Map",
    version="1.2.0",
    description="The Vesuvius Challenge organizers' whole-scroll ink map of PHerc. Paris 4, streamed: deep-zoom "
                "tiles, an OME-Zarr view, crops of any size, map pixel to scan coordinates, and the spiral at any "
                "height. No account. Data: Vesuvius Challenge open data, CC BY-NC 4.0. An ink map is model output, "
                "not a reading. MCP for agents: POST /mcp.")
# open to every origin, so neuroglancer, notebooks and other pages can stream the tiles and the zarr
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "HEAD", "POST", "OPTIONS"],
                   allow_headers=["*"], expose_headers=["Content-Length", "Content-Disposition"])
JSON_CACHE = {"Cache-Control": "public, max-age=3600"}
API = [
    ("GET", "/api/info", "Size, scale, orientation, windings, stream URLs and every source with its SHA-256"),
    ("GET", "/api/point?x=71000&y=1200", "One map pixel: winding, distance from each end, scan height and voxel"),
    ("GET", "/api/winding/60", "One winding's pixel range"),
    ("GET", "/api/crop?x0=70000&y0=600&x1=72000&y1=1800&scale=1&fmt=tif", "A region: PNG up to 60 Mpx, TIFF of any size"),
    ("GET", "/api/crop?x0=0&y0=0&x1=142283&y1=2390&scale=1&fmt=tif", "The whole map at full resolution as a tiled TIFF"),
    ("GET", "/api/section?z_mm=105&step=2", "Where the sheet crosses one scan height"),
    ("GET", "/api/ct.png?z_mm=105", "The CT slice at one height (level 4, 38.4 um), cropped to the scroll"),
    ("GET", "/api/overview.png?rows=4&width=1600", "The whole map wrapped into rows"),
    ("GET", "/api/profile", "Mean ink per 16-pixel column bin"),
    ("GET", "/api/neighbors?x=120000&y=1800", "The turns just outside and inside one pixel: where, how far, how alike"),
    ("GET", "/api/vc3d.zip", "A VC3D project: the 2.4 um CT streamed from the open data plus the fit as a segment"),
    ("GET", "/api/mesh3d.json?stride=8", "The fit's mesh as a point cloud: layout of /api/mesh3d.bin (stride 4, 8 or 16)"),
    ("GET", "/tiles/ink.dzi", "Deep Zoom descriptor; tiles at /tiles/ink_files/{level}/{col}_{row}.jpg"),
    ("GET", "/zarr/", "OME-Zarr 0.4 view (zarr v2): /zarr/{level}/{row}.{col}, 512 px zlib chunks"),
    ("POST", "/mcp", "MCP server (Streamable HTTP, JSON replies): get_info, find_winding, point_info, crop_image, "
                     "cross_section, neighbor_turns, view_link"),
    ("GET", "/docs", "Interactive API docs (OpenAPI)"),
]


@app.get("/api", tags=["about"])
def api_index():
    """Every endpoint with an example, and the stream URLs."""
    return {"endpoints": [{"method": m, "path": p, "url": SPACE_URL + p, "what": w} for m, p, w in API],
            "stream": stream_urls(), "openapi": f"{SPACE_URL}/openapi.json",
            "license": "Vesuvius Challenge open data, CC BY-NC 4.0", "note": "Model output, not a reading."}


@app.get("/api/status", tags=["about"])
def api_status():
    return {k: state[k] for k in ("phase", "done_bytes", "total_bytes", "error", "ready")} | {
        "seconds": round(time.time() - state["started"], 1)}


@app.get("/api/info", tags=["about"])
def api_info():
    need()
    return info()


@app.get("/tiles/{name}.dzi", tags=["stream"])
def dzi(name: str):
    need()
    if name not in ("ink", "wind"):
        raise HTTPException(404)
    fmt = "jpg" if name == "ink" else "png"
    xml = (f'<?xml version="1.0" encoding="UTF-8"?><Image xmlns="http://schemas.microsoft.com/deepzoom/2008" '
           f'Format="{fmt}" Overlap="0" TileSize="{TILE}"><Size Width="{D["W"]}" Height="{D["H"]}"/></Image>')
    return Response(xml, media_type="application/xml", headers={"Cache-Control": "public, max-age=3600"})


@app.get("/tiles/{name}_files/{level}/{xy}", tags=["stream"])
def tile(name: str, level: int, xy: str, cmap: str = "gray", gamma: float = Query(1.0, ge=0.2, le=5)):
    need()
    try:
        stem = xy.rsplit(".", 1)[0]
        col, row = (int(v) for v in stem.split("_"))
    except ValueError as exc:
        raise HTTPException(404) from exc
    a, rgba = tile_array(name, level, col, row, cmap, gamma)
    body = encode(a, rgba, "png" if rgba else "jpg")
    return Response(body, media_type="image/png" if rgba else "image/jpeg",
                    headers={"Cache-Control": "public, max-age=86400"})


@app.get("/api/overview.png", tags=["images"])
def api_overview(rows: int = Query(4, ge=1, le=12), width: int = Query(1600, ge=200, le=4000), cmap: str = "gray"):
    need()
    with busy():
        png = overview(rows, width, cmap)
    return Response(png, media_type="image/png", headers={"Cache-Control": "public, max-age=3600"})


@app.get("/api/profile", tags=["map"])
def api_profile():
    need()
    return {"bin_px": 16, "values": D["profile"]}


@app.get("/api/point", tags=["map"])
def api_point(x: float, y: float):
    need()
    return point(x, y)


@app.get("/api/winding/{w}", tags=["map"])
def api_winding(w: int):
    need()
    return winding_info(w)


@app.get("/api/section", tags=["scan"])
def api_section(z_mm: float = Query(..., ge=0, le=200), step: int = Query(2, ge=1, le=16)):
    need()
    with busy():
        return section(z_mm, step)


@app.get("/api/ct.png", tags=["scan"])
def api_ct(z_mm: float = Query(..., ge=0, le=200)):
    need()
    zi = int(round(z_mm / (2.4e-3 * 2 ** CT_LEVEL)))
    if not 0 <= zi < CT_SHAPE[0]:
        raise HTTPException(400, "outside the scan")
    png = ct_slice(zi)
    # a slice with a plane that could not be read is not kept, here or in the browser
    keep = "public, max-age=86400" if zi in _ct_cache else "no-store"
    return Response(png, media_type="image/png", headers={"Cache-Control": keep})


_tif_lock = threading.Lock()
_tif_live = 0
TIF_MAX_LIVE = 3             # TIFF downloads on disk and streaming at once


def _tif_done(path):
    global _tif_live
    if path:
        try:
            os.unlink(path)
        except OSError:
            pass
    with _tif_lock:
        _tif_live -= 1


class _OnceFileResponse(FileResponse):
    """A FileResponse that runs `done` however the response ends: starlette returns early on a bad or unsatisfiable
    Range header without running a background task, which would keep a download slot and a temporary file."""

    def __init__(self, *a, done, **k):
        super().__init__(*a, **k)
        self._done = done

    async def __call__(self, scope, receive, send):
        try:
            await super().__call__(scope, receive, send)
        finally:
            self._done()


@app.get("/api/crop", tags=["download"])
def api_crop(x0: int, y0: int, x1: int, y1: int, scale: int = Query(1, ge=1, le=262144), fmt: str = "png",
             cmap: str = "gray"):
    """A region of the map, reduced by `scale` (1, 2, 4 ... a power of two). fmt=png applies `cmap` and holds at most
    60 Mpx; fmt=tif is the raw map (0 to 255, gray), tiled and deflated, of any size up to the whole map at full
    resolution. Coordinates are map pixels, in either order, clamped to the map."""
    need()
    name = f"PHercParis4-spiral-output-generation-19-x{x0}-{x1}-y{y0}-{y1}-s{scale}"
    if fmt == "tif":
        import tifffile
        global _tif_live
        with _tif_lock:
            if _tif_live >= TIF_MAX_LIVE:
                raise HTTPException(503, "busy: other large downloads are streaming; try again in a moment")
            _tif_live += 1
        path = None
        try:
            with busy():
                a = crop(x0, y0, x1, y1, scale, "gray", limit=None)
                fd, path = tempfile.mkstemp(suffix=".tif")
                os.close(fd)
                tifffile.imwrite(path, a, tile=(256, 256) if min(a.shape) >= 256 else None,
                                 compression="zlib", compressionargs={"level": 1}, bigtiff=a.size > 3_000_000_000,
                                 metadata={"segment": SEG_ID, "x0": x0, "y0": y0, "scale": scale,
                                           "mm_per_px": MM_PER_PX * scale})
        except BaseException:
            _tif_done(path)          # also frees the slot when the crop is refused or the server is busy
            raise
        return _OnceFileResponse(path, media_type="image/tiff", filename=f"{name}.tif", done=lambda: _tif_done(path))
    with busy():
        png = encode(crop(x0, y0, x1, y1, scale, cmap), False, "png")
    return Response(png, media_type="image/png", headers={"Content-Disposition": f'attachment; filename="{name}.png"'})


@app.get("/api/neighbors", tags=["map"])
def api_neighbors(x: float, y: float, half_mm: float = Query(12.0, ge=2, le=40), shift_mm: float = Query(3.0, ge=0, le=10)):
    """The turns just outside and inside map pixel (x, y) at the same scan height: where they are on the map, how far
    away through the scroll, and how alike their ink is to this place, against other turns at that height."""
    need()
    with busy():
        return neighbors(x, y, half_mm, shift_mm)


@app.get("/api/vc3d.zip", tags=["download"])
def api_vc3d():
    """A VC3D project: the 2.4 um CT streamed from the open data, and this map's spiral fit as a tifxyz segment in
    that scan's voxels. Unzip, then File > Open Project in VC3D."""
    need()
    path = CACHE / f"{VC3D_NAME}-vc3d.zip"
    if not path.exists():
        with busy():
            path = vc3d_bundle()
    return FileResponse(path, media_type="application/zip", filename=f"{VC3D_NAME}-vc3d.zip",
                        headers={"Cache-Control": "public, max-age=86400"})


@app.api_route("/zarr", methods=["GET", "HEAD"], tags=["stream"])
@app.api_route("/zarr/", methods=["GET", "HEAD"], include_in_schema=False)
def zarr_root():
    """The OME-Zarr view: open it with zarr, ome-zarr-py, napari or neuroglancer (zarr://URL)."""
    need()
    return JSONResponse({"zarr": f"{SPACE_URL}/zarr/", "levels": len(zarr_levels()), "metadata": ".zattrs",
                         "python": "zarr.open_consolidated(fsspec.get_mapper('https://sharpmeow-paris4-ink-map.hf.space/zarr/'), mode='r')"},
                        headers=JSON_CACHE)


@app.api_route("/zarr/.zgroup", methods=["GET", "HEAD"], tags=["stream"])
def zarr_group():
    return JSONResponse({"zarr_format": 2}, headers=JSON_CACHE)


@app.api_route("/zarr/.zattrs", methods=["GET", "HEAD"], tags=["stream"])
def zarr_attrs():
    """OME-NGFF 0.4 multiscales: axes y and x in millimeters, one dataset per level."""
    need()
    return JSONResponse(zattrs(), headers=JSON_CACHE)


@app.api_route("/zarr/.zmetadata", methods=["GET", "HEAD"], tags=["stream"])
def zarr_meta():
    """Consolidated metadata, so a reader needs one request before the chunks."""
    need()
    meta = {".zgroup": {"zarr_format": 2}, ".zattrs": zattrs()}
    for k in zarr_levels():
        meta[f"{k}/.zarray"] = zarray(k)
        meta[f"{k}/.zattrs"] = {}
    return JSONResponse({"zarr_consolidated_format": 1, "metadata": meta}, headers=JSON_CACHE)


def _zlevel(level: str) -> int:
    need()
    if not level.isdigit() or int(level) not in zarr_levels():
        raise HTTPException(404, "no such level")
    return int(level)


@app.api_route("/zarr/{level}/.zarray", methods=["GET", "HEAD"], tags=["stream"])
def zarr_array(level: str):
    return JSONResponse(zarray(_zlevel(level)), headers=JSON_CACHE)


@app.api_route("/zarr/{level}/.zattrs", methods=["GET", "HEAD"], tags=["stream"])
def zarr_array_attrs(level: str):
    _zlevel(level)
    return JSONResponse({}, headers=JSON_CACHE)


@app.api_route("/zarr/{level}/{key}", methods=["GET", "HEAD"], tags=["stream"])
def zarr_chunk(level: str, key: str):
    """One 512 by 512 chunk, uint8, zlib; key is row.col."""
    k = _zlevel(level)
    try:
        cy, cx = (int(v) for v in key.split("."))
    except ValueError as exc:
        raise HTTPException(404, "chunk keys are row.col") from exc
    return Response(zchunk(k, cy, cx), media_type="application/octet-stream",
                    headers={"Cache-Control": "public, max-age=86400"})


@app.get("/", include_in_schema=False)
def index():
    return FileResponse(HERE / "index.html", headers={"Cache-Control": "no-cache"})


app.mount("/static", StaticFiles(directory=HERE / "static"), name="static")


@app.on_event("startup")
def _start():
    if not os.environ.get("P4_NO_LOAD"):
        threading.Thread(target=load, daemon=True).start()


import p4_mcp  # noqa: E402  (needs the functions above)

app.include_router(p4_mcp.router)
import scroll3d  # noqa: E402

app.include_router(scroll3d.router)