





import argparse
import math
import os
from copy import deepcopy

import torch
from audio_diffusion.models import DiffusionAttnUnetaD
from diffusion import sampling
from torch import nn

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel


_lowerCAmelCase							:					Optional[Any]		    =		{
    '''gwf-440k''': {
        '''url''': '''https://model-server.zqevans2.workers.dev/gwf-440k.ckpt''',
        '''sample_rate''': 48_000,
        '''sample_size''': 65_536,
    },
    '''jmann-small-190k''': {
        '''url''': '''https://model-server.zqevans2.workers.dev/jmann-small-190k.ckpt''',
        '''sample_rate''': 48_000,
        '''sample_size''': 65_536,
    },
    '''jmann-large-580k''': {
        '''url''': '''https://model-server.zqevans2.workers.dev/jmann-large-580k.ckpt''',
        '''sample_rate''': 48_000,
        '''sample_size''': 131_072,
    },
    '''maestro-uncond-150k''': {
        '''url''': '''https://model-server.zqevans2.workers.dev/maestro-uncond-150k.ckpt''',
        '''sample_rate''': 16_000,
        '''sample_size''': 65_536,
    },
    '''unlocked-uncond-250k''': {
        '''url''': '''https://model-server.zqevans2.workers.dev/unlocked-uncond-250k.ckpt''',
        '''sample_rate''': 16_000,
        '''sample_size''': 65_536,
    },
    '''honk-140k''': {
        '''url''': '''https://model-server.zqevans2.workers.dev/honk-140k.ckpt''',
        '''sample_rate''': 16_000,
        '''sample_size''': 65_536,
    },
}

def    __snake_case  (    _lowerCAmelCase       :       List[Any] , _lowerCAmelCase       :       Optional[Any]		)       ->  Tuple:
   return torch.atana(lowerCamelCase__ , lowerCamelCase__		) / math.pi * 2

def    __snake_case  (    _lowerCAmelCase       :       List[Any]		)       ->  Optional[int]:
   A_    : Dict								=	torch.sin(t * math.pi / 2		) ** 2
   A_    : Optional[Any]								=	(1 - sigma**2) ** 0.5
   return alpha_sigma_to_t(lowerCamelCase__ , lowerCamelCase__		)


class     __magic_name__     (							__UpperCAmelCase							):





     """simple docstring"""



     pass




class     __magic_name__     (							nn.Module							):





     """simple docstring"""



     def __init__(							self				:Optional[int]	,					snake_case				:Any     ):





        '''simple docstring'''




        super().__init__()

        A_    : List[Any]								=	DiffusionAttnUnetaD(snake_case	,					n_attn_layers=4     )
        A_    : int								=	deepcopy(self.diffusion     )
        A_    : str								=	torch.quasirandom.SobolEngine(1	,					scramble=snake_case     )




def    __snake_case  (    _lowerCAmelCase       :       int		)       ->  List[Any]:
   A_    : List[str]								=	MODELS_MAP[model_name]["""url"""]
   os.system(f"wget {url} ./"		)

   return f"./{model_name}.ckpt"


_lowerCAmelCase							:					int		    =		{
    '''1''': '''resnets.0''',
    '''2''': '''attentions.0''',
    '''3''': '''resnets.1''',
    '''4''': '''attentions.1''',
    '''5''': '''resnets.2''',
    '''6''': '''attentions.2''',
}
_lowerCAmelCase							:					List[Any]		    =		{
    '''8''': '''resnets.0''',
    '''9''': '''attentions.0''',
    '''10''': '''resnets.1''',
    '''11''': '''attentions.1''',
    '''12''': '''resnets.2''',
    '''13''': '''attentions.2''',
}
_lowerCAmelCase							:					Dict		    =		{
    '''1''': '''resnets.0''',
    '''2''': '''attentions.0''',
    '''3''': '''resnets.1''',
    '''4''': '''attentions.1''',
    '''5''': '''resnets.2''',
    '''6''': '''attentions.2''',
    '''8''': '''resnets.3''',
    '''9''': '''attentions.3''',
    '''10''': '''resnets.4''',
    '''11''': '''attentions.4''',
    '''12''': '''resnets.5''',
    '''13''': '''attentions.5''',
}
_lowerCAmelCase							:					Union[str, Any]		    =		{
    '''0''': '''resnets.0''',
    '''1''': '''resnets.1''',
    '''2''': '''resnets.2''',
    '''4''': '''resnets.0''',
    '''5''': '''resnets.1''',
    '''6''': '''resnets.2''',
}

