Spaces:
Running
Running
| """projtex as a hosted, agent-first service: REST API (+ OpenAPI at /docs), remote MCP at /mcp, llms.txt. | |
| Long operations are asynchronous jobs: create one, then poll it. Every job status says what to do next. | |
| """ | |
| from __future__ import annotations | |
| import base64 | |
| import io | |
| import json | |
| import logging | |
| import math | |
| import os | |
| import time | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| from typing import Any, Literal | |
| import requests | |
| from fastapi import FastAPI, File, Form, HTTPException, Request, Response, UploadFile | |
| from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse | |
| from pydantic import BaseModel, Field, model_validator | |
| from starlette.concurrency import run_in_threadpool | |
| from .jobs import BACKENDS, MAX_FACES, MAX_MESH_BYTES, MAX_RUNNING, SAMPLE_INFO, TERRAIN_BACKENDS, Identity, JobError, \ | |
| JobManager, _hm, job_files, load_project_json | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") | |
| log = logging.getLogger("projtex.server") | |
| HERE = Path(__file__).resolve().parent | |
| MANAGER = JobManager() | |
| SPACE_HOST = os.environ.get("SPACE_HOST") # set by Hugging Face Spaces | |
| # -------------------------------------------------------------------------------------------- helpers | |
| def base_url(request: Request | None) -> str: | |
| if SPACE_HOST: | |
| return f"https://{SPACE_HOST}" | |
| return str(request.base_url).rstrip("/") if request else "" | |
| def identity_from(headers, token: str | None = None) -> Identity: | |
| """Caller identity for the remote ZeroGPU Spaces: explicit token > X-HF-Token / Bearer hf_ header > the | |
| visitor identity header the Hub adds (X-IP-Token) > anonymous. Values are never logged or returned.""" | |
| headers = headers or {} | |
| tok = token or headers.get("x-hf-token") | |
| auth = headers.get("authorization") or "" | |
| if not tok and auth.lower().startswith("bearer hf_"): | |
| tok = auth.split(None, 1)[1] | |
| return Identity(token=tok or None, ip_token=headers.get("x-ip-token") or None) | |
| def elapsed(job) -> float: | |
| """Seconds since the job was created: up to now while it runs (the last event's time stood still through a | |
| long step, round 32), else up to its last event.""" | |
| return round((time.time() if job.state in ("queued", "running") else job.updated) - job.created, 1) | |
| def terrain_status(job, base: str, events: int = 12) -> dict: | |
| """The status of a terrain job: progress, the files, what each layer covers, and what to do next.""" | |
| url = f"{base}/api/jobs/{job.id}" | |
| st_file = job.root / "state.json" | |
| st = json.loads(st_file.read_text()) if st_file.exists() else {} | |
| files = {k: ([f"{url}/files/{p}" for p in v] if isinstance(v, list) else | |
| {n: f"{url}/files/{p}" for n, p in v.items()} if isinstance(v, dict) else f"{url}/files/{v}") | |
| for k, v in job_files(job).items()} | |
| out: dict[str, Any] = { | |
| "job_id": job.id, "kind": "terrain", "state": job.state, "operation": job.op, "progress": job.progress, | |
| "backend": job.params.get("backend"), | |
| "quota_identity": job.identity.kind if BACKENDS[job.params["backend"]]["uses_quota"] else "not needed", | |
| "elapsed_seconds": elapsed(job), "files": files, "error": job.error, | |
| "recent_events": job.events[-events:], | |
| } | |
| warnings = list(job.result.get("warnings") or []) + list(st.get("warnings") or []) | |
| if st.get("width_m"): | |
| size = int(job.params.get("macro_size", 2048)) | |
| out["terrain"] = {"width_m": round(st["width_m"], 2), "depth_m": round(st["depth_m"], 2), | |
| "relief_m": round(st["relief_m"], 2), "triangles": st.get("triangles"), | |
| "preset": st.get("preset"), | |
| "macro": {"size": size, "metres_per_pixel": round(max(st["width_m"], st["depth_m"]) / size, 3)}} | |
| if st.get("water_repainted_pct"): | |
| out["terrain"]["water_repainted"] = { | |
| "pct": st["water_repainted_pct"], | |
| "why": "the image model painted water beyond your water_level (it enlarges small ponds and lakes); " | |
| "that ground was repainted with its layers' colours before the tiles. The model's own " | |
| "picture is files.overview_raw."} | |
| if st.get("colours_pulled"): | |
| out["terrain"]["colours_pulled"] = { | |
| "layers": st["colours_pulled"], | |
| "why": "the image model painted these layers far from your color (their CIELAB distance from it is " | |
| "shown; narrow stripes of strong colours are often lost in its picture of the whole ground): " | |
| "the colour map was moved to within 15 of your color, lightness too, its texture kept"} | |
| if st.get("light_removed"): | |
| lr = st["light_removed"] | |
| out["terrain"]["light_removed"] = { | |
| **lr, "why": f"the image model painted the slopes lit from one side (from {lr['light_from_deg']} " | |
| "degrees, 0 = -Z, 90 = +X), which would fight your scene's own sun: that brightness was " | |
| "taken out of the colour map (texture and colours kept)"} | |
| if st.get("tiles_off_layout"): | |
| out["terrain"]["tiles_off_layout"] = { | |
| "tiles": st["tiles_off_layout"], | |
| "why": "these tiles of the colour map redrew the layout instead of refining it (narrow features " | |
| "repeated across the ground, e.g. 13 m stripes on 1 km); only their finest texture was used, " | |
| "so your layout is kept there but looks softer. A smaller ground per job keeps more detail."} | |
| if st.get("covered", 1) < 0.97: | |
| warnings.append(f"{round((1 - st['covered']) * 100, 1)} % of the ground's top-down area has no upward " | |
| "surface (holes, or a mesh that is not +Y up): those parts were filled from their " | |
| "neighbours. GLB files are +Y up; for OBJ/PLY/STL from Blender pass up: 'z'.") | |
| if (st.get("source") or {}).get("turned_over"): | |
| warnings.append("the ground's faces pointed down (wound clockwise seen from above; a double-sided " | |
| "material hides it): it was turned over, so files.terrain_glb faces up. Recalculate " | |
| "your mesh's normals outside (Blender: Mesh > Normals > Recalculate Outside) if you " | |
| "render your own mesh with one-sided materials.") | |
| if st.get("layers"): | |
| out["layers"] = [{"name": L["name"], "coverage_pct": L.get("coverage_pct"), "tile_m": L["tile_m"], | |
| "detail_px_per_m": round(1024 / L["tile_m"]), | |
| "texture_url": f"{url}/files/layers/{L['name']}.jpg"} for L in st["layers"]] | |
| # thin features (a stream, a road, a 5 m garden path on 200 m) legitimately cover well under 1 %: warn | |
| # only about layers with (almost) nothing | |
| empty = [L["name"] for L in st["layers"][1:] if (L.get("coverage_pct") or 0) < 0.002] | |
| if empty: | |
| warnings.append(f"layer(s) {empty} cover no ground: their rules (where/patches) or shapes (paths/areas) " | |
| "select nothing; loosen them (see maps/*.png for slope, height, flow; paths/areas must " | |
| "lie inside the terrain's bounds, in its coordinates)") | |
| if warnings: | |
| out["warnings"] = warnings | |
| if job.result.get("terrain", {}).get("preset") and not job.params.get("layers"): | |
| out["preset_used"] = job.result["terrain"]["preset"] | |
| q = MANAGER.queue_info(job) if job.state == "queued" else None | |
| if q: | |
| out["queue"] = q | |
| out["next"] = terrain_next(job, url, out) | |
| if job.state in ("queued", "running"): | |
| wait = (q or {}).get("estimated_start_seconds") or (job.progress or {}).get("eta_seconds") or 0 | |
| out["poll_after_seconds"] = 3 if job.params["backend"] == "mock" else int(min(45, max(10, wait / 6))) | |
| return out | |
| def terrain_next(job, url: str, st: dict) -> list[dict]: | |
| s = job.state | |
| if s in ("queued", "running"): | |
| p = job.progress or {} | |
| msg = f"{job.op} is {s}: {p.get('step', '')}" | |
| if p.get("images_total"): | |
| msg += f" (image {p['images_done']}/{p['images_total']}, about {p.get('eta_seconds')} s left)" | |
| return [{"do": "poll", "method": "GET", "url": url, "why": msg}] | |
| if s == "failed": | |
| return [{"do": "read error, fix the request and create a new terrain job", "error": job.error}] | |
| if s == "stopped": | |
| return [{"do": "the ZeroGPU quota ran out: create the job again when it refills (or with another HF token)", | |
| "error": job.error}] | |
| f = st.get("files") or {} | |
| acts = [] | |
| if f.get("contact_sheet"): | |
| acts.append({"do": "look at the result: the whole terrain, then a mid and a close view, each with the " | |
| "macro colour map alone (left) and with the detail layers (right)", "method": "GET", | |
| "url": f["contact_sheet"]}) | |
| if f.get("terrain_glb"): | |
| acts.append({"do": "simplest use: download terrain.glb (your ground with the colour map; fine from a " | |
| "distance) and import it in place of your terrain", "method": "GET", "url": f["terrain_glb"]}) | |
| if f.get("bundle_zip"): | |
| acts.append({"do": "close-up detail: download terrain_bundle.zip, unzip it, and apply the detail material " | |
| "with blender_terrain.py (Blender) or three_terrain.js (three.js); see llms.txt 'Terrain'", | |
| "method": "GET", "url": f["bundle_zip"]}) | |
| layers = [L["name"] for L in st.get("layers") or []] | |
| acts.append({"do": "optional: repaint a part you do not like with a new seed (the rest is kept)", | |
| "method": "POST", "url": f"{url}/regenerate", | |
| "body": {"views": ["overview"]}, | |
| "parts": ["overview (the whole colour map: layout and colours)", "tile_<row>_<col> (one tile of " | |
| "the refined map, row 0 at -Z)"] + [f"layer:{n}" for n in layers]}) | |
| acts.append({"do": "to change where layers go (rules, paths, areas) or their prompts, create a new terrain job " | |
| "with edited layers", "method": "POST", "url": url.rsplit("/api/jobs/", 1)[0] + "/api/terrain"}) | |
| return acts | |
| def status_payload(job, base: str, events: int = 12) -> dict: | |
| """Everything an agent needs to decide its next step.""" | |
| if job.params.get("kind") == "terrain": | |
| return terrain_status(job, base, events) | |
| url = f"{base}/api/jobs/{job.id}" | |
| d = load_project_json(job) | |
| views = [] | |
| if d: | |
| for v in d["views"]: | |
| m = v.get("meta") or {} | |
| vd = {"name": v["name"], "kind": v["kind"], "status": v["status"], "enabled": v["enabled"]} | |
| cam = v.get("camera") or {} | |
| if cam.get("eye") and cam.get("target"): # degrees above (+) or below (-) the subject's centre | |
| dx, dy, dz = (float(a) - float(b) for a, b in zip(cam["eye"], cam["target"])) | |
| vd["elevation"] = round(math.degrees(math.atan2(dy, math.hypot(dx, dz)))) | |
| for k in ("edge_agreement", "silhouette_iou", "painted_corr", "edges_moved"): | |
| if m.get(k) is not None: | |
| vd[k] = m[k] | |
| if v["status"] == "rejected" and m.get("reject_reasons"): | |
| vd["reject_reasons"] = m["reject_reasons"] | |
| if m.get("partial") and v["status"] == "generated": | |
| vd["partial"] = True | |
| if m.get("masked_pct") and v["status"] == "generated": | |
| vd["masked_pct"] = m["masked_pct"] | |
| if m.get("spill_pct") and v["status"] == "generated": | |
| vd["spill_pct"] = m["spill_pct"] | |
| if m.get("parts_kept") and v["status"] == "generated" and m.get("merged_names"): | |
| vd["parts_painted_as_one"] = [f"{a} painted like {b}" for a, b in m["merged_names"]] | |
| if m.get("lowered_from") is not None: | |
| vd["lowered_from"] = m["lowered_from"] | |
| if v.get("image"): | |
| vd["image_url"] = f"{url}/files/{v['image']}" | |
| if (job.root / "views" / v["name"] / "condition.png").exists(): | |
| vd["condition_url"] = f"{url}/files/views/{v['name']}/condition.png" | |
| views.append(vd) | |
| files = {k: ([f"{url}/files/{p}" for p in v] if isinstance(v, list) else f"{url}/files/{v}") | |
| for k, v in job_files(job).items()} | |
| ins = job.result.get("inspect") or {} | |
| stats = ins.get("stats") or {} | |
| summary = None | |
| if stats: | |
| summary = {"coverage_pct": stats.get("coverage"), "holes_pct": stats.get("holes"), | |
| "measured_over": stats.get("stats_over"), | |
| "inconsistent_pct": stats.get("inconsistent"), "views_used": stats.get("views_used"), | |
| "views_rejected": [v["name"] for v in views if v["status"] == "rejected"], | |
| "views_flagged_inconsistent": (ins.get("regenerate_views") or {}).get("views", [])} | |
| if summary["views_flagged_inconsistent"] and stats.get("flagged_views_shown") is not None: | |
| summary["flagged_views_shown_pct"] = stats["flagged_views_shown"] | |
| if stats.get("underside"): | |
| summary["underside_pct"] = stats["underside"] | |
| summary["underside_covered_pct"] = stats.get("underside_covered") | |
| if stats.get("inside"): | |
| summary["inside_pct"] = stats["inside"] | |
| if stats.get("view_mismatch") is not None: | |
| summary["view_mismatch"] = stats["view_mismatch"] | |
| merged = {v["name"]: v["parts_painted_as_one"] for v in views if v.get("parts_painted_as_one")} | |
| if merged: | |
| summary["views_parts_painted_as_one"] = merged | |
| summary["verdict"], summary["verdict_why"] = verdict(summary, {v["name"] for v in views if v["kind"] == "gap"}) | |
| out: dict[str, Any] = { | |
| "job_id": job.id, "state": job.state, "operation": job.op, "progress": job.progress, | |
| "backend": job.params.get("backend"), "mode": job.params.get("mode"), | |
| "quota_identity": job.identity.kind if BACKENDS[job.params["backend"]]["uses_quota"] else "not needed", | |
| "elapsed_seconds": elapsed(job), "summary": summary, "files": files, | |
| "views": views, "error": job.error, "recent_events": job.events[-events:], | |
| } | |
| last = dict(job.result.get("last") or {}) | |
| if last.get("images"): | |
| last["images"] = {k: f"{url}/files/{v}" for k, v in last["images"].items()} | |
| out["last_result"] = last or None | |
| if d and d.get("layers"): | |
| out["retouch_layers"] = [{"name": L["name"], "mode": L["mode"], "enabled": L.get("enabled", True), | |
| "compare_url": f"{url}/files/layers/{L['name']}/compare.png"} for L in d["layers"]] | |
| if ins.get("suggestions"): | |
| out["retouch_suggestions"] = [{"index": i, "reason": s["reason"], "area_pct": s["area_pct"]} | |
| for i, s in enumerate(ins["suggestions"])] | |
| note = BACKENDS[job.params["backend"]]["license_note"] | |
| if note: | |
| out["license_note"] = note | |
| if d and files.get("textured_glb"): | |
| out["output"] = output_notes(d) | |
| q = MANAGER.queue_info(job) if job.state == "queued" else None | |
| if q: | |
| out["queue"] = q | |
| pending = None | |
| if job.result.get("model"): | |
| out["model"] = job.result["model"] | |
| if job.result.get("pbr"): | |
| out["pbr"] = job.result["pbr"] | |
| if job.result.get("warnings"): | |
| out["warnings"] = job.result["warnings"] | |
| if job.state == "waiting_for_images": | |
| name = (job.result.get("remaining_views") or [None])[0] | |
| req = job.root / "exchange" / f"{name}.request.json" | |
| rq = json.loads(req.read_text()) if req.exists() else {} | |
| pending = {"view": name, "condition_url": f"{url}/files/exchange/{name}.condition.png", | |
| "prompt": rq.get("prompt"), "width": rq.get("width"), "height": rq.get("height"), | |
| "upload": {"method": "POST", "url": f"{url}/views/{name}/image", | |
| "body": "multipart field 'file', or JSON {\"image_url\": ...} / {\"image_base64\": ...}"}} | |
| out["pending_view"] = pending | |
| out["next"] = next_actions(job, url, out) | |
| if job.state in ("queued", "running"): | |
| wait = (q or {}).get("estimated_start_seconds") or (job.progress or {}).get("eta_seconds") or 0 | |
| out["poll_after_seconds"] = 3 if job.params["backend"] == "mock" else int(min(60, max(15, wait / 6))) | |
| return out | |
| def output_notes(d: dict) -> dict: | |
| """How the result maps onto the caller's own scene (Blender / three.js).""" | |
| m = d.get("mesh") or {} | |
| uv = (m.get("uv") or {}).get("decision") | |
| notes = { | |
| "uv_layout": {"keep": "kept: the texture is baked into your own UVs", | |
| "pack": "packed: each material's UV square was moved into one shared atlas", | |
| "regen": "regenerated: your UVs overlapped or were missing, so a new atlas was unwrapped"}.get(uv, uv), | |
| "albedo_matches_your_uvs": uv == "keep", | |
| "textured_glb": "same objects (names, transforms) and materials (names, metallic/roughness/alpha, emission, " | |
| "glass and other material extensions) as your file; every material's base colour now uses the " | |
| "one baked texture. Import it in place of your model.", | |
| } | |
| refused = (m.get("uv") or {}).get("keep_refused") | |
| if refused is not None: | |
| notes["uv_layout"] = ( | |
| f"{notes['uv_layout'].split(':')[0]} instead of kept: {refused * 100:.0f} % of your UV layout is used by " | |
| "more than one face (e.g. Smart UV Project run on each object, so every part fills the whole 0-1 " | |
| "square), and baked there the parts would show each other's paint. The GLB has the new layout; to plug " | |
| "files.albedo into your own material, unwrap all parts into one layout (Blender: select every object, " | |
| "Tab, U > Smart UV Project, one run) and send it again") | |
| if (d.get("settings") or {}).get("export_single_mesh"): | |
| notes["textured_glb"] = ("one object 'baked' with one material (single_mesh): metallic/roughness are the " | |
| "area-weighted means of your materials; same orientation as your file") | |
| orient = m.get("orientation") or {} | |
| if orient.get("faces_flipped"): | |
| notes["normals"] = (f"{orient['area_flipped_pct']} % of the surface was inside-out (normals pointing inwards, " | |
| "e.g. reversed winding); those faces were flipped so they could be painted, and the GLB " | |
| "has outward normals") | |
| if uv == "keep": | |
| notes["albedo_png"] = ("files.albedo maps onto your original model's UVs: you can plug it into your own " | |
| "material's base colour instead of importing the GLB") | |
| return notes | |
| MINOR_SHOWN_PCT = 0.5 # a flagged view showing on less of the surface does not lower the verdict | |
| VIEW_MISMATCH = 38.0 # summary.view_mismatch from here on: the bake does not reproduce the views | |
| def verdict(summ: dict, gap_views: set | None = None) -> tuple[str, str]: | |
| """good: ship it | usable: fine at a distance, optional fixes | needs work: follow `next`.""" | |
| cov, holes = summ.get("coverage_pct") or 0.0, summ.get("holes_pct") or 0.0 | |
| flagged, rejected = summ.get("views_flagged_inconsistent") or [], summ.get("views_rejected") or [] | |
| # a rejected gap view is left out and only matters through the coverage it would have added | |
| rejected = [r for r in rejected if r not in (gap_views or set())] | |
| # the part of the texture where a flagged view wins the blend and contradicts the others (older results only | |
| # have the overall disagreement between views) | |
| shown = summ.get("flagged_views_shown_pct", summ.get("inconsistent_pct") or 0.0) | |
| minor = flagged and shown < MINOR_SHOWN_PCT # it hardly paints anything that way: not worth any quota | |
| # the texture seen from the views' cameras against the generated views (90th-percentile colour difference, | |
| # median over views): good jobs 5-27, a gramophone whose parts shared one UV square 48 (round 35) | |
| mism = summ.get("view_mismatch") or 0.0 | |
| if cov < 75 or (flagged and shown >= 2.0) or mism >= VIEW_MISMATCH: | |
| why = [f"coverage {cov} %" if cov < 75 else "", | |
| f"view(s) {flagged} contradict the others on {shown} % of the surface" if flagged and shown >= 2.0 | |
| else "", | |
| f"the texture does not show what the views painted (colour difference {mism}; good jobs stay " | |
| f"under 30): parts may share UV space (see output.uv_layout; \"uv\": \"auto\" unwraps anew) or the " | |
| f"views may not line up with the model" if mism >= VIEW_MISMATCH else ""] | |
| return "needs work", "; ".join(w for w in why if w) | |
| merged = summ.get("views_parts_painted_as_one") or {} | |
| if cov < 90 or holes >= 5 or rejected or (flagged and not minor) or merged: | |
| why = [f"coverage {cov} %" if cov < 90 else "", f"holes {holes} %" if holes >= 5 else "", | |
| f"rejected views {rejected}" if rejected else "", | |
| f"view(s) {flagged} disagree with the others ({shown} % of the surface shows it)" if flagged else "", | |
| f"view(s) {sorted(merged)} painted parts in the colour of a part they touch on every try" if merged | |
| else ""] | |
| return "usable", "; ".join(w for w in why if w) | |
| return "good", f"coverage {cov} %, holes {holes} %, " + ( | |
| f"view(s) {flagged} disagree with the others where only {shown} % of the surface shows it" if minor | |
| else "all views consistent") | |
| def next_actions(job, url: str, st: dict) -> list[dict]: | |
| s = job.state | |
| if s in ("queued", "running"): | |
| p = job.progress or {} | |
| msg = f"{job.op} is {s}: {p.get('step', '')}" | |
| if p.get("total"): | |
| msg += f" (view {p.get('done', 0) + 1}/{p['total']}: {p.get('view')}, about {p.get('eta_seconds')} s left)" | |
| if st.get("queue"): | |
| q = st["queue"] | |
| msg += (f" (position {q['position']} in the queue; {q['operations_running']} operations running" | |
| + (f", start in about {q['estimated_start_seconds']} s" if q["estimated_start_seconds"] else "") | |
| + ")") | |
| return [{"do": "poll", "method": "GET", "url": url, "why": msg}] | |
| if s == "waiting_for_images": | |
| pv = st["pending_view"] | |
| return [{"do": "generate the image yourself", "why": "external backend: create an image aligned with the " | |
| "conditioning render (same framing), following the prompt", "get": pv["condition_url"]}, | |
| {"do": "upload it", "method": "POST", "url": pv["upload"]["url"]}] | |
| back = job.result.get("quota_back_at") | |
| back = back if back and back > time.time() else None | |
| when = (f"; it refills in about {_hm(back - time.time())} " | |
| f"({time.strftime('%H:%M UTC', time.gmtime(back))})") if back else "" | |
| if s == "stopped": | |
| return [{"do": "bake what is finished now, and generate the rest when the quota is back", "method": "POST", | |
| "url": f"{url}/bake", "body": {"resolution": 2048}, | |
| "why": "the ZeroGPU quota of the identity used was exhausted; finished views are kept" + when}, | |
| {"do": "later (or with another identity's quota): generate the remaining views", "method": "POST", | |
| "url": f"{url}/regenerate", | |
| "body": {"views": job.result.get("remaining_views", []), "hf_token": "<optional, your token>"}}, | |
| {"do": "or continue without quota, with images you make yourself", "method": "POST", | |
| "url": f"{url}/regenerate", | |
| "body": {"views": job.result.get("remaining_views", []), "backend": "external"}}] | |
| if s == "failed": | |
| return [{"do": "read error, fix the request and create a new job (or retry the operation)", | |
| "error": job.error}] | |
| acts = [] | |
| last = st.get("last_result") or {} | |
| if last.get("operation") == "retouch" and last.get("status") != "applied": | |
| why = {"empty_mask": "the mask covered no visible surface from that camera: aim it with " | |
| "POST /pick (render_camera + pixels) and use a sphere mask around the returned point, " | |
| "or use a retouch_suggestions index", | |
| "rejected": "the model's output did not keep the framing/geometry; try another camera, a tighter " | |
| "focus, another prompt or another backend"}.get(last.get("status"), last.get("detail")) | |
| acts.append({"do": f"retouch was not applied ({last.get('status')})", "why": why, | |
| "pick": {"method": "POST", "url": f"{url}/pick", | |
| "body": {"render_camera": "orbit00", "pixels": [[0.5, 0.5]]}}}) | |
| elif last.get("operation") == "retouch" and last.get("images", {}).get("compare"): | |
| acts.append({"do": "check the retouch (before | model input | model output | after)", "method": "GET", | |
| "url": last["images"]["compare"], | |
| "undo": {"method": "POST", "url": f"{url}/undo-retouch"}}) | |
| files = st.get("files") or {} | |
| if files.get("textured_glb"): | |
| acts.append({"do": "download the textured model", "method": "GET", "url": files["textured_glb"]}) | |
| planned = [v["name"] for v in st.get("views") or [] if v.get("status") == "planned"] | |
| if planned and job.params.get("backend") != "external": | |
| acts.append({"do": f"generate the planned view(s) {planned} that were never made" | |
| + (" once the ZeroGPU quota is back" if back else ""), "method": "POST", | |
| "url": f"{url}/regenerate", "body": {"views": planned}, | |
| "why": "the quota ran out before them" + when if back else "they were not generated"}) | |
| if files.get("contact_sheet"): | |
| acts.append({"do": "look at the result", "method": "GET", "url": files["contact_sheet"]}) | |
| if last.get("operation") == "fill_gaps": # the contact sheet's cameras do not see the underside | |
| added = {n for r in last.get("rounds") or [] for n in r.get("added") or []} | |
| below = {v["name"]: v["image_url"] for v in st.get("views") or [] if v["name"] in added and v.get("enabled") | |
| and v.get("status") == "generated" and (v.get("elevation") or 0) < -20 and v.get("image_url")} | |
| if below: | |
| acts.append({"do": "look at the views from below that were added: image models sometimes paint the top " | |
| "of the object there (laces across a sneaker's sole); disable any that does", | |
| "views": below, "disable": {"method": "POST", "url": f"{url}/disable-views", | |
| "body": {"views": ["<name>"]}}}) | |
| summ = st.get("summary") or {} | |
| flagged = summ.get("views_flagged_inconsistent") or [] | |
| rejected = summ.get("views_rejected") or [] | |
| shown = summ.get("flagged_views_shown_pct", summ.get("inconsistent_pct") or 0.0) | |
| if flagged and shown < MINOR_SHOWN_PCT: | |
| acts.append({"do": "optional, usually not worth it: regenerate views that disagree with the others", | |
| "method": "POST", "url": f"{url}/regenerate", "body": {"views": flagged}, | |
| "why": f"only {shown} % of the surface shows their differing content"}) | |
| elif flagged: | |
| acts.append({"do": "regenerate views whose content contradicts the others (new seeds)", "method": "POST", | |
| "url": f"{url}/regenerate", "body": {"views": flagged}, | |
| "if_still_flagged": {"method": "POST", "url": f"{url}/disable-views", "body": {"views": flagged}}}) | |
| merged = summ.get("views_parts_painted_as_one") or {} | |
| if merged: | |
| acts.append({"do": "look at these views: every try painted a part in the colour of the part it touches, " | |
| "although your base colours differ (a dark fretboard painted like the maple neck)", | |
| "views": merged, "method": "POST", "url": f"{url}/regenerate", "body": {"views": sorted(merged)}, | |
| "why": "if the parts should differ: regenerate (new seeds), or retouch the part with a " | |
| "prompt naming its colour; if your description makes them alike, nothing to do"}) | |
| if rejected: | |
| acts.append({"do": "optionally retry rejected views with new seeds (they are excluded from the texture)", | |
| "method": "POST", "url": f"{url}/regenerate", "body": {"views": rejected}}) | |
| if summ.get("coverage_pct") is not None and summ["coverage_pct"] < 85: | |
| acts.append({"do": "add cameras for still-uncovered surface", "method": "POST", "url": f"{url}/fill-gaps", | |
| "body": {"max_views": 3}}) | |
| elif (summ.get("holes_pct") or 0) >= 5: # the reason for "usable": surface every view missed | |
| acts.append({"do": "add cameras for the surface no view painted", "method": "POST", "url": f"{url}/fill-gaps", | |
| "body": {"max_views": 2}, | |
| "why": f"holes {summ['holes_pct']} % of the surface (filled with nearby colours for now)"}) | |
| sug = st.get("retouch_suggestions") or [] | |
| if sug and summ.get("verdict") != "good" and sug[0]["area_pct"] >= 1.0: | |
| acts.append({"do": "fix the worst remaining region locally", "method": "POST", "url": f"{url}/retouch", | |
| "body": {"suggestion": 0}, "why": f"{sug[0]['reason']} region, {sug[0]['area_pct']} % of the surface"}) | |
| if (summ.get("underside_pct") or 0) > 3 and (summ.get("underside_covered_pct") or 0) < 50: | |
| acts.append({"do": "optional, only if the underside will be seen (it faces the ground): add cameras below", | |
| "method": "POST", "url": f"{url}/fill-gaps", "body": {"max_views": 2}, | |
| "why": f"{summ['underside_pct']}% of the surface is only visible from below and is mostly " | |
| "untextured (filled with nearby colours)"}) | |
| if summ.get("verdict") == "good": | |
| acts.append({"do": "nothing else needed: the texture is complete and consistent" | |
| + (" (apart from the optional step above)" if acts and acts[-1]["do"].startswith("optional") else "")}) | |
| if back and BACKENDS.get(job.params.get("backend"), {}).get("uses_quota"): | |
| for a in acts: # these would stop again right away | |
| if a.get("url", "").endswith(("/regenerate", "/fill-gaps", "/retouch")): | |
| a["needs_quota"] = "the ZeroGPU quota is used up" + when | |
| return acts | |
| def http_error(exc: JobError) -> JobError: | |
| """Errors are re-raised as JobError so the handler returns a flat {"error", "hint"} body.""" | |
| return exc | |
| # --------------------------------------------------------------------------------------------- models | |
| class TextureRequest(BaseModel): | |
| prompt: str | None = Field(None, description="Materials and colours of what the mesh models, e.g. 'worn olive " | |
| "military paint, gold visor'. Describe materials, not new geometry.") | |
| upload_id: str | None = Field(None, description="Id from POST /api/uploads (the mesh file sent as the raw request " | |
| "body). The recommended way to send a local file.") | |
| mesh_base64: str | None = Field(None, description="The mesh file itself, base64-encoded (.glb/.obj/.ply/.stl or " | |
| ".zip, up to 20 MB). Handy for MCP clients.") | |
| sample: Literal["avocado", "lantern", "head"] | None = Field(None, description="Use a built-in sample mesh.") | |
| mesh_url: str | None = Field(None, description="Public http(s) URL of a .glb/.obj/.ply/.stl, or a .zip with a " | |
| f".gltf and its files. Max {MAX_MESH_BYTES >> 20} MB, " | |
| f"{MAX_FACES} triangles.") | |
| mode: Literal["object", "character", "environment"] | None = Field( | |
| None, description="object: orbit views; character: + face close-ups, auto up-axis and facing; environment: " | |
| "360° probes from inside (slow on this CPU machine). Default: object (or the sample's).") | |
| backend: Literal["qwen-turbo", "qwen-space", "fibo-space", "mock", "external"] = Field( | |
| "qwen-turbo", description="qwen-turbo: Qwen-Image-2.1-Turbo, 8 steps (default, fastest, least GPU quota). " | |
| "qwen-space: full Qwen-Image-2.1 (40 steps, official Space). fibo-space: Bria Fibo " | |
| "2.0 Edit Turbo (non-commercial). mock: free dry run. external: you supply images.") | |
| part_prompts: dict[str, str] | None = Field( | |
| None, description="Description per material or object name of your file, e.g. {\"Walls\": \"white " | |
| "clapboard siding\", \"Roof\": \"grey asphalt shingles\", \"Door\": \"dark green panel " | |
| "door\"}. Each view's prompt then names only the parts that view shows, which stops doors, " | |
| "windows or faces being painted where the model has none. Recommended for models with " | |
| "several named parts; keep `prompt` for the overall style. Names are listed in the job " | |
| "status (model.materials / model.objects).") | |
| subject: str | None = Field(None, description="What the mesh is, e.g. 'a sci-fi pilot helmet'.") | |
| preset: str | None = Field(None, description="object/character: minimal (4 views) | balanced (7, default) | " | |
| "dense (12) | greedy; environment: probes | probes-greedy.") | |
| probes: int | None = Field(None, ge=1, le=12, description="environment: number of 360° probes (6 views each).") | |
| gap_views: bool = Field(True, description="Add up to 3 cameras for surface the preset misses.") | |
| focus_prompt: str | None = Field(None, description="character: prompt for the face close-ups only.") | |
| lighting: Literal["flat", "soft", "natural"] | None = Field(None, description="Lighting asked of the model " | |
| "(flat = albedo-like, default).") | |
| texture_size: int | None = Field(None, description="Baked texture size: 1024, 2048 (default) or 4096.") | |
| pbr: bool | None = Field(None, description="Also make PBR materials: every view is split into albedo, roughness " | |
| "and metallic maps (a LoRA for Qwen-Image-Edit-2511 trained on " | |
| "MatSynth renders, on ZeroGPU with your quota), baked like the colour: " | |
| "files.textured_pbr_glb (base colour + metallicRoughness texture), " | |
| "albedo_pbr, roughness, metallic, metallic_roughness") | |
| view_size: int | None = Field(None, description="Generated image size per view (512-1024, default 1024).") | |
| up: Literal["auto", "y", "z", "-z", "x"] | None = Field(None, description="Up axis of the asset if not +Y " | |
| "(GLB is always +Y). The result keeps " | |
| "the input's orientation.") | |
| color_guide: bool | None = Field(None, description="Show your materials' base colours to the image model as a " | |
| "guide to which part is which (default: automatic when at " | |
| "least two materials have clearly different colours). Turn it " | |
| "off if your colours are placeholders.") | |
| single_mesh: bool = Field(False, description="Return one object with one material (a game-ready prop) instead " | |
| "of your objects and materials. Also settable later with /bake.") | |
| inside: Literal["plain", "paint"] | None = Field(None, description="The inside of a mug, vase or pot (seen only " | |
| "from above through its opening). plain " | |
| "(default): the main colour of its material. " | |
| "paint: views may paint it too (a bowl " | |
| "decorated inside).") | |
| pattern_swatches: Literal["auto", "on", "off"] | None = Field( | |
| None, description="Show each view a close-up of the decoration already painted on the part it must paint, so " | |
| "an all-over pattern (porcelain florals, folk art, fabric) continues onto every side " | |
| "instead of stopping. auto (default): on, except for parts described as a scene " | |
| "(landscape, river, mountains), where views should continue the scene rather than repeat " | |
| "its motifs.") | |
| metal: Literal["keep", "flat"] | None = Field(None, description="Mirror-like metal (metallic >= 0.9 and roughness " | |
| "<= 0.3 in your file, e.g. chrome). keep " | |
| "(default): the painted reflections, which read " | |
| "as chrome in any renderer. flat: an even base " | |
| "colour, for renderers with environment lighting " | |
| "(Cycles/Eevee with an HDRI, three.js with an " | |
| "environment map), where painted reflections " | |
| "would show twice.") | |
| uv: Literal["auto", "keep", "regen"] | None = Field(None, description="auto (default): bake into your UVs when " | |
| "they are a valid non-overlapping 0-1 layout, " | |
| "else repack per material or unwrap anew. " | |
| "keep: bake into your UVs even with small " | |
| "overlaps (more than 10 % shared: as auto, " | |
| "output.uv_layout says why). regen: always " | |
| "unwrap anew (xatlas).") | |
| seed: int | None = None | |
| hf_token: str | None = Field(None, description="Optional Hugging Face token whose ZeroGPU quota pays for " | |
| "generation. Kept in memory for this job only, never stored or " | |
| "logged. Alternatively send it as header X-HF-Token.") | |
| def _not_a_terrain(cls, data): | |
| """A terrain body sent here is refused, not run as an orbit job without its layers: a tester followed the | |
| upload's first "next" (POST /api/jobs) with layers, water_level and macro_size, and waited 5 minutes for | |
| an object texture of its polder (round 33).""" | |
| found = [k for k in TERRAIN_ONLY if isinstance(data, dict) and k in data] | |
| if found: | |
| raise JobError(f"{', '.join(found)} {'is a terrain field' if len(found) == 1 else 'are terrain fields'}: " | |
| "this endpoint textures objects", 400, | |
| "send the same body to POST /api/terrain (ground layers and a top-down colour map)") | |
| return data | |
| TERRAIN_ONLY = ("layers", "water_level", "macro_size", "heightmap", "axes") | |
| class TerrainPath(BaseModel): | |
| points: list[list[float]] = Field(..., description="[[x, z], ...] ground coordinates in metres (glTF/three.js " | |
| "axes; with axes 'z-up', Blender [x, y])") | |
| width: float = Field(4.0, description="width in metres") | |
| class TerrainLayer(BaseModel): | |
| name: str = Field(..., description="Layer name, e.g. 'grass', 'rock' (used for file names)") | |
| prompt: str = Field(..., description="What this ground looks like close up, e.g. 'short green grass with clover " | |
| "and small daisies' (it becomes a seamless photo texture)") | |
| where: dict[str, list[float | None]] | None = Field( | |
| None, description="Where the layer lies, as [min, max] ranges (null = no bound) of: slope (degrees), height " | |
| "(0 lowest .. 1 highest point), elevation (metres, the mesh's own heights: [-0.5, 1.5] for " | |
| "a beach above a sea at 0), curvature (-1 ridges .. 1 hollows and valleys), flow " | |
| "(0 .. 1 water flow: ~0.85+ streams and gullies). E.g. {\"slope\": [35, null]} for cliffs. " | |
| "No rules (and no paths/areas): everywhere, i.e. the base layer (put it first).") | |
| paths: list[TerrainPath] | None = Field(None, description="Lines this layer is drawn along: roads, trails, " | |
| "rivers, e.g. [{\"points\": [[-500, 20], [300, 80]], " | |
| "\"width\": 5}]") | |
| areas: list[list[list[float]]] | None = Field(None, description="Polygons this layer fills, [[[x, z], ...], ...]: " | |
| "fields, a village square, a car park") | |
| patches: float | None = Field(None, ge=0, le=1, description="Keep only this fraction of the layer's area, in " | |
| "natural patches (dirt patches in grass: 0.15)") | |
| breakup: float | None = Field(None, ge=0, le=1, description="How irregular the layer's edges are (default 0.5)") | |
| tile_m: float = Field(4.0, gt=0.2, le=50, description="Size in metres of one repeat of the detail texture") | |
| color: list[float] | None = Field(None, description="Optional [r, g, b] 0-255 (or 0-1) the layer shows in the " | |
| "overview guide (default: its detail texture's mean colour); " | |
| "the colour map is turned to its hue and kept within 15 " | |
| "(CIELAB) of it") | |
| triplanar: bool | None = Field(None, description="Project the detail texture from the sides on steep slopes " | |
| "(default: layers whose rules start at 30 degrees or more)") | |
| from_above: str | None = Field(None, description="How the layer looks from high above in the colour map, if " | |
| "unlike its close-up (default for forests: their tree crowns)") | |
| class TerrainHeightmap(BaseModel): | |
| upload_id: str | None = Field(None, description="From POST /api/uploads (a grey PNG, 16-bit preferred, or .npy)") | |
| url: str | None = Field(None, description="Public URL of the PNG / .npy") | |
| base64: str | None = Field(None, description="The PNG / .npy itself, base64-encoded") | |
| size_m: float | list[float] = Field(..., description="Terrain width in metres, or [width, depth]") | |
| height_m: float = Field(100.0, description="Height difference in metres between black and white") | |
| class TerrainRequest(BaseModel): | |
| prompt: str | None = Field(None, description="The place as a whole, e.g. 'an alpine valley in early summer', " | |
| "'English farmland', 'a red-rock desert canyon'") | |
| upload_id: str | None = Field(None, description="The terrain mesh from POST /api/uploads (.glb/.obj/.ply/.stl): " | |
| "the ground only, +Y up (GLB)") | |
| mesh_url: str | None = None | |
| mesh_base64: str | None = None | |
| heightmap: TerrainHeightmap | None = Field(None, description="Instead of a mesh: a heightmap image") | |
| preset: Literal["alpine", "temperate", "desert", "arctic", "volcanic"] | None = Field( | |
| None, description="Ready-made layer stack when `layers` is not given (default: chosen from the prompt's " | |
| "words; temperate otherwise)") | |
| layers: list[TerrainLayer] | None = Field(None, description="Your ground layers (up to 8), bottom to top: the " | |
| "first covers everything, each next one covers the " | |
| "earlier ones where its rules/paths/areas hold") | |
| macro_size: Literal[1024, 2048, 4096] = Field(2048, description="Size of the top-down colour map: 1024 (one " | |
| "image), 2048 (9 refined tiles), 4096 (25)") | |
| uv: Literal["planar", "keep"] = Field("planar", description="planar: terrain.glb gets top-down UVs (the macro " | |
| "map as is); keep: the macro map is baked into your " | |
| "mesh's own UVs") | |
| axes: Literal["y-up", "z-up"] = Field("y-up", description="Coordinates of paths/areas: y-up = glTF/three.js " | |
| "[x, z]; z-up = Blender [x, y]") | |
| up: Literal["y", "z"] | None = Field(None, description="Up axis of OBJ/PLY/STL files (GLB is always +Y)") | |
| water_level: float | None = Field(None, description="Height of your water plane (sea, lake, pond) in the " | |
| "mesh's own heights: the colour map paints open water below " | |
| "it, so the painted shore meets your plane") | |
| backend: Literal["qwen-turbo", "qwen-space", "mock"] = "qwen-turbo" | |
| seed: int | None = None | |
| hf_token: str | None = None | |
| class RegenerateRequest(BaseModel): | |
| views: list[str] = Field(..., description="View names (see 'views' in the job status).") | |
| seeds: dict[str, int] | None = Field(None, description="Optional seed per view (random new seeds otherwise).") | |
| prompt_extra: str | None = Field(None, description="Extra instruction appended to these views' prompts.") | |
| backend: Literal["qwen-turbo", "qwen-space", "fibo-space", "mock", "external"] | None = None | |
| hf_token: str | None = None | |
| class ViewsRequest(BaseModel): | |
| views: list[str] | |
| class FillGapsRequest(BaseModel): | |
| max_views: int = Field(3, ge=1, le=6) | |
| hf_token: str | None = None | |
| class RetouchRequest(BaseModel): | |
| suggestion: int | None = Field(None, description="Index into retouch_suggestions of the job status: " | |
| "uses its camera, mask and mode.") | |
| mode: Literal["repair", "replace", "upscale"] | None = Field(None, description="repair (fix artifacts), " | |
| "replace (new content), upscale (re-imagine a " | |
| "low-res capture at higher resolution).") | |
| prompt: str = "" | |
| view: str | None = Field(None, description="Use the camera of this view.") | |
| render_camera: str | None = Field(None, description="Use a preview render camera, e.g. 'orbit03'.") | |
| camera: dict | str | None = Field(None, description='{"eye":[x,y,z],"target":[x,y,z],"fov":40} (a view or ' | |
| 'render camera name also works here)') | |
| focus: dict | None = Field(None, description='Frame a sphere: {"point":[x,y,z],"radius":r,"direction":[dx,dy,dz]}') | |
| mask: Any = Field(None, description="'all' | 'uncovered' | 'low_quality' | {'sphere':[x,y,z,r]} | " | |
| "{'rect':[x0,y0,x1,y1]} (0-1 fractions ok) | {'circle':[cx,cy,r]} | " | |
| "{'and'|'or':[...]}") | |
| capture_size: int | None = Field(None, description="upscale: capture size, e.g. 256") | |
| output_size: int | None = None | |
| softness: float = 0.75 | |
| grow_px: int = 0 | |
| backend: Literal["qwen-turbo", "qwen-space", "fibo-space", "mock"] | None = None | |
| dry_run: bool = Field(False, description="Only show where the mask lands (the capture with the mask outlined " | |
| "in red, aim_url) in seconds: nothing is generated or changed.") | |
| hf_token: str | None = None | |
| class PbrRequest(BaseModel): | |
| hf_token: str | None = Field(None, description="Your Hugging Face token, for your own ZeroGPU quota (or send it " | |
| "as the X-HF-Token header)") | |
| class BakeRequest(BaseModel): | |
| resolution: Literal[1024, 2048, 4096] = 2048 | |
| single_mesh: bool | None = Field(None, description="true: one object, one material; false: your objects and " | |
| "materials; omitted: unchanged") | |
| class PickRequest(BaseModel): | |
| pixels: list[list[float]] = Field(..., description="[[x, y], ...] in pixels or 0-1 fractions of the image") | |
| view: str | None = None | |
| render_camera: str | None = Field(None, description="e.g. 'orbit02' (see files.renders)") | |
| class ImageSubmit(BaseModel): | |
| image_url: str | None = None | |
| image_base64: str | None = None | |
| # ------------------------------------------------------------------------------------------------ MCP | |
| from mcp.server.mcpserver import Context, Image as MCPImage, MCPServer # noqa: E402 | |
| from mcp.server.transport_security import TransportSecuritySettings # noqa: E402 | |
| LLMS_TXT = (HERE / "llms.txt").read_text() | |
| mcp = MCPServer( | |
| "projtex", | |
| instructions="Texture 3D meshes with AI-generated views projected onto the geometry. Workflow: texture_mesh " | |
| "(returns a job id at once) -> get_job every ~15 s until state is done -> view_image(contact_sheet) " | |
| "-> follow get_job's 'next' hints (regenerate_views, fill_gaps, retouch) -> download the GLB from " | |
| "files.textured_glb. Generation uses the caller's ZeroGPU quota: send your HF token as header " | |
| "X-HF-Token (or hf_token argument); backend 'mock' is a free dry run.", | |
| ) | |
| def _mcp_ident(ctx: Context | None, token: str | None) -> Identity: | |
| return identity_from({k.lower(): v for k, v in (ctx.headers or {}).items()} if ctx else {}, token) | |
| def _mcp_call(fn): | |
| try: | |
| return fn() | |
| except JobError as exc: | |
| return {"error": str(exc), "hint": exc.hint} | |
| def texture_mesh(prompt: str = "", upload_id: str = "", mesh_url: str = "", mesh_base64: str = "", sample: str = "", | |
| mode: str = "", backend: str = "qwen-turbo", subject: str = "", preset: str = "", | |
| focus_prompt: str = "", texture_size: int = 0, uv: str = "", up: str = "", | |
| part_prompts: dict[str, str] | None = None, single_mesh: bool = False, inside: str = "", | |
| metal: str = "", pattern_swatches: str = "", pbr: bool = False, hf_token: str = "", | |
| ctx: Context | None = None) -> dict: | |
| """Start texturing a mesh; returns a job id immediately (the job takes ~15 s per view with qwen-turbo). | |
| Mesh: upload_id (from REST POST /api/uploads with the file as raw body), mesh_url (public .glb/.obj/.ply/.stl/ | |
| .zip), mesh_base64 (the file itself, up to 20 MB) or sample ('avocado' | 'lantern' | 'head'). prompt: materials | |
| and colours of what the mesh models. mode: object | character | environment. backend: qwen-turbo (default, | |
| Qwen-Image-2.1-Turbo), qwen-space (full Qwen-Image-2.1), fibo-space (non-commercial), mock (free dry run), | |
| external (you supply images). preset: minimal | balanced | | |
| dense | greedy. uv: auto | keep | regen. part_prompts: {material or object name: description}; each view then | |
| names only the parts it shows (recommended for multi-part models). The result GLB keeps your objects and | |
| materials (single_mesh: one object, one material instead). inside: plain (default: the inside of a mug or vase | |
| gets its material's main colour) | paint. metal: keep (default: painted reflections) | flat (chrome-like | |
| materials get an even base colour, for renderers with environment lighting). pbr: true also bakes PBR maps | |
| (albedo, roughness, metallic; files.textured_pbr_glb). Then poll get_job.""" | |
| params = {k: v for k, v in dict(prompt=prompt, upload_id=upload_id, sample=sample, mesh_url=mesh_url, mode=mode, | |
| backend=backend, subject=subject, preset=preset, focus_prompt=focus_prompt, | |
| texture_size=texture_size, uv=uv, up=up, part_prompts=part_prompts, | |
| single_mesh=single_mesh, inside=inside, metal=metal, | |
| pattern_swatches=pattern_swatches, pbr=pbr).items() if v} | |
| def run(): | |
| return status_payload(MANAGER.create(params, _mcp_ident(ctx, hf_token or None), | |
| upload=_decode_mesh(mesh_base64 or None)), base_url(None)) | |
| return _mcp_call(run) | |
| def texture_terrain(prompt: str = "", upload_id: str = "", mesh_url: str = "", mesh_base64: str = "", | |
| heightmap: dict | None = None, preset: str = "", layers: list[dict] | None = None, | |
| macro_size: int = 2048, uv: str = "planar", axes: str = "y-up", up: str = "", | |
| backend: str = "qwen-turbo", seed: int | None = None, hf_token: str = "", | |
| ctx: Context | None = None) -> dict: | |
| """Texture a terrain (ground, landscape, heightfield): returns a job id at once (about 2 min for macro_size | |
| 2048 with qwen-turbo). Ground: a mesh (upload_id / mesh_url / mesh_base64; the ground only, +Y up) or heightmap | |
| {upload_id | url | base64, size_m, height_m}. prompt: the place ('an alpine valley in early summer'). layers | |
| (up to 8, bottom to top): [{name, prompt (close-up look), where: {slope|height|curvature|flow: [min, max]}, | |
| paths: [{points: [[x, z], ...], width}], areas: [[[x, z], ...]], patches, tile_m}]; or preset alpine | | |
| temperate | desert | arctic | volcanic (default: chosen from the prompt). macro_size 1024 | 2048 | 4096. The | |
| result: terrain.glb (your ground with a top-down colour map) and terrain_bundle.zip (splat maps, seamless | |
| detail textures, Blender / three.js materials). Then poll get_job. See the guide's 'Terrain' section.""" | |
| params = {k: v for k, v in dict(prompt=prompt, upload_id=upload_id, mesh_url=mesh_url, heightmap=heightmap, | |
| preset=preset, layers=layers, macro_size=macro_size, uv=uv, axes=axes, up=up, | |
| backend=backend, seed=seed).items() if v not in (None, "", [], {})} | |
| def run(): | |
| try: | |
| req = TerrainRequest(**params) | |
| except Exception as exc: | |
| raise JobError(f"invalid parameters: {exc}", 400, "see the TerrainRequest schema at /docs") | |
| prm = req.model_dump(exclude_none=True, exclude={"hf_token", "mesh_base64"}) | |
| hm_data = None | |
| if req.heightmap and req.heightmap.base64: | |
| hm_data = ("heightmap", _decode_mesh(req.heightmap.base64)[1]) | |
| prm["heightmap"].pop("base64", None) | |
| return status_payload(MANAGER.create_terrain(prm, _mcp_ident(ctx, hf_token or None), | |
| upload=_decode_mesh(mesh_base64 or None), heightmap_data=hm_data), | |
| base_url(None)) | |
| return _mcp_call(run) | |
| def get_job(job_id: str, include_preview: bool = True) -> list: | |
| """Status of a job: state, progress, per-view validation, coverage/quality summary, file URLs, and 'next' | |
| actions to take. When done and include_preview, also returns the contact sheet image.""" | |
| def run(): | |
| job = MANAGER.get(job_id) | |
| st = status_payload(job, base_url(None)) | |
| out: list = [json.dumps(st, indent=1)] | |
| sheet = job.root / "renders" / "contact_sheet.png" | |
| if include_preview and job.state == "done" and sheet.exists(): | |
| out.append(_mcp_image(sheet, 1024)) | |
| return out | |
| try: | |
| return run() | |
| except JobError as exc: | |
| return [json.dumps({"error": str(exc), "hint": exc.hint})] | |
| def _mcp_image(path: Path, max_size: int) -> MCPImage: | |
| from PIL import Image | |
| im = Image.open(path).convert("RGB") | |
| im.thumbnail((max_size, max_size)) | |
| buf = io.BytesIO() | |
| im.save(buf, format="JPEG", quality=88) | |
| return MCPImage(data=buf.getvalue(), format="jpeg") | |
| def view_image(job_id: str, path: str = "renders/contact_sheet.png", max_size: int = 1024) -> list: | |
| """Look at an image of a job: e.g. renders/contact_sheet.png, renders/comparison.png, renders/views_sheet.png, | |
| renders/orbit03.png, bake/albedo.png, views/<name>/generated.png, layers/<name>/compare.png.""" | |
| try: | |
| f = _safe_file(MANAGER.get(job_id), path) | |
| return [_mcp_image(f, max(128, min(max_size, 2048)))] | |
| except JobError as exc: | |
| return [json.dumps({"error": str(exc), "hint": exc.hint})] | |
| def regenerate_views(job_id: str, views: list[str], prompt_extra: str = "", backend: str = "", | |
| hf_token: str = "", ctx: Context | None = None) -> dict: | |
| """Regenerate some views with new seeds (e.g. those listed as flagged or rejected in get_job), then re-bake. | |
| A regeneration that fails validation keeps the previous image. Terrain jobs: views are parts, 'overview', | |
| 'tile_<row>_<col>' or 'layer:<name>'.""" | |
| def run(): | |
| if MANAGER.get(job_id).params.get("kind") == "terrain": | |
| return status_payload(MANAGER.regenerate_terrain(job_id, views, _mcp_ident(ctx, hf_token or None), | |
| backend or None), base_url(None)) | |
| return status_payload(MANAGER.regenerate(job_id, views, None, prompt_extra or None, backend or None, | |
| _mcp_ident(ctx, hf_token or None)), base_url(None)) | |
| return _mcp_call(run) | |
| def disable_views(job_id: str, views: list[str]) -> dict: | |
| """Exclude views from the texture (e.g. still flagged after regeneration) and re-bake.""" | |
| return _mcp_call(lambda: status_payload(MANAGER.disable_views(job_id, views), base_url(None))) | |
| def fill_gaps(job_id: str, max_views: int = 3, hf_token: str = "", ctx: Context | None = None) -> dict: | |
| """Add cameras aimed at visible surface that is still uncovered, generate them and re-bake.""" | |
| return _mcp_call(lambda: status_payload(MANAGER.fill_gaps(job_id, max_views, _mcp_ident(ctx, hf_token or None)), | |
| base_url(None))) | |
| def retouch(job_id: str, suggestion: int = -1, mode: str = "", prompt: str = "", view: str = "", | |
| render_camera: str = "", mask: Any = None, focus: dict | None = None, capture_size: int = 0, | |
| output_size: int = 0, backend: str = "", hf_token: str = "", dry_run: bool = False, | |
| ctx: Context | None = None) -> dict: | |
| """Local inpainting inside a feathered mask; the rest of the texture is untouched. Easiest: suggestion=0 (from | |
| get_job's retouch_suggestions). mode: repair | replace | upscale (capture_size=256, output_size=1024). | |
| Camera: view, render_camera ('orbit03') or focus {point, radius, direction}. mask: 'all' | 'uncovered' | | |
| 'low_quality' | {'sphere':[x,y,z,r]} | {'rect':[x0,y0,x1,y1]} | {'circle':[cx,cy,r]}. dry_run=true: only | |
| returns aim_url, the capture with the mask outlined (seconds, nothing generated): check the aim first.""" | |
| args = {k: v for k, v in dict(mode=mode or None, prompt=prompt, view=view or None, | |
| render_camera=render_camera or None, mask=mask, focus=focus, | |
| capture_size=capture_size or None, output_size=output_size or None, | |
| backend=backend or None).items() if v not in (None, "")} | |
| if suggestion >= 0: | |
| args["suggestion"] = suggestion | |
| if dry_run: | |
| return _mcp_call(lambda: aim_payload(job_id, MANAGER.retouch_aim(job_id, args), base_url(None))) | |
| return _mcp_call(lambda: status_payload(MANAGER.retouch(job_id, args, _mcp_ident(ctx, hf_token or None)), | |
| base_url(None))) | |
| def undo_retouch(job_id: str) -> dict: | |
| """Disable the most recent retouch layer and re-bake.""" | |
| return _mcp_call(lambda: status_payload(MANAGER.undo_retouch(job_id), base_url(None))) | |
| def bake(job_id: str, resolution: int = 2048, single_mesh: bool | None = None) -> dict: | |
| """Re-bake the texture at 1024, 2048 or 4096 and re-export the GLB (single_mesh: one object and one material, | |
| false: your objects and materials, omitted: unchanged).""" | |
| return _mcp_call(lambda: status_payload(MANAGER.bake(job_id, resolution, single_mesh), base_url(None))) | |
| def pick_point(job_id: str, pixels: list[list[float]], view: str = "", render_camera: str = "") -> dict: | |
| """3D point, normal and UV under image pixels [[x, y], ...] (pixels or 0-1 fractions) of a view or a preview | |
| render camera (e.g. 'orbit02'). Use it to aim retouch (focus / sphere mask) at something seen in an image.""" | |
| return _mcp_call(lambda: MANAGER.pick(job_id, pixels, view or None, render_camera or None)) | |
| def submit_view_image(job_id: str, view: str, image_url: str = "", image_base64: str = "") -> dict: | |
| """Backend 'external': upload your generated image for the pending view (see get_job.pending_view), as a public | |
| URL or base64 PNG/JPEG. It must match the conditioning render's framing. The job then continues.""" | |
| return _mcp_call(lambda: status_payload(MANAGER.submit_image(job_id, view, _image_bytes(image_url, image_base64), | |
| Identity()), base_url(None))) | |
| def service_info() -> dict: | |
| """Samples, backends (speed, quota, licence), limits and the recommended workflow.""" | |
| return _info() | |
| def guide() -> str: | |
| """The full agent guide (llms.txt): workflow, parameters, quality loop, troubleshooting.""" | |
| return LLMS_TXT | |
| def _image_bytes(image_url: str | None, image_base64: str | None) -> bytes: | |
| if image_base64: | |
| b = image_base64.split(",", 1)[1] if image_base64.startswith("data:") else image_base64 | |
| return base64.b64decode(b) | |
| if image_url: | |
| MANAGER._check_url(image_url) | |
| r = requests.get(image_url, timeout=60) | |
| r.raise_for_status() | |
| if len(r.content) > 30 * 1024 * 1024: | |
| raise JobError("image too large", 413) | |
| return r.content | |
| raise JobError("give image_url or image_base64 (or a multipart file)", 400) | |
| def _info() -> dict: | |
| return { | |
| "service": "projtex: projection texturing for 3D meshes (generate views -> project -> blend -> bake to UVs)", | |
| "samples": SAMPLE_INFO, | |
| "backends": BACKENDS, | |
| "quota": "qwen-turbo, qwen-space and fibo-space run on ZeroGPU Spaces and spend the caller's daily GPU " | |
| "quota (unauthenticated ~2 min, free account ~5 min, PRO ~40 min per day; ~7 s of GPU per view " | |
| "with qwen-turbo, ~30 s with qwen-space). Send your HF " | |
| "token (header X-HF-Token or hf_token field) so your own quota is used; otherwise the request runs " | |
| "as your forwarded visitor identity or anonymously.", | |
| "limits": {"mesh_mb": MAX_MESH_BYTES >> 20, "triangles": MAX_FACES, "texture_size": [1024, 2048, 4096], | |
| "concurrent_operations": MAX_RUNNING, "job_ttl_hours": 12}, | |
| "terrain": {"endpoint": "POST /api/terrain (MCP texture_terrain)", | |
| "about": "ground layers (splat maps + seamless detail textures) and a large top-down colour map " | |
| "painted tile by tile and projected onto a terrain mesh or heightmap; returns " | |
| "terrain.glb plus a bundle with Blender / three.js materials", | |
| "presets": {k: v["about"] for k, v in __import__("projtex.terrain", fromlist=["x"]).PRESETS.items()}, | |
| "backends": list(TERRAIN_BACKENDS), "macro_size": [1024, 2048, 4096], | |
| "images_per_job": "layers + 1 + (9 tiles at 2048 | 25 at 4096)"}, | |
| "workflow": ["local file: POST /api/uploads with the file as raw body -> upload_id", | |
| "POST /api/jobs {upload_id | mesh_url | sample, prompt} (or MCP texture_mesh) -> job id", | |
| "GET /api/jobs/{id} until state=done", | |
| "look at files.contact_sheet / comparison", "follow 'next' (regenerate, fill-gaps, retouch)", | |
| "download files.textured_glb"], | |
| "docs": {"openapi": "/docs", "guide": "/llms.txt", "mcp": "/mcp"}, | |
| "source": "projtex (headless projection texturing); runs the projection, fusion and bake on this CPU " | |
| "machine with software OpenGL", | |
| } | |
| # -------------------------------------------------------------------------------------------- FastAPI | |
| MCP_APP = mcp.streamable_http_app(streamable_http_path="/mcp", stateless_http=True, json_response=True, | |
| transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False)) | |
| async def lifespan(app: FastAPI): | |
| async with mcp.session_manager.run(): | |
| yield | |
| app = FastAPI( | |
| title="projtex: projection texturing API", | |
| version="1.0", | |
| lifespan=lifespan, | |
| description="Texture a 3D mesh by generating images from virtual cameras (Qwen-Image-2.1 via the official " | |
| "Qwen Space, or Bria Fibo 2.0 Edit Turbo), projecting them onto the geometry, blending in texture " | |
| "space and baking to the mesh's UVs. Designed for agents: every long operation is an asynchronous " | |
| "job; poll GET /api/jobs/{id} and follow its `next` list. Agent guide: /llms.txt. MCP: /mcp.", | |
| ) | |
| app.router.routes.extend(MCP_APP.routes) # MCP streamable HTTP served exactly at /mcp (no redirect) | |
| async def job_error_handler(request: Request, exc: JobError): | |
| return JSONResponse(status_code=exc.status, content={"error": str(exc), "hint": exc.hint}) | |
| def landing() -> HTMLResponse: | |
| return HTMLResponse((HERE / "static" / "index.html").read_text()) | |
| def llms() -> PlainTextResponse: | |
| return PlainTextResponse(LLMS_TXT) | |
| def static(name: str): | |
| f = (HERE / "static" / name).resolve() | |
| if f.parent != (HERE / "static").resolve() or not f.exists(): | |
| raise HTTPException(404) | |
| return FileResponse(f) | |
| def health() -> dict: | |
| return {"ok": True, "jobs": len(MANAGER.jobs), "busy": MANAGER.pending} | |
| def api_info() -> dict: | |
| return _info() | |
| def api_identity(request: Request) -> dict: | |
| ident = identity_from({k.lower(): v for k, v in request.headers.items()}) | |
| return {"identity": ident.kind, "x_ip_token_present": bool(ident.ip_token), | |
| "token_present": bool(ident.token)} | |
| def create_job(req: TextureRequest, request: Request) -> dict: | |
| params = req.model_dump(exclude_none=True, exclude={"hf_token", "mesh_base64"}) | |
| try: | |
| job = MANAGER.create(params, identity_from({k.lower(): v for k, v in request.headers.items()}, req.hf_token), | |
| upload=_decode_mesh(req.mesh_base64)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| return status_payload(job, base_url(request)) | |
| def create_terrain(req: TerrainRequest, request: Request) -> dict: | |
| """The ground is a mesh (upload_id / mesh_url / mesh_base64, +Y up) or a heightmap. Layers (or a preset) say | |
| which ground goes where; the result is terrain.glb (with the colour map) plus a bundle with the splat maps, | |
| seamless detail textures and ready-made Blender / three.js materials. Poll GET /api/jobs/{id}.""" | |
| params = req.model_dump(exclude_none=True, exclude={"hf_token", "mesh_base64"}) | |
| hm = params.get("heightmap") or {} | |
| hm_data = None | |
| if req.heightmap and req.heightmap.base64: | |
| hm_data = ("heightmap", _decode_mesh(req.heightmap.base64)[1]) | |
| hm.pop("base64", None) | |
| try: | |
| job = MANAGER.create_terrain(params, identity_from(_hdr(request), req.hf_token), | |
| upload=_decode_mesh(req.mesh_base64), heightmap_data=hm_data) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| return status_payload(job, base_url(request)) | |
| def _decode_mesh(b64: str | None) -> tuple[str, bytes] | None: | |
| if not b64: | |
| return None | |
| if b64.startswith("data:"): | |
| b64 = b64.split(",", 1)[1] | |
| try: | |
| data = base64.b64decode(b64, validate=False) | |
| except Exception as exc: | |
| raise JobError(f"mesh_base64 is not valid base64: {exc}", 400) | |
| if len(data) > 20 * 1024 * 1024: | |
| raise JobError("mesh_base64 is limited to 20 MB", 413, "use POST /api/uploads for larger files") | |
| return ("mesh", data) | |
| async def upload_mesh(request: Request, filename: str | None = None) -> dict: | |
| """Send the file bytes as the body, e.g. `curl --data-binary @house.glb -H 'content-type: model/gltf-binary' | |
| URL/api/uploads`. Then create the job with JSON: POST /api/jobs {"upload_id": ..., "prompt": ...}.""" | |
| ctype = request.headers.get("content-type", "") | |
| if ctype.startswith("multipart/form-data"): | |
| form = await request.form() | |
| f = form.get("file") or form.get("mesh") | |
| if f is None or isinstance(f, str): | |
| raise JobError("multipart upload without a 'file' field", 400, "send the mesh as field 'file', or as " | |
| "the raw request body") | |
| data, name = await f.read(), f.filename | |
| else: | |
| data, name = await request.body(), filename | |
| up = MANAGER.store_upload(data, name) | |
| base = base_url(request) | |
| if up["filename"].lower().endswith((".png", ".npy")): # a heightmap | |
| up["next"] = [{"do": "create a terrain job from this heightmap", "method": "POST", "url": f"{base}/api/terrain", | |
| "body": {"heightmap": {"upload_id": up["upload_id"], "size_m": "<width in metres>", | |
| "height_m": "<black-to-white height in metres>"}, | |
| "prompt": "<the place, e.g. an alpine valley in early summer>"}}] | |
| return up | |
| up["next"] = [{"do": "texture an object (a prop, building, vehicle, character)", "method": "POST", | |
| "url": f"{base}/api/jobs", | |
| "body": {"upload_id": up["upload_id"], "prompt": "<materials and colours>"}}, | |
| {"do": "or, if this is a terrain / the ground of a scene (with layers): create a terrain job", | |
| "method": "POST", | |
| "url": f"{base}/api/terrain", | |
| "body": {"upload_id": up["upload_id"], "prompt": "<the place, e.g. rolling farmland in summer>"}}] | |
| return up | |
| async def create_job_upload(request: Request, file: UploadFile = File(...), | |
| params: str = Form("{}", description="TextureRequest fields as JSON (without mesh). " | |
| "Fields may also be sent as separate form fields.")): | |
| try: | |
| prm_d = json.loads(params or "{}") | |
| except json.JSONDecodeError as exc: | |
| raise JobError(f"params is not valid JSON ({exc}); received {len(params)} characters ending with " | |
| f"{params[-40:]!r}", 400, | |
| "curl -F cuts a value at ';' (and treats ',' and quotes specially): send it with --form-string " | |
| "'params=...' or read it from a file with -F 'params=<params.json'. Simplest: upload the bytes " | |
| "with POST /api/uploads, then create the job with JSON (POST /api/jobs {\"upload_id\": ...}).") | |
| if not isinstance(prm_d, dict): | |
| raise JobError("params must be a JSON object", 400) | |
| form = await request.form() | |
| for k in TextureRequest.model_fields: | |
| if k in form and isinstance(form[k], str): | |
| prm_d[k] = form[k] | |
| try: | |
| prm = TextureRequest(**prm_d) | |
| except Exception as exc: | |
| raise JobError(f"invalid parameters: {exc}", 400, "see the TextureRequest schema at /docs") | |
| data = await file.read() | |
| try: | |
| job = MANAGER.create(prm.model_dump(exclude_none=True, exclude={"hf_token"}), | |
| identity_from({k.lower(): v for k, v in request.headers.items()}, prm.hf_token), | |
| upload=(file.filename or "mesh.glb", data)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| return status_payload(job, base_url(request)) | |
| def get_job_status(job_id: str, request: Request, events: int = 12) -> dict: | |
| try: | |
| return status_payload(MANAGER.get(job_id), base_url(request), events) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def _hdr(request: Request) -> dict: | |
| return {k.lower(): v for k, v in request.headers.items()} | |
| def api_regenerate(job_id: str, req: RegenerateRequest, request: Request) -> dict: | |
| try: | |
| if MANAGER.get(job_id).params.get("kind") == "terrain": | |
| job = MANAGER.regenerate_terrain(job_id, req.views, identity_from(_hdr(request), req.hf_token), | |
| req.backend) | |
| return status_payload(job, base_url(request)) | |
| job = MANAGER.regenerate(job_id, req.views, req.seeds, req.prompt_extra, req.backend, | |
| identity_from(_hdr(request), req.hf_token)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| return status_payload(job, base_url(request)) | |
| def api_disable(job_id: str, req: ViewsRequest, request: Request) -> dict: | |
| try: | |
| return status_payload(MANAGER.disable_views(job_id, req.views), base_url(request)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def api_fill_gaps(job_id: str, req: FillGapsRequest, request: Request) -> dict: | |
| try: | |
| return status_payload(MANAGER.fill_gaps(job_id, req.max_views, identity_from(_hdr(request), req.hf_token)), | |
| base_url(request)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def aim_payload(job_id: str, r: dict, base: str) -> dict: | |
| url = f"{base}/api/jobs/{job_id}/files/{r['aim']}?t={int(time.time())}" | |
| return {"dry_run": True, "aim_url": url, "mask_pixels": r.get("mask_pixels"), | |
| "next": [{"do": "look at aim_url: the red outline is where the retouch will paint; when it is right, " | |
| "send the same request without dry_run", "method": "GET", "url": url}]} | |
| def api_retouch(job_id: str, req: RetouchRequest, request: Request, response: Response) -> dict: | |
| args = req.model_dump(exclude_none=True, exclude={"hf_token", "dry_run"}) | |
| if req.dry_run: # answered at once: 200, not the 202 of a queued operation | |
| try: | |
| r = aim_payload(job_id, MANAGER.retouch_aim(job_id, args), base_url(request)) | |
| response.status_code = 200 | |
| return r | |
| except JobError as exc: | |
| raise http_error(exc) | |
| try: | |
| return status_payload(MANAGER.retouch(job_id, args, identity_from(_hdr(request), req.hf_token)), | |
| base_url(request)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def api_undo(job_id: str, request: Request) -> dict: | |
| try: | |
| return status_payload(MANAGER.undo_retouch(job_id), base_url(request)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def api_bake(job_id: str, req: BakeRequest, request: Request) -> dict: | |
| try: | |
| return status_payload(MANAGER.bake(job_id, req.resolution, req.single_mesh), base_url(request)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def api_pbr(job_id: str, request: Request, req: PbrRequest | None = None) -> dict: | |
| try: | |
| ident = identity_from(_hdr(request), req.hf_token if req else None) | |
| return status_payload(MANAGER.pbr(job_id, ident), base_url(request)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def api_pick(job_id: str, req: PickRequest) -> dict: | |
| try: | |
| return MANAGER.pick(job_id, req.pixels, req.view, req.render_camera) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| async def api_submit_image(job_id: str, view: str, request: Request, file: UploadFile | None = File(None)): | |
| try: | |
| if file is not None: | |
| data = await file.read() | |
| else: | |
| body = ImageSubmit(**(await request.json())) | |
| data = await run_in_threadpool(_image_bytes, body.image_url, body.image_base64) # may download | |
| job = await run_in_threadpool(MANAGER.submit_image, job_id, view, data, Identity()) | |
| return status_payload(job, base_url(request)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def _safe_file(job, path: str) -> Path: | |
| root = job.root.resolve() | |
| f = (root / path).resolve() | |
| allowed = ("bake/", "renders/", "views/", "layers/", "inspect/", "exchange/", "export/", "aim/", "macro/", | |
| "splat/", "maps/") | |
| rel = str(f.relative_to(root)) if str(f).startswith(str(root)) else "" | |
| if not rel or not rel.startswith(allowed) or not f.is_file(): | |
| raise JobError(f"no file {path!r}", 404, "see 'files' and 'views' in the job status") | |
| return f | |
| def api_file(job_id: str, path: str): | |
| try: | |
| return FileResponse(_safe_file(MANAGER.get(job_id), path)) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| def api_delete(job_id: str) -> dict: | |
| try: | |
| MANAGER.delete(job_id) | |
| except JobError as exc: | |
| raise http_error(exc) | |
| return {"deleted": job_id} | |