"""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'') 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)