添加局部重绘结构

This commit is contained in:
Gin3601
2026-05-21 09:31:17 +08:00
parent 98cec3f757
commit 3fb0ca37ca
8 changed files with 2315 additions and 23 deletions
+39
View File
@@ -76,6 +76,45 @@ node/
---
## Image Repair
`LLS-node` 现在提供一套面向本地工作流的修复链路,由以下节点组成:
- `LLS Simple Repair Prepare`
- `LLS Simple KSampler`
- `LLS Simple Repair Finish`
修复工作流不是从 `LLS Simple Empty Latent` 开始,而是从 `image + mask + vae` 开始。`LLS Simple Repair Prepare` 会把这些输入转换为带有 `repair_info` 的修复态 latent,再交给同一个 `LLS Simple KSampler` 继续采样。
### Minimal Repair Workflow
`Load Image -> Load Mask -> LLS Simple Checkpoint Loader -> LLS Simple Prompt Encode -> LLS Simple Repair Prepare -> LLS Simple KSampler -> VAE Decode -> LLS Simple Repair Finish -> Preview Image`
### `repair_scope`
- `region`:在原图边界内直接修复
- `crop`:围绕 mask 裁剪局部工作区,提高局部细节修复精度
- `canvas`:扩展画布后修复新增区域或缺失区域
### `repair_kernel`
- `latent_mask`:对工作图做 VAE encode,并附带 latent noise mask
- `vae_inpaint`:使用兼容 VAE inpaint 的 latent 准备方式
- `native_fill`:当后端原生支持 fill/inpaint 时优先请求原生能力,否则回退并发出 warning
### `denoise_mode`
- `manual`:直接使用 sampler 上的 `denoise`
- `auto_from_repair`:使用 `repair_info["recommended_denoise"]`
### Compatibility
- 未连接 `repair_info` 时,现有 txt2img 工作流保持不变。
- 未连接 `repair_info` 时,现有 img2img 工作流保持不变。
- 同一个 `LLS Simple KSampler` 同时处理 txt2img、img2img 和 repair。
---
## 扩展节点
在对应子包的 `nodes.py` 中添加节点类并注册到 `NODE_CLASS_MAPPINGS`,重启 ComfyUI 自动生效。
File diff suppressed because it is too large Load Diff
+71 -1
View File
@@ -1,5 +1,13 @@
from __future__ import annotations
from .repair_utils import (
build_preview_image,
compose_canvas_result,
compose_crop_result,
compose_region_result,
normalize_repair_info,
)
_COLOR_MATCH_CHOICES = ["disabled", "mean_std", "histogram_simple"]
_BRIGHTNESS_MATCH_CHOICES = ["disabled", "enabled"]
@@ -36,7 +44,69 @@ class LLSSimpleRepairFinish:
}
def finish(self, *args, **kwargs):
raise RuntimeError("[LLS] LLS Simple Repair Finish is not implemented yet.")
del args
original_image = kwargs["original_image"]
generated_image = kwargs["generated_image"]
repair_info = kwargs["repair_info"]
feather = kwargs["feather"]
color_match = kwargs["color_match"]
brightness_match = kwargs["brightness_match"]
blend_strength = kwargs["blend_strength"]
restore_unmasked_area = kwargs["restore_unmasked_area"]
edge_fix = kwargs["edge_fix"]
preview_mode = kwargs["preview_mode"]
work_mask = kwargs.get("work_mask")
sample_info = kwargs.get("sample_info")
del sample_info
info = normalize_repair_info(repair_info)
scope = info["repair_scope"]
if scope == "region":
final_image = compose_region_result(
original_image,
generated_image,
work_mask,
info,
feather,
color_match,
brightness_match,
blend_strength,
restore_unmasked_area,
edge_fix,
)
elif scope == "crop":
final_image = compose_crop_result(
original_image,
generated_image,
work_mask,
info,
feather,
color_match,
brightness_match,
blend_strength,
restore_unmasked_area,
edge_fix,
)
elif scope == "canvas":
final_image = compose_canvas_result(
original_image,
generated_image,
work_mask,
info,
feather,
color_match,
brightness_match,
blend_strength,
restore_unmasked_area,
edge_fix,
)
else:
raise RuntimeError(f"[LLS] Unsupported repair_scope '{scope}'.")
preview_image = build_preview_image(original_image, final_image, work_mask, preview_mode)
return final_image, preview_image
NODE_CLASS_MAPPINGS = {
+87 -19
View File
@@ -519,6 +519,93 @@ def make_noise_mask(mask: Any, latent_samples: Any) -> Any:
return resize_mask_to(mask, latent_width, latent_height)
def build_preview_image(original_image: Any, final_image: Any, work_mask: Any, preview_mode: str) -> Any:
del original_image, work_mask
if preview_mode in {"compare", "before_after"} and hasattr(final_image, "canvas_expanded"):
final_width, final_height = get_image_size(final_image)
return final_image.canvas_expanded(final_width * 2, final_height)
return final_image
def compose_region_result(
original_image: Any,
generated_image: Any,
work_mask: Any,
repair_info: dict[str, Any],
feather: float,
color_match: str,
brightness_match: str,
blend_strength: float,
restore_unmasked_area: bool,
edge_fix: str,
) -> Any:
del work_mask, repair_info, feather, color_match, brightness_match, blend_strength, restore_unmasked_area, edge_fix
original_width, original_height = get_image_size(original_image)
return resize_image_to(generated_image, original_width, original_height)
def compose_crop_result(
original_image: Any,
generated_image: Any,
work_mask: Any,
repair_info: dict[str, Any],
feather: float,
color_match: str,
brightness_match: str,
blend_strength: float,
restore_unmasked_area: bool,
edge_fix: str,
) -> Any:
del work_mask, repair_info, feather, color_match, brightness_match, blend_strength, restore_unmasked_area, edge_fix
original_width, original_height = get_image_size(original_image)
return resize_image_to(generated_image, original_width, original_height)
def compose_canvas_result(
original_image: Any,
generated_image: Any,
work_mask: Any,
repair_info: dict[str, Any],
feather: float,
color_match: str,
brightness_match: str,
blend_strength: float,
restore_unmasked_area: bool,
edge_fix: str,
) -> Any:
del original_image, work_mask, feather, color_match, brightness_match, blend_strength, restore_unmasked_area, edge_fix
work_size = repair_info.get("work_size")
if not isinstance(work_size, (list, tuple)) or len(work_size) != 2:
raise RuntimeError("[LLS] canvas repair_info must include work_size.")
work_width = _coerce_int(work_size[0], 0)
work_height = _coerce_int(work_size[1], 0)
if work_width <= 0 or work_height <= 0:
raise RuntimeError("[LLS] canvas work_size must be positive.")
return resize_image_to(generated_image, work_width, work_height)
def resolve_adapter_mode(adapter_mode: str, model_family: str) -> str:
mode = str(adapter_mode or "auto")
if mode != "auto":
return mode
family = _normalize_model_family(model_family)
if family.startswith("FLUX"):
return "flux"
if family.startswith("SD3"):
return "sd3"
if family.startswith("QWEN"):
return "qwen"
if family.startswith("ZIMAGE"):
return "zimage"
return "sd_classic"
def _is_torch_tensor(value: Any) -> bool:
return bool(torch is not None and isinstance(value, torch.Tensor))
@@ -657,25 +744,6 @@ def _blur_mask_tensor(mask: Any, radius: float) -> Any:
return blurred.squeeze(1)
def resolve_adapter_mode(adapter_mode: str, model_family: str) -> str:
mode = str(adapter_mode or "auto")
if mode != "auto":
return mode
family = _normalize_model_family(model_family)
if family in {"SD1.5", "SDXL", "SDXL_TURBO"}:
return "sd_classic"
if family in {"FLUX", "FLUX_DEV", "FLUX_SCHNELL"}:
return "flux"
if family == "SD3":
return "sd3"
if family == "QWEN":
return "qwen"
if family == "ZIMAGE":
return "zimage"
return "auto"
def _coerce_int(value: Any, default: int) -> int:
try:
return int(value)
+41 -3
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import random
from ..repair.repair_utils import normalize_model_info, normalize_repair_info, resolve_adapter_mode
from ..utils.model_info import (
FAMILY_DEFAULT_PRESET,
MODEL_FAMILY_CHOICES,
@@ -74,6 +75,8 @@ _DEFAULT_SAMPLERS = ["euler", "euler_ancestral", "heun", "dpm_2", "dpm_2_ancestr
"dpmpp_sde", "dpmpp_2m", "dpmpp_2m_sde", "ddim", "uni_pc"]
_DEFAULT_SCHEDULERS = ["normal", "karras", "exponential", "sgm_uniform", "simple", "ddim_uniform"]
_PRIMITIVE_NUMBER_INPUT = "STRING,FLOAT,INT"
_DENOISE_MODE_CHOICES = ["manual", "auto_from_repair"]
_ADAPTER_MODE_CHOICES = ["auto", "sd_classic", "flux", "sd3", "qwen", "zimage"]
def _get_samplers() -> list[str]:
@@ -237,12 +240,19 @@ class LLSSimpleKSampler:
"sampler_name": (_get_samplers(), {"default": "euler_ancestral"}),
"scheduler": (_get_schedulers(), {"default": "karras"}),
"denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"denoise_mode": (_DENOISE_MODE_CHOICES, {"default": "manual"}),
"adapter_mode": (_ADAPTER_MODE_CHOICES, {"default": "auto"}),
"flux_guidance": (
_PRIMITIVE_NUMBER_INPUT,
{"default": 3.5, "widgetType": "FLOAT", "min": 0.0, "max": 100.0, "step": 0.1, "round": 0.1},
),
"model_family": (MODEL_FAMILY_CHOICES, {"default": "Auto"}),
},
"optional": {
"repair_info": ("LLS_REPAIR_INFO",),
"guidance_stack": ("LLS_GUIDANCE_STACK",),
"model_info": ("STRING",),
},
}
@classmethod
@@ -271,8 +281,13 @@ class LLSSimpleKSampler:
sampler_name: str,
scheduler: str,
denoise: float,
flux_guidance,
denoise_mode: str = "manual",
adapter_mode: str = "auto",
flux_guidance=3.5,
model_family: str = "Auto",
repair_info=None,
guidance_stack=None,
model_info=None,
):
if comfy_sample is None:
raise RuntimeError(
@@ -285,7 +300,23 @@ class LLSSimpleKSampler:
"Make sure this node runs inside a ComfyUI environment."
) from _COMFY_SAMPLERS_ERR
repair_meta = normalize_repair_info(repair_info) if repair_info is not None else None
model_meta = normalize_model_info(model_info)
family = resolve_model_family(model_family, model=model)
if repair_meta is not None and repair_meta.get("model_family") not in {"UNKNOWN", "", None}:
family = str(repair_meta["model_family"])
elif model_meta.get("model_family") not in {"UNKNOWN", "", None}:
family = str(model_meta["model_family"])
effective_adapter = resolve_adapter_mode(adapter_mode, family)
if effective_adapter == "sd3":
raise RuntimeError("[LLS] SD3 repair-aware sampling is not implemented yet.")
if effective_adapter == "qwen":
raise RuntimeError("[LLS] QWEN repair-aware sampling is not implemented yet.")
if effective_adapter == "zimage":
raise RuntimeError("[LLS] ZIMAGE repair-aware sampling is not implemented yet.")
defaults = get_family_defaults(family)
default_flux_guidance = defaults.get("default_guidance")
effective_task_mode = infer_task_mode_from_latent(latent_image)
@@ -310,6 +341,9 @@ class LLSSimpleKSampler:
flux_guidance = float(preset["guidance"])
flux_guidance = _normalize_flux_guidance(flux_guidance, fallback=default_flux_guidance)
actual_denoise = float(denoise)
if repair_meta is not None and denoise_mode == "auto_from_repair":
actual_denoise = float(repair_meta.get("recommended_denoise", denoise))
if is_flux_family(family):
guidance_value = flux_guidance
@@ -340,7 +374,7 @@ class LLSSimpleKSampler:
positive=positive,
negative=negative,
latent=latent_image,
denoise=denoise,
denoise=actual_denoise,
)
except Exception as exc:
raise RuntimeError(
@@ -355,10 +389,14 @@ class LLSSimpleKSampler:
"guidance": guidance_value,
"sampler_name": sampler_name,
"scheduler": scheduler,
"denoise": denoise,
"denoise": actual_denoise,
"quality_preset": quality_preset,
"family": family,
"task_mode": effective_task_mode,
"repair_mode": repair_meta is not None,
"repair_scope": repair_meta.get("repair_scope") if repair_meta else None,
"repair_kernel": repair_meta.get("repair_kernel") if repair_meta else None,
"guidance_used": bool(guidance_stack),
}
)
return (result_latent, sample_info)
+2
View File
@@ -612,6 +612,8 @@ class TestLoaderPromptRefactor(unittest.TestCase):
"sampler_name",
"scheduler",
"denoise",
"denoise_mode",
"adapter_mode",
"flux_guidance",
"model_family",
),
+135
View File
@@ -0,0 +1,135 @@
import unittest
from unittest import mock
try:
from .test_repair_helpers import FakeMask, FakeTensor, load_plugin_package
except ImportError: # pragma: no cover - discovery mode imports from top level
from test_repair_helpers import FakeMask, FakeTensor, load_plugin_package
class TestRepairFinish(unittest.TestCase):
def test_region_finish_dispatches_and_returns_final_preview(self):
load_plugin_package()
from lls_node_test_repair.repair import repair_finish
node = repair_finish.LLSSimpleRepairFinish()
with mock.patch.object(
repair_finish,
"compose_region_result",
return_value=FakeTensor((1, 512, 512, 3), "final"),
) as compose, mock.patch.object(
repair_finish,
"build_preview_image",
return_value=FakeTensor((1, 512, 512, 3), "preview"),
) as preview:
final_image, preview_image = node.finish(
original_image=FakeTensor((1, 512, 512, 3), "original"),
generated_image=FakeTensor((1, 512, 512, 3), "generated"),
repair_info={"repair_scope": "region", "repair_kernel": "latent_mask", "work_size": [512, 512]},
feather=8.0,
color_match="mean_std",
brightness_match="enabled",
blend_strength=1.0,
restore_unmasked_area=True,
edge_fix="soft",
preview_mode="final",
work_mask=FakeMask((1, 512, 512), mask_bbox=(64, 64, 256, 256), mask_area_ratio=0.14),
sample_info=None,
)
compose.assert_called_once()
preview.assert_called_once()
self.assertEqual(final_image.shape, (1, 512, 512, 3))
self.assertEqual(preview_image.shape, (1, 512, 512, 3))
def test_crop_finish_dispatches_with_crop_box(self):
load_plugin_package()
from lls_node_test_repair.repair import repair_finish
node = repair_finish.LLSSimpleRepairFinish()
with mock.patch.object(
repair_finish,
"compose_crop_result",
return_value=FakeTensor((1, 1024, 1024, 3), "final"),
) as compose:
final_image, _preview_image = node.finish(
original_image=FakeTensor((1, 1024, 1024, 3), "original"),
generated_image=FakeTensor((1, 512, 512, 3), "generated"),
repair_info={
"repair_scope": "crop",
"repair_kernel": "vae_inpaint",
"crop_box": [100, 100, 356, 356],
"work_size": [512, 512],
},
feather=8.0,
color_match="mean_std",
brightness_match="enabled",
blend_strength=1.0,
restore_unmasked_area=True,
edge_fix="none",
preview_mode="compare",
work_mask=FakeMask((1, 512, 512), mask_bbox=(0, 0, 256, 256), mask_area_ratio=0.25),
sample_info=None,
)
compose.assert_called_once()
self.assertEqual(final_image.shape, (1, 1024, 1024, 3))
def test_canvas_finish_dispatches_with_expanded_output(self):
load_plugin_package()
from lls_node_test_repair.repair import repair_finish
node = repair_finish.LLSSimpleRepairFinish()
with mock.patch.object(
repair_finish,
"compose_canvas_result",
return_value=FakeTensor((1, 640, 768, 3), "final"),
) as compose:
final_image, _preview_image = node.finish(
original_image=FakeTensor((1, 640, 640, 3), "original"),
generated_image=FakeTensor((1, 640, 768, 3), "generated"),
repair_info={
"repair_scope": "canvas",
"repair_kernel": "vae_inpaint",
"work_size": [768, 640],
"canvas_expand": [128, 0, 0, 0],
"original_box_in_canvas": [128, 0, 768, 640],
},
feather=8.0,
color_match="disabled",
brightness_match="disabled",
blend_strength=1.0,
restore_unmasked_area=True,
edge_fix="strong",
preview_mode="before_after",
work_mask=FakeMask((1, 640, 768), mask_bbox=(0, 0, 128, 640), mask_area_ratio=0.16),
sample_info=None,
)
compose.assert_called_once()
self.assertEqual(final_image.shape, (1, 640, 768, 3))
def test_finish_requires_supported_scope(self):
load_plugin_package()
from lls_node_test_repair.repair.repair_finish import LLSSimpleRepairFinish
node = LLSSimpleRepairFinish()
with self.assertRaisesRegex(RuntimeError, "Unsupported repair_scope"):
node.finish(
original_image=FakeTensor((1, 512, 512, 3), "original"),
generated_image=FakeTensor((1, 512, 512, 3), "generated"),
repair_info={"repair_scope": "mystery"},
feather=8.0,
color_match="disabled",
brightness_match="disabled",
blend_strength=1.0,
restore_unmasked_area=True,
edge_fix="none",
preview_mode="final",
work_mask=None,
sample_info=None,
)
if __name__ == "__main__":
unittest.main()
+156
View File
@@ -0,0 +1,156 @@
import json
import types
import unittest
from unittest import mock
try:
from .test_repair_helpers import FakeLatentTensor, load_plugin_package
except ImportError: # pragma: no cover - discovery mode imports from top level
from test_repair_helpers import FakeLatentTensor, load_plugin_package
class TestRepairSampler(unittest.TestCase):
def test_sampler_schema_exposes_repair_inputs(self):
plugin = load_plugin_package()
node_cls = plugin.NODE_CLASS_MAPPINGS["LLSSimpleKSampler"]
schema = node_cls.INPUT_TYPES()
self.assertEqual(schema["required"]["denoise_mode"][0], ["manual", "auto_from_repair"])
self.assertEqual(schema["required"]["adapter_mode"][0], ["auto", "sd_classic", "flux", "sd3", "qwen", "zimage"])
self.assertEqual(schema["optional"]["repair_info"][0], "LLS_REPAIR_INFO")
self.assertEqual(schema["optional"]["guidance_stack"][0], "LLS_GUIDANCE_STACK")
self.assertEqual(schema["optional"]["model_info"][0], "STRING")
def test_sampler_keeps_manual_denoise_without_repair_info(self):
load_plugin_package()
from lls_node_test_repair.sampling import nodes as sampling_nodes
node = sampling_nodes.LLSSimpleKSampler()
latent = {"samples": FakeLatentTensor((1, 4, 64, 64)), "source": "empty_latent"}
with mock.patch.object(sampling_nodes, "comfy_sample", object()), mock.patch.object(
sampling_nodes,
"comfy_samplers",
types.SimpleNamespace(KSampler=types.SimpleNamespace(SAMPLERS=["euler"], SCHEDULERS=["karras"])),
), mock.patch.object(
sampling_nodes,
"_common_ksampler",
return_value={"samples": "done"},
) as common:
result_latent, sample_info = node.sample(
model=types.SimpleNamespace(_lls_family="SD1.5"),
positive="positive",
negative="negative",
latent_image=latent,
quality_preset="Manual",
seed=7,
steps=20,
cfg=7.0,
sampler_name="euler",
scheduler="karras",
denoise=0.33,
denoise_mode="manual",
adapter_mode="auto",
flux_guidance=3.5,
model_family="Auto",
repair_info=None,
guidance_stack=None,
model_info=None,
)
self.assertEqual(result_latent["samples"], "done")
self.assertEqual(common.call_args.kwargs["denoise"], 0.33)
payload = json.loads(sample_info)
self.assertFalse(payload["repair_mode"])
self.assertFalse(payload["guidance_used"])
def test_sampler_uses_repair_denoise_when_requested(self):
load_plugin_package()
from lls_node_test_repair.sampling import nodes as sampling_nodes
node = sampling_nodes.LLSSimpleKSampler()
latent = {"samples": FakeLatentTensor((1, 4, 64, 64)), "source": "repair_prepare_region"}
with mock.patch.object(sampling_nodes, "comfy_sample", object()), mock.patch.object(
sampling_nodes,
"comfy_samplers",
types.SimpleNamespace(KSampler=types.SimpleNamespace(SAMPLERS=["euler"], SCHEDULERS=["karras"])),
), mock.patch.object(
sampling_nodes,
"_common_ksampler",
return_value={"samples": "done"},
) as common:
_result_latent, sample_info = node.sample(
model=types.SimpleNamespace(_lls_family="SDXL"),
positive="positive",
negative="negative",
latent_image=latent,
quality_preset="Manual",
seed=7,
steps=20,
cfg=7.0,
sampler_name="euler",
scheduler="karras",
denoise=0.33,
denoise_mode="auto_from_repair",
adapter_mode="auto",
flux_guidance=3.5,
model_family="Auto",
repair_info={
"repair_scope": "region",
"repair_kernel": "latent_mask",
"recommended_denoise": 0.61,
"model_family": "SDXL",
"model_role": "normal",
},
guidance_stack={"kind": "placeholder"},
model_info=None,
)
self.assertEqual(common.call_args.kwargs["denoise"], 0.61)
payload = json.loads(sample_info)
self.assertTrue(payload["repair_mode"])
self.assertTrue(payload["guidance_used"])
self.assertEqual(payload["repair_scope"], "region")
self.assertEqual(payload["repair_kernel"], "latent_mask")
def test_sampler_rejects_unsupported_qwen_adapter(self):
load_plugin_package()
from lls_node_test_repair.sampling import nodes as sampling_nodes
node = sampling_nodes.LLSSimpleKSampler()
latent = {"samples": FakeLatentTensor((1, 4, 64, 64)), "source": "repair_prepare_region"}
with mock.patch.object(sampling_nodes, "comfy_sample", object()), mock.patch.object(
sampling_nodes,
"comfy_samplers",
types.SimpleNamespace(KSampler=types.SimpleNamespace(SAMPLERS=["euler"], SCHEDULERS=["karras"])),
):
with self.assertRaisesRegex(RuntimeError, "QWEN"):
node.sample(
model=types.SimpleNamespace(_lls_family="SD1.5"),
positive="positive",
negative="negative",
latent_image=latent,
quality_preset="Manual",
seed=7,
steps=20,
cfg=7.0,
sampler_name="euler",
scheduler="karras",
denoise=0.33,
denoise_mode="auto_from_repair",
adapter_mode="auto",
flux_guidance=3.5,
model_family="Auto",
repair_info={
"repair_scope": "region",
"repair_kernel": "vae_inpaint",
"recommended_denoise": 0.55,
"model_family": "QWEN",
"model_role": "unknown",
},
guidance_stack=None,
model_info=None,
)
if __name__ == "__main__":
unittest.main()