refactor: Move assets to correct categories (fence→environment, chest→items, campfire already in workstations)
This commit is contained in:
@@ -1,183 +1,152 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script - generates single asset to verify ComfyUI workflow
|
||||
"""
|
||||
import os
|
||||
import json
|
||||
import time
|
||||
import requests
|
||||
from pathlib import Path
|
||||
import copy
|
||||
import subprocess
|
||||
|
||||
COMFYUI_URL = "http://127.0.0.1:8000"
|
||||
OUTPUT_DIR = "/Users/davidkotnik/repos/novafarma/assets/images"
|
||||
|
||||
# Test prompt with all requirements
|
||||
TEST_PROMPT = "Top-down 2.5D video game sprite, 128x128 pixels, smooth vector art style like Stardew Valley, NO pixels, vibrant colors, clean edges, GREEN SCREEN background RGBA(0,255,0,255) pure green chroma key, Normal farm cow black and white Holstein pattern friendly face 4-direction sprite"
|
||||
|
||||
def queue_comfy_prompt(prompt_text: str, output_name: str) -> dict:
|
||||
"""Send prompt to ComfyUI"""
|
||||
workflow = {
|
||||
"3": {
|
||||
"inputs": {
|
||||
"seed": int(time.time()),
|
||||
"steps": 25,
|
||||
"cfg": 7.5,
|
||||
"sampler_name": "euler",
|
||||
"scheduler": "normal",
|
||||
"denoise": 1,
|
||||
"model": ["4", 0],
|
||||
"positive": ["6", 0],
|
||||
"negative": ["7", 0],
|
||||
"latent_image": ["5", 0]
|
||||
},
|
||||
"class_type": "KSampler"
|
||||
workflow = {
|
||||
"3": {
|
||||
"inputs": {
|
||||
"seed": int(time.time()),
|
||||
"steps": 25,
|
||||
"cfg": 7.5,
|
||||
"sampler_name": "euler",
|
||||
"scheduler": "normal",
|
||||
"denoise": 1,
|
||||
"model": ["4", 0],
|
||||
"positive": ["6", 0],
|
||||
"negative": ["7", 0],
|
||||
"latent_image": ["5", 0]
|
||||
},
|
||||
"4": {
|
||||
"inputs": {
|
||||
"ckpt_name": "sd_xl_base_1.0.safetensors"
|
||||
},
|
||||
"class_type": "CheckpointLoaderSimple"
|
||||
"class_type": "KSampler"
|
||||
},
|
||||
"4": {
|
||||
"inputs": {
|
||||
"ckpt_name": "v1-5-pruned-emaonly.safetensors"
|
||||
},
|
||||
"5": {
|
||||
"inputs": {
|
||||
"width": 128,
|
||||
"height": 128,
|
||||
"batch_size": 1
|
||||
},
|
||||
"class_type": "EmptyLatentImage"
|
||||
"class_type": "CheckpointLoaderSimple"
|
||||
},
|
||||
"5": {
|
||||
"inputs": {
|
||||
"width": 512,
|
||||
"height": 512,
|
||||
"batch_size": 1
|
||||
},
|
||||
"6": {
|
||||
"inputs": {
|
||||
"text": prompt_text,
|
||||
"clip": ["4", 1]
|
||||
},
|
||||
"class_type": "CLIPTextEncode"
|
||||
"class_type": "EmptyLatentImage"
|
||||
},
|
||||
"6": {
|
||||
"inputs": {
|
||||
"text": "",
|
||||
"clip": ["4", 1],
|
||||
"class_type": "CLIPTextEncode"
|
||||
},
|
||||
"7": {
|
||||
"inputs": {
|
||||
"text": "blurry, low quality, pixelated, photo, realistic",
|
||||
"clip": ["4", 1]
|
||||
},
|
||||
"7": {
|
||||
"inputs": {
|
||||
"text": "blurry, low quality, pixelated, voxel, 3D render, realistic photo",
|
||||
"clip": ["4", 1]
|
||||
},
|
||||
"class_type": "CLIPTextEncode"
|
||||
"class_type": "CLIPTextEncode"
|
||||
},
|
||||
"8": {
|
||||
"inputs": {
|
||||
"samples": ["3", 0],
|
||||
"vae": ["4", 2]
|
||||
},
|
||||
"8": {
|
||||
"inputs": {
|
||||
"samples": ["3", 0],
|
||||
"vae": ["4", 2]
|
||||
},
|
||||
"class_type": "VAEDecode"
|
||||
"class_type": "VAEDecode"
|
||||
},
|
||||
"9": {
|
||||
"inputs": {
|
||||
"filename_prefix": "",
|
||||
"images": ["8", 0],
|
||||
},
|
||||
"9": {
|
||||
"inputs": {
|
||||
"filename_prefix": output_name,
|
||||
"images": ["8", 0]
|
||||
},
|
||||
"class_type": "SaveImage"
|
||||
}
|
||||
"class_type": "SaveImage"
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
def queue_prompt(prompt):
|
||||
p = {"prompt": prompt}
|
||||
data = json.dumps(p).encode('utf-8')
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{COMFYUI_URL}/prompt",
|
||||
json={"prompt": workflow}
|
||||
)
|
||||
return response.json()
|
||||
res = requests.post(COMFYUI_URL + "/prompt", data=data)
|
||||
if res.status_code == 200:
|
||||
print("🚀 SUCCESS: Poslano v ComfyUI!")
|
||||
return res.json()
|
||||
else:
|
||||
print(f"❌ Napaka: {res.text}")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"❌ Error: {e}")
|
||||
print(f"❌ UFI (ComfyUI) ni prižgan! {e}")
|
||||
return None
|
||||
|
||||
def wait_for_completion(prompt_id: str, timeout: int = 180) -> bool:
|
||||
"""Wait for generation"""
|
||||
start = time.time()
|
||||
while time.time() - start < timeout:
|
||||
try:
|
||||
response = requests.get(f"{COMFYUI_URL}/history/{prompt_id}")
|
||||
history = response.json()
|
||||
|
||||
r = requests.get(f"{COMFYUI_URL}/history/{prompt_id}")
|
||||
history = r.json()
|
||||
if prompt_id in history:
|
||||
status = history[prompt_id].get("status", {})
|
||||
if status.get("completed", False):
|
||||
status = history[prompt_id].get('status', {})
|
||||
if status.get('completed'):
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Polling: {e}")
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(2)
|
||||
|
||||
return False
|
||||
|
||||
def download_image(prompt_id: str, output_path: Path) -> bool:
|
||||
"""Download image"""
|
||||
def download_image(prompt_id: str, prefix: str):
|
||||
try:
|
||||
response = requests.get(f"{COMFYUI_URL}/history/{prompt_id}")
|
||||
history = response.json()
|
||||
|
||||
r = requests.get(f"{COMFYUI_URL}/history/{prompt_id}")
|
||||
history = r.json()
|
||||
if prompt_id not in history:
|
||||
return False
|
||||
|
||||
outputs = history[prompt_id].get("outputs", {})
|
||||
for node_id, node_output in outputs.items():
|
||||
if "images" in node_output:
|
||||
for img in node_output["images"]:
|
||||
filename = img["filename"]
|
||||
subfolder = img.get("subfolder", "")
|
||||
|
||||
img_url = f"{COMFYUI_URL}/view"
|
||||
params = {
|
||||
"filename": filename,
|
||||
"subfolder": subfolder,
|
||||
"type": "output"
|
||||
}
|
||||
|
||||
img_response = requests.get(img_url, params=params)
|
||||
if img_response.status_code == 200:
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(output_path, 'wb') as f:
|
||||
f.write(img_response.content)
|
||||
outputs = history[prompt_id].get('outputs', {})
|
||||
for node_id, out in outputs.items():
|
||||
if 'images' in out:
|
||||
for img in out['images']:
|
||||
filename = img['filename']
|
||||
subfolder = img.get('subfolder', '')
|
||||
url = f"{COMFYUI_URL}/view"
|
||||
params = {'filename': filename, 'subfolder': subfolder, 'type': 'output'}
|
||||
img_resp = requests.get(url, params=params)
|
||||
if img_resp.status_code == 200:
|
||||
out_path = os.path.join(OUTPUT_DIR, f"{prefix}_{filename}")
|
||||
os.makedirs(os.path.dirname(out_path), exist_ok=True)
|
||||
with open(out_path, 'wb') as f:
|
||||
f.write(img_resp.content)
|
||||
print(f"✅ Image saved: {out_path}")
|
||||
# subprocess.run(["open", out_path])
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Download error: {e}")
|
||||
return False
|
||||
|
||||
print("=" * 70)
|
||||
print("🧪 TEST: Single Asset Generation")
|
||||
print("=" * 70)
|
||||
print(f"📡 Server: {COMFYUI_URL}")
|
||||
print(f"🎨 Prompt: {TEST_PROMPT[:80]}...")
|
||||
print()
|
||||
# Updated generate_character to use new helpers
|
||||
|
||||
# Check connection
|
||||
try:
|
||||
response = requests.get(f"{COMFYUI_URL}/system_stats")
|
||||
print("✅ ComfyUI connected!")
|
||||
except:
|
||||
print("❌ ComfyUI NOT responding!")
|
||||
exit(1)
|
||||
|
||||
# Queue test
|
||||
print("\n🚀 Queuing test generation...")
|
||||
result = queue_comfy_prompt(TEST_PROMPT, "test_cow")
|
||||
if not result or "prompt_id" not in result:
|
||||
print("❌ Failed to queue!")
|
||||
exit(1)
|
||||
|
||||
prompt_id = result["prompt_id"]
|
||||
print(f"⏳ Queued: {prompt_id}")
|
||||
|
||||
# Wait
|
||||
print("⏳ Waiting for completion...")
|
||||
if wait_for_completion(prompt_id):
|
||||
output_path = Path(OUTPUT_DIR) / "zivali" / "test_cow.png"
|
||||
if download_image(prompt_id, output_path):
|
||||
print(f"✅ SUCCESS! Saved to: {output_path}")
|
||||
print("\n📊 Check the image:")
|
||||
print(f" - Size should be 128x128")
|
||||
print(f" - Background should be green RGBA(0,255,0,255)")
|
||||
print(f" - Style should be smooth 2D vector art")
|
||||
def generate_character(name, prompt_text, prefix):
|
||||
wf = copy.deepcopy(workflow)
|
||||
wf["6"]["inputs"]["text"] = prompt_text
|
||||
wf["9"]["inputs"]["filename_prefix"] = prefix
|
||||
result = queue_prompt(wf)
|
||||
if not result or 'prompt_id' not in result:
|
||||
print('❌ Failed to get prompt_id')
|
||||
return
|
||||
pid = result['prompt_id']
|
||||
if wait_for_completion(pid):
|
||||
download_image(pid, prefix)
|
||||
else:
|
||||
print("❌ Failed to download!")
|
||||
else:
|
||||
print("⏱️ Timeout!")
|
||||
print('❌ Generation timed out')
|
||||
|
||||
# Generate 5 Kai images
|
||||
kai_prompt = "Top-down 2.5D, (bold black outlines:1.4), dark hand-drawn stylized indie game art, high contrast noir style, character Kai with pink and green dreadlocks, friendly smile"
|
||||
for i in range(5):
|
||||
generate_character("Kai", kai_prompt, f"noir_kai_{i}")
|
||||
|
||||
# Generate 5 Gronk images
|
||||
gronk_prompt = "Top-down 2.5D, (bold black outlines:1.4), dark hand-drawn stylized indie game art, high contrast noir style, character Gronk with pink dreadlocks, stretched ears, facial piercings, holding a vape"
|
||||
for i in range(5):
|
||||
generate_character("Gronk", gronk_prompt, f"noir_gronk_{i}")
|
||||
Reference in New Issue
Block a user