Spaces:
Running on Zero
Running on Zero
File size: 9,011 Bytes
ca9a89c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """高层 API 图片参数兼容层；顺序与能力校验复用 UI 的执行计划。"""
import re
from PIL import Image
from core.execution_plan import (
MODE_MODEL_PK, MODE_MULTI_MODEL_GRID, MODE_MULTI_REFERENCE, MODE_SINGLE,
available_run_modes, build_execution_plan,
)
from core.reference_inputs import ImageInputError, REFERENCE_CHAIN_SPECS, SOURCE_IMAGE_KEYS, reference_choices
from core.runtime_config import CONFIG
NATIVE_CHAIN_ALIASES = {
"reference_edit": "reference_latent",
"mage_flow_reference_edit": "reference_image",
"joyai_reference_edit": "joyai_image",
"boogu_edit": "boogu_image_edit",
}
_IMAGE_KEY = re.compile(r"(?:图|img|image)([0-9０-９]+)", re.IGNORECASE)
def normalize_image_request(params: dict) -> None:
"""只修改 handle_run 的副本。显式混用入口时拒绝，不猜优先级。"""
numbered = {}
for key in list(params):
match = _IMAGE_KEY.fullmatch(key) if isinstance(key, str) else None
if not match:
continue
index = int(match.group(1))
if index in numbered:
raise ImageInputError(f"图片编号 {index} 重复；图{index} / img{index} / image{index} 只能提供一个字段。")
numbered[index] = params.pop(key)
if numbered:
if "images" in params or params.get("image") is not None:
raise ImageInputError("编号字段、images 列表和 image 源图不能混用，请只保留一个入口。")
if sorted(numbered) != list(range(1, len(numbered) + 1)):
raise ImageInputError("图片编号必须从 1 开始且连续，不能缺少 img1 或跳号。")
params["images"] = [numbered[i] for i in range(1, len(numbered) + 1)]
if "images" in params:
images = params["images"]
if not isinstance(images, list) or not images:
raise ImageInputError("images 必须是非空、有序图片列表。")
if any(not isinstance(image, (str, Image.Image)) or (isinstance(image, str) and not image.strip()) for image in images):
raise ImageInputError("images 不允许空图片；删除空项后重新按顺序编号。")
if params.get("image") is not None:
raise ImageInputError("image 与 images 不能混用。")
if params["task_type"] == "reference":
params.setdefault("run_mode", MODE_MULTI_REFERENCE)
if params.get("image") is not None:
if not isinstance(params["image"], (str, Image.Image)):
raise ImageInputError("image 必须是图片 URL 或 Base64 Data URI。")
params["images"] = [params.pop("image")]
elif params["task_type"] == "txt2img" and params.get("image") is not None:
raise ImageInputError("txt2img 不使用 image 源图；参考图编辑请用 task_type=reference 和 images。")
if params.get("image") is not None and not isinstance(params["image"], (str, Image.Image)):
raise ImageInputError("image 必须是图片 URL 或 Base64 Data URI。")
for field in ("run_mode", "reference_role"):
if field in params and not isinstance(params[field], str):
raise ImageInputError(f"{field} 必须是字符串。")
if "extra_models" in params and (
not isinstance(params["extra_models"], list)
or not all(isinstance(model, str) for model in params["extra_models"])
):
raise ImageInputError("extra_models 必须是模型名称列表。")
if "use_model_defaults" in params and not isinstance(params["use_model_defaults"], bool):
raise ImageInputError("use_model_defaults 必须是 boolean。")
def validate_image_plan(params: dict) -> None:
"""不下载图片、不分配 GPU，提交队列前校验完整调用语义。"""
values = {
"task_type": params["task_type"], "model_display_name": params["model"],
"positive_prompt": params["prompt"], "negative_prompt": params.get("negative_prompt") or "",
"seed": params.get("seed", -1), "batch_size": params.get("batch_size", 1),
}
task = params["task_type"]
if task in SOURCE_IMAGE_KEYS and params.get("image") is not None:
values[SOURCE_IMAGE_KEYS[task]] = params["image"]
for item in params.get("chain", []):
chain = NATIVE_CHAIN_ALIASES.get(item["injector_type"], item["injector_type"])
if chain in REFERENCE_CHAIN_SPECS:
if params.get("images"):
raise ImageInputError("images 与原生参考 chain 不能混用；自动路由用 images，旧调用只用 chain。")
if item.get("image") is None:
raise ImageInputError(f"{chain} 缺少 image。")
if not isinstance(item["image"], (str, Image.Image)) or (isinstance(item["image"], str) and not item["image"].strip()):
raise ImageInputError(f"{chain}.image 必须是图片 URL 或 Base64 Data URI。")
values.setdefault(REFERENCE_CHAIN_SPECS[chain][0], []).append(item["image"])
mode = params.get("run_mode", MODE_SINGLE)
if mode not in {MODE_MODEL_PK, MODE_MULTI_MODEL_GRID} and params.get("extra_models"):
raise ImageInputError("extra_models 只用于 model_pk / multi_model_grid。")
if mode != MODE_MULTI_REFERENCE and params.get("reference_role", "auto") != "auto":
raise ImageInputError("reference_role 只用于 reference / multi_reference。")
if mode in {MODE_MODEL_PK, MODE_MULTI_MODEL_GRID} and (
params.get("chain") or params.get("vae_id") or params.get("pid") or params.get("pid_settings")
):
raise ImageInputError("模型对比仅使用基础模型，不支持 chain / VAE / PiD 覆盖。")
build_execution_plan(
values, mode=mode, images=params.get("images"),
extra_models=params.get("extra_models"), reference_role=params.get("reference_role", "auto"),
use_model_defaults=params.get("use_model_defaults", True),
)
def image_workflow_contract(model: str) -> dict:
"""在现有 discovery 上增量公开高层图片契约，不要求 AI 猜 injector。"""
roles = reference_choices(model)
return {
"reference_supported": bool(roles),
"reference_roles": {
role: {"chain": spec["chain"], "max_images": spec["max_images"]}
for role, spec in roles.items()
},
"run_modes_by_task": {
task: [mode for _, mode in available_run_modes(task, model)]
for task in ["txt2img", *SOURCE_IMAGE_KEYS, "reference"]
},
"parameters_schema": {
"type": "object",
"properties": {
"images": {"type": "array", "minItems": 1, "maxItems": CONFIG.max_multi_images,
"items": {"type": "string", "description": "Public HTTP(S) image URL or Base64 Data URI; not a local path."}},
"run_mode": {"type": "string", "enum": ["single", "model_pk", "multi_independent", "multi_model_grid", "multi_reference"],
"description": "Default single; task_type=reference defaults to multi_reference. Use run_modes_by_task."},
"reference_role": {"type": "string", "enum": list(roles) or ["auto"], "default": "auto"},
"extra_models": {"type": "array", "items": {"type": "string"}, "description": "Only for model_pk / multi_model_grid; current model is included automatically."},
"use_model_defaults": {"type": "boolean", "default": True},
},
"patternProperties": {r"^(图|img|image)[1-9][0-9]*$": {"type": "string", "description": "Alternative to images; indices must start at 1, be contiguous and unique."}},
},
"rules": [
"Use task_type=reference with images, prompt, width, height for instruction editing / fusion; no native chain needed. Model must support reference.",
"images order is 1-based: 图1 = img1 = image1. Numbered prompt references become image 1; quoted/backticked text is literal. No auto translation or prompt expansion.",
"Do not mix images, numbered fields, image, or native reference chain. Legacy txt2img + chain and single-image image calls remain supported.",
"multi_independent / multi_model_grid use one shared prompt per image. Say 当前图片 / the current image; cross-image numbered references are rejected.",
"Leave steps/cfg/sampler/scheduler omitted to use model defaults. Negative prompts depend on model/CFG and may be unused at CFG=1.",
"Prefer async_execution=true; poll the same task_id with get_task_status until completed/failed; do not resubmit on a polling timeout.",
"Read result.image_references, effective_prompt, captions and jobs. Existing result.images remains a flat URL list; partial_success signals mixed results.",
],
"limits": {"max_images": CONFIG.max_multi_images, "max_pk_models": CONFIG.max_pk_models,
"max_jobs": CONFIG.max_plan_jobs, "max_outputs": CONFIG.max_plan_outputs},
}
|