_lowerCAmelCase							:					Dict		    =		{
    '''skip''': '''conv_skip''',
    '''main.0''': '''conv_1''',
    '''main.1''': '''group_norm_1''',
    '''main.3''': '''conv_2''',
    '''main.4''': '''group_norm_2''',
}

_lowerCAmelCase							:					str		    =		{
    '''norm''': '''group_norm''',
    '''qkv_proj''': ['''query''', '''key''', '''value'''],
    '''out_proj''': ['''proj_attn'''],
}

def    __snake_case  (    _lowerCAmelCase       :       Dict		)       ->  Any:





   if name.startswith("skip"		):
      return name.replace("skip" , RES_CONV_MAP["skip"]		)

   # name has to be of format main.{digit}
   if not name.startswith("main."		):
      raise ValueError(f"ResConvBlock error with {name}"		)

   return name.replace(name[:6] , RES_CONV_MAP[name[:6]]		)

def    __snake_case  (    _lowerCAmelCase       :       Union[str, Any]		)       ->  Any:





   for key, value in ATTN_MAP.items():
      if name.startswith(lowerCamelCase__		) and not isinstance(lowerCamelCase__ , lowerCamelCase__		):
         return name.replace(lowerCamelCase__ , lowerCamelCase__		)
      elif name.startswith(lowerCamelCase__		):
         return [name.replace(lowerCamelCase__ , lowerCamelCase__		) for v in value]
   raise ValueError(f"Attn error with {name}"		)

def    __snake_case  (    _lowerCAmelCase       :       Dict , _lowerCAmelCase       :       List[Any]=13		)       ->  Any:
   A_    : Tuple								=	input_string

   if string.split("."		)[0] == "timestep_embed":
      return string.replace("timestep_embed" , "time_proj"		)

   A_    : str								=	0
   if string.startswith("net.3."		):
      depth += 1
      A_    : List[Any]								=	string[6:]
   elif string.startswith("net."		):
      A_    : Union[str, Any]								=	string[4:]

   while string.startswith("main.7."		):
      depth += 1
      A_    : Optional[Any]								=	string[7:]

   if string.startswith("main."		):
      A_    : int								=	string[5:]

   # mid block
   if string[:2].isdigit():
      A_    : Optional[int]								=	string[:2]
      A_    : List[str]								=	string[2:]
   else:
      A_    : int								=	string[0]
      A_    : Optional[Any]								=	string[1:]

   if depth == max_depth:
      A_    : int								=	MID_NUM_TO_LAYER[layer_num]
      A_    : Tuple								=	"""mid_block"""
   elif depth > 0 and int(lowerCamelCase__		) < 7:
      A_    : Any								=	DOWN_NUM_TO_LAYER[layer_num]
      A_    : Union[str, Any]								=	f"down_blocks.{depth}"
   elif depth > 0 and int(lowerCamelCase__		) > 7:
      A_    : str								=	UP_NUM_TO_LAYER[layer_num]
      A_    : Optional[int]								=	f"up_blocks.{max_depth - depth - 1}"
   elif depth == 0:
      A_    : Optional[int]								=	DEPTH_0_TO_LAYER[layer_num]
      A_    : str								=	f"up_blocks.{max_depth - 1}" if int(lowerCamelCase__		) > 3 else """down_blocks.0"""

   if not string_left.startswith("."		):
      raise ValueError(f"Naming error with {input_string} and string_left: {string_left}."		)

   A_    : str								=	string_left[1:]

   if "resnets" in new_layer:
      A_    : Dict								=	convert_resconv_naming(lowerCamelCase__		)
   elif "attentions" in new_layer:
      A_    : List[Any]								=	convert_attn_naming(lowerCamelCase__		)
      A_    : str								=	new_string_left

   if not isinstance(lowerCamelCase__ , lowerCamelCase__		):
      A_    : Optional[int]								=	prefix + """.""" + new_layer + """.""" + string_left
   else:
      A_    : Union[str, Any]								=	[prefix + """.""" + new_layer + """.""" + s for s in string_left]
   return new_string

