
"""simple docstring"""




import concurrent.futures
import json
import os
import shutil
import tempfile
import unittest

from transformers import AutoTokenizer, PreTrainedTokenizerFast
from transformers.testing_utils import require_tokenizers

from ..test_tokenization_common import TokenizerTesterMixin







@require_tokenizers
class 						lowercase(_lowercase						,	unittest.TestCase					):
     __snake_case:  Any  				=    PreTrainedTokenizerFast
     __snake_case:  Optional[int]  				=    False
     __snake_case:  List[Any]  				=    True
     __snake_case:  Dict  				=    'tokenizer_file'



     def  lowercase__     (       self      )		->    Optional[Any]:







         """simple docstring"""



         a__    = False  # because we don't have pretrained_vocab_files_map
         super().setUp()
         a__    = True

         a__    = ['robot-test/dummy-tokenizer-fast', 'robot-test/dummy-tokenizer-wordlevel']
         a__    = 'SaulLu/dummy-tokenizer-bytelevel-bpe'

         # Inclusion of 2 tokenizers to test different types of models (Unigram and WordLevel for the moment)
         a__    = [(PreTrainedTokenizerFast, model_path, {}) for model_path in model_paths]

         a__    = PreTrainedTokenizerFast.from_pretrained(model_paths[0]      )
         tokenizer.save_pretrained(self.tmpdirname      )



     def  lowercase__     (       self      )		->    Any:







         """simple docstring"""



         pass



     def  lowercase__     (       self      )		->    Optional[int]:







         """simple docstring"""



         pass



     def  lowercase__     (       self      )		->    int:







         """simple docstring"""



         pass



     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         pass



     def  lowercase__     (       self      )		->    Any:







         """simple docstring"""



         a__    = self.tmpdirname
         # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
         for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
             with self.subTest(f'{tokenizer.__class__.__name__} ({pretrained_name})'      ):
                 try:
                     a__    = tempfile.mkdtemp()
                     a__    = self.rust_tokenizer_class.from_pretrained(__SCREAMING_SNAKE_CASE    ,							**__SCREAMING_SNAKE_CASE      )

                     tokenizer.save_pretrained(self.tmpdirname      )
                     super().test_training_new_tokenizer()
                 finally:
                     # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                     # is restored
                     shutil.rmtree(self.tmpdirname      )
                     a__    = tmpdirname_orig



     def  lowercase__     (       self      )		->    int:







         """simple docstring"""



         a__    = self.tmpdirname
         # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
         for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
             with self.subTest(f'{tokenizer.__class__.__name__} ({pretrained_name})'      ):
                 try:
                     a__    = tempfile.mkdtemp()
                     a__    = self.rust_tokenizer_class.from_pretrained(__SCREAMING_SNAKE_CASE    ,							**__SCREAMING_SNAKE_CASE      )

                     tokenizer.save_pretrained(self.tmpdirname      )
                     super().test_training_new_tokenizer_with_special_tokens_change()
                 finally:
                     # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                     # is restored
                     shutil.rmtree(self.tmpdirname      )
                     a__    = tmpdirname_orig




     def  lowercase__     (       self      )		->    List[Any]:







         """simple docstring"""



         a__    = self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name      )

         a__    = ('a' for _ in range(1_0_0_0      ))
         a__    = tokenizer.train_new_from_iterator(text_iterator=__SCREAMING_SNAKE_CASE    ,							length=1_0_0_0    ,							vocab_size=5_0      )

         a__    = new_tokenizer.encode('a🤗'      )
         self.assertEqual(__SCREAMING_SNAKE_CASE    ,							[6_4, 1_7_2, 2_5_3, 9_7, 2_4_5]      )







