| import torch
|
| from PIL import Image
|
| from comfy.cli_args import args, LatentPreviewMethod
|
| from comfy.taesd.taesd import TAESD
|
| import comfy.model_management
|
| import folder_paths
|
| import comfy.utils
|
| import logging
|
|
|
| MAX_PREVIEW_RESOLUTION = args.preview_size
|
|
|
| def preview_to_image(latent_image):
|
| latents_ubyte = (((latent_image + 1.0) / 2.0).clamp(0, 1)
|
| .mul(0xFF)
|
| )
|
| if comfy.model_management.directml_enabled:
|
| latents_ubyte = latents_ubyte.to(dtype=torch.uint8)
|
| latents_ubyte = latents_ubyte.to(device="cpu", dtype=torch.uint8, non_blocking=comfy.model_management.device_supports_non_blocking(latent_image.device))
|
|
|
| return Image.fromarray(latents_ubyte.numpy())
|
|
|
| class LatentPreviewer:
|
| def decode_latent_to_preview(self, x0):
|
| pass
|
|
|
| def decode_latent_to_preview_image(self, preview_format, x0):
|
| preview_image = self.decode_latent_to_preview(x0)
|
| return ("JPEG", preview_image, MAX_PREVIEW_RESOLUTION)
|
|
|
| class TAESDPreviewerImpl(LatentPreviewer):
|
| def __init__(self, taesd):
|
| self.taesd = taesd
|
|
|
| def decode_latent_to_preview(self, x0):
|
| x_sample = self.taesd.decode(x0[:1])[0].movedim(0, 2)
|
| return preview_to_image(x_sample)
|
|
|
|
|
| class Latent2RGBPreviewer(LatentPreviewer):
|
| def __init__(self, latent_rgb_factors, latent_rgb_factors_bias=None):
|
| self.latent_rgb_factors = torch.tensor(latent_rgb_factors, device="cpu").transpose(0, 1)
|
| self.latent_rgb_factors_bias = None
|
| if latent_rgb_factors_bias is not None:
|
| self.latent_rgb_factors_bias = torch.tensor(latent_rgb_factors_bias, device="cpu")
|
|
|
| def decode_latent_to_preview(self, x0):
|
| self.latent_rgb_factors = self.latent_rgb_factors.to(dtype=x0.dtype, device=x0.device)
|
| if self.latent_rgb_factors_bias is not None:
|
| self.latent_rgb_factors_bias = self.latent_rgb_factors_bias.to(dtype=x0.dtype, device=x0.device)
|
|
|
| if x0.ndim == 5:
|
| x0 = x0[0, :, 0]
|
| else:
|
| x0 = x0[0]
|
|
|
| latent_image = torch.nn.functional.linear(x0.movedim(0, -1), self.latent_rgb_factors, bias=self.latent_rgb_factors_bias)
|
|
|
|
|
| return preview_to_image(latent_image)
|
|
|
|
|
| def get_previewer(device, latent_format):
|
| previewer = None
|
| method = args.preview_method
|
| if method != LatentPreviewMethod.NoPreviews:
|
|
|
| taesd_decoder_path = None
|
| if latent_format.taesd_decoder_name is not None:
|
| taesd_decoder_path = next(
|
| (fn for fn in folder_paths.get_filename_list("vae_approx")
|
| if fn.startswith(latent_format.taesd_decoder_name)),
|
| ""
|
| )
|
| taesd_decoder_path = folder_paths.get_full_path("vae_approx", taesd_decoder_path)
|
|
|
| if method == LatentPreviewMethod.Auto:
|
| method = LatentPreviewMethod.Latent2RGB
|
|
|
| if method == LatentPreviewMethod.TAESD:
|
| if taesd_decoder_path:
|
| taesd = TAESD(None, taesd_decoder_path, latent_channels=latent_format.latent_channels).to(device)
|
| previewer = TAESDPreviewerImpl(taesd)
|
| else:
|
| logging.warning("Warning: TAESD previews enabled, but could not find models/vae_approx/{}".format(latent_format.taesd_decoder_name))
|
|
|
| if previewer is None:
|
| if latent_format.latent_rgb_factors is not None:
|
| previewer = Latent2RGBPreviewer(latent_format.latent_rgb_factors, latent_format.latent_rgb_factors_bias)
|
| return previewer
|
|
|
| def prepare_callback(model, steps, x0_output_dict=None):
|
| preview_format = "JPEG"
|
| if preview_format not in ["JPEG", "PNG"]:
|
| preview_format = "JPEG"
|
|
|
| previewer = get_previewer(model.load_device, model.model.latent_format)
|
|
|
| pbar = comfy.utils.ProgressBar(steps)
|
| def callback(step, x0, x, total_steps):
|
| if x0_output_dict is not None:
|
| x0_output_dict["x0"] = x0
|
|
|
| preview_bytes = None
|
| if previewer:
|
| preview_bytes = previewer.decode_latent_to_preview_image(preview_format, x0)
|
| pbar.update_absolute(step + 1, total_steps, preview_bytes)
|
| return callback
|
|
|
|
|
