




"""simple docstring"""

import json
import os
import shutil
import warnings
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import List

from ..utils import logging
from . import BaseTransformersCLICommand


try:
		from cookiecutter.main import cookiecutter

		_lowerCamelCase  :			Optional[Any]        =       True
except ImportError:
		_lowerCamelCase  :			str        =       False

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

def 		lowercase_      (		_UpperCAmelCase	):




	"""simple docstring"""






	return AddNewModelCommand(args.testing  ,      args.testing_file  ,      path=args.path	)

class        lowercase      (    __UpperCAmelCase):







				@staticmethod
				def 					a_  (							_lowerCamelCase   :					ArgumentParser       ):



					"""simple docstring"""

					A_  :				int	   =		parser.add_parser('''add-new-model'''       )
					add_new_model_parser.add_argument('''--testing'''      ,     action='''store_true'''      ,     help='''If in testing mode.'''       )
					add_new_model_parser.add_argument('''--testing_file'''      ,     type=_lowerCamelCase      ,     help='''Configuration file on which to run.'''       )
					add_new_model_parser.add_argument(
					    '''--path'''      ,     type=_lowerCamelCase      ,     help='''Path to cookiecutter. Should only be used for testing purposes.'''       )
					add_new_model_parser.set_defaults(func=_lowerCamelCase       )







				def __init__(							self   :					Any      ,     _lowerCamelCase   :					bool      ,     _lowerCamelCase   :					str      ,     _lowerCamelCase   :					Tuple=None      ,     *_lowerCamelCase   :					Tuple       ):



					"""simple docstring"""

					A_  :				str	   =		testing
					A_  :				List[Any]	   =		testing_file
					A_  :				str	   =		path







				def 					a_  (							self   :					Tuple       ):



					"""simple docstring"""

					warnings.warn(
					    '''The command `transformers-cli add-new-model` is deprecated and will be removed in v5 of Transformers. '''
					    '''It is not actively maintained anymore, so might give a result that won\'t pass all tests and quality '''
					    '''checks, you should use `transformers-cli add-new-model-like` instead.'''       )
					if not _has_cookiecutter:
						raise ImportError(
						    '''Model creation dependencies are required to use the `add_new_model` command. Install them by running '''
						    '''the following at the root of your `transformers` clone:\n\n\t$ pip install -e .[modelcreation]\n'''       )
					# Ensure that there is no other `cookiecutter-template-xxx` directory in the current working directory
					A_  :				str	   =		[directory for directory in os.listdir() if '''cookiecutter-template-''' == directory[:22]]
					if len(_lowerCamelCase       ) > 0:
						raise ValueError(
						    '''Several directories starting with `cookiecutter-template-` in current working directory. '''
						    '''Please clean your directory by removing all folders starting with `cookiecutter-template-` or '''
						    '''change your working directory.'''       )

					A_  :				List[Any]	   =		(
					    Path(_lowerCamelCase       ).parent.parent.parent.parent if self._path is None else Path(self._path       ).parent.parent
					)
					A_  :				Tuple	   =		path_to_transformer_root / '''templates''' / '''adding_a_new_model'''

					# Execute cookiecutter
					if not self._testing:
						cookiecutter(str(_lowerCamelCase       )       )
					else:
						with open(self._testing_file      ,     '''r'''       ) as configuration_file:
							A_  :				Dict	   =		json.load(_lowerCamelCase       )

						cookiecutter(
						    str(path_to_cookiecutter if self._path is None else self._path       )      ,     no_input=_lowerCamelCase      ,     extra_context=_lowerCamelCase      ,     )

					A_  :				Optional[Any]	   =		[directory for directory in os.listdir() if '''cookiecutter-template-''' in directory[:22]][0]

					# Retrieve configuration
					with open(directory + '''/configuration.json'''      ,     '''r'''       ) as configuration_file:
						A_  :				str	   =		json.load(_lowerCamelCase       )

					A_  :				Any	   =		configuration['''lowercase_modelname''']
					A_  :				Dict	   =		configuration['''generate_tensorflow_pytorch_and_flax''']
					os.remove(F"""{directory}/configuration.json"""       )

					A_  :				Tuple	   =		'''PyTorch''' in generate_tensorflow_pytorch_and_flax
					A_  :				Dict	   =		'''TensorFlow''' in generate_tensorflow_pytorch_and_flax
					A_  :				int	   =		'''Flax''' in generate_tensorflow_pytorch_and_flax

					A_  :				Optional[Any]	   =		F"""{path_to_transformer_root}/src/transformers/models/{lowercase_model_name}"""
					os.makedirs(_lowerCamelCase      ,     exist_ok=_lowerCamelCase       )
					os.makedirs(F"""{path_to_transformer_root}/tests/models/{lowercase_model_name}"""      ,     exist_ok=_lowerCamelCase       )

					# Tests require submodules as they have parent imports
					with open(F"""{path_to_transformer_root}/tests/models/{lowercase_model_name}/__init__.py"""      ,     '''w'''       ):
						pass

					shutil.move(
					    F"""{directory}/__init__.py"""      ,     F"""{model_dir}/__init__.py"""      ,     )
					shutil.move(
					    F"""{directory}/configuration_{lowercase_model_name}.py"""      ,     F"""{model_dir}/configuration_{lowercase_model_name}.py"""      ,     )

					def remove_copy_lines(_lowerCamelCase   :					Optional[int]       ):
						with open(_lowerCamelCase      ,     '''r'''       ) as f:
							A_  :				List[Any]	   =		f.readlines()
						with open(_lowerCamelCase      ,     '''w'''       ) as f:
							for line in lines:
								if "# Copied from transformers." not in line:
									f.write(_lowerCamelCase       )

					if output_pytorch:
						if not self._testing:
							remove_copy_lines(F"""{directory}/modeling_{lowercase_model_name}.py"""       )

						shutil.move(
						    F"""{directory}/modeling_{lowercase_model_name}.py"""      ,     F"""{model_dir}/modeling_{lowercase_model_name}.py"""      ,     )

						shutil.move(
						    F"""{directory}/test_modeling_{lowercase_model_name}.py"""      ,     F"""{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_{lowercase_model_name}.py"""      ,     )
					else:
						os.remove(F"""{directory}/modeling_{lowercase_model_name}.py"""       )
						os.remove(F"""{directory}/test_modeling_{lowercase_model_name}.py"""       )

					if output_tensorflow:
						if not self._testing:
							remove_copy_lines(F"""{directory}/modeling_tf_{lowercase_model_name}.py"""       )

						shutil.move(
						    F"""{directory}/modeling_tf_{lowercase_model_name}.py"""      ,     F"""{model_dir}/modeling_tf_{lowercase_model_name}.py"""      ,     )

						shutil.move(
						    F"""{directory}/test_modeling_tf_{lowercase_model_name}.py"""      ,     F"""{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_tf_{lowercase_model_name}.py"""      ,     )
					else:
						os.remove(F"""{directory}/modeling_tf_{lowercase_model_name}.py"""       )
						os.remove(F"""{directory}/test_modeling_tf_{lowercase_model_name}.py"""       )

					if output_flax:
						if not self._testing:
							remove_copy_lines(F"""{directory}/modeling_flax_{lowercase_model_name}.py"""       )

						shutil.move(
						    F"""{directory}/modeling_flax_{lowercase_model_name}.py"""      ,     F"""{model_dir}/modeling_flax_{lowercase_model_name}.py"""      ,     )

						shutil.move(
						    F"""{directory}/test_modeling_flax_{lowercase_model_name}.py"""      ,     F"""{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_flax_{lowercase_model_name}.py"""      ,     )
					else:
						os.remove(F"""{directory}/modeling_flax_{lowercase_model_name}.py"""       )
						os.remove(F"""{directory}/test_modeling_flax_{lowercase_model_name}.py"""       )

					shutil.move(
					    F"""{directory}/{lowercase_model_name}.md"""      ,     F"""{path_to_transformer_root}/docs/source/en/model_doc/{lowercase_model_name}.md"""      ,     )

					shutil.move(
					    F"""{directory}/tokenization_{lowercase_model_name}.py"""      ,     F"""{model_dir}/tokenization_{lowercase_model_name}.py"""      ,     )

					shutil.move(
					    F"""{directory}/tokenization_fast_{lowercase_model_name}.py"""      ,     F"""{model_dir}/tokenization_{lowercase_model_name}_fast.py"""      ,     )

					from os import fdopen, remove
					from shutil import copymode, move
					from tempfile import mkstemp

					def replace(_lowerCamelCase   :					str      ,     _lowerCamelCase   :					str      ,     _lowerCamelCase   :					List[str]       ):
						# Create temp file
						A_   , A_  :				Union[str, Any]	   =		mkstemp()
						A_  :				List[Any]	   =		False
						with fdopen(_lowerCamelCase      ,     '''w'''       ) as new_file:
							with open(_lowerCamelCase       ) as old_file:
								for line in old_file:
									new_file.write(_lowerCamelCase       )
									if line_to_copy_below in line:
										A_  :				Union[str, Any]	   =		True
										for line_to_copy in lines_to_copy:
											new_file.write(_lowerCamelCase       )

						if not line_found:
							raise ValueError(F"""Line {line_to_copy_below} was not found in file."""       )

						# Copy the file permissions from the old file to the new file
						copymode(_lowerCamelCase      ,     _lowerCamelCase       )
						# Remove original file
						remove(_lowerCamelCase       )
						# Move new file
						move(_lowerCamelCase      ,     _lowerCamelCase       )

					def skip_units(_lowerCamelCase   :					str       ):
						return (
						    ("generating PyTorch" in line and not output_pytorch)
						    or ("generating TensorFlow" in line and not output_tensorflow)
						    or ("generating Flax" in line and not output_flax)
						)

					def replace_in_files(_lowerCamelCase   :					Any       ):
						with open(_lowerCamelCase       ) as datafile:
							A_  :				Optional[Any]	   =		[]
							A_  :				Optional[int]	   =		False
							A_  :				str	   =		False
							for line in datafile:
								if "# To replace in: " in line and "##" not in line:
									A_  :				Dict	   =		line.split('''"'''       )[1]
									A_  :				str	   =		skip_units(_lowerCamelCase       )
								elif "# Below: " in line and "##" not in line:
									A_  :				Optional[int]	   =		line.split('''"'''       )[1]
									A_  :				Any	   =		skip_units(_lowerCamelCase       )
								elif "# End." in line and "##" not in line:
									if not skip_file and not skip_snippet:
										replace(_lowerCamelCase      ,     _lowerCamelCase      ,     _lowerCamelCase       )

									A_  :				int	   =		[]
								elif "# Replace with" in line and "##" not in line:
									A_  :				Any	   =		[]
								elif "##" not in line:
									lines_to_copy.append(_lowerCamelCase       )

						remove(_lowerCamelCase       )

					replace_in_files(F"""{directory}/to_replace_{lowercase_model_name}.py"""       )
					os.rmdir(_lowerCamelCase       )











"""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