def    __snake_case  (    _lowerCAmelCase       :       Any		)       ->  List[str]:
   A_    : int								=	{}
   for k, v in state_dict.items():
      if k.endswith("kernel"		):
         # up- and downsample layers, don't have trainable weights
         continue

      A_    : Tuple								=	rename(lowerCamelCase__		)

      # check if we need to transform from Conv => Linear for attention
      if isinstance(lowerCamelCase__ , lowerCamelCase__		):
         A_    : Optional[int]								=	transform_conv_attns(lowerCamelCase__ , lowerCamelCase__ , lowerCamelCase__		)
      else:
         A_    : Dict								=	v

   return new_state_dict

def    __snake_case  (    _lowerCAmelCase       :       Optional[int] , _lowerCAmelCase       :       List[str] , _lowerCAmelCase       :       Dict		)       ->  Optional[Any]:





   if len(lowerCamelCase__		) == 1:
      if len(v.shape		) == 3:
         # weight
         A_    : List[str]								=	v[:, :, 0]
      else:
         # bias
         A_    : List[str]								=	v
   else:
      # qkv matrices
      A_    : Dict								=	v.shape[0]
      A_    : Optional[int]								=	trippled_shape // 3
      for i in range(3		):
         if len(v.shape		) == 3:
            A_    : List[Any]								=	v[i * single_shape : (i + 1) * single_shape, :, 0]
         else:
            A_    : int								=	v[i * single_shape : (i + 1) * single_shape]
   return new_state_dict

def    __snake_case  (    _lowerCAmelCase       :       str		)       ->  Optional[int]:
   A_    : Any								=	torch.device("cuda" if torch.cuda.is_available() else "cpu"		)

   A_    : Tuple								=	args.model_path.split("/"		)[-1].split("."		)[0]
   if not os.path.isfile(args.model_path		):
      assert (
          model_name == args.model_path
      ), f"Make sure to provide one of the official model names {MODELS_MAP.keys()}"
      A_    : Optional[int]								=	download(lowerCamelCase__		)

   A_    : Tuple								=	MODELS_MAP[model_name]["""sample_rate"""]
   A_    : Dict								=	MODELS_MAP[model_name]["""sample_size"""]

   A_    : Optional[int]								=	Object()
   A_    : List[str]								=	sample_size
   A_    : Any								=	sample_rate
   A_    : Optional[int]								=	0

   A_    : List[str]								=	UNetaDModel(sample_size=lowerCamelCase__ , sample_rate=lowerCamelCase__		)
   A_    : List[str]								=	diffusers_model.state_dict()

   A_    : List[Any]								=	DiffusionUncond(lowerCamelCase__		)
   orig_model.load_state_dict(torch.load(args.model_path , map_location=lowerCamelCase__		)["state_dict"]		)
   A_    : Dict								=	orig_model.diffusion_ema.eval()
   A_    : int								=	orig_model.state_dict()
   A_    : Union[str, Any]								=	rename_orig_weights(lowerCamelCase__		)

   A_    : str								=	set(renamed_state_dict.keys()		) - set(diffusers_state_dict.keys()		)
   A_    : Union[str, Any]								=	set(diffusers_state_dict.keys()		) - set(renamed_state_dict.keys()		)

   assert len(lowerCamelCase__		) == 0, f"Problem with {renamed_minus_diffusers}"
   assert all(k.endswith("kernel"		) for k in list(lowerCamelCase__		)		), f"Problem with {diffusers_minus_renamed}"

   for key, value in renamed_state_dict.items():
      assert (
          diffusers_state_dict[key].squeeze().shape == value.squeeze().shape
      ), f"Shape for {key} doesn\'t match. Diffusers: {diffusers_state_dict[key].shape} vs. {value.shape}"
      if key == "time_proj.weight":
         A_    : Tuple								=	value.squeeze()

      A_    : List[Any]								=	value

   diffusers_model.load_state_dict(lowerCamelCase__		)

   A_    : int								=	100
   A_    : List[str]								=	33

   A_    : str								=	IPNDMScheduler(num_train_timesteps=lowerCamelCase__		)

   A_    : List[str]								=	torch.manual_seed(lowerCamelCase__		)
   A_    : Union[str, Any]								=	torch.randn([1, 2, config.sample_size] , generator=lowerCamelCase__		).to(lowerCamelCase__		)

   A_    : Optional[Any]								=	torch.linspace(1 , 0 , steps + 1 , device=lowerCamelCase__		)[:-1]
   A_    : str								=	get_crash_schedule(lowerCamelCase__		)

   A_    : str								=	DanceDiffusionPipeline(unet=lowerCamelCase__ , scheduler=lowerCamelCase__		)

   A_    : Tuple								=	torch.manual_seed(33		)
   A_    : List[Any]								=	pipe(num_inference_steps=lowerCamelCase__ , generator=lowerCamelCase__		).audios

   A_    : int								=	sampling.iplms_sample(lowerCamelCase__ , lowerCamelCase__ , lowerCamelCase__ , {}		)
   A_    : Optional[Any]								=	generated.clamp(-1 , 1		)

   A_    : int								=	(generated - audio).abs().sum()
   A_    : Tuple								=	(generated - audio).abs().max()

   if args.save:
      pipe.save_pretrained(args.checkpoint_path		)

   print("Diff sum" , lowerCamelCase__		)
   print("Diff max" , lowerCamelCase__		)

   assert diff_max < 1e-3, f"Diff max: {diff_max} is too much :-/"

   print(f"Conversion for {model_name} successful!"		)


