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

   snake_case_ :  Any     =					True
except ImportError:
   snake_case_ :  Optional[int]     =					False

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



def  __snake_case     (    _UpperCAmelCase					:       Namespace):
   return AddNewModelCommand(args.testing,      args.testing_file,      path=args.path)


class 						lowercase__	(  snake_case_   ):






   '''simple docstring'''







   @staticmethod
   def       UpperCAmelCase				(			lowerCamelCase__   ):

      '''simple docstring'''







      UpperCamelCase       =  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__   ):

      '''simple docstring'''







      UpperCamelCase       =  testing
      UpperCamelCase       =  testing_file
      UpperCamelCase       =  path







   def       UpperCAmelCase				(			self   ):

      '''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
      UpperCamelCase       =  [directory for directory in os.listdir() if '''cookiecutter-template-''' == directory[:2_2]]
      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       =  (
          Path(lowerCamelCase__   ).parent.parent.parent.parent if self._path is None else Path(self._path   ).parent.parent
      )
      UpperCamelCase       =  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       =  json.load(lowerCamelCase__   )

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

      UpperCamelCase       =  [directory for directory in os.listdir() if '''cookiecutter-template-''' in directory[:2_2]][0]

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

      UpperCamelCase       =  configuration['''lowercase_modelname''']
      UpperCamelCase       =  configuration['''generate_tensorflow_pytorch_and_flax''']
      os.remove(f'{directory}/configuration.json'   )

      UpperCamelCase       =  '''PyTorch''' in generate_tensorflow_pytorch_and_flax
      UpperCamelCase       =  '''TensorFlow''' in generate_tensorflow_pytorch_and_flax
      UpperCamelCase       =  '''Flax''' in generate_tensorflow_pytorch_and_flax

      UpperCamelCase       =  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       =  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     , UpperCamelCase       =  mkstemp()
         UpperCamelCase       =  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       =  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       =  []
            UpperCamelCase       =  False
            UpperCamelCase       =  False
            for line in datafile:
               if "# To replace in: " in line and "##" not in line:
                  UpperCamelCase       =  line.split('''"'''   )[1]
                  UpperCamelCase       =  skip_units(lowerCamelCase__   )
               elif "# Below: " in line and "##" not in line:
                  UpperCamelCase       =  line.split('''"'''   )[1]
                  UpperCamelCase       =  skip_units(lowerCamelCase__   )
               elif "# End." in line and "##" not in line:
                  if not skip_file and not skip_snippet:
                     replace(lowerCamelCase__ ,   lowerCamelCase__ ,   lowerCamelCase__   )

                  UpperCamelCase       =  []
               elif "# Replace with" in line and "##" not in line:
                  UpperCamelCase       =  []
               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 queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
   from ..models.auto import AutoTokenizer


class 						lowercase__	:






   '''simple docstring'''







   def       UpperCAmelCase				(			self ,   lowerCamelCase__   ):

      '''simple docstring'''







      raise NotImplementedError()







   def       UpperCAmelCase				(			self   ):

      '''simple docstring'''







      raise NotImplementedError()




