Spaces:
Running
Running
Download app.py from SharpMeow/segment-lens: direct link, hf CLI and curl.
- Browser
- Download file 18.2 kB
-
https://huggingface.co/spaces/SharpMeow/segment-lens/resolve/main/app.py
- Command line
-
hf download hf://spaces/SharpMeow/segment-lens/app.py
-
curl -L -o app.py https://huggingface.co/spaces/SharpMeow/segment-lens/resolve/main/app.py
18.2 kB
| """Segment Lens on Hugging Face: the 3D page (index.html, app.js) at /, and the same atlas for AI agents and scripts | |
| at /agents, as API endpoints (gradio_client) and MCP tools (/agents/gradio_api/mcp/). | |
| Everything comes from the Segment Atlas dataset (SharpMeow/vesuvius-segment-atlas): atlas.jsonl, one row per | |
| published tifxyz mesh, and lens/*.slg.gz, the decimated previews (format: "Lens files" in docs/tools/atlas.md of | |
| ChaseHendrick/Scrolls). Nothing is written and no scan data is read. Previews exist only for scrolls on no prize | |
| list, as on the page. | |
| python3 app.py # serves on port 7860 (SEGMENT_LENS_DATA=URL/ reads another copy of the atlas) | |
| """ | |
| import functools | |
| import gzip | |
| import io | |
| import json | |
| import math | |
| import os | |
| import struct | |
| import threading | |
| import time | |
| import urllib.request | |
| from collections import OrderedDict | |
| from pathlib import Path | |
| VERSION = "1.1.0" | |
| HERE = Path(__file__).resolve().parent | |
| DATA = os.environ.get("SEGMENT_LENS_DATA", | |
| "https://huggingface.co/datasets/SharpMeow/vesuvius-segment-atlas/resolve/main/") | |
| S3 = "https://vesuvius-challenge-open-data.s3.us-east-1.amazonaws.com/" | |
| MAGIC = b"SLG1" | |
| ATLAS_TTL_S = 3600 | |
| LENS_CACHE = 32 | |
| SORTS = ("area_cm2", "filled_share", "within_10pct", "holes_mm2", "folds_mm2", "stretched_mm2", "grid_h", "segment") | |
| # the page's colors (app.js COLORS), as 0-255 RGB: plain, hole, fold, stretched | |
| COLORS = {"base": (204, 214, 219), "hole": (89, 166, 255), "fold": (255, 92, 92), "str": (255, 166, 51)} | |
| WEB_FILES = ("index.html", "app.js", "style.css", "connect.html", "connect.json", "CONNECT.md", "onescroll.html") | |
| SUMMARY_KEYS = ("id", "scroll", "segment", "mesh", "kind", "volume", "voxel_um", "area_cm2", "filled_share", | |
| "within_10pct", "holes_mm2", "folds_mm2", "stretched_mm2", "clean", "prize_lists") | |
| class LensError(ValueError): | |
| """A refusal an agent can act on: unknown id, no preview, bad argument.""" | |
| # ---------------------------------------------------------------- data | |
| def _get(url, timeout=60): | |
| req = urllib.request.Request(url, headers={"User-Agent": f"segment-lens/{VERSION}"}) | |
| with urllib.request.urlopen(req, timeout=timeout) as r: | |
| return r.read() | |
| _atlas = {"rows": None, "at": 0.0} | |
| _atlas_lock = threading.Lock() | |
| def atlas_rows(get=_get): | |
| """The atlas rows, fetched at most once an hour.""" | |
| with _atlas_lock: | |
| if _atlas["rows"] is None or time.time() - _atlas["at"] > ATLAS_TTL_S: | |
| text = get(DATA + "atlas.jsonl").decode("utf-8") | |
| _atlas["rows"] = [json.loads(line) for line in text.splitlines() if line.strip()] | |
| _atlas["at"] = time.time() | |
| return _atlas["rows"] | |
| def find_row(segment_id, rows=None): | |
| rows = atlas_rows() if rows is None else rows | |
| segment_id = (segment_id or "").strip() | |
| for r in rows: | |
| if r["id"] == segment_id: | |
| return r | |
| raise LensError(f"no mesh with id {segment_id!r}; list_segments gives the ids") | |
| def parse_lens(blob): | |
| """(h, w, k, xyz (h, w, 3) float64 with NaN for missing nodes, flags (h, w) uint8) from lens bytes: the reader | |
| of kit.atlas.read_lens and the page's parseLens.""" | |
| import numpy as np | |
| raw = gzip.decompress(blob) | |
| if raw[:4] != MAGIC: | |
| raise LensError("not a Segment Lens grid (bad magic)") | |
| h, w, k = struct.unpack("<HHH", raw[4:10]) | |
| v = struct.unpack("<6f", raw[10:34]) | |
| lo, hi = np.array(v[:3]), np.array(v[3:]) | |
| n = h * w | |
| if len(raw) != 34 + 7 * n: | |
| raise LensError("Segment Lens grid has the wrong length") | |
| q = np.frombuffer(raw, "<u2", 3 * n, 34).reshape(3, h, w) | |
| flags = np.frombuffer(raw, np.uint8, n, 34 + 6 * n).reshape(h, w) | |
| span = np.where(hi - lo > 0, hi - lo, 1.0) | |
| xyz = np.stack([lo[i] + q[i] / 65534.0 * span[i] for i in range(3)], -1) | |
| xyz[q[0] == 65535] = np.nan | |
| return h, w, k, xyz, flags | |
| _lens = OrderedDict() | |
| _lens_lock = threading.Lock() | |
| def load_lens(row, get=_get): | |
| if not row.get("lens"): | |
| why = ("its scroll is on a prize list, so it has numbers but no preview (public media rule)" | |
| if row.get("prize_lists") else "it was not measured or has no filled nodes") | |
| raise LensError(f"{row['id']} has no preview: {why}") | |
| with _lens_lock: | |
| if row["lens"] in _lens: | |
| _lens.move_to_end(row["lens"]) | |
| return _lens[row["lens"]] | |
| g = parse_lens(get(DATA + "lens/" + row["lens"])) | |
| with _lens_lock: | |
| _lens[row["lens"]] = g | |
| while len(_lens) > LENS_CACHE: | |
| _lens.popitem(last=False) | |
| return g | |
| def flag_counts(flags): | |
| f = flags.ravel() | |
| return {"nodes": int(f.size), "hole": int(((f & 1) > 0).sum()), "fold": int(((f & 2) > 0).sum()), | |
| "stretched": int(((f & 4) > 0).sum())} | |
| def node_colors(flags): | |
| """One RGB per preview node, with the page's precedence: fold over stretched over hole over plain.""" | |
| import numpy as np | |
| c = np.empty(flags.shape + (3,), np.uint8) | |
| c[:] = COLORS["base"] | |
| c[(flags & 1) > 0] = COLORS["hole"] | |
| c[(flags & 4) > 0] = COLORS["str"] | |
| c[(flags & 2) > 0] = COLORS["fold"] | |
| return c | |
| # ---------------------------------------------------------------- pictures | |
| def render_grid(flags, xyz, size=512): | |
| """The preview grid flat (rows down, columns across), each node colored by its flag; missing nodes black.""" | |
| import numpy as np | |
| from PIL import Image | |
| c = node_colors(flags) | |
| c[np.isnan(xyz[..., 0])] = (12, 16, 20) | |
| h, w = flags.shape | |
| s = max(1, size // max(h, w)) | |
| return Image.fromarray(np.repeat(np.repeat(c, s, 0), s, 1)) | |
| def render_3d(flags, xyz, azimuth=30.0, elevation=20.0, size=512): | |
| """An orthographic picture of the surface from (azimuth, elevation) degrees, the scan's z up: each grid cell | |
| filled as two triangles' worth of points, shaded by its facing, colored by flag; nearest wins.""" | |
| import numpy as np | |
| from PIL import Image | |
| ok = ~np.isnan(xyz[..., 0]) | |
| if ok.sum() < 4: | |
| raise LensError("the preview has too few nodes to draw") | |
| # x, y across, z up (the page rotates the same way) | |
| p = xyz[..., [0, 1, 2]].copy() | |
| center = np.nanmean(p[ok], 0) | |
| p -= center | |
| scale = np.nanmax(np.abs(p[ok])) or 1.0 | |
| p /= scale | |
| a, e = math.radians(azimuth), math.radians(elevation) | |
| right = np.array([math.cos(a), math.sin(a), 0.0]) | |
| view = np.array([-math.sin(a) * math.cos(e), math.cos(a) * math.cos(e), math.sin(e)]) | |
| up = np.cross(right, view) | |
| # normals from the grid's neighbors, for shading | |
| dy = np.gradient(np.nan_to_num(p), axis=0) | |
| dx = np.gradient(np.nan_to_num(p), axis=1) | |
| nrm = np.cross(dx, dy) | |
| nrm /= np.linalg.norm(nrm, axis=-1, keepdims=True) + 1e-9 | |
| light = np.abs(nrm @ view) * 0.75 + 0.25 | |
| # densify: 4 x 4 samples per cell, bilinear between the cell's corners | |
| h, w = flags.shape | |
| t = np.linspace(0, 1, 4, endpoint=False) | |
| rr, cc = np.meshgrid(t, t, indexing="ij") | |
| P00, P01, P10, P11 = p[:-1, :-1], p[:-1, 1:], p[1:, :-1], p[1:, 1:] | |
| cell_ok = ok[:-1, :-1] & ok[:-1, 1:] & ok[1:, :-1] & ok[1:, 1:] | |
| pts = (P00[..., None, None, :] * ((1 - rr) * (1 - cc))[..., None] + P01[..., None, None, :] * ((1 - rr) * cc)[..., None] | |
| + P10[..., None, None, :] * (rr * (1 - cc))[..., None] + P11[..., None, None, :] * (rr * cc)[..., None]) | |
| pts = pts[cell_ok].reshape(-1, 3) | |
| col = (node_colors(flags)[:-1, :-1][cell_ok].astype(np.float32) * light[:-1, :-1][cell_ok][:, None]) | |
| col = np.repeat(col, 16, 0) | |
| u = pts @ right | |
| v = pts @ up | |
| d = pts @ view | |
| half = size / 2 * 0.92 | |
| px = np.clip((u * half + size / 2).astype(int), 0, size - 1) | |
| py = np.clip((size / 2 - v * half).astype(int), 0, size - 1) | |
| order = np.argsort(d) # far first, near last, so near overwrite | |
| img = np.full((size, size, 3), (5, 8, 11), np.uint8) | |
| img[py[order], px[order]] = np.clip(col[order], 0, 255).astype(np.uint8) | |
| return Image.fromarray(img) | |
| # ---------------------------------------------------------------- tools | |
| def page_url(segment_id=""): | |
| host = os.environ.get("SPACE_HOST") | |
| base = f"https://{host}/" if host else "/" | |
| return base + (f"#id={segment_id}" if segment_id else "") | |
| def kit_command(row): | |
| return f"python -m kit tifxyz check {row['url'].replace(S3, 's3://vesuvius-challenge-open-data/')} --png check.png" | |
| def summary(row): | |
| out = {k: row.get(k) for k in SUMMARY_KEYS} | |
| out["has_preview"] = bool(row.get("lens")) | |
| return out | |
| def list_scrolls(): | |
| """Every scroll in the Segment Atlas, with how many top-level meshes it has, how many have a 3D preview, and | |
| their total area. | |
| Returns: | |
| JSON: scrolls (scroll, meshes, previews, area_cm2, prize_lists), total_meshes, version, open_in_browser. | |
| """ | |
| rows = [r for r in atlas_rows() if r.get("kind") == "mesh"] | |
| by = {} | |
| for r in rows: | |
| s = by.setdefault(r["scroll"], {"scroll": r["scroll"], "meshes": 0, "previews": 0, "area_cm2": 0.0, | |
| "prize_lists": sorted(set(r.get("prize_lists") or []))}) | |
| s["meshes"] += 1 | |
| s["previews"] += bool(r.get("lens")) | |
| s["area_cm2"] = round(s["area_cm2"] + (r.get("area_cm2") or 0.0), 3) | |
| return {"scrolls": sorted(by.values(), key=lambda s: s["scroll"]), "total_meshes": len(rows), | |
| "version": VERSION, "open_in_browser": page_url()} | |
| def list_segments(scroll: str = "", sort: str = "area_cm2", top: int = 50, previews_only: bool = False): | |
| """The measured meshes of one scroll (or all scrolls), largest first by default: each mesh's id, area, filled | |
| share, isometry (share of edges within 10 percent) and the area of its holes, folds and stretched cells. | |
| Args: | |
| scroll: a scroll name as list_scrolls gives it, such as PHerc0139; empty for every scroll. | |
| sort: one of area_cm2, filled_share, within_10pct, holes_mm2, folds_mm2, stretched_mm2, grid_h, segment. | |
| Numbers sort largest first, segment names A to Z. | |
| top: how many rows to return, 1 to 500. | |
| previews_only: only meshes with a 3D preview (scrolls on no prize list). | |
| Returns: | |
| JSON: segments (one summary per mesh, with has_preview), count (before the top cut), and the call's settings. | |
| """ | |
| if sort not in SORTS: | |
| raise LensError(f"sort must be one of {', '.join(SORTS)}") | |
| top = int(top) | |
| if not 1 <= top <= 500: | |
| raise LensError("top must be 1 to 500") | |
| scroll = (scroll or "").strip() | |
| rows = [r for r in atlas_rows() if r.get("kind") == "mesh" and (not scroll or r["scroll"] == scroll)] | |
| if scroll and not rows: | |
| raise LensError(f"no meshes for scroll {scroll!r}; list_scrolls gives the names") | |
| if previews_only: | |
| rows = [r for r in rows if r.get("lens")] | |
| if sort == "segment": | |
| rows = sorted(rows, key=lambda r: r["segment"]) | |
| else: | |
| rows = sorted(rows, key=lambda r: -(r.get(sort) if r.get(sort) is not None else -math.inf)) | |
| return {"segments": [summary(r) for r in rows[:top]], "count": len(rows), "scroll": scroll or None, | |
| "sort": sort, "top": top} | |
| def get_segment(segment_id: str): | |
| """Everything the Segment Atlas knows about one mesh: its open-data folder, grid size, area, isometry, | |
| holes, folds and stretched cells (nodes and mm2), z range, the kit command that reproduces the numbers, and, | |
| when it has a preview, the preview's flag counts. | |
| Args: | |
| segment_id: the mesh's id, as list_segments gives it (SCROLL/SEGMENT/MESH). | |
| Returns: | |
| JSON: the atlas row, plus kit_command, preview (grid size and flag counts, or why there is none) and | |
| open_in_browser. | |
| """ | |
| row = find_row(segment_id) | |
| out = dict(row) | |
| out["kit_command"] = kit_command(row) if row.get("url") else None | |
| try: | |
| h, w, k, xyz, flags = load_lens(row) | |
| out["preview"] = {"grid_w": w, "grid_h": h, "every_kth_node": k, "flags": flag_counts(flags)} | |
| except LensError as exc: | |
| out["preview"] = {"none": str(exc)} | |
| out["open_in_browser"] = page_url(row["id"]) | |
| return out | |
| def segment_picture(segment_id: str, view: str = "3d", azimuth: float = 30.0, elevation: float = 20.0): | |
| """A picture of one mesh's preview with its flags painted on, as on the page: holes blue, folds red, | |
| stretched cells orange, plain gray. Geometry only: no CT, no ink, and a clean surface is not proof the trace | |
| stays on one sheet. Only meshes with a preview (scrolls on no prize list) can be drawn. | |
| Args: | |
| segment_id: the mesh's id, as list_segments gives it. | |
| view: "3d" for the surface seen from azimuth and elevation (the scan's z up), or "grid" for the flat grid. | |
| azimuth: degrees around the scan's z axis, for view 3d. | |
| elevation: degrees above the horizontal, -90 to 90, for view 3d. | |
| Returns: | |
| A PNG picture, and JSON: id, view, grid size, every_kth_node, flag counts, open_in_browser. | |
| """ | |
| if view not in ("3d", "grid"): | |
| raise LensError('view must be "3d" or "grid"') | |
| if not -90 <= float(elevation) <= 90: | |
| raise LensError("elevation must be -90 to 90") | |
| row = find_row(segment_id) | |
| h, w, k, xyz, flags = load_lens(row) | |
| img = render_grid(flags, xyz) if view == "grid" else render_3d(flags, xyz, float(azimuth), float(elevation)) | |
| info = {"id": row["id"], "view": view, "grid_w": w, "grid_h": h, "every_kth_node": k, | |
| "flags": flag_counts(flags), "open_in_browser": page_url(row["id"])} | |
| if view == "3d": | |
| info.update(azimuth=float(azimuth), elevation=float(elevation)) | |
| return img, info | |
| # ---------------------------------------------------------------- app | |
| AGENTS_TEXT = """\ | |
| # Segment Lens for agents and scripts | |
| The same Segment Atlas the [3D page](/) shows, as API endpoints and MCP tools. Every answer that names a mesh | |
| includes `open_in_browser`, a link that opens it on the page for a person. | |
| - `list_scrolls`: every scroll, its mesh and preview counts and total area. | |
| - `list_segments`: one scroll's meshes (or all), sorted, with area, filled share, isometry and flagged areas. | |
| - `get_segment`: one mesh's full atlas row, the `kit tifxyz check` command and the preview's flag counts. | |
| - `segment_picture`: a picture of one mesh with holes, folds and stretched cells painted on (3D or flat grid). | |
| **MCP**: add the server `{url}/agents/gradio_api/mcp/`. | |
| **Python**: | |
| ```python | |
| from gradio_client import Client | |
| c = Client("{url}/agents/") | |
| top = c.predict("PHerc0139", "area_cm2", 5, True, api_name="/list_segments") | |
| seg = c.predict(top["segments"][0]["id"], api_name="/get_segment") | |
| ``` | |
| Geometry only: no CT, no ink, and nothing here judges whether a segment stays on one sheet. Data: Vesuvius | |
| Challenge open data via the Segment Atlas, CC BY-NC 4.0. Segment Lens {version}. | |
| """ | |
| def fix_mcp_multiple_outputs(): | |
| """Gradio 6.28's MCP server joins a queued tool's several outputs into one text (the same patch as Scroll Lens), | |
| so segment_picture's image would reach an agent as text. Hand them over one by one.""" | |
| try: | |
| from gradio.mcp import GradioMCPServer | |
| from anyio.to_thread import run_sync | |
| except ImportError: | |
| return | |
| async def without_progress(self, job): | |
| result = await run_sync(job.result) | |
| return list(result) if isinstance(result, tuple) else [result] | |
| GradioMCPServer._execute_tool_without_progress = without_progress | |
| def build_demo(): | |
| import gradio as gr | |
| def guard(fn): | |
| # functools.wraps keeps the name, docstring and signature, which the API and the MCP tool list come from | |
| def run(*a, **kw): | |
| try: | |
| return fn(*a, **kw) | |
| except LensError as exc: | |
| raise gr.Error(str(exc)) | |
| return run | |
| url = page_url().rstrip("/") if page_url() != "/" else "https://sharpmeow-segment-lens.hf.space" | |
| with gr.Blocks(title="Segment Lens for agents") as demo: | |
| gr.Markdown(AGENTS_TEXT.format(url=url, version=VERSION)) | |
| with gr.Row(): | |
| sc = gr.Textbox(value="", label="scroll") | |
| so = gr.Dropdown(list(SORTS), value="area_cm2", label="sort") | |
| tp = gr.Number(value=50, precision=0, label="top") | |
| pv = gr.Checkbox(value=False, label="previews_only") | |
| sid = gr.Textbox(value="", label="segment_id") | |
| with gr.Row(): | |
| vw = gr.Dropdown(["3d", "grid"], value="3d", label="view") | |
| az = gr.Number(value=30.0, label="azimuth") | |
| el = gr.Number(value=20.0, label="elevation") | |
| with gr.Row(): | |
| b1, b2, b3, b4 = (gr.Button(n) for n in ("list_scrolls", "list_segments", "get_segment", "segment_picture")) | |
| pic = gr.Image(type="pil", label="picture", format="png") | |
| out = gr.JSON(label="answer") | |
| b1.click(guard(list_scrolls), None, out, api_name="list_scrolls") | |
| b2.click(guard(list_segments), [sc, so, tp, pv], out, api_name="list_segments") | |
| b3.click(guard(get_segment), [sid], out, api_name="get_segment") | |
| b4.click(guard(segment_picture), [sid, vw, az, el], [pic, out], api_name="segment_picture") | |
| return demo | |
| def build_app(): | |
| import gradio as gr | |
| from fastapi import FastAPI | |
| from fastapi.responses import FileResponse, RedirectResponse | |
| fix_mcp_multiple_outputs() | |
| app = FastAPI() | |
| # ssr_mode=False: on Spaces (GRADIO_SSR_MODE and GRADIO_SERVER_PORT set) server-side rendering starts a | |
| # Node server on port 7860, so uvicorn could not bind there and the Space stopped at startup | |
| app = gr.mount_gradio_app(app, build_demo(), path="/agents", mcp_server=True, footer_links=["api"], ssr_mode=False) | |
| def index(): | |
| return FileResponse(HERE / "index.html") | |
| def serve(name): | |
| def f(): | |
| return FileResponse(HERE / name) | |
| return f | |
| for name in WEB_FILES: | |
| app.add_api_route("/" + name, serve(name), methods=["GET"]) | |
| def agents(): | |
| return RedirectResponse("/agents/") | |
| return app | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(build_app(), host="0.0.0.0", port=int(os.environ.get("PORT", "7860"))) | |