


import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, ByTaTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
    __snake_case   							=						'''pt'''
elif is_tf_available():
    __snake_case   							=						'''tf'''
else:
    __snake_case   							=						'''jax'''



class 	__lowerCamelCase			(_a				,      unittest.TestCase						):
       _lowercase											=     ByTaTokenizer
       _lowercase											=     False







       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




            super().setUp()
            __UpperCamelCase =			ByTaTokenizer()
            tokenizer.save_pretrained(self.tmpdirname					)







       @cached_property
       def 		snake_case_	(			self: int					):






            '''simple docstring'''




            return ByTaTokenizer.from_pretrained('google/byt5-small'					)







       def 		snake_case_	(			self: Optional[int],**A_: int					):






            '''simple docstring'''




            return self.tokenizer_class.from_pretrained(self.tmpdirname,**A_					)







       def 		snake_case_	(			self: Any,A_: List[Any],A_: List[str]=False,A_: List[Any]=20,A_: List[Any]=5					):






            '''simple docstring'''




            __UpperCamelCase =			[]
            for i in range(len(A_					)					):
                 try:
                      __UpperCamelCase =			tokenizer.decode([i],clean_up_tokenization_spaces=A_					)
                 except UnicodeDecodeError:
                      pass
                 toks.append((i, tok)					)

            __UpperCamelCase =			list(filter(lambda A_					: re.match(r'^[ a-zA-Z]+$',t[1]					),A_					)					)
            __UpperCamelCase =			list(filter(lambda A_					: [t[0]] == tokenizer.encode(t[1],add_special_tokens=A_					),A_					)					)
            if max_length is not None and len(A_					) > max_length:
                 __UpperCamelCase =			toks[:max_length]
            if min_length is not None and len(A_					) < min_length and len(A_					) > 0:
                 while len(A_					) < min_length:
                      __UpperCamelCase =			toks + toks
        # toks_str = [t[1] for t in toks]
            __UpperCamelCase =			[t[0] for t in toks]

            # Ensure consistency
            __UpperCamelCase =			tokenizer.decode(A_,clean_up_tokenization_spaces=A_					)
            if " " not in output_txt and len(A_					) > 1:
                 __UpperCamelCase =			(
                     tokenizer.decode([toks_ids[0]],clean_up_tokenization_spaces=A_					)
                     + ' '
                     + tokenizer.decode(toks_ids[1:],clean_up_tokenization_spaces=A_					)
                 )
            if with_prefix_space:
                 __UpperCamelCase =			' ' + output_txt
            __UpperCamelCase =			tokenizer.encode(A_,add_special_tokens=A_					)
            return output_txt, output_ids







       def 		snake_case_	(			self: List[Any]					):






            '''simple docstring'''




            __UpperCamelCase =			self.ta_base_tokenizer
            __UpperCamelCase =			tokenizer(['hi</s>', 'I went to the gym</s>', '</s>']					)
            __UpperCamelCase =			tokenizer(['hi', 'I went to the gym', '']					)
            self.assertListEqual(batch_with_eos_added['input_ids'],batch_without_eos_added['input_ids']					)







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			self.ta_base_tokenizer
            __UpperCamelCase =			'Unicode €.'
            __UpperCamelCase =			tokenizer(A_					)
            __UpperCamelCase =			[88, 113, 108, 102, 114, 103, 104, 35, 229, 133, 175, 49, 1]
            self.assertEqual(encoded['input_ids'],A_					)

            # decoding
            __UpperCamelCase =			tokenizer.decode(A_					)
            self.assertEqual(A_,'Unicode €.</s>'					)

            __UpperCamelCase =			tokenizer('e è é ê ë'					)
            __UpperCamelCase =			[104, 35, 198, 171, 35, 198, 172, 35, 198, 173, 35, 198, 174, 1]
            self.assertEqual(encoded['input_ids'],A_					)
            # decoding
            __UpperCamelCase =			tokenizer.decode(A_					)
            self.assertEqual(A_,'e è é ê ë</s>'					)

            # encode/decode, but with `encode` instead of `__call__`
            self.assertEqual(tokenizer.decode(tokenizer.encode('e è é ê ë'					)					),'e è é ê ë</s>'					)







       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




            __UpperCamelCase =			self.ta_base_tokenizer
            __UpperCamelCase =			['A long paragraph for summarization.', 'Another paragraph for summarization.']
            # fmt: off
            __UpperCamelCase =			[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 1, 0]
            # fmt: on
            __UpperCamelCase =			tokenizer(A_,padding=A_,return_tensors=A_					)
            self.assertIsInstance(A_,A_					)

            if FRAMEWORK != "jax":
                 __UpperCamelCase =			list(batch.input_ids.numpy()[0]					)
            else:
                 __UpperCamelCase =			list(batch.input_ids.tolist()[0]					)

            self.assertListEqual(A_,A_					)

            self.assertEqual((2, 37),batch.input_ids.shape					)
            self.assertEqual((2, 37),batch.attention_mask.shape					)







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			self.ta_base_tokenizer
            __UpperCamelCase =			['A long paragraph for summarization.', 'Another paragraph for summarization.']
            __UpperCamelCase =			tokenizer(A_,padding=A_,return_tensors=A_					)
            # check if input_ids are returned and no decoder_input_ids
            self.assertIn('input_ids',A_					)
            self.assertIn('attention_mask',A_					)
            self.assertNotIn('decoder_input_ids',A_					)
            self.assertNotIn('decoder_attention_mask',A_					)







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			self.ta_base_tokenizer
            __UpperCamelCase =			[
                'Summary of the text.',
                'Another summary.',
            ]
            __UpperCamelCase =			tokenizer(
                text_target=A_,max_length=32,padding='max_length',truncation=A_,return_tensors=A_					)
            self.assertEqual(32,targets['input_ids'].shape[1]					)







       def 		snake_case_	(			self: Union[str, Any]					):






            '''simple docstring'''




            __UpperCamelCase =			self.ta_base_tokenizer
            __UpperCamelCase =			['A long paragraph for summarization. </s>']
            __UpperCamelCase =			['Summary of the text. </s>']
            # fmt: off
            __UpperCamelCase =			[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 35, 1]
            __UpperCamelCase =			[86, 120, 112, 112, 100, 117, 124, 35, 114, 105, 35, 119, 107, 104, 35, 119, 104, 123, 119, 49, 35, 1]
            # fmt: on

            __UpperCamelCase =			tokenizer(A_,text_target=A_					)

            self.assertEqual(A_,batch['input_ids'][0]					)
            self.assertEqual(A_,batch['labels'][0]					)







       def 		snake_case_	(			self: int					):






            '''simple docstring'''




            __UpperCamelCase =			self.get_tokenizers()
            for tokenizer in tokenizers:
                 with self.subTest(F'''{tokenizer.__class__.__name__}'''					):
                      self.assertNotEqual(tokenizer.model_max_length,42					)

        # Now let's start the test
            __UpperCamelCase =			self.get_tokenizers()
            for tokenizer in tokenizers:
                 with self.subTest(F'''{tokenizer.__class__.__name__}'''					):
                      # Isolate this from the other tests because we save additional tokens/etc
                      __UpperCamelCase =			tempfile.mkdtemp()

                      __UpperCamelCase =			' He is very happy, UNwant\u00E9d,running'
                      __UpperCamelCase =			tokenizer.encode(A_,add_special_tokens=A_					)
                      tokenizer.save_pretrained(A_					)

                      __UpperCamelCase =			tokenizer.__class__.from_pretrained(A_					)
                      __UpperCamelCase =			after_tokenizer.encode(A_,add_special_tokens=A_					)
                      self.assertListEqual(A_,A_					)

                      shutil.rmtree(A_					)

            __UpperCamelCase =			self.get_tokenizers(model_max_length=42					)
            for tokenizer in tokenizers:
                 with self.subTest(F'''{tokenizer.__class__.__name__}'''					):
                      # Isolate this from the other tests because we save additional tokens/etc
                      __UpperCamelCase =			tempfile.mkdtemp()

                      __UpperCamelCase =			' He is very happy, UNwant\u00E9d,running'
                      tokenizer.add_tokens(['bim', 'bambam']					)
                      __UpperCamelCase =			tokenizer.additional_special_tokens
                      additional_special_tokens.append('new_additional_special_token'					)
                      tokenizer.add_special_tokens({'additional_special_tokens': additional_special_tokens}					)
                      __UpperCamelCase =			tokenizer.encode(A_,add_special_tokens=A_					)
                      tokenizer.save_pretrained(A_					)

                      __UpperCamelCase =			tokenizer.__class__.from_pretrained(A_					)
                      __UpperCamelCase =			after_tokenizer.encode(A_,add_special_tokens=A_					)
                      self.assertListEqual(A_,A_					)
                      self.assertIn('new_additional_special_token',after_tokenizer.additional_special_tokens					)
                      self.assertEqual(after_tokenizer.model_max_length,42					)

                      __UpperCamelCase =			tokenizer.__class__.from_pretrained(A_,model_max_length=43					)
                      self.assertEqual(tokenizer.model_max_length,43					)

                      shutil.rmtree(A_					)







       def 		snake_case_	(			self: Tuple					):






            '''simple docstring'''




            __UpperCamelCase =			[]
            if self.test_slow_tokenizer:
                 tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())					)

            if self.test_rust_tokenizer:
                 tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())					)

            for tokenizer_class, tokenizer_utils in tokenizer_list:
                 with tempfile.TemporaryDirectory() as tmp_dir:
                      tokenizer_utils.save_pretrained(A_					)

                      with open(os.path.join(A_,'special_tokens_map.json'					),encoding='utf-8'					) as json_file:
                           __UpperCamelCase =			json.load(A_					)

                      with open(os.path.join(A_,'tokenizer_config.json'					),encoding='utf-8'					) as json_file:
                           __UpperCamelCase =			json.load(A_					)

                      __UpperCamelCase =			[F'''<extra_id_{i}>''' for i in range(125					)]

                      __UpperCamelCase =			added_tokens_extra_ids + [
                          'an_additional_special_token'
                      ]
                      __UpperCamelCase =			added_tokens_extra_ids + [
                          'an_additional_special_token'
                      ]

                      with open(os.path.join(A_,'special_tokens_map.json'					),'w',encoding='utf-8'					) as outfile:
                           json.dump(A_,A_					)
                      with open(os.path.join(A_,'tokenizer_config.json'					),'w',encoding='utf-8'					) as outfile:
                           json.dump(A_,A_					)

                      # the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
                      # into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
                      # "special_tokens_map.json" files
                      __UpperCamelCase =			tokenizer_class.from_pretrained(
                          A_,)
                      self.assertIn(
                          'an_additional_special_token',tokenizer_without_change_in_init.additional_special_tokens					)
                      # self.assertIn("an_additional_special_token",tokenizer_without_change_in_init.get_vocab()) # ByT5Tokenization no vocab
                      self.assertEqual(
                          ['an_additional_special_token'],tokenizer_without_change_in_init.convert_ids_to_tokens(
                              tokenizer_without_change_in_init.convert_tokens_to_ids(['an_additional_special_token']					)					),)

                      # Now we test that we can change the value of additional_special_tokens in the from_pretrained
                      __UpperCamelCase =			added_tokens_extra_ids + [AddedToken('a_new_additional_special_token',lstrip=A_					)]
                      __UpperCamelCase =			tokenizer_class.from_pretrained(
                          A_,additional_special_tokens=A_,)

                      self.assertIn('a_new_additional_special_token',tokenizer.additional_special_tokens					)
                      self.assertEqual(
                          ['a_new_additional_special_token'],tokenizer.convert_ids_to_tokens(
                              tokenizer.convert_tokens_to_ids(['a_new_additional_special_token']					)					),)







       def 		snake_case_	(			self: str					):






            '''simple docstring'''




            __UpperCamelCase =			[]
            if self.test_slow_tokenizer:
                 tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())					)

            if self.test_rust_tokenizer:
                 tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())					)

            for tokenizer_class, tokenizer_utils in tokenizer_list:
                 with tempfile.TemporaryDirectory() as tmp_dir:
                      tokenizer_utils.save_pretrained(A_					)

                      __UpperCamelCase =			tokenizer_class.from_pretrained(A_					)

                      self.assertTrue(tokenizer.decode([255]					) == ''					)







       def 		snake_case_	(			self: Union[str, Any]					):






            '''simple docstring'''




            pass







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            pass







       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




            pass







       def 		snake_case_	(			self: Tuple					):






            '''simple docstring'''




            pass







       def 		snake_case_	(			self: Optional[Any]					):






            '''simple docstring'''




            __UpperCamelCase =			self.get_tokenizers(fast=A_,do_lower_case=A_					)
            for tokenizer in tokenizers:
                 with self.subTest(F'''{tokenizer.__class__.__name__}'''					):
                      __UpperCamelCase =			['t', 'h', 'i', 's', ' ', 'i', 's', ' ', 'a', ' ', 't', 'e', 'x', 't', '</s>']
                      __UpperCamelCase =			tokenizer.convert_tokens_to_string(A_					)

                      self.assertIsInstance(A_,A_					)







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			self.get_tokenizers()
            for tokenizer in tokenizers:
                 with self.subTest(F'''{tokenizer.__class__.__name__}'''					):
                      __UpperCamelCase =			[
                          'bos_token',
                          'eos_token',
                          'unk_token',
                          'sep_token',
                          'pad_token',
                          'cls_token',
                          'mask_token',
                      ]

                      __UpperCamelCase =			0
                      __UpperCamelCase =			tokenizer.convert_ids_to_tokens(
                          A_,skip_special_tokens=A_					)

                      for attr in attributes_list:
                           setattr(A_,attr + '_id',A_					)
                           self.assertEqual(getattr(A_,A_					),A_					)
                           self.assertEqual(getattr(A_,attr + '_id'					),A_					)

                           setattr(A_,attr + '_id',A_					)
                           self.assertEqual(getattr(A_,A_					),A_					)
                           self.assertEqual(getattr(A_,attr + '_id'					),A_					)

                      setattr(A_,'additional_special_tokens_ids',[]					)
                      self.assertListEqual(getattr(A_,'additional_special_tokens'					),[]					)
                      self.assertListEqual(getattr(A_,'additional_special_tokens_ids'					),[]					)

                      setattr(A_,'additional_special_tokens_ids',[token_id_to_test_setters]					)
                      self.assertListEqual(getattr(A_,'additional_special_tokens'					),[token_to_test_setters]					)
                      self.assertListEqual(getattr(A_,'additional_special_tokens_ids'					),[token_id_to_test_setters]					)