if __name__ == "__main__":
 _lowerCAmelCase							:					Optional[Any]		    =		argparse.ArgumentParser()

 parser.add_argument('''--model_path''', default=None, type=str, required=True, help='''Path to the model to convert.''')
 parser.add_argument(
     '''--save''', default=True, type=bool, required=False, help='''Whether to save the converted model or not.'''
 )
 parser.add_argument('''--checkpoint_path''', default=None, type=str, required=True, help='''Path to the output model.''')
 _lowerCAmelCase							:					Dict		    =		parser.parse_args()

 main(args)





'''simple docstring'''




import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


lowerCamelCase							:Any				    =					logging.get_logger(__name__)  # pylint: disable=invalid-name







class 	_lowerCAmelCase							(			__UpperCAmelCase ):


	def __init__(self       ,      lowercase       ,      lowercase       ,      lowercase       ,      lowercase       ,      lowercase       ,      lowercase       ,      lowercase       ,      ):
				super().__init__()
				self.register_modules(
				    vae=lowercase       ,      text_encoder=lowercase       ,      tokenizer=lowercase       ,      unet=lowercase       ,      scheduler=lowercase       ,      safety_checker=lowercase       ,      feature_extractor=lowercase       ,      )


	def     _a	(self       ,      lowercase = "auto"							):
				if slice_size == "auto":
							# half the attention head size is usually a good trade-off between
							# speed and memory
							A_	:						int	=     self.unet.config.attention_head_dim // 2
				self.unet.set_attention_slice(lowercase							)


	def     _a	(self							):
				self.enable_attention_slicing(lowercase							)


	@torch.no_grad()
	def __call__(self       ,      lowercase       ,      lowercase = 512       ,      lowercase = 512       ,      lowercase = 50       ,      lowercase = 7.5       ,      lowercase = None       ,      lowercase = 1       ,      lowercase = 0.0       ,      lowercase = None       ,      lowercase = None       ,      lowercase = "pil"       ,      lowercase = True       ,      lowercase = None       ,      lowercase = 1       ,      lowercase = None       ,      **lowercase       ,      ):

				if isinstance(lowercase       ,      lowercase							):
							A_	:						Union[str, Any]	=     1
				elif isinstance(lowercase       ,      lowercase							):
							A_	:						Any	=     len(lowercase							)
				else:
							raise ValueError(F'`prompt` has to be of type `str` or `list` but is {type(lowercase							)}'							)

				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(lowercase       ,      lowercase							) or callback_steps <= 0)
				):
							raise ValueError(
							    F'`callback_steps` has to be a positive integer but is {callback_steps} of type'
							    F' {type(lowercase							)}.'							)

				# get prompt text embeddings
				A_	:						Optional[Any]	=     self.tokenizer(
				    lowercase       ,      padding="""max_length"""       ,      max_length=self.tokenizer.model_max_length       ,      return_tensors="""pt"""       ,      )
				A_	:						Dict	=     text_inputs.input_ids

				if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
							A_	:						Union[str, Any]	=     self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]							)
							logger.warning(
							    """The following part of your input was truncated because CLIP can only handle sequences up to"""
							    F' {self.tokenizer.model_max_length} tokens: {removed_text}'							)
							A_	:						Dict	=     text_input_ids[:, : self.tokenizer.model_max_length]

				if text_embeddings is None:
							A_	:						Any	=     self.text_encoder(text_input_ids.to(self.device							)							)[0]

				# duplicate text embeddings for each generation per prompt, using mps friendly method
				A_,    A_,    A_	:						Tuple	=     text_embeddings.shape
				A_	:						Optional[Any]	=     text_embeddings.repeat(1       ,      lowercase       ,      1							)
				A_	:						Any	=     text_embeddings.view(bs_embed * num_images_per_prompt       ,      lowercase       ,      -1							)

				# 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.
				A_	:						List[str]	=     guidance_scale > 1.0
				# get unconditional embeddings for classifier free guidance
				if do_classifier_free_guidance:
							A_	:						List[str]
							if negative_prompt is None:
										A_	:						Optional[int]	=     [""""""]
							elif type(lowercase							) is not type(lowercase							):
										raise TypeError(
										    F'`negative_prompt` should be the same type to `prompt`, but got {type(lowercase							)} !='
										    F' {type(lowercase							)}.'							)
							elif isinstance(lowercase       ,      lowercase							):
										A_	:						Dict	=     [negative_prompt]
							elif batch_size != len(lowercase							):
										raise ValueError(
										    F'`negative_prompt`: {negative_prompt} has batch size {len(lowercase							)}, but `prompt`:'
										    F' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
										    """ the batch size of `prompt`."""							)
							else:
										A_	:						Dict	=     negative_prompt

							A_	:						int	=     text_input_ids.shape[-1]
							A_	:						List[Any]	=     self.tokenizer(
							    lowercase       ,      padding="""max_length"""       ,      max_length=lowercase       ,      truncation=lowercase       ,      return_tensors="""pt"""       ,      )
							A_	:						str	=     self.text_encoder(uncond_input.input_ids.to(self.device							)							)[0]

							# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
							A_	:						Optional[Any]	=     uncond_embeddings.shape[1]
							A_	:						str	=     uncond_embeddings.repeat(lowercase       ,      lowercase       ,      1							)
							A_	:						List[str]	=     uncond_embeddings.view(batch_size * num_images_per_prompt       ,      lowercase       ,      -1							)

							# For classifier free guidance, we need to do two forward passes.
							# Here we concatenate the unconditional and text embeddings into a single batch
							# to avoid doing two forward passes
							A_	:						Dict	=     torch.cat([uncond_embeddings, text_embeddings]							)

				# 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`.
				A_	:						int	=     (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
				A_	:						Dict	=     (batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
				A_	:						Dict	=     text_embeddings.dtype
				if latents is None:
							if self.device.type == "mps":
										# randn does not exist on mps
										A_	:						Tuple	=     torch.randn(
										    lowercase       ,      generator=lowercase       ,      device="""cpu"""       ,      dtype=lowercase							).to(self.device							)
										A_	:						int	=     torch.randn(lowercase       ,      generator=lowercase       ,      device="""cpu"""       ,      dtype=lowercase							).to(
										    self.device							)
							else:
										A_	:						int	=     torch.randn(
										    lowercase       ,      generator=lowercase       ,      device=self.device       ,      dtype=lowercase							)
										A_	:						str	=     torch.randn(lowercase       ,      generator=lowercase       ,      device=self.device       ,      dtype=lowercase							)
				else:
							if latents_reference.shape != latents_shape:
										raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'							)
							A_	:						str	=     latents_reference.to(self.device							)
							A_	:						Tuple	=     latents.to(self.device							)

				# This is the key part of the pipeline where we
				# try to ensure that the generated images w/ the same seed
				# but different sizes actually result in similar images
				A_	:						Optional[int]	=     (latents_shape[3] - latents_shape_reference[3]) // 2
				A_	:						Optional[int]	=     (latents_shape[2] - latents_shape_reference[2]) // 2
				A_	:						Optional[int]	=     latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
				A_	:						int	=     latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
				A_	:						Optional[Any]	=     0 if dx < 0 else dx
				A_	:						Optional[Any]	=     0 if dy < 0 else dy
				A_	:						Optional[int]	=     max(-dx       ,      0							)
				A_	:						List[str]	=     max(-dy       ,      0							)
				# import pdb
				# pdb.set_trace()
				A_	:						str	=     latents_reference[:, :, dy : dy + h, dx : dx + w]

				# set timesteps
				self.scheduler.set_timesteps(lowercase							)

				# Some schedulers like PNDM have timesteps as arrays
				# It's more optimized to move all timesteps to correct device beforehand
				A_	:						Any	=     self.scheduler.timesteps.to(self.device							)

				# scale the initial noise by the standard deviation required by the scheduler
				A_	:						Tuple	=     latents * self.scheduler.init_noise_sigma

				# 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]
				A_	:						Dict	=     """eta""" in set(inspect.signature(self.scheduler.step							).parameters.keys()							)
				A_	:						Any	=     {}
				if accepts_eta:
							A_	:						Optional[int]	=     eta

				for i, t in enumerate(self.progress_bar(lowercase							)							):
							# expand the latents if we are doing classifier free guidance
							A_	:						Tuple	=     torch.cat([latents] * 2							) if do_classifier_free_guidance else latents
							A_	:						Tuple	=     self.scheduler.scale_model_input(lowercase       ,      lowercase							)

							# predict the noise residual
							A_	:						List[str]	=     self.unet(lowercase       ,      lowercase       ,      encoder_hidden_states=lowercase							).sample

							# perform guidance
							if do_classifier_free_guidance:
										A_,    A_	:						str	=     noise_pred.chunk(2							)
										A_	:						Optional[Any]	=     noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

							# compute the previous noisy sample x_t -> x_t-1
							A_	:						List[str]	=     self.scheduler.step(lowercase       ,      lowercase       ,      lowercase       ,      **lowercase							).prev_sample

							# call the callback, if provided
							if callback is not None and i % callback_steps == 0:
										callback(lowercase       ,      lowercase       ,      lowercase							)

				A_	:						List[str]	=     1 / 0.1_82_15 * latents
				A_	:						List[str]	=     self.vae.decode(lowercase							).sample

				A_	:						Optional[int]	=     (image / 2 + 0.5).clamp(0       ,      1							)

				# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
				A_	:						Optional[int]	=     image.cpu().permute(0       ,      2       ,      3       ,      1							).float().numpy()

				if self.safety_checker is not None:
							A_	:						Union[str, Any]	=     self.feature_extractor(self.numpy_to_pil(lowercase							)       ,      return_tensors="""pt"""							).to(
							    self.device							)
							A_,    A_	:						Optional[int]	=     self.safety_checker(
							    images=lowercase       ,      clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype							)							)
				else:
							A_	:						Tuple	=     None

				if output_type == "pil":
							A_	:						Tuple	=     self.numpy_to_pil(lowercase							)

				if not return_dict:
							return (image, has_nsfw_concept)

				return StableDiffusionPipelineOutput(images=lowercase       ,      nsfw_content_detected=lowercase							)