@require_tokenizers
class 						lowercase(unittest.TestCase					):



     def  lowercase__     (       self      )		->    int:







         """simple docstring"""



         a__    = AutoTokenizer.from_pretrained('bert-base-cased'      )
         a__    = json.loads(tokenizer._tokenizer.to_str()      )
         a__    = len(__SCREAMING_SNAKE_CASE      )

         with tempfile.TemporaryDirectory() as tmp_dir:
             # Hack to save this in the tokenizer_config.json
             a__    = ['tokenizer.4.0.0.json']
             tokenizer.save_pretrained(__SCREAMING_SNAKE_CASE      )
             json.dump(__SCREAMING_SNAKE_CASE    ,							open(os.path.join(__SCREAMING_SNAKE_CASE    ,							'tokenizer.4.0.0.json'      )    ,							'w'      )      )

             # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
             a__    = AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE      )
             self.assertEqual(len(__SCREAMING_SNAKE_CASE      )    ,							len(__SCREAMING_SNAKE_CASE      ) + 1      )
             a__    = json.loads(new_tokenizer._tokenizer.to_str()      )
             self.assertIn('huggingface'    ,							json_tokenizer['model']['vocab']      )

             # Will need to be adjusted if we reach v42 and this test is still here.
             # Should pick the old tokenizer file as the version of Transformers is < 4.0.0
             shutil.move(os.path.join(__SCREAMING_SNAKE_CASE    ,							'tokenizer.4.0.0.json'      )    ,							os.path.join(__SCREAMING_SNAKE_CASE    ,							'tokenizer.42.0.0.json'      )      )
             a__    = ['tokenizer.42.0.0.json']
             tokenizer.save_pretrained(__SCREAMING_SNAKE_CASE      )
             a__    = AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE      )
             self.assertEqual(len(__SCREAMING_SNAKE_CASE      )    ,							len(__SCREAMING_SNAKE_CASE      )      )
             a__    = json.loads(new_tokenizer._tokenizer.to_str()      )
             self.assertNotIn('huggingface'    ,							json_tokenizer['model']['vocab']      )




     def  lowercase__     (       self      )		->    Tuple:







         """simple docstring"""



         a__    = 'hf-internal-testing/test-two-tokenizers'

         # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
         a__    = AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE      )
         self.assertEqual(len(__SCREAMING_SNAKE_CASE      )    ,							2_8_9_9_7      )
         a__    = json.loads(tokenizer._tokenizer.to_str()      )
         self.assertIn('huggingface'    ,							json_tokenizer['model']['vocab']      )

         # Testing an older version by monkey-patching the version in the module it's used.
         import transformers as old_transformers

         a__    = '3.0.0'
         a__    = old_transformers.models.auto.AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE      )
         self.assertEqual(len(__SCREAMING_SNAKE_CASE      )    ,							2_8_9_9_6      )
         a__    = json.loads(old_tokenizer._tokenizer.to_str()      )
         self.assertNotIn('huggingface'    ,							json_tokenizer['model']['vocab']      )







@require_tokenizers
class 						lowercase(unittest.TestCase					):



     def  lowercase__     (       self      )		->    Any:







         """simple docstring"""



         a__    = PreTrainedTokenizerFast.from_pretrained('robot-test/dummy-tokenizer-wordlevel'      )
         a__    = 'The Matrix is a 1999 science fiction action film.'

         with concurrent.futures.ThreadPoolExecutor() as executor:
             a__    = [executor.submit(self.fetch    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      ) for i in range(1_0      )]
             a__    = [future.result() for future in futures]
             self.assertEqual(__SCREAMING_SNAKE_CASE    ,							[[1, 1_0, 0, 8, 0, 1_8, 0, 0, 0, 2] for i in range(1_0      )]      )




     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )		->    Optional[Any]:







         """simple docstring"""



         return tokenizer.encode(__SCREAMING_SNAKE_CASE    ,							truncation='longest_first'    ,							padding='longest'      )




"""simple docstring"""




import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


a		:					Tuple    	=					namedtuple('ExampleNamedTuple', 'a b c')