'''simple docstring'''

import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment







def __UpperCamelCase( _A       :					Any	,      _A       :					List[str]=()	,      _A       :					List[str]=None	,      _A       :					Dict="no"	,      _A       :					List[str]="29500"     ):
   '''simple docstring'''
   UpperCAmelCase__				:					int 					=	False
   UpperCAmelCase__				:					List[str] 					=	False
   if any(key.startswith('''KAGGLE'''     ) for key in os.environ.keys()     ):
      UpperCAmelCase__				:					List[str] 					=	True
   elif "IPython" in sys.modules:
      UpperCAmelCase__				:					List[Any] 					=	'''google.colab''' in str(sys.modules['''IPython'''].get_ipython()     )

   try:
      UpperCAmelCase__				:					Tuple 					=	PrecisionType(mixed_precision.lower()     )
   except ValueError:
      raise ValueError(
          F'''Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.'''     )

   if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''	,      _A     ) is not None):
      # TPU launch
      import torch_xla.distributed.xla_multiprocessing as xmp

      if len(AcceleratorState._shared_state     ) > 0:
         raise ValueError(
             '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
             '''your training function. Restart your notebook and make sure no cells initializes an '''
             '''`Accelerator`.'''     )
      if num_processes is None:
         UpperCAmelCase__				:					Tuple 					=	8

      UpperCAmelCase__				:					Optional[Any] 					=	PrepareForLaunch(_A	,      distributed_type='''TPU'''     )
      print(F'''Launching a training on {num_processes} TPU cores.'''     )
      xmp.spawn(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )
   elif in_colab:
      # No need for a distributed launch otherwise as it's either CPU or one GPU.
      if torch.cuda.is_available():
         print('''Launching training on one GPU.'''     )
      else:
         print('''Launching training on one CPU.'''     )
      function(*_A     )
   else:
      if num_processes is None:
         raise ValueError(
             '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.'''     )

      if num_processes > 1:
         # Multi-GPU launch
         from torch.multiprocessing import start_processes
         from torch.multiprocessing.spawn import ProcessRaisedException

         if len(AcceleratorState._shared_state     ) > 0:
            raise ValueError(
                '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
                '''inside your training function. Restart your notebook and make sure no cells initializes an '''
                '''`Accelerator`.'''     )

         if torch.cuda.is_initialized():
            raise ValueError(
                '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
                '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
                '''function.'''     )

         # torch.distributed will expect a few environment variable to be here. We set the ones common to each
         # process here (the other ones will be set be the launcher).
         with patch_environment(
             world_size=_A	,      master_addr='''127.0.01'''	,      master_port=_A	,      mixed_precision=_A     ):
            UpperCAmelCase__				:					str 					=	PrepareForLaunch(_A	,      distributed_type='''MULTI_GPU'''     )
            print(F'''Launching training on {num_processes} GPUs.'''     )
            try:
               start_processes(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )
            except ProcessRaisedException as e:
               if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
                  raise RuntimeError(
                      '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
                      '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
                      '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
                      '''which one is problematic.'''     ) from e

      else:
         # No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
         if is_mps_available():
            UpperCAmelCase__				:					Union[str, Any] 					=	'''1'''
            print('''Launching training on MPS.'''     )
         elif torch.cuda.is_available():
            print('''Launching training on one GPU.'''     )
         else:
            print('''Launching training on CPU.'''     )
         function(*_A     )







def __UpperCamelCase( _A       :					List[str]	,      _A       :					Optional[Any]=()	,      _A       :					str=2     ):
   '''simple docstring'''
   from torch.multiprocessing import start_processes

   with tempfile.NamedTemporaryFile() as tmp_file:
      # torch.distributed will expect a few environment variable to be here. We set the ones common to each
      # process here (the other ones will be set be the launcher).
      with patch_environment(
          world_size=_A	,      master_addr='''127.0.01'''	,      master_port='''29500'''	,      accelerate_mixed_precision='''no'''	,      accelerate_debug_rdv_file=tmp_file.name	,      accelerate_use_cpu='''yes'''	,      ):
         UpperCAmelCase__				:					Optional[int] 					=	PrepareForLaunch(_A	,      debug=_A     )
         start_processes(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )




