""" Enhanced KSampler nodes with tiling support """ import os import sys import torch import importlib.util import numpy as np # Import our advanced noise generation library sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from noise_generator import create_spectral_diverse_noise, create_hierarchical_noise # Add the ComfyUI root directory to Python path to access the main nodes comfy_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) if comfy_dir not in sys.path: sys.path.insert(0, comfy_dir) # Import the common_ksampler function, MAX_RESOLUTION from the main nodes.py try: from nodes import common_ksampler, MAX_RESOLUTION except ImportError: # Fallback: try to import from nodes module in current directory spec = importlib.util.spec_from_file_location("nodes", os.path.join(comfy_dir, "nodes.py")) nodes_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(nodes_module) common_ksampler = nodes_module.common_ksampler MAX_RESOLUTION = nodes_module.MAX_RESOLUTION # Import required ComfyUI modules import comfy.samplers import comfy.utils import comfy.model_management # --- Detail Daemon scheduling functions --- def make_detail_daemon_schedule( steps, start, end, bias, amount, exponent, start_offset, end_offset, fade, smooth, ): start = min(start, end) mid = start + bias * (end - start) multipliers = np.zeros(steps) start_idx, mid_idx, end_idx = [ int(round(x * (steps - 1))) for x in [start, mid, end] ] if mid_idx - start_idx + 1 > 0: start_values = np.linspace(0, 1, mid_idx - start_idx + 1) if smooth: start_values = 0.5 * (1 - np.cos(start_values * np.pi)) start_values = start_values**exponent if start_values.any(): start_values *= amount - start_offset start_values += start_offset multipliers[start_idx : mid_idx + 1] = start_values if end_idx - mid_idx + 1 > 0: end_values = np.linspace(1, 0, end_idx - mid_idx + 1) if smooth: end_values = 0.5 * (1 - np.cos(end_values * np.pi)) end_values = end_values**exponent if end_values.any(): end_values *= amount - end_offset end_values += end_offset multipliers[mid_idx : end_idx + 1] = end_values multipliers[:start_idx] = start_offset multipliers[end_idx + 1 :] = end_offset multipliers *= 1 - fade return multipliers def get_dd_schedule( sigma: float, sigmas: torch.Tensor, dd_schedule: torch.Tensor, ) -> float: sched_len = len(dd_schedule) if ( sched_len < 2 or len(sigmas) < 2 or sigma <= 0 or not (sigmas[-1] <= sigma <= sigmas[0]) ): return 0.0 # First, we find the index of the closest sigma in the list to what the model was # called with. deltas = (sigmas[:-1] - sigma).abs() idx = int(deltas.argmin()) if ( (idx == 0 and sigma >= sigmas[0]) or (idx == sched_len - 1 and sigma <= sigmas[-2]) or deltas[idx] == 0 ): # Either exact match or closest to head/tail of the DD schedule so we # can't interpolate to another schedule item. return dd_schedule[idx].item() # If we're here, that means the sigma is in between two sigmas in the # list. idxlow, idxhigh = (idx, idx - 1) if sigma > sigmas[idx] else (idx + 1, idx) # We find the low/high neighbor sigmas - our sigma is somewhere between them. nlow, nhigh = sigmas[idxlow], sigmas[idxhigh] if nhigh - nlow == 0: # Shouldn't be possible, but just in case... Avoid divide by zero. return dd_schedule[idxlow].item() # Ratio of how close we are to the high neighbor. ratio = ((sigma - nlow) / (nhigh - nlow)).clamp(0, 1) # Mix the DD schedule high/low items according to the ratio. return torch.lerp(dd_schedule[idxlow], dd_schedule[idxhigh], ratio).item() class DetailDaemonModelWrapper: """ A nested wrapper for ComfyUI models. - The outer wrapper intercepts the __call__ to adjust sigmas for Detail Daemon. - It proxies all other attributes to the original model, but returns a specialized proxy for the inner 'model' attribute to ensure compatibility. """ def __init__(self, model, sigmas, cfg, dds_make_schedule): self.original_model = model self.dd_schedule = torch.tensor( dds_make_schedule(len(sigmas) - 1), dtype=torch.float32, device="cpu", ) self.sigmas_cpu = sigmas.detach().clone().cpu() self.sigma_max = float(self.sigmas_cpu[0]) self.sigma_min = float(self.sigmas_cpu[-1]) + 1e-5 self.cfg = cfg self._inner_model_proxy = self.InnerModelProxy(self.original_model) def __call__(self, x, sigma, **extra_args): """Intercepts the denoising call to adjust sigma.""" sigma_float = float(sigma.max().detach().cpu()) if not (self.sigma_min <= sigma_float <= self.sigma_max): return self.original_model(x, sigma, **extra_args) dd_adjustment = get_dd_schedule(sigma_float, self.sigmas_cpu, self.dd_schedule) * 0.1 adjusted_sigma = sigma * max(1e-6, 1.0 - dd_adjustment * self.cfg) return self.original_model(x, adjusted_sigma, **extra_args) def __getattr__(self, name): """Forwards attribute access to the original model, except for 'model'.""" if name == 'model': return self._inner_model_proxy return getattr(self.original_model, name) class InnerModelProxy: """ A proxy for the inner model object that ensures 'latent_format' is always available, falling back to the outer model's get_model_object if needed. """ def __init__(self, outer_model): self.outer_model = outer_model self.inner_model = outer_model.model def __getattr__(self, name): """ Forwards attribute access to the inner model, with a special case for 'latent_format' to prevent crashes in utilities like latent preview. """ if name == 'latent_format': if hasattr(self.inner_model, 'latent_format'): return self.inner_model.latent_format # Fallback for models where latent_format is not on the inner model return self.outer_model.get_model_object("latent_format") return getattr(self.inner_model, name) # --- End Detail Daemon --- class KSamplerMultiSeed: """ KSampler Multi-Seed Node This node is extracted from the main ComfyUI nodes.py and provides multi-seed sampling functionality. It automatically uses the latest common_ksampler function from ComfyUI, ensuring compatibility with updates. Features optional noise injection for generating variations. """ @classmethod def INPUT_TYPES(s): return { "required": { "model": ("MODEL", {"tooltip": "The model used for denoising the input latent."}), "positive": ("CONDITIONING", { "tooltip": "The conditioning describing the attributes you want to include in the image."}), "negative": ("CONDITIONING", { "tooltip": "The conditioning describing the attributes you want to exclude from the image."}), "latent_image": ("LATENT", {"tooltip": "The latent image to denoise."}), "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "The amount of denoising applied."}), "steps": ("INT", {"default": 20, "min": 1, "max": 10000, "tooltip": "The number of steps used in the denoising process."}), "cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0, "step": 0.1, "round": 0.01, "tooltip": "Classifier-Free Guidance scale."}), "seed_count": ("INT", {"default": 1, "min": 1, "max": 1000, "tooltip": "The number of seeds to generate images with."}), "seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff, "control_after_generate": True, "tooltip": "The starting random seed. It will be incremented for each image in the batch."}), "sampler_name": (comfy.samplers.KSampler.SAMPLERS, {"tooltip": "The algorithm used when sampling."}), "scheduler": (comfy.samplers.KSampler.SCHEDULERS, {"tooltip": "The scheduler controls how noise is gradually removed."}), "injected_noise": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01, "round": 0.01, "tooltip": "Strength of noise injection for variation generation. 0.0=disabled, >0.0=blend base and variation noise."}), }, "optional": { "dd_enabled": ("BOOLEAN", {"default": False, "label_on": "DD Enabled", "label_off": "DD Disabled"}), "detail_amount": ("FLOAT", {"default": 0.1, "min": -5.0, "max": 5.0, "step": 0.01, "tooltip": "Overall strength of the detail adjustment."}), "dd_start": ("FLOAT", {"default": 0.2, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Start of the adjustment curve as a fraction of total steps."}), "dd_end": ("FLOAT", {"default": 0.8, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "End of the adjustment curve as a fraction of total steps."}), "dd_bias": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Curve bias; >0.5 peaks later, <0.5 peaks earlier."}), "dd_exponent": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.05, "tooltip": "Exponent for the curve shape."}), "dd_start_offset": ("FLOAT", {"default": 0.0, "min": -1.0, "max": 1.0, "step": 0.01, "tooltip": "Adjustment multiplier before the curve starts."}), "dd_end_offset": ("FLOAT", {"default": 0.0, "min": -1.0, "max": 1.0, "step": 0.01, "tooltip": "Adjustment multiplier after the curve ends."}), "dd_fade": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.05, "tooltip": "Fade the entire effect in or out."}), "dd_smooth": ("BOOLEAN", {"default": True, "tooltip": "Apply smoothing to the curve."}), } } RETURN_TYPES = ("LATENT",) OUTPUT_TOOLTIPS = ("A batch of denoised latents, one for each seed.",) FUNCTION = "sample" CATEGORY = "pirog/sampling" DESCRIPTION = "Generates multiple images by incrementing the seed for each generation for each latent in the input batch." def slerp(self, val, low, high): """Spherical linear interpolation for noise blending""" dims = low.shape low = low.reshape(dims[0], -1) high = high.reshape(dims[0], -1) low_norm = low / torch.norm(low, dim=1, keepdim=True) high_norm = high / torch.norm(high, dim=1, keepdim=True) low_norm[low_norm != low_norm] = 0.0 high_norm[high_norm != high_norm] = 0.0 omega = torch.acos((low_norm * high_norm).sum(1)) so = torch.sin(omega) res = (torch.sin((1.0 - val) * omega) / so).unsqueeze(1) * low + (torch.sin(val * omega) / so).unsqueeze(1) * high return res.reshape(dims) def sample(self, model, positive, negative, latent_image, denoise, steps, cfg, seed_count, seed, sampler_name, scheduler, injected_noise=0.0, dd_enabled=False, detail_amount=0.1, dd_start=0.2, dd_end=0.8, dd_bias=0.5, dd_exponent=1.0, dd_start_offset=0.0, dd_end_offset=0.0, dd_fade=0.0, dd_smooth=True): output_latents = [] input_latents = latent_image["samples"] # Prepare sampler and sigmas once if needed for noise injection or Detail Daemon sigmas = None if injected_noise > 0.0 or dd_enabled: device = comfy.model_management.get_torch_device() comfy.model_management.load_model_gpu(model) sampler = comfy.samplers.KSampler(model, steps=steps, device=device, sampler=sampler_name, scheduler=scheduler, denoise=1.0, model_options=model.model_options) sigmas = sampler.sigmas # Prepare model wrapper if Detail Daemon is enabled model_to_use = model if dd_enabled and sigmas is not None: def dds_make_schedule(schedule_steps): return make_detail_daemon_schedule( schedule_steps, dd_start, dd_end, dd_bias, detail_amount, dd_exponent, dd_start_offset, dd_end_offset, dd_fade, dd_smooth ) model_to_use = DetailDaemonModelWrapper(model, sigmas, cfg, dds_make_schedule) total_generations = input_latents.shape[0] * seed_count pbar = comfy.utils.ProgressBar(total_generations) for latent_sample in input_latents: for i in range(seed_count): current_seed = seed + i # Reshape the single sample to a batch of 1 for the sampler work_latent = latent_sample.unsqueeze(0) # Apply noise injection if enabled if injected_noise > 0.0: # Generate base noise from main seed generator = torch.manual_seed(seed) base_noise = torch.randn(work_latent.shape, generator=generator, device="cpu", dtype=work_latent.dtype) # Generate variation noise from current seed (naturally varies per iteration) generator = torch.manual_seed(current_seed + 12345) # Offset to ensure different noise variation_noise = torch.randn(work_latent.shape, generator=generator, device="cpu", dtype=work_latent.dtype) # Blend noises using SLERP blended_noise = self.slerp(injected_noise, base_noise, variation_noise) # Calculate sigma scaling factor using pre-calculated sigmas start_at_step = round(steps - steps * denoise) end_at_step = steps sigma = sigmas[start_at_step] - sigmas[end_at_step] sigma /= model.model.latent_format.scale_factor sigma = sigma.detach().cpu().item() # Apply scaled noise to latent work_latent = work_latent + blended_noise.to(work_latent.device) * sigma latent_for_sampler = {"samples": work_latent} # Call the common ksampler function from the main ComfyUI nodes # This ensures we always use the latest version result_latent, = common_ksampler(model_to_use, current_seed, steps, cfg, sampler_name, scheduler, positive, negative, latent_for_sampler, denoise=denoise) output_latents.append(result_latent["samples"]) pbar.update(1) # Combine all the generated latents into a single batch tensor final_samples = torch.cat(output_latents, dim=0) # Create the final output dictionary final_latent = {"samples": final_samples} return (final_latent,) class KSamplerMultiSeedPlus: """ KSamplerMultiSeed+ Node - Integrated Sampling Pipeline This node combines the KSamplerMultiSeed functionality with integrated VAE encode/decode and empty latent image generation. It provides a complete pipeline in one node: - For denoise=1.0: Creates empty latent → samples → decodes to image - For denoise<1.0: Encodes input image → samples → decodes to image Uses minimal approach by calling vanilla ComfyUI functions directly. """ def __init__(self): self.device = comfy.model_management.intermediate_device() @classmethod def INPUT_TYPES(s): return { "required": { "model": ("MODEL", {"tooltip": "The model used for denoising."}), "vae": ("VAE", {"tooltip": "The VAE model used for encoding/decoding."}), "positive": ("CONDITIONING", { "tooltip": "The conditioning describing the attributes you want to include in the image."}), "negative": ("CONDITIONING", { "tooltip": "The conditioning describing the attributes you want to exclude from the image."}), "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "The amount of denoising applied. 1.0=new image, <1.0=img2img."}), "steps": ("INT", {"default": 20, "min": 1, "max": 10000, "tooltip": "The number of steps used in the denoising process."}), "cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0, "step": 0.1, "round": 0.01, "tooltip": "Classifier-Free Guidance scale."}), "seed_count": ("INT", {"default": 1, "min": 1, "max": 1000, "tooltip": "The number of seeds to generate images with."}), "seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff, "control_after_generate": True, "tooltip": "The starting random seed. It will be incremented for each image in the batch."}), "sampler_name": (comfy.samplers.KSampler.SAMPLERS, {"tooltip": "The algorithm used when sampling."}), "scheduler": (comfy.samplers.KSampler.SCHEDULERS, {"tooltip": "The scheduler controls how noise is gradually removed."}), "width": ("INT", {"default": 512, "min": 16, "max": MAX_RESOLUTION, "step": 8, "tooltip": "The width of the generated image in pixels (used when denoise=1.0)."}), "height": ("INT", {"default": 512, "min": 16, "max": MAX_RESOLUTION, "step": 8, "tooltip": "The height of the generated image in pixels (used when denoise=1.0)."}), "noise_type": (["vanilla", "spectral-diverse", "hierarchical"], { "default": "vanilla", "tooltip": "🎲 Noise Generation Method:\n\n" "• vanilla: Standard ComfyUI noise (torch.randn) - reliable baseline\n" "• spectral-diverse: Frequency-controlled noise with pink/blue/hybrid patterns - enhanced diversity\n" "• hierarchical: Multi-scale latent-aware noise with statistical modeling - maximum quality\n\n" "Advanced methods produce more diverse and potentially higher quality results." }), "injected_noise": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01, "round": 0.01, "tooltip": "Strength of noise injection for variation generation. 0.0=disabled, >0.0=blend base and variation noise."}), "vertical_splits": ("INT", {"default": 1, "min": 1, "max": 8, "step": 1, "tooltip": "Number of vertical splits (1 = no splitting, 2+ = process image in tiles)"}), "horizontal_splits": ("INT", {"default": 1, "min": 1, "max": 8, "step": 1, "tooltip": "Number of horizontal splits (1 = no splitting, 2+ = process image in tiles)"}), "overlap": ("INT", {"default": 64, "min": 0, "max": 256, "step": 8, "tooltip": "Fixed pixel overlap for tile borders (multiple of 8 recommended). Helps reduce seam artifacts between tiles."}), "tile_supersampling": ("FLOAT", {"default": 1.0, "min": 1.0, "max": 4.0, "step": 0.1, "tooltip": "Supersampling factor for each tile before processing. The tile is scaled by this amount, processed, and then scaled back down."}), "supersampling_min_resolution": ("INT", {"default": 512, "min": 256, "max": 4096, "step": 64, "tooltip": "The minimum resolution for the longest side of a tile after supersampling. If smaller, the tile will be upscaled to this size."}), "supersampling_max_resolution": ("INT", {"default": 2048, "min": 256, "max": 8192, "step": 64, "tooltip": "The maximum resolution for the longest side of a tile after supersampling. If larger, the tile will be downscaled to this size."}), }, "optional": { "input_image": ("IMAGE", {"tooltip": "Input image for img2img (used when denoise<1.0)."}), "dd_enabled": ("BOOLEAN", {"default": False, "label_on": "DD Enabled", "label_off": "DD Disabled"}), "detail_amount": ("FLOAT", {"default": 0.1, "min": -5.0, "max": 5.0, "step": 0.01, "tooltip": "Overall strength of the detail adjustment."}), "dd_start": ("FLOAT", {"default": 0.2, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Start of the adjustment curve as a fraction of total steps."}), "dd_end": ("FLOAT", {"default": 0.8, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "End of the adjustment curve as a fraction of total steps."}), "dd_bias": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Curve bias; >0.5 peaks later, <0.5 peaks earlier."}), "dd_exponent": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.05, "tooltip": "Exponent for the curve shape."}), "dd_start_offset": ("FLOAT", {"default": 0.0, "min": -1.0, "max": 1.0, "step": 0.01, "tooltip": "Adjustment multiplier before the curve starts."}), "dd_end_offset": ("FLOAT", {"default": 0.0, "min": -1.0, "max": 1.0, "step": 0.01, "tooltip": "Adjustment multiplier after the curve ends."}), "dd_fade": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.05, "tooltip": "Fade the entire effect in or out."}), "dd_smooth": ("BOOLEAN", {"default": True, "tooltip": "Apply smoothing to the curve."}), } } RETURN_TYPES = ("IMAGE",) OUTPUT_TOOLTIPS = ("Generated images, one for each seed.",) FUNCTION = "sample_integrated" CATEGORY = "pirog/sampling" DESCRIPTION = "Integrated sampling pipeline: Creates or encodes latent → multi-seed sampling → decodes to images. For denoise=1.0 uses width/height to create empty latent. For denoise<1.0 uses input_image." def slerp(self, val, low, high): """Spherical linear interpolation for noise blending""" dims = low.shape low = low.reshape(dims[0], -1) high = high.reshape(dims[0], -1) low_norm = low / torch.norm(low, dim=1, keepdim=True) high_norm = high / torch.norm(high, dim=1, keepdim=True) low_norm[low_norm != low_norm] = 0.0 high_norm[high_norm != high_norm] = 0.0 omega = torch.acos((low_norm * high_norm).sum(1)) so = torch.sin(omega) res = (torch.sin((1.0 - val) * omega) / so).unsqueeze(1) * low + (torch.sin(val * omega) / so).unsqueeze(1) * high return res.reshape(dims) def split_image_into_tiles(self, image, vertical_splits, horizontal_splits, overlap): """Split image into tiles with expansion borders and generate proper gradient masks for img2img""" _, height, width, channels = image.shape tiles = [] masks = [] bboxes = [] # Use linspace to create perfectly spaced coordinates, handling any image size y_coords = np.linspace(0, height, vertical_splits + 1, dtype=int) x_coords = np.linspace(0, width, horizontal_splits + 1, dtype=int) def create_gradient_mask(tile_w, tile_h, core_w, core_h, core_x, core_y, device): """Create a 2D gradient mask for seamless tile blending.""" # Create horizontal gradient (ramp) h_mask = torch.ones((1, tile_w), dtype=torch.float32, device=device) if core_x > 0: # Left expansion left_ramp = torch.linspace(0.0, 1.0, steps=core_x, device=device) h_mask[0, :core_x] = left_ramp if core_x + core_w < tile_w: # Right expansion right_ramp = torch.linspace(1.0, 0.0, steps=tile_w - (core_x + core_w), device=device) h_mask[0, core_x + core_w:] = right_ramp # Create vertical gradient (ramp) v_mask = torch.ones((tile_h, 1), dtype=torch.float32, device=device) if core_y > 0: # Top expansion top_ramp = torch.linspace(0.0, 1.0, steps=core_y, device=device) v_mask[:core_y, 0] = top_ramp if core_y + core_h < tile_h: # Bottom expansion bottom_ramp = torch.linspace(1.0, 0.0, steps=tile_h - (core_y + core_h), device=device) v_mask[core_y + core_h:, 0] = bottom_ramp # Combine gradients # The multiplication will correctly form corners and edges. return v_mask * h_mask for i in range(vertical_splits): for j in range(horizontal_splits): y1, y2 = y_coords[i], y_coords[i+1] x1, x2 = x_coords[j], x_coords[j+1] core_h = y2 - y1 core_w = x2 - x1 # Use a fixed pixel overlap for expansion expansion_h = overlap expansion_w = overlap y_start = max(0, y1 - expansion_h) y_end = min(height, y2 + expansion_h) x_start = max(0, x1 - expansion_w) x_end = min(width, x2 + expansion_w) # Skip tiles that have no area if y_start >= y_end or x_start >= x_end: continue tile = image[:, y_start:y_end, x_start:x_end, :] tiles.append(tile) orig_x_in_tile = x1 - x_start orig_y_in_tile = y1 - y_start actual_tile_h = y_end - y_start actual_tile_w = x_end - x_start gradient_mask = create_gradient_mask(actual_tile_w, actual_tile_h, core_w, core_h, orig_x_in_tile, orig_y_in_tile, image.device) masks.append(gradient_mask) bbox_info = { 'orig_x': x1, 'orig_y': y1, 'orig_w': core_w, 'orig_h': core_h, 'exp_x_start': x_start, 'exp_y_start': y_start, 'exp_x_end': x_end, 'exp_y_end': y_end, 'tile_index': i * horizontal_splits + j } bboxes.append(bbox_info) # Return lists of tensors, as they may have different sizes return tiles, masks, bboxes def reassemble_tiles(self, original_image, processed_tiles, tile_masks, bboxes, vertical_splits, horizontal_splits): """Reassemble tiles using a robust weighted accumulation method for seamless blending.""" _, height, width, channels = original_image.shape device = original_image.device dtype = original_image.dtype # Initialize accumulation tensors result_acc = torch.zeros_like(original_image) weight_sum_acc = torch.zeros((1, height, width, 1), device=device, dtype=dtype) for i, (processed_tile, mask) in enumerate(zip(processed_tiles, tile_masks)): bbox_info = bboxes[i] exp_y_start, exp_y_end = bbox_info['exp_y_start'], bbox_info['exp_y_end'] exp_x_start, exp_x_end = bbox_info['exp_x_start'], bbox_info['exp_x_end'] orig_tile_h = exp_y_end - exp_y_start orig_tile_w = exp_x_end - exp_x_start # The mask should always match the original tile dimensions. # VAE encode/decode can sometimes slightly alter dimensions, so we ensure mask matches. if mask.shape[0] != orig_tile_h or mask.shape[1] != orig_tile_w: mask = torch.nn.functional.interpolate( mask.unsqueeze(0).unsqueeze(0), size=(orig_tile_h, orig_tile_w), mode='bilinear' ).squeeze(0).squeeze(0) # Expand mask to have a channel dimension for broadcasting mask_expanded = mask.unsqueeze(-1) # Accumulate weighted pixel data result_acc[0, exp_y_start:exp_y_end, exp_x_start:exp_x_end, :] += processed_tile[0] * mask_expanded # Accumulate weights weight_sum_acc[0, exp_y_start:exp_y_end, exp_x_start:exp_x_end, :] += mask_expanded # Avoid division by zero by clamping weights weight_sum_clamped = torch.clamp(weight_sum_acc, min=1e-6) # Normalize the accumulated result final_result = result_acc / weight_sum_clamped # Where no tile contributed (weight is near zero), use the original image content untouched_mask = weight_sum_acc <= 1e-6 final_result[untouched_mask.expand_as(final_result)] = original_image[untouched_mask.expand_as(original_image)] return final_result def scale_image_tensor(self, image_tensor, scale_multiplier, min_res, max_res, resolution_step=8): """Proportionally scales a torch image tensor (B, H, W, C) with resolution limits.""" if scale_multiplier == 1.0: return image_tensor _, orig_h, orig_w, _ = image_tensor.shape # Calculate initial scaled dimensions new_w, new_h = int(orig_w * scale_multiplier), int(orig_h * scale_multiplier) # Apply min/max resolution limits based on the longest side longest_side = max(new_w, new_h) if longest_side < min_res: scale_factor = min_res / longest_side new_w, new_h = int(new_w * scale_factor), int(new_h * scale_factor) elif longest_side > max_res: scale_factor = max_res / longest_side new_w, new_h = int(new_w * scale_factor), int(new_h * scale_factor) # Ensure dimensions are a multiple of resolution_step new_w = max(1, round(new_w / resolution_step)) * resolution_step new_h = max(1, round(new_h / resolution_step)) * resolution_step if new_w == orig_w and new_h == orig_h: return image_tensor # Permute to BCHW for interpolate, scale, and permute back to BHWC scaled_tensor = torch.nn.functional.interpolate( image_tensor.permute(0, 3, 1, 2), size=(new_h, new_w), mode='bicubic', align_corners=False ) return scaled_tensor.permute(0, 2, 3, 1) def sample_integrated(self, model, vae, positive, negative, denoise, steps, cfg, seed_count, seed, sampler_name, scheduler, width, height, noise_type="vanilla", injected_noise=0.0, vertical_splits=1, horizontal_splits=1, overlap=64, tile_supersampling=1.0, supersampling_min_resolution=512, supersampling_max_resolution=2048, input_image=None, dd_enabled=False, detail_amount=0.1, dd_start=0.2, dd_end=0.8, dd_bias=0.5, dd_exponent=1.0, dd_start_offset=0.0, dd_end_offset=0.0, dd_fade=0.0, dd_smooth=True): # Determine if we're doing txt2img (denoise=1.0) or img2img (denoise<1.0) is_txt2img = abs(denoise - 1.0) < 0.001 # Check if we need tiling use_tiling = vertical_splits > 1 or horizontal_splits > 1 # --- Create a dictionary for Detail Daemon parameters for easy passing --- dd_params = { "dd_enabled": dd_enabled, "detail_amount": detail_amount, "dd_start": dd_start, "dd_end": dd_end, "dd_bias": dd_bias, "dd_exponent": dd_exponent, "dd_start_offset": dd_start_offset, "dd_end_offset": dd_end_offset, "dd_fade": dd_fade, "dd_smooth": dd_smooth, } if is_txt2img: # Create latent image (txt2img pipeline) if input_image is not None: pass # Generate initial latent - detect required channel count from model try: # Try to get the required latent channels from the model if hasattr(model, 'model') and hasattr(model.model, 'latent_format'): latent_channels = model.model.latent_format.latent_channels else: # Fallback: use 4 channels for SD/SDXL models latent_channels = 4 # Additional fallback based on model type detection if latent_channels is None or latent_channels <= 0: # Try to detect if it's a Flux model by checking model properties try: # Check if this might be a Flux model based on common indicators model_config = getattr(model.model, 'model_config', {}) if hasattr(model.model, 'diffusion_model'): # Check for Flux-specific architecture indicators diff_model = model.model.diffusion_model if (hasattr(diff_model, 'x_embedder') or getattr(model_config, 'unet_config', {}).get('in_channels') == 16): latent_channels = 16 # Flux models use 16 channels else: latent_channels = 4 # SD/SDXL models use 4 channels else: latent_channels = 4 # Default fallback except: latent_channels = 4 # Safe fallback except: # Final fallback latent_channels = 4 latent_shape = [1, latent_channels, height // 8, width // 8] latent = torch.zeros(latent_shape, device=self.device) latent_image = {"samples": latent} else: # Encode input image (img2img pipeline) if input_image is None: raise ValueError("input_image is required when denoise < 1.0 (img2img mode)") # Use vanilla VAE encode approach encoded_tensor = vae.encode(input_image[:,:,:,:3]) latent_image = {"samples": encoded_tensor} # Handle tiling or standard processing if use_tiling and not is_txt2img: # Tiled img2img processing return self._process_tiled_img2img(model, vae, positive, negative, denoise, steps, cfg, seed_count, seed, sampler_name, scheduler, noise_type, injected_noise, input_image, vertical_splits, horizontal_splits, overlap, tile_supersampling, supersampling_min_resolution, supersampling_max_resolution, dd_params) else: # Standard processing (txt2img or non-tiled img2img) return self._process_standard(model, vae, positive, negative, latent_image, denoise, steps, cfg, seed_count, seed, sampler_name, scheduler, noise_type, injected_noise, is_txt2img, dd_params) def _process_standard(self, model, vae, positive, negative, latent_image, denoise, steps, cfg, seed_count, seed, sampler_name, scheduler, noise_type, injected_noise, is_txt2img, dd_params): """Standard processing without tiling""" output_latents = [] input_latents = latent_image["samples"] # Prepare sampler and sigmas once if needed for noise injection or Detail Daemon sigmas = None if injected_noise > 0.0 or dd_params.get("dd_enabled", False): device = comfy.model_management.get_torch_device() comfy.model_management.load_model_gpu(model) sampler = comfy.samplers.KSampler(model, steps=steps, device=device, sampler=sampler_name, scheduler=scheduler, denoise=1.0, model_options=model.model_options) sigmas = sampler.sigmas # Prepare model wrapper if Detail Daemon is enabled model_to_use = model if dd_params.get("dd_enabled", False) and sigmas is not None: def dds_make_schedule(schedule_steps): return make_detail_daemon_schedule( schedule_steps, dd_params["dd_start"], dd_params["dd_end"], dd_params["dd_bias"], dd_params["detail_amount"], dd_params["dd_exponent"], dd_params["dd_start_offset"], dd_params["dd_end_offset"], dd_params["dd_fade"], dd_params["dd_smooth"] ) model_to_use = DetailDaemonModelWrapper(model, sigmas, cfg, dds_make_schedule) total_generations = input_latents.shape[0] * seed_count pbar = comfy.utils.ProgressBar(total_generations) for latent_sample in input_latents: for i in range(seed_count): current_seed = seed + i # For advanced noise methods in txt2img mode, replace zeros with our noise if is_txt2img and noise_type != "vanilla": # Get the shape of the current latent sample base_latent = latent_sample.unsqueeze(0) try: if noise_type == "spectral-diverse": # Generate noise directly on the target device with exact shape noise = create_spectral_diverse_noise( base_latent.shape, current_seed, device=base_latent.device, noise_type="hybrid") # Ensure exact compatibility current_latent = noise.to(device=base_latent.device, dtype=base_latent.dtype) elif noise_type == "hierarchical": # Generate noise directly on the target device with exact shape noise = create_hierarchical_noise( base_latent.shape, current_seed, device=base_latent.device, diversity_strength=1.0) # Ensure exact compatibility current_latent = noise.to(device=base_latent.device, dtype=base_latent.dtype) else: current_latent = base_latent except Exception: # Fallback to vanilla noise if advanced noise generation fails current_latent = base_latent work_latent = current_latent else: # Use the existing latent (vanilla mode or img2img) work_latent = latent_sample.unsqueeze(0) # Apply noise injection if enabled if injected_noise > 0.0: # Generate base noise from main seed generator = torch.manual_seed(seed) base_noise = torch.randn(work_latent.shape, generator=generator, device="cpu", dtype=work_latent.dtype) # Generate variation noise from current seed (naturally varies per iteration) generator = torch.manual_seed(current_seed + 12345) # Offset to ensure different noise variation_noise = torch.randn(work_latent.shape, generator=generator, device="cpu", dtype=work_latent.dtype) # Blend noises using SLERP blended_noise = self.slerp(injected_noise, base_noise, variation_noise) # Calculate sigma scaling factor device = comfy.model_management.get_torch_device() comfy.model_management.load_model_gpu(model) sampler = comfy.samplers.KSampler(model, steps=steps, device=device, sampler=sampler_name, scheduler=scheduler, denoise=1.0, model_options=model.model_options) sigmas = sampler.sigmas start_at_step = round(steps - steps * denoise) end_at_step = steps sigma = sigmas[start_at_step] - sigmas[end_at_step] sigma /= model.model.latent_format.scale_factor sigma = sigma.detach().cpu().item() # Apply scaled noise to latent work_latent = work_latent + blended_noise.to(work_latent.device) * sigma latent_for_sampler = {"samples": work_latent} # Call the common ksampler function from the main ComfyUI nodes # This ensures we always use the latest version result_latent, = common_ksampler(model_to_use, current_seed, steps, cfg, sampler_name, scheduler, positive, negative, latent_for_sampler, denoise=denoise) output_latents.append(result_latent["samples"]) pbar.update(1) # Combine all the generated latents into a single batch tensor final_samples = torch.cat(output_latents, dim=0) # Decode latents to images using vanilla VAE decode approach images = vae.decode(final_samples) if len(images.shape) == 5: # Combine batches images = images.reshape(-1, images.shape[-3], images.shape[-2], images.shape[-1]) return (images,) def _prepare_inpaint_conditioning(self, conditioning, concat_latent, latent_mask): c = [] for t in conditioning: # Create a deep enough copy to avoid modifying the original conditioning n = [t[0], t[1].copy()] n[1]["concat_latent_image"] = concat_latent # The mask is inverted for ComfyUI's inpainting model n[1]["concat_mask"] = 1.0 - latent_mask c.append(n) return c def _process_tiled_img2img(self, model, vae, positive, negative, denoise, steps, cfg, seed_count, seed, sampler_name, scheduler, noise_type, injected_noise, input_image, vertical_splits, horizontal_splits, overlap, tile_supersampling=1.0, supersampling_min_resolution=512, supersampling_max_resolution=2048, dd_params=None): """Tiled img2img processing with proper batch support.""" if input_image is None: raise ValueError("input_image is required for tiled processing") all_results = [] # STEP 1: UPSCALE THE ENTIRE IMAGE BATCH FIRST if tile_supersampling > 1.0: image_to_process = self.scale_image_tensor( input_image, tile_supersampling, supersampling_min_resolution, supersampling_max_resolution ) else: image_to_process = input_image # The main loop to iterate over each image in the input batch for i in range(image_to_process.shape[0]): single_image = image_to_process[i:i+1] # Keep it as a batch of 1 [1, H, W, C] # Now, run the multi-seed process on this single image for seed_idx in range(seed_count): current_seed = seed + seed_idx # --- Detail Daemon model wrapping per-seed, as sigmas can vary per tile --- # We can't pre-wrap the model for tiling because sigmas might differ for each tile's sampler instance. # Instead, we will wrap the model inside the tile loop. # STEP 2: SPLIT THE (POTENTIALLY UPSCALED) IMAGE tiles, tile_masks, bboxes = self.split_image_into_tiles(single_image, vertical_splits, horizontal_splits, overlap) processed_tiles = [] # Process each tile for the current image and seed for tile_idx, tile in enumerate(tiles): # Store original tile dimensions _, orig_tile_h, orig_tile_w, _ = tile.shape # --- VAE COMPLIANCE: PRE-PADDING --- # Calculate required padding to make dimensions divisible by 8 pad_h = (8 - orig_tile_h % 8) % 8 pad_w = (8 - orig_tile_w % 8) % 8 # Pad the tile if necessary. We use replicate padding to avoid hard edges. if pad_h > 0 or pad_w > 0: # Pytorch's pad format is (left, right, top, bottom) padded_tile = torch.nn.functional.pad(tile.permute(0, 3, 1, 2), (0, pad_w, 0, pad_h), mode='replicate').permute(0, 2, 3, 1) else: padded_tile = tile # Encode the compliant tile concat_latent = vae.encode(padded_tile) # Apply noise injection if enabled if injected_noise > 0.0: generator = torch.manual_seed(seed) base_noise = torch.randn(concat_latent.shape, generator=generator, device="cpu", dtype=concat_latent.dtype) generator = torch.manual_seed(current_seed + tile_idx + 12345) variation_noise = torch.randn(concat_latent.shape, generator=generator, device="cpu", dtype=concat_latent.dtype) blended_noise = self.slerp(injected_noise, base_noise, variation_noise) device = comfy.model_management.get_torch_device() comfy.model_management.load_model_gpu(model) sampler = comfy.samplers.KSampler(model, steps=steps, device=device, sampler=sampler_name, scheduler=scheduler, denoise=1.0, model_options=model.model_options) sigmas = sampler.sigmas start_at_step = round(steps - steps * denoise) end_at_step = steps sigma = sigmas[start_at_step] - sigmas[end_at_step] sigma /= model.model.latent_format.scale_factor sigma = sigma.detach().cpu().item() concat_latent = concat_latent + blended_noise.to(concat_latent.device) * sigma # The noise mask for the sampler MUST be all ones to allow full generation in overlap zones. latent_mask = torch.ones_like(concat_latent[:, :1, :, :]) # Set up the latent structure for the sampler latent_for_sampler = { "samples": concat_latent, "noise_mask": latent_mask # Use the ones mask } # --- Detail Daemon model wrapping for this specific tile --- model_to_use = model if dd_params and dd_params.get("dd_enabled", False): device = comfy.model_management.get_torch_device() comfy.model_management.load_model_gpu(model) sampler = comfy.samplers.KSampler(model, steps=steps, device=device, sampler=sampler_name, scheduler=scheduler, denoise=denoise, model_options=model.model_options) sigmas = sampler.sigmas def dds_make_schedule(schedule_steps): return make_detail_daemon_schedule( schedule_steps, dd_params["dd_start"], dd_params["dd_end"], dd_params["dd_bias"], dd_params["detail_amount"], dd_params["dd_exponent"], dd_params["dd_start_offset"], dd_params["dd_end_offset"], dd_params["dd_fade"], dd_params["dd_smooth"] ) model_to_use = DetailDaemonModelWrapper(model, sigmas, cfg, dds_make_schedule) # Set inpainting-specific conditioning info positive_inp = self._prepare_inpaint_conditioning(positive, concat_latent, latent_mask) negative_inp = self._prepare_inpaint_conditioning(negative, concat_latent, latent_mask) # Sample tile result_latent, = common_ksampler( model_to_use, current_seed + tile_idx, steps, cfg, sampler_name, scheduler, positive_inp, negative_inp, latent_for_sampler, denoise=denoise ) # Decode tile back to image decoded_image = vae.decode(result_latent["samples"]) # --- VAE COMPLIANCE: POST-CROP --- # Crop the decoded image back to the ORIGINAL tile dimensions # This removes the padding and restores pixel-perfect alignment. tile_image = decoded_image[:, :orig_tile_h, :orig_tile_w, :] processed_tiles.append(tile_image) # Reassemble the tiles for the current image and seed result_image = self.reassemble_tiles(single_image, processed_tiles, tile_masks, bboxes, vertical_splits, horizontal_splits) all_results.append(result_image) # Stack all final results from all images and seeds into a single batch return (torch.cat(all_results, dim=0),)