nodes/Gin3601-LLS-node
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修改图像预览节点

Browse Source
This commit is contained in:
Gin3601
2026-05-22 15:25:59 +08:00
parent dd8b0533f4
commit 5bea2122d0
8 changed files with 340 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files
image/nodes.py
+31 -26
View File
@@ -18,8 +18,6 @@ from ..utils.model_info import (
info_to_json,
parse_jsonish_info,
resolve_model_family,
resolve_model_name,
resolve_text_encoder_names,
resolve_vae_name,
)
@@ -47,6 +45,13 @@ def _round_to_multiple(value: int, multiple: int) -> int:
return max(multiple, ((int(value) + multiple - 1) // multiple) * multiple)
def _first_non_empty(*values):
for value in values:
if value not in (None, ""):
return value
return ""
def _get_image_dimensions(image) -> tuple[int, int, int]:
shape = tuple(getattr(image, "shape", ()))
if len(shape) < 4:
@@ -356,9 +361,6 @@ class LLSSaveImage:
"save_metadata": ("BOOLEAN", {"default": True}),
},
"optional": {
"model": ("MODEL",),
"clip": ("CLIP",),
"vae": ("VAE",),
"prompt_info": ("STRING", {"forceInput": True}),
"latent_info": ("STRING", {"forceInput": True}),
"sample_info": ("STRING", {"forceInput": True}),
@@ -374,9 +376,6 @@ class LLSSaveImage:
def _build_metadata(
self,
image,
model=None,
clip=None,
vae=None,
prompt_info: str | None = None,
latent_info: str | None = None,
sample_info: str | None = None,
@@ -389,19 +388,31 @@ class LLSSaveImage:
decode = parse_jsonish_info(decode_info)
upscale = parse_jsonish_info(upscale_info)
family = prompt.get("model_family") or latent.get("model_family") or decode.get("model_family")
if not family:
family = resolve_model_family("Auto", model=model, clip=clip)
text_encoder_name = resolve_text_encoder_names(clip).get("text_encoder_name", "")
checkpoint_name = resolve_model_name(
model=model,
clip=clip,
fallback=prompt.get("checkpoint_name") or latent.get("checkpoint_name") or decode.get("checkpoint_name") or "",
family = _first_non_empty(
prompt.get("model_family"),
latent.get("model_family"),
decode.get("model_family"),
sample.get("model_family"),
sample.get("family"),
)
vae_name = resolve_vae_name(
vae,
fallback=decode.get("vae_name") or latent.get("vae_name") or "",
checkpoint_name = _first_non_empty(
prompt.get("checkpoint_name"),
latent.get("checkpoint_name"),
decode.get("checkpoint_name"),
sample.get("checkpoint_name"),
sample.get("model_name"),
)
text_encoder_name = _first_non_empty(
prompt.get("text_encoder_name"),
prompt.get("text_encoder_name_1"),
prompt.get("text_encoder_name_2"),
latent.get("text_encoder_name"),
decode.get("text_encoder_name"),
)
vae_name = _first_non_empty(
decode.get("vae_name"),
latent.get("vae_name"),
prompt.get("vae_name"),
)
metadata = {
@@ -433,9 +444,6 @@ class LLSSaveImage:
filename_prefix: str = "LLS",
output_mode: str = "save",
save_metadata: bool = True,
model=None,
clip=None,
vae=None,
prompt_info: str | None = None,
latent_info: str | None = None,
sample_info: str | None = None,
@@ -462,9 +470,6 @@ class LLSSaveImage:
if save_metadata:
merged_extra_pnginfo["lls_metadata"] = self._build_metadata(
image=image,
model=model,
clip=clip,
vae=vae,
prompt_info=prompt_info,
latent_info=latent_info,
sample_info=sample_info,
pro_edit/backends/registry.py
+67 -12
View File
@@ -87,6 +87,40 @@ def _resolve_fallback_backend_name(family: str) -> str:
return "generic"
def _build_routing_result(backend, *, routing_reason, capabilities, profile, execution_path):
return backend, RoutingResult(
backend.backend_name,
routing_reason,
capabilities,
profile,
execution_path,
)
def _resolve_profile_route(*, profile, capabilities, native_backend_name):
if native_backend_name is not None:
backend = get_backend(native_backend_name)
validate_manual_backend(native_backend_name, backend, profile)
return _build_routing_result(
backend,
routing_reason="profile.backend_type",
capabilities=capabilities,
profile=profile,
execution_path="native_edit",
)
fallback_backend_name = _resolve_fallback_backend_name(str(profile.get("family") or ""))
backend = get_backend(fallback_backend_name)
validate_manual_backend(fallback_backend_name, backend, profile)
return _build_routing_result(
backend,
routing_reason="profile.family_fallback",
capabilities=capabilities,
profile=profile,
execution_path="fallback_repair",
)
def resolve_backend(backend_mode: str, *, model=None, model_info=None, edit_info=None):
_ensure_builtin_backends()
mode = str(backend_mode or "auto").strip().lower() or "auto"
@@ -105,17 +139,38 @@ def resolve_backend(backend_mode: str, *, model=None, model_info=None, edit_info
if mode != "auto":
backend = get_backend(mode)
validate_manual_backend(mode, backend, profile)
routing_reason = "manual.override" if native_backend_name is not None else "manual.family_fallback"
execution_path = "native_edit" if native_backend_name is not None else "fallback_repair"
return backend, RoutingResult(backend.backend_name, routing_reason, capabilities, profile, execution_path)
if backend.supports(profile):
routing_reason = "manual.override" if native_backend_name is not None else "manual.family_fallback"
execution_path = "native_edit" if native_backend_name is not None else "fallback_repair"
validate_manual_backend(mode, backend, profile)
return _build_routing_result(
backend,
routing_reason=routing_reason,
capabilities=capabilities,
profile=profile,
execution_path=execution_path,
)
if native_backend_name is not None:
backend = get_backend(native_backend_name)
validate_manual_backend(native_backend_name, backend, profile)
return backend, RoutingResult(backend.backend_name, "profile.backend_type", capabilities, profile, "native_edit")
backend, routed = _resolve_profile_route(
profile=profile,
capabilities=capabilities,
native_backend_name=native_backend_name,
)
fallback_reason = (
"manual.incompatible_native_fallback"
if routed.execution_path == "native_edit"
else "manual.incompatible_family_fallback"
)
return _build_routing_result(
backend,
routing_reason=fallback_reason,
capabilities=capabilities,
profile=profile,
execution_path=routed.execution_path,
)
fallback_backend_name = _resolve_fallback_backend_name(str(profile.get("family") or ""))
backend = get_backend(fallback_backend_name)
validate_manual_backend(fallback_backend_name, backend, profile)
return backend, RoutingResult(backend.backend_name, "profile.family_fallback", capabilities, profile, "fallback_repair")
return _resolve_profile_route(
profile=profile,
capabilities=capabilities,
native_backend_name=native_backend_name,
)
pro_edit/pro_edit_bridge.py
+100 -133
View File
@@ -1,171 +1,138 @@
from __future__ import annotations
import random
import importlib
from .backends.registry import resolve_backend
from .pro_edit_utils import BACKEND_MODE_CHOICES, normalize_edit_info, set_conditioning_values
from ..sampling.nodes import (
_QUALITY_PRESETS,
_common_ksampler,
_get_samplers,
_get_schedulers,
_normalize_flux_guidance,
)
from ..utils.model_info import (
FAMILY_DEFAULT_PRESET,
MODEL_FAMILY_CHOICES,
get_family_defaults,
get_sampling_preset,
info_to_json,
)
from ..sampling.nodes import _common_ksampler, _get_samplers, _get_schedulers
try:
comfy_core_nodes = importlib.import_module("nodes")
except Exception:
comfy_core_nodes = None
def _apply_sampler_strategy(strategy: str, *, positive, negative, flux_guidance, defaults):
normalized = str(strategy or "").strip().lower() or "standard_k"
if normalized == "standard_k":
return positive, negative, None
if normalized == "flux_guided":
guidance_value = _normalize_flux_guidance(flux_guidance, defaults.get("default_guidance"))
positive = set_conditioning_values(positive, {"guidance": guidance_value})
negative = set_conditioning_values(negative, {"guidance": guidance_value})
return positive, negative, guidance_value
raise RuntimeError(f"[LLS] Unsupported sampler_strategy '{strategy}'.")
def _run_native_ksampler_advanced(
model,
add_noise,
noise_seed,
steps,
cfg,
sampler_name,
scheduler,
positive,
negative,
latent_image,
start_at_step,
end_at_step,
return_with_leftover_noise,
denoise,
):
if comfy_core_nodes is None or not hasattr(comfy_core_nodes, "KSamplerAdvanced"):
return None
sampler = comfy_core_nodes.KSamplerAdvanced()
result = sampler.sample(
model=model,
add_noise=add_noise,
noise_seed=int(noise_seed),
steps=int(steps),
cfg=float(cfg),
sampler_name=sampler_name,
scheduler=scheduler,
positive=positive,
negative=negative,
latent_image=latent_image,
start_at_step=int(start_at_step),
end_at_step=int(end_at_step),
return_with_leftover_noise=return_with_leftover_noise,
denoise=float(denoise),
)
if isinstance(result, tuple):
return result[0]
if isinstance(result, list):
return result[0]
return result
class LLSProKSamplerBridge:
CATEGORY = "LLS/Image Edit"
FUNCTION = "sample"
RETURN_TYPES = ("LATENT", "STRING")
RETURN_NAMES = ("latent", "sample_info")
RETURN_TYPES = ("LATENT",)
RETURN_NAMES = ("latent",)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": ("MODEL",),
"add_noise": (["enable", "disable"], {"advanced": True}),
"noise_seed": ("INT", {"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF, "control_after_generate": True}),
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
"cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0, "step": 0.1, "round": 0.01}),
"sampler_name": (_get_samplers(),),
"scheduler": (_get_schedulers(),),
"positive": ("CONDITIONING",),
"negative": ("CONDITIONING",),
"latent_image": ("LATENT",),
"backend_mode": (BACKEND_MODE_CHOICES, {"default": "auto"}),
"quality_preset": (_QUALITY_PRESETS, {"default": FAMILY_DEFAULT_PRESET}),
"seed": ("INT", {"default": -1, "min": -1, "max": 0xFFFFFFFFFFFFFFFF}),
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
"cfg": ("FLOAT", {"default": 7.0, "min": 0.0, "max": 100.0, "step": 0.1}),
"sampler_name": (_get_samplers(), {"default": "euler"}),
"scheduler": (_get_schedulers(), {"default": "normal"}),
"denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"denoise_mode": (["manual", "auto_from_edit"], {"default": "manual"}),
"flux_guidance": ("STRING,FLOAT,INT", {"default": 3.5, "widgetType": "FLOAT"}),
"model_family": (MODEL_FAMILY_CHOICES, {"default": "Auto"}),
},
"optional": {
"edit_info": ("LLS_EDIT_INFO",),
"model_info": ("STRING",),
},
"start_at_step": ("INT", {"default": 0, "min": 0, "max": 10000, "advanced": True}),
"end_at_step": ("INT", {"default": 10000, "min": 0, "max": 10000, "advanced": True}),
"return_with_leftover_noise": (["disable", "enable"], {"advanced": True}),
}
}
def sample(
self,
model,
positive,
negative,
latent_image,
backend_mode,
quality_preset,
seed,
add_noise,
noise_seed,
steps,
cfg,
sampler_name,
scheduler,
denoise,
denoise_mode,
flux_guidance,
model_family,
edit_info=None,
model_info=None,
positive,
negative,
latent_image,
start_at_step,
end_at_step,
return_with_leftover_noise,
denoise=1.0,
):
effective_model_info = model_info
normalized_model_family = str(model_family or "").strip()
if isinstance(model_info, dict):
effective_model_info = dict(model_info)
if normalized_model_family not in {"", "Auto", "auto"}:
effective_model_info.setdefault("family", normalized_model_family)
effective_model_info.setdefault("model_family", normalized_model_family)
elif model_info is None and normalized_model_family not in {"", "Auto", "auto"}:
effective_model_info = {"family": normalized_model_family}
_, routing = resolve_backend(
backend_mode,
result_latent = _run_native_ksampler_advanced(
model=model,
model_info=effective_model_info,
edit_info=edit_info,
)
profile = routing.profile
defaults = get_family_defaults(profile["family"])
if quality_preset == FAMILY_DEFAULT_PRESET:
steps = int(defaults["default_steps"])
cfg = float(defaults["default_cfg"])
sampler_name = str(defaults["default_sampler"])
scheduler = str(defaults["default_scheduler"])
denoise = float(defaults["default_denoise"])
else:
preset = get_sampling_preset({"family": profile["family"]}, quality_preset)
if preset is not None:
steps = int(preset["steps"])
cfg = float(preset["cfg"])
sampler_name = str(preset["sampler_name"])
scheduler = str(preset["scheduler"])
denoise = float(preset["denoise"])
normalized_edit_info = normalize_edit_info(edit_info)
actual_denoise = float(denoise)
if denoise_mode == "auto_from_edit":
actual_denoise = float(normalized_edit_info.get("recommended_denoise", actual_denoise))
positive, negative, guidance_value = _apply_sampler_strategy(
profile["sampler_strategy"],
positive=positive,
negative=negative,
flux_guidance=flux_guidance,
defaults=defaults,
)
actual_seed = random.randint(0, 0xFFFFFFFFFFFFFFFF) if int(seed) == -1 else int(seed)
result_latent = _common_ksampler(
model=model,
seed=actual_seed,
steps=int(steps),
cfg=float(cfg),
add_noise=add_noise,
noise_seed=noise_seed,
steps=steps,
cfg=cfg,
sampler_name=sampler_name,
scheduler=scheduler,
positive=positive,
negative=negative,
latent=latent_image,
denoise=actual_denoise,
latent_image=latent_image,
start_at_step=start_at_step,
end_at_step=end_at_step,
return_with_leftover_noise=return_with_leftover_noise,
denoise=denoise,
)
sample_info = info_to_json(
{
"backend_name": routing.backend_name,
"routing_reason": routing.routing_reason,
"execution_path": routing.execution_path,
"family": profile["family"],
"model_role": profile["role"],
"profile_id": profile["profile_id"],
"backend_type": profile["backend_type"],
"sampler_strategy": profile["sampler_strategy"],
"seed": actual_seed,
"steps": int(steps),
"cfg": float(cfg),
"guidance": guidance_value,
"sampler_name": sampler_name,
"scheduler": scheduler,
"denoise": actual_denoise,
"denoise_mode": denoise_mode,
"quality_preset": quality_preset,
}
)
return result_latent, sample_info
if result_latent is None:
force_full_denoise = str(return_with_leftover_noise or "disable") != "enable"
disable_noise = str(add_noise or "enable") == "disable"
result_latent = _common_ksampler(
model=model,
seed=int(noise_seed),
steps=int(steps),
cfg=float(cfg),
sampler_name=sampler_name,
scheduler=scheduler,
positive=positive,
negative=negative,
latent=latent_image,
denoise=float(denoise),
disable_noise=disable_noise,
start_step=int(start_at_step),
last_step=int(end_at_step),
force_full_denoise=force_full_denoise,
)
return (result_latent,)
NODE_CLASS_MAPPINGS = {"LLSProKSamplerBridge": LLSProKSamplerBridge}
tests/test_loader_prompt_refactor.py
+32 -12
View File
@@ -815,32 +815,42 @@ class TestLoaderPromptRefactor(unittest.TestCase):
self.assertEqual(payload["size_source"], "input_image")
self.assertEqual(payload["resize_mode"], "none")
def test_save_image_merges_metadata_from_native_objects_and_info_strings(self):
def test_save_image_schema_uses_info_strings_only(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSaveImage"]
required = node_cls.INPUT_TYPES()["required"]
optional = node_cls.INPUT_TYPES()["optional"]
self.assertEqual(required["image"][0], "IMAGE")
self.assertEqual(required["filename_prefix"][0], "STRING")
self.assertEqual(required["output_mode"][0], ["save", "preview_only"])
self.assertEqual(required["save_metadata"][0], "BOOLEAN")
self.assertNotIn("model", optional)
self.assertNotIn("clip", optional)
self.assertNotIn("vae", optional)
self.assertEqual(optional["prompt_info"], ("STRING", {"forceInput": True}))
self.assertEqual(optional["latent_info"], ("STRING", {"forceInput": True}))
self.assertEqual(optional["sample_info"], ("STRING", {"forceInput": True}))
self.assertEqual(optional["decode_info"], ("STRING", {"forceInput": True}))
self.assertEqual(optional["upscale_info"], ("STRING", {"forceInput": True}))
def test_save_image_merges_metadata_from_info_strings(self):
load_plugin_package()
from lls_node_test_refactor.image import nodes as image_nodes
model = TaggedValue("MODEL::SD15")
model._lls_family = "SD1.5"
model._lls_model_name = "sd15.safetensors"
clip = TaggedValue("CLIP::SD15")
clip._lls_text_encoder_name = "clip_l.safetensors"
vae = TaggedValue("VAE::SD15")
vae._lls_vae_name = "VAE::SD15"
with mock.patch.object(image_nodes, "comfy_core_nodes", CoreNodesStub()):
node = image_nodes.LLSSaveImage()
result = node.save(
image=FakeTensor((1, 512, 512, 3)),
filename_prefix="LLS",
save_metadata=True,
model=model,
clip=clip,
vae=vae,
prompt_info=json.dumps(
{
"positive_prompt": "cat",
"negative_prompt": "bad",
"model_family": "SD1.5",
"checkpoint_name": "sd15.safetensors",
"text_encoder_name": "clip_l.safetensors",
}
),
sample_info=json.dumps(
@@ -854,12 +864,22 @@ class TestLoaderPromptRefactor(unittest.TestCase):
"denoise": 1.0,
}
),
decode_info=json.dumps(
{
"vae_name": "VAE::SD15",
"width": 512,
"height": 512,
"batch_size": 1,
}
),
upscale_info=json.dumps({"mode": "none", "scale": 1.0}),
)
metadata = result["ui"]["images"][0]["extra_pnginfo"]["lls_metadata"]
self.assertEqual(metadata["positive_prompt"], "cat")
self.assertEqual(metadata["checkpoint_name"], "sd15.safetensors")
self.assertEqual(metadata["text_encoder_name"], "clip_l.safetensors")
self.assertEqual(metadata["vae_name"], "VAE::SD15")
self.assertEqual(metadata["steps"], 20)
self.assertEqual(metadata["upscale_mode"], "none")
tests/test_pro_edit_bridge.py
+65 -212
View File
@@ -1,4 +1,3 @@
import json
import unittest
from unittest import mock
@@ -14,233 +13,87 @@ class TestProEditBridge(unittest.TestCase):
self.bridge_module = import_plugin_submodule(self.plugin, "pro_edit.pro_edit_bridge")
self.node = self.bridge_module.LLSProKSamplerBridge()
def test_bridge_uses_auto_from_edit_denoise(self):
model = FakeModel(
family="SDXL",
model_role="inpaint",
supports_inpaint_native=True,
supports_image_edit_native=False,
preferred_edit_backend="sdxl",
profile_id="sdxl_inpaint",
backend_type="sdxl_native",
sampler_strategy="standard_k",
loader_strategy="sdxl_checkpoint",
)
def test_bridge_prefers_native_advanced_ksampler_when_available(self):
model = FakeModel()
positive = make_conditioning("positive")
negative = make_conditioning("negative")
latent = {"samples": FakeLatentTensor((1, 4, 64, 64)), "source": "pro_edit_prepare_region"}
captured = {}
native_result = {"samples": FakeLatentTensor((1, 4, 64, 64)), "source": "native"}
with mock.patch.object(self.bridge_module, "_common_ksampler", side_effect=lambda **kwargs: kwargs["latent"]):
result_latent, sample_info_json = self.node.sample(
class NativeKSamplerAdvanced:
def sample(self, **kwargs):
captured.update(kwargs)
return (native_result,)
with mock.patch.object(self.bridge_module, "comfy_core_nodes", mock.Mock(KSamplerAdvanced=NativeKSamplerAdvanced)), mock.patch.object(
self.bridge_module,
"_common_ksampler",
side_effect=AssertionError("_common_ksampler should not be used when native KSamplerAdvanced is available"),
):
(result_latent,) = self.node.sample(
model=model,
positive=make_conditioning("positive"),
negative=make_conditioning("negative"),
latent_image=latent,
backend_mode="auto",
quality_preset="Manual",
seed=7,
add_noise="disable",
noise_seed=11,
steps=20,
cfg=7.0,
sampler_name="euler",
scheduler="normal",
denoise=0.25,
denoise_mode="auto_from_edit",
flux_guidance=3.5,
model_family="Auto",
edit_info={
"backend_name": "sdxl",
"model_family": "SDXL",
"model_role": "inpaint",
"supports_inpaint_native": True,
"supports_image_edit_native": False,
"preferred_edit_backend": "sdxl",
"profile_id": "sdxl_inpaint",
"backend_type": "sdxl_native",
"sampler_strategy": "standard_k",
"recommended_denoise": 0.63,
},
model_info=None,
positive=positive,
negative=negative,
latent_image=latent,
start_at_step=3,
end_at_step=17,
return_with_leftover_noise="enable",
)
sample_info = json.loads(sample_info_json)
self.assertEqual(result_latent["source"], "pro_edit_prepare_region")
self.assertEqual(sample_info["backend_name"], "sdxl")
self.assertEqual(sample_info["profile_id"], "sdxl_inpaint")
self.assertEqual(sample_info["sampler_strategy"], "standard_k")
self.assertEqual(sample_info["denoise"], 0.63)
self.assertEqual(sample_info["denoise_mode"], "auto_from_edit")
self.assertIs(result_latent, native_result)
self.assertIs(captured["model"], model)
self.assertIs(captured["positive"], positive)
self.assertIs(captured["negative"], negative)
self.assertIs(captured["latent_image"], latent)
self.assertEqual(captured["noise_seed"], 11)
self.assertEqual(captured["steps"], 20)
self.assertEqual(captured["cfg"], 7.0)
self.assertEqual(captured["sampler_name"], "euler")
self.assertEqual(captured["scheduler"], "normal")
self.assertEqual(captured["denoise"], 1.0)
self.assertEqual(captured["add_noise"], "disable")
self.assertEqual(captured["start_at_step"], 3)
self.assertEqual(captured["end_at_step"], 17)
self.assertEqual(captured["return_with_leftover_noise"], "enable")
def test_bridge_applies_flux_guidance_for_flux_backend(self):
model = FakeModel(
family="FLUX_DEV",
model_role="edit",
supports_inpaint_native=False,
supports_image_edit_native=True,
preferred_edit_backend="flux",
profile_id="flux_edit",
backend_type="flux_edit",
sampler_strategy="flux_guided",
loader_strategy="flux_split_or_bundle",
)
latent = {"samples": FakeLatentTensor((1, 16, 64, 64)), "source": "pro_edit_prepare_region"}
def test_bridge_falls_back_to_common_ksampler_without_native_node(self):
latent = {"samples": FakeLatentTensor((1, 4, 64, 64))}
captured = {}
with mock.patch.object(self.bridge_module, "_common_ksampler", side_effect=lambda **kwargs: kwargs["latent"]):
_, sample_info_json = self.node.sample(
model=model,
def fake_common_ksampler(**kwargs):
captured.update(kwargs)
return kwargs["latent"]
with mock.patch.object(self.bridge_module, "comfy_core_nodes", None), mock.patch.object(
self.bridge_module,
"_common_ksampler",
side_effect=fake_common_ksampler,
):
(_result_latent,) = self.node.sample(
model=FakeModel(),
add_noise="enable",
noise_seed=21,
steps=10,
cfg=5.5,
sampler_name="heun",
scheduler="karras",
positive=make_conditioning("positive"),
negative=make_conditioning("negative"),
latent_image=latent,
backend_mode="auto",
quality_preset="Manual",
seed=9,
steps=12,
cfg=1.0,
sampler_name="euler",
scheduler="simple",
denoise=0.8,
denoise_mode="manual",
flux_guidance=4.2,
model_family="Auto",
edit_info={
"backend_name": "flux",
"model_family": "FLUX_DEV",
"model_role": "edit",
"supports_inpaint_native": False,
"supports_image_edit_native": True,
"preferred_edit_backend": "flux",
"profile_id": "flux_edit",
"backend_type": "flux_edit",
"sampler_strategy": "flux_guided",
"recommended_denoise": 0.8,
},
model_info=None,
start_at_step=0,
end_at_step=10000,
return_with_leftover_noise="disable",
)
sample_info = json.loads(sample_info_json)
self.assertEqual(sample_info["backend_name"], "flux")
self.assertEqual(sample_info["profile_id"], "flux_edit")
self.assertEqual(sample_info["sampler_strategy"], "flux_guided")
self.assertEqual(sample_info["guidance"], 4.2)
def test_bridge_manual_backend_mismatch_raises(self):
model = FakeModel(
family="SDXL",
model_role="inpaint",
supports_inpaint_native=True,
supports_image_edit_native=False,
preferred_edit_backend="sdxl",
profile_id="sdxl_inpaint",
backend_type="sdxl_native",
sampler_strategy="standard_k",
loader_strategy="sdxl_checkpoint",
)
latent = {"samples": FakeLatentTensor((1, 4, 64, 64)), "source": "pro_edit_prepare_region"}
with self.assertRaisesRegex(RuntimeError, "backend 'flux' is incompatible"):
self.node.sample(
model=model,
positive=make_conditioning("positive"),
negative=make_conditioning("negative"),
latent_image=latent,
backend_mode="flux",
quality_preset="Manual",
seed=11,
steps=20,
cfg=7.0,
sampler_name="euler",
scheduler="normal",
denoise=0.5,
denoise_mode="manual",
flux_guidance=3.5,
model_family="Auto",
edit_info=None,
model_info=None,
)
def test_bridge_rejects_unknown_sampler_strategy(self):
model = FakeModel(
family="FLUX_DEV",
model_role="edit",
supports_inpaint_native=False,
supports_image_edit_native=True,
preferred_edit_backend="flux",
profile_id="flux_edit",
backend_type="flux_edit",
sampler_strategy="mystery_strategy",
loader_strategy="flux_split_or_bundle",
)
latent = {"samples": FakeLatentTensor((1, 16, 64, 64)), "source": "pro_edit_prepare_region"}
with self.assertRaisesRegex(RuntimeError, "Unsupported sampler_strategy 'mystery_strategy'"):
self.node.sample(
model=model,
positive=make_conditioning("positive"),
negative=make_conditioning("negative"),
latent_image=latent,
backend_mode="auto",
quality_preset="Manual",
seed=9,
steps=12,
cfg=1.0,
sampler_name="euler",
scheduler="simple",
denoise=0.8,
denoise_mode="manual",
flux_guidance=4.2,
model_family="Auto",
edit_info=None,
model_info=None,
)
def test_bridge_routes_base_profile_through_fallback_execution_path(self):
model = FakeModel(
family="SDXL",
model_role="base",
supports_inpaint_native=False,
supports_image_edit_native=False,
preferred_edit_backend=None,
profile_id="sdxl_base",
backend_type="none",
sampler_strategy="standard_k",
loader_strategy="sdxl_checkpoint",
)
latent = {
"samples": FakeLatentTensor((1, 4, 64, 64)),
"noise_mask": FakeLatentTensor((1, 4, 64, 64)),
"source": "pro_edit_fallback_region",
}
with mock.patch.object(self.bridge_module, "_common_ksampler", side_effect=lambda **kwargs: kwargs["latent"]):
result_latent, sample_info_json = self.node.sample(
model=model,
positive=make_conditioning("positive"),
negative=make_conditioning("negative"),
latent_image=latent,
backend_mode="auto",
quality_preset="Manual",
seed=13,
steps=20,
cfg=7.0,
sampler_name="euler",
scheduler="normal",
denoise=0.4,
denoise_mode="manual",
flux_guidance=3.5,
model_family="Auto",
edit_info={
"backend_name": "sdxl",
"profile_id": "sdxl_base",
"backend_type": "none",
"sampler_strategy": "standard_k",
"execution_path": "fallback_repair",
"recommended_denoise": 0.4,
},
model_info=None,
)
sample_info = json.loads(sample_info_json)
self.assertEqual(result_latent["source"], "pro_edit_fallback_region")
self.assertEqual(sample_info["backend_name"], "sdxl")
self.assertEqual(sample_info["profile_id"], "sdxl_base")
self.assertEqual(sample_info["execution_path"], "fallback_repair")
self.assertFalse(captured["disable_noise"])
self.assertTrue(captured["force_full_denoise"])
if __name__ == "__main__":
tests/test_pro_edit_registration.py
+14 -17
View File
@@ -86,35 +86,32 @@ class TestProEditRegistration(unittest.TestCase):
bridge_cls = plugin.NODE_CLASS_MAPPINGS["LLSProKSamplerBridge"]
bridge_schema = bridge_cls.INPUT_TYPES()
bridge_required = bridge_schema["required"]
bridge_optional = bridge_schema["optional"]
bridge_optional = bridge_schema.get("optional", {})
self.assertEqual(bridge_cls.CATEGORY, "LLS/Image Edit")
self.assertEqual(bridge_cls.FUNCTION, "sample")
self.assertEqual(bridge_cls.RETURN_TYPES, ("LATENT", "STRING"))
self.assertEqual(bridge_cls.RETURN_NAMES, ("latent", "sample_info"))
self.assertEqual(bridge_cls.RETURN_TYPES, ("LATENT",))
self.assertEqual(bridge_cls.RETURN_NAMES, ("latent",))
self.assertEqual(
set(bridge_required),
{
"model",
"positive",
"negative",
"latent_image",
"backend_mode",
"quality_preset",
"seed",
"add_noise",
"noise_seed",
"steps",
"cfg",
"sampler_name",
"scheduler",
"denoise",
"denoise_mode",
"flux_guidance",
"model_family",
"positive",
"negative",
"latent_image",
"start_at_step",
"end_at_step",
"return_with_leftover_noise",
},
)
self.assertEqual(bridge_required["backend_mode"][0], ["auto", "sdxl", "flux"])
self.assertEqual(bridge_required["denoise_mode"][0], ["manual", "auto_from_edit"])
self.assertEqual(bridge_optional["edit_info"], ("LLS_EDIT_INFO",))
self.assertEqual(bridge_optional["model_info"], ("STRING",))
self.assertEqual(bridge_required["add_noise"][0], ["enable", "disable"])
self.assertEqual(bridge_required["return_with_leftover_noise"][0], ["disable", "enable"])
self.assertEqual(bridge_optional, {})
finish_cls = plugin.NODE_CLASS_MAPPINGS["LLSProImageEditFinish"]
finish_schema = finish_cls.INPUT_TYPES()
tests/test_pro_edit_registry.py
+27 -3
View File
@@ -70,7 +70,7 @@ class TestProEditRegistry(unittest.TestCase):
self.assertEqual(routing.routing_reason, "profile.backend_type")
self.assertEqual(routing.profile["profile_id"], "sdxl_inpaint")
def test_manual_flux_override_rejects_sdxl_only_model(self):
def test_manual_flux_override_falls_back_to_profile_backend_for_sdxl_native_profile(self):
model = FakeModel(
family="SDXL",
model_role="inpaint",
@@ -83,8 +83,12 @@ class TestProEditRegistry(unittest.TestCase):
loader_strategy="sdxl_checkpoint",
)
with self.assertRaisesRegex(RuntimeError, "backend 'flux' is incompatible with profile"):
self.registry.resolve_backend("flux", model=model)
backend, routing = self.registry.resolve_backend("flux", model=model)
self.assertEqual(backend.backend_name, "sdxl")
self.assertEqual(routing.routing_reason, "manual.incompatible_native_fallback")
self.assertEqual(routing.execution_path, "native_edit")
self.assertEqual(routing.profile["profile_id"], "sdxl_inpaint")
def test_auto_routes_sdxl_base_profile_to_family_fallback(self):
model = FakeModel(
@@ -145,6 +149,26 @@ class TestProEditRegistry(unittest.TestCase):
self.assertEqual(routing.routing_reason, "manual.family_fallback")
self.assertEqual(routing.execution_path, "fallback_repair")
def test_manual_flux_override_falls_back_to_generic_for_sd15_base_profile(self):
model = FakeModel(
family="SD1.5",
model_role="base",
supports_inpaint_native=False,
supports_image_edit_native=False,
preferred_edit_backend=None,
profile_id="sd15_base",
backend_type="none",
sampler_strategy="standard_k",
loader_strategy="sd15_checkpoint",
)
backend, routing = self.registry.resolve_backend("flux", model=model)
self.assertEqual(backend.backend_name, "generic")
self.assertEqual(routing.routing_reason, "manual.incompatible_family_fallback")
self.assertEqual(routing.execution_path, "fallback_repair")
self.assertEqual(routing.profile["profile_id"], "sd15_base")
if __name__ == "__main__":
unittest.main()
测试工作流.json
+4 -27
View File
@@ -1533,12 +1533,6 @@
"name": "latent_image",
"type": "LATENT",
"link": 105
},
{
"name": "edit_info",
"shape": 7,
"type": "LLS_EDIT_INFO",
"link": 118
}
],
"outputs": [
@@ -1548,11 +1542,6 @@
"links": [
121
]
},
{
"name": "sample_info",
"type": "STRING",
"links": null
}
],
"properties": {
@@ -1561,19 +1550,16 @@
"Node name for S&R": "LLSProKSamplerBridge"
},
"widgets_values": [
"auto",
"Manual",
"enable",
54136361244841,
"randomize",
20,
1,
"euler",
"simple",
1,
"manual",
3.5,
"Auto",
""
0,
10000,
"disable"
]
},
{
@@ -1817,7 +1803,6 @@
"name": "edit_info",
"type": "LLS_EDIT_INFO",
"links": [
118,
119
]
},
@@ -2348,14 +2333,6 @@
4,
"CONDITIONING"
],
[
118,
67,
3,
68,
4,
"LLS_EDIT_INFO"
],
[
119,
67,