# THESIS: Turn five model tasks into one character-to-animation workbench; # refuse the generic prompt form as the product's primary structure. # OWN-WORLD: White proof sheets, near-black ink rails, cobalt registration # marks, vermilion actions, square frame cells, and indexed palette strips. # STORY: Choose a real LPC base, dress it, derive directions and walks, correct # pixels and palette, then export; every result becomes the next stage's input. # FIRST VIEWPORT: Source rail left, dominant active proof center, next action # right, with the palette and proof log below. # FORM: Screenprint registration workbench; staged per tab from approved comps # A+B+C. Direction seed e120291a. import argparse import copy import json import os import secrets import shutil import tempfile import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from pathlib import Path from urllib.parse import urlsplit, urlunsplit import gradio as gr import requests from dotenv import load_dotenv from PIL import Image, ImageDraw from postprocess import ( apply_shared_palette, process_output, reference_palette, save_gif, shared_palette, ) ROOT = Path(__file__).parent WORKFLOW = ROOT / "workflows" / "LPC_FourDirection_Walk_API.json" BASE_DIR = ROOT / "assets" / "bases" OUTPUT_DIR = ROOT / "outputs" BASE_STANDING = BASE_DIR / "standing_south.png" BASE_WALKS = { "north": BASE_DIR / "walk_north_4x2.png", "west": BASE_DIR / "walk_west_4x2.png", "south": BASE_DIR / "walk_south_4x2.png", "east": BASE_DIR / "walk_east_4x2.png", } DIRECTIONS = ("north", "west", "south", "east") MAX_PARALLEL_REQUESTS = 4 DIRECTION_LABELS = { "north": "북쪽", "west": "서쪽", "south": "남쪽", "east": "동쪽", } def load_environment(): for directory in (ROOT, *ROOT.parents): candidate = directory / ".env" if candidate.exists(): load_dotenv(candidate, override=False) return candidate return None ENV_PATH = load_environment() TASKS = { "Rotate standing character": ( "TASK_ROTATE_STANDING: Turn the south-facing standing female LPC " "character to face {direction}. Preserve the exact hairstyle, hair " "color, clothing, shoes and accessories. Keep the character centered " "on a pure white background." ), "Standing to walk frame 1": ( "TASK_STANDING_TO_WALK_FIRST_FRAME: Convert this {direction}-facing " "standing LPC character into the first frame of the {direction}-facing " "walking animation. Preserve the exact appearance and pure white background." ), "Propagate frame 1 appearance": ( "TASK_PROPAGATE_APPEARANCE: Use frame 1 as the appearance reference. " "Apply exactly the same hairstyle, hair color, clothing, shoes and " "accessories to frames 2 through 8. Preserve every walking pose, frame " "order, 4 by 2 layout and pure white background." ), "Dress 4x2 walk sheet": ( "TASK_DRESS_WALK_SHEET: Dress the female LPC character in all 8 " "{direction}-facing walking frames with {appearance}. Preserve every " "pose, frame order, 4 by 2 layout and pure white background." ), "Dress standing character": ( "TASK_DRESS_STANDING: Dress the south-facing standing female LPC " "character with {appearance}. Preserve the pose and pure white background." ), } TASK_LABELS = { "Rotate standing character": "Standing 방향 바꾸기", "Standing to walk frame 1": "걷기 첫 프레임 만들기", "Propagate frame 1 appearance": "첫 프레임 외형 전파", "Dress 4x2 walk sheet": "4×2 걷기 시트 단장", "Dress standing character": "Standing 캐릭터 단장", } PALETTE_CHOICES = [ ("기준 팔레트 고정", "Lock reference palette"), ("새 색상 허용 · 최종 32색", "Allow new colors (32)"), ] RESOLUTION_CHOICES = [ ("업스케일 유지", "Upscaled"), ("LPC 원본 크기", "Native LPC"), ] FORMAT_CHOICES = [ ("PNG 시트", "PNG sheet"), ("GIF 애니메이션", "GIF"), ] def initial_state(): return { "active_standing": "", "active_palette": "", "directions": {}, "walks": {}, "history": [], } def state_copy(state): return copy.deepcopy(state) if state else initial_state() def make_prompt(task, direction, appearance): if task not in TASKS: raise ValueError(f"알 수 없는 작업입니다: {task}") if task.startswith("Dress") and not appearance.strip(): raise ValueError("머리, 옷, 신발, 장식을 설명해 주세요.") return TASKS[task].format( direction=direction.lower(), appearance=appearance.strip() ) def normalize_server(value): raw = (value or "").strip() if not raw: raise ValueError("ComfyUI 서버 주소를 입력해 주세요.") if raw.count("://") != 1: raise ValueError( "서버 주소가 중복되었거나 형식이 잘못되었습니다. " "예: https://cloud.comfy.org" ) parsed = urlsplit(raw) if parsed.scheme not in {"http", "https"} or not parsed.hostname: raise ValueError("서버 주소는 http:// 또는 https://로 시작해야 합니다.") clean = parsed._replace(query="", fragment="") return urlunsplit(clean).rstrip("/") def resolve_api_key(value): return ( (value or "").strip() or os.getenv("COMFY_API_KEY", "").strip() or os.getenv("COMFY_CLOUD_API_KEY", "").strip() ) def is_cloud(server): return urlsplit(server).hostname == "cloud.comfy.org" def headers(api_key): return {"X-API-Key": api_key} if api_key else {} def request_error(error): if isinstance(error, requests.exceptions.ConnectionError): return ( "서버에 연결하지 못했습니다. 주소가 중복 입력되지 않았는지와 " "인터넷 연결을 확인해 주세요." ) if isinstance(error, requests.exceptions.Timeout): return "서버 응답 시간이 초과되었습니다. 잠시 후 다시 시도해 주세요." if isinstance(error, requests.exceptions.HTTPError): status = error.response.status_code if error.response is not None else "알 수 없음" if status in {401, 403}: return "API 키가 거부되었습니다. Comfy Cloud 키를 다시 확인해 주세요." return f"ComfyUI가 HTTP {status} 오류를 반환했습니다." return str(error) def upload_image(server, api_key, image_path): with open(image_path, "rb") as image: response = requests.post( f"{server}/api/upload/image", headers=headers(api_key), files={"image": (Path(image_path).name, image, "image/png")}, data={"type": "input", "overwrite": "true"}, timeout=120, ) response.raise_for_status() return response.json()["name"] def submit(server, api_key, workflow): response = requests.post( f"{server}/api/prompt", headers=headers(api_key), json={"prompt": workflow}, timeout=120, ) response.raise_for_status() data = response.json() return data.get("prompt_id") or data["job_id"] def first_image(value): if isinstance(value, dict): if {"filename", "type"} <= value.keys(): return value for child in value.values(): found = first_image(child) if found: return found elif isinstance(value, list): for child in value: found = first_image(child) if found: return found return None def wait_for_result(server, api_key, job_id, cloud): deadline = time.time() + 900 while time.time() < deadline: if cloud: response = requests.get( f"{server}/api/jobs/{job_id}", headers=headers(api_key), timeout=60, ) response.raise_for_status() data = response.json() status = data.get("status") if status == "completed": return data if status in {"failed", "cancelled"}: raise RuntimeError(data.get("error") or f"작업이 {status} 상태입니다.") else: response = requests.get( f"{server}/api/history/{job_id}", headers=headers(api_key), timeout=60, ) response.raise_for_status() data = response.json() if job_id in data: return data[job_id] time.sleep(2) raise TimeoutError("15분 안에 생성이 완료되지 않았습니다.") def download_image(server, api_key, output): response = requests.get( f"{server}/api/view", headers=headers(api_key), params={ "filename": output["filename"], "subfolder": output.get("subfolder", ""), "type": output.get("type", "output"), }, timeout=120, ) response.raise_for_status() with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as handle: handle.write(response.content) return handle.name def persist(path, prefix): OUTPUT_DIR.mkdir(exist_ok=True) source = Path(path) stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") target = OUTPUT_DIR / f"{prefix}_{stamp}{source.suffix.lower()}" shutil.copy2(source, target) return str(target) def add_history(state, path, label): state["history"].insert(0, {"path": path, "label": label}) state["history"] = state["history"][:24] def history_value(state): return [ (entry["path"], entry["label"]) for entry in (state or {}).get("history", []) if Path(entry["path"]).exists() ] def direction_values(state): directions = (state or {}).get("directions", {}) return tuple(directions.get(direction) for direction in DIRECTIONS) def walk_value(state): walks = (state or {}).get("walks", {}) return [ (walks[direction], f"{DIRECTION_LABELS[direction]} 걷기") for direction in DIRECTIONS if direction in walks and Path(walks[direction]).exists() ] def palette_swatch(colors, prefix="palette"): while len(colors) < 32: colors.append(colors[-1]) swatch = 34 preview = Image.new("RGB", (swatch * 16, swatch * 2), "white") draw = ImageDraw.Draw(preview) for index, color in enumerate(colors[:32]): x = (index % 16) * swatch y = (index // 16) * swatch draw.rectangle((x, y, x + swatch - 2, y + swatch - 2), fill=color) with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as handle: preview.save(handle.name) return persist(handle.name, prefix) def palette_preview(image_path, prefix="palette"): image = Image.open(image_path).convert("RGB").resize( (64, 64), Image.Resampling.NEAREST ) return palette_swatch(reference_palette(image), prefix) def finalize_shared_palette(paths, prefix): colors = shared_palette(paths.values()) outputs = apply_shared_palette(paths.values(), colors) finalized = { direction: persist(path, f"{prefix}_{direction}") for direction, path in zip(paths, outputs) } return finalized, palette_swatch(colors, f"{prefix}_palette") def animation_preview(sheet_path, fps=8): sheet = Image.open(sheet_path).convert("RGB") return persist( save_gif(sheet, reference_palette(sheet), fps), "walk_preview", ) def generate( server_value, api_key_value, image_path, task, direction, appearance, seed, steps, cfg, lora_strength, palette_reference_path=None, palette_mode="Lock reference palette", align_frames=True, pixel_snap=True, output_resolution="Upscaled", output_format="PNG sheet", fps=8, prefix="result", ): if not image_path: raise ValueError("입력 이미지를 선택해 주세요.") server = normalize_server(server_value) api_key = resolve_api_key(api_key_value) cloud = is_cloud(server) if cloud and not api_key: raise ValueError( "Comfy Cloud API 키가 필요합니다. 설정 탭 또는 .env에 " "COMFY_API_KEY를 입력해 주세요." ) workflow = json.loads(WORKFLOW.read_text(encoding="utf-8")) workflow["1"]["inputs"]["image"] = upload_image(server, api_key, image_path) workflow["4"]["inputs"]["strength_model"] = float(lora_strength) workflow["9"]["inputs"]["text"] = make_prompt(task, direction, appearance) workflow["14"]["inputs"]["noise_seed"] = int(seed) workflow["15"]["inputs"]["steps"] = int(steps) workflow["17"]["inputs"]["cfg"] = float(cfg) workflow["2"]["inputs"]["megapixels"] = ( 2.0 if task in {"Dress 4x2 walk sheet", "Propagate frame 1 appearance"} else 1.0 ) job_id = submit(server, api_key, workflow) result = wait_for_result(server, api_key, job_id, cloud) output = first_image(result.get("outputs", result)) if not output: raise RuntimeError("ComfyUI 작업은 끝났지만 이미지 출력이 없습니다.") raw_path = download_image(server, api_key, output) final_path = process_output( raw_path, image_path, task, palette_reference_path, palette_mode, align_frames, pixel_snap, output_resolution, output_format, fps, ) return persist(final_path, prefix), job_id def compose_first_frame(first_frame_path, direction): base = Image.open(BASE_WALKS[direction]).convert("RGB") first = Image.open(first_frame_path).convert("RGB").resize( (base.width // 4, base.height // 2), Image.Resampling.NEAREST ) base.paste(first, (0, 0)) with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as handle: base.save(handle.name) return handle.name def dress_core( source, appearance, server, api_key, seed, steps, cfg, strength, state, ): final, job_id = generate( server, api_key, source, "Dress standing character", "south", appearance, seed, steps, cfg, strength, palette_mode="Allow new colors (32)", align_frames=False, prefix="standing_dressed", ) state = state_copy(state) palette = palette_preview(final) state["active_standing"] = final state["active_palette"] = palette state["directions"] = {"south": final} state["walks"] = {} add_history(state, final, "Standing 외형 완료") return state, final, palette, job_id def direction_job( direction, index, source, server, api_key, seed, steps, cfg, strength, ): if direction == "south": return source, "기존 Standing 사용" return generate( server, api_key, source, "Rotate standing character", direction, "", int(seed) + index, steps, cfg, strength, palette_mode="Defer shared palette", prefix=f"standing_{direction}_pending", ) def direction_results( selected, source, server, api_key, seed, steps, cfg, strength, state, progress=None, ): state = state if state is not None else initial_state() source = source or state.get("active_standing") if not source: raise ValueError("먼저 Standing 캐릭터를 완성해 주세요.") selected = list(dict.fromkeys(selected or [])) if not selected: raise ValueError("생성할 방향을 하나 이상 선택해 주세요.") for direction in selected: state["directions"].pop(direction, None) total = len(selected) completed = 0 failures = [] workers = min(MAX_PARALLEL_REQUESTS, total) with ThreadPoolExecutor(max_workers=workers) as executor: futures = { executor.submit( direction_job, direction, index, source, server, api_key, seed, steps, cfg, strength, ): direction for index, direction in enumerate(selected) } for future in as_completed(futures): direction = futures[future] try: final, job_id = future.result() except Exception as error: failures.append(f"{DIRECTION_LABELS[direction]}: {request_error(error)}") continue state["directions"][direction] = final completed += 1 add_history( state, final, f"{DIRECTION_LABELS[direction]} Standing · 팔레트 대기 · {job_id}", ) if progress: progress( (completed, total), desc=f"{DIRECTION_LABELS[direction]} 완료 · {completed}/{total}", ) yield state, f"{DIRECTION_LABELS[direction]} 완료 · {completed}/{total}" if failures: raise RuntimeError(" / ".join(failures)) finalized, palette = finalize_shared_palette( {direction: state["directions"][direction] for direction in selected}, "standing_shared", ) state["directions"].update(finalized) state["active_palette"] = palette if "south" in finalized: state["active_standing"] = finalized["south"] for direction, path in finalized.items(): add_history(state, path, f"{DIRECTION_LABELS[direction]} Standing · 공통 팔레트") if progress: progress(1, desc="전체 방향 공통 32색 팔레트 적용") yield state, "전체 방향 완료 · 공통 32색 팔레트를 적용했습니다." def walk_job( direction, index, standing, server, api_key, seed, steps, cfg, strength, ): first, _ = generate( server, api_key, standing, "Standing to walk frame 1", direction, "", int(seed) + index * 2, steps, cfg, strength, palette_mode="Defer shared palette", align_frames=False, prefix=f"walk_first_{direction}_pending", ) composite = compose_first_frame(first, direction) final, job_id = generate( server, api_key, composite, "Propagate frame 1 appearance", direction, "", int(seed) + index * 2 + 1, steps, cfg, strength, palette_mode="Defer shared palette", align_frames=True, prefix=f"walk_{direction}_pending", ) return final, job_id def walk_results( selected, server, api_key, seed, steps, cfg, strength, state, progress=None, ): state = state if state is not None else initial_state() selected = list(dict.fromkeys(selected or [])) if not selected: raise ValueError("생성할 방향을 하나 이상 선택해 주세요.") missing = [ direction for direction in selected if direction not in state.get("directions", {}) ] if missing: labels = ", ".join(DIRECTION_LABELS[direction] for direction in missing) raise ValueError(f"먼저 다음 방향의 Standing을 만들어 주세요: {labels}") for direction in selected: state["walks"].pop(direction, None) total = len(selected) completed = 0 failures = [] workers = min(MAX_PARALLEL_REQUESTS, total) with ThreadPoolExecutor(max_workers=workers) as executor: futures = { executor.submit( walk_job, direction, index, state["directions"][direction], server, api_key, seed, steps, cfg, strength, ): direction for index, direction in enumerate(selected) } for future in as_completed(futures): direction = futures[future] try: final, job_id = future.result() except Exception as error: failures.append(f"{DIRECTION_LABELS[direction]}: {request_error(error)}") continue state["walks"][direction] = final completed += 1 add_history( state, final, f"{DIRECTION_LABELS[direction]} 걷기 · 팔레트 대기 · {job_id}", ) if progress: progress( (completed, total), desc=f"{DIRECTION_LABELS[direction]} 걷기 완료 · {completed}/{total}", ) yield state, f"{DIRECTION_LABELS[direction]} 걷기 완료 · {completed}/{total}" if failures: raise RuntimeError(" / ".join(failures)) finalized, palette = finalize_shared_palette( {direction: state["walks"][direction] for direction in selected}, "walk_shared", ) state["walks"].update(finalized) state["active_palette"] = palette for direction, path in finalized.items(): add_history(state, path, f"{DIRECTION_LABELS[direction]} 걷기 · 공통 팔레트") if progress: progress(1, desc="전체 방향 공통 32색 팔레트 적용") yield state, "전체 걷기 완료 · 공통 32색 팔레트를 적용했습니다." def format_failure(error): return f"오류 · {request_error(error)}" def ui_dress( source, appearance, server, api_key, seed, steps, cfg, strength, state, ): try: state, final, palette, job_id = dress_core( source, appearance, server, api_key, seed, steps, cfg, strength, state, ) return ( final, palette, final, f"완료 · Standing 외형 생성 · 시드 {int(seed)} · {job_id}", state, history_value(state), ) except Exception as error: return None, None, None, format_failure(error), state, history_value(state) def ui_directions( selected, source, server, api_key, seed, steps, cfg, strength, state, progress=gr.Progress(), ): try: for state, message in direction_results( selected, source, server, api_key, seed, steps, cfg, strength, state, progress, ): yield ( *direction_values(state), f"{message} · 시드 {int(seed)}", state, history_value(state), ) except Exception as error: yield ( *direction_values(state), format_failure(error), state, history_value(state), ) def ui_walks( selected, server, api_key, seed, steps, cfg, strength, state, progress=gr.Progress(), ): try: for state, message in walk_results( selected, server, api_key, seed, steps, cfg, strength, state, progress, ): available = [ state["walks"][direction] for direction in selected or [] if direction in state["walks"] ] latest = available[-1] if available else None animation = animation_preview(latest) if latest else None yield ( walk_value(state), latest, animation, latest, f"{message} · 시드 {int(seed)}", state, history_value(state), ) except Exception as error: yield ( walk_value(state), None, None, None, format_failure(error), state, history_value(state), ) def estimate_jobs(selected): selected = selected or [] rotations = sum(direction != "south" for direction in selected) total = 1 + rotations + len(selected) * 2 return ( f"전체 실행 예상: {total}개 작업 " f"(외형 1 + 방향 {rotations} + 걷기 {len(selected) * 2}) · " "방향별 최대 4개 병렬" ) def new_seed(): return secrets.randbelow(2_147_483_647) + 1 def ui_run_all( source, appearance, selected, server, api_key, seed, steps, cfg, strength, state, progress=gr.Progress(), ): try: if not selected: raise ValueError("생성할 방향을 하나 이상 선택해 주세요.") total = 1 + sum(direction != "south" for direction in selected) + len(selected) * 2 progress((0, total), desc="Standing 외형 생성") state, active, palette, _ = dress_core( source, appearance, server, api_key, seed, steps, cfg, strength, state, ) yield ( active, palette, *direction_values(state), walk_value(state), None, None, active, "Standing 완료 · 방향 작업을 병렬로 시작합니다.", state, history_value(state), ) for state, message in direction_results( selected, active, server, api_key, int(seed) + 100, steps, cfg, strength, state, progress, ): yield ( active, state.get("active_palette") or palette, *direction_values(state), walk_value(state), None, None, active, message, state, history_value(state), ) for state, message in walk_results( selected, server, api_key, int(seed) + 200, steps, cfg, strength, state, progress, ): available = [ state["walks"][direction] for direction in selected if direction in state["walks"] ] latest = available[-1] if available else None yield ( active, state.get("active_palette") or palette, *direction_values(state), walk_value(state), latest, animation_preview(latest) if latest else None, latest, f"{message} · 전체 {total}개 작업 · 시드 {int(seed)}", state, history_value(state), ) except Exception as error: yield ( (state or {}).get("active_standing") or None, None, *direction_values(state), walk_value(state), None, None, None, format_failure(error), state, history_value(state), ) def use_active(state): path = (state or {}).get("active_standing") if not path: return None, "먼저 Standing 캐릭터를 완성해 주세요." return path, "활성 캐릭터를 입력으로 가져왔습니다." def walk_base(direction): return str(BASE_WALKS[direction]) def ui_single_walk( direction, source, server, api_key, seed, steps, cfg, strength, state, progress=gr.Progress(), ): state = state_copy(state) if source: state["directions"][direction] = source if direction == "south" and not state.get("active_standing"): state["active_standing"] = source yield from ui_walks( [direction], server, api_key, seed, steps, cfg, strength, state, progress, ) def local_process( source, palette_source, palette_mode, align, resolution, output_format, fps, set_active, state, ): try: if not source: raise ValueError("처리할 이미지를 선택해 주세요.") with Image.open(source) as image: sheet = image.width >= image.height * 1.5 task = "Dress 4x2 walk sheet" if sheet else "Dress standing character" final = process_output( source, source, task, palette_source, palette_mode, align and sheet, True, resolution, output_format, fps, ) final = persist(final, "pixel_palette") palette = palette_preview(palette_source or final, "palette") state = state_copy(state) state["active_palette"] = palette if set_active and not sheet and output_format == "PNG sheet": state["active_standing"] = final state["directions"]["south"] = final add_history(state, final, "픽셀·팔레트 처리") return ( final, final, palette, palette, "완료 · Perfect Pixel 격자와 32색 팔레트를 적용했습니다.", state, history_value(state), state.get("active_standing") or None, ) except Exception as error: return ( None, None, None, None, format_failure(error), state, history_value(state), (state or {}).get("active_standing") or None, ) def check_connection(server_value, api_key_value): try: server = normalize_server(server_value) api_key = resolve_api_key(api_key_value) if is_cloud(server) and not api_key: raise ValueError("Comfy Cloud API 키가 설정되지 않았습니다.") response = requests.get( f"{server}/api/prompt", headers=headers(api_key), timeout=15, ) response.raise_for_status() return ( f"연결 확인 · {urlsplit(server).hostname} 응답 " f"(HTTP {response.status_code})" ) except Exception as error: return format_failure(error) def prompt_preview(task, direction, appearance): try: return make_prompt(task, direction, appearance) except Exception as error: return str(error) def ui_advanced( source, task, direction, appearance, server, api_key, seed, steps, cfg, strength, palette_source, palette_mode, align, resolution, output_format, fps, state, ): try: final, job_id = generate( server, api_key, source, task, direction, appearance, seed, steps, cfg, strength, palette_source, palette_mode, align, True, resolution, output_format, fps, prefix="advanced", ) state = state_copy(state) add_history(state, final, f"고급 단일 작업 · {TASK_LABELS[task]}") return ( final, final, f"완료 · {TASK_LABELS[task]} · {job_id}", state, history_value(state), ) except Exception as error: return None, None, format_failure(error), state, history_value(state) def render_stage_strip(state): state = state or initial_state() direction_count = len(state.get("directions", {})) walk_count = len(state.get("walks", {})) if not state.get("active_standing"): current = 0 elif direction_count < 4: current = 2 elif walk_count < 4: current = 3 else: current = 4 stages = [ ("베이스", "실제 LPC 원본 선택"), ("외형", "머리·복장·장식"), ("방향", f"{direction_count}/4 완료"), ("걷기", f"{walk_count}/4 완료"), ("내보내기", "PNG·GIF"), ] items = "".join( f'
베이스 캐릭터에서 외형, 방향, 8프레임 걷기, 32색 내보내기까지 한 흐름으로 만듭니다.