#!/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 후 사용")