From 1d481c0dc75cac2cd1342012971f82a39792e83e Mon Sep 17 00:00:00 2001 From: Gin3601 Date: Wed, 20 May 2026 16:24:04 +0800 Subject: [PATCH] =?UTF-8?q?Qwen=E7=B3=BB=E5=88=97=E6=9C=89=E5=8D=95?= =?UTF-8?q?=E7=8B=AC=E7=9A=84=E5=B0=81=E8=A3=85=E8=8A=82=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- __init__.py | 1 + .../plans/2026-05-20-qwen-minimal-nodes.md | 395 +++++++++++++ qwen/__init__.py | 8 + qwen/discovery.py | 115 ++++ qwen/nodes.py | 114 ++++ qwen/runtime.py | 266 +++++++++ tests/test_qwen_nodes.py | 535 ++++++++++++++++++ 7 files changed, 1434 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md create mode 100644 qwen/__init__.py create mode 100644 qwen/discovery.py create mode 100644 qwen/nodes.py create mode 100644 qwen/runtime.py create mode 100644 tests/test_qwen_nodes.py diff --git a/__init__.py b/__init__.py index 144eba4..6ee9240 100644 --- a/__init__.py +++ b/__init__.py @@ -42,6 +42,7 @@ _SUBPACKAGES: list[str] = [ "model_loader", # 第 1 节：模型加载与管理 "conditioning", # 第 2 节：文本编码与条件生成 "sampling", # 第 3 节：图像采样与去噪 + "qwen", # 第 4 节：Qwen 一体化节点 "latent", # 第 4 节：Latent 空间操作 "image", # 第 5 节：图像处理与后处理 "upscale", # 第 5 节（超分子域）：图像超分 diff --git a/docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md b/docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md new file mode 100644 index 0000000..dbe6e71 --- /dev/null +++ b/docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md @@ -0,0 +1,395 @@ +# Qwen Minimal Nodes Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add two minimal high-level Qwen nodes that hide model/clip/vae wiring and directly output `IMAGE` for text-to-image and official image edit workflows. + +**Architecture:** Introduce a dedicated `qwen/` subpackage with three focused modules: discovery for compatible model/resource filtering, runtime for thin wrappers around official ComfyUI Qwen nodes, and user-facing nodes that expose the compressed UX. Keep Qwen isolated from the existing SD1.5/SDXL/FLUX simple-chain code and validate compatibility both in the widget choices and at runtime. + +**Tech Stack:** Python, unittest, ComfyUI core nodes, `comfy_extras.nodes_qwen`, `comfy_extras.nodes_flux`, `comfy_extras.nodes_model_advanced`, `comfy_extras.nodes_cfg` + +--- + +### Task 1: Add Qwen discovery helpers and red tests + +**Files:** +- Create: `qwen/discovery.py` +- Create: `tests/test_qwen_nodes.py` +- Test: `tests/test_qwen_nodes.py` + +- [ ] **Step 1: Write the failing tests** + +```python +class QwenFolderPathsStub: + def __init__(self): + self._files = { + "diffusion_models": [ + "qwen_image_fp8_e4m3fn.safetensors", + "qwen_image_2512_fp8_e4m3fn.safetensors", + "qwen_image_edit_2511_bf16.safetensors", + "qwen_image_layered_bf16.safetensors", + "flux1-dev.safetensors", + ], + "text_encoders": ["qwen_2.5_vl_7b_fp8_scaled.safetensors"], + "vae": ["qwen_image_vae.safetensors"], + } + + def get_filename_list(self, category): + return list(self._files.get(category, [])) + + +def test_qwen_text_node_filters_only_text_models(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen.nodes import LLSQwenTextToImage + + with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()): + choices = LLSQwenTextToImage.INPUT_TYPES()["required"]["model_name"][0] + + assert choices == [ + "qwen_image_2512_fp8_e4m3fn.safetensors", + "qwen_image_fp8_e4m3fn.safetensors", + ] + + +def test_qwen_edit_node_filters_only_edit_models(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen.nodes import LLSQwenImageEdit + + with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()): + choices = LLSQwenImageEdit.INPUT_TYPES()["required"]["model_name"][0] + + assert choices == ["qwen_image_edit_2511_bf16.safetensors"] + + +def test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen.nodes import LLSQwenTextToImage + + stub = QwenFolderPathsStub() + stub._files["diffusion_models"] = ["flux1-dev.safetensors", "sdxl.safetensors"] + with mock.patch.object(qwen_discovery, "folder_paths", stub): + choices = LLSQwenTextToImage.INPUT_TYPES()["required"]["model_name"][0] + + assert choices == ["(no qwen text-to-image models found)"] +``` + +- [ ] **Step 2: Run the focused tests to verify they fail** + +Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_filters_only_text_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_filters_only_edit_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist -v` +Expected: FAIL because the `qwen/` package and node classes do not exist yet. + +- [ ] **Step 3: Write the minimal discovery implementation** + +```python +TEXT_PLACEHOLDER = "(no qwen text-to-image models found)" +EDIT_PLACEHOLDER = "(no qwen image edit models found)" + + +def is_qwen_text_to_image_model(name: str) -> bool: + lowered = name.lower() + return ( + "qwen" in lowered + and "image" in lowered + and "edit" not in lowered + and "layered" not in lowered + ) + + +def is_qwen_image_edit_model(name: str) -> bool: + lowered = name.lower() + return "qwen" in lowered and "image" in lowered and "edit" in lowered and "layered" not in lowered + + +def get_qwen_text_model_choices() -> list[str]: + names = [name for name in _get_filename_list("diffusion_models") if is_qwen_text_to_image_model(name)] + return sorted(dict.fromkeys(names)) or [TEXT_PLACEHOLDER] + + +def get_qwen_edit_model_choices() -> list[str]: + names = [name for name in _get_filename_list("diffusion_models") if is_qwen_image_edit_model(name)] + return sorted(dict.fromkeys(names)) or [EDIT_PLACEHOLDER] +``` + +- [ ] **Step 4: Run the focused tests to verify they pass** + +Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_filters_only_text_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_filters_only_edit_models tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist -v` +Expected: PASS with the correct model filtering and placeholder behavior. + +- [ ] **Step 5: Commit the discovery slice** + +```bash +git add qwen/discovery.py tests/test_qwen_nodes.py +git commit -m "feat: add qwen model discovery helpers" +``` + +### Task 2: Add runtime wrappers for official Qwen flows with red tests + +**Files:** +- Create: `qwen/runtime.py` +- Modify: `tests/test_qwen_nodes.py` +- Test: `tests/test_qwen_nodes.py` + +- [ ] **Step 1: Write the failing runtime tests** + +```python +def test_qwen_text_runtime_rejects_incompatible_model_name(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()): + with self.assertRaisesRegex(RuntimeError, "not compatible with LLSQwenTextToImage"): + qwen_runtime.run_qwen_text_to_image( + model_name="qwen_image_edit_2511_bf16.safetensors", + prompt="a cat", + width=1024, + height=1024, + steps=20, + seed=1, + batch_size=1, + ) + + +def test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + stub = QwenFolderPathsStub() + stub._files["text_encoders"] = [] + with mock.patch.object(qwen_discovery, "folder_paths", stub): + with self.assertRaisesRegex(RuntimeError, "Missing Qwen text encoder"): + qwen_runtime.run_qwen_text_to_image( + model_name="qwen_image_fp8_e4m3fn.safetensors", + prompt="a cat", + width=1024, + height=1024, + steps=20, + seed=1, + batch_size=1, + ) + + +def test_qwen_text_runtime_executes_minimal_official_pipeline(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()), \ + mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub()), \ + mock.patch.object(qwen_runtime, "nodes_sd3", SD3NodesStub()), \ + mock.patch.object(qwen_runtime, "nodes_model_advanced", ModelAdvancedStub()): + image = qwen_runtime.run_qwen_text_to_image( + model_name="qwen_image_fp8_e4m3fn.safetensors", + prompt="a cat", + width=1024, + height=1024, + steps=20, + seed=99, + batch_size=2, + ) + + self.assertEqual(image, "IMAGE::decoded") +``` + +- [ ] **Step 2: Run the focused runtime tests to verify they fail** + +Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_rejects_incompatible_model_name tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_executes_minimal_official_pipeline -v` +Expected: FAIL because the Qwen runtime module and wrappers do not exist yet. + +- [ ] **Step 3: Write the minimal runtime implementation** + +```python +def run_qwen_text_to_image(model_name, prompt, width, height, steps, seed, batch_size): + validate_qwen_text_model_name(model_name) + clip_name = resolve_qwen_text_encoder_name() + vae_name = resolve_qwen_vae_name() + if clip_name is None: + raise RuntimeError("[LLS] Missing Qwen text encoder. Place a compatible Qwen VL text encoder in ComfyUI/models/text_encoders/.") + if vae_name is None: + raise RuntimeError("[LLS] Missing Qwen VAE. Place qwen_image_vae in ComfyUI/models/vae/.") + + model = comfy_core_nodes.UNETLoader().load_unet(model_name, "default")[0] + clip = comfy_core_nodes.CLIPLoader().load_clip(clip_name, type="qwen_image", device="default")[0] + vae = comfy_core_nodes.VAELoader().load_vae(vae_name)[0] + + model = nodes_model_advanced.ModelSamplingAuraFlow().patch_aura(model, 3.1)[0] + positive = comfy_core_nodes.CLIPTextEncode().encode(clip, prompt)[0] + negative = comfy_core_nodes.CLIPTextEncode().encode(clip, "")[0] + latent = nodes_sd3.EmptySD3LatentImage().generate(width, height, batch_size)[0] + sampled = comfy_core_nodes.KSampler().sample( + model, seed, steps, 4.0, "euler", "simple", positive, negative, latent, denoise=1.0 + )[0] + return comfy_core_nodes.VAEDecode().decode(vae, sampled)[0] +``` + +- [ ] **Step 4: Extend the runtime implementation for official image edit** + +```python +def run_qwen_image_edit(model_name, image, prompt, steps, seed): + validate_qwen_edit_model_name(model_name) + clip_name = resolve_qwen_text_encoder_name() + vae_name = resolve_qwen_vae_name() + ... + scaled = nodes_flux.FluxKontextImageScale.execute(image).result[0] + positive = nodes_qwen.TextEncodeQwenImageEditPlus.execute(clip, prompt, vae=vae, image1=scaled).result[0] + negative = nodes_qwen.TextEncodeQwenImageEditPlus.execute(clip, "", vae=vae, image1=scaled).result[0] + positive = nodes_flux.FluxKontextMultiReferenceLatentMethod.execute(positive, "index_timestep_zero").result[0] + negative = nodes_flux.FluxKontextMultiReferenceLatentMethod.execute(negative, "index_timestep_zero").result[0] + latent = comfy_core_nodes.VAEEncode().encode(vae, scaled)[0] + model = nodes_model_advanced.ModelSamplingAuraFlow().patch_aura(model, 3.1)[0] + model = nodes_cfg.CFGNorm.execute(model, 1.0).result[0] + sampled = comfy_core_nodes.KSampler().sample( + model, seed, steps, 4.0, "euler", "simple", positive, negative, latent, denoise=1.0 + )[0] + return comfy_core_nodes.VAEDecode().decode(vae, sampled)[0] +``` + +- [ ] **Step 5: Run the focused runtime tests to verify they pass** + +Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_rejects_incompatible_model_name tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_runtime_executes_minimal_official_pipeline tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_runtime_executes_minimal_official_pipeline -v` +Expected: PASS with clear validation failures and both minimal pipelines returning decoded images through the official ComfyUI node wrappers. + +- [ ] **Step 6: Commit the runtime slice** + +```bash +git add qwen/runtime.py tests/test_qwen_nodes.py +git commit -m "feat: add qwen runtime wrappers" +``` + +### Task 3: Add user-facing nodes and register the new subpackage + +**Files:** +- Create: `qwen/__init__.py` +- Create: `qwen/nodes.py` +- Modify: `__init__.py` +- Modify: `tests/test_qwen_nodes.py` +- Test: `tests/test_qwen_nodes.py` + +- [ ] **Step 1: Write the failing node registration and schema tests** + +```python +def test_plugin_registers_qwen_nodes(self): + plugin = load_plugin_package() + self.assertIn("LLSQwenTextToImage", plugin.NODE_CLASS_MAPPINGS) + self.assertIn("LLSQwenImageEdit", plugin.NODE_CLASS_MAPPINGS) + + +def test_qwen_text_node_returns_image_only(self): + plugin = load_plugin_package() + node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenTextToImage"] + required = node_cls.INPUT_TYPES()["required"] + + self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",)) + self.assertEqual( + tuple(required.keys()), + ("model_name", "prompt", "width", "height", "steps", "seed", "batch_size"), + ) + + +def test_qwen_edit_node_returns_image_only(self): + plugin = load_plugin_package() + node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenImageEdit"] + required = node_cls.INPUT_TYPES()["required"] + + self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",)) + self.assertEqual(tuple(required.keys()), ("model_name", "image", "prompt", "steps", "seed")) +``` + +- [ ] **Step 2: Run the focused node tests to verify they fail** + +Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_plugin_registers_qwen_nodes tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_returns_image_only tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_returns_image_only -v` +Expected: FAIL because the Qwen subpackage is not yet registered and the node classes do not exist. + +- [ ] **Step 3: Write the minimal node layer** + +```python +class LLSQwenTextToImage: + CATEGORY = "LLS/Qwen" + FUNCTION = "generate" + RETURN_TYPES = ("IMAGE",) + RETURN_NAMES = ("image",) + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "model_name": (get_qwen_text_model_choices(),), + "prompt": ("STRING", {"default": "", "multiline": True}), + "width": ("INT", {"default": 1024, "min": 16, "max": 8192, "step": 16}), + "height": ("INT", {"default": 1024, "min": 16, "max": 8192, "step": 16}), + "steps": ("INT", {"default": 20, "min": 1, "max": 10000}), + "seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}), + "batch_size": ("INT", {"default": 1, "min": 1, "max": 64}), + } + } + + def generate(self, model_name, prompt, width, height, steps, seed, batch_size): + return (run_qwen_text_to_image(model_name, prompt, width, height, steps, seed, batch_size),) +``` + +- [ ] **Step 4: Register the subpackage** + +```python +_SUBPACKAGES: list[str] = [ + "model_loader", + "conditioning", + "sampling", + "qwen", + "latent", + ... +] +``` + +- [ ] **Step 5: Run the node tests to verify they pass** + +Run: `python3 -m unittest tests.test_qwen_nodes.TestQwenNodes.test_plugin_registers_qwen_nodes tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_returns_image_only tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_returns_image_only tests.test_qwen_nodes.TestQwenNodes.test_qwen_text_node_executes_runtime tests.test_qwen_nodes.TestQwenNodes.test_qwen_edit_node_executes_runtime -v` +Expected: PASS with both nodes registered, exposing only `IMAGE`, and delegating to the runtime layer. + +- [ ] **Step 6: Commit the node layer** + +```bash +git add __init__.py qwen/__init__.py qwen/nodes.py tests/test_qwen_nodes.py +git commit -m "feat: add minimal qwen image nodes" +``` + +### Task 4: Run the full regression suite for the plugin + +**Files:** +- Modify: `tests/test_qwen_nodes.py` (only if a verification run exposes a mismatch) +- Test: `tests/test_qwen_nodes.py` + +- [ ] **Step 1: Run the dedicated Qwen test file** + +Run: `python3 -m unittest tests.test_qwen_nodes -v` +Expected: PASS with zero failures for discovery, validation, runtime, registration, and execution-path coverage. + +- [ ] **Step 2: Run the full existing test suite** + +Run: `python3 -m unittest discover -s tests -v` +Expected: PASS with zero regressions in the existing loader/prompt/latent/upscale suites. + +- [ ] **Step 3: Review the changed files before closeout** + +```bash +git status --short +``` + +Expected: +- `M __init__.py` +- `A qwen/__init__.py` +- `A qwen/discovery.py` +- `A qwen/runtime.py` +- `A qwen/nodes.py` +- `A tests/test_qwen_nodes.py` +- `A docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md` + +- [ ] **Step 4: Commit the verified final state** + +```bash +git add __init__.py qwen tests/test_qwen_nodes.py docs/superpowers/plans/2026-05-20-qwen-minimal-nodes.md +git commit -m "test: verify minimal qwen nodes" +``` diff --git a/qwen/__init__.py b/qwen/__init__.py new file mode 100644 index 0000000..01e2c20 --- /dev/null +++ b/qwen/__init__.py @@ -0,0 +1,8 @@ +""" +node.qwen +========= +功能域：Qwen 一体化图像节点 +""" +from .nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS + +__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"] diff --git a/qwen/discovery.py b/qwen/discovery.py new file mode 100644 index 0000000..89bcf5e --- /dev/null +++ b/qwen/discovery.py @@ -0,0 +1,115 @@ +""" +Qwen model/resource discovery helpers. +""" +from __future__ import annotations + +try: + import folder_paths +except Exception: + folder_paths = None + + +TEXT_MODEL_PLACEHOLDER = "(no qwen text-to-image models found)" +EDIT_MODEL_PLACEHOLDER = "(no qwen image edit models found)" + +_QWEN_TEXT_ENCODER_PATTERNS = ( + "qwen_2.5_vl_7b_fp8_scaled.safetensors", + "qwen_2.5_vl_7b", + "qwen25_7b", +) +_QWEN_VAE_PATTERNS = ( + "qwen_image_vae.safetensors", + "qwen_image_vae", +) + + +def _get_filename_list(category: str) -> list[str]: + if folder_paths is None: + return [] + try: + return list(folder_paths.get_filename_list(category)) + except Exception: + return [] + + +def _sorted_unique(items: list[str]) -> list[str]: + return sorted(dict.fromkeys(item for item in items if item)) + + +def _match_by_patterns(names: list[str], patterns: tuple[str, ...]) -> str | None: + lowered = {name.lower(): name for name in names} + for pattern in patterns: + pattern_lower = pattern.lower() + for candidate_lower, candidate in lowered.items(): + if candidate_lower == pattern_lower or pattern_lower in candidate_lower: + return candidate + return None + + +def is_qwen_text_to_image_model(name: str | None) -> bool: + if not name: + return False + lowered = name.lower() + return ( + "qwen" in lowered + and "image" in lowered + and "edit" not in lowered + and "layered" not in lowered + ) + + +def is_qwen_image_edit_model(name: str | None) -> bool: + if not name: + return False + lowered = name.lower() + return "qwen" in lowered and "image" in lowered and "edit" in lowered and "layered" not in lowered + + +def get_qwen_text_model_choices() -> list[str]: + names = [name for name in _get_filename_list("diffusion_models") if is_qwen_text_to_image_model(name)] + return _sorted_unique(names) or [TEXT_MODEL_PLACEHOLDER] + + +def get_qwen_edit_model_choices() -> list[str]: + names = [name for name in _get_filename_list("diffusion_models") if is_qwen_image_edit_model(name)] + return _sorted_unique(names) or [EDIT_MODEL_PLACEHOLDER] + + +def validate_qwen_text_model_name(model_name: str) -> str: + if not is_qwen_text_to_image_model(model_name): + raise RuntimeError( + f"[LLS] Model '{model_name}' is not compatible with LLSQwenTextToImage." + ) + return model_name + + +def validate_qwen_edit_model_name(model_name: str) -> str: + if not is_qwen_image_edit_model(model_name): + raise RuntimeError( + f"[LLS] Model '{model_name}' is not compatible with LLSQwenImageEdit." + ) + return model_name + + +def resolve_qwen_text_encoder_name() -> str | None: + names = _get_filename_list("text_encoders") + matched = _match_by_patterns(names, _QWEN_TEXT_ENCODER_PATTERNS) + if matched is not None: + return matched + for name in names: + lowered = name.lower() + if "qwen" in lowered and ("vl" in lowered or "qwen25_7b" in lowered): + return name + return None + + +def resolve_qwen_vae_name() -> str | None: + names = _get_filename_list("vae") + matched = _match_by_patterns(names, _QWEN_VAE_PATTERNS) + if matched is not None: + return matched + for name in names: + lowered = name.lower() + if "qwen" in lowered and "vae" in lowered and "layered" not in lowered: + return name + return None diff --git a/qwen/nodes.py b/qwen/nodes.py new file mode 100644 index 0000000..8cbca57 --- /dev/null +++ b/qwen/nodes.py @@ -0,0 +1,114 @@ +""" +Minimal high-level Qwen nodes. +""" +from __future__ import annotations + +import importlib + +from . import discovery, runtime + +try: + comfy_core_nodes = importlib.import_module("nodes") +except Exception: + comfy_core_nodes = None + +_MAX_RESOLUTION = int(getattr(comfy_core_nodes, "MAX_RESOLUTION", 8192)) + + +class LLSQwenTextToImage: + CATEGORY = "LLS/Qwen" + FUNCTION = "generate" + RETURN_TYPES = ("IMAGE",) + RETURN_NAMES = ("image",) + DESCRIPTION = ( + "Minimal Qwen text-to-image node. Internally loads the Qwen model, text encoder, " + "and VAE, then runs the official ComfyUI Qwen generation flow." + ) + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "model_name": (discovery.get_qwen_text_model_choices(),), + "prompt": ("STRING", {"default": "", "multiline": True}), + "width": ("INT", {"default": 1024, "min": 16, "max": _MAX_RESOLUTION, "step": 16}), + "height": ("INT", {"default": 1024, "min": 16, "max": _MAX_RESOLUTION, "step": 16}), + "steps": ("INT", {"default": 20, "min": 1, "max": 10000}), + "seed": ("INT", {"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF}), + "batch_size": ("INT", {"default": 1, "min": 1, "max": 64}), + } + } + + def generate( + self, + model_name: str, + prompt: str, + width: int, + height: int, + steps: int, + seed: int, + batch_size: int, + ): + return ( + runtime.run_qwen_text_to_image( + model_name=model_name, + prompt=prompt, + width=width, + height=height, + steps=steps, + seed=seed, + batch_size=batch_size, + ), + ) + + +class LLSQwenImageEdit: + CATEGORY = "LLS/Qwen" + FUNCTION = "generate" + RETURN_TYPES = ("IMAGE",) + RETURN_NAMES = ("image",) + DESCRIPTION = ( + "Minimal Qwen image-edit node. Internally loads the Qwen edit model, text encoder, " + "and VAE, then runs the official ComfyUI Qwen image-edit flow." + ) + + @classmethod + def INPUT_TYPES(cls): + return { + "required": { + "model_name": (discovery.get_qwen_edit_model_choices(),), + "image": ("IMAGE",), + "prompt": ("STRING", {"default": "", "multiline": True}), + "steps": ("INT", {"default": 20, "min": 1, "max": 10000}), + "seed": ("INT", {"default": 0, "min": 0, "max": 0xFFFFFFFFFFFFFFFF}), + } + } + + def generate( + self, + model_name: str, + image, + prompt: str, + steps: int, + seed: int, + ): + return ( + runtime.run_qwen_image_edit( + model_name=model_name, + image=image, + prompt=prompt, + steps=steps, + seed=seed, + ), + ) + + +NODE_CLASS_MAPPINGS: dict[str, type] = { + "LLSQwenTextToImage": LLSQwenTextToImage, + "LLSQwenImageEdit": LLSQwenImageEdit, +} + +NODE_DISPLAY_NAME_MAPPINGS: dict[str, str] = { + "LLSQwenTextToImage": "LLS Qwen Text To Image", + "LLSQwenImageEdit": "LLS Qwen Image Edit", +} diff --git a/qwen/runtime.py b/qwen/runtime.py new file mode 100644 index 0000000..58205a2 --- /dev/null +++ b/qwen/runtime.py @@ -0,0 +1,266 @@ +""" +Runtime wrappers for official ComfyUI Qwen image pipelines. +""" +from __future__ import annotations + +import importlib +from typing import Any + +try: + import folder_paths +except Exception: + folder_paths = None + +try: + comfy_core_nodes = importlib.import_module("nodes") +except Exception as exc: + comfy_core_nodes = None + _CORE_NODES_ERR = exc +else: + _CORE_NODES_ERR = None + +try: + nodes_sd3 = importlib.import_module("comfy_extras.nodes_sd3") +except Exception as exc: + nodes_sd3 = None + _SD3_ERR = exc +else: + _SD3_ERR = None + +try: + nodes_model_advanced = importlib.import_module("comfy_extras.nodes_model_advanced") +except Exception as exc: + nodes_model_advanced = None + _MODEL_ADVANCED_ERR = exc +else: + _MODEL_ADVANCED_ERR = None + +try: + nodes_qwen = importlib.import_module("comfy_extras.nodes_qwen") +except Exception as exc: + nodes_qwen = None + _QWEN_ERR = exc +else: + _QWEN_ERR = None + +try: + nodes_flux = importlib.import_module("comfy_extras.nodes_flux") +except Exception as exc: + nodes_flux = None + _FLUX_ERR = exc +else: + _FLUX_ERR = None + +try: + nodes_cfg = importlib.import_module("comfy_extras.nodes_cfg") +except Exception as exc: + nodes_cfg = None + _CFG_ERR = exc +else: + _CFG_ERR = None + +from . import discovery + + +_QWEN_CLIP_TYPE = "qwen_image" +_QWEN_MODEL_SHIFT = 3.1 +_QWEN_CFG = 4.0 +_QWEN_SAMPLER = "euler" +_QWEN_SCHEDULER = "simple" +_QWEN_CFG_NORM_STRENGTH = 1.0 +_QWEN_REFERENCE_LATENTS_METHOD = "index_timestep_zero" + + +def _unwrap_first(result: Any): + if hasattr(result, "result"): + values = result.result + if values is None: + return None + return values[0] + if isinstance(result, tuple): + return result[0] + if isinstance(result, list): + return result[0] + return result + + +def _require_module(module: Any, label: str, import_error: Exception | None = None): + if module is None: + raise RuntimeError( + f"[LLS] Required ComfyUI runtime component '{label}' is unavailable. " + "Use a ComfyUI build that includes official Qwen image support." + ) from import_error + return module + + +def _require_class(module: Any, class_name: str, import_error: Exception | None = None): + _require_module(module, class_name, import_error) + cls = getattr(module, class_name, None) + if cls is None: + raise RuntimeError( + f"[LLS] Required ComfyUI runtime component '{class_name}' is unavailable. " + "Use a ComfyUI build that includes official Qwen image support." + ) from import_error + return cls + + +def _load_qwen_clip(clip_name: str): + clip_loader_cls = _require_class(comfy_core_nodes, "CLIPLoader", _CORE_NODES_ERR) + loader = clip_loader_cls() + try: + return loader.load_clip(clip_name, type=_QWEN_CLIP_TYPE, device="default")[0] + except TypeError: + return loader.load_clip(clip_name, type=_QWEN_CLIP_TYPE)[0] + + +def _load_qwen_vae(vae_name: str): + vae_loader_cls = _require_class(comfy_core_nodes, "VAELoader", _CORE_NODES_ERR) + return vae_loader_cls().load_vae(vae_name)[0] + + +def _load_qwen_model(model_name: str): + unet_loader_cls = _require_class(comfy_core_nodes, "UNETLoader", _CORE_NODES_ERR) + return unet_loader_cls().load_unet(model_name, "default")[0] + + +def _patch_qwen_model_sampling(model): + patcher_cls = _require_class(nodes_model_advanced, "ModelSamplingAuraFlow", _MODEL_ADVANCED_ERR) + return patcher_cls().patch_aura(model, _QWEN_MODEL_SHIFT)[0] + + +def _apply_cfg_norm(model): + cfg_cls = _require_class(nodes_cfg, "CFGNorm", _CFG_ERR) + return _unwrap_first(cfg_cls.execute(model, _QWEN_CFG_NORM_STRENGTH)) + + +def _encode_clip_text(clip, prompt: str): + encoder_cls = _require_class(comfy_core_nodes, "CLIPTextEncode", _CORE_NODES_ERR) + return encoder_cls().encode(clip, prompt)[0] + + +def _create_empty_qwen_latent(width: int, height: int, batch_size: int): + latent_cls = _require_class(nodes_sd3, "EmptySD3LatentImage", _SD3_ERR) + latent_node = latent_cls() + if hasattr(latent_node, "generate"): + return latent_node.generate(width, height, batch_size)[0] + return latent_node.execute(width, height, batch_size)[0] + + +def _encode_qwen_edit_conditioning(clip, prompt: str, vae, image): + encoder_cls = _require_class(nodes_qwen, "TextEncodeQwenImageEditPlus", _QWEN_ERR) + return _unwrap_first(encoder_cls.execute(clip, prompt, vae=vae, image1=image)) + + +def _scale_qwen_edit_image(image): + scaler_cls = _require_class(nodes_flux, "FluxKontextImageScale", _FLUX_ERR) + return _unwrap_first(scaler_cls.execute(image)) + + +def _apply_reference_latents_method(conditioning): + method_cls = _require_class(nodes_flux, "FluxKontextMultiReferenceLatentMethod", _FLUX_ERR) + return _unwrap_first(method_cls.execute(conditioning, _QWEN_REFERENCE_LATENTS_METHOD)) + + +def _encode_image_to_latent(vae, image): + encoder_cls = _require_class(comfy_core_nodes, "VAEEncode", _CORE_NODES_ERR) + return encoder_cls().encode(vae, image)[0] + + +def _sample_qwen(model, positive, negative, latent, steps: int, seed: int): + sampler_cls = _require_class(comfy_core_nodes, "KSampler", _CORE_NODES_ERR) + return sampler_cls().sample( + model, + int(seed), + int(steps), + _QWEN_CFG, + _QWEN_SAMPLER, + _QWEN_SCHEDULER, + positive, + negative, + latent, + denoise=1.0, + )[0] + + +def _decode_qwen_latent(vae, latent): + decoder_cls = _require_class(comfy_core_nodes, "VAEDecode", _CORE_NODES_ERR) + return decoder_cls().decode(vae, latent)[0] + + +def _resolve_qwen_resources(model_name: str, validator): + validator(model_name) + clip_name = discovery.resolve_qwen_text_encoder_name() + vae_name = discovery.resolve_qwen_vae_name() + + if clip_name is None: + raise RuntimeError( + "[LLS] Missing Qwen text encoder. Place a compatible Qwen VL text encoder in " + "ComfyUI/models/text_encoders/." + ) + if vae_name is None: + raise RuntimeError( + "[LLS] Missing Qwen VAE. Place qwen_image_vae.safetensors in ComfyUI/models/vae/." + ) + + return model_name, clip_name, vae_name + + +def run_qwen_text_to_image( + model_name: str, + prompt: str, + width: int, + height: int, + steps: int, + seed: int, + batch_size: int, +): + try: + model_name, clip_name, vae_name = _resolve_qwen_resources( + model_name, + discovery.validate_qwen_text_model_name, + ) + model = _patch_qwen_model_sampling(_load_qwen_model(model_name)) + clip = _load_qwen_clip(clip_name) + vae = _load_qwen_vae(vae_name) + positive = _encode_clip_text(clip, prompt) + negative = _encode_clip_text(clip, "") + latent = _create_empty_qwen_latent(int(width), int(height), int(batch_size)) + sampled = _sample_qwen(model, positive, negative, latent, int(steps), int(seed)) + return _decode_qwen_latent(vae, sampled) + except Exception as exc: + if str(exc).startswith("[LLS]"): + raise + raise RuntimeError(f"[LLS] Qwen text-to-image failed: {exc}") from exc + + +def run_qwen_image_edit( + model_name: str, + image, + prompt: str, + steps: int, + seed: int, +): + try: + if image is None: + raise RuntimeError("[LLS] Missing source image. Connect an IMAGE input.") + + model_name, clip_name, vae_name = _resolve_qwen_resources( + model_name, + discovery.validate_qwen_edit_model_name, + ) + model = _patch_qwen_model_sampling(_load_qwen_model(model_name)) + clip = _load_qwen_clip(clip_name) + vae = _load_qwen_vae(vae_name) + scaled = _scale_qwen_edit_image(image) + positive = _encode_qwen_edit_conditioning(clip, prompt, vae, scaled) + negative = _encode_qwen_edit_conditioning(clip, "", vae, scaled) + positive = _apply_reference_latents_method(positive) + negative = _apply_reference_latents_method(negative) + latent = _encode_image_to_latent(vae, scaled) + model = _apply_cfg_norm(model) + sampled = _sample_qwen(model, positive, negative, latent, int(steps), int(seed)) + return _decode_qwen_latent(vae, sampled) + except Exception as exc: + if str(exc).startswith("[LLS]"): + raise + raise RuntimeError(f"[LLS] Qwen image edit failed: {exc}") from exc diff --git a/tests/test_qwen_nodes.py b/tests/test_qwen_nodes.py new file mode 100644 index 0000000..fa42b1b --- /dev/null +++ b/tests/test_qwen_nodes.py @@ -0,0 +1,535 @@ +import importlib.util +import pathlib +import sys +import unittest +from unittest import mock + + +ROOT = pathlib.Path(__file__).resolve().parents[1] + + +def load_plugin_package(): + for name in list(sys.modules): + if name == "lls_node_test_qwen" or name.startswith("lls_node_test_qwen."): + sys.modules.pop(name) + + spec = importlib.util.spec_from_file_location( + "lls_node_test_qwen", + ROOT / "__init__.py", + submodule_search_locations=[str(ROOT)], + ) + module = importlib.util.module_from_spec(spec) + sys.modules["lls_node_test_qwen"] = module + spec.loader.exec_module(module) + return module + + +class NodeOutputStub: + def __init__(self, *args): + self.args = args + + @property + def result(self): + return self.args + + def __getitem__(self, index): + return self.args[index] + + +class QwenFolderPathsStub: + def __init__(self): + self._files = { + "diffusion_models": [ + "qwen_image_fp8_e4m3fn.safetensors", + "qwen_image_2512_fp8_e4m3fn.safetensors", + "qwen_image_edit_2511_bf16.safetensors", + "qwen_image_layered_bf16.safetensors", + "flux1-dev.safetensors", + ], + "text_encoders": ["qwen_2.5_vl_7b_fp8_scaled.safetensors"], + "vae": ["qwen_image_vae.safetensors"], + } + + def get_filename_list(self, category): + return list(self._files.get(category, [])) + + +class QwenClipStub: + def __init__(self, clip_name, clip_type): + self.clip_name = clip_name + self.clip_type = clip_type + + +class QwenVAEStub: + def __init__(self, vae_name): + self.vae_name = vae_name + + +class CoreQwenNodesStub: + last_unet_call = None + last_clip_call = None + last_text_encode_calls = [] + last_ksampler_call = None + last_vae_encode_call = None + last_vae_decode_call = None + last_vae_load_call = None + + @classmethod + def reset(cls): + cls.last_unet_call = None + cls.last_clip_call = None + cls.last_text_encode_calls = [] + cls.last_ksampler_call = None + cls.last_vae_encode_call = None + cls.last_vae_decode_call = None + cls.last_vae_load_call = None + + class UNETLoader: + def load_unet(self, unet_name, weight_dtype): + CoreQwenNodesStub.last_unet_call = { + "unet_name": unet_name, + "weight_dtype": weight_dtype, + } + return (f"MODEL::{unet_name}",) + + class CLIPLoader: + def load_clip(self, clip_name, type="stable_diffusion", device="default"): + CoreQwenNodesStub.last_clip_call = { + "clip_name": clip_name, + "type": type, + "device": device, + } + return (QwenClipStub(clip_name, type),) + + class VAELoader: + def load_vae(self, vae_name): + CoreQwenNodesStub.last_vae_load_call = {"vae_name": vae_name} + return (QwenVAEStub(vae_name),) + + class CLIPTextEncode: + def encode(self, clip, text): + CoreQwenNodesStub.last_text_encode_calls.append( + { + "clip_name": clip.clip_name, + "clip_type": clip.clip_type, + "text": text, + } + ) + return (f"COND::{text}",) + + class KSampler: + def sample( + self, + model, + seed, + steps, + cfg, + sampler_name, + scheduler, + positive, + negative, + latent_image, + denoise=1.0, + ): + CoreQwenNodesStub.last_ksampler_call = { + "model": model, + "seed": seed, + "steps": steps, + "cfg": cfg, + "sampler_name": sampler_name, + "scheduler": scheduler, + "positive": positive, + "negative": negative, + "latent_image": latent_image, + "denoise": denoise, + } + return ({"samples": f"SAMPLED::{seed}::{steps}"},) + + class VAEDecode: + def decode(self, vae, samples): + CoreQwenNodesStub.last_vae_decode_call = { + "vae_name": vae.vae_name, + "samples": samples, + } + return (f"IMAGE::{samples['samples']}",) + + class VAEEncode: + def encode(self, vae, pixels): + CoreQwenNodesStub.last_vae_encode_call = { + "vae_name": vae.vae_name, + "pixels": pixels, + } + return ({"samples": f"ENCODED::{pixels}"},) + + +class SD3NodesStub: + last_generate_call = None + + @classmethod + def reset(cls): + cls.last_generate_call = None + + class EmptySD3LatentImage: + def generate(self, width, height, batch_size=1): + SD3NodesStub.last_generate_call = { + "width": width, + "height": height, + "batch_size": batch_size, + } + return ( + { + "samples": f"EMPTY::{width}x{height}::{batch_size}", + "downscale_ratio_spacial": 8, + }, + ) + + +class ModelAdvancedStub: + last_call = None + + @classmethod + def reset(cls): + cls.last_call = None + + class ModelSamplingAuraFlow: + def patch_aura(self, model, shift): + ModelAdvancedStub.last_call = {"model": model, "shift": shift} + return (f"AURA::{model}::{shift}",) + + +class FluxNodesStub: + last_scale_call = None + last_reference_calls = [] + + @classmethod + def reset(cls): + cls.last_scale_call = None + cls.last_reference_calls = [] + + class FluxKontextImageScale: + @classmethod + def execute(cls, image): + FluxNodesStub.last_scale_call = {"image": image} + return NodeOutputStub(f"SCALED::{image}") + + class FluxKontextMultiReferenceLatentMethod: + @classmethod + def execute(cls, conditioning, reference_latents_method): + FluxNodesStub.last_reference_calls.append( + { + "conditioning": conditioning, + "reference_latents_method": reference_latents_method, + } + ) + return NodeOutputStub(f"{conditioning}::{reference_latents_method}") + + +class QwenExtraNodesStub: + last_calls = [] + + @classmethod + def reset(cls): + cls.last_calls = [] + + class TextEncodeQwenImageEditPlus: + @classmethod + def execute(cls, clip, prompt, vae=None, image1=None, image2=None, image3=None): + QwenExtraNodesStub.last_calls.append( + { + "clip_name": clip.clip_name, + "prompt": prompt, + "vae_name": getattr(vae, "vae_name", None), + "image1": image1, + "image2": image2, + "image3": image3, + } + ) + return NodeOutputStub(f"EDIT_COND::{prompt}::{image1}") + + +class CFGNodesStub: + last_call = None + + @classmethod + def reset(cls): + cls.last_call = None + + class CFGNorm: + @classmethod + def execute(cls, model, strength): + CFGNodesStub.last_call = {"model": model, "strength": strength} + return NodeOutputStub(f"CFGNORM::{model}::{strength}") + + +class TestQwenNodes(unittest.TestCase): + def setUp(self): + CoreQwenNodesStub.reset() + SD3NodesStub.reset() + ModelAdvancedStub.reset() + FluxNodesStub.reset() + QwenExtraNodesStub.reset() + CFGNodesStub.reset() + + def test_plugin_registers_qwen_nodes(self): + plugin = load_plugin_package() + + self.assertIn("LLSQwenTextToImage", plugin.NODE_CLASS_MAPPINGS) + self.assertIn("LLSQwenImageEdit", plugin.NODE_CLASS_MAPPINGS) + + def test_qwen_text_node_filters_only_text_models(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import nodes as qwen_nodes + + with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()): + choices = qwen_nodes.LLSQwenTextToImage.INPUT_TYPES()["required"]["model_name"][0] + + self.assertEqual( + choices, + [ + "qwen_image_2512_fp8_e4m3fn.safetensors", + "qwen_image_fp8_e4m3fn.safetensors", + ], + ) + + def test_qwen_edit_node_filters_only_edit_models(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import nodes as qwen_nodes + + with mock.patch.object(qwen_discovery, "folder_paths", QwenFolderPathsStub()): + choices = qwen_nodes.LLSQwenImageEdit.INPUT_TYPES()["required"]["model_name"][0] + + self.assertEqual(choices, ["qwen_image_edit_2511_bf16.safetensors"]) + + def test_qwen_text_node_uses_placeholder_when_no_compatible_models_exist(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import nodes as qwen_nodes + + stub = QwenFolderPathsStub() + stub._files["diffusion_models"] = ["flux1-dev.safetensors", "sdxl.safetensors"] + with mock.patch.object(qwen_discovery, "folder_paths", stub): + choices = qwen_nodes.LLSQwenTextToImage.INPUT_TYPES()["required"]["model_name"][0] + + self.assertEqual(choices, ["(no qwen text-to-image models found)"]) + + def test_qwen_text_runtime_rejects_incompatible_model_name(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + stub = QwenFolderPathsStub() + with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object( + qwen_runtime, + "folder_paths", + stub, + ): + with self.assertRaisesRegex(RuntimeError, "not compatible with LLSQwenTextToImage"): + qwen_runtime.run_qwen_text_to_image( + model_name="qwen_image_edit_2511_bf16.safetensors", + prompt="a cat", + width=1024, + height=1024, + steps=20, + seed=1, + batch_size=1, + ) + + def test_qwen_edit_runtime_rejects_incompatible_model_name(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + stub = QwenFolderPathsStub() + with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object( + qwen_runtime, + "folder_paths", + stub, + ): + with self.assertRaisesRegex(RuntimeError, "not compatible with LLSQwenImageEdit"): + qwen_runtime.run_qwen_image_edit( + model_name="qwen_image_fp8_e4m3fn.safetensors", + image="IMAGE::input", + prompt="edit the cat", + steps=20, + seed=1, + ) + + def test_qwen_text_runtime_raises_when_qwen_text_encoder_is_missing(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + stub = QwenFolderPathsStub() + stub._files["text_encoders"] = [] + with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object( + qwen_runtime, + "folder_paths", + stub, + ): + with self.assertRaisesRegex(RuntimeError, "Missing Qwen text encoder"): + qwen_runtime.run_qwen_text_to_image( + model_name="qwen_image_fp8_e4m3fn.safetensors", + prompt="a cat", + width=1024, + height=1024, + steps=20, + seed=1, + batch_size=1, + ) + + def test_qwen_text_runtime_raises_when_qwen_vae_is_missing(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + stub = QwenFolderPathsStub() + stub._files["vae"] = [] + with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object( + qwen_runtime, + "folder_paths", + stub, + ): + with self.assertRaisesRegex(RuntimeError, "Missing Qwen VAE"): + qwen_runtime.run_qwen_text_to_image( + model_name="qwen_image_fp8_e4m3fn.safetensors", + prompt="a cat", + width=1024, + height=1024, + steps=20, + seed=1, + batch_size=1, + ) + + def test_qwen_text_runtime_executes_minimal_official_pipeline(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + stub = QwenFolderPathsStub() + with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object( + qwen_runtime, + "folder_paths", + stub, + ), mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), mock.patch.object( + qwen_runtime, + "nodes_sd3", + SD3NodesStub, + ), mock.patch.object(qwen_runtime, "nodes_model_advanced", ModelAdvancedStub): + image = qwen_runtime.run_qwen_text_to_image( + model_name="qwen_image_fp8_e4m3fn.safetensors", + prompt="a cat", + width=1024, + height=1152, + steps=20, + seed=99, + batch_size=2, + ) + + self.assertEqual(image, "IMAGE::SAMPLED::99::20") + self.assertEqual(CoreQwenNodesStub.last_clip_call["type"], "qwen_image") + self.assertEqual(SD3NodesStub.last_generate_call, {"width": 1024, "height": 1152, "batch_size": 2}) + self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 4.0) + self.assertEqual(CoreQwenNodesStub.last_ksampler_call["sampler_name"], "euler") + self.assertEqual(CoreQwenNodesStub.last_ksampler_call["scheduler"], "simple") + self.assertEqual(CoreQwenNodesStub.last_text_encode_calls[0]["text"], "a cat") + self.assertEqual(CoreQwenNodesStub.last_text_encode_calls[1]["text"], "") + self.assertEqual(ModelAdvancedStub.last_call["shift"], 3.1) + + def test_qwen_edit_runtime_executes_minimal_official_pipeline(self): + load_plugin_package() + from lls_node_test_qwen.qwen import discovery as qwen_discovery + from lls_node_test_qwen.qwen import runtime as qwen_runtime + + stub = QwenFolderPathsStub() + with mock.patch.object(qwen_discovery, "folder_paths", stub), mock.patch.object( + qwen_runtime, + "folder_paths", + stub, + ), mock.patch.object(qwen_runtime, "comfy_core_nodes", CoreQwenNodesStub), mock.patch.object( + qwen_runtime, + "nodes_model_advanced", + ModelAdvancedStub, + ), mock.patch.object(qwen_runtime, "nodes_qwen", QwenExtraNodesStub), mock.patch.object( + qwen_runtime, + "nodes_flux", + FluxNodesStub, + ), mock.patch.object(qwen_runtime, "nodes_cfg", CFGNodesStub): + image = qwen_runtime.run_qwen_image_edit( + model_name="qwen_image_edit_2511_bf16.safetensors", + image="IMAGE::input", + prompt="turn the cat blue", + steps=30, + seed=123, + ) + + self.assertEqual(image, "IMAGE::SAMPLED::123::30") + self.assertEqual(FluxNodesStub.last_scale_call, {"image": "IMAGE::input"}) + self.assertEqual(QwenExtraNodesStub.last_calls[0]["prompt"], "turn the cat blue") + self.assertEqual(QwenExtraNodesStub.last_calls[0]["image1"], "SCALED::IMAGE::input") + self.assertEqual(QwenExtraNodesStub.last_calls[1]["prompt"], "") + self.assertEqual(len(FluxNodesStub.last_reference_calls), 2) + self.assertEqual(FluxNodesStub.last_reference_calls[0]["reference_latents_method"], "index_timestep_zero") + self.assertEqual(CFGNodesStub.last_call["strength"], 1.0) + self.assertEqual(CoreQwenNodesStub.last_vae_encode_call["pixels"], "SCALED::IMAGE::input") + self.assertEqual(CoreQwenNodesStub.last_ksampler_call["cfg"], 4.0) + + def test_qwen_text_node_returns_image_only(self): + plugin = load_plugin_package() + node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenTextToImage"] + required = node_cls.INPUT_TYPES()["required"] + + self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",)) + self.assertEqual(node_cls.RETURN_NAMES, ("image",)) + self.assertEqual( + tuple(required.keys()), + ("model_name", "prompt", "width", "height", "steps", "seed", "batch_size"), + ) + + def test_qwen_edit_node_returns_image_only(self): + plugin = load_plugin_package() + node_cls = plugin.NODE_CLASS_MAPPINGS["LLSQwenImageEdit"] + required = node_cls.INPUT_TYPES()["required"] + + self.assertEqual(node_cls.RETURN_TYPES, ("IMAGE",)) + self.assertEqual(node_cls.RETURN_NAMES, ("image",)) + self.assertEqual(tuple(required.keys()), ("model_name", "image", "prompt", "steps", "seed")) + + def test_qwen_text_node_executes_runtime(self): + load_plugin_package() + from lls_node_test_qwen.qwen import nodes as qwen_nodes + node = qwen_nodes.LLSQwenTextToImage() + + with mock.patch.object(qwen_nodes.runtime, "run_qwen_text_to_image", return_value="IMAGE::node"): + result = node.generate( + model_name="qwen_image_fp8_e4m3fn.safetensors", + prompt="a cat", + width=1024, + height=1024, + steps=20, + seed=1, + batch_size=1, + ) + + self.assertEqual(result, ("IMAGE::node",)) + + def test_qwen_edit_node_executes_runtime(self): + load_plugin_package() + from lls_node_test_qwen.qwen import nodes as qwen_nodes + node = qwen_nodes.LLSQwenImageEdit() + + with mock.patch.object(qwen_nodes.runtime, "run_qwen_image_edit", return_value="IMAGE::node-edit"): + result = node.generate( + model_name="qwen_image_edit_2511_bf16.safetensors", + image="IMAGE::input", + prompt="edit the cat", + steps=20, + seed=1, + ) + + self.assertEqual(result, ("IMAGE::node-edit",)) + + +if __name__ == "__main__": + unittest.main()