code: 
  '''simple docstring'''
  # Copyright (c) 2021-, NVIDIA CORPORATION.  All rights reserved.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  ####################################################################################################
  
  #
  # Note: If when running this conversion script you're getting an exception:
  #     ModuleNotFoundError: No module named 'megatron.model.enums'
  # you need to tell python where to find the clone of Megatron-LM, e.g.:
  #
  # cd /tmp
  # git clone https://github.com/NVIDIA/Megatron-LM
  # PYTHONPATH=/tmp/Megatron-LM python src/transformers/models/megatron_bert/convert_megatron_bert_checkpoint.py ...
  #
  # if you already have it cloned elsewhere, simply adjust the path to the existing path
  #
  # If the training was done using a Megatron-LM fork, e.g.,
  # https://github.com/microsoft/Megatron-DeepSpeed/ then chances are that you need to have that one
  # in your path, i.e., /path/to/Megatron-DeepSpeed/
  #
  
  import argparse
  import os
  import re
  import zipfile
  
  import torch
  
  from transformers import MegatronBertConfig
  
  
  
  
  def       _UpperCamelCase	(					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__=0 )     ->       Dict:
  
  
  		'''simple docstring'''
  
  
  		if name is None:
  				snake_case     :					Any     =	None
  		else:
  				snake_case     :					List[str]     =	'''.''' * max(0   ,					spaces - 2 ) + '''# {:''' + str(50 - spaces ) + '''s}'''
  				snake_case     :					Dict     =	fmt.format(SCREAMING_SNAKE_CASE__ )
  
  		# Print and recurse (if needed).
  		if isinstance(SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ ):
  				if msg is not None:
  						print(SCREAMING_SNAKE_CASE__ )
  				for k in val.keys():
  						recursive_print(SCREAMING_SNAKE_CASE__   ,					val[k]   ,					spaces + 2 )
  		elif isinstance(SCREAMING_SNAKE_CASE__   ,					torch.Tensor ):
  				print(SCREAMING_SNAKE_CASE__   ,					''':'''   ,					val.size() )
  		else:
  				print(SCREAMING_SNAKE_CASE__   ,					''':'''   ,					SCREAMING_SNAKE_CASE__ )
  
  
  
  
  def       _UpperCamelCase	(					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ )     ->       Dict:
  
  
  		'''simple docstring'''
  
  
  		snake_case     :					List[Any]     =	param.size()
  		if checkpoint_version == 1.0:
  				# version 1.0 stores [num_heads * hidden_size * num_splits, :]
  				snake_case     :					int     =	(num_heads, hidden_size, num_splits) + input_shape[1:]
  				snake_case     :					Union[str, Any]     =	param.view(*SCREAMING_SNAKE_CASE__ )
  				snake_case     :					List[Any]     =	param.transpose(0   ,					2 )
  				snake_case     :					Optional[int]     =	param.transpose(1   ,					2 ).contiguous()
  		elif checkpoint_version >= 2.0:
  				# other versions store [num_heads * num_splits * hidden_size, :]
  				snake_case     :					Any     =	(num_heads, num_splits, hidden_size) + input_shape[1:]
  				snake_case     :					Tuple     =	param.view(*SCREAMING_SNAKE_CASE__ )
  				snake_case     :					Any     =	param.transpose(0   ,					1 ).contiguous()
  		snake_case     :					Any     =	param.view(*SCREAMING_SNAKE_CASE__ )
  		return param
  
  
  
  
  def       _UpperCamelCase	(					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ )     ->       Tuple:
  
  
  		'''simple docstring'''
  
  
  		snake_case     :					Optional[Any]     =	{}
  
  		# old versions did not store training args
  		snake_case     :					List[Any]     =	input_state_dict.get('''args'''   ,					SCREAMING_SNAKE_CASE__ )
  		if ds_args is not None:
  				# do not make the user write a config file when the exact dimensions/sizes are already in the checkpoint
  				# from pprint import pprint
  				# pprint(vars(ds_args))
  
  				snake_case     :					Any     =	ds_args.tokenizer_type
  				snake_case     :					Optional[int]     =	ds_args.padded_vocab_size
  				snake_case     :					Tuple     =	ds_args.max_position_embeddings
  				snake_case     :					str     =	ds_args.hidden_size
  				snake_case     :					Any     =	ds_args.num_layers
  				snake_case     :					List[Any]     =	ds_args.num_attention_heads
  				snake_case     :					str     =	ds_args.ffn_hidden_size if '''ffn_hidden_size''' in ds_args else 4 * ds_args.hidden_size
  				# pprint(config)
  
  		# The number of heads.
  		snake_case     :					Optional[int]     =	config.num_attention_heads
  		# The hidden_size per head.
  		snake_case     :					str     =	config.hidden_size // heads
  		# Megatron-LM checkpoint version
  		if "checkpoint_version" in input_state_dict.keys():
  				snake_case     :					Optional[Any]     =	input_state_dict['''checkpoint_version''']
  		else:
  				snake_case     :					str     =	0.0
  
  		# The model.
  		snake_case     :					Dict     =	input_state_dict['''model''']
  		# The language model.
  		snake_case     :					Optional[Any]     =	model['''language_model''']
  		# The embeddings.
  		snake_case     :					Tuple     =	lm['''embedding''']
  
  		# The word embeddings.
  		snake_case     :					Optional[Any]     =	embeddings['''word_embeddings''']['''weight''']
  		# Truncate the embedding table to vocab_size rows.
  		snake_case     :					int     =	word_embeddings[: config.vocab_size, :]
  		# Store the word embeddings.
  		snake_case     :					Optional[Any]     =	word_embeddings
  
  		# The position embeddings.
  		snake_case     :					Union[str, Any]     =	embeddings['''position_embeddings''']['''weight''']
  		assert pos_embeddings.size(0 ) == config.max_position_embeddings and pos_embeddings.size(1 ) == config.hidden_size
  		# Store the position embeddings.
  		snake_case     :					List[str]     =	pos_embeddings
  
  		# The token-type embeddings.
  		snake_case     :					Optional[int]     =	embeddings['''tokentype_embeddings''']['''weight''']
  		# Store the position embeddings.
  		snake_case     :					Union[str, Any]     =	tokentype_embeddings
  
  		# The transformer.
  		snake_case     :					int     =	lm['''transformer'''] if '''transformer''' in lm.keys() else lm['''encoder''']
  
  		# The regex to extract layer names.
  		snake_case     :					Optional[int]     =	re.compile(R'''layers\.(\d+)\.([a-z0-9_.]+)\.([a-z]+)''' )
  
  		# The simple map of names for "automated" rules.
  		snake_case     :					Tuple     =	{
  		    '''attention.dense''': '''.attention.output.dense.''',
  		    '''self_attention.dense''': '''.attention.output.dense.''',
  		    '''mlp.dense_h_to_4h''': '''.intermediate.dense.''',
  		    '''mlp.dense_4h_to_h''': '''.output.dense.''',
  		}
  
  		# Keep track of the attention/query/value tensor.
  		snake_case     :					Optional[Any]     =	None
  
  		# Extract the layers.
  		for key, val in transformer.items():
  				# Match the name.
  				snake_case     :					str     =	layer_re.match(SCREAMING_SNAKE_CASE__ )
  
  				# Stop if that's not a layer
  				if m is None:
  						break
  
  				# The index of the layer.
  				snake_case     :					int     =	int(m.group(1 ) )
  				# The name of the operation.
  				snake_case     :					Dict     =	m.group(2 )
  				# Is it a weight or a bias?
  				snake_case     :					Any     =	m.group(3 )
  
  				# The name of the layer.
  				snake_case     :					Dict     =	F'bert.encoder.layer.{layer_idx}'
  
  				# For layernorm(s), simply store the layer norm.
  				if op_name.endswith('''layernorm''' ):
  						snake_case     :					Optional[int]     =	'''attention.ln''' if op_name.startswith('''input''' ) else '''ln'''
  						snake_case     :					int     =	val
  
  				# Transpose the QKV matrix.
  				elif (
  				    op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
  				) and weight_or_bias == "weight":
  						# Make sure the QKV pointer is nil.
  						assert attention_qkv_weight is None, ""
  
  						snake_case     :					Union[str, Any]     =	fix_query_key_value_ordering(SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					3   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ )
  						# Store the tensor as we need the bias as well to interleave QKV and biases.
  						snake_case     :					Tuple     =	out_val
  
  				# Transpose the bias.
  				elif (
  				    op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
  				) and weight_or_bias == "bias":
  						# Make sure we read the weight tensor.
  						assert attention_qkv_weight is not None, ""
  
  						# Split the QKV matrix into Q, K and V. Megatron stores Q,K,V interleaved.
  						snake_case     :					Tuple     =	attention_qkv_weight[0 * config.hidden_size : 1 * config.hidden_size, :]
  						snake_case     :					str     =	attention_qkv_weight[1 * config.hidden_size : 2 * config.hidden_size, :]
  						snake_case     :					Union[str, Any]     =	attention_qkv_weight[2 * config.hidden_size : 3 * config.hidden_size, :]
  
  						snake_case     :					Union[str, Any]     =	fix_query_key_value_ordering(SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					3   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ )
  						# Split the bias.
  						snake_case     :					Optional[int]     =	out_val[0 * config.hidden_size : 1 * config.hidden_size]
  						snake_case     :					List[Any]     =	out_val[1 * config.hidden_size : 2 * config.hidden_size]
  						snake_case     :					str     =	out_val[2 * config.hidden_size : 3 * config.hidden_size]
  
  						# Store.
  						snake_case     :					Union[str, Any]     =	q
  						snake_case     :					int     =	q_bias
  						snake_case     :					Union[str, Any]     =	k
  						snake_case     :					Optional[Any]     =	k_bias
  						snake_case     :					Tuple     =	v
  						snake_case     :					Tuple     =	v_bias
  
  						# Clear the stored tensor.
  						snake_case     :					Optional[Any]     =	None
  
  				# Copy weights and biases as is.
  				elif weight_or_bias in ["weight", "bias"]:
  						snake_case     :					Union[str, Any]     =	megatron_to_transformers[op_name]
  						snake_case     :					Optional[Any]     =	val
  
      # The final layernorm.
  		snake_case     :					Optional[int]     =	transformer['''final_layernorm.weight''']
  		snake_case     :					int     =	transformer['''final_layernorm.bias''']
  
  		# The pooler.
  		snake_case     :					List[str]     =	lm['''pooler''']
  
  		# Store the matrix and the bias.
  		snake_case     :					int     =	pooler['''dense.weight''']
  		snake_case     :					Tuple     =	pooler['''dense.bias''']
  
  		# The LM head from Megatron (for RACE).
  		snake_case     :					Optional[Any]     =	model['''lm_head''']
  
  		# The transform matrix.
  		snake_case     :					Optional[int]     =	lm_head['''dense.weight''']
  		snake_case     :					Dict     =	lm_head['''dense.bias''']
  
  		# The transform LN.
  		snake_case     :					Optional[int]     =	lm_head['''layernorm.weight''']
  		snake_case     :					int     =	lm_head['''layernorm.bias''']
  
  		# For the decoder, we replicate the weights.
  		snake_case     :					Optional[int]     =	word_embeddings
  		snake_case     :					List[str]     =	lm_head['''bias''']
  
  		# The classifier from Megatron (for MLNI).
  		snake_case     :					int     =	model['''binary_head''']
  
  		# Store the classifier.
  		snake_case     :					List[Any]     =	binary_head['''weight''']
  		snake_case     :					Tuple     =	binary_head['''bias''']
  
  		# It should be done!
  		return output_state_dict
  
  
  
  
  def       _UpperCamelCase	(					)     ->       int:
  
  
  		'''simple docstring'''
  
  
  		snake_case     :					Optional[Any]     =	argparse.ArgumentParser()
  		parser.add_argument('''--print-checkpoint-structure'''   ,					action='''store_true''' )
  		parser.add_argument('''path_to_checkpoint'''   ,					type=SCREAMING_SNAKE_CASE__   ,					help='''Path to the ZIP file containing the checkpoint''' )
  		parser.add_argument(
  		    '''--config_file'''   ,					default=''''''   ,					type=SCREAMING_SNAKE_CASE__   ,					help='''An optional config json file describing the pre-trained model.'''   ,					)
  		snake_case     :					Tuple     =	parser.parse_args()
  
  		# Extract the basename.
  		snake_case     :					Dict     =	os.path.dirname(args.path_to_checkpoint )
  
  		# Load the model.
  		# the .zip is very optional, let's keep it for backward compatibility
  		print(F'Extracting PyTorch state dictionary from "{args.path_to_checkpoint}"' )
  		if args.path_to_checkpoint.endswith('''.zip''' ):
  				with zipfile.ZipFile(args.path_to_checkpoint   ,					'''r''' ) as checkpoint:
  						with checkpoint.open('''release/mp_rank_00/model_optim_rng.pt''' ) as pytorch_dict:
  								snake_case     :					List[str]     =	torch.load(SCREAMING_SNAKE_CASE__   ,					map_location='''cpu''' )
  		else:
  				snake_case     :					str     =	torch.load(args.path_to_checkpoint   ,					map_location='''cpu''' )
  
  		if args.config_file == "":
  				# Default config of megatron-bert 345m
  				snake_case     :					List[str]     =	MegatronBertConfig()
  
  				# different megatron-bert-*-345m models have different vocab sizes, so override the default
  				# config (which is for megatron-bert-cased-345m) with the actual vocab dimension
  				snake_case     :					Optional[int]     =	input_state_dict['''model''']['''lm_head''']['''bias'''].numel()
  		else:
  				snake_case     :					List[str]     =	MegatronBertConfig.from_json_file(args.config_file )
  
  		# Convert.
  		print('''Converting''' )
  		snake_case     :					List[Any]     =	convert_megatron_checkpoint(SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ )
  
  		# Print the structure of converted state dict.
  		if args.print_checkpoint_structure:
  				recursive_print(SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ )
  
  		# Store the config to file.
  		print('''Saving config''' )
  		config.save_pretrained(SCREAMING_SNAKE_CASE__ )
  
  		# Store the state_dict to file.
  		snake_case     :					Optional[int]     =	os.path.join(SCREAMING_SNAKE_CASE__   ,					'''pytorch_model.bin''' )
  		print(F'Saving checkpoint to "{output_checkpoint_file}"' )
  		torch.save(SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ )
  
  
  ####################################################################################################
  
  if __name__ == "__main__":
  						main()
  
  ####################################################################################################
  
  
  
