monster/wardrobegod
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
294af7cf1d
wardrobegod/server.py
type-two 294af7cf1d Phase 3: the 2D paper-doll tier — 392 garments you can actually change 
The flat cut-outs were the cheapest content on the board and the only tier with a shipping
consumer, but wardrobegod couldn't show one: scan() filtered PNGs out of garments/ (fixed in
Phase 0) and there was no compositor, no slots, no preview. Now there is a working dress-up.

· library/doll/ seeded with 90sDJsim's 441 shipped sprites (165 top, 110 bottom, 60 shoes,
  57 hat, base body) — an instant starter wardrobe rather than a cold start.
· tools/doll.py wraps djsim's compositor. It deliberately does NOT import doll_composite:
  that module pulls numpy at import time for a keyer and an HSV recolour we never use (the
  farm's RMBG-2.0 does our keying), and numpy isn't installable here under PEP 668. Instead
  the measured anchors are parsed out of its source with ast, so there is still exactly one
  source of truth and they cannot drift; place() is a pure-PIL port.
· slugify() reproduces djsim's item_art() _art_slug character for character, verified against
  it. That equality IS the drop-in contract — a near-miss silently falls back to generic slot
  art instead of erroring, so it's the kind of bug you'd ship without noticing.
· /api/doll/gen runs the proven pipeline: flux_local on SOLID GREEN -> farm RMBG-2.0 ->
  staged to 704x1408. Green is not a preference: grey and checkerboard backgrounds eat
  garments in the key, and flux can't spell, so no text ever goes in the prompt.
· /api/doll/compose composes server-side with PIL, so the browser preview and the exported
  PNG are the same bytes rather than two renderers that drift.
· /api/doll/export writes into a LIVE game directory, so it defaults to a dry run and refuses
  any stem without _i_ — overwriting a generic slot fallback would restyle every unarted item
  in a shipping game.
· UI: a doll 2D tab with per-slot pickers, randomise, strip, and export.

Verified: composed a dressed doll from real sprites (correct base->bottom->shoes->top->hat
order), catalogue reports 392 layers, export dry-run and the generic-art guard both behave.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-24 16:07:13 +10:00

566 lines
28 KiB
Python
Raw Blame History

#!/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
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')
# 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'))
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')
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')
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'
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)
rows.append({'name': f, 'path': p, 'kb': os.path.getsize(p) // 1024,
'home': os.path.basename(dd),
'kind': 'img' if f.lower().endswith(IMG_EXT) else 'model'})
out[key] = rows
out['outfits'] = sorted(f[:-5] for f in os.listdir(OUTFITS) if f.endswith('.json'))
return out
def doll_catalogue():
return DOLL.catalogue() if DOLL else {}
# The prompt that produced the 441 shipped sprites, lifted verbatim from 90sDJsim's
# gen_wardrobe.py. Two hard-won rules are baked in and must not be "improved":
# · SOLID BRIGHT GREEN background — grey/checkerboard backgrounds eat garments in the key,
# and plaid survives green. This is why we don't generate on white for the doll tier.
# · no text/words/logos — flux can't spell, so band tees are ART, never readable words.
DOLL_PROMPT = ('1990s Australian {phrase}, product photo, flat lay, centered, '
'solid bright green background, no text, no words, no logos, no person, no hanger')
# The 3D tier wants a plain-white product shot instead — nothing gets keyed on green there.
FLAT_PROMPT = ('product photo of a single {phrase} laid flat on a plain white background, front view, '
'no mannequin, no person, no text, soft even lighting, 1990s Australian op-shop garment, '
'slightly worn')
def run_blender(args, jid=None):
"""One headless Blender op; stdout tail lands in the job log (when there is a job)."""
cmd = [BLENDER, '-b', '--python', 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'):
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)
with urllib.request.urlopen(req, timeout=120) as r:
body = r.read()
try:
return json.loads(body)
except ValueError:
return body
def mb_wait(job_id, jid, timeout=900):
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):
assets = mb_req(f'/api/assets?parent_job={job_id}')
rows = assets.get('assets', assets) if isinstance(assets, dict) else assets
return rows or []
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 == '/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 == '/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)
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/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/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)
jid = new_job('doll', f'{slot}: {phrase[:40]}')
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'] = 'flux on green'
j = mb_req('/api/jobs', {'operator': 'flux_local',
'params': {'prompt': DOLL_PROMPT.format(phrase=phrase),
'width': 1024, 'height': 1024}})
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/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 <slot>_i_<slug>.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)
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']]
out = os.path.join(DIRS['out'], slug(a.get('name') or 'outfit') + '.glb')
run_blender(['assemble', b, out] + gs, 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)
prompt = body.get('prompt', '')
style = ('product photo of a single {} laid flat on a plain white background, '
'front view, no mannequin, no person, no text, soft even lighting, '
'1990s Australian op-shop garment, slightly worn').format(prompt)
jid = new_job('gen', 'flux: ' + prompt[:40])
def fx(jid):
tgt = os.path.join(DIRS['gen'], slug(prompt) + '.png')
if CF_ACCT and CF_TOKEN: # Cloudflare Workers AI direct — the flux.mjs contract
import base64
JOBS[jid]['note'] = 'cloudflare flux-1-schnell'
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': 8, 'width': 768, 'height': 768}).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
j = mb_req('/api/jobs', {'operator': 'flux_local', 'asset_id': None,
'params': {'prompt': style, 'model': 'flux2-klein-4b',
'steps': 4, 'width': 768, 'height': 768,
'seed': int(time.time()) % 99991}})
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 image came back')
return mb_download(outs[0], tgt)
job_thread(jid, fx)
return self.j({'job': jid})
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.
JOBS[jid]['note'] = 'local white-keyer'
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': 'trellis_mac', 'asset_id': aid, 'params': {}})
mb_wait(j.get('id') or j.get('job_id'), jid, timeout=1200)
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/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()
Reference in New Issue View Git Blame Copy Permalink