#!/usr/bin/env python3 """Reproducible multi-engine image→3D benchmark via the public PicoBerry /v1 API. Same 3 reference images × 6 engines × 3 repeats, engine defaults, texture on. Records: status, wall time (5s resolution), vertices/faces/topology (files.stats), file sizes, list price (credits, from /v1/models?category=image-to-3d at run time), thumbnails. No subjective scoring. Modes python3 bench.py --summarize [results.json] FREE, local: re-aggregate a results file (the published object form with `runs`, or the raw list form) and print the per-engine table. Never calls the API. python3 bench.py PAID: runs the matrix (needs PICOBERRY_API_KEY). Reuses inputs/*.png|webp when present (published inputs work as-is), resumes pending/orphaned runs instead of re-submitting. """ import json, os, sys, time, threading, urllib.request, urllib.error, uuid, mimetypes from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone BASE = os.environ.get('PICOBERRY_API_BASE', 'https://api.picoberry.ai').rstrip('/') KEY = os.environ.get('PICOBERRY_API_KEY', '') OUT = os.path.dirname(os.path.abspath(__file__)) INPUTS = os.path.join(OUT, 'inputs'); RESULTS = os.path.join(OUT, 'results'); PENDING = os.path.join(OUT, 'pending.json') os.makedirs(INPUTS, exist_ok=True); os.makedirs(RESULTS, exist_ok=True) ENGINES = os.environ.get('BENCH_ENGINES', 'tripo,tripo-p2,tripo-v3.1,tripo-v3.1-ultra,meshy6,hunyuan-3.1').split(',') REPEATS = int(os.environ.get('BENCH_REPEATS', '3')) CONCURRENCY = int(os.environ.get('BENCH_CONCURRENCY', '6')) PROMPTS = { 'stylized-prop': 'A stylized wooden barrel with two dark iron bands, game prop, front three-quarter view, whole object centered and fully visible, plain light grey studio background, soft even lighting, no text, no watermark', 'hard-surface': 'A hard-surface sci-fi blaster pistol with crisp panel lines, matte dark metal with orange accents, side view, whole object centered and fully visible, plain light grey studio background, soft even lighting, no text, no watermark', 'character': 'A stylized armored knight character standing in a T-pose, full body, front view, whole body centered and fully visible, plain light grey studio background, soft even lighting, clean concept art, no text, no watermark', } now = lambda: datetime.now(timezone.utc).isoformat(timespec='seconds').replace('+00:00', 'Z') lock = threading.Lock() def req(method, path, json_body=None, form=None, timeout=120): url = BASE + path headers = {'Authorization': f'Bearer {KEY}'} data = None if json_body is not None: data = json.dumps(json_body).encode(); headers['Content-Type'] = 'application/json' elif form is not None: boundary = 'B' + uuid.uuid4().hex parts = [] for k, v in form.items(): if isinstance(v, tuple): fname, blob, ctype = v parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"; filename="{fname}"\r\nContent-Type: {ctype}\r\n\r\n'.encode() + blob + b'\r\n') else: parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()) data = b''.join(parts) + f'--{boundary}--\r\n'.encode() headers['Content-Type'] = f'multipart/form-data; boundary={boundary}' r = urllib.request.Request(url, data=data, method=method, headers=headers) try: with urllib.request.urlopen(r, timeout=timeout) as res: return res.status, json.loads(res.read().decode() or '{}') except urllib.error.HTTPError as e: body = e.read().decode(errors='replace') try: return e.code, json.loads(body) except Exception: return e.code, {'raw': body[:300]} def poll(asset_id, max_s=1800, every=5): t0 = time.time() while time.time() - t0 < max_s: st, d = req('GET', f'/v1/assets/{asset_id}', timeout=60) a = d.get('data') if st == 200 else None if a and a.get('taskStatus') in (2, 3): return a, now() time.sleep(every) return None, now() def head_size(url): try: r = urllib.request.Request(url, method='HEAD') with urllib.request.urlopen(r, timeout=60) as res: return int(res.headers.get('Content-Length') or 0) except Exception: return None def download(url, path): try: with urllib.request.urlopen(url, timeout=120) as res, open(path, 'wb') as f: f.write(res.read()) return True except Exception: return False def load_runs(path): """results.json in either shape: the raw list this script writes while running, or the published object ({'runs': [...], 'engines': {...}, ...}). Returns the list of run records.""" if not os.path.exists(path): return [] data = json.load(open(path)) return data['runs'] if isinstance(data, dict) else data def input_file(key): """Published inputs are .webp, fresh generations are .png — accept both, with the right MIME.""" for ext, mime in (('png', 'image/png'), ('webp', 'image/webp'), ('jpg', 'image/jpeg')): path = os.path.join(INPUTS, f'{key}.{ext}') if os.path.exists(path): return path, f'{key}.{ext}', mime return None, None, None def summarize(path): import statistics as st runs = load_runs(path) if not runs: print('no runs in', path); return engines = [] for r in runs: if r['engine'] not in engines: engines.append(r['engine']) print(f"{'engine':18s} {'ok':>5s} {'wall med':>9s} {'wall range':>12s} {'faces med':>11s} {'faces range':>22s} {'GLB MB':>7s}") for e in engines: rs = [r for r in runs if r['engine'] == e]; ok = [r for r in rs if r.get('status') == 'succeeded'] walls = [r['wallSeconds'] for r in ok]; faces = [r['stats']['faces'] for r in ok if r.get('stats') and 'faces' in r['stats']] mb = [r['modelBytes'] / 1e6 for r in ok if r.get('modelBytes')] print(f"{e:18s} {len(ok):>2d}/{len(rs):<2d} {st.median(walls) if walls else 0:>8.0f}s {min(walls) if walls else 0:>5d}-{max(walls) if walls else 0:<6d} " f"{st.median(faces) if faces else 0:>11,.0f} {min(faces) if faces else 0:>10,}-{max(faces) if faces else 0:<10,} {st.median(mb) if mb else 0:>7.1f}") def gen_inputs(): meta_path = os.path.join(INPUTS, 'inputs.json') if os.path.exists(meta_path): return json.load(open(meta_path)) # Published inputs (inputs/*.webp) without inputs.json → reuse them verbatim instead of paying for new ones. reused = {} for key in PROMPTS: path, fname, _ = input_file(key) if path: reused[key] = {'assetId': None, 'prompt': PROMPTS[key], 'model': 'reused published input', 'file': f'inputs/{fname}'} if len(reused) == len(PROMPTS): json.dump(reused, open(meta_path, 'w'), indent=2) print('reusing existing inputs:', ', '.join(v['file'] for v in reused.values()), flush=True) return reused out = {} for key, prompt in PROMPTS.items(): st, d = req('POST', '/v1/images', json_body={'prompt': prompt, 'aspectRatio': '1:1'}) if st not in (200, 201, 202) or not d.get('data', {}).get('id'): raise SystemExit(f'image gen failed {key}: {st} {d}') aid = d['data']['id']; sub = now() a, done = poll(aid, max_s=900) if not a or a.get('taskStatus') != 2: raise SystemExit(f'image gen not succeeded {key}: {a and a.get("status")}') url = a['files']['image'] path = os.path.join(INPUTS, f'{key}.png') download(url, path) out[key] = {'assetId': aid, 'prompt': prompt, 'model': (a.get('details') or {}).get('textToImage', {}).get('model'), 'submittedAt': sub, 'completedAt': done, 'file': f'inputs/{key}.png'} print('input ready', key, aid, out[key]['model'], flush=True) json.dump(out, open(meta_path, 'w'), indent=2) return out def run_job(job, inputs, results, results_path): key, engine, rep = job['input'], job['engine'], job['repeat'] path, fname, mime = input_file(key) if not path: path = os.path.join(OUT, inputs[key]['file']); fname = os.path.basename(path); mime = 'image/webp' if fname.endswith('.webp') else 'image/png' blob = open(path, 'rb').read() rec = {'input': key, 'engine': engine, 'repeat': rep, 'submittedAt': job.get('submittedAt') or now()} if job.get('assetId'): rec['assetId'] = job['assetId']; rec['adopted'] = True else: for attempt in range(4): st, d = req('POST', '/v1/models/from-image', form={'image': (fname, blob, mime), 'engine': engine, 'texture': 'true'}) if st in (200, 201, 202) and d.get('data', {}).get('id'): rec['assetId'] = d['data']['id'] with lock: pend = json.load(open(PENDING)) if os.path.exists(PENDING) else [] pend.append({'assetId': rec['assetId'], 'input': key, 'engine': engine, 'repeat': rep, 'submittedAt': rec['submittedAt']}) json.dump(pend, open(PENDING, 'w'), indent=2) break rec.setdefault('submitErrors', []).append({'status': st, 'body': d}) if st == 429 or st >= 500: time.sleep(30 * (attempt + 1)); continue break if 'assetId' not in rec: rec['status'] = 'submit_failed'; rec['completedAt'] = now() else: a, done = poll(rec['assetId']) rec['completedAt'] = done if not a: rec['status'] = 'timeout' else: rec['status'] = 'succeeded' if a.get('taskStatus') == 2 else 'failed' rec['taskStartedAt'] = a.get('taskStartedAt'); rec['apiStatus'] = a.get('status') det = a.get('details') or {} rec['engineReported'] = (det.get('imageTo3D') or {}).get('model') rec['error'] = det.get('error') or det.get('failReason') or a.get('errorMessage') files = a.get('files') or {} rec['hasTextures'] = files.get('hasTextures') if rec['status'] == 'succeeded': if files.get('stats'): try: with urllib.request.urlopen(files['stats'], timeout=60) as res: rec['stats'] = json.loads(res.read().decode()) except Exception as e: rec['stats'] = {'error': str(e)} rec['modelBytes'] = head_size(files['model']) if files.get('model') else None rec['previewBytes'] = head_size(files['preview']) if files.get('preview') else None if files.get('thumbnail'): tp = os.path.join(RESULTS, f'{engine}__{key}__{rep}.webp') if download(files['thumbnail'], tp): rec['thumbnail'] = os.path.relpath(tp, OUT) t0 = datetime.fromisoformat(rec['submittedAt'].replace('Z', '+00:00')); t1 = datetime.fromisoformat(rec['completedAt'].replace('Z', '+00:00')) rec['wallSeconds'] = round((t1 - t0).total_seconds()) with lock: results.append(rec); json.dump(results, open(results_path, 'w'), indent=2) print(f"{rec['status']:9s} {engine:17s} {key:14s} #{rep} {rec['wallSeconds']}s faces={ (rec.get('stats') or {}).get('faces') }", flush=True) return rec def main(): st, cat = req('GET', '/v1/models?category=image-to-3d') # image-to-3D prices (task type 1); the default '3d' is text-to-3D catalog = {m['name']: {k: m.get(k) for k in ('label', 'cost', 'textureCost', 'category', 'provider', 'engine', 'polygonConfig')} for m in cat.get('data', [])} json.dump({'fetchedAt': now(), 'models': catalog}, open(os.path.join(OUT, 'catalog.json'), 'w'), indent=2) st, cr = req('GET', '/v1/credits'); credits_before = cr.get('data', {}).get('totalAmount') inputs = gen_inputs() results_path = os.path.join(OUT, 'results.json') results = load_runs(results_path) done = {(r['input'], r['engine'], r['repeat']) for r in results if r.get('status') in ('succeeded', 'failed')} # 이미 제출된(진행 중) 자산을 다시 제출하지 않는다 — pending.json + 계정의 최근 자산(엔진·입력 해시로 대조). import hashlib inv = {hashlib.sha256(open(os.path.join(OUT, v['file']), 'rb').read()).hexdigest(): k for k, v in inputs.items()} known = {r.get('assetId') for r in results} | set() adopt = [] pend = json.load(open(PENDING)) if os.path.exists(PENDING) else [] for pjob in pend: if pjob['assetId'] not in known and (pjob['input'], pjob['engine'], pjob['repeat']) not in done: adopt.append(pjob); known.add(pjob['assetId']) st, la = req('GET', '/v1/assets?limit=100') run_start = min(v['completedAt'] for v in inputs.values()) for a in (la.get('data') or []): if a.get('type') != 0 or a['id'] in known or a.get('createdAt', '') < run_start: continue eng = ((a.get('details') or {}).get('imageTo3D') or {}).get('model'); key = inv.get((a.get('hashExt') or '').split('.')[0]) if eng in ENGINES and key: taken = {(r['input'], r['engine'], r['repeat']) for r in results} | {(j['input'], j['engine'], j['repeat']) for j in adopt} slot = next((n for n in range(1, REPEATS + 1) if (key, eng, n) not in taken), None) if slot: adopt.append({'input': key, 'engine': eng, 'repeat': slot, 'assetId': a['id'], 'submittedAt': a['createdAt']}); known.add(a['id']) taken = {(r['input'], r['engine'], r['repeat']) for r in results} | {(j['input'], j['engine'], j['repeat']) for j in adopt} jobs = adopt + [{'input': k, 'engine': e, 'repeat': n} for n in range(1, REPEATS + 1) for k in inputs for e in ENGINES if (k, e, n) not in taken] print(f'adopting {len(adopt)} in-flight assets', flush=True) print(f'{len(jobs)} jobs, concurrency {CONCURRENCY}, credits before {credits_before}', flush=True) with ThreadPoolExecutor(max_workers=CONCURRENCY) as ex: list(ex.map(lambda j: run_job(j, inputs, results, results_path), jobs)) st, cr = req('GET', '/v1/credits'); credits_after = cr.get('data', {}).get('totalAmount') json.dump({'startedAt': results[0]['submittedAt'] if results else None, 'finishedAt': now(), 'creditsBefore': credits_before, 'creditsAfter': credits_after, 'engines': ENGINES, 'repeats': REPEATS, 'concurrency': CONCURRENCY, 'settings': {'texture': True, 'polycount': 'engine default', 'input': 'single image, 1:1'}}, open(os.path.join(OUT, 'run-meta.json'), 'w'), indent=2) print('done; credits', credits_before, '->', credits_after) if __name__ == '__main__': if '--summarize' in sys.argv: i = sys.argv.index('--summarize') summarize(sys.argv[i + 1] if len(sys.argv) > i + 1 else os.path.join(OUT, 'results.json')) else: if not KEY: raise SystemExit('PICOBERRY_API_KEY is not set — this mode generates and costs credits. Use --summarize for the free local re-aggregation.') main()