cbpark84
1c919f1106
- run_batch(): sequential|concurrent|auto - chain(): 파이프라인 체이닝 - schema(): 스키마 자동 추출 - ResultCache: LRU+TTL 캐시 - preview_callback: 실시간 프리뷰 - upload_file/audio/video() - BridgeInputSEED: random/increment/fixed - MCPServer: MCP Tool 자동 생성
79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""워크플로우 JSON 파서 및 BridgeInput 값 패처."""
|
|
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Tuple, Union
|
|
|
|
from .types import convert_value
|
|
|
|
BRIDGE_INPUT_PREFIX = "BridgeInput"
|
|
|
|
|
|
def load_workflow(source: Union[str, dict, Path]) -> dict:
|
|
"""워크플로우를 파일 경로 또는 dict로 로드한다."""
|
|
if isinstance(source, dict):
|
|
return source
|
|
path = Path(source)
|
|
with path.open("r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def find_bridge_inputs(workflow: dict) -> Dict[str, Tuple[str, str]]:
|
|
"""
|
|
워크플로우에서 BridgeInput 노드를 탐색한다.
|
|
|
|
Returns:
|
|
{ "input_name": (node_id, type_name) }
|
|
예: { "prompt": ("1", "STRING"), "steps": ("3", "INT"), "image": ("5", "ANY") }
|
|
"""
|
|
result: Dict[str, Tuple[str, str]] = {}
|
|
for node_id, node in workflow.items():
|
|
if not isinstance(node, dict):
|
|
continue
|
|
class_type = node.get("class_type", "")
|
|
if class_type.startswith(BRIDGE_INPUT_PREFIX):
|
|
name = node.get("inputs", {}).get("name")
|
|
type_name = class_type[len(BRIDGE_INPUT_PREFIX):] # "STRING", "INT", "ANY" 등
|
|
if name:
|
|
result[name] = (node_id, type_name)
|
|
return result
|
|
|
|
|
|
def patch_inputs(workflow: dict, inputs: Dict[str, Any]) -> dict:
|
|
"""
|
|
원본 워크플로우를 deepcopy 후 BridgeInput 노드의 value를 패치한다.
|
|
|
|
스칼라 타입 (STRING/INT/FLOAT/BOOLEAN):
|
|
- inputs.value (위젯) 와 inputs.value_link (연결 포트) 모두 패치
|
|
- value_link가 연결된 경우 런타임에 우선되므로 위젯 값도 동기화
|
|
복합 타입 / ANY:
|
|
- inputs.value 패치
|
|
"""
|
|
patched = copy.deepcopy(workflow)
|
|
bridge_map = find_bridge_inputs(patched)
|
|
|
|
PRIMITIVE = {"STRING", "INT", "FLOAT", "BOOLEAN"}
|
|
|
|
for name, value in inputs.items():
|
|
if name not in bridge_map:
|
|
available = list(bridge_map.keys())
|
|
raise ValueError(
|
|
f"BridgeInput '{name}' not found in workflow. "
|
|
f"Available inputs: {available}"
|
|
)
|
|
node_id, type_name = bridge_map[name]
|
|
converted = convert_value(value, type_name)
|
|
node_inputs = patched[node_id]["inputs"]
|
|
|
|
if type_name in PRIMITIVE:
|
|
# 위젯 값 패치
|
|
node_inputs["value"] = converted
|
|
# value_link가 이미 노드 링크([node_id, idx])로 연결된 경우 건드리지 않음
|
|
# value_link가 없거나 null이면 value(위젯)가 자동으로 사용됨
|
|
else:
|
|
# 복합 타입 / ANY
|
|
node_inputs["value"] = converted
|
|
|
|
return patched
|