class 						lowercase(unittest.TestCase					):



     def  lowercase__     (       self      )		->    Union[str, Any]:







         """simple docstring"""



         a__    = torch.randn(5    ,							2      )
         a__    = torch.device('cuda'      ) if torch.cuda.is_available() else torch.device('cpu'      )

         a__    = send_to_device(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta.cpu()    ,							__SCREAMING_SNAKE_CASE      )      )

         a__    = send_to_device((tensor, [tensor, tensor], 1)    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta[0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta[1]    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta[1][0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta[1][1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta[2]    ,							1      )

         a__    = send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['a'].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta['b']    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['b'][0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta['b'][1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta['c']    ,							1      )

         a__    = send_to_device(ExampleNamedTuple(a=__SCREAMING_SNAKE_CASE    ,							b=[tensor, tensor]    ,							c=1      )    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta.a.cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta.b    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta.b[0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta.b[1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta.c    ,							1      )

         a__    = send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}      )    ,							__SCREAMING_SNAKE_CASE      )
         self.assertIsInstance(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['a'].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertIsInstance(resulta['b']    ,							__SCREAMING_SNAKE_CASE      )
         self.assertTrue(torch.equal(resulta['b'][0].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertTrue(torch.equal(resulta['b'][1].cpu()    ,							__SCREAMING_SNAKE_CASE      )      )
         self.assertEqual(resulta['c']    ,							1      )



     def  lowercase__     (       self      )		->    Dict:







         """simple docstring"""



         with self.assertRaises(__SCREAMING_SNAKE_CASE      ) as cm:
             a__    = recursively_apply(torch.tensor    ,							(torch.tensor(1      ), 1)    ,							error_on_other_type=__SCREAMING_SNAKE_CASE      )
         self.assertEqual(
             str(cm.exception      )    ,							'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'    ,							)



     def  lowercase__     (       self      )		->    Any:







         """simple docstring"""



         a__    = torch.tensor([1, 2, 3, 4, 5]      )
         self.assertEqual(listify(__SCREAMING_SNAKE_CASE      )    ,							[1, 2, 3, 4, 5]      )

         a__    = torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )
         self.assertEqual(listify(__SCREAMING_SNAKE_CASE      )    ,							[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )

         a__    = torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )
         self.assertEqual(
             listify(__SCREAMING_SNAKE_CASE      )    ,							[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )



     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         with patch_environment(aa=1    ,							BB=2      ):
             self.assertEqual(os.environ.get('AA'      )    ,							'1'      )
             self.assertEqual(os.environ.get('BB'      )    ,							'2'      )

         self.assertNotIn('AA'    ,							os.environ      )
         self.assertNotIn('BB'    ,							os.environ      )



     def  lowercase__     (       self      )		->    Tuple:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE    ,							keep_fpaa_wrapper=__SCREAMING_SNAKE_CASE      )
         a__    = pickle.dumps(__SCREAMING_SNAKE_CASE      )



     @require_cuda
     def  lowercase__     (       self      )		->    Union[str, Any]:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = torch.cuda.amp.autocast(dtype=torch.floataa      )(model.forward      )
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE    ,							keep_fpaa_wrapper=__SCREAMING_SNAKE_CASE      )
         a__    = pickle.dumps(__SCREAMING_SNAKE_CASE      )



     @require_cuda
     @require_torch_min_version(version='2.0'      )
     def  lowercase__     (       self      )		->    Optional[int]:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = model.forward
         a__    = torch.cuda.amp.autocast(dtype=torch.floataa      )(model.forward      )
         a__    = convert_outputs_to_fpaa(model.forward      )
         a__    = torch.compile(model.forward    ,							backend='inductor'      )
         a__    = torch.randn(4    ,							1_0      ).cuda()
         a__    = model(__SCREAMING_SNAKE_CASE      )



     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         a__    = RegressionModel()
         # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
         a__    = torch.nn.parallel.DataParallel(__SCREAMING_SNAKE_CASE      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE      )

         self.assertEqual(__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE      )



     @require_torch_min_version(version='2.0'      )
     def  lowercase__     (       self      )		->    str:







         """simple docstring"""



         a__    = RegressionModel()
         a__    = torch.compile(__SCREAMING_SNAKE_CASE      )

         # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
         a__    = torch.nn.parallel.DataParallel(__SCREAMING_SNAKE_CASE      )
         a__    = torch.compile(__SCREAMING_SNAKE_CASE      )
         a__    = extract_model_from_parallel(__SCREAMING_SNAKE_CASE      )

         self.assertEqual(compiled_model._orig_mod    ,							compiled_model_unwrapped._orig_mod      )




     def  lowercase__     (       self      )		->    Optional[Any]:







         """simple docstring"""



         self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3]      )]      )    ,							torch.device('cpu'      )      )
         self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3]      )}      )    ,							torch.device('cpu'      )      )
         self.assertIsNone(find_device([1, 'a']      )      )


