if mem_eff_attn:

logger.info("Enable memory efficient attention for U-Net")

# これはDiffusersのU-Netではなく自前のU-Netなので置き換えなくても良い

unet.set_use_memory_efficient_attention(False, True)

elif xformers:

logger.info("Enable xformers for U-Net")

try:

import xformers.ops

except ImportError:

raise ImportError("No xformers / xformersがインストールされていないようです")

unet.set_use_memory_efficient_attention(True, False)

elif sdpa:

logger.info("Enable SDPA for U-Net")

unet.set_use_memory_efficient_attention(False, False)

if mem_eff_attn:

replace_vae_attn_to_memory_efficient()

elif xformers:

replace_vae_attn_to_xformers()

elif sdpa:

logger.info("VAE Attention.forward has been replaced to FlashAttention (not xformers)")

flash_func = FlashAttentionFunction

def forward_flash_attn(self, hidden_states, **kwargs):

q_bucket_size = 512

k_bucket_size = 1024

residual = hidden_states

batch, channel, height, width = hidden_states.shape

# norm

hidden_states = self.group_norm(hidden_states)

hidden_states = hidden_states.view(batch, channel, height * width).transpose(1, 2)

# proj to q, k, v

query_proj = self.to_q(hidden_states)

key_proj = self.to_k(hidden_states)

value_proj = self.to_v(hidden_states)

query_proj, key_proj, value_proj = map(

lambda t: rearrange(t, "b n (h d) -> b h n d", h=self.heads), (query_proj, key_proj, value_proj)

)

out = flash_func.apply(query_proj, key_proj, value_proj, None, False, q_bucket_size, k_bucket_size)

out = rearrange(out, "b h n d -> b n (h d)")

# compute next hidden_states

# linear proj

hidden_states = self.to_out[0](hidden_states)

# dropout

hidden_states = self.to_out[1](hidden_states)

hidden_states = hidden_states.transpose(-1, -2).reshape(batch, channel, height, width)

# res connect and rescale

hidden_states = (hidden_states + residual) / self.rescale_output_factor

return hidden_states

def forward_flash_attn_0_14(self, hidden_states, **kwargs):

if not hasattr(self, "to_q"):

self.to_q = self.query

self.to_k = self.key

self.to_v = self.value

self.to_out = [self.proj_attn, torch.nn.Identity()]

self.heads = self.num_heads

return forward_flash_attn(self, hidden_states, **kwargs)

if diffusers.__version__ < "0.15.0":

diffusers.models.attention.AttentionBlock.forward = forward_flash_attn_0_14

else:

logger.info("VAE: Attention.forward has been replaced to xformers")

import xformers.ops

def forward_xformers(self, hidden_states, **kwargs):

residual = hidden_states

batch, channel, height, width = hidden_states.shape

# norm

hidden_states = self.group_norm(hidden_states)

hidden_states = hidden_states.view(batch, channel, height * width).transpose(1, 2)

# proj to q, k, v

query_proj = self.to_q(hidden_states)

key_proj = self.to_k(hidden_states)

value_proj = self.to_v(hidden_states)

query_proj, key_proj, value_proj = map(

lambda t: rearrange(t, "b n (h d) -> b h n d", h=self.heads), (query_proj, key_proj, value_proj)

)

query_proj = query_proj.contiguous()

key_proj = key_proj.contiguous()

value_proj = value_proj.contiguous()

out = xformers.ops.memory_efficient_attention(query_proj, key_proj, value_proj, attn_bias=None)

out = rearrange(out, "b h n d -> b n (h d)")

# compute next hidden_states

# linear proj

hidden_states = self.to_out[0](hidden_states)

# dropout

hidden_states = self.to_out[1](hidden_states)

hidden_states = hidden_states.transpose(-1, -2).reshape(batch, channel, height, width)

# res connect and rescale

hidden_states = (hidden_states + residual) / self.rescale_output_factor

return hidden_states

def forward_xformers_0_14(self, hidden_states, **kwargs):

if not hasattr(self, "to_q"):

self.to_q = self.query

self.to_k = self.key

self.to_v = self.value

self.to_out = [self.proj_attn, torch.nn.Identity()]

self.heads = self.num_heads

return forward_xformers(self, hidden_states, **kwargs)