class 						lowercase__	(  snake_case_   ):






   '''simple docstring'''







   def __init__(			self ,   lowerCamelCase__ ,   lowerCamelCase__ = False ,   **lowerCamelCase__   ):

      '''simple docstring'''







      UpperCamelCase       =  tokenizer
      UpperCamelCase       =  skip_prompt
      UpperCamelCase       =  decode_kwargs

      # variables used in the streaming process
      UpperCamelCase       =  []
      UpperCamelCase       =  0
      UpperCamelCase       =  True







   def       UpperCAmelCase				(			self ,   lowerCamelCase__   ):

      '''simple docstring'''







      if len(value.shape   ) > 1 and value.shape[0] > 1:
         raise ValueError('''TextStreamer only supports batch size 1'''   )
      elif len(value.shape   ) > 1:
         UpperCamelCase       =  value[0]

      if self.skip_prompt and self.next_tokens_are_prompt:
         UpperCamelCase       =  False
         return

      # Add the new token to the cache and decodes the entire thing.
      self.token_cache.extend(value.tolist()   )
      UpperCamelCase       =  self.tokenizer.decode(self.token_cache ,   **self.decode_kwargs   )

      # After the symbol for a new line, we flush the cache.
      if text.endswith('''\n'''   ):
         UpperCamelCase       =  text[self.print_len :]
         UpperCamelCase       =  []
         UpperCamelCase       =  0
      # If the last token is a CJK character, we print the characters.
      elif len(lowerCamelCase__   ) > 0 and self._is_chinese_char(ord(text[-1]   )   ):
         UpperCamelCase       =  text[self.print_len :]
         self.print_len += len(lowerCamelCase__   )
      # Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
      # which may change with the subsequent token -- there are probably smarter ways to do this!)
      else:
         UpperCamelCase       =  text[self.print_len : text.rfind(''' '''   ) + 1]
         self.print_len += len(lowerCamelCase__   )

      self.on_finalized_text(lowerCamelCase__   )







   def       UpperCAmelCase				(			self   ):

      '''simple docstring'''







      if len(self.token_cache   ) > 0:
         UpperCamelCase       =  self.tokenizer.decode(self.token_cache ,   **self.decode_kwargs   )
         UpperCamelCase       =  text[self.print_len :]
         UpperCamelCase       =  []
         UpperCamelCase       =  0
      else:
         UpperCamelCase       =  ''''''

      UpperCamelCase       =  True
      self.on_finalized_text(lowerCamelCase__ ,   stream_end=lowerCamelCase__   )







   def       UpperCAmelCase				(			self ,   lowerCamelCase__ ,   lowerCamelCase__ = False   ):

      '''simple docstring'''







      print(lowerCamelCase__ ,   flush=lowerCamelCase__ ,   end='''''' if not stream_end else None   )







   def       UpperCAmelCase				(			self ,   lowerCamelCase__   ):

      '''simple docstring'''







      if (
          (cp >= 0x4_E_0_0 and cp <= 0x9_F_F_F)
          or (cp >= 0x3_4_0_0 and cp <= 0x4_D_B_F)  #
          or (cp >= 0x2_0_0_0_0 and cp <= 0x2_A_6_D_F)  #
          or (cp >= 0x2_A_7_0_0 and cp <= 0x2_B_7_3_F)  #
          or (cp >= 0x2_B_7_4_0 and cp <= 0x2_B_8_1_F)  #
          or (cp >= 0x2_B_8_2_0 and cp <= 0x2_C_E_A_F)  #
          or (cp >= 0xF_9_0_0 and cp <= 0xF_A_F_F)
          or (cp >= 0x2_F_8_0_0 and cp <= 0x2_F_A_1_F)  #
      ):  #
         return True

      return False




class 						lowercase__	(  snake_case_   ):






   '''simple docstring'''







   def __init__(			self ,   lowerCamelCase__ ,   lowerCamelCase__ = False ,   lowerCamelCase__ = None ,   **lowerCamelCase__   ):

      '''simple docstring'''







      super().__init__(lowerCamelCase__ ,   lowerCamelCase__ ,   **lowerCamelCase__   )
      UpperCamelCase       =  Queue()
      UpperCamelCase       =  None
      UpperCamelCase       =  timeout







   def       UpperCAmelCase				(			self ,   lowerCamelCase__ ,   lowerCamelCase__ = False   ):

      '''simple docstring'''







      self.text_queue.put(lowerCamelCase__ ,   timeout=self.timeout   )
      if stream_end:
         self.text_queue.put(self.stop_signal ,   timeout=self.timeout   )







   def __iter__(			self   ):

      '''simple docstring'''







      return self







   def       UpperCAmelCase				(			self   ):

      '''simple docstring'''







      UpperCamelCase       =  self.text_queue.get(timeout=self.timeout   )
      if value == self.stop_signal:
         raise StopIteration()
      else:
         return value





