""" MetaHub Save Image Node for ComfyUI Advanced image saving with dual metadata support (A1111/Civitai + Image MetaHub) """ import time from pathlib import Path try: from . import metadata_utils as utils from .workflow_extractor import WorkflowExtractor except ImportError: import metadata_utils as utils from workflow_extractor import WorkflowExtractor _HINT_SHOWN = False class MetaHubSaveImage: """ Simple entry-point node. Keeps the UI focused on normal saving while delegating all metadata logic to MetaHubSaveNode. """ @classmethod def INPUT_TYPES(cls): return { "required": { "images": ("IMAGE",), }, "optional": { "filename_pattern": ("STRING", { "default": "ComfyUI_%counter%", "tooltip": "Filename pattern with placeholders" }), "file_format": (["PNG", "JPEG", "WebP"], { "default": "PNG", "tooltip": "File format" }), "output_path": ("STRING", { "default": "", "tooltip": "Custom output directory (empty = ComfyUI default)" }), "tags": ("STRING", { "default": "", "tooltip": "Optional Image MetaHub tags (comma-separated)" }), "notes": ("STRING", { "multiline": True, "default": "", "tooltip": "Optional Image MetaHub notes" }), "project_name": ("STRING", { "default": "", "tooltip": "Optional Image MetaHub project name" }), }, "hidden": { "prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "unique_id": "UNIQUE_ID" }, } RETURN_TYPES = () FUNCTION = "save_images" OUTPUT_NODE = True CATEGORY = "image/save" DESCRIPTION = "Save images with Image MetaHub metadata. Connect images and generate." def save_images( self, images, filename_pattern="ComfyUI_%counter%", file_format="PNG", output_path="", tags="", notes="", project_name="", prompt=None, extra_pnginfo=None, unique_id=None, ): return MetaHubSaveNode().save_images( images=images, filename_pattern=filename_pattern, file_format=file_format, output_path=output_path, user_tags=tags, notes=notes, project_name=project_name, prompt=prompt, extra_pnginfo=extra_pnginfo, unique_id=unique_id, save_node_class_type="MetaHubSaveImage", save_node_display_name="MetaHub Save Image", ) class MetaHubSaveNode: """ ComfyUI custom node for saving images with comprehensive metadata. Features: - A1111/Civitai compatible metadata (tEXt chunk "parameters") - Image MetaHub metadata (iTXt chunk "imagemetahub_data") - Auto-detection of workflow parameters (sampler, prompts, model, VAE, LoRAs) - SHA256 model hashes (AutoV2/Civitai format) - Graceful degradation (never interrupts generation) """ @classmethod def INPUT_TYPES(cls): return { "required": { "images": ("IMAGE",), }, "optional": { "filename_pattern": ("STRING", { "default": "ComfyUI_%counter%", "tooltip": "Filename pattern with placeholders" }), "file_format": (["PNG", "JPEG", "WebP"], { "default": "PNG", "tooltip": "File format" }), "quality": ("INT", { "default": 95, "min": 1, "max": 100, "tooltip": "JPEG/WebP quality" }), "output_path": ("STRING", { "default": "", "tooltip": "Custom output directory (empty = ComfyUI default)" }), "user_tags": ("STRING", { "default": "", "tooltip": "Optional Image MetaHub tags (comma-separated)" }), "notes": ("STRING", { "multiline": True, "default": "", "tooltip": "Optional Image MetaHub notes" }), "project_name": ("STRING", { "default": "", "tooltip": "Optional Image MetaHub project name" }), "seed": ("INT", { "default": None, "forceInput": True, "tooltip": "Override seed" }), "steps": ("INT", { "default": None, "forceInput": True, "tooltip": "Override steps" }), "cfg": ("FLOAT", { "default": None, "forceInput": True, "tooltip": "Override CFG scale" }), "sampler_name": ("STRING", { "default": None, "forceInput": True, "tooltip": "Override sampler name" }), "scheduler": ("STRING", { "default": None, "forceInput": True, "tooltip": "Override scheduler" }), "model_name": ("STRING", { "default": None, "forceInput": True, "tooltip": "Override model filename" }), "positive": ("STRING", { "multiline": True, "default": None, "forceInput": True, "tooltip": "Override positive prompt" }), "negative": ("STRING", { "multiline": True, "default": None, "forceInput": True, "tooltip": "Override negative prompt" }), "denoise": ("FLOAT", { "default": None, "forceInput": True, "tooltip": "Override denoise strength" }), "vae_name": ("STRING", { "default": None, "forceInput": True, "tooltip": "Override VAE filename" }), "generation_time": ("FLOAT", { "default": 0.0, "min": 0.0, "tooltip": "Generation time in seconds" }), "filename_prefix": ("STRING", { "default": "ComfyUI", "tooltip": "Deprecated: use filename_pattern" }), "upscale_model": ("STRING", { "default": "", "tooltip": "Upscale model name" }), # Performance metrics overrides (advanced users) "vram_peak_mb": ("FLOAT", { "default": None, "forceInput": True, "tooltip": "Override VRAM peak (MB)" }), "gpu_device_override": ("STRING", { "default": None, "forceInput": True, "tooltip": "Override GPU device name" }), "generation_time_override": ("FLOAT", { "default": None, "forceInput": True, "tooltip": "Timestamp from MetaHub Timer Node (elapsed time calculated automatically)" }), }, "hidden": { "prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "unique_id": "UNIQUE_ID" }, } RETURN_TYPES = () FUNCTION = "save_images" OUTPUT_NODE = True CATEGORY = "image/save" DESCRIPTION = "Save images with A1111/Civitai and Image MetaHub metadata" def save_images( self, images, filename_pattern="ComfyUI_%counter%", file_format="PNG", quality=95, output_path="", seed=None, steps=None, cfg=None, sampler_name=None, scheduler=None, model_name=None, positive=None, negative=None, denoise=None, vae_name=None, user_tags="", notes="", project_name="", generation_time=0.0, filename_prefix="ComfyUI", upscale_model="", vram_peak_mb=None, gpu_device_override=None, generation_time_override=None, prompt=None, extra_pnginfo=None, unique_id=None, save_node_class_type="MetaHubSaveNode", save_node_display_name="MetaHub Save Image Advanced", ): global _HINT_SHOWN try: workflow_json = utils.get_workflow_json(extra_pnginfo) prompt_data = prompt if isinstance(prompt, dict) else workflow_json.get("prompt", {}) if not isinstance(prompt_data, dict): prompt_data = {} parent_image = None if isinstance(extra_pnginfo, dict): parent_image = extra_pnginfo.get("parent_image") workflow_json = utils.ensure_prompt_in_workflow(workflow_json, prompt_data) save_node_id = str(unique_id) if unique_id is not None else None utils.ensure_metahub_save_node( workflow_json, save_node_id, class_type=save_node_class_type, display_name=save_node_display_name, ) imh_attribution = utils.extract_workflow_attribution(workflow_json, save_node_id) extractor = WorkflowExtractor(prompt_data) extracted, missing_fields = extractor.extract( save_node_id=save_node_id ) lora_list = extracted.get("lora_list") or utils.extract_loras_from_workflow(workflow_json) def resolve_value(manual_value, extracted_value, default_value): if manual_value is not None: return manual_value if extracted_value is not None: return extracted_value return default_value def normalize_prompt_override(value): if isinstance(value, bool): return None return value def normalize_int(value, default_value): try: return int(value) except (TypeError, ValueError): return default_value def normalize_float(value, default_value): try: return float(value) except (TypeError, ValueError): return default_value seed_value = normalize_int(resolve_value(seed, extracted.get("seed"), 0), 0) steps_value = normalize_int(resolve_value(steps, extracted.get("steps"), 20), 20) cfg_value = normalize_float(resolve_value(cfg, extracted.get("cfg"), 7.0), 7.0) sampler_value = resolve_value(sampler_name, extracted.get("sampler_name"), "euler") scheduler_value = resolve_value(scheduler, extracted.get("scheduler"), "normal") model_name_value = resolve_value(model_name, extracted.get("model_name"), "") positive_override = normalize_prompt_override(positive) negative_override = normalize_prompt_override(negative) positive_value = resolve_value(positive_override, extracted.get("positive"), "") negative_value = resolve_value(negative_override, extracted.get("negative"), "") denoise_value = normalize_float(resolve_value(denoise, extracted.get("denoise"), 1.0), 1.0) vae_name_value = resolve_value(vae_name, extracted.get("vae_name"), "") metadata_fields = [ "seed", "steps", "cfg", "sampler_name", "scheduler", "model_name", "positive", "negative", "denoise", "vae_name", ] manual_inputs = { "seed": seed, "steps": steps, "cfg": cfg, "sampler_name": sampler_name, "scheduler": scheduler, "model_name": model_name, "positive": positive_override, "negative": negative_override, "denoise": denoise, "vae_name": vae_name, } metadata_sources = utils.build_metadata_sources(manual_inputs, extracted, metadata_fields) metadata_status = utils.build_metadata_status(metadata_sources) quality_value = normalize_int(quality if quality is not None else 95, 95) quality_value = max(1, min(100, quality_value)) default_pattern = "ComfyUI_%counter%" pattern_value = (filename_pattern or "").strip() if pattern_value and pattern_value != default_pattern: final_pattern = pattern_value elif filename_prefix and str(filename_prefix).strip(): final_pattern = f"{filename_prefix}_%counter%" else: final_pattern = default_pattern if model_name_value: model_hash = utils.calculate_model_hash(model_name_value, model_type="checkpoint") else: model_hash = "0000000000" lora_hashes = utils.calculate_lora_hashes(lora_list) if lora_list else {} height, width = images[0].shape[0], images[0].shape[1] # Determine final generation time # generation_time_override is a TIMESTAMP from Timer Node, calculate elapsed time print(f"[MetaHub Save] generation_time_override={generation_time_override}, generation_time={generation_time}") if generation_time_override is not None and generation_time_override > 0: # Calculate elapsed time from Timer timestamp final_time = time.time() - generation_time_override print(f"[MetaHub Save] Calculated elapsed time from Timer: {final_time:.2f}s") elif generation_time > 0: final_time = generation_time print(f"[MetaHub Save] Using legacy generation_time: {final_time}s") else: final_time = 0.0 print(f"[MetaHub Save] No generation time provided") # Collect GPU metrics (auto-detect) gpu_metrics = utils.collect_gpu_metrics() # Collect version info version_info = utils.collect_version_info() # Calculate derived metrics generation_time_ms = None if final_time > 0: generation_time_ms = int(final_time * 1000) steps_per_second = None if generation_time_ms and generation_time_ms > 0 and steps_value > 0: steps_per_second = round((steps_value / (generation_time_ms / 1000)), 2) params = { "positive": positive_value, "negative": negative_value, "steps": steps_value, "sampler": sampler_value, "scheduler": scheduler_value, "cfg": cfg_value, "seed": seed_value, "width": width, "height": height, "model_name": model_name_value, "model_hash": model_hash, "vae_name": vae_name_value, "denoise": denoise_value, "upscale_model": upscale_model, "generation_time": generation_time, "user_tags": user_tags, "notes": notes, "project_name": project_name, "lora_list": lora_list, "lora_hashes": lora_hashes, "parent_image": parent_image, "generation_type": extracted.get("generation_type"), "source_image": extracted.get("source_image"), "imh_attribution": imh_attribution, # Performance metrics (Tier 1, 2, 3) "vram_peak_mb": vram_peak_mb if vram_peak_mb is not None else gpu_metrics.get("vram_peak_mb"), "gpu_device": gpu_device_override if gpu_device_override else gpu_metrics.get("gpu_device"), "generation_time_ms": generation_time_ms, "steps_per_second": steps_per_second, "comfyui_version": version_info.get("comfyui_version"), "torch_version": version_info.get("torch_version"), "python_version": version_info.get("python_version"), "metadata_status": metadata_status, "metadata_sources": metadata_sources, } a1111_metadata = utils.build_a1111_metadata(params) imh_metadata = utils.build_imh_metadata(params, workflow_json) output_dir = utils.get_output_directory(output_path) saved_paths = utils.save_image_batch( images, output_dir, final_pattern, params, file_format, quality_value, a1111_metadata, imh_metadata, ) warn_fields = { "seed", "steps", "cfg", "sampler_name", "scheduler", "model_name", "positive", "negative", "loras", } label_map = { "sampler_name": "sampler", "model_name": "model", } missing_warn = [] for field in missing_fields: if field not in warn_fields: continue if field == "loras": if lora_list: continue missing_warn.append("loras") continue if manual_inputs.get(field) is not None: continue missing_warn.append(label_map.get(field, field)) if missing_warn: missing_warn = sorted(set(missing_warn)) print( "[ImageMetaHub-Save] ⚠ Some params not detected " f"({', '.join(missing_warn)}), image saved with available metadata" ) auto_summary_parts = [] if seed is None and extracted.get("seed") is not None: auto_summary_parts.append(f"seed={seed_value}") if steps is None and extracted.get("steps") is not None: auto_summary_parts.append(f"steps={steps_value}") if model_name is None and extracted.get("model_name"): model_display = Path(str(model_name_value)).stem or str(model_name_value) auto_summary_parts.append(f"model={model_display}") auto_summary = ", ".join(auto_summary_parts) for file_path in saved_paths: suffix = "" if not _HINT_SHOWN and auto_summary: suffix = f" (auto-detected: {auto_summary})" print(f"[ImageMetaHub-Save] ✓ Saved: {file_path.name}{suffix}") if not _HINT_SHOWN: print("💡 Organize your AI images → github.com/LuqP2/ImageMetaHub") _HINT_SHOWN = True # Build preview structure for ComfyUI UI output_base = utils.get_output_directory("") # Get default ComfyUI output return utils.build_ui_preview(saved_paths, output_base) except Exception as e: print(f"[ImageMetaHub-Save] Warning: Save failed: {e}") raise RuntimeError(f"MetaHub Save Image failed: {e}") from e NODE_CLASS_MAPPINGS = { "MetaHubSaveImage": MetaHubSaveImage, "MetaHubSaveNode": MetaHubSaveNode, } NODE_DISPLAY_NAME_MAPPINGS = { "MetaHubSaveImage": "MetaHub Save Image", "MetaHubSaveNode": "MetaHub Save Image Advanced", }