style_context: 
  '''simple docstring'''
  from typing import Callable, List, Optional, Union
  
  import torch
  
  from ...models import UNetaDModel
  from ...schedulers import CMStochasticIterativeScheduler
  from ...utils import (
      is_accelerate_available,
      is_accelerate_version,
      logging,
      randn_tensor,
      replace_example_docstring,
  )
  from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput
  
  
  lowercase__						=			logging.get_logger(__name__)  # pylint: disable=invalid-name
  
  
  lowercase__						=			"\n    Examples:\n        ```py\n        >>> import torch\n\n        >>> from diffusers import ConsistencyModelPipeline\n\n        >>> device = \"cuda\"\n        >>> # Load the cd_imagenet64_l2 checkpoint.\n        >>> model_id_or_path = \"openai/diffusers-cd_imagenet64_l2\"\n        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)\n        >>> pipe.to(device)\n\n        >>> # Onestep Sampling\n        >>> image = pipe(num_inference_steps=1).images[0]\n        >>> image.save(\"cd_imagenet64_l2_onestep_sample.png\")\n\n        >>> # Onestep sampling, class-conditional image generation\n        >>> # ImageNet-64 class label 145 corresponds to king penguins\n        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]\n        >>> image.save(\"cd_imagenet64_l2_onestep_sample_penguin.png\")\n\n        >>> # Multistep sampling, class-conditional image generation\n        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:\n        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77\n        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]\n        >>> image.save(\"cd_imagenet64_l2_multistep_sample_penguin.png\")\n        ```\n"
  
  
  
  
  
  
  class   snake_case__					(			__SCREAMING_SNAKE_CASE				):
  
  
  
  
  
  
  
  		"""simple docstring"""
  
  		def __init__(			self							: List[Any]     ,  UpperCamelCase__							: UNetaDModel     ,  UpperCamelCase__							: CMStochasticIterativeScheduler			)	-> None:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				super().__init__()
  
  				self.register_modules(
  				    unet=UpperCamelCase__     ,  scheduler=UpperCamelCase__     ,  )
  
  				snake_case     :					str     =	None
  
  
  
  		def 			lowerCAmelCase							(			self							: Tuple     ,  UpperCamelCase__							: str=0			)	-> int:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				if is_accelerate_available() and is_accelerate_version('''>='''     ,  '''0.14.0'''			):
  						from accelerate import cpu_offload
  				else:
  						raise ImportError('''`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher'''			)
  
  				snake_case     :					Optional[Any]     =	torch.device(f'cuda:{gpu_id}'			)
  
  				if self.device.type != "cpu":
  						self.to('''cpu'''     ,  silence_dtype_warnings=UpperCamelCase__			)
  						torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)
  
  				for cpu_offloaded_model in [self.unet]:
  						cpu_offload(UpperCamelCase__     ,  UpperCamelCase__			)
  
  				if self.safety_checker is not None:
  						cpu_offload(self.safety_checker     ,  execution_device=UpperCamelCase__     ,  offload_buffers=UpperCamelCase__			)
  
  
  
  		def 			lowerCAmelCase							(			self							: Optional[int]     ,  UpperCamelCase__							: int=0			)	-> Dict:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				if is_accelerate_available() and is_accelerate_version('''>='''     ,  '''0.17.0.dev0'''			):
  						from accelerate import cpu_offload_with_hook
  				else:
  						raise ImportError('''`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'''			)
  
  				snake_case     :					Union[str, Any]     =	torch.device(f'cuda:{gpu_id}'			)
  
  				if self.device.type != "cpu":
  						self.to('''cpu'''     ,  silence_dtype_warnings=UpperCamelCase__			)
  						torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)
  
  				snake_case     :					List[Any]     =	None
  				for cpu_offloaded_model in [self.unet]:
  						snake_case ,snake_case     :					Union[str, Any]     =	cpu_offload_with_hook(UpperCamelCase__     ,  UpperCamelCase__     ,  prev_module_hook=UpperCamelCase__			)
  
  				if self.safety_checker is not None:
  						snake_case ,snake_case     :					Optional[Any]     =	cpu_offload_with_hook(self.safety_checker     ,  UpperCamelCase__     ,  prev_module_hook=UpperCamelCase__			)
  
  				# We'll offload the last model manually.
  				snake_case     :					Optional[Any]     =	hook
  
  
  
  		@property
  		# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  		def 			lowerCAmelCase							(			self							: int			)	-> List[str]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				if not hasattr(self.unet     ,  '''_hf_hook'''			):
  						return self.device
  				for module in self.unet.modules():
  						if (
  						    hasattr(UpperCamelCase__     ,  '''_hf_hook'''			)
  						    and hasattr(module._hf_hook     ,  '''execution_device'''			)
  						    and module._hf_hook.execution_device is not None
  						):
  								return torch.device(module._hf_hook.execution_device			)
  				return self.device
  
  
  
  		def 			lowerCAmelCase							(			self							: int     ,  UpperCamelCase__							: Optional[Any]     ,  UpperCamelCase__							: Dict     ,  UpperCamelCase__							: int     ,  UpperCamelCase__							: List[str]     ,  UpperCamelCase__							: int     ,  UpperCamelCase__							: Optional[int]     ,  UpperCamelCase__							: Any     ,  UpperCamelCase__							: Optional[int]=None			)	-> Tuple:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				snake_case     :					Union[str, Any]     =	(batch_size, num_channels, height, width)
  				if isinstance(UpperCamelCase__     ,  UpperCamelCase__			) and len(UpperCamelCase__			) != batch_size:
  						raise ValueError(
  						    f'You have passed a list of generators of length {len(UpperCamelCase__			)}, but requested an effective batch'
  						    f' size of {batch_size}. Make sure the batch size matches the length of the generators.'			)
  
  				if latents is None:
  						snake_case     :					List[Any]     =	randn_tensor(UpperCamelCase__     ,  generator=UpperCamelCase__     ,  device=UpperCamelCase__     ,  dtype=UpperCamelCase__			)
  				else:
  						snake_case     :					Tuple     =	latents.to(device=UpperCamelCase__     ,  dtype=UpperCamelCase__			)
  
  				# scale the initial noise by the standard deviation required by the scheduler
  				snake_case     :					str     =	latents * self.scheduler.init_noise_sigma
  				return latents
  
  
  
  		def 			lowerCAmelCase							(			self							: Any     ,  UpperCamelCase__							: torch.FloatTensor     ,  UpperCamelCase__							: str = "pil"			)	-> Union[str, Any]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				if output_type not in ["pt", "np", "pil"]:
  						raise ValueError(
  						    f'output_type={output_type} is not supported. Make sure to choose one of [\'pt\', \'np\', or \'pil\']'			)
  
  				# Equivalent to diffusers.VaeImageProcessor.denormalize
  				snake_case     :					Any     =	(sample / 2 + 0.5).clamp(0     ,  1			)
  				if output_type == "pt":
  						return sample
  
  				# Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
  				snake_case     :					Optional[int]     =	sample.cpu().permute(0     ,  2     ,  3     ,  1			).numpy()
  				if output_type == "np":
  						return sample
  
  				# Output_type must be 'pil'
  				snake_case     :					Tuple     =	self.numpy_to_pil(UpperCamelCase__			)
  				return sample
  
  
  
  		def 			lowerCAmelCase							(			self							: Tuple     ,  UpperCamelCase__							: List[str]     ,  UpperCamelCase__							: Dict     ,  UpperCamelCase__							: List[str]=None			)	-> List[str]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				if self.unet.config.num_class_embeds is not None:
  						if isinstance(UpperCamelCase__     ,  UpperCamelCase__			):
  								snake_case     :					str     =	torch.tensor(UpperCamelCase__     ,  dtype=torch.int			)
  						elif isinstance(UpperCamelCase__     ,  UpperCamelCase__			):
  								assert batch_size == 1, "Batch size must be 1 if classes is an int"
  								snake_case     :					Dict     =	torch.tensor([class_labels]     ,  dtype=torch.int			)
  						elif class_labels is None:
  								# Randomly generate batch_size class labels
  								# TODO: should use generator here? int analogue of randn_tensor is not exposed in ...utils
  								snake_case     :					List[Any]     =	torch.randint(0     ,  self.unet.config.num_class_embeds     ,  size=(batch_size,)			)
  						snake_case     :					Dict     =	class_labels.to(UpperCamelCase__			)
  				else:
  						snake_case     :					int     =	None
  				return class_labels
  
  
  
  		def 			lowerCAmelCase							(			self							: List[Any]     ,  UpperCamelCase__							: int     ,  UpperCamelCase__							: Any     ,  UpperCamelCase__							: List[Any]     ,  UpperCamelCase__							: List[Any]     ,  UpperCamelCase__							: List[str]     ,  UpperCamelCase__							: Union[str, Any]			)	-> Union[str, Any]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				if num_inference_steps is None and timesteps is None:
  						raise ValueError('''Exactly one of `num_inference_steps` or `timesteps` must be supplied.'''			)
  
  				if num_inference_steps is not None and timesteps is not None:
  						logger.warning(
  						    f'Both `num_inference_steps`: {num_inference_steps} and `timesteps`: {timesteps} are supplied;'
  						    ''' `timesteps` will be used over `num_inference_steps`.'''			)
  
  				if latents is not None:
  						snake_case     :					List[Any]     =	(batch_size, 3, img_size, img_size)
  						if latents.shape != expected_shape:
  								raise ValueError(f'The shape of latents is {latents.shape} but is expected to be {expected_shape}.'			)
  
  				if (callback_steps is None) or (
  				    callback_steps is not None and (not isinstance(UpperCamelCase__     ,  UpperCamelCase__			) or callback_steps <= 0)
  				):
  						raise ValueError(
  						    f'`callback_steps` has to be a positive integer but is {callback_steps} of type'
  						    f' {type(UpperCamelCase__			)}.'			)
  
  
  
  		@torch.no_grad()
  		@replace_example_docstring(UpperCamelCase__			)
  		def __call__(			self							: Optional[int]     ,  UpperCamelCase__							: int = 1     ,  UpperCamelCase__							: Optional[Union[torch.Tensor, List[int], int]] = None     ,  UpperCamelCase__							: int = 1     ,  UpperCamelCase__							: List[int] = None     ,  UpperCamelCase__							: Optional[Union[torch.Generator, List[torch.Generator]]] = None     ,  UpperCamelCase__							: Optional[torch.FloatTensor] = None     ,  UpperCamelCase__							: Optional[str] = "pil"     ,  UpperCamelCase__							: bool = True     ,  UpperCamelCase__							: Optional[Callable[[int, int, torch.FloatTensor], None]] = None     ,  UpperCamelCase__							: int = 1     ,  )	-> str:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				snake_case     :					Optional[int]     =	self.unet.config.sample_size
  				snake_case     :					List[Any]     =	self._execution_device
  
  				# 1. Check inputs
  				self.check_inputs(UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__			)
  
  				# 2. Prepare image latents
  				# Sample image latents x_0 ~ N(0, sigma_0^2 * I)
  				snake_case     :					List[str]     =	self.prepare_latents(
  				    batch_size=UpperCamelCase__     ,  num_channels=self.unet.config.in_channels     ,  height=UpperCamelCase__     ,  width=UpperCamelCase__     ,  dtype=self.unet.dtype     ,  device=UpperCamelCase__     ,  generator=UpperCamelCase__     ,  latents=UpperCamelCase__     ,  )
  
  				# 3. Handle class_labels for class-conditional models
  				snake_case     :					List[str]     =	self.prepare_class_labels(UpperCamelCase__     ,  UpperCamelCase__     ,  class_labels=UpperCamelCase__			)
  
  				# 4. Prepare timesteps
  				if timesteps is not None:
  						self.scheduler.set_timesteps(timesteps=UpperCamelCase__     ,  device=UpperCamelCase__			)
  						snake_case     :					Union[str, Any]     =	self.scheduler.timesteps
  						snake_case     :					Optional[int]     =	len(UpperCamelCase__			)
  				else:
  						self.scheduler.set_timesteps(UpperCamelCase__			)
  						snake_case     :					List[str]     =	self.scheduler.timesteps
  
  				# 5. Denoising loop
  				# Multistep sampling: implements Algorithm 1 in the paper
  				with self.progress_bar(total=UpperCamelCase__			) as progress_bar:
  						for i, t in enumerate(UpperCamelCase__			):
  								snake_case     :					Any     =	self.scheduler.scale_model_input(UpperCamelCase__     ,  UpperCamelCase__			)
  								snake_case     :					int     =	self.unet(UpperCamelCase__     ,  UpperCamelCase__     ,  class_labels=UpperCamelCase__     ,  return_dict=UpperCamelCase__			)[0]
  
  								snake_case     :					Dict     =	self.scheduler.step(UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__     ,  generator=UpperCamelCase__			)[0]
  
  								# call the callback, if provided
  								progress_bar.update()
  								if callback is not None and i % callback_steps == 0:
  										callback(UpperCamelCase__     ,  UpperCamelCase__     ,  UpperCamelCase__			)
  
          # 6. Post-process image sample
  				snake_case     :					Any     =	self.postprocess_image(UpperCamelCase__     ,  output_type=UpperCamelCase__			)
  
  				# Offload last model to CPU
  				if hasattr(self     ,  '''final_offload_hook'''			) and self.final_offload_hook is not None:
  						self.final_offload_hook.offload()
  
  				if not return_dict:
  						return (image,)
  
  				return ImagePipelineOutput(images=UpperCamelCase__			)
  
  
  