if diffusers.__version__ < "0.15.0":

diffusers.models.attention.AttentionBlock.forward = forward_xformers_0_14

else:

logger.info("VAE: Attention.forward has been replaced to sdpa")

def forward_sdpa(self, hidden_states, **kwargs):

residual = hidden_states

batch, channel, height, width = hidden_states.shape

# norm

hidden_states = self.group_norm(hidden_states)

hidden_states = hidden_states.view(batch, channel, height * width).transpose(1, 2)

# proj to q, k, v

query_proj = self.to_q(hidden_states)

key_proj = self.to_k(hidden_states)

value_proj = self.to_v(hidden_states)

query_proj, key_proj, value_proj = map(

lambda t: rearrange(t, "b n (h d) -> b n h d", h=self.heads), (query_proj, key_proj, value_proj)

)

out = torch.nn.functional.scaled_dot_product_attention(

query_proj, key_proj, value_proj, attn_mask=None, dropout_p=0.0, is_causal=False

)

out = rearrange(out, "b n h d -> b n (h d)")

# compute next hidden_states

# linear proj

hidden_states = self.to_out[0](hidden_states)

# dropout

hidden_states = self.to_out[1](hidden_states)

hidden_states = hidden_states.transpose(-1, -2).reshape(batch, channel, height, width)

# res connect and rescale

hidden_states = (hidden_states + residual) / self.rescale_output_factor

return hidden_states

def forward_sdpa_0_14(self, hidden_states, **kwargs):

if not hasattr(self, "to_q"):

self.to_q = self.query

self.to_k = self.key

self.to_v = self.value

self.to_out = [self.proj_attn, torch.nn.Identity()]

self.heads = self.num_heads

return forward_sdpa(self, hidden_states, **kwargs)

if diffusers.__version__ < "0.15.0":

diffusers.models.attention.AttentionBlock.forward = forward_sdpa_0_14

else:

r"""

def __init__(

self,

device,

vae: AutoencoderKL,

text_encoder: CLIPTextModel,

tokenizer: CLIPTokenizer,

unet: InferUNet2DConditionModel,

scheduler: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler],

clip_skip: int,

clip_model: CLIPModel,

clip_guidance_scale: float,

clip_image_guidance_scale: float,

vgg16_model: torchvision.models.VGG,

vgg16_guidance_scale: float,

vgg16_layer_no: int,

# safety_checker: StableDiffusionSafetyChecker,

# feature_extractor: CLIPFeatureExtractor,

):

super().__init__()

self.device = device

self.clip_skip = clip_skip

if hasattr(scheduler.config, "steps_offset") and scheduler.config.steps_offset != 1:

deprecation_message = (

f"The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"

f" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure "

"to update the config accordingly as leaving `steps_offset` might led to incorrect results"

" in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,"

" it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`"

" file"

)

deprecate("steps_offset!=1", "1.0.0", deprecation_message, standard_warn=False)

new_config = dict(scheduler.config)

new_config["steps_offset"] = 1

scheduler._internal_dict = FrozenDict(new_config)

if hasattr(scheduler.config, "clip_sample") and scheduler.config.clip_sample is True:

deprecation_message = (

f"The configuration file of this scheduler: {scheduler} has not set the configuration `clip_sample`."

" `clip_sample` should be set to False in the configuration file. Please make sure to update the"

" config accordingly as not setting `clip_sample` in the config might lead to incorrect results in"

" future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it would be very"

" nice if you could open a Pull request for the `scheduler/scheduler_config.json` file"

)

deprecate("clip_sample not set", "1.0.0", deprecation_message, standard_warn=False)

new_config = dict(scheduler.config)

new_config["clip_sample"] = False

scheduler._internal_dict = FrozenDict(new_config)

self.vae = vae

self.text_encoder = text_encoder

self.tokenizer = tokenizer

self.unet = unet

self.scheduler = scheduler

self.safety_checker = None

# Textual Inversion

self.token_replacements = {}

# XTI

self.token_replacements_XTI = {}

# CLIP guidance

self.clip_guidance_scale = clip_guidance_scale

self.clip_image_guidance_scale = clip_image_guidance_scale

self.clip_model = clip_model

self.normalize = transforms.Normalize(mean=FEATURE_EXTRACTOR_IMAGE_MEAN, std=FEATURE_EXTRACTOR_IMAGE_STD)

self.make_cutouts = MakeCutouts(FEATURE_EXTRACTOR_SIZE)

# VGG16 guidance

self.vgg16_guidance_scale = vgg16_guidance_scale

if self.vgg16_guidance_scale > 0.0:

return_layers = {f"{vgg16_layer_no}": "feat"}

self.vgg16_feat_model = torchvision.models._utils.IntermediateLayerGetter(

vgg16_model.features, return_layers=return_layers

)

self.vgg16_normalize = transforms.Normalize(mean=VGG16_IMAGE_MEAN, std=VGG16_IMAGE_STD)

# ControlNet

self.control_nets: List[ControlNetInfo] = []

self.control_net_enabled = True # control_netsが空ならTrueでもFalseでもControlNetは動作しない

self.gradual_latent: GradualLatent = None

# Textual Inversion

def add_token_replacement(self, target_token_id, rep_token_ids):

self.token_replacements[target_token_id] = rep_token_ids

def set_enable_control_net(self, en: bool):

self.control_net_enabled = en

def replace_token(self, tokens, layer=None):

new_tokens = []

for token in tokens:

if token in self.token_replacements:

replacer_ = self.token_replacements[token]

if layer:

replacer = []

for r in replacer_:

if r in self.token_replacements_XTI:

replacer.append(self.token_replacements_XTI[r][layer])

else:

replacer = replacer_

new_tokens.extend(replacer)

else:

new_tokens.append(token)

return new_tokens

def add_token_replacement_XTI(self, target_token_id, rep_token_ids):

self.token_replacements_XTI[target_token_id] = rep_token_ids

def set_control_nets(self, ctrl_nets):

self.control_nets = ctrl_nets

def set_gradual_latent(self, gradual_latent):

if gradual_latent is None:

logger.info("gradual_latent is disabled")

self.gradual_latent = None

else:

logger.info(f"gradual_latent is enabled: {gradual_latent}")

self.gradual_latent = gradual_latent # (ds_ratio, start_timesteps, every_n_steps, ratio_step)

# region xformersとか使う部分：独自に書き換えるので関係なし

def enable_xformers_memory_efficient_attention(self):

r"""

