diff --git a/__init__.py b/__init__.py index a5b691bc..3910d80c 100644 --- a/__init__.py +++ b/__init__.py @@ -18,6 +18,7 @@ try: # pragma: no cover - import fallback for pytest collection from .py.nodes.lora_info import LoraInfoLM from .py.nodes.lora_syntax_to_path import LoraSyntaxToPath from .py.nodes.create_hook_lora import CreateHookLoraLM + from .py.nodes.load_image_metadata import LoadImageMetadataLM from .py.nodes.metadata_overwrite import MetadataOverwriteLM from .py.metadata_collector import init as init_metadata_collector except ( @@ -70,6 +71,7 @@ except ( MetadataOverwriteLM = importlib.import_module( "py.nodes.metadata_overwrite" ).MetadataOverwriteLM + LoadImageMetadataLM = importlib.import_module("py.nodes.load_image_metadata").LoadImageMetadataLM init_metadata_collector = importlib.import_module("py.metadata_collector").init NODE_CLASS_MAPPINGS = { @@ -93,6 +95,7 @@ NODE_CLASS_MAPPINGS = { LoraSyntaxToPath.NAME: LoraSyntaxToPath, CreateHookLoraLM.NAME: CreateHookLoraLM, MetadataOverwriteLM.NAME: MetadataOverwriteLM, + LoadImageMetadataLM.NAME: LoadImageMetadataLM, } WEB_DIRECTORY = "./web/comfyui" diff --git a/docs/load-image-metadata.md b/docs/load-image-metadata.md new file mode 100644 index 00000000..ab3a3271 --- /dev/null +++ b/docs/load-image-metadata.md @@ -0,0 +1,208 @@ +# Load Image Metadata (LoraManager) + +Load a source image and reuse its prompts, local models, LoRAs, and sampling settings. +The node lives under **Lora Manager โ†’ loaders**. Restart ComfyUI after installing +this change and refresh the page. This Python node needs no Vue widget build. + +## Wiring a checkpoint workflow + +1. Upload/select an image in **Load Image Metadata (LoraManager)**. +2. Convert `ckpt_name` on **Checkpoint Loader (LoraManager)** to an input and + connect `model_name`. Leave its randomization control fixed. +3. Connect the checkpoint's MODEL and CLIP to **Lora Loader (LoraManager)**. + Connect the metadata node's `lora_stack` to that loader. Leave its LoRA widget + empty unless you intentionally want additional LoRAs. +4. Connect the LoRA loader's CLIP to two CLIP Text Encode nodes. Connect metadata + `positive` and `negative` to their text inputs, and their conditioning outputs + to KSampler. Connect the LoRA loader's MODEL to KSampler. +5. Convert KSampler's seed, steps, cfg, sampler_name, scheduler, and denoise + widgets to inputs and connect the corresponding metadata outputs. +6. For text-to-image, connect width/height to an appropriate Empty Latent node. + For img2img, encode the `image` output with the appropriate VAE instead. +7. Connect KSampler's samples and the checkpoint's VAE to VAE Decode, then Save Image. +8. Connect `readable_report` to a text display node for prompts, sampling settings, + model/LoRA names, local resolution status and warnings. The original `report` + output remains notes followed by formatted JSON; it is not a pure JSON string. + +`model_name`, `sampler_name`, and `scheduler` use COMBO outputs for converted +dropdown inputs in current ComfyUI. `model_name` contains the matching local +checkpoint or diffusion-model filename. The report identifies the resolved type; +connect it to the appropriate loader. Lookup searches both categories regardless +of how the original metadata labels the model. + +For a diffusion-model workflow, connect `model_name` to **Unet Loader +(LoraManager)** and select the correct text encoder(s), VAE, latent node and +architecture-specific conditioning separately. These settings do not reconstruct +an entire workflow or guarantee pixel-identical reproduction. + +## Selection and overrides + +`prefer_saved_image_metadata` is enabled by default. It prefers the saved +A1111-style generation parameters (including ComfyUI exports in that format) +over the workflow. The report identifies this source; `sampler_node_id` is +ignored in this mode when valid saved parameters are available. If saved +parameters are absent or malformed, the node tries workflow metadata and +reports any parsing failure. + +Disable the flag to prefer workflow extraction. Only active samplers are +eligible: muted/bypassed sampler nodes and samplers inside muted/bypassed +subgraph instances are excluded. This uses the saved UI workflow's mode flags +when available, including nested subgraphs, and any modes in the API graph. +Explicitly selecting an inactive sampler produces an error report and the +usual saved-parameter/default recovery; it never extracts that inactive stage. + +With one supported active sampler, leave `sampler_node_id` blank. With several, enter +its original node ID. Reports list candidate IDs when selection is ambiguous. +Native subgraphs in API prompt metadata use colon-qualified paths: `1481:1783` +means node 1783 inside subgraph instance 1481. Nested paths such as `10:20:30` +are supported; slash notation (`1481/1783`) is also accepted. A container ID +(`1481`) or leaf ID (`1783`) is accepted only if it identifies one sampler. +An exact sampler ID takes precedence over abbreviated matching. + +Selection follows that sampler's graph, rather than mixing branches. Supported +sampling nodes include KSampler, KSamplerAdvanced and SamplerCustomAdvanced with +standard RandomNoise, CFGGuider/BasicGuider, BasicScheduler and KSamplerSelect +components. BasicGuider's CFG is 1; its architecture-specific lack of negative +conditioning is reported. Known Image Saver parameter/selector outputs and +rgthree seed values can be read without executing those nodes. + +Detail Daemon's underlying sampler name is recovered, but its sampling effects +are explicitly unsupported. Other custom model/conditioning nodes can still +require defaults or overrides. If a requested stage cannot be read and global +image parameters are used instead, the report explicitly says those parameters +cannot verify the selected stage. Subgraph traversal requires the expanded API +prompt; UI-workflow-only subgraph definitions are not expanded or executed. + +Extraction errors do not stop this node. If an API prompt uses unsupported +samplers, the node first tries the image's saved generation parameters. Any +remaining unavailable or invalid extracted fields use the SDXL starter defaults; +valid extracted fields are preserved. `readable_report` starts with **โŒ ERROR** +and explains each recovery or substitution. This also applies to existing nodes +saved with `missing_settings=strict`; that legacy option no longer blocks +extraction recovery. New nodes default to `use_defaults`. + +The report uses emoji section markers (๐Ÿ–ผ๏ธ image, ๐Ÿ“ฆ model, โš™๏ธ sampling, ๐Ÿงฉ LoRAs, +โž•/โž– prompts) and โŒ/โš ๏ธ/โ„น๏ธ status markers. It is plain text, so colors depend on the +connected display node. Missing/ambiguous local files still appear in +`missing_files`. An empty model output requires selecting a local model manually. +Invalid explicit overrides and unreadable image files remain execution errors. + +`overrides_json` replaces extracted values, for example: + +```json +{ + "scheduler": "normal", + "model_name": "portraits/model.safetensors", + "seed": 12345, + "loras": [["styles/ink.safetensors", 0.7, 0.3]] +} +``` + +Supported keys: `positive`, `negative`, `model_name`, `seed`, +`steps`, `cfg`, `sampler_name`, `scheduler`, `width`, `height`, `denoise`, `loras`. +LoRA entries are `[name, model_strength, clip_strength]`; `"loras": []` explicitly +clears the extracted stack. Legacy `checkpoint_name` and `unet_name` override +keys remain accepted as aliases for `model_name`; supply only one model key. +Exact relative or absolute local +business paths disambiguate duplicate basenames. Matching falls back to a unique +filename or extensionless filename, then an exact unique catalog `file_name` or +`model_name` alias. Version dots are preserved when stripping known file +extensions. It never downloads or fuzzy-matches models, and stale entries whose +files no longer exist are excluded. + +Images with no metadata automatically use a bottle-inspired SDXL starter preset, +even with an existing saved `strict` setting: a glass-bottle/galaxy landscape +prompt, negative `text, watermark`, seed 0, 20 steps, CFG 7, Euler/normal, +1024ร—1024 and denoise 1, with no LoRAs. These settings are clearly identified as +synthetic defaults in both reports. Source image pixels and mask are unchanged. +Overrides take precedence. The node selects `sd_xl_base_1.0.safetensors` only +when uniquely indexed; otherwise choose an SDXL checkpoint manually or supply +`model_name`. Malformed or unsupported metadata also recovers with an explicit ERROR report. + +## Supported metadata and limits + +- PNG API prompt metadata; JPEG/WebP EXIF parameter comments; ComfyUI WebP + `prompt:`/`workflow:` EXIF fields. +- Standard KSampler, core checkpoint/UNet/LoRA loaders, LoRA Manager checkpoint, + UNet, LoRA/text loaders and LoRA stacks. LoRA application order and separate + model/CLIP strengths are preserved, including intentional repeated entries. + Different LoRA chains on model and prompt CLIP branches require an explicit + stack override rather than being silently merged. +- Literal CLIPTextEncode text and supported primitive value connections. Prompt + polarity comes from sampler wiring, never from words such as โ€œuglyโ€. +- A1111/Forge generation text with explicit sampler alias mappings. Recognized + LoRA directives become stack entries and are removed from prompt text. Literal + tags in ComfyUI encoder text remain literal; graph loaders determine its stack. +- A1111 `Automatic`/absent schedules do not reliably identify a ComfyUI schedule. + The node substitutes `normal` and reports the missing information as an ERROR; + an explicit override can select a different schedule. +- UI-workflow-only fallback supports known core widget layouts, with a report + warning. Saved widgets can differ from executed values (for example a seed + randomized after generation). Custom widget layouts are not guessed. +- KSamplerAdvanced partial/noise settings require an explicit denoise override; + this is an intentional approximation, not a reconstruction of those controls. +- Distinct SDXL/Flux encoder prompts, combined/regional/zeroed conditioning, + arbitrary custom nodes, dynamic wildcards and unsupported custom sampling components are + not automatically reconstructed. Supply explicit overrides or retain the + original workflow for those cases. +- Width/height come from a recognized latent source or fall back to source-image + dimensions; resized/upscaled images can therefore need dimension overrides. +- VAE, text encoder choice, CLIP skip, ControlNet and architecture-specific + conditioning still need the appropriate nodes. No embedded code is executed + and no external metadata service is contacted. + +LoRA Manager must have indexed the required models. Library resolution includes +its configured extra folders and preserves business paths through symlinks. + +## Extraction without a local catalog + +The parser extracts names before attempting local resolution. In recovery mode, +`report` includes `source_resources` with original model names, LoRA names and +strengths, and embedded resource hashes even when none are installed. The model +output sockets remain empty and the resolved stack excludes missing files. + +Combined sampler labels such as `Euler a SGM Uniform`, `Euler Normal` and +`er_sde simple` are split into sampler and scheduler. Multiline parameter blocks +and their nested JSON resource lists are supported. If prompt LoRA tags are +absent, one hash-name entry and one weighted resource can be matched offline; +multiple entries require an explicit mapping rather than guessing from order. +A single resource also disambiguates duplicated identical prompt tags. + +The `Model` field in A1111-style metadata does not distinguish checkpoints from +standalone diffusion models. The node searches both indexed categories by name, +then reports the matched type. Local model type and filename cannot be verified +without an indexed library. Multiple equally good matches are reported as +ambiguous; specify a relative path through `model_name` to disambiguate. + +## โ€œImage contains no supported generation metadataโ€ + +For older versions, this means extraction failed before any library lookup. +The current node uses the starter preset when metadata is entirely absent. The error identifies the +actual server file, its format, byte size and metadata keys. PNG text chunks are +read both before and after pixel data. If no generation metadata remains, upload +the original saved file: clipboard copies and re-encoded/exported images may +lose it. `use_defaults` supplies replacement settings; it does not recover the +original prompts or seed. + +## Missing local resources + +`missing_files` is a text output listing unresolved checkpoints/UNets and LoRAs. +LoRA entries include both model and CLIP weights and the resolution failure. +It is empty when all requested resources resolve. Missing and ambiguous LoRAs +are excluded from `lora_stack`, including in strict mode, so downstream loaders +receive only resolved files. Valid entries keep their original order and weights. +Unresolved model-name sockets are empty: select a model manually or override its +name before connecting that socket to a loader. + +## Output layout and upgrade + +The outputs start with `image`, `mask`, `positive`, `negative`, **`model_name`**, +**`lora_stack`**, **`lora_stack_text`**, followed by the sampling settings and reports. +`lora_stack_text` lists each resolved stack path with model and CLIP weights in +application order. It is empty for an empty stack; unresolved files appear only +in `missing_files`, with their requested weights. + +This replaces the former separate checkpoint/UNet sockets and renames `lost_list` +to `missing_files`. Restart ComfyUI, refresh, and recreate existing instances of +this node; reconnect the model and stack outputs to avoid stale saved slot indices. +Sampling and report output indices remain unchanged. No Vue build is required. diff --git a/py/nodes/load_image_metadata.py b/py/nodes/load_image_metadata.py new file mode 100644 index 00000000..a9c11346 --- /dev/null +++ b/py/nodes/load_image_metadata.py @@ -0,0 +1,431 @@ +"""Load an image and expose locally resolved generation settings.""" +from __future__ import annotations + +import hashlib +import json +import os +from typing import Any + +import folder_paths # pyright: ignore[reportMissingImports] + +from ..utils.exif_utils import ExifUtils +from ..utils.generation_metadata import ( + GenerationMetadata, + MetadataError, + extract_generation_metadata, + finite_number, + split_lora_tags, +) +from ..utils.utils import _format_model_name_for_comfyui +from .checkpoint_loader import CheckpointLoaderLM + + +DEFAULTS = { + "positive": "", "negative": "", "seed": 0, "steps": 20, "cfg": 7.0, + "sampler_name": "euler", "scheduler": "normal", "denoise": 1.0, +} +# An SDXL-sized starter preset inspired by ComfyUI's bottle example. These +# values are explicitly synthetic, never presented as recovered metadata. +EMPTY_IMAGE_DEFAULTS = { + **DEFAULTS, + "positive": "beautiful scenery inside a glass bottle, purple galaxy, intricate miniature landscape, highly detailed", + "negative": "text, watermark", + "width": 1024, + "height": 1024, +} +ALLOWED_OVERRIDES = set(DEFAULTS) | {"model_name", "checkpoint_name", "unet_name", "width", "height", "loras"} + + +def parse_overrides(text: str) -> dict[str, Any]: + try: + value = json.loads(text or "{}") + except ValueError as exc: + raise MetadataError(f"Invalid overrides_json: {exc}") from exc + if not isinstance(value, dict): + raise MetadataError("overrides_json must be an object") + unknown = set(value) - ALLOWED_OVERRIDES + if unknown: + raise MetadataError(f"Unknown override keys: {', '.join(sorted(unknown))}") + model_keys = [key for key in ("model_name", "checkpoint_name", "unet_name") if key in value] + if len(model_keys) > 1: + raise MetadataError("Specify only one model_name override (checkpoint_name/unet_name are legacy aliases)") + if model_keys: + key = model_keys[0] + name = value.pop(key) + if not isinstance(name, str) or not name.strip(): + raise MetadataError("model_name override must be nonempty text") + value["model_name"] = name.strip() + return value + + +_MODEL_FILE_EXTENSIONS = (".safetensors", ".ckpt", ".pt", ".pth", ".bin", ".gguf") + + +def _model_stem(name: str) -> str: + """Remove a known file extension, retaining dots in model/version names.""" + for extension in _MODEL_FILE_EXTENSIONS: + if name.lower().endswith(extension): + return name[:-len(extension)] + return name + + +def resolve_resource(name: str, resources: list[dict[str, Any]], roots: list[str]) -> dict[str, Any]: + """Match paths, filenames, then exact catalog aliases; never fuzzy-match.""" + if not isinstance(name, str) or not name.strip(): + raise MetadataError("Missing model name") + normalized = name.strip().replace("\\", "/") + levels: list[list[dict[str, Any]]] = [[], [], [], []] + for item in resources: + file_path = item.get("file_path") + if not file_path: + continue + path = file_path.replace("\\", "/") + relative = _format_model_name_for_comfyui(file_path, roots).replace("\\", "/") + exact = normalized in (path, relative, _model_stem(path), _model_stem(relative)) + basename = normalized.rsplit("/", 1)[-1] == path.rsplit("/", 1)[-1] + stem = _model_stem(normalized.rsplit("/", 1)[-1]) == _model_stem(path.rsplit("/", 1)[-1]) + aliases = [item.get("file_name"), item.get("model_name")] + alias = any( + isinstance(value, str) and normalized in (value.strip(), _model_stem(value.strip())) + for value in aliases + ) + # Stat only plausible matches, not every file in a large library for + # each LoRA. Missing cached files must never win a match. + if not (exact or basename or stem or alias) or not os.path.isfile(file_path): + continue + if exact: + levels[0].append(item) + if basename: + levels[1].append(item) + if stem: + levels[2].append(item) + if alias: + levels[3].append(item) + for matches in levels: + unique = {os.path.abspath(item["file_path"]): item for item in matches} + if len(unique) == 1: + return next(iter(unique.values())) + if unique: + raise MetadataError(f"Ambiguous local model '{name}': {', '.join(unique)}. Specify its relative path in overrides_json.") + raise MetadataError(f"Model '{name}' could not be matched to an existing file in the local LoRA Manager catalog") + + +class LoadImageMetadataLM: + NAME = "Load Image Metadata (LoraManager)" + CATEGORY = "Lora Manager/loaders" + DESCRIPTION = ( + "Load an image and recover prompts, LoRAs and sampling settings from its metadata. " + "Connect lora_stack to Lora Loader. Convert loader/sampler widgets to inputs for the other outputs. " + "Extraction failures use starter defaults and are shown as ERROR messages in readable_report." + ) + RETURN_TYPES = ( + "IMAGE", "MASK", "STRING", "STRING", "COMBO", "LORA_STACK", "STRING", + "INT", "INT", "FLOAT", "COMBO", "COMBO", "INT", "INT", "FLOAT", "STRING", "STRING", "STRING", + ) + RETURN_NAMES = ( + "image", "mask", "positive", "negative", "model_name", "lora_stack", "lora_stack_text", + "seed", "steps", "cfg", "sampler_name", "scheduler", "width", "height", "denoise", "report", "readable_report", "missing_files", + ) + FUNCTION = "load_metadata" + + @classmethod + def INPUT_TYPES(cls) -> dict[str, Any]: + from nodes import LoadImage # pyright: ignore[reportMissingImports] + + return {"required": { + "image": LoadImage.INPUT_TYPES()["required"]["image"], + "sampler_node_id": ("STRING", {"default": "", "tooltip": "Leave empty for a single sampler. Subgraphs: use the full API ID, e.g. 1481:1783 (or 1481/1783). A container or leaf ID works only when unique."}), + "missing_settings": (["use_defaults", "strict"], {"tooltip": "Extraction errors always return defaults and an ERROR report, including for saved strict settings. Unresolved files are listed in missing_files."}), + "overrides_json": ("STRING", {"default": "{}", "multiline": True, "dynamicPrompts": False, "tooltip": 'Explicit replacements, e.g. {"scheduler":"normal", "model_name":"folder/model.safetensors"}. Use "loras": [] to clear the recovered stack.'}), + "prefer_saved_image_metadata": ("BOOLEAN", {"default": True, "tooltip": "Prefer saved A1111-style generation parameters. Disable to select an active workflow sampler; muted/bypassed samplers are excluded."}), + }} + + @classmethod + def VALIDATE_INPUTS(cls, image: str, **kwargs: Any) -> bool | str: + if not folder_paths.exists_annotated_filepath(image): + return f"Invalid image file: {image}" + return True + + @classmethod + def IS_CHANGED(cls, image: str, **kwargs: Any) -> str: + digest = hashlib.sha256() + with open(folder_paths.get_annotated_filepath(image), "rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + @staticmethod + def _source_diagnostics(path: str) -> str: + """Describe the actual selected file without including prompt contents.""" + from PIL import Image + + try: + with Image.open(path) as source: + if source.format == "PNG": + source.load() + details = ( + f"File: {path}\nFormat: {source.format}; " + f"size: {os.path.getsize(path)} bytes; " + f"metadata keys: {', '.join(sorted(source.info)) or '(none)'}" + ) + return details + except (OSError, ValueError) as exc: + return f"File: {path}\nCould not inspect image metadata: {exc}" + + @staticmethod + def _library() -> tuple[list[dict[str, Any]], list[str], list[dict[str, Any]], list[str]]: + from ..services.service_registry import ServiceRegistry + + async def snapshot() -> tuple[list[dict[str, Any]], list[str], list[dict[str, Any]], list[str]]: + models = await ServiceRegistry.get_checkpoint_scanner() + loras = await ServiceRegistry.get_lora_scanner() + model_cache = await models.get_cached_data() + lora_cache = await loras.get_cached_data() + return list(model_cache.raw_data), models.get_model_roots(), list(lora_cache.raw_data), loras.get_model_roots() + + return CheckpointLoaderLM._run_async(snapshot) + + def load_metadata( + self, image: str, sampler_node_id: str = "", missing_settings: str = "use_defaults", + overrides_json: str = "{}", prefer_saved_image_metadata: bool = True, + ) -> tuple[Any, ...]: + import comfy.samplers # pyright: ignore[reportMissingImports] + from nodes import LoadImage # pyright: ignore[reportMissingImports] + + overrides = parse_overrides(overrides_json) + if missing_settings not in ("strict", "use_defaults"): + raise MetadataError("Invalid missing_settings policy") + path = folder_paths.get_annotated_filepath(image) + pixels, mask = LoadImage().load_image(image) + fields = {} + no_metadata = False + try: + fields = ExifUtils._load_structured_metadata(path) + no_metadata = not any(fields.values()) + if no_metadata: + extracted = GenerationMetadata( + values=dict(EMPTY_IMAGE_DEFAULTS), + notes=[ + "ERROR: No generation metadata found. Using the SDXL bottle starter preset; these settings were not extracted from the image.", + self._source_diagnostics(path), + ], + ) + else: + extracted = extract_generation_metadata(fields, sampler_node_id, prefer_saved_image_metadata) + except (ValueError, TypeError, KeyError, OSError, RecursionError) as exc: + error = f"ERROR: Metadata extraction failed: {exc}" + extracted = GenerationMetadata(issues={"source": str(exc)}) + # An unsupported API graph need not make valid saved generation + # parameters unusable. Do not execute or infer custom graph nodes. + if (fields.get("prompt") or fields.get("workflow")) and (fields.get("parameters") or fields.get("comment")): + try: + extracted = extract_generation_metadata({ + "parameters": fields.get("parameters"), "comment": fields.get("comment"), + }) + extracted.notes.append(error + "; recovered saved generation parameters instead.") + if sampler_node_id.strip(): + extracted.notes.append("ERROR: Global saved parameters cannot verify the requested sampler stage; they are an image-level fallback.") + except (ValueError, TypeError, KeyError, RecursionError) as fallback_exc: + extracted.notes.append(f"ERROR: Parameter fallback failed: {fallback_exc}") + if "source" in extracted.issues: + extracted.notes.extend([error, self._source_diagnostics(path)]) + source_resources = {"checkpoint_name": extracted.values.get("checkpoint_name"), "unet_name": extracted.values.get("unet_name"), "loras": list(extracted.loras), "resource_hints": extracted.resource_hints} + values = extracted.values + notes = extracted.notes + for key, value in overrides.items(): + values[key] = value + extracted.issues.pop(key, None) + notes.append(f"Explicit override: {key}.") + if "model_name" in overrides: + extracted.issues.pop("model", None) + values.pop("checkpoint_name", None) + values.pop("unet_name", None) + if "loras" in overrides: + extracted.loras = self._override_loras(overrides["loras"]) + notes.extend(f"ERROR: {key}: {message}" for key, message in extracted.issues.items()) + # Discard incomplete graph results instead of outputting half a LoRA + # chain or a prompt known to differ from its conditioning. + for key in extracted.issues: + if key not in overrides: + values.pop(key, None) + if "loras" in extracted.issues and "loras" not in overrides: + extracted.loras = [] + if "model" in extracted.issues and "model_name" not in overrides: + values.pop("checkpoint_name", None) + values.pop("unet_name", None) + for key, default in EMPTY_IMAGE_DEFAULTS.items(): + if key not in values: + values[key] = default + notes.append(f"ERROR: Missing {key}; using default {default!r}.") + # Validate independently so one invalid value cannot erase the other + # successfully extracted settings. Invalid explicit overrides still + # identify a user configuration error rather than an extraction error. + for key in EMPTY_IMAGE_DEFAULTS: + trial = {**EMPTY_IMAGE_DEFAULTS, key: values[key]} + try: + self._validate_values(trial, comfy.samplers.KSampler.SAMPLERS, comfy.samplers.KSampler.SCHEDULERS, True, []) + values[key] = trial[key] + except (ValueError, TypeError, OverflowError) as exc: + if key in overrides: + raise MetadataError(f"Invalid override {key}: {exc}") from exc + values[key] = EMPTY_IMAGE_DEFAULTS[key] + notes.append(f"ERROR: Invalid {key}: {exc}; using default {values[key]!r}.") + # Only A1111 directives represent LoRA application. In ComfyUI graphs, + # literal tags in encoder text are not executed by CLIPTextEncode. + for key in ("positive", "negative"): + try: + clean, tags = split_lora_tags(values[key]) + except (ValueError, TypeError) as exc: + if key in overrides: + raise MetadataError(f"Invalid override {key}: {exc}") from exc + values[key] = EMPTY_IMAGE_DEFAULTS[key] + notes.append(f"ERROR: Invalid LoRA directive in {key}: {exc}; using starter prompt.") + continue + if tags: + if notes and notes[0] == "A1111/Forge parameters.": + if "loras" not in overrides: + extracted.loras.extend(tags) + values[key] = clean + else: + notes.append(f"Literal LoRA tags retained in {key}; the embedded ComfyUI graph determines the stack.") + try: + models, roots, loras, lora_roots = self._library() + except Exception as exc: + models, roots, loras, lora_roots = [], [], [], [] + notes.append(f"ERROR: Local library lookup failed: {exc}. Extracted names remain in source_resources.") + if (no_metadata or "source" in extracted.issues) and "model_name" not in overrides: + base_candidates = [ + item for item in models + if item.get("sub_type") == "checkpoint" + and os.path.basename(item.get("file_path", "")).lower() == "sd_xl_base_1.0.safetensors" + and os.path.isfile(item["file_path"]) + ] + if len(base_candidates) == 1: + values["model_name"] = _format_model_name_for_comfyui(base_candidates[0]["file_path"], roots) + notes.append("Starter checkpoint: indexed sd_xl_base_1.0.safetensors.") + else: + notes.append("Select an SDXL checkpoint manually, or set model_name in overrides_json. No unambiguous SDXL base checkpoint was found.") + missing_entries = [] + name = values.get("model_name") or values.get("checkpoint_name") or values.get("unet_name") + values.pop("checkpoint_name", None) + values.pop("unet_name", None) + values["model_name"] = "" + values["model_type"] = "" + if name: + try: + # A1111's generic Model label can refer to either category. + # Search both together so duplicate names remain ambiguous. + available_models = [item for item in models if item.get("sub_type") in ("checkpoint", "diffusion_model")] + item = resolve_resource(name, available_models, roots) + values["model_name"] = _format_model_name_for_comfyui(item["file_path"], roots) + values["model_type"] = item["sub_type"] + notes.append(f"Resolved model_name: {values['model_name']} ({values['model_type']}).") + except MetadataError as exc: + missing_entries.append(f"Model: {name} โ€” {exc}") + notes.append(f"WARNING {exc}; model_name is empty.") + if not values["model_name"]: + notes.append("WARNING No model resolved. Select a model manually on your loader.") + stack = [] + for name, model_strength, clip_strength in extracted.loras: + try: + item = resolve_resource(name, loras, lora_roots) + stack.append((os.path.abspath(item["file_path"]), model_strength, clip_strength)) + except MetadataError as exc: + missing_entries.append(f"LoRA: {name} | model weight: {model_strength:g} | CLIP weight: {clip_strength:g} โ€” {exc}") + notes.append(f"WARNING Skipped LoRA: {exc}.") + notes.append(f"Resolved {len(stack)} LoRA entries; preserve stack order and avoid adding them again in the loader widget.") + notes.append("Metadata settings do not restore VAE, text encoders, ControlNet, regional conditioning or the original latent pipeline.") + lora_stack_text = "\n".join( + f"{path} | model weight: {model_strength:g} | CLIP weight: {clip_strength:g}" + for path, model_strength, clip_strength in stack + ) + missing_files = "\n".join(missing_entries) + report = "\n".join(notes) + "\n\n" + json.dumps({**values, "loras": stack, "lora_stack_text": lora_stack_text, "source_resources": source_resources, "missing_files": missing_files}, ensure_ascii=False, indent=2) + readable_report = self._readable_report(image, values, extracted.loras, stack, source_resources, notes) + return (pixels, mask, values["positive"], values["negative"], values["model_name"], + stack, lora_stack_text, values["seed"], values["steps"], + values["cfg"], values["sampler_name"], values["scheduler"], values["width"], + values["height"], values["denoise"], report, readable_report, missing_files) + + @staticmethod + def _readable_report( + image: str, values: dict[str, Any], requested_loras: list[tuple[str, float, float]], + stack: list[tuple[str, float, float]], source: dict[str, Any], notes: list[str], + ) -> str: + errors = [note for note in notes if note.startswith("ERROR")] + lines = ["๐Ÿ–ผ๏ธ IMAGE GENERATION SETTINGS", f"Image: {image}"] + if errors: + lines.extend(["", "โŒ ERROR โ€” RECOVERED SETTINGS / DEFAULTS", *errors]) + else: + lines.append("โœ… Metadata extracted") + lines.extend(["", "๐Ÿ“ฆ MODEL"]) + for key, label in (("checkpoint_name", "Checkpoint"), ("unet_name", "UNet")): + if source.get(key): + lines.append(f"{label} recorded in image: {source[key]}") + if values["model_name"]: + lines.append(f"Model resolved locally: {values['model_name']} ({values['model_type']})") + else: + lines.append("No local model resolved.") + lines.extend([ + "", "โš™๏ธ SAMPLING", f"Seed: {values['seed']}", f"Steps: {values['steps']}", + f"CFG: {values['cfg']:g}", f"Sampler: {values['sampler_name']}", + f"Scheduler: {values['scheduler']}", f"Size: {values['width']} ร— {values['height']}", + f"Denoise: {values['denoise']:g}", "", "๐Ÿงฉ LORAS", + ]) + if requested_loras: + for name, model_strength, clip_strength in requested_loras: + lines.append(f"- {name} (model: {model_strength:g}, CLIP: {clip_strength:g})") + else: + lines.append("No LoRA entries extracted or selected.") + for hint in source.get("resource_hints", []): + if hint.get("name") not in {entry[0] for entry in requested_loras}: + lines.append(f"- Recorded resource: {hint['name']} (strength unresolved)") + lines.append(f"Resolved locally: {len(stack)} of {len(requested_loras)} requested entries.") + lines.extend(["", "โž• POSITIVE PROMPT", values["positive"] or "(empty)", + "", "โž– NEGATIVE PROMPT", values["negative"] or "(empty)", + "", "๐Ÿ“‹ NOTES AND WARNINGS"]) + lines.extend(f"{'โŒ' if note.startswith('ERROR') else 'โš ๏ธ' if note.startswith('WARNING') else 'โ„น๏ธ'} {note}" for note in notes) + return "\n".join(lines) + + @staticmethod + def _override_loras(value: Any) -> list[tuple[str, float, float]]: + if not isinstance(value, list): + raise MetadataError("loras override must be a list of [name, model_strength, clip_strength]") + entries = [] + for entry in value: + if not isinstance(entry, list) or len(entry) != 3 or not isinstance(entry[0], str): + raise MetadataError("Each LoRA override must be [name, model_strength, clip_strength]") + entries.append((entry[0], finite_number(entry[1]), finite_number(entry[2]))) + return entries + + @staticmethod + def _validate_values(values: dict[str, Any], samplers: list[str], schedulers: list[str], strict: bool, notes: list[str]) -> None: + for key in ("positive", "negative"): + if not isinstance(values[key], str): + raise MetadataError(f"{key} must be text") + for key, low, high in (("seed", 0, 2**64 - 1), ("steps", 1, 10000), ("width", 1, 16384), ("height", 1, 16384)): + raw = values[key] + try: + number = int(raw) + if isinstance(raw, bool) or (isinstance(raw, float) and raw != number) or not low <= number <= high: + raise ValueError() + except (ValueError, TypeError, OverflowError) as exc: + raise MetadataError(f"{key} must be an integer between {low} and {high}") from exc + values[key] = number + for key, low, high in (("cfg", 0, 100), ("denoise", 0, 1)): + try: + number = finite_number(values[key]) + if not low <= number <= high: + raise ValueError() + except (ValueError, TypeError) as exc: + raise MetadataError(f"{key} must be a finite number between {low} and {high}") from exc + values[key] = number + for key, choices in (("sampler_name", samplers), ("scheduler", schedulers)): + if values[key] not in choices: + if strict: + raise MetadataError(f"Unsupported {key}: {values[key]!r}; set an explicit override") + fallback = DEFAULTS[key] + if fallback not in choices: + raise MetadataError(f"Default {key} {fallback!r} is unavailable in this ComfyUI installation") + notes.append(f"WARNING Replaced unsupported {key} {values[key]!r} with {fallback!r}.") + values[key] = fallback diff --git a/py/nodes/lora_loader.py b/py/nodes/lora_loader.py index 962af5ce..fe498e12 100644 --- a/py/nodes/lora_loader.py +++ b/py/nodes/lora_loader.py @@ -1,5 +1,6 @@ import importlib import logging +import os import comfy.sd # pyright: ignore[reportMissingImports] import comfy.utils # pyright: ignore[reportMissingImports] @@ -37,7 +38,9 @@ def _collect_stack_entries(lora_stack): for lora_path, model_strength, clip_strength in lora_stack: lora_name = extract_lora_name(lora_path) - absolute_lora_path, trigger_words = get_lora_info_absolute(lora_name) + absolute_lora_path, trigger_words = get_lora_info_absolute( + lora_path if os.path.isabs(lora_path) else lora_name + ) entries.append({ "name": lora_name, "absolute_path": absolute_lora_path, diff --git a/py/utils/exif_utils.py b/py/utils/exif_utils.py index 7ed97cc5..f7eca769 100644 --- a/py/utils/exif_utils.py +++ b/py/utils/exif_utils.py @@ -177,6 +177,11 @@ class ExifUtils: return brotli_meta with Image.open(image_path) as img: + # PNG text chunks may legally follow IDAT. Pillow reads those only + # when loading the image, so inspecting info immediately after open + # can incorrectly report a metadata-free image. + if img.format == "PNG": + img.load() info = getattr(img, "info", {}) or {} if "parameters" in info: @@ -193,6 +198,18 @@ class ExifUtils: exif[piexif.ExifIFD.UserComment] ) + # ComfyUI's WebP exporter stores JSON in EXIF Make/Model with + # prompt:/workflow: prefixes instead of UserComment. + exif = img.getexif() + for tag in (piexif.ImageIFD.Make, piexif.ImageIFD.Model): + text = ExifUtils._decode_exif_text(exif.get(tag)) + if not text: + continue + for key in ("prompt", "workflow"): + prefix = key + ":" + if text.startswith(prefix) and not metadata[key]: + metadata[key] = text[len(prefix):].rstrip("\x00") + try: exif_dict = piexif.load(image_path) except Exception as e: diff --git a/py/utils/generation_metadata.py b/py/utils/generation_metadata.py new file mode 100644 index 00000000..f5fec236 --- /dev/null +++ b/py/utils/generation_metadata.py @@ -0,0 +1,578 @@ +"""Offline extraction of reusable generation settings from image metadata. + +Embedded graphs are data: only explicit adapters are followed, never executed. +""" +from __future__ import annotations + +import json +import math +import re +from dataclasses import dataclass, field +from typing import Any + + +class MetadataError(ValueError): + """Metadata cannot be interpreted without a user decision.""" + + +@dataclass +class GenerationMetadata: + values: dict[str, Any] = field(default_factory=dict) + loras: list[tuple[str, float, float]] = field(default_factory=list) + issues: dict[str, str] = field(default_factory=dict) + notes: list[str] = field(default_factory=list) + resource_hints: list[dict[str, Any]] = field(default_factory=list) + + +LORA_PATTERN = re.compile(r"]+?):([+-]?[\d.eE]+)(?::([+-]?[\d.eE]+))?>", re.I) +SAMPLERS = { + "euler": "euler", "euler a": "euler_ancestral", "heun": "heun", + "lms": "lms", "dpm2": "dpm_2", "dpm2 a": "dpm_2_ancestral", + "dpm++ 2m": "dpmpp_2m", "dpm++ 2s a": "dpmpp_2s_ancestral", + "dpm++ sde": "dpmpp_sde", "dpm++ 2m sde": "dpmpp_2m_sde", + "dpm++ 3m sde": "dpmpp_3m_sde", "ddim": "ddim", "uni pc": "uni_pc", +} + + +def finite_number(value: Any) -> float: + if isinstance(value, bool): + raise MetadataError("Boolean is not a numeric generation setting") + number = float(value) + if not math.isfinite(number): + raise MetadataError("Generation settings must be finite numbers") + return number + + +def split_lora_tags(text: str) -> tuple[str, list[tuple[str, float, float]]]: + loras = [] + + def remove(match: re.Match[str]) -> str: + model = finite_number(match[2]) + clip = finite_number(match[3]) if match[3] is not None else model + loras.append((match[1].strip(), model, clip)) + return "" + + clean = LORA_PATTERN.sub(remove, text).strip() + if re.search(r" dict[str, Any]: + if isinstance(value, str): + if len(value) > 16 * 1024 * 1024: + raise MetadataError("Metadata exceeds the 16 MiB parsing limit") + value = json.loads(value) + if not isinstance(value, dict): + raise MetadataError("Expected a metadata JSON object") + return value + + +class GraphReader: + """Follow a selected sampler's inputs without mixing workflow branches.""" + + def __init__(self, graph: dict[str, Any], inactive_ids: set[str] | None = None) -> None: + if len(graph) > 10000: + raise MetadataError("Workflow exceeds the 10,000 node parsing limit") + self.graph = {str(key): value for key, value in graph.items()} + self.inactive_ids = inactive_ids or set() + self.result = GenerationMetadata() + + def node(self, link: Any, seen: tuple[str, ...]) -> tuple[str, str, dict[str, Any]]: + if not (isinstance(link, list) and len(link) == 2 and isinstance(link[1], int)): + raise MetadataError("Expected a workflow connection") + node_id = str(link[0]) + if node_id in seen or len(seen) >= 100: + raise MetadataError("Cyclic or excessively deep workflow connection") + node = self.graph.get(node_id) + if not isinstance(node, dict) or not isinstance(node.get("inputs"), dict): + raise MetadataError(f"Missing or malformed node {node_id}") + return node_id, node.get("class_type", ""), node["inputs"] + + def scalar(self, value: Any, seen: tuple[str, ...] = ()) -> Any: + if not isinstance(value, list): + if isinstance(value, (str, int, float)) and not isinstance(value, bool): + return value + raise MetadataError("Missing or non-scalar setting") + node_id, kind, inputs = self.node(value, seen) + if kind == "Input Parameters (Image Saver)": + keys = ("seed", "steps", "cfg", "sampler", "scheduler", "denoise") + if not 0 <= value[1] < len(keys): + raise MetadataError(f"Unsupported parameter output {value[1]} on {node_id}") + return self.scalar(inputs.get(keys[value[1]]), (*seen, node_id)) + if value[1] != 0: + raise MetadataError(f"Unsupported output {value[1]} on {kind} ({node_id})") + keys = { + "PrimitiveNode": "value", "PrimitiveInt": "value", "PrimitiveFloat": "value", + "PrimitiveString": "value", "PrimitiveStringMultiline": "value", + "easy int": "value", "easy float": "value", "easy string": "value", + "Seed (rgthree)": "seed", + "Sampler Selector (Image Saver)": "sampler_name", + "Scheduler Selector (Image Saver)": "scheduler", + "Text (LoraManager)": "text", "Reroute": "value", + } + if kind not in keys: + raise MetadataError(f"Unsupported value node {kind} ({node_id})") + resolved = self.scalar(inputs.get(keys[kind]), (*seen, node_id)) + if kind == "Text (LoraManager)" and isinstance(resolved, str) and re.search(r"__[^\n]+?__|\{[^{}]*\|[^{}]*\}", resolved): + raise MetadataError("Dynamic text expansion requires an explicit prompt override") + return resolved + + def text(self, link: Any, seen: tuple[str, ...] = ()) -> str: + node_id, kind, inputs = self.node(link, seen) + if link[1] != 0: + raise MetadataError(f"Unsupported conditioning output on {kind} ({node_id})") + if kind in ("CLIPTextEncode", "Prompt (LoraManager)"): + if kind == "Prompt (LoraManager)" and any(k.startswith("trigger_words") for k in inputs): + raise MetadataError("Prompt has dynamic trigger words; provide an explicit prompt override") + value = self.scalar(inputs.get("text"), (*seen, node_id)) + if not isinstance(value, str): + raise MetadataError("Prompt is not text") + if kind == "Prompt (LoraManager)" and re.search(r"__[^\n]+?__|\{[^{}]*\|[^{}]*\}", value): + raise MetadataError("Dynamic prompt expansion cannot be recovered from source text; provide an explicit prompt override") + return value + if kind in ("CLIPTextEncodeSDXL", "CLIPTextEncodeFlux"): + keys = ("text_g", "text_l") if kind == "CLIPTextEncodeSDXL" else ("clip_l", "t5xxl") + texts = [self.scalar(inputs.get(key), (*seen, node_id)) for key in keys] + if texts[0] != texts[1] or not isinstance(texts[0], str): + raise MetadataError(f"{kind} has distinct encoder prompts; a single string cannot reproduce it") + self.result.notes.append(f"{kind}: restore architecture-specific conditioning separately.") + return texts[0] + if kind == "ConditioningZeroOut": + raise MetadataError("Zeroed conditioning is not equivalent to encoding an empty prompt") + raise MetadataError(f"Unsupported conditioning node {kind} ({node_id}); use a prompt override") + + def widget_loras(self, value: Any) -> list[tuple[str, float, float]]: + if isinstance(value, dict): + value = value.get("__value__") + if isinstance(value, list) and len(value) == 1 and isinstance(value[0], list): + value = value[0] + if not isinstance(value, list): + raise MetadataError("Unsupported LoRA widget data") + entries = [] + for item in value: + if not isinstance(item, dict): + raise MetadataError("Malformed LoRA widget entry") + if item.get("active", False): + name = item.get("name") + if not isinstance(name, str) or not name: + raise MetadataError("LoRA name is missing") + strength = finite_number(item.get("strength")) + entries.append((name, strength, finite_number(item.get("clipStrength", strength)))) + return entries + + def stack(self, link: Any, seen: tuple[str, ...] = ()) -> list[tuple[str, float, float]]: + node_id, kind, inputs = self.node(link, seen) + if link[1] != 0: + raise MetadataError("Unsupported LoRA stack output") + seen = (*seen, node_id) + if kind == "Lora Stacker (LoraManager)": + previous = self.stack(inputs["lora_stack"], seen) if "lora_stack" in inputs else [] + return previous + self.widget_loras(inputs.get("loras", [])) + if kind == "Lora Stack Combiner (LoraManager)": + entries = [] + keys = [key for key in inputs if re.fullmatch(r"lora_stack\d+", key)] + for key in sorted(keys, key=lambda key: int(key[len("lora_stack"):])): + entries.extend(self.stack(inputs[key], seen)) + return entries + raise MetadataError(f"Unsupported LoRA stack node {kind} ({node_id})") + + def model(self, link: Any, seen: tuple[str, ...] = ()) -> None: + node_id, kind, inputs = self.node(link, seen) + if link[1] != 0: + raise MetadataError("Unsupported model output") + seen = (*seen, node_id) + loaders = { + "CheckpointLoaderSimple": ("checkpoint_name", "ckpt_name"), + "CheckpointLoader": ("checkpoint_name", "ckpt_name"), + "Checkpoint Loader (LoraManager)": ("checkpoint_name", "ckpt_name"), + "UNETLoader": ("unet_name", "unet_name"), + "Unet Loader (LoraManager)": ("unet_name", "unet_name"), + } + if kind in loaders: + output, key = loaders[kind] + self.result.values[output] = self.scalar(inputs.get(key), seen) + return + if kind in ("LoraLoader", "LoraLoaderModelOnly", "Lora Loader (LoraManager)", "LoraLoaderLM", "LoRA Text Loader (LoraManager)"): + self.model(inputs.get("model"), seen) + if "lora_stack" in inputs: + self.result.loras.extend(self.stack(inputs["lora_stack"], seen)) + if kind in ("LoraLoader", "LoraLoaderModelOnly"): + strength = finite_number(self.scalar(inputs.get("strength_model"), seen)) + clip = 0.0 if kind == "LoraLoaderModelOnly" else finite_number(self.scalar(inputs.get("strength_clip"), seen)) + name = self.scalar(inputs.get("lora_name"), seen) + if not isinstance(name, str): + raise MetadataError("LoRA name is not text") + self.result.loras.append((name, strength, clip)) + elif kind == "LoRA Text Loader (LoraManager)": + _, entries = split_lora_tags(self.scalar(inputs.get("lora_syntax"), seen)) + self.result.loras.extend(entries) + else: + self.result.loras.extend(self.widget_loras(inputs.get("loras", []))) + return + raise MetadataError(f"Unsupported model node {kind} ({node_id}); model/LoRA chain is incomplete") + + def clip_loras(self, link: Any, seen: tuple[str, ...] = ()) -> list[tuple[str, float]]: + """Check that prompt CLIP branches actually use the recovered LoRA stack.""" + node_id, kind, inputs = self.node(link, seen) + seen = (*seen, node_id) + if kind in ("CheckpointLoaderSimple", "CheckpointLoader", "Checkpoint Loader (LoraManager)") and link[1] == 1: + return [] + if kind in ("CLIPLoader", "DualCLIPLoader", "TripleCLIPLoader") and link[1] == 0: + return [] + if kind in ("LoraLoader", "Lora Loader (LoraManager)", "LoraLoaderLM", "LoRA Text Loader (LoraManager)") and link[1] == 1: + previous = self.clip_loras(inputs.get("clip"), seen) + entries = self.stack(inputs["lora_stack"], seen) if "lora_stack" in inputs else [] + if kind == "LoraLoader": + entries.append((self.scalar(inputs.get("lora_name")), 0, finite_number(self.scalar(inputs.get("strength_clip"))))) + elif kind == "LoRA Text Loader (LoraManager)": + _, parsed = split_lora_tags(self.scalar(inputs.get("lora_syntax"))) + entries.extend(parsed) + else: + entries.extend(self.widget_loras(inputs.get("loras", []))) + return previous + [(name, clip) for name, _, clip in entries if clip != 0] + raise MetadataError(f"Unsupported CLIP branch {kind} ({node_id}); restore text encoder/conditioning separately") + + def select_sampler(self, sampler_id: str) -> str: + candidates = [key for key, node in self.graph.items() if isinstance(node, dict) and node.get("class_type") in ("KSampler", "KSamplerAdvanced", "SamplerCustomAdvanced") + and node.get("mode", 0) == 0 + and not any(key == prefix or key.startswith(prefix + ":") for prefix in self.inactive_ids)] + selector = sampler_id.strip() + if selector in candidates: + return selector + # ComfyUI API prompts expand native subgraphs into colon-qualified IDs. + # Accept slash paths too, as well as an unambiguous container/leaf ID. + selector = selector.replace("/", ":") + if selector in self.graph and selector not in candidates: + raise MetadataError(f"Sampler {selector} is muted, bypassed or unsupported; active sampler IDs: {', '.join(candidates) or 'none'}") + if selector in candidates: + return selector + matches = candidates if not selector else [key for key in candidates if key.startswith(selector + ":") or key.endswith(":" + selector)] + if len(matches) == 1: + return matches[0] + choices = ", ".join(matches or candidates) or "none" + raise MetadataError(f"Choose a unique sampler_node_id; supported sampler IDs: {choices}") + + def custom_sampler_inputs(self, inputs: dict[str, Any]) -> dict[str, Any]: + """Adapt the core advanced sampling pipeline without executing any nodes.""" + result = {"latent_image": inputs.get("latent_image")} + adapters = ( + ("noise", {"RandomNoise": {"seed": "noise_seed"}}, ("seed",)), + ("guider", { + "CFGGuider": {"cfg": "cfg", "model": "model", "positive": "positive", "negative": "negative"}, + "BasicGuider": {"model": "model", "positive": "conditioning"}, + }, ("cfg", "model", "positive", "negative")), + ("sigmas", {"BasicScheduler": {"steps": "steps", "scheduler": "scheduler", "denoise": "denoise"}}, ("steps", "scheduler", "denoise")), + ) + for key, kinds, fields in adapters: + try: + link = inputs.get(key) + node_id, kind, upstream = self.node(link, ()) + if link[1] != 0 or kind not in kinds: + raise MetadataError(f"Unsupported {key} node {kind} ({node_id})") + for output, source in kinds[kind].items(): + result[output] = upstream.get(source) + if kind == "BasicGuider": + result["cfg"] = 1.0 + self.result.issues["negative"] = "BasicGuider has no negative conditioning; restore that architecture-specific setup separately" + except MetadataError as exc: + for field in fields: + self.result.issues[field] = str(exc) + try: + link = inputs.get("sampler") + seen = () + while True: + node_id, kind, upstream = self.node(link, seen) + seen = (*seen, node_id) + if link[1] != 0: + raise MetadataError("Unsupported sampler output") + if kind == "KSamplerSelect": + result["sampler_name"] = upstream.get("sampler_name") + break + if kind == "DetailDaemonSamplerNode": + self.result.issues["sampler_effects"] = "Detail Daemon modifies sampling; recovered base sampler settings do not reproduce this effect" + link = upstream.get("sampler") + continue + raise MetadataError(f"Unsupported sampler node {kind} ({node_id})") + except MetadataError as exc: + self.result.issues["sampler_name"] = str(exc) + return result + + def read(self, sampler_id: str) -> GenerationMetadata: + sampler_id = self.select_sampler(sampler_id) + node = self.graph[sampler_id] + inputs = node.get("inputs") + if not isinstance(inputs, dict): + raise MetadataError("Malformed sampler inputs") + self.result.notes.append(f"ComfyUI API graph; sampler {sampler_id} ({node['class_type']}).") + if node["class_type"] == "SamplerCustomAdvanced": + inputs = self.custom_sampler_inputs(inputs) + for output, key in {"seed": "noise_seed" if node["class_type"] == "KSamplerAdvanced" else "seed", "steps": "steps", "cfg": "cfg", "sampler_name": "sampler_name", "scheduler": "scheduler"}.items(): + try: + self.result.values[output] = self.scalar(inputs.get(key)) + except (ValueError, TypeError) as exc: + self.result.issues[output] = str(exc) + if node["class_type"] == "KSamplerAdvanced": + self.result.issues["denoise"] = "KSamplerAdvanced start/end/noise settings cannot be represented by denoise alone" + else: + try: + self.result.values["denoise"] = self.scalar(inputs.get("denoise", 1.0)) + except (ValueError, TypeError) as exc: + self.result.issues["denoise"] = str(exc) + for key in ("positive", "negative"): + try: + self.result.values[key] = self.text(inputs.get(key)) + except (ValueError, TypeError) as exc: + self.result.issues[key] = str(exc) + try: + self.model(inputs.get("model")) + except (ValueError, TypeError) as exc: + self.result.issues["model"] = str(exc) + self.result.issues["loras"] = "Model/LoRA chain could not be fully recovered" + expected_clip = [(name, clip) for name, _, clip in self.result.loras if clip != 0] + for polarity in ("positive", "negative"): + if polarity in self.result.issues: + continue + try: + _, _, encoder = self.node(inputs.get(polarity), ()) + if "clip" in encoder: + actual_clip = self.clip_loras(encoder["clip"]) + if actual_clip != expected_clip: + self.result.issues["loras"] = "Model and prompt CLIP branches use different LoRAs; explicitly choose a reusable stack with a loras override" + except MetadataError as exc: + self.result.issues[polarity] = str(exc) + try: + _, kind, latent = self.node(inputs.get("latent_image"), ()) + if kind in ("EmptyLatentImage", "EmptySD3LatentImage"): + for key in ("width", "height"): + self.result.values[key] = self.scalar(latent.get(key)) + else: + self.result.notes.append("Latent dimensions unavailable; using image dimensions. Restore the original latent/img2img setup separately.") + except MetadataError: + self.result.notes.append("Latent dimensions unavailable; using image dimensions.") + return self.result + + +def _parameter_fields(text: str) -> dict[str, str]: + """Split multiline parameters without splitting JSON objects or quoted names.""" + parts = [] + start = 0 + depth = 0 + quoted = False + escaped = False + for index, char in enumerate(text): + if quoted: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + quoted = False + elif char == '"': + quoted = True + elif char in "[{": + depth += 1 + elif char in "]}": + depth = max(0, depth - 1) + elif char == "," and depth == 0: + parts.append(text[start:index]) + start = index + 1 + parts.append(text[start:]) + fields = {} + for part in parts: + match = re.match(r"^\s*([\w ]+):\s*([\s\S]*)$", part) + if match: + fields[match[1].strip()] = match[2].strip() + return fields + + +def _parameter_loras(fields: dict[str, str], result: GenerationMetadata) -> None: + for key in ("positive", "negative"): + result.values[key], entries = split_lora_tags(result.values[key]) + result.loras.extend(entries) + try: + hashes = json.loads(fields.get("Hashes", "{}")) + resources = json.loads(fields.get("Civitai resources", "[]")) + if not isinstance(hashes, dict) or not isinstance(resources, list): + raise ValueError("Invalid resource containers") + except (ValueError, TypeError) as exc: + result.issues["loras"] = f"Malformed embedded resource metadata: {exc}" + return + names = [(key[5:], value) for key, value in hashes.items() if key.upper().startswith("LORA:")] + weighted = [item for item in resources if isinstance(item, dict) and "weight" in item] + result.resource_hints = [{"name": name, "hash": value} for name, value in names] + if result.loras: + if len(names) == 1 and len(weighted) == 1: + strength = finite_number(weighted[0]["weight"]) + single = (names[0][0], strength, strength) + if len(result.loras) > 1 and all(entry == single for entry in result.loras): + result.loras = [single] + result.notes.append("Repeated identical prompt tags collapsed to the single LoRA recorded in resource metadata.") + return + # Without a catalog there is no general mapping between a hash name and + # a Civitai version ID. One name and one resource are unambiguous; multiple + # resources must not be paired by their incidental JSON ordering. + if len(names) == 1 and len(weighted) == 1: + strength = finite_number(weighted[0]["weight"]) + result.loras.append((names[0][0], strength, strength)) + result.resource_hints[0].update(weighted[0]) + result.notes.append("LoRA name recovered from Hashes and its sole resource weight; separate CLIP strength was not saved, so model strength is used for both.") + elif names or weighted: + result.issues["loras"] = "LoRA resource names/weights cannot be paired unambiguously without a catalog; provide an explicit loras override" + + +def parse_parameters(text: str) -> GenerationMetadata: + match = re.search(r"^Steps:\s*\d+.*$", text, re.M) + if not match: + raise MetadataError("No supported A1111/Forge generation parameters found") + prompt = text[:match.start()].strip() + positive, separator, negative = prompt.partition("Negative prompt:") + fields = _parameter_fields(text[match.start():]) + result = GenerationMetadata(notes=["A1111/Forge parameters."]) + result.values.update(positive=positive.strip(), negative=negative.strip() if separator else "") + for output, key in {"seed": "Seed", "steps": "Steps", "cfg": "CFG scale", "sampler_name": "Sampler", "scheduler": "Schedule type", "checkpoint_name": "Model", "denoise": "Denoising strength"}.items(): + if key in fields: + result.values[output] = fields[key].strip().strip('"') + result.values.setdefault("denoise", 1.0) + size = re.fullmatch(r"(\d+)x(\d+)", fields.get("Size", "").strip()) + if size: + result.values.update(width=int(size[1]), height=int(size[2])) + sampler = str(result.values.get("sampler_name", "")).lower().strip() + for suffix, scheduler in ( + (" sgm uniform", "sgm_uniform"), (" sgm_uniform", "sgm_uniform"), + (" karras", "karras"), (" exponential", "exponential"), + (" simple", "simple"), ("_simple", "simple"), + (" normal", "normal"), ("_normal", "normal"), ("_sgm_uniform", "sgm_uniform"), + (" ddim uniform", "ddim_uniform"), + (" beta", "beta"), (" linear quadratic", "linear_quadratic"), + ): + if sampler.endswith(suffix): + sampler = sampler[:-len(suffix)] + result.values.setdefault("scheduler", scheduler) + break + result.values["sampler_name"] = SAMPLERS.get(sampler, sampler) + if "scheduler" in result.values: + result.values["scheduler"] = result.values["scheduler"].lower() + if result.values["scheduler"] == "automatic": + result.values.pop("scheduler") + if "scheduler" not in result.values: + result.issues["scheduler"] = "A1111 scheduler is unspecified/Automatic; choose an explicit ComfyUI scheduler" + for key in ("Clip skip", "Hires upscale", "Hires steps", "Hires upscaler"): + if key in fields: + result.notes.append(f"Restore separately: {key}: {fields[key]}") + _parameter_loras(fields, result) + return result + + +def inactive_workflow_nodes(workflow: dict[str, Any]) -> set[str]: + """Map muted/bypassed instances and nested nodes to API-qualified IDs.""" + inactive: set[str] = set() + definitions = {str(item["id"]): item for item in workflow.get("definitions", {}).get("subgraphs", []) if isinstance(item, dict) and "id" in item} + count = 0 + + def visit(container: dict[str, Any], prefix: str, ancestors: tuple[str, ...]) -> None: + nonlocal count + for node in container.get("nodes", []): + count += 1 + if count > 10000 or len(ancestors) > 100: + raise MetadataError("Workflow subgraph traversal limit exceeded") + if not isinstance(node, dict) or "id" not in node: + continue + node_id = prefix + str(node["id"]) + if node.get("mode", 0) != 0: + inactive.add(node_id) + continue + kind = node.get("type") + if kind in definitions: + if kind in ancestors: + raise MetadataError("Cyclic workflow subgraph definition") + visit(definitions[kind], node_id + ":", (*ancestors, kind)) + + visit(workflow, "", ()) + return inactive + + +def extract_generation_metadata( + fields: dict[str, Any], sampler_id: str = "", prefer_saved_image_metadata: bool = True, +) -> GenerationMetadata: + parameters = fields.get("parameters") or fields.get("comment") + saved_text = isinstance(parameters, str) and bool(parameters.strip()) and not parameters.lstrip().startswith("{") + recovery_notes = [] + if prefer_saved_image_metadata and saved_text: + try: + result = parse_parameters(parameters) + result.notes.append("Source: saved image generation parameters (preferred).") + if sampler_id.strip(): + result.notes.append("sampler_node_id is ignored while using saved image generation parameters.") + return result + except (ValueError, TypeError) as exc: + recovery_notes.append(f"ERROR: Saved image metadata could not be parsed: {exc}; trying workflow metadata.") + prompt = fields.get("prompt") + workflow = _json_object(fields["workflow"]) if fields.get("workflow") else None + if prompt: + try: + graph = _json_object(prompt) + except (ValueError, TypeError) as exc: + raise MetadataError(f"Malformed embedded prompt: {exc}") from exc + result = GraphReader(graph, inactive_workflow_nodes(workflow) if workflow else None).read(sampler_id.strip()) + elif isinstance(parameters, str) and parameters.lstrip().startswith("{"): + result = GraphReader(_json_object(parameters), inactive_workflow_nodes(workflow) if workflow else None).read(sampler_id.strip()) + elif workflow: + result = GraphReader(workflow_to_prompt(workflow)).read(sampler_id.strip()) + result.notes.insert(0, "UI workflow fallback: only known core widget layouts are supported; saved widget values may differ from executed values.") + elif saved_text: + result = parse_parameters(parameters) + result.notes.append("Source: saved image generation parameters; no workflow metadata available.") + else: + raise MetadataError("Image contains no supported generation metadata") + result.notes.extend(recovery_notes) + return result + + +def workflow_to_prompt(workflow: dict[str, Any]) -> dict[str, Any]: + """Decode only known core widget layouts; preserve links to unknown nodes.""" + nodes = workflow.get("nodes") + links = workflow.get("links", []) + if not isinstance(nodes, list) or not isinstance(links, list) or len(nodes) > 10000: + raise MetadataError("Malformed or excessively large UI workflow") + link_map = {} + for link in links: + if isinstance(link, list) and len(link) >= 5: + link_map[str(link[0])] = [str(link[1]), link[2]] + layouts = { + "CheckpointLoaderSimple": ["ckpt_name"], + "UNETLoader": ["unet_name", "weight_dtype"], + "LoraLoader": ["lora_name", "strength_model", "strength_clip"], + "LoraLoaderModelOnly": ["lora_name", "strength_model"], + "CLIPTextEncode": ["text"], + "EmptyLatentImage": ["width", "height", "batch_size"], + "EmptySD3LatentImage": ["width", "height", "batch_size"], + "KSampler": ["seed", "control_after_generate", "steps", "cfg", "sampler_name", "scheduler", "denoise"], + "PrimitiveNode": ["value"], + "PrimitiveInt": ["value"], "PrimitiveFloat": ["value"], + "PrimitiveString": ["value"], "PrimitiveStringMultiline": ["value"], + } + graph = {} + for node in nodes: + if not isinstance(node, dict) or "id" not in node: + raise MetadataError("Malformed workflow node") + kind = node.get("type", "") + widgets = node.get("widgets_values", []) + inputs = {} + layout = layouts.get(kind) + if node.get("mode", 0) != 0: + kind = "Unsupported muted/bypassed " + kind + elif layout is not None: + if not isinstance(widgets, list): + raise MetadataError(f"Unsupported widget layout for {kind}") + if kind == "KSampler" and len(widgets) == 6: + layout = [key for key in layout if key != "control_after_generate"] + for key, value in zip(layout, widgets): + inputs[key] = value + for slot in node.get("inputs", []): + if not isinstance(slot, dict) or not isinstance(slot.get("name"), str): + raise MetadataError("Malformed workflow input") + if slot.get("link") is not None: + inputs[slot["name"]] = link_map.get(str(slot["link"]), ["missing", 0]) + graph[str(node["id"])] = {"class_type": kind, "inputs": inputs} + return graph diff --git a/py/utils/utils.py b/py/utils/utils.py index db60e3f8..adbda081 100644 --- a/py/utils/utils.py +++ b/py/utils/utils.py @@ -114,6 +114,16 @@ def get_lora_info_absolute(lora_name): scanner = await ServiceRegistry.get_lora_scanner() cache = await scanner.get_cached_data() + # Stack producers can resolve an exact business path. Preserve it even + # when several indexed LoRAs share the same basename. + if os.path.isabs(lora_name): + for item in cache.raw_data: + file_path = item.get("file_path") + if file_path and os.path.abspath(file_path) == os.path.abspath(lora_name): + civitai = item.get("civitai") or {} + return file_path, civitai.get("trainedWords", []) + return lora_name, [] + lora_name_normalized = lora_name.replace("\\", "/") lora_name_no_ext = lora_name_normalized for ext in (".safetensors", ".ckpt", ".pt", ".bin"): diff --git a/tests/nodes/test_load_image_metadata.py b/tests/nodes/test_load_image_metadata.py new file mode 100644 index 00000000..17de96c9 --- /dev/null +++ b/tests/nodes/test_load_image_metadata.py @@ -0,0 +1,430 @@ +import json +import sys +import types +from pathlib import Path + +import piexif +import piexif.helper +import pytest +from PIL import Image, PngImagePlugin + +from py.nodes.load_image_metadata import LoadImageMetadataLM, MetadataError, resolve_resource +from py.utils.exif_utils import ExifUtils + + +PARAMETERS = 'cat \nNegative prompt: blur\nSteps: 25, Sampler: Euler, Schedule type: Normal, CFG scale: 6.5, Seed: 18446744073709551615, Size: 768x1024, Model: base' + + +@pytest.fixture +def runtime(tmp_path, monkeypatch): + import comfy + import folder_paths + import nodes + + image_path = tmp_path / "input.png" + info = PngImagePlugin.PngInfo() + info.add_text("parameters", PARAMETERS) + Image.new("RGB", (16, 24)).save(image_path, pnginfo=info) + model = tmp_path / "base.safetensors" + lora = tmp_path / "style.safetensors" + model.touch() + lora.touch() + library = ([{"file_path": str(model), "sub_type": "checkpoint"}], [str(tmp_path)], [{"file_path": str(lora)}], [str(tmp_path)]) + monkeypatch.setattr(LoadImageMetadataLM, "_library", staticmethod(lambda: library)) + monkeypatch.setattr(folder_paths, "get_annotated_filepath", lambda name: str(image_path), raising=False) + monkeypatch.setattr(folder_paths, "exists_annotated_filepath", lambda name: image_path.exists(), raising=False) + pixels = types.SimpleNamespace(shape=(1, 24, 16, 3)) + mask = object() + class LoadImage: + @classmethod + def INPUT_TYPES(cls): + return {"required": {"image": (["input.png"], {"image_upload": True})}} + + def load_image(self, name): + return pixels, mask + monkeypatch.setattr(nodes, "LoadImage", LoadImage, raising=False) + samplers = types.ModuleType("comfy.samplers") + samplers.KSampler = types.SimpleNamespace(SAMPLERS=["euler", "dpmpp_2m"], SCHEDULERS=["normal", "karras"]) + monkeypatch.setitem(sys.modules, "comfy.samplers", samplers) + monkeypatch.setattr(comfy, "samplers", samplers, raising=False) + return image_path, library, pixels, mask + + +def test_full_node_contract_with_real_png_metadata(runtime): + _, library, pixels, mask = runtime + result = LoadImageMetadataLM().load_metadata("input.png") + assert len(result) == len(LoadImageMetadataLM.RETURN_TYPES) + assert result[:4] == (pixels, mask, "cat", "blur") + assert result[5] == [(library[2][0]["file_path"], .7, .2)] + assert result[7:15] == (2**64 - 1, 25, 6.5, "euler", "normal", 768, 1024, 1.0) + assert "Resolved 1 LoRA" in result[15] + assert LoadImageMetadataLM.INPUT_TYPES()["required"]["image"][1]["image_upload"] + + +@pytest.mark.parametrize("extension", ["webp", "jpg"]) +def test_exif_parameters_from_real_image(runtime, extension): + image_path, *_ = runtime + exif = piexif.dump({"Exif": {piexif.ExifIFD.UserComment: piexif.helper.UserComment.dump(PARAMETERS, encoding="unicode")}}) + alternate = image_path.with_suffix("." + extension) + Image.new("RGB", (16, 24)).save(alternate, exif=exif) + fields = ExifUtils._load_structured_metadata(str(alternate)) + assert "Steps: 25" in fields["parameters"] + + +def test_missing_lora_strict_or_explicit_skip(runtime): + runtime[1][2].clear() + strict_result = LoadImageMetadataLM().load_metadata("input.png") + assert strict_result[5] == [] + assert "LoRA: style | model weight: 0.7 | CLIP weight: 0.2" in strict_result[17] + result = LoadImageMetadataLM().load_metadata("input.png", missing_settings="use_defaults") + assert result[5] == [] + assert "Skipped LoRA" in result[15] + + +def test_overrides_replace_loras_and_preserve_large_seed(runtime): + result = LoadImageMetadataLM().load_metadata("input.png", overrides_json=json.dumps({"seed": 2**64 - 2, "loras": [], "positive": "changed"})) + assert result[2] == "changed" + assert result[5] == [] + assert result[7] == 2**64 - 2 + + +def test_no_metadata_can_be_inspected_with_defaults(runtime): + Image.new("RGB", (16, 24)).save(runtime[0]) + assert LoadImageMetadataLM().load_metadata("input.png")[12:14] == (1024, 1024) + result = LoadImageMetadataLM().load_metadata("input.png", missing_settings="use_defaults") + assert result[12:14] == (1024, 1024) + assert "No model resolved" in result[15] + + +@pytest.mark.parametrize("override", [{"seed": -1}, {"steps": 2.5}, {"cfg": float("nan")}, {"sampler_name": "made_up"}, {"positive": ["1", 0]}, {"unknown": 1}]) +def test_invalid_override_rejected(runtime, override): + with pytest.raises((MetadataError, ValueError)): + LoadImageMetadataLM().load_metadata("input.png", overrides_json=json.dumps(override)) + + +def test_duplicate_basenames_require_path(tmp_path): + items = [] + for folder in ("a", "b"): + directory = tmp_path / folder + directory.mkdir() + path = directory / "same.safetensors" + path.touch() + items.append({"file_path": str(path)}) + with pytest.raises(MetadataError, match="Ambiguous"): + resolve_resource("same", items, [str(tmp_path)]) + assert resolve_resource("b/same.safetensors", items, [str(tmp_path)]) == items[1] + assert resolve_resource("b/same", items, [str(tmp_path)]) == items[1] + + +def test_file_hash_detects_replacement_and_accepts_all_inputs(runtime): + before = LoadImageMetadataLM.IS_CHANGED("input.png", sampler_node_id="", missing_settings="strict", overrides_json="{}") + Image.new("RGB", (32, 32)).save(runtime[0]) + assert before != LoadImageMetadataLM.IS_CHANGED("input.png") + + +def test_comfy_webp_exif_prompt_fields(runtime): + image_path, *_ = runtime + graph = {"1": {"class_type": "KSampler", "inputs": {"seed": 42}}} + exif = piexif.dump({"0th": { + piexif.ImageIFD.Make: "prompt:" + json.dumps(graph), + piexif.ImageIFD.Model: 'workflow:{"nodes": []}', + }}) + alternate = image_path.with_suffix(".webp") + Image.new("RGB", (16, 24)).save(alternate, exif=exif) + fields = ExifUtils._load_structured_metadata(str(alternate)) + assert json.loads(fields["prompt"]) == graph + assert json.loads(fields["workflow"]) == {"nodes": []} + + + +def test_report_preserves_extracted_names_without_catalog(runtime): + runtime[1][0].clear() + runtime[1][2].clear() + result = LoadImageMetadataLM().load_metadata("input.png", missing_settings="use_defaults") + payload = json.loads(result[15].split("\n\n", 1)[1]) + assert result[4:7] == ("", [], "") + assert payload["source_resources"]["checkpoint_name"] == "base" + assert payload["source_resources"]["loras"] == [["style", .7, .2]] + + +# These user-provided images are optional local integration fixtures, not assets +# required by the public test suite. +_SAMPLE_PNGS = sorted((Path(__file__).resolve().parents[2] / "_tmp").glob("*.png")) +_SAMPLE_PNGS = [path for path in _SAMPLE_PNGS if path.stem.endswith("_")] + + +@pytest.mark.parametrize("sample", _SAMPLE_PNGS or [pytest.param(None, marks=pytest.mark.skip(reason="No local PNG samples"))], ids=lambda path: path.name if path else "no-samples") +def test_local_png_node_without_catalog(runtime, monkeypatch, sample): + import comfy.samplers + import folder_paths + + runtime[1][0].clear() + runtime[1][2].clear() + monkeypatch.setattr(folder_paths, "get_annotated_filepath", lambda name: str(sample)) + monkeypatch.setattr(comfy.samplers.KSampler, "SAMPLERS", ["euler", "euler_ancestral", "er_sde"]) + monkeypatch.setattr(comfy.samplers.KSampler, "SCHEDULERS", ["normal", "simple", "sgm_uniform"]) + result = LoadImageMetadataLM().load_metadata(sample.name, missing_settings="use_defaults") + payload = json.loads(result[15].split("\n\n", 1)[1]) + assert result[2] and result[3] + assert "I", len(payload)) + chunk_type + payload + + struct.pack(">I", zlib.crc32(chunk_type + payload) & 0xFFFFFFFF)) + # Place metadata immediately before IEND, after all pixel data. + image_path.write_bytes(original[:-12] + chunk + original[-12:]) + result = LoadImageMetadataLM().load_metadata("input.png") + assert result[2:4] == ("cat", "blur") + assert result[4] == "base.safetensors" + assert result[7] == 2**64 - 1 + + +def test_missing_metadata_report_identifies_actual_file(runtime): + Image.new("RGB", (16, 24)).save(runtime[0]) + message = LoadImageMetadataLM().load_metadata("input.png")[15] + assert str(runtime[0]) in message + assert "Format: PNG" in message + assert "metadata keys: (none)" in message + assert "settings were not extracted" in message + + + +def test_readable_report_contains_settings_prompts_and_missing_resources(runtime): + runtime[1][0].clear() + runtime[1][2].clear() + result = LoadImageMetadataLM().load_metadata("input.png", missing_settings="use_defaults") + readable = result[16] + assert LoadImageMetadataLM.RETURN_NAMES[16] == "readable_report" + assert "Checkpoint recorded in image: base" in readable + assert "No local model resolved." in readable + assert "Seed: 18446744073709551615" in readable + assert "Sampler: euler" in readable + assert "Size: 768 ร— 1024" in readable + assert "style (model: 0.7, CLIP: 0.2)" in readable + assert "Resolved locally: 0 of 1 requested entries." in readable + assert "POSITIVE PROMPT\ncat" in readable + assert "NEGATIVE PROMPT\nblur" in readable + assert "WARNING" in readable + assert json.loads(result[15].split("\n\n", 1)[1])["seed"] == 2**64 - 1 + + + +def test_empty_metadata_starter_respects_overrides_and_indexed_model(runtime): + Image.new("RGB", (16, 24)).save(runtime[0]) + base = runtime[0].parent / "sd_xl_base_1.0.safetensors" + base.touch() + runtime[1][0].append({"file_path": str(base), "sub_type": "checkpoint"}) + result = LoadImageMetadataLM().load_metadata("input.png", overrides_json='{"seed": 123, "positive": "custom prompt", "width": 768}') + assert result[2] == "custom prompt" + assert result[4] == base.name + assert result[7] == 123 + assert result[12:14] == (768, 1024) + assert result[5] == [] + + +def test_user_example_png_runs_with_saved_strict_setting(runtime, monkeypatch): + import folder_paths + + path = Path(__file__).resolve().parents[2] / "_tmp" / "example.png" + if not path.exists(): + pytest.skip("No local example.png fixture") + monkeypatch.setattr(folder_paths, "get_annotated_filepath", lambda name: str(path)) + assert not any(ExifUtils._load_structured_metadata(str(path)).values()) + runtime[1][0].clear() + runtime[1][2].clear() + result = LoadImageMetadataLM().load_metadata("example.png", missing_settings="strict") + assert "glass bottle" in result[2] + assert result[3] == "text, watermark" + assert result[4:7] == ("", [], "") + assert result[7:15] == (0, 20, 7.0, "euler", "normal", 1024, 1024, 1.0) + assert "starter preset" in result[16] + + + +def test_missing_files_includes_model_and_lora_in_strict_mode(runtime): + runtime[1][0].clear() + runtime[1][2].clear() + result = LoadImageMetadataLM().load_metadata("input.png", missing_settings="strict") + assert result[4:7] == ("", [], "") + assert "Model: base" in result[17] + assert "LoRA: style | model weight: 0.7 | CLIP weight: 0.2" in result[17] + assert LoadImageMetadataLM.RETURN_NAMES[17] == "missing_files" + + +def test_missing_files_keeps_valid_stack_entries(runtime): + result = LoadImageMetadataLM().load_metadata("input.png", overrides_json=json.dumps({"loras": [["style", .7, .2], ["missing", -.5, 0]]})) + assert result[5] == [(runtime[1][2][0]["file_path"], .7, .2)] + assert "LoRA: missing | model weight: -0.5 | CLIP weight: 0" in result[17] + assert "LoRA: style" not in result[17] + assert LoadImageMetadataLM().load_metadata("input.png")[17] == "" + + +@pytest.mark.parametrize("subtype", ["checkpoint", "diffusion_model"]) +def test_generic_model_name_resolves_both_model_categories(runtime, subtype): + runtime[1][0][0]["sub_type"] = subtype + result = LoadImageMetadataLM().load_metadata("input.png") + assert result[4] == "base.safetensors" + assert result[17] == "" + assert subtype in result[16] + assert LoadImageMetadataLM.RETURN_NAMES[4:7] == ("model_name", "lora_stack", "lora_stack_text") + assert result[6] == f"{runtime[1][2][0]['file_path']} | model weight: 0.7 | CLIP weight: 0.2" + + +def test_duplicate_model_names_across_categories_require_path(runtime): + directory = runtime[0].parent / "unet" + directory.mkdir() + model = directory / "base.safetensors" + model.touch() + runtime[1][0].append({"file_path": str(model), "sub_type": "diffusion_model"}) + # The exact root-relative name wins when present. + result = LoadImageMetadataLM().load_metadata("input.png", overrides_json='{"model_name":"unet/base.safetensors"}') + assert result[4] == "unet/base.safetensors" + result = LoadImageMetadataLM().load_metadata("input.png", overrides_json='{"model_name":"old/base.safetensors"}') + assert result[4] == "" + assert "Ambiguous" in result[17] + + +@pytest.mark.parametrize("key", ["model_name", "checkpoint_name", "unet_name"]) +def test_model_override_aliases(runtime, key): + runtime[1][0][0]["sub_type"] = "diffusion_model" + result = LoadImageMetadataLM().load_metadata("input.png", overrides_json=json.dumps({key: "base.safetensors"})) + assert result[4] == "base.safetensors" + + +@pytest.mark.parametrize("policy", ["strict", "use_defaults"]) +def test_unsupported_sampler_returns_defaults_and_error(runtime, policy): + info = PngImagePlugin.PngInfo() + info.add_text("prompt", json.dumps({"1": {"class_type": "CustomSampler", "inputs": {}}})) + Image.new("RGB", (16, 24)).save(runtime[0], pnginfo=info) + result = LoadImageMetadataLM().load_metadata("input.png", missing_settings=policy) + assert result[7:15] == (0, 20, 7.0, "euler", "normal", 1024, 1024, 1.0) + assert "glass bottle" in result[2] + assert result[5] == [] + assert "โŒ ERROR" in result[16] + assert "supported sampler IDs: none" in result[16] + assert "โš™๏ธ SAMPLING" in result[16] + + +def test_unsupported_graph_uses_valid_parameters_before_defaults(runtime): + info = PngImagePlugin.PngInfo() + info.add_text("prompt", json.dumps({"1": {"class_type": "CustomSampler", "inputs": {}}})) + info.add_text("parameters", PARAMETERS) + Image.new("RGB", (16, 24)).save(runtime[0], pnginfo=info) + result = LoadImageMetadataLM().load_metadata("input.png", missing_settings="strict", prefer_saved_image_metadata=False) + assert result[2] == "cat" + assert result[7] == 2**64 - 1 + assert result[8] == 25 + assert "recovered saved generation parameters" in result[16] + assert "โŒ ERROR" in result[16] + + +def test_invalid_extracted_number_preserves_other_settings(runtime): + info = PngImagePlugin.PngInfo() + info.add_text("parameters", PARAMETERS.replace("CFG scale: 6.5", "CFG scale: nan")) + Image.new("RGB", (16, 24)).save(runtime[0], pnginfo=info) + result = LoadImageMetadataLM().load_metadata("input.png") + assert result[9] == 7.0 + assert result[8] == 25 + assert "ERROR: Invalid cfg" in result[16] + + + +def test_actual_custom_sampler_png_uses_saved_parameters(runtime, monkeypatch): + import comfy.samplers + import folder_paths + + path = Path(__file__).resolve().parents[2] / "_tmp" / "20260613-122517_S4_unnamedaANIMA_v10_617459040116303.png" + if not path.exists(): + pytest.skip("No local custom sampler PNG") + monkeypatch.setattr(folder_paths, "get_annotated_filepath", lambda name: str(path)) + monkeypatch.setattr(comfy.samplers.KSampler, "SAMPLERS", ["euler", "er_sde"]) + monkeypatch.setattr(comfy.samplers.KSampler, "SCHEDULERS", ["normal", "simple"]) + result = LoadImageMetadataLM().load_metadata(path.name, missing_settings="strict", prefer_saved_image_metadata=False) + assert result[7:15] == (617459040116303, 30, 4.0, "er_sde", "simple", 1664, 1088, 1.0) + assert result[2] + assert "โŒ ERROR" in result[16] + assert "recovered saved generation parameters" in result[16] + + +@pytest.mark.parametrize("selector", ["1481:1783", "1481/1783", "1481", "1783"]) +def test_actual_png_subgraph_sampler_selection(runtime, monkeypatch, selector): + import comfy.samplers + import folder_paths + + path = Path(__file__).resolve().parents[2] / "_tmp" / "20260613-122517_S4_unnamedaANIMA_v10_617459040116303.png" + if not path.exists(): + pytest.skip("No local custom sampler PNG") + monkeypatch.setattr(folder_paths, "get_annotated_filepath", lambda name: str(path)) + monkeypatch.setattr(comfy.samplers.KSampler, "SAMPLERS", ["euler", "er_sde"]) + monkeypatch.setattr(comfy.samplers.KSampler, "SCHEDULERS", ["normal", "simple"]) + result = LoadImageMetadataLM().load_metadata(path.name, sampler_node_id=selector, prefer_saved_image_metadata=False) + assert result[7:12] == (617459040116303, 30, 4.0, "er_sde", "simple") + assert "sampler 1481:1783" in result[16] + assert "Detail Daemon" in result[16] + assert "recovered saved generation parameters" not in result[16] + + +def test_source_preference_flag_defaults_true(runtime): + assert LoadImageMetadataLM.INPUT_TYPES()["required"]["prefer_saved_image_metadata"][1]["default"] is True + info = PngImagePlugin.PngInfo() + info.add_text("parameters", PARAMETERS) + info.add_text("prompt", json.dumps({"1": {"class_type": "CustomSampler", "inputs": {}}})) + Image.new("RGB", (16, 24)).save(runtime[0], pnginfo=info) + result = LoadImageMetadataLM().load_metadata("input.png") + assert result[7] == 2**64 - 1 + assert "saved image generation parameters (preferred)" in result[16] + assert "โŒ ERROR" not in result[16] + + + +@pytest.mark.parametrize("name", ["Kroma.v2.1", "Kroma.v2.1.safetensors", " Kroma.v2.1 "]) +def test_model_resolution_preserves_dotted_extensionless_names(tmp_path, name): + directory = tmp_path / "Krea 2" + directory.mkdir() + path = directory / "Kroma.v2.1.safetensors" + path.touch() + item = {"file_path": str(path)} + assert resolve_resource(name, [item], [str(tmp_path)]) == item + + +def test_model_resolution_accepts_unique_catalog_model_name(tmp_path): + path = tmp_path / "local-renamed.safetensors" + path.touch() + item = {"file_path": str(path), "model_name": "Kroma catalog name"} + assert resolve_resource("Kroma catalog name", [item], [str(tmp_path)]) == item + + +def test_catalog_alias_ambiguity_and_stale_entries(tmp_path): + items = [] + for name in ("a", "b"): + path = tmp_path / (name + ".safetensors") + path.touch() + items.append({"file_path": str(path), "model_name": "Kroma"}) + with pytest.raises(MetadataError, match="Ambiguous"): + resolve_resource("Kroma", items, [str(tmp_path)]) + items.append({"file_path": str(tmp_path / "absent.safetensors"), "model_name": "missing"}) + with pytest.raises(MetadataError, match="could not be matched"): + resolve_resource("missing", items, [str(tmp_path)]) + assert resolve_resource("a.safetensors", items, [str(tmp_path)]) == items[0] diff --git a/tests/nodes/test_lora_loader.py b/tests/nodes/test_lora_loader.py index 814dab82..cf53d2f1 100644 --- a/tests/nodes/test_lora_loader.py +++ b/tests/nodes/test_lora_loader.py @@ -200,3 +200,19 @@ def test_lora_loader_qwen_model_raises_clear_error_when_helper_import_fails(monk [], lora_stack=[("stack_qwen.safetensors", 0.6, 0.1)], ) + + +def test_stack_entry_keeps_resolved_absolute_path(monkeypatch): + from py.nodes.lora_loader import _collect_stack_entries + + seen = [] + + def resolve(name): + seen.append(name) + return name, ["trigger"] + + monkeypatch.setattr("py.nodes.lora_loader.get_lora_info_absolute", resolve) + result = _collect_stack_entries([("/models/b/same.safetensors", .7, .3)]) + assert seen == ["/models/b/same.safetensors"] + assert result[0]["absolute_path"] == "/models/b/same.safetensors" + assert result[0]["clip_strength"] == .3 diff --git a/tests/utils/test_generation_metadata.py b/tests/utils/test_generation_metadata.py new file mode 100644 index 00000000..14da3bb9 --- /dev/null +++ b/tests/utils/test_generation_metadata.py @@ -0,0 +1,250 @@ +import json + +import pytest + +from py.utils.generation_metadata import ( + GraphReader, MetadataError, extract_generation_metadata, parse_parameters, split_lora_tags, +) + + +def graph(): + return { + "1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "base.safetensors"}}, + "2": {"class_type": "CLIPTextEncode", "inputs": {"text": "ugly monster, (detail:1.2)", "clip": ["1", 1]}}, + "3": {"class_type": "CLIPTextEncode", "inputs": {"text": "sunshine", "clip": ["1", 1]}}, + "4": {"class_type": "EmptyLatentImage", "inputs": {"width": 768, "height": 1024}}, + "5": {"class_type": "KSampler", "inputs": {"model": ["1", 0], "positive": ["2", 0], "negative": ["3", 0], "latent_image": ["4", 0], "seed": 18446744073709551615, "steps": 25, "cfg": 6.5, "sampler_name": "euler", "scheduler": "normal", "denoise": 1}}, + } + + +def test_traces_polarity_without_content_heuristics(): + result = GraphReader(graph()).read("") + assert result.values["positive"] == "ugly monster, (detail:1.2)" + assert result.values["negative"] == "sunshine" + assert result.values["seed"] == 2**64 - 1 + assert result.values["width"] == 768 + assert not result.issues + + +def test_multiple_samplers_require_selection_and_do_not_mix(): + data = graph() + data["6"] = {"class_type": "KSampler", "inputs": {**data["5"]["inputs"], "seed": 42}} + with pytest.raises(MetadataError, match="5, 6"): + GraphReader(data).read("") + assert GraphReader(data).read("6").values["seed"] == 42 + + +def test_model_lora_order_repeated_entries_and_clip_strength(): + data = graph() + data["6"] = {"class_type": "LoraLoader", "inputs": {"model": ["1", 0], "lora_name": "same.safetensors", "strength_model": .7, "strength_clip": .3}} + data["7"] = {"class_type": "Lora Loader (LoraManager)", "inputs": {"model": ["6", 0], "loras": {"__value__": [{"name": "same", "active": True, "strength": .4, "clipStrength": 0}, {"name": "disabled", "active": False}]}}} + data["5"]["inputs"]["model"] = ["7", 0] + result = GraphReader(data).read("") + assert result.loras == [("same.safetensors", .7, .3), ("same", .4, 0)] + + +def test_linked_primitive_and_cycle_detection(): + data = graph() + data["6"] = {"class_type": "PrimitiveInt", "inputs": {"value": 123}} + data["5"]["inputs"]["seed"] = ["6", 0] + assert GraphReader(data).read("").values["seed"] == 123 + data["6"]["inputs"]["value"] = ["6", 0] + assert "Cyclic" in GraphReader(data).read("").issues["seed"] + + +def test_unsupported_conditioning_is_not_silently_flattened(): + data = graph() + data["2"]["class_type"] = "ConditioningCombine" + assert "Unsupported conditioning" in GraphReader(data).read("").issues["positive"] + + +def test_parameters_sampler_mapping_and_clean_prompts(): + result = parse_parameters('portrait (detail:1.2) \nsecond line\nNegative prompt: blur\nmore blur\nSteps: 25, Sampler: DPM++ 2M Karras, CFG scale: 7, Seed: 123, Size: 512x768, Model: base') + assert result.values["sampler_name"] == "dpmpp_2m" + assert result.values["scheduler"] == "karras" + assert result.values["negative"] == "blur\nmore blur" + clean, loras = split_lora_tags(result.values["positive"]) + assert clean == "portrait (detail:1.2) \nsecond line" + assert loras == [] + assert result.loras == [("style", .7, .2)] + + +def test_unspecified_a1111_scheduler_requires_decision(): + result = parse_parameters("cat\nSteps: 20, Sampler: Euler a, Seed: 1, CFG scale: 7") + assert result.values["sampler_name"] == "euler_ancestral" + assert "scheduler" in result.issues + + +@pytest.mark.parametrize("value", ["", "", ""]) +def test_bad_lora_strength(value): + with pytest.raises(ValueError): + split_lora_tags(value) + + +def test_malformed_and_missing_metadata(): + with pytest.raises(MetadataError, match="Malformed"): + extract_generation_metadata({"prompt": "{"}) + with pytest.raises(MetadataError, match="no supported"): + extract_generation_metadata({}) + assert extract_generation_metadata({"comment": json.dumps(graph())}).values["steps"] == 25 + + +def test_core_ui_workflow_fallback(): + workflow = {"nodes": [ + {"id": 1, "type": "CheckpointLoaderSimple", "widgets_values": ["base.safetensors"]}, + {"id": 2, "type": "CLIPTextEncode", "widgets_values": ["positive"]}, + {"id": 3, "type": "CLIPTextEncode", "widgets_values": ["negative"]}, + {"id": 4, "type": "KSampler", "widgets_values": [42, "fixed", 20, 7, "euler", "normal", 1], "inputs": [ + {"name": "model", "link": 1}, {"name": "positive", "link": 2}, {"name": "negative", "link": 3}]}, + ], "links": [[1, 1, 0, 4, 0, "MODEL"], [2, 2, 0, 4, 1, "CONDITIONING"], [3, 3, 0, 4, 2, "CONDITIONING"]]} + result = extract_generation_metadata({"workflow": json.dumps(workflow)}) + assert result.values["positive"] == "positive" + assert result.values["seed"] == 42 + assert "UI workflow fallback" in result.notes[0] + + +def test_stack_combiner_uses_numeric_order(): + data = {str(i): {"class_type": "Lora Stacker (LoraManager)", "inputs": {"loras": [{"name": str(i), "strength": 1, "active": True}]}} for i in (1, 2, 10)} + data["20"] = {"class_type": "Lora Stack Combiner (LoraManager)", "inputs": {"lora_stack10": ["10", 0], "lora_stack2": ["2", 0], "lora_stack1": ["1", 0]}} + assert [entry[0] for entry in GraphReader(data).stack(["20", 0])] == ["1", "2", "10"] + + +def test_model_and_clip_lora_mismatch_requires_override(): + data = graph() + data["6"] = {"class_type": "LoraLoader", "inputs": {"model": ["1", 0], "clip": ["1", 1], "lora_name": "style", "strength_model": .7, "strength_clip": .3}} + data["5"]["inputs"]["model"] = ["6", 0] + assert "different LoRAs" in GraphReader(data).read("").issues["loras"] + data["2"]["inputs"]["clip"] = ["6", 1] + data["3"]["inputs"]["clip"] = ["6", 1] + assert not GraphReader(data).read("").issues + + +def test_malformed_sampler_inputs(): + data = graph() + data["5"]["inputs"] = None + with pytest.raises(MetadataError, match="Malformed sampler"): + GraphReader(data).read("") + + +@pytest.mark.parametrize("label,sampler,scheduler", [ + ("Euler a SGM Uniform", "euler_ancestral", "sgm_uniform"), + ("Euler simple", "euler", "simple"), + ("Euler Normal", "euler", "normal"), + ("er_sde simple", "er_sde", "simple"), +]) +def test_combined_sampler_scheduler_labels(label, sampler, scheduler): + result = parse_parameters(f"cat\nSteps: 30, Sampler: {label}, Seed: 42, CFG scale: 5") + assert result.values["sampler_name"] == sampler + assert result.values["scheduler"] == scheduler + assert not result.issues + + +def test_multiline_settings_and_single_resource_weight(): + result = parse_parameters('cat\nNegative prompt: blur\nSteps: 30, Sampler: Euler Normal, Seed: 42, CFG scale: 5, Clip skip: 0, extra text,\nmore text\n, Model: example, Hashes: {"model":"123", "LORA:style, special":"456"}, Civitai resources: [{"air":"urn:model"}, {"air":"urn:lora", "weight":0.74}]') + assert result.values["checkpoint_name"] == "example" + assert result.values["negative"] == "blur" + assert result.loras == [("style, special", .74, .74)] + assert result.resource_hints[0]["hash"] == "456" + + +def test_multiple_resource_weights_are_not_paired_by_order(): + result = parse_parameters('cat\nSteps: 20, Sampler: Euler Normal, Hashes: {"LORA:first":"aaa","LORA:second":"bbb"}, Civitai resources: [{"weight":0.5},{"weight":0.8}]') + assert result.loras == [] + assert "loras" in result.issues + assert [item["name"] for item in result.resource_hints] == ["first", "second"] + + + +def test_duplicate_tags_with_single_authoritative_resource(): + result = parse_parameters('cat \nSteps: 10, Sampler: Euler simple, Hashes: {"LORA:style":"abc"}, Civitai resources: [{"weight":0.45}]') + assert result.loras == [("style", .45, .45)] + assert "