




import argparse

import torch

from diffusers.pipelines.stable_diffusion.convert_from_ckpt import download_from_original_stable_diffusion_ckpt


if __name__ == "__main__":
		_lowerCamelCase  :			Optional[int]        =       argparse.ArgumentParser()

		parser.add_argument(
		    '--checkpoint_path', default=None, type=str, required=True, help='Path to the checkpoint to convert.'
		)
		# !wget https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml
		parser.add_argument(
		    '--original_config_file',
		    default=None,
		    type=str,
		    help='The YAML config file corresponding to the original architecture.',
		)
		parser.add_argument(
		    '--num_in_channels',
		    default=None,
		    type=int,
		    help='The number of input channels. If `None` number of input channels will be automatically inferred.',
		)
		parser.add_argument(
		    '--scheduler_type',
		    default='pndm',
		    type=str,
		    help='Type of scheduler to use. Should be one of [\'pndm\', \'lms\', \'ddim\', \'euler\', \'euler-ancestral\', \'dpm\']',
		)
		parser.add_argument(
		    '--pipeline_type',
		    default=None,
		    type=str,
		    help=(
		        'The pipeline type. One of \'FrozenOpenCLIPEmbedder\', \'FrozenCLIPEmbedder\', \'PaintByExample\''
		        '. If `None` pipeline will be automatically inferred.'
		    ),
		)
		parser.add_argument(
		    '--image_size',
		    default=None,
		    type=int,
		    help=(
		        'The image size that the model was trained on. Use 512 for Stable Diffusion v1.X and Stable Siffusion v2'
		        ' Base. Use 768 for Stable Diffusion v2.'
		    ),
		)
		parser.add_argument(
		    '--prediction_type',
		    default=None,
		    type=str,
		    help=(
		        'The prediction type that the model was trained on. Use \'epsilon\' for Stable Diffusion v1.X and Stable'
		        ' Diffusion v2 Base. Use \'v_prediction\' for Stable Diffusion v2.'
		    ),
		)
		parser.add_argument(
		    '--extract_ema',
		    action='store_true',
		    help=(
		        'Only relevant for checkpoints that have both EMA and non-EMA weights. Whether to extract the EMA weights'
		        ' or not. Defaults to `False`. Add `--extract_ema` to extract the EMA weights. EMA weights usually yield'
		        ' higher quality images for inference. Non-EMA weights are usually better to continue fine-tuning.'
		    ),
		)
		parser.add_argument(
		    '--upcast_attention',
		    action='store_true',
		    help=(
		        'Whether the attention computation should always be upcasted. This is necessary when running stable'
		        ' diffusion 2.1.'
		    ),
		)
		parser.add_argument(
		    '--from_safetensors',
		    action='store_true',
		    help='If `--checkpoint_path` is in `safetensors` format, load checkpoint with safetensors instead of PyTorch.',
		)
		parser.add_argument(
		    '--to_safetensors',
		    action='store_true',
		    help='Whether to store pipeline in safetensors format or not.',
		)
		parser.add_argument('--dump_path', default=None, type=str, required=True, help='Path to the output model.')
		parser.add_argument('--device', type=str, help='Device to use (e.g. cpu, cuda:0, cuda:1, etc.)')
		parser.add_argument(
		    '--stable_unclip',
		    type=str,
		    default=None,
		    required=False,
		    help='Set if this is a stable unCLIP model. One of \'txt2img\' or \'img2img\'.',
		)
		parser.add_argument(
		    '--stable_unclip_prior',
		    type=str,
		    default=None,
		    required=False,
		    help='Set if this is a stable unCLIP txt2img model. Selects which prior to use. If `--stable_unclip` is set to `txt2img`, the karlo prior (https://huggingface.co/kakaobrain/karlo-v1-alpha/tree/main/prior) is selected by default.',
		)
		parser.add_argument(
		    '--clip_stats_path',
		    type=str,
		    help='Path to the clip stats file. Only required if the stable unclip model\'s config specifies `model.params.noise_aug_config.params.clip_stats_path`.',
		    required=False,
		)
		parser.add_argument(
		    '--controlnet', action='store_true', default=None, help='Set flag if this is a controlnet checkpoint.'
		)
		parser.add_argument('--half', action='store_true', help='Save weights in half precision.')
		parser.add_argument(
		    '--vae_path',
		    type=str,
		    default=None,
		    required=False,
		    help='Set to a path, hub id to an already converted vae to not convert it again.',
		)
		_lowerCamelCase  :			Optional[Any]        =       parser.parse_args()

		_lowerCamelCase  :			Optional[Any]        =       download_from_original_stable_diffusion_ckpt(
		    checkpoint_path=args.checkpoint_path,
		    original_config_file=args.original_config_file,
		    image_size=args.image_size,
		    prediction_type=args.prediction_type,
		    model_type=args.pipeline_type,
		    extract_ema=args.extract_ema,
		    scheduler_type=args.scheduler_type,
		    num_in_channels=args.num_in_channels,
		    upcast_attention=args.upcast_attention,
		    from_safetensors=args.from_safetensors,
		    device=args.device,
		    stable_unclip=args.stable_unclip,
		    stable_unclip_prior=args.stable_unclip_prior,
		    clip_stats_path=args.clip_stats_path,
		    controlnet=args.controlnet,
		    vae_path=args.vae_path,
		)

		if args.half:
				pipe.to(torch_dtype=torch.floataa)

		if args.controlnet:
				# only save the controlnet model
				pipe.controlnet.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)
		else:
				pipe.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)











