code: 
  import argparse
  import json
  import os
  import re
  from collections import OrderedDict
  from dataclasses import dataclass, field
  from functools import partial
  from pathlib import Path
  from pprint import pprint
  from typing import Dict, List, Tuple
  
  import torch
  import torch.nn as nn
  from classy_vision.models.regnet import RegNet, RegNetParams
  from huggingface_hub import cached_download, hf_hub_url
  from torch import Tensor
  from vissl.models.model_helpers import get_trunk_forward_outputs
  
  from transformers import AutoImageProcessor, RegNetConfig, RegNetForImageClassification, RegNetModel
  from transformers.modeling_utils import PreTrainedModel
  from transformers.utils import logging
  
  
  logging.set_verbosity_info()
  lowercase__						=			logging.get_logger()
  
  
  
  
  
  
  @dataclass
  class   snake_case__					:
  
  
  
  
  
  
  
  		"""simple docstring"""
  
  		lowerCamelCase           =	42
  		lowerCamelCase           =	field(default_factory=lowerCAmelCase_				)
  		lowerCamelCase           =	field(default_factory=lowerCAmelCase_				)
  		lowerCamelCase           =	field(default_factory=lowerCAmelCase_				)
  
  
  
  		def 			lowerCAmelCase							(			self							: Optional[Any]     ,  UpperCamelCase__							: int     ,  UpperCamelCase__							: Union[str, Any]     ,  UpperCamelCase__							: Optional[Any]     ,  UpperCamelCase__							: Optional[Any]			)	-> Optional[int]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				snake_case     :					Optional[Any]     =	len(list(m.modules()			)			) == 1 or isinstance(__SCREAMING_SNAKE_CASE     ,  nn.Convad			) or isinstance(__SCREAMING_SNAKE_CASE     ,  nn.BatchNormad			)
  				if has_not_submodules:
  						self.traced.append(__SCREAMING_SNAKE_CASE			)
  						snake_case     :					List[str]     =	m
  
  
  
  		def __call__(			self							: Optional[int]     ,  UpperCamelCase__							: Union[str, Any]			)	-> Any:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				for name, m in self.module.named_modules():
  						self.handles.append(m.register_forward_hook(partial(self._forward_hook     ,  name=__SCREAMING_SNAKE_CASE			)			)			)
  				self.module(__SCREAMING_SNAKE_CASE			)
  				[x.remove() for x in self.handles]
  				return self
  
  
  
  		@property
  		def 			lowerCAmelCase							(			self							: Optional[int]			)	-> Any:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				return {k: v for k, v in self.nameamodule.items() if len(list(v.state_dict().keys()			)			) > 0}
  
  
  
  
  
  
  
  class   snake_case__					(			nn.Module				):
  
  
  
  
  
  
  
  		"""simple docstring"""
  
  		def __init__(			self							: Dict     ,  UpperCamelCase__							: Optional[int]			)	-> Optional[int]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				super().__init__()
  
  				snake_case     :					str     =	[]
  				# - get the stem
  				feature_blocks.append(('''conv1''', model.stem)			)
  				# - get all the feature blocks
  				for k, v in model.trunk_output.named_children():
  						assert k.startswith('''block'''			), f'Unexpected layer name {k}'
  						snake_case     :					List[Any]     =	len(__SCREAMING_SNAKE_CASE			) + 1
  						feature_blocks.append((f'res{block_index}', v)			)
  
  				snake_case     :					Optional[int]     =	nn.ModuleDict(__SCREAMING_SNAKE_CASE			)
  
  
  
  		def 			lowerCAmelCase							(			self							: Any     ,  UpperCamelCase__							: Tuple			)	-> Optional[int]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				return get_trunk_forward_outputs(
  				    __SCREAMING_SNAKE_CASE     ,  out_feat_keys=__SCREAMING_SNAKE_CASE     ,  feature_blocks=self._feature_blocks     ,  )
  
  
  
  
  
  
  
  class   snake_case__					(			lowerCAmelCase_				):
  
  
  
  
  
  
  
  		"""simple docstring"""
  
  		def 			lowerCAmelCase							(			self							: Tuple			)	-> Optional[int]:
  
  
  
  
  
  
  				"""simple docstring"""
  
  
  
  
  
  
  
  				return [(8, 2, 2, 8, 1.0), (8, 2, 7, 8, 1.0), (8, 2, 17, 8, 1.0), (8, 2, 1, 8, 1.0)]
  
  
  
  
  
  
  
  def       _UpperCamelCase	(					SCREAMING_SNAKE_CASE__ )     ->       Dict[str, str]:
  
  
  		'''simple docstring'''
  
  
  		snake_case     :					int     =	RegNetConfig(depths=[2, 7, 17, 1]   ,					hidden_sizes=[8, 8, 8, 8]   ,					groups_width=8 )
  		if "in1k" in model_name:
  				snake_case     :					Optional[int]     =	RegNetForImageClassification(SCREAMING_SNAKE_CASE__ )
  		else:
  				snake_case     :					Optional[int]     =	RegNetModel(SCREAMING_SNAKE_CASE__ )
  		# create from model (with small weights)
  		snake_case     :					Any     =	FakeRegNetVisslWrapper(
  		    RegNet(FakeRegNetParams(depth=27   ,					group_width=1010   ,					w_a=1744   ,					w_a=620.83   ,					w_m=2.52 ) ) )
  
  		with torch.no_grad():
  				snake_case     :					Tuple     =	from_model.eval()
  				snake_case     :					Any     =	our_model.eval()
  
  				snake_case     :					List[str]     =	torch.randn((1, 3, 32, 32) )
  				# trace both
  				snake_case     :					Union[str, Any]     =	Tracker(SCREAMING_SNAKE_CASE__ )
  				snake_case     :					Dict     =	dest_tracker(SCREAMING_SNAKE_CASE__ ).parametrized
  
  				pprint(dest_tracker.nameamodule )
  				snake_case     :					Union[str, Any]     =	Tracker(SCREAMING_SNAKE_CASE__ )
  				snake_case     :					Tuple     =	src_tracker(SCREAMING_SNAKE_CASE__ ).parametrized
  
  		# convert the keys -> module dict to keys -> params
  		def to_params_dict(SCREAMING_SNAKE_CASE__ ):
  				snake_case     :					Optional[Any]     =	OrderedDict()
  				for name, module in dict_with_modules.items():
  						for param_name, param in module.state_dict().items():
  								snake_case     :					Optional[Any]     =	param
  				return params_dict
  
  		snake_case     :					Tuple     =	{}
  
  		snake_case     :					Optional[int]     =	to_params_dict(SCREAMING_SNAKE_CASE__ )
  		snake_case     :					Union[str, Any]     =	to_params_dict(SCREAMING_SNAKE_CASE__ )
  
  		for (src_key, src_param), (dest_key, dest_param) in zip(src_state_dict.items()   ,					dst_state_dict.items() ):
  				snake_case     :					Dict     =	dest_key
  				logger.info(F'{src_key} -> {dest_key}' )
  		# if "in1k" was in the model_name it means it must have a classification head (was finetuned)
  		if "in1k" in model_name:
  				snake_case     :					Dict     =	'''classifier.1.weight'''
  				snake_case     :					Tuple     =	'''classifier.1.bias'''
  
  		return from_to_ours_keys
  
  
  
  
  def       _UpperCamelCase	(					SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__ = None   ,					SCREAMING_SNAKE_CASE__ = True )     ->       List[Any]:
  
  
  		'''simple docstring'''
  
  
  		snake_case     :					Tuple     =	'''imagenet-1k-id2label.json'''
  		snake_case     :					Any     =	1000
  
  		snake_case     :					Tuple     =	'''huggingface/label-files'''
  		snake_case     :					Optional[Any]     =	num_labels
  		snake_case     :					int     =	json.load(open(cached_download(hf_hub_url(SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					repo_type='''dataset''' ) )   ,					'''r''' ) )
  		snake_case     :					Optional[Any]     =	{int(SCREAMING_SNAKE_CASE__ ): v for k, v in idalabel.items()}
  
  		snake_case     :					List[Any]     =	idalabel
  		snake_case     :					Union[str, Any]     =	{v: k for k, v in idalabel.items()}
  
  		snake_case     :					Optional[Any]     =	partial(SCREAMING_SNAKE_CASE__   ,					num_labels=SCREAMING_SNAKE_CASE__   ,					idalabel=SCREAMING_SNAKE_CASE__   ,					labelaid=SCREAMING_SNAKE_CASE__ )
  
  		snake_case     :					Dict     =	{
  		    '''regnet-y-10b-seer''': ImageNetPreTrainedConfig(
  		        depths=[2, 7, 17, 1]   ,					hidden_sizes=[2020, 4040, 1_1110, 2_8280]   ,					groups_width=1010 ),
  		    # finetuned on imagenet
  		    '''regnet-y-10b-seer-in1k''': ImageNetPreTrainedConfig(
  		        depths=[2, 7, 17, 1]   ,					hidden_sizes=[2020, 4040, 1_1110, 2_8280]   ,					groups_width=1010 ),
  		}
  
  		# add seer weights logic
  		def load_using_classy_vision(SCREAMING_SNAKE_CASE__ ) -> Tuple[Dict, Dict]:
  				snake_case     :					int     =	torch.hub.load_state_dict_from_url(SCREAMING_SNAKE_CASE__   ,					model_dir=str(SCREAMING_SNAKE_CASE__ )   ,					map_location='''cpu''' )
  				# check if we have a head, if yes add it
  				snake_case     :					Any     =	files['''classy_state_dict''']['''base_model''']['''model''']
  				return model_state_dict["trunk"], model_state_dict["heads"]
  
  		snake_case     :					int     =	{
  		    '''regnet-y-10b-seer''': partial(
  		        SCREAMING_SNAKE_CASE__   ,					'''https://dl.fbaipublicfiles.com/vissl/model_zoo/seer_regnet10B/model_iteration124500_conso.torch'''   ,					),
  		    '''regnet-y-10b-seer-in1k''': partial(
  		        SCREAMING_SNAKE_CASE__   ,					'''https://dl.fbaipublicfiles.com/vissl/model_zoo/seer_finetuned/seer_10b_finetuned_in1k_model_phase28_conso.torch'''   ,					),
  		}
  
  		snake_case     :					Optional[Any]     =	get_from_to_our_keys(SCREAMING_SNAKE_CASE__ )
  
  		if not (save_directory / F'{model_name}.pth').exists():
  				logger.info('''Loading original state_dict.''' )
  				snake_case ,snake_case     :					int     =	names_to_from_model[model_name]()
  				snake_case     :					Union[str, Any]     =	from_state_dict_trunk
  				if "in1k" in model_name:
  						# add the head
  						snake_case     :					int     =	{**from_state_dict_trunk, **from_state_dict_head}
  				logger.info('''Done!''' )
  
  				snake_case     :					int     =	{}
  
  				snake_case     :					List[str]     =	list(from_state_dict.keys() )
  				snake_case     :					Union[str, Any]     =	R'''\.block.-part.'''
  				# this is "interesting", so the original checkpoints have `block[0,1]-part` in each key name, we remove it
  				for key in from_state_dict.keys():
  						# remove the weird "block[0,1]-part" from the key
  						snake_case     :					int     =	re.sub(SCREAMING_SNAKE_CASE__   ,					''''''   ,					SCREAMING_SNAKE_CASE__ )
  						# now src_key from the model checkpoints is the one we got from the original model after tracing, so use it to get the correct destination key
  						snake_case     :					Optional[int]     =	from_to_ours_keys[src_key]
  						# store the parameter with our key
  						snake_case     :					int     =	from_state_dict[key]
  						not_used_keys.remove(SCREAMING_SNAKE_CASE__ )
  				# check that all keys have been updated
  				assert len(SCREAMING_SNAKE_CASE__ ) == 0, F'Some keys where not used {",".join(SCREAMING_SNAKE_CASE__ )}'
  
  				logger.info(F'The following keys were not used: {",".join(SCREAMING_SNAKE_CASE__ )}' )
  
  				# save our state dict to disk
  				torch.save(SCREAMING_SNAKE_CASE__   ,					save_directory / F'{model_name}.pth' )
  
  				del converted_state_dict
  		else:
  				logger.info('''The state_dict was already stored on disk.''' )
  		if push_to_hub:
  				logger.info(F'Token is {os.environ["HF_TOKEN"]}' )
  				logger.info('''Loading our model.''' )
  				# create our model
  				snake_case     :					Optional[int]     =	names_to_config[model_name]
  				snake_case     :					List[Any]     =	RegNetModel
  				if "in1k" in model_name:
  						snake_case     :					Tuple     =	RegNetForImageClassification
  				snake_case     :					str     =	our_model_func(SCREAMING_SNAKE_CASE__ )
  				# place our model to the meta device (so remove all the weights)
  				our_model.to(torch.device('''meta''' ) )
  				logger.info('''Loading state_dict in our model.''' )
  				# load state dict
  				snake_case     :					Union[str, Any]     =	our_model.state_dict().keys()
  				PreTrainedModel._load_pretrained_model_low_mem(
  				    SCREAMING_SNAKE_CASE__   ,					SCREAMING_SNAKE_CASE__   ,					[save_directory / F'{model_name}.pth'] )
  				logger.info('''Finally, pushing!''' )
  				# push it to hub
  				our_model.push_to_hub(
  				    repo_path_or_name=save_directory / model_name   ,					commit_message='''Add model'''   ,					output_dir=save_directory / model_name   ,					)
  				snake_case     :					Union[str, Any]     =	384
  				# we can use the convnext one
  				snake_case     :					Tuple     =	AutoImageProcessor.from_pretrained('''facebook/convnext-base-224-22k-1k'''   ,					size=SCREAMING_SNAKE_CASE__ )
  				image_processor.push_to_hub(
  				    repo_path_or_name=save_directory / model_name   ,					commit_message='''Add image processor'''   ,					output_dir=save_directory / model_name   ,					)
  
  
  if __name__ == "__main__":
  						lowercase__						=			argparse.ArgumentParser()
  						# Required parameters
  						parser.add_argument(
  						    "--model_name",
  						    default=None,
  						    type=str,
  						    help=(
  						        "The name of the model you wish to convert, it must be one of the supported regnet* architecture,"
  						        " currently: regnetx-*, regnety-*. If `None`, all of them will the converted."
  						    ),
  						)
  						parser.add_argument(
  						    "--pytorch_dump_folder_path",
  						    default=None,
  						    type=Path,
  						    required=True,
  						    help="Path to the output PyTorch model directory.",
  						)
  						parser.add_argument(
  						    "--push_to_hub",
  						    default=True,
  						    type=bool,
  						    required=False,
  						    help="If True, push model and image processor to the hub.",
  						)
  
  						lowercase__						=			parser.parse_args()
  
  						lowercase__						=			args.pytorch_dump_folder_path
  						pytorch_dump_folder_path.mkdir(exist_ok=True, parents=True)
  						convert_weights_and_push(pytorch_dump_folder_path, args.model_name, args.push_to_hub)
  
  
  
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__			)
  
  
  