self.unet.set_use_memory_efficient_attention_xformers(True)

def disable_xformers_memory_efficient_attention(self):

r"""

self.unet.set_use_memory_efficient_attention_xformers(False)

def enable_attention_slicing(self, slice_size: Optional[Union[str, int]] = "auto"):

r"""

if slice_size == "auto":

# half the attention head size is usually a good trade-off between

# speed and memory

slice_size = self.unet.config.attention_head_dim // 2

self.unet.set_attention_slice(slice_size)

def disable_attention_slicing(self):

r"""

# set slice_size = `None` to disable `attention slicing`

self.enable_attention_slicing(None)

def enable_sequential_cpu_offload(self):

r"""

# accelerateが必要になるのでとりあえず省略

raise NotImplementedError("cpu_offload is omitted.")

# if is_accelerate_available():

# from accelerate import cpu_offload

# else:

# raise ImportError("Please install accelerate via `pip install accelerate`")

# device = self.device

# for cpu_offloaded_model in [self.unet, self.text_encoder, self.vae, self.safety_checker]:

# if cpu_offloaded_model is not None:

# cpu_offload(cpu_offloaded_model, device)

# endregion

@torch.no_grad()

def __call__(

self,

prompt: Union[str, List[str]],

negative_prompt: Optional[Union[str, List[str]]] = None,

init_image: Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]] = None,

mask_image: Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]] = None,

height: int = 512,

width: int = 512,

num_inference_steps: int = 50,

guidance_scale: float = 7.5,

negative_scale: float = None,

strength: float = 0.8,

# num_images_per_prompt: Optional[int] = 1,

eta: float = 0.0,

generator: Optional[torch.Generator] = None,

latents: Optional[torch.FloatTensor] = None,

max_embeddings_multiples: Optional[int] = 3,

output_type: Optional[str] = "pil",

vae_batch_size: float = None,

return_latents: bool = False,

# return_dict: bool = True,

callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,

