


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

		__A				=True
except ImportError:
		__A				=False

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







def       _UpperCamelCase (						UpperCamelCase__   ):
			return AddNewModelCommand(args.testing	, args.testing_file	, path=args.path   )


class 						_snake_case   (       a__		):




					@staticmethod
					def 			snake_case__  ( _lowerCamelCase):
								UpperCAmelCase__					:   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							,							_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase=None							,							*_lowerCamelCase):
								UpperCAmelCase__					:   Dict       				=       testing
								UpperCAmelCase__					:   Optional[int]       				=       testing_file
								UpperCAmelCase__					:   Union[str, Any]       				=       path




					def 			snake_case__  ( self):
								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
								UpperCAmelCase__					:   int       				=       [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.""")

								UpperCAmelCase__					:   str       				=       (
								    Path(_lowerCamelCase).parent.parent.parent.parent if self._path is None else Path(self._path).parent.parent
								)
								UpperCAmelCase__					:   List[Any]       				=       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:
														UpperCAmelCase__					:   Any       				=       json.load(_lowerCamelCase)

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

								UpperCAmelCase__					:   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:
											UpperCAmelCase__					:   Dict       				=       json.load(_lowerCamelCase)

								UpperCAmelCase__					:   Dict       				=       configuration["""lowercase_modelname"""]
								UpperCAmelCase__					:   Tuple       				=       configuration["""generate_tensorflow_pytorch_and_flax"""]
								os.remove(f'''{directory}/configuration.json''')

								UpperCAmelCase__					:   Optional[int]       				=       """PyTorch""" in generate_tensorflow_pytorch_and_flax
								UpperCAmelCase__					:   Tuple       				=       """TensorFlow""" in generate_tensorflow_pytorch_and_flax
								UpperCAmelCase__					:   List[Any]       				=       """Flax""" in generate_tensorflow_pytorch_and_flax

								UpperCAmelCase__					:   Optional[int]       				=       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):
											with open(_lowerCamelCase							,							"""r""") as f:
														UpperCAmelCase__					:   int       				=       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							,							_lowerCamelCase							,							_lowerCamelCase):
											# Create temp file
											UpperCAmelCase__					:   Any       				=       mkstemp()
											UpperCAmelCase__					:   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:
																							UpperCAmelCase__					:   Tuple       				=       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):
											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):
											with open(_lowerCamelCase) as datafile:
														UpperCAmelCase__					:   int       				=       []
														UpperCAmelCase__					:   Any       				=       False
														UpperCAmelCase__					:   Any       				=       False
														for line in datafile:
																	if "# To replace in: " in line and "##" not in line:
																				UpperCAmelCase__					:   Union[str, Any]       				=       line.split("""\"""")[1]
																				UpperCAmelCase__					:   Union[str, Any]       				=       skip_units(_lowerCamelCase)
																	elif "# Below: " in line and "##" not in line:
																				UpperCAmelCase__					:   List[Any]       				=       line.split("""\"""")[1]
																				UpperCAmelCase__					:   Optional[int]       				=       skip_units(_lowerCamelCase)
																	elif "# End." in line and "##" not in line:
																				if not skip_file and not skip_snippet:
																							replace(_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase)

																				UpperCAmelCase__					:   Tuple       				=       []
																	elif "# Replace with" in line and "##" not in line:
																				UpperCAmelCase__					:   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'''



from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer


@dataclass
class 						_snake_case   (       a__		):
					lowerCAmelCase :torch.FloatTensor


class 						_snake_case   (       a__    ,				a__		):




					@register_to_config
					def __init__( self							,							_lowerCamelCase = 3							,							_lowerCamelCase = 3							,							_lowerCamelCase = ("DownEncoderBlock2D",)							,							_lowerCamelCase = ("UpDecoderBlock2D",)							,							_lowerCamelCase = (64,)							,							_lowerCamelCase = 1							,							_lowerCamelCase = "silu"							,							_lowerCamelCase = 3							,							_lowerCamelCase = 32							,							_lowerCamelCase = 256							,							_lowerCamelCase = 32							,							_lowerCamelCase = None							,							_lowerCamelCase = 0.18215							,							_lowerCamelCase = "group"							,							):
								super().__init__()

								# pass init params to Encoder
								UpperCAmelCase__					:   str       				=       Encoder(
								    in_channels=_lowerCamelCase							,							out_channels=_lowerCamelCase							,							down_block_types=_lowerCamelCase							,							block_out_channels=_lowerCamelCase							,							layers_per_block=_lowerCamelCase							,							act_fn=_lowerCamelCase							,							norm_num_groups=_lowerCamelCase							,							double_z=_lowerCamelCase							,							)

								UpperCAmelCase__					:   Optional[Any]       				=       vq_embed_dim if vq_embed_dim is not None else latent_channels

								UpperCAmelCase__					:   Any       				=       nn.Convad(_lowerCamelCase							,							_lowerCamelCase							,							1)
								UpperCAmelCase__					:   Optional[int]       				=       VectorQuantizer(_lowerCamelCase							,							_lowerCamelCase							,							beta=0.25							,							remap=_lowerCamelCase							,							sane_index_shape=_lowerCamelCase)
								UpperCAmelCase__					:   Optional[int]       				=       nn.Convad(_lowerCamelCase							,							_lowerCamelCase							,							1)

								# pass init params to Decoder
								UpperCAmelCase__					:   str       				=       Decoder(
								    in_channels=_lowerCamelCase							,							out_channels=_lowerCamelCase							,							up_block_types=_lowerCamelCase							,							block_out_channels=_lowerCamelCase							,							layers_per_block=_lowerCamelCase							,							act_fn=_lowerCamelCase							,							norm_num_groups=_lowerCamelCase							,							norm_type=_lowerCamelCase							,							)




					@apply_forward_hook
					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase = True):
								UpperCAmelCase__					:   Union[str, Any]       				=       self.encoder(_lowerCamelCase)
								UpperCAmelCase__					:   str       				=       self.quant_conv(_lowerCamelCase)

								if not return_dict:
											return (h,)

								return VQEncoderOutput(latents=_lowerCamelCase)




					@apply_forward_hook
					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase = False							,							_lowerCamelCase = True):
								# also go through quantization layer
								if not force_not_quantize:
											UpperCAmelCase__				,    UpperCAmelCase__				,    UpperCAmelCase__					:   Optional[Any]       				=       self.quantize(_lowerCamelCase)
								else:
											UpperCAmelCase__					:   Union[str, Any]       				=       h
								UpperCAmelCase__					:   Any       				=       self.post_quant_conv(_lowerCamelCase)
								UpperCAmelCase__					:   Any       				=       self.decoder(_lowerCamelCase							,							quant if self.config.norm_type == """spatial""" else None)

								if not return_dict:
											return (dec,)

								return DecoderOutput(sample=_lowerCamelCase)




					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase = True):
								UpperCAmelCase__					:   Dict       				=       sample
								UpperCAmelCase__					:   Dict       				=       self.encode(_lowerCamelCase).latents
								UpperCAmelCase__					:   List[str]       				=       self.decode(_lowerCamelCase).sample

								if not return_dict:
											return (dec,)

								return DecoderOutput(sample=_lowerCamelCase)