From 41f8f9f77792cb4ec9bb84e3cbb2895c4e694815 Mon Sep 17 00:00:00 2001 From: magicwang1111 <131337141+magicwang1111@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:58:05 +0800 Subject: [PATCH] Upload original reference images without loading tensors --- README.md | 6 + py/image_files.py | 83 +++++++++ py/nodes.py | 27 ++- py/nodes_v4.py | 99 ++++++++--- tests/test_original_image_files.py | 264 +++++++++++++++++++++++++++++ 5 files changed, 449 insertions(+), 30 deletions(-) create mode 100644 py/image_files.py create mode 100644 tests/test_original_image_files.py diff --git a/README.md b/README.md index cdf0f08..7cbffc6 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,12 @@ Visible video generation modes: Documented newer models such as `kling-image-o1` and `kling-video-o1` still exist in the internal capability map, but are hidden from the UI until live endpoint support is confirmed for normal accounts. +## Original image uploads + +`Advanced Element Create` and the Kling 4.0 image/video nodes can use original JPEG/PNG files from `LoadImage`, including equal-size `Batch Images` inputs, without decoding them into tensors. Existing inputs and workflow connections stay the same. Subject images upload as files; Kling 4.0 sends the original file bytes as base64. + +This applies to RGB images in the current input directory. Rotated EXIF images, PNGs with EXIF, transparent or animated images, mixed-size batches, and processed images keep the existing tensor path. If another branch needs the image tensor, that branch still loads it. + ## Important limitations - Native 4K video output is exposed through the video `mode` dropdown. Models that do not support `mode=4k` will be rejected by the node capability checks before submission. diff --git a/py/image_files.py b/py/image_files.py new file mode 100644 index 0000000..b0adefc --- /dev/null +++ b/py/image_files.py @@ -0,0 +1,83 @@ +from pathlib import Path +import re + +import folder_paths +from PIL import Image + + +def _plain_png(path): + # PNG EXIF can follow IDAT. Pillow.getexif() decodes the image to find it; + # skip compressed chunks instead, and leave such files on the tensor path. + with open(path, "rb") as source: + source.seek(8) + while header := source.read(8): + if len(header) != 8: + return False + length, kind = int.from_bytes(header[:4], "big"), header[4:] + if kind in {b"eXIf", b"acTL", b"tRNS"}: + return False + if kind == b"IEND": + return True + source.seek(length + 4, 1) + return False + + +def _source_files(dynprompt, link, visiting): + if not isinstance(link, list) or len(link) != 2 or link[1] != 0: + return None + node_id = link[0] + if node_id in visiting or not dynprompt.has_node(node_id): + return None + node = dynprompt.get_node(node_id) + inputs = node["inputs"] + if node["class_type"] == "LoadImage" and set(inputs) == {"image"}: + filename = inputs["image"] + if not isinstance(filename, str): + return None + root = Path(folder_paths.get_input_directory()).resolve() + filename, annotated_root = folder_paths.annotated_filepath(filename) + if annotated_root is not None and Path(annotated_root).resolve() != root: + return None + path = (root / filename).resolve() + if Path(filename).is_absolute() or not path.is_relative_to(root): + raise ValueError("Reference image must be inside the current input directory.") + try: + with Image.open(path) as image: + if image.format not in {"JPEG", "PNG"} or image.mode != "RGB": + return None + if image.format == "JPEG" and image.getexif().get(274, 1) != 1: + return None + if image.format == "PNG" and not _plain_png(path): + return None + return [(str(path), image.size)] + except (OSError, Image.DecompressionBombError): + return None + + if node["class_type"] != "BatchImagesNode" or not inputs: + return None + if any(re.fullmatch(r"images\.image\d+", key) is None for key in inputs): + return None + files = [] + for key in sorted(inputs, key=lambda key: int(key.removeprefix("images.image"))): + sources = _source_files(dynprompt, inputs[key], visiting | {node_id}) + if sources is None: + return None + files.extend(sources) + # BatchImages resizes to its first input. Only bypass a batch that would + # concatenate unchanged RGB images of equal dimensions. + if any(size != files[0][1] for _, size in files): + return None + return files + + +def original_image_files(dynprompt, unique_id, input_name): + if dynprompt is None or unique_id is None: + return None + link = dynprompt.get_node(unique_id)["inputs"].get(input_name) + sources = _source_files(dynprompt, link, set()) + return [path for path, _ in sources] if sources is not None else None + + +def required_image_inputs(dynprompt, unique_id, images): + return [name for name, image in images.items() + if image is None and original_image_files(dynprompt, unique_id, name) is None] diff --git a/py/nodes.py b/py/nodes.py index bd4233d..24558e3 100644 --- a/py/nodes.py +++ b/py/nodes.py @@ -9,6 +9,7 @@ from .api.capabilities import CUSTOM_VOICE_LIST_TYPE, DEFAULT_IMAGE_RESOLUTIONS, VIDEO_GENERATION_MODES, get_image_capability, get_video_capability, validate_image_generation_request, \ validate_video_generation_request from .api.exceptions import KLingAPIError +from .image_files import original_image_files, required_image_inputs from .api.providers import ( OFFICIAL_PROVIDER, TENCENT_VOD_PROVIDER, @@ -3494,13 +3495,14 @@ class AdvancedCustomElementCreateNode: "element_description": ("STRING", {"multiline": True, "default": ""}), }, "optional": { - "image": ("IMAGE",), - "image_list": ("IMAGE",), + "image": ("IMAGE", {"lazy": True}), + "image_list": ("IMAGE", {"lazy": True}), "video_url": ("STRING", {"multiline": False, "default": ""}), "element_voice_id": ("STRING", {"multiline": False, "default": ""}), "extra_payload_json": ("STRING", {"multiline": True, "default": ""}), "reference_video_input": ("VIDEO", {"tooltip": "连接 Load Video，主体资产视频不能超过 8 秒；超长将提示更换，不自动截断。video_character 模式下优先使用此输入，否则使用 video_url。"}), - } + }, + "hidden": {"dynprompt": "DYNPROMPT", "unique_id": "UNIQUE_ID"}, } RETURN_TYPES = (ELEMENT_TYPE, "STRING", "STRING") @@ -3509,6 +3511,10 @@ class AdvancedCustomElementCreateNode: OUTPUT_NODE = False CATEGORY = NODE_CATEGORY + def check_lazy_status(self, dynprompt=None, unique_id=None, **kwargs): + images = {name: kwargs[name] for name in ("image", "image_list") if name in kwargs} + return required_image_inputs(dynprompt, unique_id, images) + def create( self, element_type, @@ -3521,6 +3527,8 @@ class AdvancedCustomElementCreateNode: element_voice_id="", extra_payload_json="", reference_video_input=None, + dynprompt=None, + unique_id=None, ): provider = _resolve_video_provider(provider) generator = AdvancedCustomElements() @@ -3541,10 +3549,15 @@ class AdvancedCustomElementCreateNode: generator.reference_type = "video_refer" merged_image_list = [] - if image is not None: - merged_image_list.extend(_upload_image_batch_references(image)) - if image_list is not None: - merged_image_list.extend(_upload_image_batch_references(image_list)) + for name, value in (("image", image), ("image_list", image_list)): + if value is not None: + merged_image_list.extend(_upload_image_batch_references(value)) + else: + files = original_image_files(dynprompt, unique_id, name) + if files is not None: + merged_image_list.extend(_upload_file_to_temporary_media_host(path) for path in files) + elif dynprompt is not None and name in dynprompt.get_node(unique_id)["inputs"]: + raise ValueError("Reference image changed before upload; run the workflow again.") if element_type in ("image_subject", "multi_image_subject"): # Kling's advanced element API requires one frontal image plus 1-3 diff --git a/py/nodes_v4.py b/py/nodes_v4.py index 5d9c77a..c437213 100644 --- a/py/nodes_v4.py +++ b/py/nodes_v4.py @@ -1,5 +1,6 @@ from urllib.parse import urlsplit from pathlib import Path +import base64 import re import tempfile @@ -12,6 +13,29 @@ from .api.video_v4 import (VIDEO_ELEMENT_TYPES, build_payload, known_duration, r validate_video_metadata) from .nodes import (_image_batch_to_base64_list, _normalize_element_list, _normalize_voice_list, _runtime_client, _upload_file_to_temporary_media_host) +from .image_files import original_image_files, required_image_inputs + + +def _image_inputs(values, lazy=False): + images = {name: value for name, value in values.items() + if name in {"first_frame", "middle_frames", "last_frame", "reference_images"} + or re.fullmatch(r"image_\d+", name)} + middle = values.get("middle_frame") + if lazy and isinstance(middle, tuple): + middle = middle[0] + for name, value in (middle or {}).items(): + if lazy: + image, key = value + images[key] = image + else: + images[f"middle_frame.{name}"] = value + return images + + +def _encode_images(images, files=None): + if files is not None: + return [base64.b64encode(Path(path).read_bytes()).decode("ascii") for path in files] + return _image_batch_to_base64_list(images) def _common_inputs(): @@ -37,24 +61,31 @@ def _frame_schema_inputs(): io.Boolean.Input("watermark", default=False, optional=True), io.String.Input("callback_url", default="", optional=True), io.String.Input("external_task_id", default="", optional=True), - io.Image.Input("first_frame", optional=True, tooltip="单张首帧，所有图片按首帧→中间帧→尾帧→参考图统一编号。"), - io.Image.Input("middle_frames", optional=True, tooltip="旧工作流的中间帧批次，排在动态中间帧之前。"), + io.Image.Input("first_frame", optional=True, lazy=True, tooltip="单张首帧，所有图片按首帧→中间帧→尾帧→参考图统一编号。"), + io.Image.Input("middle_frames", optional=True, lazy=True, tooltip="旧工作流的中间帧批次，排在动态中间帧之前。"), io.Autogrow.Input("middle_frame", optional=True, template=io.Autogrow.TemplateNames( - io.Image.Input("frame", tooltip="中间帧，连接后自动增加接口；支持单张或批次，按接口顺序编号。"), + io.Image.Input("frame", lazy=True, tooltip="中间帧，连接后自动增加接口；支持单张或批次，按接口顺序编号。"), names=[f"middle_frame_{n}" for n in range(1, 11)], min=0)), - io.Image.Input("last_frame", optional=True, tooltip="单张尾帧，接在所有中间帧之后编号。"), + io.Image.Input("last_frame", optional=True, lazy=True, tooltip="单张尾帧，接在所有中间帧之后编号。"), ] -def _frames(first_frame, middle_frames, last_frame, middle_frame=None): +def _frames(first_frame, middle_frames, last_frame, middle_frame=None, original_files=None): + original_files = original_files or {} contents = [] - middle = [("middle_frame", images) for name, images in sorted( - (middle_frame or {}).items(), key=lambda item: int(item[0].rsplit("_", 1)[1]))] + middle_values = dict(middle_frame or {}) + for key in original_files: + if key.startswith("middle_frame."): + middle_values.setdefault(key.removeprefix("middle_frame."), None) + middle = [("middle_frame", f"middle_frame.{name}", images) for name, images in sorted( + middle_values.items(), key=lambda item: int(item[0].rsplit("_", 1)[1]))] middle_index = 0 - for kind, images in [("first_frame", first_frame), ("middle_frame", middle_frames), *middle, ("last_frame", last_frame)]: - if images is None: + for kind, key, images in [("first_frame", "first_frame", first_frame), + ("middle_frame", "middle_frames", middle_frames), *middle, + ("last_frame", "last_frame", last_frame)]: + if images is None and key not in original_files: continue - encoded = _image_batch_to_base64_list(images) + encoded = _encode_images(images, original_files.get(key)) if kind != "middle_frame" and len(encoded) != 1: raise ValueError(f"{kind} requires a single image; use middle_frames for a batch.") for value in encoded: @@ -152,16 +183,34 @@ class Kling4Image2Video(io.ComfyNode): node_id="Comfyui-Kling-Wrapper Kling 4.0 Image2Video", category=NODE_CATEGORY + "/Kling 4.0", inputs=_frame_schema_inputs(), + hidden=[io.Hidden.dynprompt, io.Hidden.unique_id], outputs=[io.String.Output(display_name="url"), io.String.Output(display_name="video_id")], ) + @classmethod + def check_lazy_status(cls, **kwargs): + hidden = cls.hidden + return required_image_inputs(hidden.dynprompt if hidden else None, + hidden.unique_id if hidden else None, + _image_inputs(kwargs, lazy=True)) + @classmethod def execute(cls, **kwargs): - return io.NodeOutput(*cls().generate(**kwargs)) + files = {} + if cls.hidden is not None: + for name, image in _image_inputs(kwargs).items(): + if image is None: + paths = original_image_files(cls.hidden.dynprompt, cls.hidden.unique_id, name) + if paths is not None: + files[name] = paths + elif name in cls.hidden.dynprompt.get_node(cls.hidden.unique_id)["inputs"]: + raise ValueError("Reference image changed before upload; run the workflow again.") + return io.NodeOutput(*cls().generate(**kwargs, original_files=files)) def generate(self, prompt, resolution="1080p", duration="10", first_frame=None, - middle_frames=None, last_frame=None, watermark=False, callback_url="", external_task_id="", middle_frame=None): - prompt, contents = _unify_image_ids(prompt, _frames(first_frame, middle_frames, last_frame, middle_frame)) + middle_frames=None, last_frame=None, watermark=False, callback_url="", external_task_id="", middle_frame=None, + original_files=None): + prompt, contents = _unify_image_ids(prompt, _frames(first_frame, middle_frames, last_frame, middle_frame, original_files)) return _run("image", prompt, resolution, duration, contents=contents, watermark=watermark, callback_url=callback_url, external_task_id=external_task_id) @@ -173,7 +222,7 @@ class Kling4OmniVideo(Kling4Image2Video): inputs = _frame_schema_inputs() inputs.insert(3, io.Combo.Input("aspect_ratio", options=["adaptive", "21:9", "16:9", "9:16", "1:1"], default="adaptive")) inputs.extend([ - io.Image.Input("reference_images", optional=True, tooltip="旧参考图片批次，接在关键帧之后统一分配 image_N。"), + io.Image.Input("reference_images", optional=True, lazy=True, tooltip="旧参考图片批次，接在关键帧之后统一分配 image_N。"), io.Video.Input("base_video", optional=True, tooltip="待编辑视频；有此输入时为 @video_1。与 base_video_url 二选一。"), io.String.Input("base_video_url", optional=True, default=""), io.Video.Input("feature_video", optional=True, tooltip="特征参考视频；接在 base_video 后编号，无 base_video 时为 @video_1。"), @@ -182,28 +231,32 @@ class Kling4OmniVideo(Kling4Image2Video): io.Custom(CUSTOM_VOICE_LIST_TYPE).Input("custom_voice_list", optional=True, tooltip="现有音色列表：@voice_1、@voice_2…"), ]) for index in range(1, 11): - inputs.append(io.Image.Input(f"image_{index}", optional=True, tooltip="单张参考图；引用编号在所有关键帧之后，查看接口旁标签。")) + inputs.append(io.Image.Input(f"image_{index}", optional=True, lazy=True, tooltip="单张参考图；引用编号在所有关键帧之后，查看接口旁标签。")) for index in range(1, 6): inputs.append(io.Video.Input(f"feature_video_{index}", optional=True, tooltip="参考视频，实际 @video 编号显示在连接口旁。")) return io.Schema( node_id="Comfyui-Kling-Wrapper Kling 4.0 OmniVideo", category=NODE_CATEGORY + "/Kling 4.0", inputs=inputs, + hidden=[io.Hidden.dynprompt, io.Hidden.unique_id], outputs=[io.String.Output(display_name="url"), io.String.Output(display_name="video_id")], ) def generate(self, prompt, resolution="1080p", duration="10", aspect_ratio="adaptive", first_frame=None, middle_frames=None, last_frame=None, reference_images=None, base_video=None, base_video_url="", feature_video=None, feature_video_urls="", - element_list=None, custom_voice_list=None, watermark=False, callback_url="", external_task_id="", middle_frame=None, **dynamic_inputs): + element_list=None, custom_voice_list=None, watermark=False, callback_url="", external_task_id="", middle_frame=None, + original_files=None, **dynamic_inputs): + original_files = original_files or {} if base_video is not None and base_video_url.strip(): raise ValueError("Provide base_video or base_video_url, not both.") urls = [_video_url(line) for line in feature_video_urls.splitlines() if line.strip()] base_url = _video_url(base_video_url) if base_video_url.strip() else None - contents = _frames(first_frame, middle_frames, last_frame, middle_frame) + contents = _frames(first_frame, middle_frames, last_frame, middle_frame, original_files) dynamic_images = [] dynamic_videos = [] - for name, value in dynamic_inputs.items(): - if value is None: + image_files = {name: None for name in original_files if re.fullmatch(r"image_\d+", name)} + for name, value in (image_files | dynamic_inputs).items(): + if value is None and name not in original_files: continue match = re.fullmatch(r"(image|feature_video)_([1-9]\d*)", name) if not match: @@ -212,15 +265,15 @@ class Kling4OmniVideo(Kling4Image2Video): if index > (10 if kind == "image" else 5): raise ValueError(f"Kling 4.0 input index out of range: {name}") (dynamic_images if kind == "image" else dynamic_videos).append((index, value)) - if dynamic_images and reference_images is not None: + if dynamic_images and (reference_images is not None or "reference_images" in original_files): raise ValueError("Use image_N inputs or legacy reference_images, not both.") for index, image in sorted(dynamic_images): - encoded = _image_batch_to_base64_list(image) + encoded = _encode_images(image, original_files.get(f"image_{index}")) if len(encoded) != 1: raise ValueError(f"image_{index} requires a single image; connect each reference separately.") contents.append({"type": "refer_image", "url": encoded[0], "id": f"image_{index}"}) - if reference_images is not None: - for index, value in enumerate(_image_batch_to_base64_list(reference_images), 1): + if reference_images is not None or "reference_images" in original_files: + for index, value in enumerate(_encode_images(reference_images, original_files.get("reference_images")), 1): contents.append({"type": "refer_image", "url": value, "id": f"image_{index}"}) videos = [] if base_video is not None or base_url: diff --git a/tests/test_original_image_files.py b/tests/test_original_image_files.py new file mode 100644 index 0000000..3ab137b --- /dev/null +++ b/tests/test_original_image_files.py @@ -0,0 +1,264 @@ +import base64 +import importlib +import io +import os +from pathlib import Path +import sys +import tempfile +import zlib +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +from PIL import Image +import torch + + +ROOT = Path(__file__).resolve().parents[1] +for directory in (ROOT.parent.parent, ROOT): + sys.path.insert(0, str(directory)) + +from comfy.cli_args import args + +args.cpu = True +import execution +import nodes as core_nodes +from comfy_execution.graph import DynamicPrompt +from comfy_extras.nodes_post_processing import BatchImagesNode + +kling = importlib.import_module("py.nodes") +v4 = importlib.import_module("py.nodes_v4") +image_files = importlib.import_module("py.image_files") + + +class Collect: + @classmethod + def INPUT_TYPES(cls): + return {"required": {"value": ("STRING",)}} + + RETURN_TYPES = () + FUNCTION = "collect" + OUTPUT_NODE = True + + def collect(self, value): + return {"ui": {"value": [value]}} + + +class OriginalImageTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name).resolve() + self.owner = self.root / "alice" + self.owner.mkdir() + directory = patch.object(image_files.folder_paths, "get_input_directory", return_value=str(self.owner)) + directory.start() + self.addCleanup(directory.stop) + mapping = patch.dict(core_nodes.NODE_CLASS_MAPPINGS, { + "BatchImagesNode": BatchImagesNode, + "KlingElement": kling.AdvancedCustomElementCreateNode, + "KlingOmni": v4.Kling4OmniVideo, + "KlingImage": v4.Kling4Image2Video, + "Collect": Collect, + }) + mapping.start() + self.addCleanup(mapping.stop) + + def picture(self, name="photo.jpg", size=(24, 16), mode="RGB", **options): + path = self.owner / name + Image.new(mode, size, "red").save(path, **options) + return path + + def graph(self, filename="photo.jpg"): + return { + "source": {"class_type": "LoadImage", "inputs": {"image": filename}}, + "target": {"class_type": "KlingElement", "inputs": {"image": ["source", 0]}}, + } + + async def execute(self, prompt): + prompt["collect"] = {"class_type": "Collect", "inputs": {"value": ["target", 0]}} + server = SimpleNamespace(client_id=None, send_sync=Mock()) + executor = execution.PromptExecutor(server, cache_type=execution.CacheType.NONE, + cache_args={"ram": 0, "ram_inactive": 0}) + await executor.execute_async(prompt, "original-image-test", execute_outputs=["collect"]) + self.assertTrue(executor.success, executor.status_messages) + return executor + + def test_headers_only_and_original_bytes(self): + for extension in ("jpg", "png"): + with self.subTest(extension=extension): + path = self.picture(f"photo.{extension}") + graph = DynamicPrompt(self.graph(path.name)) + with patch.object(Image.Image, "load", side_effect=AssertionError("must not decode")): + paths = image_files.original_image_files(graph, "target", "image") + self.assertEqual(paths, [str(path)]) + self.assertEqual(base64.b64decode(v4._encode_images(None, paths)[0]), path.read_bytes()) + + def test_rotated_transparent_and_animated_images_fall_back(self): + exif = Image.Exif() + exif[274] = 8 + paths = [self.picture("rotated.jpg", exif=exif), + self.picture("alpha.png", mode="RGBA"), + self.picture("key.png", transparency=(255, 0, 0))] + animated = self.owner / "animated.png" + Image.new("RGB", (24, 16), "red").save( + animated, save_all=True, append_images=[Image.new("RGB", (24, 16), "blue")]) + paths.append(animated) + for path in paths: + with self.subTest(path=path.name): + self.assertIsNone(image_files.original_image_files(DynamicPrompt(self.graph(path.name)), "target", "image")) + + def test_png_exif_after_pixel_data_is_detected_without_decoding(self): + path = self.picture("late-exif.png") + exif = Image.Exif() + exif[274] = 8 + data = exif.tobytes()[6:] + chunk = b"eXIf" + data + chunk = len(data).to_bytes(4, "big") + chunk + zlib.crc32(chunk).to_bytes(4, "big") + original = path.read_bytes() + path.write_bytes(original[:-12] + chunk + original[-12:]) + with patch.object(Image.Image, "load", side_effect=AssertionError("must not decode")): + self.assertIsNone(image_files.original_image_files(DynamicPrompt(self.graph(path.name)), "target", "image")) + + def test_batch_order_duplicates_and_mixed_sizes(self): + first = self.picture("a.png") + second = self.picture("b.jpg") + graph = self.graph("a.png") + graph["second"] = {"class_type": "LoadImage", "inputs": {"image": "b.jpg"}} + graph["batch"] = {"class_type": "BatchImagesNode", "inputs": { + "images.image2": ["second", 0], "images.image0": ["source", 0], "images.image1": ["second", 0]}} + graph["target"]["inputs"]["image"] = ["batch", 0] + self.assertEqual(image_files.original_image_files(DynamicPrompt(graph), "target", "image"), + [str(first), str(second), str(second)]) + self.picture("b.jpg", size=(12, 8)) + self.assertIsNone(image_files.original_image_files(DynamicPrompt(graph), "target", "image")) + + def test_processed_missing_and_mask_inputs_fall_back(self): + self.picture() + for source in ( + {"class_type": "ImageScale", "inputs": {"image": ["source", 0]}}, + {"class_type": "LoadImage", "inputs": {"image": "missing.jpg"}}, + ): + graph = self.graph() + graph["processed"] = source + graph["target"]["inputs"]["image"] = ["processed", 0] + self.assertIsNone(image_files.original_image_files(DynamicPrompt(graph), "target", "image")) + graph = self.graph() + graph["target"]["inputs"]["image"] = ["source", 1] + self.assertIsNone(image_files.original_image_files(DynamicPrompt(graph), "target", "image")) + + def test_current_account_directory_is_enforced(self): + other = self.root / "bob" + other.mkdir() + path = other / "private.jpg" + Image.new("RGB", (24, 16)).save(path) + for filename in ("../bob/private.jpg", str(path)): + with self.subTest(filename=filename), self.assertRaisesRegex(ValueError, "current input directory"): + image_files.original_image_files(DynamicPrompt(self.graph(filename)), "target", "image") + + def test_symlink_cannot_escape_current_account(self): + outside = self.root / "outside.jpg" + Image.new("RGB", (24, 16)).save(outside) + try: + os.symlink(outside, self.owner / "link.jpg") + except OSError as error: + self.skipTest(str(error)) + with self.assertRaisesRegex(ValueError, "current input directory"): + image_files.original_image_files(DynamicPrompt(self.graph("link.jpg")), "target", "image") + + async def test_subject_creation_skips_load_and_batch_and_uploads_original_files(self): + path = self.picture() + graph = self.graph() + graph["batch"] = {"class_type": "BatchImagesNode", "inputs": { + "images.image0": ["source", 0], "images.image1": ["source", 0]}} + graph["target"]["inputs"].update(element_type="multi_image_subject", element_name="subject", + element_description="subject description", image_list=["batch", 0]) + uploaded = [] + + def upload(filename): + uploaded.append(Path(filename).read_bytes()) + return "https://example.com/original.jpg" + + provider = Mock() + provider.create_element.return_value = SimpleNamespace( + element={"element_id": "element", "reference_type": "image_refer"}, element_id="element") + with patch.object(core_nodes.LoadImage, "load_image", side_effect=AssertionError("LoadImage executed")), \ + patch.object(BatchImagesNode, "execute", side_effect=AssertionError("Batch executed")), \ + patch.object(kling, "_resolve_video_provider", return_value=kling.TENCENT_VOD_PROVIDER), \ + patch.object(kling, "_upload_file_to_temporary_media_host", side_effect=upload), \ + patch.object(kling, "_video_provider") as backend: + backend.return_value.__enter__.return_value = provider + await self.execute(graph) + self.assertEqual(uploaded, [path.read_bytes()] * 3) + self.assertTrue(path.exists()) + + async def test_omni_skips_sources_and_keeps_dynamic_frame_order(self): + path = self.picture() + graph = self.graph() + graph["target"] = {"class_type": "KlingOmni", "inputs": { + "prompt": "@image_1", "resolution": "720p", "duration": "10", "aspect_ratio": "adaptive", + "middle_frame.middle_frame_2": ["source", 0], "middle_frame.middle_frame_1": ["source", 0], + "image_1": ["source", 0]}} + with patch.object(core_nodes.LoadImage, "load_image", side_effect=AssertionError("LoadImage executed")), \ + patch.object(v4, "_run", return_value=("url", "id")) as run: + await self.execute(graph) + contents = run.call_args.args[5] + self.assertEqual([item["type"] for item in contents], ["middle_frame", "middle_frame", "refer_image"]) + self.assertEqual([item["id"] for item in contents], ["image_1", "image_2", "image_3"]) + self.assertTrue(all(base64.b64decode(item["url"]) == path.read_bytes() for item in contents)) + + async def test_rotated_photo_executes_existing_loader(self): + exif = Image.Exif() + exif[274] = 8 + self.picture(exif=exif) + graph = self.graph() + graph["target"] = {"class_type": "KlingImage", "inputs": { + "prompt": "photo", "resolution": "720p", "duration": "10", "first_frame": ["source", 0]}} + with patch.object(core_nodes.LoadImage, "load_image", autospec=True, return_value=(torch.ones(1, 24, 16, 3), None)) as load, \ + patch.object(v4, "_run", return_value=("url", "id")) as run: + await self.execute(graph) + load.assert_called_once() + with Image.open(io.BytesIO(base64.b64decode(run.call_args.kwargs["contents"][0]["url"]))) as image: + self.assertEqual(image.size, (16, 24)) + + async def test_resize_in_dynamic_middle_frame_uses_processed_pixels(self): + self.picture() + graph = self.graph() + graph["resize"] = {"class_type": "ImageScale", "inputs": { + "image": ["source", 0], "upscale_method": "nearest-exact", "width": 8, "height": 8, "crop": "disabled"}} + graph["target"] = {"class_type": "KlingImage", "inputs": { + "prompt": "photo", "resolution": "720p", "duration": "10", "middle_frame.middle_frame_1": ["resize", 0]}} + with patch.object(core_nodes.LoadImage, "load_image", autospec=True, return_value=(torch.ones(1, 16, 24, 3), None)) as load, \ + patch.object(v4, "_run", return_value=("url", "id")) as run: + await self.execute(graph) + load.assert_called_once() + with Image.open(io.BytesIO(base64.b64decode(run.call_args.kwargs["contents"][0]["url"]))) as image: + self.assertEqual(image.size, (8, 8)) + + async def test_mixed_size_batch_keeps_existing_resize_behavior(self): + self.picture("a.jpg") + self.picture("b.jpg", size=(12, 8)) + graph = self.graph("a.jpg") + graph["second"] = {"class_type": "LoadImage", "inputs": {"image": "b.jpg"}} + graph["batch"] = {"class_type": "BatchImagesNode", "inputs": { + "images.image0": ["source", 0], "images.image1": ["second", 0]}} + graph["target"] = {"class_type": "KlingOmni", "inputs": { + "prompt": "photo", "resolution": "720p", "duration": "10", "aspect_ratio": "adaptive", + "reference_images": ["batch", 0]}} + + def load_image(instance, image): + return torch.ones(1, 16, 24, 3) if image == "a.jpg" else torch.ones(1, 8, 12, 3), None + + with patch.object(core_nodes.LoadImage, "load_image", autospec=True, side_effect=load_image) as load, \ + patch.object(v4, "_run", return_value=("url", "id")) as run: + await self.execute(graph) + self.assertEqual(load.call_count, 2) + contents = run.call_args.args[5] + self.assertEqual(len(contents), 2) + for item in contents: + with Image.open(io.BytesIO(base64.b64decode(item["url"]))) as image: + self.assertEqual(image.size, (24, 16)) + + +if __name__ == "__main__": + unittest.main()