is_cancelled_callback: Optional[Callable[[], bool]] = None,

callback_steps: Optional[int] = 1,

img2img_noise=None,

clip_prompts=None,

clip_guide_images=None,

networks: Optional[List[LoRANetwork]] = None,

**kwargs,

):

r"""

num_images_per_prompt = 1 # fixed

if isinstance(prompt, str):

batch_size = 1

prompt = [prompt]

elif isinstance(prompt, list):

batch_size = len(prompt)

else:

raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(prompt)}")

reginonal_network = " AND " in prompt[0]

vae_batch_size = (

batch_size

if vae_batch_size is None

else (int(vae_batch_size) if vae_batch_size >= 1 else max(1, int(batch_size * vae_batch_size)))

)

if strength < 0 or strength > 1:

raise ValueError(f"The value of strength should in [0.0, 1.0] but is {strength}")

if height % 8 != 0 or width % 8 != 0:

raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}.")

if (callback_steps is None) or (

callback_steps is not None and (not isinstance(callback_steps, int) or callback_steps <= 0)

):

raise ValueError(

f"`callback_steps` has to be a positive integer but is {callback_steps} of type" f" {type(callback_steps)}."

)

# get prompt text embeddings

# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)

# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`

# corresponds to doing no classifier free guidance.

do_classifier_free_guidance = guidance_scale > 1.0

if not do_classifier_free_guidance and negative_scale is not None:

logger.warning(f"negative_scale is ignored if guidance scalle <= 1.0")

negative_scale = None

# get unconditional embeddings for classifier free guidance

if negative_prompt is None:

negative_prompt = [""] * batch_size

elif isinstance(negative_prompt, str):

negative_prompt = [negative_prompt] * batch_size

if batch_size != len(negative_prompt):

raise ValueError(

f"`negative_prompt`: {negative_prompt} has batch size {len(negative_prompt)}, but `prompt`:"

f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"

" the batch size of `prompt`."

)

if not self.token_replacements_XTI:

text_embeddings, uncond_embeddings, prompt_tokens = get_weighted_text_embeddings(

pipe=self,

prompt=prompt,

uncond_prompt=negative_prompt if do_classifier_free_guidance else None,

max_embeddings_multiples=max_embeddings_multiples,

clip_skip=self.clip_skip,

**kwargs,

)

if negative_scale is not None:

_, real_uncond_embeddings, _ = get_weighted_text_embeddings(

pipe=self,

prompt=prompt, # こちらのトークン長に合わせてuncondを作るので75トークン超で必須

uncond_prompt=[""] * batch_size,

max_embeddings_multiples=max_embeddings_multiples,

clip_skip=self.clip_skip,

**kwargs,

)

if self.token_replacements_XTI:

text_embeddings_concat = []

for layer in [

"IN01",

"IN02",

"IN04",

"IN05",

"IN07",

"IN08",

"MID",

"OUT03",

"OUT04",

"OUT05",

"OUT06",

"OUT07",

"OUT08",

"OUT09",

"OUT10",

"OUT11",

]:

text_embeddings, uncond_embeddings, prompt_tokens = get_weighted_text_embeddings(

pipe=self,

prompt=prompt,

uncond_prompt=negative_prompt if do_classifier_free_guidance else None,

max_embeddings_multiples=max_embeddings_multiples,

clip_skip=self.clip_skip,

layer=layer,

**kwargs,

)

if do_classifier_free_guidance:

if negative_scale is None:

text_embeddings_concat.append(torch.cat([uncond_embeddings, text_embeddings]))

else:

text_embeddings_concat.append(torch.cat([uncond_embeddings, text_embeddings, real_uncond_embeddings]))

text_embeddings = torch.stack(text_embeddings_concat)

else:

if do_classifier_free_guidance:

if negative_scale is None:

text_embeddings = torch.cat([uncond_embeddings, text_embeddings])

else:

text_embeddings = torch.cat([uncond_embeddings, text_embeddings, real_uncond_embeddings])

# CLIP guidanceで使用するembeddingsを取得する

if self.clip_guidance_scale > 0:

clip_text_input = prompt_tokens

if clip_text_input.shape[1] > self.tokenizer.model_max_length:

# TODO 75文字を超えたら警告を出す？

logger.info(f"trim text input {clip_text_input.shape}")

clip_text_input = torch.cat(

[clip_text_input[:, : self.tokenizer.model_max_length - 1], clip_text_input[:, -1].unsqueeze(1)], dim=1

)

logger.info(f"trimmed {clip_text_input.shape}")

for i, clip_prompt in enumerate(clip_prompts):

if clip_prompt is not None: # clip_promptがあれば上書きする

clip_text_input[i] = self.tokenizer(

clip_prompt,

padding="max_length",

max_length=self.tokenizer.model_max_length,

truncation=True,

return_tensors="pt",

).input_ids.to(self.device)

text_embeddings_clip = self.clip_model.get_text_features(clip_text_input)

text_embeddings_clip = text_embeddings_clip / text_embeddings_clip.norm(p=2, dim=-1, keepdim=True) # prompt複数件でもOK

if (

self.clip_image_guidance_scale > 0

or self.vgg16_guidance_scale > 0

and clip_guide_images is not None

or self.control_nets

):

if isinstance(clip_guide_images, PIL.Image.Image):

clip_guide_images = [clip_guide_images]

if self.clip_image_guidance_scale > 0:

clip_guide_images = [preprocess_guide_image(im) for im in clip_guide_images]

clip_guide_images = torch.cat(clip_guide_images, dim=0)

clip_guide_images = self.normalize(clip_guide_images).to(self.device).to(text_embeddings.dtype)

image_embeddings_clip = self.clip_model.get_image_features(clip_guide_images)

image_embeddings_clip = image_embeddings_clip / image_embeddings_clip.norm(p=2, dim=-1, keepdim=True)

if len(image_embeddings_clip) == 1:

image_embeddings_clip = image_embeddings_clip.repeat((batch_size, 1, 1, 1))

elif self.vgg16_guidance_scale > 0:

size = (width // VGG16_INPUT_RESIZE_DIV, height // VGG16_INPUT_RESIZE_DIV) # とりあえず1/4に（小さいか?）

clip_guide_images = [preprocess_vgg16_guide_image(im, size) for im in clip_guide_images]

clip_guide_images = torch.cat(clip_guide_images, dim=0)

clip_guide_images = self.vgg16_normalize(clip_guide_images).to(self.device).to(text_embeddings.dtype)

image_embeddings_vgg16 = self.vgg16_feat_model(clip_guide_images)["feat"]

if len(image_embeddings_vgg16) == 1:

image_embeddings_vgg16 = image_embeddings_vgg16.repeat((batch_size, 1, 1, 1))

else:

# ControlNetのhintにguide imageを流用する

# 前処理はControlNet側で行う

pass

# set timesteps

self.scheduler.set_timesteps(num_inference_steps, self.device)

latents_dtype = text_embeddings.dtype

init_latents_orig = None

mask = None

if init_image is None:

# get the initial random noise unless the user supplied it

# Unlike in other pipelines, latents need to be generated in the target device

# for 1-to-1 results reproducibility with the CompVis implementation.

# However this currently doesn't work in `mps`.

latents_shape = (

batch_size * num_images_per_prompt,

self.unet.in_channels,

height // 8,

width // 8,

)

if latents is None:

if self.device.type == "mps":

# randn does not exist on mps

latents = torch.randn(

latents_shape,

generator=generator,

device="cpu",

dtype=latents_dtype,

).to(self.device)

else:

latents = torch.randn(

latents_shape,

generator=generator,

device=self.device,

dtype=latents_dtype,

)

else:

if latents.shape != latents_shape:

raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}")

latents = latents.to(self.device)

timesteps = self.scheduler.timesteps.to(self.device)

# scale the initial noise by the standard deviation required by the scheduler

latents = latents * self.scheduler.init_noise_sigma

else:

# image to tensor

if isinstance(init_image, PIL.Image.Image):

init_image = [init_image]

if isinstance(init_image[0], PIL.Image.Image):

init_image = [preprocess_image(im) for im in init_image]

init_image = torch.cat(init_image)

if isinstance(init_image, list):

init_image = torch.stack(init_image)

# mask image to tensor

if mask_image is not None:

if isinstance(mask_image, PIL.Image.Image):

mask_image = [mask_image]

if isinstance(mask_image[0], PIL.Image.Image):

mask_image = torch.cat([preprocess_mask(im) for im in mask_image]) # H*W, 0 for repaint

# encode the init image into latents and scale the latents

init_image = init_image.to(device=self.device, dtype=latents_dtype)

if init_image.size()[-2:] == (height // 8, width // 8):

init_latents = init_image

else:

if vae_batch_size >= batch_size:

init_latent_dist = self.vae.encode(init_image).latent_dist

init_latents = init_latent_dist.sample(generator=generator)

else:

clean_memory()

init_latents = []

for i in tqdm(range(0, min(batch_size, len(init_image)), vae_batch_size)):

init_latent_dist = self.vae.encode(

init_image[i : i + vae_batch_size] if vae_batch_size > 1 else init_image[i].unsqueeze(0)

).latent_dist

init_latents.append(init_latent_dist.sample(generator=generator))

init_latents = torch.cat(init_latents)

init_latents = 0.18215 * init_latents

if len(init_latents) == 1:

init_latents = init_latents.repeat((batch_size, 1, 1, 1))

init_latents_orig = init_latents

# preprocess mask

if mask_image is not None:

mask = mask_image.to(device=self.device, dtype=latents_dtype)

if len(mask) == 1:

mask = mask.repeat((batch_size, 1, 1, 1))

# check sizes

if not mask.shape == init_latents.shape:

raise ValueError("The mask and init_image should be the same size!")

# get the original timestep using init_timestep

offset = self.scheduler.config.get("steps_offset", 0)

init_timestep = int(num_inference_steps * strength) + offset

init_timestep = min(init_timestep, num_inference_steps)

timesteps = self.scheduler.timesteps[-init_timestep]

timesteps = torch.tensor([timesteps] * batch_size * num_images_per_prompt, device=self.device)

# add noise to latents using the timesteps

latents = self.scheduler.add_noise(init_latents, img2img_noise, timesteps)

t_start = max(num_inference_steps - init_timestep + offset, 0)

timesteps = self.scheduler.timesteps[t_start:].to(self.device)

# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature

# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.

# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502

# and should be between [0, 1]

accepts_eta = "eta" in set(inspect.signature(self.scheduler.step).parameters.keys())

extra_step_kwargs = {}

if accepts_eta:

extra_step_kwargs["eta"] = eta

num_latent_input = (3 if negative_scale is not None else 2) if do_classifier_free_guidance else 1

if self.control_nets:

guided_hints = original_control_net.get_guided_hints(self.control_nets, num_latent_input, batch_size, clip_guide_images)

if reginonal_network:

num_sub_and_neg_prompts = len(text_embeddings) // batch_size

# last subprompt and negative prompt

text_emb_last = []

for j in range(batch_size):

text_emb_last.append(text_embeddings[(j + 1) * num_sub_and_neg_prompts - 2])

text_emb_last.append(text_embeddings[(j + 1) * num_sub_and_neg_prompts - 1])

text_emb_last = torch.stack(text_emb_last)

else:

text_emb_last = text_embeddings

enable_gradual_latent = False

if self.gradual_latent:

if not hasattr(self.scheduler, "set_gradual_latent_params"):

logger.info("gradual_latent is not supported for this scheduler. Ignoring.")

logger.info(f'{self.scheduler.__class__.__name__}')

else:

enable_gradual_latent = True

step_elapsed = 1000

current_ratio = self.gradual_latent.ratio

# first, we downscale the latents to the specified ratio / 最初に指定された比率にlatentsをダウンスケールする

height, width = latents.shape[-2:]

org_dtype = latents.dtype

if org_dtype == torch.bfloat16:

latents = latents.float()

latents = torch.nn.functional.interpolate(

latents, scale_factor=current_ratio, mode="bicubic", align_corners=False

).to(org_dtype)

# apply unsharp mask / アンシャープマスクを適用する

if self.gradual_latent.gaussian_blur_ksize:

latents = self.gradual_latent.apply_unshark_mask(latents)

for i, t in enumerate(tqdm(timesteps)):

resized_size = None

if enable_gradual_latent:

# gradually upscale the latents / latentsを徐々にアップスケールする

if (

t < self.gradual_latent.start_timesteps

and current_ratio < 1.0