"""simple docstring"""

import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import TransformeraDModel, VQDiffusionPipeline, VQDiffusionScheduler, VQModel
from diffusers.pipelines.vq_diffusion.pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings
from diffusers.utils import load_numpy, slow, torch_device
from diffusers.utils.testing_utils import require_torch_gpu


_lowerCamelCase  :			Any        =       False

class        lowercase      (    unittest.TestCase):







				def 					a_  (							self   :					Tuple       ):



					"""simple docstring"""

					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()







				@property
				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					return 12







				@property
				def 					a_  (							self   :					List[str]       ):



					"""simple docstring"""

					return 12







				@property
				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					return 32







				@property
				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					torch.manual_seed(0       )
					A_  :				Union[str, Any]	   =		VQModel(
					    block_out_channels=[32, 64]      ,     in_channels=3      ,     out_channels=3      ,     down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']      ,     up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']      ,     latent_channels=3      ,     num_vq_embeddings=self.num_embed      ,     vq_embed_dim=3      ,     )
					return model







				@property
				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				int	   =		CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )
					return tokenizer







				@property
				def 					a_  (							self   :					int       ):



					"""simple docstring"""

					torch.manual_seed(0       )
					A_  :				Optional[int]	   =		CLIPTextConfig(
					    bos_token_id=0      ,     eos_token_id=2      ,     hidden_size=self.text_embedder_hidden_size      ,     intermediate_size=37      ,     layer_norm_eps=1E-05      ,     num_attention_heads=4      ,     num_hidden_layers=5      ,     pad_token_id=1      ,     vocab_size=10_00      ,     )
					return CLIPTextModel(_lowerCamelCase       )







				@property
				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					torch.manual_seed(0       )

					A_  :				Optional[Any]	   =		12
					A_  :				Optional[int]	   =		12

					A_  :				int	   =		{
					    '''attention_bias''': True,
					    '''cross_attention_dim''': 32,
					    '''attention_head_dim''': height * width,
					    '''num_attention_heads''': 1,
					    '''num_vector_embeds''': self.num_embed,
					    '''num_embeds_ada_norm''': self.num_embeds_ada_norm,
					    '''norm_num_groups''': 32,
					    '''sample_size''': width,
					    '''activation_fn''': '''geglu-approximate''',
					}

					A_  :				Tuple	   =		TransformeraDModel(**_lowerCamelCase       )
					return model







				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		'''cpu'''

					A_  :				Union[str, Any]	   =		self.dummy_vqvae
					A_  :				str	   =		self.dummy_text_encoder
					A_  :				List[Any]	   =		self.dummy_tokenizer
					A_  :				int	   =		self.dummy_transformer
					A_  :				Any	   =		VQDiffusionScheduler(self.num_embed       )
					A_  :				Optional[Any]	   =		LearnedClassifierFreeSamplingEmbeddings(learnable=_lowerCamelCase       )

					A_  :				Dict	   =		VQDiffusionPipeline(
					    vqvae=_lowerCamelCase      ,     text_encoder=_lowerCamelCase      ,     tokenizer=_lowerCamelCase      ,     transformer=_lowerCamelCase      ,     scheduler=_lowerCamelCase      ,     learned_classifier_free_sampling_embeddings=_lowerCamelCase      ,     )
					A_  :				List[Any]	   =		pipe.to(_lowerCamelCase       )
					pipe.set_progress_bar_config(disable=_lowerCamelCase       )

					A_  :				List[Any]	   =		'''teddy bear playing in the pool'''

					A_  :				List[Any]	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				List[Any]	   =		pipe([prompt]      ,     generator=_lowerCamelCase      ,     num_inference_steps=2      ,     output_type='''np'''       )
					A_  :				Any	   =		output.images

					A_  :				List[str]	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				List[Any]	   =		pipe(
					    [prompt]      ,     generator=_lowerCamelCase      ,     output_type='''np'''      ,     return_dict=_lowerCamelCase      ,     num_inference_steps=2       )[0]

					A_  :				Optional[int]	   =		image[0, -3:, -3:, -1]
					A_  :				Optional[int]	   =		image_from_tuple[0, -3:, -3:, -1]

					assert image.shape == (1, 24, 24, 3)

					A_  :				Optional[int]	   =		np.array([0.6551, 0.6168, 0.5008, 0.5676, 0.5659, 0.4295, 0.6073, 0.5599, 0.4992]       )

					assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2
					assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1E-2







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		'''cpu'''

					A_  :				int	   =		self.dummy_vqvae
					A_  :				List[str]	   =		self.dummy_text_encoder
					A_  :				Optional[Any]	   =		self.dummy_tokenizer
					A_  :				Any	   =		self.dummy_transformer
					A_  :				Any	   =		VQDiffusionScheduler(self.num_embed       )
					A_  :				Optional[int]	   =		LearnedClassifierFreeSamplingEmbeddings(
					    learnable=_lowerCamelCase      ,     hidden_size=self.text_embedder_hidden_size      ,     length=tokenizer.model_max_length       )

					A_  :				int	   =		VQDiffusionPipeline(
					    vqvae=_lowerCamelCase      ,     text_encoder=_lowerCamelCase      ,     tokenizer=_lowerCamelCase      ,     transformer=_lowerCamelCase      ,     scheduler=_lowerCamelCase      ,     learned_classifier_free_sampling_embeddings=_lowerCamelCase      ,     )
					A_  :				List[Any]	   =		pipe.to(_lowerCamelCase       )
					pipe.set_progress_bar_config(disable=_lowerCamelCase       )

					A_  :				Any	   =		'''teddy bear playing in the pool'''

					A_  :				str	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				Optional[Any]	   =		pipe([prompt]      ,     generator=_lowerCamelCase      ,     num_inference_steps=2      ,     output_type='''np'''       )
					A_  :				Tuple	   =		output.images

					A_  :				Optional[int]	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				List[str]	   =		pipe(
					    [prompt]      ,     generator=_lowerCamelCase      ,     output_type='''np'''      ,     return_dict=_lowerCamelCase      ,     num_inference_steps=2       )[0]

					A_  :				Optional[int]	   =		image[0, -3:, -3:, -1]
					A_  :				Optional[Any]	   =		image_from_tuple[0, -3:, -3:, -1]

					assert image.shape == (1, 24, 24, 3)

					A_  :				str	   =		np.array([0.6693, 0.6075, 0.4959, 0.5701, 0.5583, 0.4333, 0.6171, 0.5684, 0.4988]       )

					assert np.abs(image_slice.flatten() - expected_slice       ).max() < 2.0
					assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1E-2

@slow
@require_torch_gpu
class        lowercase      (    unittest.TestCase):







				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()







				def 					a_  (							self   :					str       ):



					"""simple docstring"""

					A_  :				str	   =		load_numpy(
					    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
					    '''/vq_diffusion/teddy_bear_pool_classifier_free_sampling.npy'''       )

					A_  :				int	   =		VQDiffusionPipeline.from_pretrained('''microsoft/vq-diffusion-ithq'''       )
					A_  :				Tuple	   =		pipeline.to(_lowerCamelCase       )
					pipeline.set_progress_bar_config(disable=_lowerCamelCase       )

					# requires GPU generator for gumbel softmax
					# don't use GPU generator in tests though
					A_  :				Dict	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				Union[str, Any]	   =		pipeline(
					    '''teddy bear playing in the pool'''      ,     num_images_per_prompt=1      ,     generator=_lowerCamelCase      ,     output_type='''np'''      ,     )

					A_  :				Optional[int]	   =		output.images[0]

					assert image.shape == (2_56, 2_56, 3)
					assert np.abs(expected_image - image       ).max() < 2.0





