Files
settled-reach/tooling/db/trellis_connector.py
T
jpmschweitzerandClaude Opus 4.6 e836be1ad4 feat(skills): add glb-gen skill and Trellis/image connectors
New skill for converting concept images to game-ready .glb models via
Trellis (image-to-3D) and Blender post-processing (scale normalization,
material setup, recolor mask generation).

Connectors: trellis_connector.py (Gradio API), image_connector.py
(Gemini API for concept art).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-17 23:33:49 +01:00

315 lines
10 KiB
Python
Executable File

#!/usr/bin/env python3
"""
Trellis 3D model generator connector — Gradio API wrapper.
Talks to the Trellis Gradio app at tower-of-joy:11510.
Pipeline: upload image → start session → image_to_3d → extract_glb → download .glb
Usage:
python3 trellis_connector.py health
python3 trellis_connector.py generate image.png [--output model.glb] [--simplify 0.95] [--texture-size 1024] [--seed 42] [--timeout 600]
"""
import base64
import json
import os
import shutil
import sys
import time
import urllib.error
import urllib.request
import urllib.parse
CONFIG_PATH = os.path.join(os.path.dirname(__file__), "config.json")
def load_config():
with open(CONFIG_PATH) as f:
return json.load(f)
def get_base_url():
config = load_config()
return config.get("trellis_url", "http://tower-of-joy:11510")
def health():
"""Check if the Trellis API is reachable."""
base = get_base_url()
try:
req = urllib.request.Request(f"{base}/info", method="GET")
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
endpoints = list(data.get("named_endpoints", {}).keys())
print(json.dumps({
"ok": True,
"url": base,
"endpoints": endpoints
}, indent=2))
except Exception as e:
print(json.dumps({
"ok": False,
"url": base,
"error": str(e)
}, indent=2))
sys.exit(1)
def _call_api(base, endpoint, data, timeout=600):
"""Call a Gradio API endpoint. Tries sync /api/ first, falls back to SSE /gradio_api/call/."""
# Trellis uses the sync /api/ pattern
api_url = f"{base}/api{endpoint}"
payload = json.dumps({"data": data})
req = urllib.request.Request(
api_url,
data=payload.encode(),
headers={"Content-Type": "application/json"},
method="POST"
)
print(f" Calling {endpoint}...", file=sys.stderr)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
result = json.loads(resp.read())
# Sync Gradio returns {"data": [...], "is_generating": false, ...}
if isinstance(result, dict) and "data" in result:
return result["data"]
return result
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8", errors="replace")
raise RuntimeError(f"{endpoint} failed ({e.code}): {body[:300]}")
def _upload_image(base, image_path):
"""Upload an image file to the Gradio server and return the file reference."""
upload_url = f"{base}/upload"
with open(image_path, "rb") as f:
image_data = f.read()
filename = os.path.basename(image_path)
# Gradio upload expects multipart/form-data with a 'files' field
boundary = "----TrellisConnectorBoundary"
body = (
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="files"; filename="{filename}"\r\n'
f"Content-Type: image/png\r\n"
f"\r\n"
).encode() + image_data + f"\r\n--{boundary}--\r\n".encode()
req = urllib.request.Request(
upload_url,
data=body,
headers={
"Content-Type": f"multipart/form-data; boundary={boundary}",
},
method="POST"
)
print(f" Uploading {filename}...", file=sys.stderr)
with urllib.request.urlopen(req, timeout=30) as resp:
result = json.loads(resp.read())
# Gradio returns a list of uploaded file paths
if isinstance(result, list) and len(result) > 0:
return result[0]
raise RuntimeError(f"Upload failed: {result}")
def _download_file(url, output_path, base):
"""Download a file from the Gradio server."""
if url.startswith("/"):
url = f"{base}{url}"
elif not url.startswith("http"):
url = f"{base}/file={url}"
print(f" Downloading to {output_path}...", file=sys.stderr)
req = urllib.request.Request(url, method="GET")
with urllib.request.urlopen(req, timeout=120) as resp:
with open(output_path, "wb") as f:
shutil.copyfileobj(resp, f)
return os.path.getsize(output_path)
def _check_available(base):
"""Quick check if Trellis is reachable. Fail fast with a clear message."""
try:
req = urllib.request.Request(f"{base}/info", method="GET")
urllib.request.urlopen(req, timeout=5)
except Exception:
print(json.dumps({
"ok": False,
"error": f"Trellis is not available at {base}. The service may be switched off to save system resources. Start it before generating 3D models."
}, indent=2))
sys.exit(1)
def generate(image_path, output=None, simplify=0.95, texture_size=1024,
seed=0, timeout=600):
"""
Generate a 3D model from an image.
Pipeline:
1. Start session
2. Upload and preprocess image
3. Generate 3D from image
4. Extract GLB
5. Download GLB file
Args:
image_path: Path to the input image (PNG recommended)
output: Output .glb file path (default: auto-named)
simplify: Mesh simplification factor (0.9-0.98, default 0.95)
texture_size: Texture resolution (512-2048, default 1024)
seed: Random seed (default 0)
timeout: Max wait time per step in seconds
"""
base = get_base_url()
_check_available(base)
start_time = time.time()
if not os.path.isfile(image_path):
print(json.dumps({"ok": False, "error": f"Image not found: {image_path}"}), indent=2)
sys.exit(1)
if output is None:
name = os.path.splitext(os.path.basename(image_path))[0]
output = f"{name}.glb"
# Step 1: Start session
print("Step 1/5: Starting session...", file=sys.stderr)
_call_api(base, "/start_session", [], timeout=30)
# Step 2: Upload and preprocess image
print("Step 2/5: Uploading and preprocessing image...", file=sys.stderr)
uploaded_path = _upload_image(base, image_path)
file_ref = {
"path": uploaded_path,
"meta": {"_type": "gradio.FileData"}
}
preprocess_result = _call_api(base, "/preprocess_image_1", [file_ref], timeout=60)
# _call_api returns the "data" array directly
if isinstance(preprocess_result, list) and len(preprocess_result) > 0:
preprocessed_ref = preprocess_result[0]
else:
preprocessed_ref = preprocess_result
# Step 3: Get seed
print("Step 3/5: Generating 3D model...", file=sys.stderr)
seed_result = _call_api(base, "/get_seed", [True, seed], timeout=10)
if isinstance(seed_result, list) and seed_result:
actual_seed = seed_result[0]
else:
actual_seed = seed
# Step 4: Image to 3D
# Parameters: image, multiimages, seed, ss_guidance, ss_steps, slat_guidance, slat_steps, algo
gen_result = _call_api(base, "/image_to_3d", [
preprocessed_ref, # image
[], # multiimages (empty)
actual_seed, # seed
7.5, # ss_guidance_strength
12, # ss_sampling_steps
3.0, # slat_guidance_strength
12, # slat_sampling_steps
"stochastic", # multiimage_algo
], timeout=timeout)
# Step 5: Extract GLB
print("Step 4/5: Extracting GLB...", file=sys.stderr)
glb_result = _call_api(base, "/extract_glb", [simplify, texture_size], timeout=120)
# _call_api returns the "data" array: [model_viewer_data, download_button_data]
glb_url = None
if isinstance(glb_result, list):
for item in glb_result:
if isinstance(item, dict):
url = item.get("url") or item.get("path")
if url:
glb_url = url
break
if not glb_url:
print(json.dumps({
"ok": False,
"error": "Could not extract GLB URL from response",
"response": glb_result
}, indent=2))
sys.exit(1)
# Step 6: Download
print("Step 5/5: Downloading GLB...", file=sys.stderr)
os.makedirs(os.path.dirname(os.path.abspath(output)), exist_ok=True)
file_size = _download_file(glb_url, output, base)
elapsed = round(time.time() - start_time, 1)
print(json.dumps({
"ok": True,
"file": output,
"size_bytes": file_size,
"simplify": simplify,
"texture_size": texture_size,
"seed": actual_seed,
"generation_time_s": elapsed,
"source_image": image_path
}, indent=2))
def main():
if len(sys.argv) < 2:
print("Usage:")
print(" trellis_connector.py health")
print(" trellis_connector.py generate image.png [--output model.glb] [--simplify 0.95] [--texture-size 1024] [--seed N] [--timeout N]")
sys.exit(1)
cmd = sys.argv[1]
if cmd == "health":
health()
elif cmd == "generate":
if len(sys.argv) < 3:
print("Error: image path required", file=sys.stderr)
sys.exit(1)
image_path = sys.argv[2]
output = None
simplify = 0.95
texture_size = 1024
seed = 0
timeout = 600
i = 3
while i < len(sys.argv):
if sys.argv[i] == "--output" and i + 1 < len(sys.argv):
output = sys.argv[i + 1]
i += 2
elif sys.argv[i] == "--simplify" and i + 1 < len(sys.argv):
simplify = float(sys.argv[i + 1])
i += 2
elif sys.argv[i] == "--texture-size" and i + 1 < len(sys.argv):
texture_size = int(sys.argv[i + 1])
i += 2
elif sys.argv[i] == "--seed" and i + 1 < len(sys.argv):
seed = int(sys.argv[i + 1])
i += 2
elif sys.argv[i] == "--timeout" and i + 1 < len(sys.argv):
timeout = int(sys.argv[i + 1])
i += 2
else:
print(f"Unknown argument: {sys.argv[i]}", file=sys.stderr)
sys.exit(1)
generate(image_path, output=output, simplify=simplify,
texture_size=texture_size, seed=seed, timeout=timeout)
else:
print(f"Unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()