`scripts/_regen_v4_deco.py`

100 lines
#!/usr/bin/env python3
"""Targeted _v4 regen for ONE brief (deco-01) to finally land a CHARTREUSE ground.
Prior v1/v2/v3 rendered dark olive/near-black. Reuses gen_briefs.py machinery
(same Replicate SDXL model, poll loop, cost accounting) with a color-anchored prompt
that front-loads the flat ground color HARD + strong dark-background negatives +
guidance_scale bumped to 9.5 so the color prompt binds harder.
NEVER overwrites v1/v2/v3. Curator-mode, is_published untouched (no publish here)."""
import os, re, json, time, urllib.request, urllib.error

ROOT = "/Users/macstudio3/Projects/pattern-vault"
OUT = f"{ROOT}/assets/generated"
os.makedirs(OUT, exist_ok=True)

tok = None
with open(os.path.expanduser("~/Projects/secrets-manager/.env")) as f:
    m = re.search(r"REPLICATE_API_TOKEN=([A-Za-z0-9_-]+)", f.read())
    if m: tok = m.group(1)
assert tok, "no replicate token"

# Color-anchored prompt: front-load the flat ground color HARD (verbatim from task).
JOBS = {
    "BRIEF-2026-deco-01": {
        "prompt": ("flat solid CHARTREUSE yellow-green background color #B0C000 filling the entire canvas, "
                   "Art-Deco geometric wallpaper printed on top in thin METALLIC GOLD linework - "
                   "fan and sunburst arcs, stepped chevrons, concentric semicircles, crisp 1920s Gatsby symmetry, "
                   "seamless tile, flat vector poster art."),
        "negative": ("black background, dark background, navy, charcoal, palm, fronds, leaves, botanical, "
                     "flowers, birds, animals, photorealistic, gradient background"),
        "w": 1024, "h": 1024,
    },
}

UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
def post(url, payload):
    hdr = {"Authorization": f"Bearer {tok}", "Content-Type": "application/json", "User-Agent": UA}
    req = urllib.request.Request(url, data=json.dumps(payload).encode(), headers=hdr, method="POST")
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)

def get(url):
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {tok}", "User-Agent": UA})
    with urllib.request.urlopen(req, timeout=60) as r:
        return json.load(r)

MODEL = "stability-ai/sdxl"
VERSION = get(f"https://api.replicate.com/v1/models/{MODEL}")["latest_version"]["id"]
print(f"using {MODEL}@{VERSION[:12]}", flush=True)

results = []
total_secs = 0.0

# Exactly 2 variants, tagged _v4a/_v4b. Fresh seeds 505/606. guidance_scale 9.5.
VARIANTS = [("v4a", 505), ("v4b", 606)]
for bid, spec in JOBS.items():
    for vlabel, seed in VARIANTS:
        tag = f"{bid}_{vlabel}"
        payload = {"version": VERSION, "input": {
            "prompt": spec["prompt"], "negative_prompt": spec["negative"],
            "width": spec["w"], "height": spec["h"], "num_inference_steps": 30,
            "guidance_scale": 9.5, "seed": seed, "refine": "no_refiner"
        }}
        ok = False
        for attempt in range(4):
            try:
                pred = post("https://api.replicate.com/v1/predictions", payload); ok = True; break
            except urllib.error.HTTPError as e:
                body = e.read().decode()[:150]
                if e.code == 429:
                    print(f"[{tag}] 429 throttled, backoff {6*(attempt+1)}s", flush=True)
                    time.sleep(6*(attempt+1)); continue
                print(f"[{tag}] HTTP {e.code}: {body}", flush=True); break
        if not ok:
            results.append({"tag": tag, "status": "error"}); continue
        for _ in range(60):
            if pred.get("status") in ("succeeded", "failed", "canceled"): break
            time.sleep(2); pred = get(pred["urls"]["get"])
        st = pred.get("status")
        secs = (pred.get("metrics") or {}).get("predict_time", 0) or 0
        total_secs += secs
        if st == "succeeded":
            out = pred["output"]
            imgurl = out[0] if isinstance(out, list) else out
            dest = f"{OUT}/{tag}.png"
            urllib.request.urlretrieve(imgurl, dest)
            sz = os.path.getsize(dest)
            print(f"[{tag}] OK {secs:.1f}s -> {dest} ({sz//1024}KB)", flush=True)
            results.append({"tag": tag, "brief": bid, "path": dest, "status": "ok",
                            "predict_time": secs, "prompt": spec["prompt"], "seed": seed,
                            "guidance_scale": 9.5})
        else:
            print(f"[{tag}] {st}: {pred.get('error')}", flush=True)
            results.append({"tag": tag, "brief": bid, "status": st})
        time.sleep(1.5)

RATE = 0.000725
cost = total_secs * RATE
print(f"\n=== TOTAL predict_time {total_secs:.1f}s  est cost ${cost:.4f} (rate ${RATE}/s A40) ===", flush=True)
json.dump({"results": results, "total_predict_secs": total_secs, "est_cost_usd": round(cost, 4)},
          open(f"{OUT}/_regen_v4_deco_manifest.json", "w"), indent=2)