modelbeast/server/operators/flux_local/run.py
MODELBEAST 76f065d3a6 A/B image-gen: local FLUX (mflux/MLX) + OpenRouter nano-banana operators
- flux_local: prompt->image fully on-device via mflux (schnell/dev, steps,
  quantize, seed, size). Both FLUX repos are HF-gated as of 2026-07 (schnell
  included) — clean GatedRepoError surfaces with a hint; needs owner HF token.
- openrouter_image: prompt->image via OpenRouter chat/completions with
  modalities [image,text]; parses data-URL images from the response; model
  picker for nano-banana / nano-banana-pro. Gated on OPENROUTER_API_KEY.
- settings: openrouter_key added to vault (masked/redacted/env-injected)
- scripts/install_mflux.sh; venv installed
- A/B flow: run both with the same prompt, judge in Compare mode

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-12 22:31:19 +10:00

69 lines
2.1 KiB
Python

import argparse
import json
import subprocess
import sys
import time
from pathlib import Path
ap = argparse.ArgumentParser()
ap.add_argument("--input", action="append", default=[])
ap.add_argument("--outdir", required=True)
ap.add_argument("--params", default="{}")
a = ap.parse_args()
p = json.loads(a.params)
if not p.get("prompt"):
print("ERROR: prompt is required")
sys.exit(1)
model = p.get("model", "schnell")
steps = int(p.get("steps", 4))
outdir = Path(a.outdir)
out_png = outdir / f"flux_{model}_s{p.get('seed', 42)}.png"
# mflux-generate lives next to this venv's python
cli = Path(sys.executable).parent / "mflux-generate"
if not cli.exists():
print(f"ERROR: mflux not installed ({cli} missing). Run scripts/install_mflux.sh")
sys.exit(1)
cmd = [
str(cli),
"--model", model,
"--prompt", p["prompt"],
"--steps", str(steps),
"--width", str(p.get("width", 1024)),
"--height", str(p.get("height", 1024)),
"--seed", str(p.get("seed", 42)),
"--output", str(out_png),
]
quant = str(p.get("quantize", "8"))
if quant in ("4", "8"):
cmd += ["--quantize", quant]
if model == "dev":
cmd += ["--guidance", str(p.get("guidance", 3.5))]
print("+", " ".join(cmd), flush=True)
print("(first run downloads weights from HuggingFace — can take a while)", flush=True)
t0 = time.time()
res = subprocess.run(cmd, stdout=sys.stdout, stderr=subprocess.STDOUT)
if res.returncode != 0:
print("HINT: if this failed with a gated-repo/auth error, the model needs an "
"HF license accept + token (Settings → HuggingFace token). schnell is ungated.")
sys.exit(res.returncode)
elapsed = round(time.time() - t0, 1)
if not out_png.exists():
# mflux may append suffixes; grab any png it wrote
pngs = sorted(outdir.glob("*.png"))
if not pngs:
print("ERROR: no image produced")
sys.exit(1)
out_png = pngs[-1]
(outdir / "result.json").write_text(json.dumps({"outputs": [
{"path": out_png.name,
"meta": {"tool": "mflux", "model": model, "steps": steps,
"seed": p.get("seed", 42), "quantize": quant, "seconds": elapsed}}]}))
print(f"done in {elapsed}s: {out_png.name}", flush=True)