File size: 9,046 Bytes
478cb8f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | #!/usr/bin/env python3
"""Phase A calibration sweeps on inputs_v2, seed 777. Outputs under sweeps/ prefixes."""
import json, os, shutil, sys, time, urllib.request
HOST = "http://127.0.0.1:7865"
SEED = 777
W, H = 896, 1152
HW, HH = 1344, 1728
OUT_ROOT = "/workspace/outputs_v2/sweeps"
COMFY_OUT = "/workspace/ComfyUI/output"
PAIRS = [("r1.jpg", "t2.jpg"), ("r2.jpg", "t5.jpg"), ("r4.jpg", "t3.jpg")] # variety subset
def post(prompt):
req = urllib.request.Request(f"{HOST}/prompt",
data=json.dumps({"prompt": prompt}).encode(), headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req) as r:
d = json.loads(r.read())
if "prompt_id" not in d:
raise RuntimeError(d)
return d["prompt_id"]
def core(p, style_img, comp_img, guidance=3.5, redux=0.5):
p["u"] = {"class_type": "UNETLoader", "inputs": {"unet_name": "flux1-dev.safetensors", "weight_dtype": "default"}}
p["c"] = {"class_type": "DualCLIPLoader", "inputs": {"clip_name1": "t5xxl_fp16.safetensors",
"clip_name2": "clip_l.safetensors", "type": "flux", "device": "default"}}
p["v"] = {"class_type": "VAELoader", "inputs": {"vae_name": "ae.safetensors"}}
p["txt"] = {"class_type": "CLIPTextEncode", "inputs": {"clip": ["c", 0], "text": ""}}
p["guid"] = {"class_type": "FluxGuidance", "inputs": {"conditioning": ["txt", 0], "guidance": guidance}}
p["neg"] = {"class_type": "ConditioningZeroOut", "inputs": {"conditioning": ["guid", 0]}}
p["si"] = {"class_type": "LoadImage", "inputs": {"image": style_img}}
p["cvl"] = {"class_type": "CLIPVisionLoader", "inputs": {"clip_name": "sigclip_vision_patch14_384.safetensors"}}
p["cve"] = {"class_type": "CLIPVisionEncode", "inputs": {"clip_vision": ["cvl", 0], "image": ["si", 0], "crop": "center"}}
p["sml"] = {"class_type": "StyleModelLoader", "inputs": {"style_model_name": "flux1-redux-dev.safetensors"}}
p["sma"] = {"class_type": "StyleModelApply", "inputs": {"conditioning": ["guid", 0],
"style_model": ["sml", 0], "clip_vision_output": ["cve", 0], "strength": redux, "strength_type": "attn_bias"}}
p["ci"] = {"class_type": "LoadImage", "inputs": {"image": comp_img}}
p["rs"] = {"class_type": "ImageResize+", "inputs": {"image": ["ci", 0], "width": W, "height": H,
"interpolation": "lanczos", "method": "fill / crop", "condition": "always", "multiple_of": 0}}
p["depth"] = {"class_type": "DepthAnythingV2Preprocessor", "inputs": {
"image": ["rs", 0], "ckpt_name": "depth_anything_v2_vitl.pth", "resolution": W}}
def save(p, src, prefix):
p["dec"] = {"class_type": "VAEDecode", "inputs": {"samples": src, "vae": ["v", 0]}}
p["save"] = {"class_type": "SaveImage", "inputs": {"images": ["dec", 0], "filename_prefix": prefix}}
def cnet_wf(style_img, comp_img, tag, *, guidance=3.5, scheduler="simple", redux=0.5,
depth_str=0.7, canny=False, radv=False, hires=False):
p = {}
core(p, style_img, comp_img, guidance=guidance, redux=redux)
pos = ["sma", 0]
if radv:
del p["sma"], p["cve"]
p["radv"] = {"class_type": "ReduxAdvanced", "inputs": {"conditioning": ["guid", 0],
"style_model": ["sml", 0], "clip_vision": ["cvl", 0], "image": ["si", 0],
"downsampling_factor": 3, "downsampling_function": "area",
"mode": "center crop (square)", "weight": 1.0, "autocrop_margin": 0.1}}
pos = ["radv", 0]
p["cnl"] = {"class_type": "ControlNetLoader", "inputs": {"control_net_name": "FLUX.1-dev-ControlNet-Union-Pro-2.0.safetensors"}}
p["cn"] = {"class_type": "ControlNetApplySD3", "inputs": {"positive": pos, "negative": ["neg", 0],
"control_net": ["cnl", 0], "vae": ["v", 0], "image": ["depth", 0],
"strength": depth_str, "start_percent": 0.0, "end_percent": 0.8}}
last = "cn"
if canny:
p["cne"] = {"class_type": "Canny", "inputs": {"image": ["rs", 0], "low_threshold": 0.2, "high_threshold": 0.5}}
p["cn2"] = {"class_type": "ControlNetApplySD3", "inputs": {"positive": ["cn", 0], "negative": ["cn", 1],
"control_net": ["cnl", 0], "vae": ["v", 0], "image": ["cne", 0],
"strength": 0.35, "start_percent": 0.0, "end_percent": 0.6}}
last = "cn2"
p["msf"] = {"class_type": "ModelSamplingFlux", "inputs": {"model": ["u", 0],
"max_shift": 1.15, "base_shift": 0.5, "width": W, "height": H}}
p["lat"] = {"class_type": "EmptySD3LatentImage", "inputs": {"width": W, "height": H, "batch_size": 1}}
p["ks"] = {"class_type": "KSampler", "inputs": {"model": ["msf", 0], "positive": [last, 0],
"negative": [last, 1], "latent_image": ["lat", 0], "seed": SEED, "steps": 32, "cfg": 1.0,
"sampler_name": "euler", "scheduler": scheduler, "denoise": 1.0}}
final = ["ks", 0]
if hires:
p["up"] = {"class_type": "LatentUpscaleBy", "inputs": {"samples": ["ks", 0],
"upscale_method": "bislerp", "scale_by": 1.5}}
p["msf2"] = {"class_type": "ModelSamplingFlux", "inputs": {"model": ["u", 0],
"max_shift": 1.15, "base_shift": 0.5, "width": HW, "height": HH}}
p["ks2"] = {"class_type": "KSampler", "inputs": {"model": ["msf2", 0], "positive": pos,
"negative": ["neg", 0], "latent_image": ["up", 0], "seed": SEED, "steps": 32, "cfg": 1.0,
"sampler_name": "euler", "scheduler": scheduler, "denoise": 0.30}}
final = ["ks2", 0]
save(p, final, f"sweeps/{tag}")
return p
def bfl_wf(style_img, comp_img, tag, guidance):
p = {}
core(p, style_img, comp_img, guidance=guidance)
p["lora"] = {"class_type": "LoraLoaderModelOnly", "inputs": {"model": ["u", 0],
"lora_name": "flux1-depth-dev-lora.safetensors", "strength_model": 1.0}}
p["ip2p"] = {"class_type": "InstructPixToPixConditioning", "inputs": {"positive": ["sma", 0],
"negative": ["neg", 0], "vae": ["v", 0], "pixels": ["depth", 0]}}
p["msf"] = {"class_type": "ModelSamplingFlux", "inputs": {"model": ["lora", 0],
"max_shift": 1.15, "base_shift": 0.5, "width": W, "height": H}}
p["ks"] = {"class_type": "KSampler", "inputs": {"model": ["msf", 0], "positive": ["ip2p", 0],
"negative": ["ip2p", 1], "latent_image": ["ip2p", 2], "seed": SEED, "steps": 32, "cfg": 1.0,
"sampler_name": "euler", "scheduler": "simple", "denoise": 1.0}}
save(p, ["ks", 0], f"sweeps/{tag}")
return p
def main():
q = []
def tagname(r, t): return f"{r.split('.')[0]}x{t.split('.')[0]}"
for r, t in PAIRS: # A1: bfl guidance
for gv in [4.0, 7.0, 10.0]:
q.append((f"A1-bfl-g{gv:g}-{tagname(r,t)}", bfl_wf(r, t, f"A1-bfl-g{gv:g}-{tagname(r,t)}", gv)))
for r, t in PAIRS: # A2: guidance x scheduler
for gv in [2.5, 3.0, 3.5]:
for sch in ["simple", "beta"]:
tag = f"A2-g{gv:g}-{sch}-{tagname(r,t)}"
q.append((tag, cnet_wf(r, t, tag, guidance=gv, scheduler=sch)))
for r, t in PAIRS[:2]: # A3: redux x depth strength
for rx in [0.4, 0.5, 0.7]:
for ds in [0.55, 0.7, 0.85]:
tag = f"A3-rx{rx:g}-ds{ds:g}-{tagname(r,t)}"
q.append((tag, cnet_wf(r, t, tag, redux=rx, depth_str=ds)))
for r, t in PAIRS: # A4: hires
tag = f"A4-hires-{tagname(r,t)}"
q.append((tag, cnet_wf(r, t, tag, hires=True)))
for r, t in PAIRS: # A5: ReduxAdvanced
tag = f"A5-radv-{tagname(r,t)}"
q.append((tag, cnet_wf(r, t, tag, radv=True)))
for r, t in PAIRS[:2]: # A6: canny stack
tag = f"A6-canny-{tagname(r,t)}"
q.append((tag, cnet_wf(r, t, tag, canny=True)))
ids = {}
for tag, prompt in q:
ids[post(prompt)] = tag
print("queued", tag, flush=True)
pending, errors = set(ids), []
while pending:
time.sleep(10)
for pid in list(pending):
try:
with urllib.request.urlopen(f"{HOST}/history/{pid}") as r:
h = json.loads(r.read())
except Exception:
continue
if pid not in h: continue
st = h[pid].get("status", {})
if st.get("completed"):
pending.discard(pid); print(f"done {ids[pid]} ({len(ids)-len(pending)}/{len(ids)})", flush=True)
elif st.get("status_str") == "error":
pending.discard(pid); errors.append(ids[pid])
msgs = [m for m in st.get("messages", []) if m[0] == "execution_error"]
print(f"ERROR {ids[pid]}: {(msgs[-1][1].get('exception_message','?') if msgs else '?')[:300]}", flush=True)
os.makedirs(OUT_ROOT, exist_ok=True)
src = os.path.join(COMFY_OUT, "sweeps")
for f in sorted(os.listdir(src)):
shutil.copy(os.path.join(src, f), os.path.join(OUT_ROOT, f.split("_")[0] + ".png"))
print(f"COLLECTED {len(os.listdir(OUT_ROOT))} sweep outputs", flush=True)
if errors:
print("ERRORS:", errors); sys.exit(1)
print("SWEEPS COMPLETE", flush=True)
if __name__ == "__main__":
main()
|