175 lines
6.9 KiB
Python
175 lines
6.9 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
meeting-103 v2.1 라운드 파이프라인
|
||
|
|
- 결과 회수 (HTTP /view)
|
||
|
|
- 썸네일 생성 (1MB 미만)
|
||
|
|
- 자동 평가 (black/pastel/aspect)
|
||
|
|
- 자산 라이브러리 7종 누적
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import urllib.request
|
||
|
|
from pathlib import Path
|
||
|
|
from PIL import Image
|
||
|
|
|
||
|
|
REPO_ROOT = Path("/mnt/ssd1/dev/projects/8460s-image-rd")
|
||
|
|
COMFY = "http://100.123.6.62:8188"
|
||
|
|
|
||
|
|
THRESHOLDS = {"byeolyi": 8, "hanja": 4.5, "hanok": 4.5}
|
||
|
|
PROTECTED = {"qwen-image-2.0", "qwen-image-2512", "qwen-image-layered", "qwen-edit-2511",
|
||
|
|
"flux-dev", "flux-schnell", "flux-kontext", "flux-fill", "flux-redux",
|
||
|
|
"pulid-flux", "pony-v6", "illustrious", "lightning-lora-qwen"}
|
||
|
|
|
||
|
|
|
||
|
|
def http_view(filename):
|
||
|
|
"""ComfyUI HTTP /view 으로 PNG 다운로드 (bytes 반환)"""
|
||
|
|
url = f"{COMFY}/view?filename={filename}&type=output"
|
||
|
|
with urllib.request.urlopen(url, timeout=20) as r:
|
||
|
|
return r.read()
|
||
|
|
|
||
|
|
|
||
|
|
def history_outputs(prompt_id):
|
||
|
|
"""prompt_id → 출력 파일명 목록"""
|
||
|
|
url = f"{COMFY}/history/{prompt_id}"
|
||
|
|
with urllib.request.urlopen(url, timeout=10) as r:
|
||
|
|
data = json.load(r)
|
||
|
|
if not data:
|
||
|
|
return []
|
||
|
|
info = list(data.values())[0]
|
||
|
|
outs = info.get("outputs", {})
|
||
|
|
return [f["filename"] for n, i in outs.items() for f in i.get("images", [])]
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_round_results(round_n, combos):
|
||
|
|
"""combos = [{"id": "1-1", "prompt_id": "...", "expected_filename": "..."}]"""
|
||
|
|
round_dir = REPO_ROOT / "results" / "meeting-103" / f"round{round_n}"
|
||
|
|
round_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
results = []
|
||
|
|
for c in combos:
|
||
|
|
fns = history_outputs(c["prompt_id"]) if c.get("prompt_id") else []
|
||
|
|
fn = fns[0] if fns else c.get("expected_filename")
|
||
|
|
if not fn:
|
||
|
|
results.append({**c, "status": "no_filename"})
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
data = http_view(fn)
|
||
|
|
out_png = round_dir / f"{c['id']}.png"
|
||
|
|
out_png.write_bytes(data)
|
||
|
|
size = len(data)
|
||
|
|
# 썸네일 (1MB 미만 의무)
|
||
|
|
img = Image.open(out_png).convert("RGB")
|
||
|
|
thumb = img.copy()
|
||
|
|
thumb.thumbnail((512, 768))
|
||
|
|
thumb_path = round_dir / f"{c['id']}_thumb.png"
|
||
|
|
thumb.save(thumb_path, "PNG", optimize=True)
|
||
|
|
assert thumb_path.stat().st_size < 1_000_000, "thumbnail >1MB"
|
||
|
|
# 자동 평가
|
||
|
|
w, h = img.size
|
||
|
|
pixels = list(img.getdata())
|
||
|
|
total = len(pixels)
|
||
|
|
near_black = sum(1 for r, g, b in pixels if r < 30 and g < 30 and b < 30)
|
||
|
|
pastel = sum(1 for r, g, b in pixels if (r > 150 and b > 150) or (r > 200 and g > 150 and b > 150))
|
||
|
|
results.append({
|
||
|
|
**c, "status": "OK" if size > 50000 else "FAILED_SMALL",
|
||
|
|
"size_bytes": size, "width": w, "height": h,
|
||
|
|
"aspect": round(w / h, 4),
|
||
|
|
"black_pct": round(near_black / total * 100, 2),
|
||
|
|
"pastel_pct": round(pastel / total * 100, 1),
|
||
|
|
"thumb_size_kb": thumb_path.stat().st_size // 1024,
|
||
|
|
})
|
||
|
|
except Exception as e:
|
||
|
|
results.append({**c, "status": f"ERROR:{type(e).__name__}", "error": str(e)[:120]})
|
||
|
|
return results
|
||
|
|
|
||
|
|
|
||
|
|
def update_models_performance(round_n, combo_results):
|
||
|
|
"""자산 라이브러리 1 — 모델 차원별 점수 누적"""
|
||
|
|
path = REPO_ROOT / "assets-library" / "models-performance.json"
|
||
|
|
data = json.loads(path.read_text())
|
||
|
|
for c in combo_results:
|
||
|
|
if c.get("status") != "OK":
|
||
|
|
continue
|
||
|
|
for model in c.get("models_used", []):
|
||
|
|
mdata = data["models"].setdefault(model, {
|
||
|
|
"rounds_used": 0,
|
||
|
|
"dimensions": {dim: [] for dim in data["_dimensions"]},
|
||
|
|
"avg_scores": {}, "best_combinations": [], "verdict": ""
|
||
|
|
})
|
||
|
|
mdata["rounds_used"] = max(mdata["rounds_used"], round_n)
|
||
|
|
scores = c.get("scores", {})
|
||
|
|
for dim, score in scores.items():
|
||
|
|
if score is not None and dim in mdata["dimensions"]:
|
||
|
|
mdata["dimensions"][dim].append(score)
|
||
|
|
for dim, arr in mdata["dimensions"].items():
|
||
|
|
if arr:
|
||
|
|
mdata["avg_scores"][dim] = round(sum(arr) / len(arr), 2)
|
||
|
|
path.write_text(json.dumps(data, indent=2, ensure_ascii=False))
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def save_workflow_to_library(combo, workflow_json, round_n):
|
||
|
|
"""자산 라이브러리 2 — winner/partial 워크플로 보존"""
|
||
|
|
scores = combo.get("scores", {})
|
||
|
|
if not scores:
|
||
|
|
return None
|
||
|
|
if all(scores.get(d, 0) >= t for d, t in THRESHOLDS.items()):
|
||
|
|
category = "winner"
|
||
|
|
elif max(scores.values()) >= 5:
|
||
|
|
category = "partial"
|
||
|
|
else:
|
||
|
|
return None
|
||
|
|
best_dim = max(scores, key=scores.get)
|
||
|
|
model = (combo.get("models_used") or ["unknown"])[0]
|
||
|
|
name = f"{category}-R{round_n}-{combo['id']}-{model}.json"
|
||
|
|
path = REPO_ROOT / "assets-library" / "workflows" / name
|
||
|
|
path.write_text(json.dumps(workflow_json, indent=2, ensure_ascii=False))
|
||
|
|
# 인덱스 갱신
|
||
|
|
idx_path = REPO_ROOT / "assets-library" / "workflows" / "_index.json"
|
||
|
|
idx = json.loads(idx_path.read_text())
|
||
|
|
entry = {"name": name, "category": category, "best_dimension": best_dim,
|
||
|
|
"scores": scores, "round": round_n, "model": model}
|
||
|
|
idx.setdefault(category, []).append(entry)
|
||
|
|
idx["count"] = len(idx.get("winner", [])) + len(idx.get("partial", []))
|
||
|
|
idx_path.write_text(json.dumps(idx, indent=2, ensure_ascii=False))
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def save_results_meta(round_n, results, learning=None):
|
||
|
|
"""라운드 메타 JSON 저장 (그리드 보조)"""
|
||
|
|
path = REPO_ROOT / "results" / "meeting-103" / f"round{round_n}" / "meta.json"
|
||
|
|
payload = {
|
||
|
|
"round": round_n,
|
||
|
|
"captured_at": __import__("datetime").datetime.now().isoformat(),
|
||
|
|
"combos": results,
|
||
|
|
"learning": learning or {}
|
||
|
|
}
|
||
|
|
path.write_text(json.dumps(payload, indent=2, ensure_ascii=False))
|
||
|
|
return path
|
||
|
|
|
||
|
|
|
||
|
|
def extract_learning(results):
|
||
|
|
"""라운드 학습 추출 (RAG-style)"""
|
||
|
|
passed = []
|
||
|
|
failed = []
|
||
|
|
for r in results:
|
||
|
|
if r.get("status") != "OK":
|
||
|
|
failed.append({"id": r["id"], "reason": r.get("status")})
|
||
|
|
continue
|
||
|
|
if r.get("black_pct", 0) > 5:
|
||
|
|
failed.append({"id": r["id"], "reason": "검정 5%+", "black_pct": r["black_pct"]})
|
||
|
|
else:
|
||
|
|
passed.append({"id": r["id"], "pastel_pct": r.get("pastel_pct"), "size": r.get("size_bytes")})
|
||
|
|
best = sorted([r for r in results if r.get("status") == "OK"],
|
||
|
|
key=lambda x: x.get("pastel_pct", 0), reverse=True)
|
||
|
|
return {
|
||
|
|
"passed_count": len(passed),
|
||
|
|
"failed_count": len(failed),
|
||
|
|
"best_pastel": best[0]["id"] if best else None,
|
||
|
|
"next_round_hint": "auto-determine based on round learning"
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
print("round_pipeline.py — 라이브러리, import 후 사용")
|