#!/usr/bin/env python3 """WARDROBEGOD — one skeleton, infinite fits. The wardrobe generator bench. Stdlib server + three.js page (no build step, house style — sibling of NPCFACTORY/imagelab). python3 server.py → http://localhost:8150 WG_HOST=0.0.0.0 python3 server.py → expose on the tailnet What it does: · library panes over bodies / garments / generated images / finished outfits · loads FBX or GLB on a 3D stage (FBX auto-converts to GLB via local headless Blender) · Blender jobs: convert · scale-to-height · decimate-keep-weights · FIT (weight-transfer a garment onto a rigged body) · ASSEMBLE (body + fitted garments → one dressed GLB) · the garment GENERATOR pipeline (MODELBEAST, $0): flux image → bg_remove → either a hanging-garment TEXTURE (racks/paper-doll) or trellis_mac → rigid 3D wearable (hats/shoes/bags — bone-parent those; cloth that must bend gets FIT instead) Env: MB_HOST (default m3ultra :8777) · MB_TOKEN (bearer; generator disabled without it) WG_PORT (8150) · WG_HOST (127.0.0.1) """ import hashlib, json, mimetypes, os, re, shutil, subprocess, sys, threading, time, urllib.parse, urllib.request, uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ROOT = os.path.dirname(os.path.abspath(__file__)) LIB = os.path.join(ROOT, 'library') DIRS = {'bodies': os.path.join(LIB, 'bodies'), 'garments': os.path.join(LIB, 'garments'), 'gen': os.path.join(LIB, 'gen'), 'out': os.path.join(ROOT, 'out')} OUTFITS = os.path.join(LIB, 'outfits') # saved dress-ups: body + attachments + clip + tints THUMBS = os.path.join(ROOT, '.thumbs') # lazy 256px previews, keyed on source path hash MANIFEST = os.path.join(LIB, 'index.json') # slot / layer / tags per asset — the wardrobe's spine EXTRA_BODIES = [os.path.expanduser('~/Documents/anatomy'), os.path.expanduser('~/Documents/thriftgod/web/assets/models')] CACHE = os.path.join(ROOT, '.glbcache') BLENDER = os.environ.get('WG_BLENDER', '/Applications/Blender.app/Contents/MacOS/Blender') OPS = os.path.join(ROOT, 'blender_ops.py') TOOLS = os.path.join(ROOT, 'tools') # reskin trio (NPCFACTORY) + the doll compositor (90sDJsim) MIRPAMO = os.environ.get('WG_MIRPAMO', os.path.expanduser('~/Documents/MIRPAMO/bin/mirpamo')) # The 3GOD depot MUST be addressed direct over the tailnet. Its auth trusts the raw socket peer # against an allow-list, so through the digalot.fyi Cloudflare front the peer is CF and every # write 403s — measured: /api/list reports authed=false via CF, authed=true direct. (The shipped # prop_campaign.py --publish defaults to the CF URL and is broken for this exact reason.) GOD3 = os.environ.get('WG_GOD3', 'http://100.94.195.115:8788') # Verified against each consumer's actual loader code, not its docs. "ready" means a file drop # alone makes the asset appear; everything else needs a registry edit first, and exporting # without one produces a file nobody loads. EXPORT_TARGETS = { 'djsim-doll': {'ready': True, 'why': 'djsim resolves doll art by filename and self-heals on restart'}, '3god': {'ready': True, 'why': 'HTTP depot; thriftgod and procity both read their props from it'}, 'djsim-ped': {'ready': False, 'why': 'peds load from a hardcoded array, not a directory scan', 'registry': 'ultra 90sDJsim/web/world/crowd/rigs.js PED_NAMES (:20-27) — the ' 'array that owns the street; index.html:639 has a second copy'}, 'procity-ped': {'ready': False, 'why': 'same pattern — loadPedFleet reads a fixed name list', 'registry': 'PROCITY/web/models/peds/ + the ped name array; publish props via ' 'pipeline/publish.py on m3ultra (the only box with _normalized/)'}, 'thriftgod-prop': {'ready': False, 'why': 'thriftgod has NO garment consumer; props are hardcoded hero lists', 'registry': 'thriftgod/web/index.html HERO_FLOOR / HERO_COUNTER (:1322-1325) ' '— the filename must match a literal string exactly'}, 'not-tonight': {'ready': False, 'why': 'only ingests public/props/.png and requires a manifest entry; ' 'there is no doll, garment or character art path at all', 'registry': 'not-tonight/public/props/manifest.json (+ tools/props_import.py)'}, } def god3_name(name): """Reproduce the depot's clean_name: illegal characters are DELETED, not substituted. That deletion is the hazard — 'shop!-cat.glb' cleans to 'shop-cat.glb' and silently serves an existing, different mesh. Case is significant and spaces/underscores survive verbatim. """ n = re.sub(r'[^A-Za-z0-9._ -]', '', os.path.basename(name)) n = re.sub(r'^[^A-Za-z0-9]+', '', n) return n or 'asset.glb' # Creds come off disk, never off the command line. .env beside server.py (gitignored) holds the # CF Workers AI pair; MB_TOKEN lives in backnforth/.env — read it there rather than making every # launch remember `export MB_TOKEN=…` (without it the farm tiers silently die: no RMBG-2.0, no # image-edit try-on, no trellis). First file to define a key wins, so a local .env always overrides. def _load_env(path): if not os.path.exists(path): return for line in open(path): if '=' in line and not line.lstrip().startswith('#'): k, v = line.split('=', 1) os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) _load_env(os.path.join(ROOT, '.env')) _load_env(os.path.expanduser('~/Documents/backnforth/.env')) KIT_LIB = os.path.expanduser('~/Documents/character_kit/modular/parts_library/library.json') MB = os.environ.get('MB_HOST', 'http://100.89.131.57:8777') MB_TOKEN = os.environ.get('MB_TOKEN', '') CF_ACCT = os.environ.get('CLOUDFLARE_ACCOUNT_ID', '') CF_TOKEN = os.environ.get('CLOUDFLARE_API_TOKEN', '') # 90sDJsim's doll pipeline and its live art live on ultra — export writes over the tailnet. DJSIM_HOST = os.environ.get('WG_DJSIM_HOST', 'johnking@100.91.239.7') DJSIM_DOLL = os.environ.get('WG_DJSIM_DOLL', 'Documents/90sDJsim/web/world/media/doll') PORT = int(os.environ.get('WG_PORT', 8150)) HOST = os.environ.get('WG_HOST', '127.0.0.1') # POST body cap - tiny stdlib server, do not let a peer stream unbounded bytes at it MAX_BODY = int(os.environ.get('WG_MAX_BODY', 512 * 1024 * 1024)) for d in list(DIRS.values()) + [CACHE, OUTFITS]: os.makedirs(d, exist_ok=True) sys.path.insert(0, os.path.join(ROOT, 'tools')) try: # the 2D paper-doll tier (PIL + numpy); 3D still works without it import doll as DOLL except Exception as _e: DOLL = None print(f'2D doll tier off ({_e.__class__.__name__}: {_e}) — pip install pillow numpy to enable') try: # registry edits into consumers that load from hardcoded lists import registry as REG except Exception as _e: REG = None print(f'registry edits off ({_e.__class__.__name__}: {_e})') JOBS = {} # id → {status, note, out, log} MODEL_EXT = ('.glb', '.gltf', '.fbx', '.obj') IMG_EXT = ('.png', '.jpg', '.jpeg', '.webp') def slug(s): return re.sub(r'[^a-z0-9]+', '-', (s or '').lower()).strip('-')[:60] or 'x' # Asset classification lives in the FILENAME, matching the convention already in use across the # fleet (~/Documents/FBX: a '-nsfw' suffix before the extension, 106 of 124 files tagged, plus an # `nsfw` column in _INVENTORY.csv). Filenames beat a sidecar database here because assets get # rsynced between six machines constantly — a tag in the name travels with the file and can't # drift out of sync, and it's visible in Finder. NSFW_RE = re.compile(r'-nsfw(?=[.\s(]|$)', re.I) # -swim is the middle tier: swimwear/underwear/lingerie. Clothed, so it isn't # nsfw, but a bikini body is still wrong for a shopfront NPC or a kids' scene, # so consumers that ask for "dressed" shouldn't silently get one. SWIM_RE = re.compile(r'-swim(?=[.\s(]|$)', re.I) def is_nsfw(path): return bool(NSFW_RE.search(os.path.basename(path))) def is_swim(path): return bool(SWIM_RE.search(os.path.basename(path))) def rating(path): """Strictest tier that applies: nsfw > swim > safe.""" return 'nsfw' if is_nsfw(path) else 'swim' if is_swim(path) else 'safe' def tagged_name(name, nsfw, swim=False): """Set the content-tier suffix, preserving the extension. Both tags are stripped first, so this moves an asset between tiers rather than stacking suffixes. nsfw wins if both are asked for.""" stem, dot, ext = name.rpartition('.') if not dot: stem, ext = name, '' stem = SWIM_RE.sub('', NSFW_RE.sub('', stem)).rstrip('-') if nsfw: stem += '-nsfw' elif swim: stem += '-swim' return stem + (dot + ext if dot else '') def allowed(path): p = os.path.realpath(path) roots = list(DIRS.values()) + EXTRA_BODIES + [CACHE, OUTFITS, os.path.join(LIB, 'doll')] return any(p.startswith(os.path.realpath(r) + os.sep) or p == os.path.realpath(r) for r in roots) def scan(): out = {} for key, d in DIRS.items(): # garments hold BOTH kinds: fitted/harvested GLBs and flat cut-out PNGs (the paper-doll # tier). Listing models only used to hide every /api/hangable output — the whole 2D route # dead-ended in a directory the UI refused to show. exts = MODEL_EXT + IMG_EXT if key == 'garments' else MODEL_EXT if key in ('bodies', 'out') else IMG_EXT rows = [] dirs = [d] + (EXTRA_BODIES if key == 'bodies' else []) for dd in dirs: if not os.path.isdir(dd): continue for f in sorted(os.listdir(dd)): if f.lower().endswith(exts): p = os.path.join(dd, f) row = {'name': f, 'path': p, 'kb': os.path.getsize(p) // 1024, 'home': os.path.basename(dd), 'nsfw': is_nsfw(f), 'swim': is_swim(f), 'rating': rating(f), 'kind': 'img' if f.lower().endswith(IMG_EXT) else 'model'} if key == 'garments': row.update(manifest_for(p, f)) elif key == 'bodies': rec = load_manifest()['items'].get(f, {}) row['body_type'] = rec.get('body_type') or DEFAULT_BODY_TYPE row['height'] = body_height(rec) row['tags'] = rec.get('tags', []) rows.append(row) out[key] = rows out['outfits'] = sorted(f[:-5] for f in os.listdir(OUTFITS) if f.endswith('.json')) return out # Canonical slots + draw order. The layer int drives BOTH the 2D doll stack and the 3D # normal-push, so one garment's ordering is the same in either product. SLOTS = ['head', 'torso', 'outer', 'legs', 'feet', 'hand', 'accessory', 'bag'] LAYER = {'skin': 0, 'legs': 20, 'feet': 25, 'torso': 30, 'outer': 40, 'head': 50, 'hand': 55, 'accessory': 60, 'bag': 70} # Guessing a slot from the filename is a convenience, never an assertion — every guess is # overridable and the manifest records which ones were inferred rather than set. SLOT_HINTS = [ ('head', r'hat|cap|beanie|helmet|hood|bucket|fedora|akubra|visor|glasses|sunnies'), ('feet', r'shoe|boot|sneaker|trainer|sandal|thong|volley|plimsoll|ugg|blundstone'), ('legs', r'pant|trouser|jean|short|skirt|trackie|jogger|legging|stubbies|cargo'), ('outer', r'jacket|coat|parka|windbreaker|hoodie|anorak|blazer|cardigan'), ('torso', r'shirt|tee|top|jumper|sweater|knit|singlet|polo|blouse|flanno|flannel'), ('bag', r'bag|backpack|rucksack|satchel|bumbag'), ('hand', r'glove|mitt|watch|bracelet'), ] def guess_slot(name): n = name.lower() for slot, pat in SLOT_HINTS: if re.search(pat, n): return slot return None # Body types carry their own target height. This is the whole reason unitfix refuses to # normalise everything to one number: a 'small' and an 'obese' character are not the same height, # and flattening them would erase the range the library exists to hold. Export applies the height # for the body's declared type; unitfix only ever repairs physically impossible scales. BODY_TYPES = { 'small': 1.58, 'medium': 1.72, 'large': 1.88, 'tall': 1.95, 'child': 1.30, 'obese': 1.74, 'stocky': 1.68, } DEFAULT_BODY_TYPE = 'medium' def body_height(rec): t = (rec or {}).get('body_type') or DEFAULT_BODY_TYPE return (rec or {}).get('height') or BODY_TYPES.get(t, BODY_TYPES[DEFAULT_BODY_TYPE]) def load_manifest(): if os.path.exists(MANIFEST): try: return json.load(open(MANIFEST)) except ValueError: pass return {'schemaVersion': 1, 'items': {}} def save_manifest(m): tmp = MANIFEST + '.tmp' with open(tmp, 'w') as f: json.dump(m, f, indent=1, sort_keys=True) os.replace(tmp, MANIFEST) # atomic — a half-written index would blank the wardrobe def manifest_for(path, name): """The record for one asset: stored fields win, the rest are inferred and marked as such.""" m = load_manifest() rec = dict(m['items'].get(os.path.basename(path), {})) if 'slot' not in rec: g = guess_slot(name) if g: rec['slot'], rec['slot_inferred'] = g, True rec.setdefault('layer', LAYER.get(rec.get('slot'), 30)) rec.setdefault('tags', []) rec['nsfw'] = is_nsfw(name) rec['swim'] = is_swim(name) rec['rating'] = rating(name) return rec def thumb_img(src, tgt): """Thumbnail a 2D asset — cropped to its content first. Doll layers are staged on a 704x1408 canvas with the garment occupying a slot-sized patch of it, so a plain downscale renders a mostly-empty tile with a tiny garment in it. Cropping to the alpha bbox makes a jumper look like a jumper at 256px. """ from PIL import Image im = Image.open(src).convert('RGBA') bb = im.getbbox() # None on a fully transparent image if bb: im = im.crop(bb) im.thumbnail((256, 256), Image.LANCZOS) im.save(tgt) def doll_catalogue(): """The 2D layers, enriched with the same manifest records the 3D garments use. The two catalogues were separate, so tag search only covered 3D and the 392 doll layers were invisible to it. They share one manifest now; the doll's slot comes from its filename prefix (which is authoritative — djsim resolves art by it), so it is never a guess here. """ cat = DOLL.catalogue() if DOLL else {} m = load_manifest() for slot, items in cat.items(): for it in items: rec = dict(m['items'].get(os.path.basename(it['path']), {})) it['slot'] = DOLL_SLOT_MAP.get(slot, slot) # doll slot names -> canonical slots it['layer'] = LAYER.get(it['slot'], 30) it['tags'] = rec.get('tags', []) it['title'] = rec.get('title') or it['key'].replace('_', ' ') it['nsfw'] = is_nsfw(it['path']) it['swim'] = is_swim(it['path']) it['rating'] = rating(it['path']) it['tier'] = '2d' return cat # djsim's doll uses top/bottom/shoes/hat/bag; the library's canonical slots are broader. DOLL_SLOT_MAP = {'top': 'torso', 'bottom': 'legs', 'shoes': 'feet', 'hat': 'head', 'bag': 'bag'} CANON_TO_DOLL = {v: k for k, v in DOLL_SLOT_MAP.items()} # --------------------------------------------------------------------------------------- # Prompt templates. `doll2d` is the one that actually shipped the 441 sprites; the green # variant is 90sDJsim's later gen_wardrobe fallback. Rules baked in, do not "improve": # · no text/words/logos — flux can't spell, so band tees are ART, never readable words. # · green is for PLAID and busy patterns. A bench run measured klein-4b on pure white # keying cleanly (7/7 flood seeds) on both a dark solid and a busy multicolour knit, so # white is the default and green is the escape hatch, not the other way round. TEMPLATES = { # 'no hanger, no coat hanger' is not padding: without it a wire hanger showed up in most # SD product shots (and some flux ones), and a hanger keys into the cutout as garment. 'flat3d': ('product photo of a single {phrase} laid flat on a plain white background, front view, ' 'no mannequin, no person, no hanger, no coat hanger, no text, soft even lighting, ' '1990s Australian op-shop garment, slightly worn'), 'doll2d': ('paper doll clothing item for a dress-up game, {phrase}, {year} australia, {view}, ' 'floating centered on a plain pure white background, painterly stylized 3d game ' 'illustration, full item visible, no body, no person, no mannequin, no hanger, ' 'no coat hanger, no text, no logos'), 'doll2d_green': ('1990s Australian {phrase}, product photo, flat lay, centered, ' 'solid bright green background, no text, no words, no logos, no person, no hanger'), # rigid props/hats/bags headed for TRELLIS: three-quarter view reconstructs far better than # a flat lay, and 'op-shop garment slightly worn' is wrong for a hard hat or a vinyl record. 'object3d': ('product photo of a single {phrase}, three-quarter view, centered, floating on a ' 'plain white background, studio lighting, sharp focus, entire object visible, ' 'no person, no hands, no mannequin, no text, no logos'), } VIEWS = {'hat': 'front view', 'top': 'front view laid flat with the sleeves down', 'bottom': 'front view laid perfectly flat on the ground', 'shoes': 'the pair side by side, front view', 'bag': 'front view'} DOLL_PROMPT = TEMPLATES['doll2d_green'] # back-compat alias FLAT_PROMPT = TEMPLATES['flat3d'] # The SD lane's default. Hyper_Realism was the only checkpoint the farm could reach and it is a # person-photoreal merge — it framed every garment as worn-and-cropped no matter how hard you # negative-prompted. juggernautXL is a product/material specialist; measured on the same # prompt+seed it returns a clean isolated flat-lay. SD_CKPT = os.environ.get('WG_SD_CKPT', 'juggernautXL_ragnarok.safetensors') SD_NEG = ('person, human, body, model, mannequin, face, hands, legs, worn by someone, ' 'hanger, coat hanger, (worst quality, low quality:1.4), blurry, watermark, text') def gen_seed(phrase, variant=0): """Reproducible seed from the phrase — reroll by bumping `variant`. Replaces two broken schemes: /api/gen used `time.time() % 99991` (nothing was ever reproducible) and /api/doll/gen inherited the operator default seed=42 (the same phrase produced a byte-identical image forever, so there was no reroll at all). """ return int(hashlib.sha1(f'{phrase}_{variant}'.encode()).hexdigest()[:7], 16) def run_blender(args, jid=None, script=None): """One headless Blender op; stdout tail lands in the job log (when there is a job). script= runs a different bpy file (the reskin trio lifted from NPCFACTORY) instead of blender_ops.py. """ cmd = [BLENDER, '-b', '--python', script or OPS, '--'] + [str(a) for a in args] r = subprocess.run(cmd, capture_output=True, text=True, timeout=1800) tail = '\n'.join((r.stdout + r.stderr).strip().splitlines()[-12:]) if jid and jid in JOBS: JOBS[jid]['log'] = tail if r.returncode != 0 or 'Error' in r.stderr: raise RuntimeError(tail[-400:]) def glb_of(path, jid=None): """FBX/OBJ → cached GLB (stage + ops always speak GLB); GLB passes through. The cache key carries a hash of the full source path: keying on the basename alone let two different files that slug the same (anatomy/jacket_v2.fbx vs garments/jacket-v2.fbx) collide onto one cache entry and serve each other's geometry. """ if path.lower().endswith(('.glb', '.gltf')): return path src = os.path.realpath(path) key = slug(os.path.basename(path)) + '-' + hashlib.sha1(src.encode()).hexdigest()[:8] tgt = os.path.join(CACHE, key + '.glb') if not os.path.exists(tgt) or os.path.getmtime(tgt) < os.path.getmtime(path): run_blender(['convert', path, tgt], jid) return tgt def new_job(kind, note): reap() jid = uuid.uuid4().hex[:10] JOBS[jid] = {'status': 'running', 'kind': kind, 'note': note, 'out': None, 'log': '', 't': time.time()} return jid def reap(max_age=3600, keep=200): """Drop finished jobs the page has long stopped polling — JOBS never pruned itself.""" now = time.time() dead = [k for k, v in list(JOBS.items()) if v.get('status') in ('done', 'error') and now - v.get('t', now) > max_age] for k in dead: JOBS.pop(k, None) if len(JOBS) > keep: for k, _ in sorted(JOBS.items(), key=lambda kv: kv[1].get('t', 0))[:len(JOBS) - keep]: JOBS.pop(k, None) def job_thread(jid, fn): def go(): try: JOBS[jid]['out'] = fn(jid) JOBS[jid]['status'] = 'done' except Exception as e: JOBS[jid]['status'] = 'error' JOBS[jid]['log'] = (JOBS[jid]['log'] + '\n' + str(e)).strip()[-1500:] threading.Thread(target=go, daemon=True).start() # ---------- MODELBEAST client (token stays server-side, never in the page) ---------- def mb_req(path, data=None, raw=None, ctype='application/json', tries=4): # the farm intermittently 404s/5xxs valid requests under load (seen three # times 2026-07-29: a job poll, a gen, a reskin — each one blip killed a # multi-minute pipeline). Retry briefly; job-create 404s created nothing, # so the retry cannot double-submit. last = None for attempt in range(tries): req = urllib.request.Request(MB + path, data=raw if raw is not None else (json.dumps(data).encode() if data else None)) req.add_header('Authorization', 'Bearer ' + MB_TOKEN) if data is not None or raw is not None: req.add_header('Content-Type', ctype) try: with urllib.request.urlopen(req, timeout=120) as r: body = r.read() break except urllib.error.HTTPError as e: if e.code in (404, 500, 502, 503) and attempt < tries - 1: last = e time.sleep(5 * (attempt + 1)) continue raise else: raise last try: return json.loads(body) except ValueError: return body def mb_wait(job_id, jid, timeout=3600): t0 = time.time() while time.time() - t0 < timeout: j = mb_req(f'/api/jobs/{job_id}') st = j.get('status') JOBS[jid]['note'] = f"farm: {st}" if st in ('done', 'completed', 'succeeded'): return j if st in ('error', 'failed', 'cancelled'): raise RuntimeError('farm job failed: ' + str(j.get('error') or st)) time.sleep(3) raise RuntimeError('farm job timed out') def mb_outputs(job_id): """Output assets for one farm job. The farm IGNORES ?parent_job= — a bogus id still returns the whole asset table (measured: 2,578 rows, 34 distinct parents). Taking rows[0] only ever worked because the table comes back newest-first, so two jobs in flight would happily download each other's image. Always filter client-side. job.asset_ids is [] even on success, so it's no help either. """ assets = mb_req(f'/api/assets?parent_job={job_id}') rows = assets.get('assets', assets) if isinstance(assets, dict) else assets if isinstance(rows, dict): rows = rows.get('items', []) return [r for r in (rows or []) if isinstance(r, dict) and str(r.get('parent_job')) == str(job_id)] def mb_download(asset, tgt): aid = asset.get('id') or asset.get('asset_id') data = mb_req(f'/api/assets/{aid}/file') if not isinstance(data, bytes): # mb_req parses any JSON-looking 200 into a dict. Writing that dict into a .png/.glb # used to succeed silently and mark the job done — you got a "mesh" full of error JSON. raise RuntimeError(f'farm returned no file for asset {aid}: {str(data)[:200]}') with open(tgt, 'wb') as f: f.write(data) return tgt def mb_upload(path): import mimetypes as mt boundary = uuid.uuid4().hex name = os.path.basename(path) ctype = mt.guess_type(path)[0] or 'application/octet-stream' with open(path, 'rb') as f: payload = f.read() body = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\n' f'Content-Type: {ctype}\r\n\r\n').encode() + payload + f'\r\n--{boundary}--\r\n'.encode() out = mb_req('/api/assets', raw=body, ctype=f'multipart/form-data; boundary={boundary}') return out.get('id') or out.get('asset_id') or (out.get('asset') or {}).get('id') # ---------- HTTP ---------- class H(BaseHTTPRequestHandler): def log_message(self, *a): pass def send(self, code, body, ctype='application/json'): self.send_response(code) self.send_header('Content-Type', ctype) self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def j(self, obj, code=200): self.send(code, json.dumps(obj).encode()) def do_GET(self): u = urllib.parse.urlparse(self.path) q = dict(urllib.parse.parse_qsl(u.query)) if u.path == '/': return self.send(200, open(os.path.join(ROOT, 'web', 'index.html'), 'rb').read(), 'text/html') if u.path.startswith('/vendor/'): # vendored three.js — no CDN, works offline f = os.path.realpath(os.path.join(ROOT, 'web', u.path.lstrip('/'))) web = os.path.realpath(os.path.join(ROOT, 'web')) if not f.startswith(web + os.sep) or not os.path.isfile(f): return self.j({'error': 'not found'}, 404) ctype = 'text/javascript' if f.endswith('.js') else (mimetypes.guess_type(f)[0] or 'application/octet-stream') return self.send(200, open(f, 'rb').read(), ctype) if u.path == '/api/kit': # socket-kit parts library (character_kit canonical) try: lib = json.load(open(KIT_LIB)) except Exception as e: return self.j({'error': 'kit library unreadable: ' + str(e)[:80]}, 500) parts = [{k: p.get(k) for k in ('id', 'slot', 'source_rig', 'style', 'tags')} for p in lib.get('parts', [])] return self.j({'slots': lib.get('slots', []), 'parts': parts}) if u.path == '/api/lib': return self.j({'lib': scan(), 'mb': bool(MB_TOKEN), 'cf': bool(CF_ACCT and CF_TOKEN), 'blender': os.path.exists(BLENDER), 'doll': doll_catalogue()}) if u.path.startswith('/api/job/'): jid = u.path.rsplit('/', 1)[1] return self.j(JOBS.get(jid) or {'status': 'unknown'}) if u.path.startswith('/api/outfit/'): f = os.path.join(OUTFITS, slug(u.path.rsplit('/', 1)[1]) + '.json') if not os.path.isfile(f): return self.j({'error': 'no such outfit'}, 404) return self.j(json.load(open(f))) if u.path == '/file': p = q.get('p', '') if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'nope'}, 403) return self.send(200, open(p, 'rb').read(), mimetypes.guess_type(p)[0] or 'application/octet-stream') if u.path == '/thumb': # lazy 256px thumb, cached; the grid needs these p = q.get('p', '') if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'nope'}, 403) key = hashlib.sha1(os.path.realpath(p).encode()).hexdigest()[:12] tgt = os.path.join(THUMBS, key + '.png') if not os.path.exists(tgt) or os.path.getmtime(tgt) < os.path.getmtime(p): try: if p.lower().endswith(IMG_EXT): thumb_img(p, tgt) else: run_blender(['thumb', glb_of(p), tgt, 256]) except Exception as e: return self.j({'error': str(e)[-200:]}, 500) return self.send(200, open(tgt, 'rb').read(), 'image/png') if u.path == '/glb': # stage loader: any model file → GLB (convert+cache) p = q.get('p', '') if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'nope'}, 403) try: g = glb_of(p) except Exception as e: return self.j({'error': str(e)[-300:]}, 500) return self.send(200, open(g, 'rb').read(), 'model/gltf-binary') return self.j({'error': 'not found'}, 404) def do_POST(self): u = urllib.parse.urlparse(self.path) q = dict(urllib.parse.parse_qsl(u.query)) n = int(self.headers.get('Content-Length') or 0) if n > MAX_BODY: return self.j({'error': 'body too large'}, 413) raw = self.rfile.read(n) if n else b'' if u.path == '/api/upload': # file input → library dir to, name = q.get('to', 'bodies'), os.path.basename(q.get('name', 'upload.glb')) if to not in DIRS or not name.lower().endswith(MODEL_EXT + IMG_EXT): return self.j({'error': 'bad target'}, 400) p = os.path.join(DIRS[to], name) with open(p, 'wb') as f: f.write(raw) return self.j({'ok': True, 'path': p}) body = json.loads(raw or b'{}') if u.path == '/api/thumbs/warm': # pre-render missing thumbs so the grid isn't cold jid = new_job('thumbs', 'warming thumbnails') def fx(jid): lib = scan() todo = [] for key in ('garments', 'bodies', 'out'): for it in lib.get(key, []): p = it['path'] k = hashlib.sha1(os.path.realpath(p).encode()).hexdigest()[:12] t = os.path.join(THUMBS, k + '.png') if not os.path.exists(t) or os.path.getmtime(t) < os.path.getmtime(p): todo.append((p, t)) for d in (doll_catalogue() or {}).values(): for it in d: p = it['path'] k = hashlib.sha1(os.path.realpath(p).encode()).hexdigest()[:12] t = os.path.join(THUMBS, k + '.png') if not os.path.exists(t): todo.append((p, t)) done = fail = 0 # Deliberately SERIAL: each 3D thumb launches its own Blender, and firing a # grid's worth at once buries the machine (and every request blocks a server # thread). Slow and quiet beats fast and unusable. for i, (p, t) in enumerate(todo): JOBS[jid]['note'] = f'{i + 1}/{len(todo)} {os.path.basename(p)[:28]}' try: if p.lower().endswith(IMG_EXT): thumb_img(p, t) else: run_blender(['thumb', glb_of(p), t, 256]) done += 1 except Exception: fail += 1 # a single unreadable asset must not abort the sweep return f'{done} rendered, {fail} failed, {len(todo)} queued' job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/rig': # bench step 1: raw mesh → rigged mixamorig # MIRPAMO is the local offline Mixamo (auto-rig + retarget). Its skeleton is 22 bones # with NO fingers, which is fine for locomotion but leaves cuff/forearm verts # under-driven if you later hang a 65-bone donor garment on it — say so rather than # let it surprise you downstream. p = body.get('path', '') if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'path outside library'}, 403) if not os.path.exists(MIRPAMO): return self.j({'error': 'MIRPAMO not found at ' + MIRPAMO}, 400) jid = new_job('rig', 'rig ' + os.path.basename(p)) def fx(jid): out = os.path.join(DIRS['bodies'], tagged_name(slug(os.path.basename(p).rsplit('.', 1)[0]) + '-rigged.glb', is_nsfw(p), is_swim(p))) cmd = [MIRPAMO, 'rig', p, '-o', out] if body.get('tris'): cmd += ['--tris', str(int(body['tris']))] r = subprocess.run(cmd, capture_output=True, text=True, timeout=3600) JOBS[jid]['log'] = '\n'.join((r.stdout + r.stderr).strip().splitlines()[-12:]) if r.returncode != 0 or not os.path.exists(out): raise RuntimeError(JOBS[jid]['log'][-400:] or 'rig failed') return out job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/meta': # set slot / layer / tags on a garment p = body.get('path', '') if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'path outside library'}, 403) m = load_manifest() rec = m['items'].setdefault(os.path.basename(p), {}) if body.get('slot') in SLOTS: rec['slot'] = body['slot'] rec['layer'] = LAYER.get(body['slot'], 30) rec.pop('slot_inferred', None) # an explicit choice is no longer a guess if isinstance(body.get('tags'), list): rec['tags'] = [str(t).strip().lower() for t in body['tags'] if str(t).strip()] if body.get('title'): rec['title'] = str(body['title'])[:80] if body.get('body_type') in BODY_TYPES: rec['body_type'] = body['body_type'] rec.pop('height', None) # type drives height unless overridden below if body.get('height'): try: rec['height'] = max(0.3, min(3.0, float(body['height']))) except (TypeError, ValueError): pass save_manifest(m) return self.j({'ok': True, 'rec': rec}) if u.path == '/api/tag': # set the content tier on an asset p, nsfw = body.get('path', ''), bool(body.get('nsfw')) swim = bool(body.get('swim')) if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'path outside library'}, 403) d, name = os.path.split(p) new = tagged_name(name, nsfw, swim) if new == name: return self.j({'ok': True, 'path': p, 'note': 'already tagged that way'}) tgt = os.path.join(d, new) if os.path.exists(tgt): return self.j({'error': 'a file with that name already exists'}, 409) os.rename(p, tgt) # the convert cache is keyed on the source path, so a rename orphans its entry return self.j({'ok': True, 'path': tgt, 'nsfw': nsfw, 'swim': swim and not nsfw, 'rating': rating(tgt)}) if u.path == '/api/outfit': # save a dress-up: body + attachments + clip + tints # The rigid-attach sliders used to live only in browser memory ("preview only") — every # hat placement died on reload. This is the persistence half of that tier. name = slug(body.get('name') or 'outfit') b = body.get('body') or '' if not allowed(b) or not os.path.isfile(b): return self.j({'error': 'body outside library'}, 403) rec = {'name': name, 'body': b, 'anim': body.get('anim'), 'attach': body.get('attach') or {}, 'tint': body.get('tint') or {}, 'saved': time.strftime('%Y-%m-%dT%H:%M:%S')} for att in rec['attach'].values(): if not allowed(att.get('path', '')): return self.j({'error': 'attachment outside library'}, 403) with open(os.path.join(OUTFITS, name + '.json'), 'w') as f: json.dump(rec, f, indent=2) return self.j({'ok': True, 'name': name}) if u.path == '/api/reskin': # 3D tier: paint a garment onto the body's own UVs # Sidesteps the fact that every TRELLIS body is unstructured with no shared topology — # instead of fitting a cloth mesh, we repaint the atlas the body already has. Needs a # single-mesh / single-UV / single-material body (multi-material rigs shard the bake). if not MB_TOKEN: return self.j({'error': 'reskin needs the farm image-edit operator — MB_TOKEN missing'}, 400) b, g = body.get('body', ''), body.get('garment', '') for p in (b, g): if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'path outside library'}, 403) desc = body.get('desc') or 'the garment in image 2' jid = new_job('reskin', 'reskin ' + os.path.basename(b)) def fx(jid): src = glb_of(b, jid) name = slug(os.path.basename(b).rsplit('.', 1)[0]) + '--' + slug(os.path.basename(g).rsplit('.', 1)[0]) wd = os.path.join(DIRS['gen'], '_reskin', name) os.makedirs(wd, exist_ok=True) JOBS[jid]['note'] = 'rendering bind-pose plates' run_blender([src, wd], jid, script=os.path.join(TOOLS, 'render_plates.py')) gid = mb_upload(g) outs = {} for side in ('front', 'back'): plate = os.path.join(wd, f'base_{side}.png') if not os.path.exists(plate): raise RuntimeError(f'no {side} plate rendered') JOBS[jid]['note'] = f'image-edit: dressing the {side} plate' j = mb_req('/api/jobs', { 'operator': 'mflux_image_edit', 'asset_ids': [mb_upload(plate), gid], 'params': {'prompt': f'put {desc} from image 2 onto the person in image 1. ' 'Keep the exact same pose, body proportions, framing and ' 'plain background. Full body visible, head to toe.'}}) ji = j.get('id') or j.get('job_id') mb_wait(ji, jid, timeout=900) o = mb_outputs(ji) if not o: raise RuntimeError(f'image-edit returned nothing for the {side} plate') edited = os.path.join(wd, f'edit_{side}.png') mb_download(o[0], edited) # the edit model redraws the figure at its own scale/offset; the projection bake # assumes the original framing, so unaligned plates make the body sample background aligned = os.path.join(wd, f'aligned_{side}.png') subprocess.run(['python3', os.path.join(TOOLS, 'align_plate.py'), plate, edited, aligned], capture_output=True, text=True, timeout=300) outs[side] = aligned if os.path.exists(aligned) else edited JOBS[jid]['note'] = 'baking onto the body atlas (Cycles)' tgt = os.path.join(DIRS['garments'], name + '.reskin.png') run_blender([src, outs['front'], outs['back'], tgt], jid, script=os.path.join(TOOLS, 'bake_skin.py')) if not os.path.exists(tgt): raise RuntimeError('bake produced no texture') return tgt job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/doll/gen': # 2D tier: prompt → staged paper-doll layer if not DOLL: return self.j({'error': '2D tier unavailable — needs pillow + numpy'}, 400) if not MB_TOKEN: return self.j({'error': 'the doll pipeline runs on the farm — MB_TOKEN missing'}, 400) slot, keyname = body.get('slot', 'top'), body.get('key') or body.get('phrase', '') phrase = body.get('phrase', '') if slot not in ('hat', 'top', 'bottom', 'shoes', 'bag'): return self.j({'error': 'slot must be hat|top|bottom|shoes|bag'}, 400) if not phrase: return self.j({'error': 'need a garment phrase'}, 400) # doll2d (painterly, pure white) is the template the 441 shipped sprites were made # with; doll2d_green is the escape hatch for plaid and busy patterns where a white # key gets confused. Default to what shipped. tpl = body.get('template', 'doll2d') if tpl not in TEMPLATES: return self.j({'error': 'template must be one of ' + '|'.join(TEMPLATES)}, 400) variant = int(body.get('variant', 0) or 0) style = TEMPLATES[tpl].format(phrase=phrase, year=body.get('year', 1990), view=VIEWS.get(slot, 'front view')) jid = new_job('doll', f'{slot}: {phrase[:40]}') JOBS[jid]['prompt'] = style def fx(jid): stage_dir = os.path.join(DIRS['gen'], '_doll') os.makedirs(stage_dir, exist_ok=True) base = DOLL.slugify(keyname) raw = os.path.join(stage_dir, f'{slot}_{base}_flux.png') cut = os.path.join(stage_dir, f'{slot}_{base}_keyed.png') JOBS[jid]['note'] = f'flux ({tpl})' # 704, not 1024: the doll canvas is 704 wide, so generating bigger only means # downscaling before the key and softening the edges we depend on. j = mb_req('/api/jobs', {'operator': 'flux_local', 'params': {'prompt': style, 'model': 'flux2-klein-4b', 'steps': 4, 'width': 704, 'height': 704, 'seed': gen_seed(phrase, variant)}}) gid = j.get('id') or j.get('job_id') mb_wait(gid, jid) outs = mb_outputs(gid) if not outs: raise RuntimeError('flux returned no image') mb_download(outs[0], raw) JOBS[jid]['note'] = 'keying (farm RMBG-2.0)' k = mb_req('/api/jobs', {'operator': 'bg_remove_local', 'asset_id': mb_upload(raw), 'params': {}}) kid = k.get('id') or k.get('job_id') mb_wait(kid, jid) kouts = mb_outputs(kid) if not kouts: raise RuntimeError('bg_remove returned nothing') mb_download(kouts[0], cut) JOBS[jid]['note'] = 'staging to 704x1408' return DOLL.stage(cut, slot, base) job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/doll/compose': # stack staged layers → one dressed doll preview if not DOLL: return self.j({'error': '2D tier unavailable'}, 400) stems = [s for s in (body.get('stems') or []) if isinstance(s, str) and '/' not in s] out = os.path.join(DIRS['out'], slug(body.get('name') or 'doll') + '.png') base = os.path.join(DOLL.DOLL_DIR, 'base.png') DOLL.compose(out, stems, base=base if os.path.exists(base) else None) return self.j({'ok': True, 'path': out}) if u.path == '/api/export': # the hub: one door, honest about each target tgt, p = body.get('target', ''), body.get('path', '') if tgt not in EXPORT_TARGETS: return self.j({'error': 'target must be one of ' + '|'.join(EXPORT_TARGETS)}, 400) t = EXPORT_TARGETS[tgt] if not t['ready']: # Verified against each consumer's real loader: these ingest by a HARDCODED array # or a manifest, so dropping a file does nothing until its NAME is in that list. # We can make that edit — but it is source surgery on a live game, so it dry-runs # unless explicitly told to write. if not REG: return self.j({'ok': False, 'ready': False, 'target': tgt, 'why': t['why'], 'registry': t.get('registry')}, 409) nm = body.get('name') or os.path.basename(p).rsplit('.', 1)[0] try: res = REG.add(tgt, nm, do_write=bool(body.get('write'))) except Exception as e: return self.j({'error': str(e)[:300]}, 500) return self.j({'ok': True, 'ready': False, 'registry_edit': res, 'why': t['why'], 'note': ('written — back up files are alongside each source' if body.get('write') else 'DRY RUN — re-send with write:true to actually edit the source')}) if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'path outside library'}, 403) if is_nsfw(p): return self.j({'error': 'refusing to publish an nsfw-tagged asset'}, 403) if tgt == '3god': name = god3_name(body.get('name') or os.path.basename(p)) if not name.lower().endswith('.glb'): return self.j({'error': '3GOD wants a .glb'}, 400) jid = new_job('export', f'3god ← {name}') def fx(jid): # The depot DELETES illegal characters rather than substituting, so two # different sources can clean to one name and the second silently serves the # first's mesh. Check before writing. try: listing = json.load(urllib.request.urlopen(GOD3 + '/api/list', timeout=30)) have = {a.get('file') or a.get('name') for a in (listing.get('assets') or listing.get('items') or [])} except Exception: have = set() if name in have and not body.get('overwrite'): raise RuntimeError(f'{name} already exists in the depot — pass overwrite:true ' f'if you really mean to replace it') data = open(p, 'rb').read() req = urllib.request.Request( GOD3 + '/api/upload?name=' + urllib.parse.quote(name), data=data, headers={'Content-Type': 'model/gltf-binary'}, method='POST') with urllib.request.urlopen(req, timeout=600) as r: out = r.read()[:200] JOBS[jid]['log'] = out.decode('utf-8', 'replace') return GOD3 + '/a/' + name job_thread(jid, fx) return self.j({'job': jid, 'name': name, 'depot': GOD3}) return self.j({'error': 'unhandled target'}, 500) if u.path == '/api/doll/export': # staged layers → 90sDJsim's live media/doll if not DOLL: return self.j({'error': '2D tier unavailable'}, 400) stems = [s for s in (body.get('stems') or []) if isinstance(s, str) and '/' not in s] if not stems: return self.j({'error': 'nothing to export'}, 400) # djsim's loader resolves _i_.png and self-heals on new files, so a plain # copy wires the garment with no code change. NEVER touch the generic slot fallbacks — # overwriting those would restyle every unarted item in a shipping game. bad = [s for s in stems if '_i_' not in s] if bad: return self.j({'error': 'refusing to overwrite generic slot art: ' + bad[0]}, 400) # The house rule "bases stay in the library, only dressed outfits ship into games" # has lived as one unenforced line in the README. This is the enforcement. nsfw = [s for s in stems if is_nsfw(s)] if nsfw: return self.j({'error': f'refusing to export nsfw-tagged asset into a game ' f'directory: {nsfw[0]}'}, 403) files = [(s, os.path.join(DOLL.DOLL_DIR, s + '.png')) for s in stems] missing = [s for s, p in files if not os.path.isfile(p)] if missing: return self.j({'error': 'not staged: ' + missing[0]}, 404) dest = f'{DJSIM_HOST}:{DJSIM_DOLL}' if body.get('dry', True): return self.j({'ok': True, 'dry': True, 'files': [s for s, _ in files], 'dest': dest, 'note': f'DRY RUN — would copy {len(files)} layer/s to {dest}. ' f'Re-send with dry:false to actually write.'}) jid = new_job('export', f'{len(files)} layer/s → djsim') def fx(jid): cmd = ['rsync', '-a'] + [p for _, p in files] + [dest + '/'] r = subprocess.run(cmd, capture_output=True, text=True, timeout=600) if r.returncode != 0: raise RuntimeError((r.stderr or r.stdout)[-400:]) return dest job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/blender': # {op, args:{...}} → background Blender job op = body.get('op') a = body.get('args', {}) # Every other endpoint gates its path args through allowed(); this one never did, so # any tailnet peer could hand WG_HOST=0.0.0.0 an arbitrary on-disk file to import. ins = [a[k] for k in ('path', 'body', 'garment') if a.get(k)] + list(a.get('garments') or []) if not ins: return self.j({'error': 'no input path'}, 400) for p in ins: if not allowed(p) or not os.path.isfile(p): return self.j({'error': 'path outside library: ' + os.path.basename(str(p))}, 403) jid = new_job(op, body.get('note', op)) def fx(jid): if op == 'scale': src = glb_of(a['path'], jid) out = os.path.join(DIRS['bodies'], slug(a.get('name') or os.path.basename(a['path']).rsplit('.', 1)[0]) + '.glb') run_blender(['scale', src, out, a['height']], jid) elif op == 'decimate': src = glb_of(a['path'], jid) out = os.path.join(DIRS['bodies'], slug(os.path.basename(a['path']).rsplit('.', 1)[0]) + f"-{int(a['tris'])//1000}k.glb") run_blender(['decimate', src, out, int(a['tris'])], jid) elif op == 'fit': b, g = glb_of(a['body'], jid), glb_of(a['garment'], jid) mode = a.get('mode', 'garment') base = slug(os.path.basename(a['garment']).rsplit('.', 1)[0]) out = os.path.join(DIRS['garments'] if mode == 'garment' else DIRS['out'], base + ('-fitted.glb' if mode == 'garment' else '-dressed.glb')) run_blender(['fit', b, g, out, mode, a.get('inflate', 3)], jid) elif op == 'assemble': b = glb_of(a['body'], jid) gs = [glb_of(g, jid) for g in a['garments']] # An assembled result INHERITS the body's nsfw tag. Clearing it automatically # because "a garment was added" would be a silent wrong call — a hat does not # clothe a nude base — and the failure mode is shipping a nude model into a # game. Untagging stays a deliberate act once you've looked at the result. out = os.path.join(DIRS['out'], tagged_name(slug(a.get('name') or 'outfit') + '.glb', is_nsfw(a['body']))) run_blender(['assemble', b, out] + gs, jid) elif op == 'unitfix': src = glb_of(a['path'], jid) base = slug(os.path.basename(a['path']).rsplit('.', 1)[0]) out = os.path.join(DIRS['bodies'], tagged_name(base + '-m.glb', is_nsfw(a['path']))) run_blender(['unitfix', src, out, a.get('height', 1.72)], jid) elif op == 'convert': src = a['path'] out = os.path.join(DIRS['bodies'], slug(os.path.basename(src).rsplit('.', 1)[0]) + '.glb') run_blender(['convert', src, out], jid) else: raise RuntimeError('unknown op') return out job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/gen': # the garment pipeline, step 1: flux image if not (CF_ACCT and CF_TOKEN) and not MB_TOKEN: return self.j({'error': 'no image backend — need .env CF creds or MB_TOKEN'}, 400) phrase = body.get('phrase') or body.get('prompt', '') if not phrase: return self.j({'error': 'need a garment phrase'}, 400) tpl = body.get('template', 'flat3d') if tpl not in TEMPLATES: return self.j({'error': 'template must be one of ' + '|'.join(TEMPLATES)}, 400) # klein-4b is the DEFAULT because it measurably renders fabric better: at matched # prompt/seed it holds corduroy wale across the garment where CF flux-1-schnell # flattens it to smooth canvas. CF stays available — it's ~2s and never queues # behind the farm's serial GPU lane, which is worth real money on a one-at-a-time UI. backend = body.get('backend', 'klein' if MB_TOKEN else 'cf') variant = int(body.get('variant', 0) or 0) seed = body.get('seed') seed = gen_seed(phrase, variant) if seed in (None, '') else int(seed) wide = int(body.get('width') or (704 if tpl.startswith('doll2d') else 1024)) steps = int(body.get('steps') or 4) style = TEMPLATES[tpl].format(phrase=phrase, year=body.get('year', 1990), view=VIEWS.get(body.get('view', 'top'), 'front view')) if body.get('lora') and backend != 'sd': pass # flux_local has no lora param at all — silently ignoring beats pretending jid = new_job('gen', f'{backend}: {phrase[:36]}') JOBS[jid]['prompt'] = style JOBS[jid]['seed'] = seed def fx(jid): tgt = os.path.join(DIRS['gen'], f'{slug(phrase)}-v{variant}.png') if backend == 'cf' and CF_ACCT and CF_TOKEN: import base64 JOBS[jid]['note'] = 'cloudflare flux-1-schnell (softer fabric)' req = urllib.request.Request( f'https://api.cloudflare.com/client/v4/accounts/{CF_ACCT}/ai/run/@cf/black-forest-labs/flux-1-schnell', data=json.dumps({'prompt': style, 'steps': max(steps, 8), 'width': wide, 'height': wide, 'seed': seed}).encode(), headers={'Authorization': 'Bearer ' + CF_TOKEN, 'Content-Type': 'application/json'}) with urllib.request.urlopen(req, timeout=120) as r: out = json.load(r) if not out.get('success'): raise RuntimeError(str(out.get('errors'))[:300]) with open(tgt, 'wb') as f: f.write(base64.b64decode(out['result']['image'])) return tgt if backend == 'sd': # SDXL via ComfyUI. juggernautXL is a product/material specialist and beats # the SD1.5 default outright: measured side by side on the same prompt+seed it # gives a full isolated flat-lay, where Hyper_Realism frames the garment as # worn and crops it at the edge. `lora` may be a list — the operator chains them. JOBS[jid]['note'] = 'comfyui_sd ' + body.get('checkpoint', SD_CKPT) params = {'prompt': style, 'negative': body.get('negative', SD_NEG), 'checkpoint': body.get('checkpoint', SD_CKPT), 'steps': int(body.get('steps') or 25), 'cfg': float(body.get('cfg', 7.0)), 'width': wide, 'height': wide, 'seed': seed} if body.get('lora'): params['lora'] = body['lora'] params['lora_weight'] = body.get('lora_weight', 0.8) j = mb_req('/api/jobs', {'operator': 'comfyui_sd', 'params': params}) gid = j.get('id') or j.get('job_id') mb_wait(gid, jid, timeout=900) outs = mb_outputs(gid) if not outs: raise RuntimeError('no image came back') return mb_download(outs[0], tgt) JOBS[jid]['note'] = 'flux_local klein-4b' j = mb_req('/api/jobs', {'operator': 'flux_local', 'params': {'prompt': style, 'model': 'flux2-klein-4b', 'steps': steps, 'width': wide, 'height': wide, 'seed': seed}}) gid = j.get('id') or j.get('job_id') mb_wait(gid, jid) outs = mb_outputs(gid) if not outs: raise RuntimeError('no image came back') return mb_download(outs[0], tgt) job_thread(jid, fx) return self.j({'job': jid, 'seed': seed, 'prompt': style}) if u.path == '/api/rmbg': # step 2: cut it out (farm RMBG, else local white-keyer) p = body.get('path', '') if not allowed(p): return self.j({'error': 'nope'}, 403) jid = new_job('rmbg', 'cutout ' + os.path.basename(p)) def fx(jid): tgt = p.rsplit('.', 1)[0] + '-cut.png' if MB_TOKEN: # RMBG-2.0 on the farm — best quality aid = mb_upload(p) j = mb_req('/api/jobs', {'operator': 'bg_remove_local', 'asset_id': aid, 'params': {}}) mb_wait(j.get('id') or j.get('job_id'), jid) outs = mb_outputs(j.get('id') or j.get('job_id')) if not outs: raise RuntimeError('no cutout came back') return mb_download(outs[0], tgt) # local fallback: border-flood white-keyer (never naive white→alpha — traps ledger). # flood from the edges through near-white pixels only, so white ON the garment survives. # Measured limit: flat3d product shots come back with a studio-grey backdrop and a soft # drop shadow (border 215-244, ZERO pixels ≥250), so this >232 gate only partially seeds # and leaves grey fringing. RMBG-2.0 is semantic and doesn't care — warn loudly here. JOBS[jid]['note'] = 'local white-keyer (no farm token — grey backdrops will fringe)' from PIL import Image im = Image.open(p).convert('RGBA') w, h = im.size px = im.load() near = lambda c: c[0] > 232 and c[1] > 232 and c[2] > 232 seen = bytearray(w * h) stack = [(x, y) for x in range(w) for y in (0, h - 1)] + \ [(x, y) for y in range(h) for x in (0, w - 1)] while stack: x, y = stack.pop() i = y * w + x if seen[i] or not near(px[x, y]): continue seen[i] = 1 px[x, y] = (0, 0, 0, 0) if x > 0: stack.append((x - 1, y)) if x < w - 1: stack.append((x + 1, y)) if y > 0: stack.append((x, y - 1)) if y < h - 1: stack.append((x, y + 1)) im.save(tgt) return tgt job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/to3d': # step 3a: rigid wearable via TRELLIS (hats/shoes/bags) if not MB_TOKEN: return self.j({'error': 'MB_TOKEN not set'}, 400) p = body.get('path', '') if not allowed(p): return self.j({'error': 'nope'}, 403) jid = new_job('to3d', '3D: ' + os.path.basename(p) + ' (~5 min on the farm)') def fx(jid): aid = mb_upload(p) j = mb_req('/api/jobs', {'operator': 'trellis2_mlx', 'asset_id': aid, 'params': {}}) mb_wait(j.get('id') or j.get('job_id'), jid, timeout=7200) outs = mb_outputs(j.get('id') or j.get('job_id')) glbs = [o for o in outs if str(o.get('name', o.get('filename', ''))).lower().endswith('.glb')] or outs if not glbs: raise RuntimeError('no mesh came back') tgt = os.path.join(DIRS['garments'], slug(os.path.basename(p).rsplit('.', 1)[0]) + '-rigid.glb') return mb_download(glbs[0], tgt) job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/kit/assemble': # build a character from socket parts choice = body.get('parts') or {} if 'torso' not in choice: return self.j({'error': 'torso is required — it sets the target rig'}, 400) try: lib = json.load(open(KIT_LIB)) except Exception as e: return self.j({'error': 'kit library unreadable'}, 500) by_id = {p['id']: p for p in lib.get('parts', [])} spec_parts = {} for slot, pid in choice.items(): p = by_id.get(pid) if not p or p.get('slot') != slot: return self.j({'error': f'unknown part {pid} for slot {slot}'}, 400) blend = os.path.normpath(os.path.join(os.path.dirname(KIT_LIB), p['file'])) spec_parts[slot] = {'blend': blend, 'object': p['object'], 'source_rig': p['source_rig']} target = spec_parts['torso']['source_rig'] g = spec_parts.get('genitals') if g and g['source_rig'] != target: return self.j({'error': 'genitals is an additive same-rig slot — pick the torso rig'}, 400) name = 'kit-' + '-'.join(slug(spec_parts[s2]['object']) for s2 in ('torso',) ) name += '-' + str(int(time.time()) % 100000) wd = os.path.join(DIRS['gen'], '_kit') os.makedirs(wd, exist_ok=True) spec_path = os.path.join(wd, name + '.json') json.dump({'target_tag': target, 'parts': spec_parts}, open(spec_path, 'w')) out_glb = os.path.join(DIRS['out'], name + '.glb') out_png = os.path.join(DIRS['gen'], name + '.png') jid = new_job('kit', 'assemble ' + name) def fx(jid): run_blender([spec_path, out_glb, out_png], jid, script=os.path.join(TOOLS, 'kit_assemble.py')) if not os.path.exists(out_glb): raise RuntimeError('assembly produced no GLB') JOBS[jid]['png'] = out_png return out_glb job_thread(jid, fx) return self.j({'job': jid}) if u.path == '/api/hangable': # step 3b: cutout → hanging-garment texture pack p = body.get('path', '') if not allowed(p): return self.j({'error': 'nope'}, 403) tgt = os.path.join(DIRS['garments'], os.path.basename(p)) shutil.copyfile(p, tgt) return self.j({'ok': True, 'path': tgt, 'note': 'texture saved to garments — use on rack planes (fittings.js garment({image}))'}) return self.j({'error': 'not found'}, 404) if __name__ == '__main__': print(f'WARDROBEGOD on http://{HOST}:{PORT} · blender={os.path.exists(BLENDER)} · farm token={"yes" if MB_TOKEN else "NO (generator off)"}') ThreadingHTTPServer((HOST, PORT), H).serve_forever()