
"""simple docstring"""




import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


a		:					List[str]    	=					logging.get_logger(__name__)


a		:					Tuple    	=					{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

# See all MVP models at https://huggingface.co/models?filter=mvp
a		:					Optional[Any]    	=					{
    '''vocab_file''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/vocab.json''',
    },
    '''added_tokens.json''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/added_tokens.json''',
    },
    '''merges_file''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/merges.txt''',
    },
}

a		:					Any    	=					{
    '''RUCAIBox/mvp''': 10_24,
}



@lru_cache()
def 						__magic_name__		(				)    ->	List[str]:
    a__    = (
        list(range(ord('!'    )       ,		ord('~'    ) + 1    )    ) + list(range(ord('¡'    )       ,		ord('¬'    ) + 1    )    ) + list(range(ord('®'    )       ,		ord('ÿ'    ) + 1    )    )
    )
    a__    = bs[:]
    a__    = 0
    for b in range(2**8    ):
        if b not in bs:
            bs.append(UpperCamelCase    )
            cs.append(2**8 + n    )
            n += 1
    a__    = [chr(UpperCamelCase    ) for n in cs]
    return dict(zip(UpperCamelCase       ,		UpperCamelCase    )    )



def 						__magic_name__		(				UpperCamelCase    :		Union[str, Any]    )    ->	Optional[Any]:
    a__    = set()
    a__    = word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)    )
        a__    = char
    return pairs







class 						lowercase(lowercase__					):
     __snake_case:  str  				=    VOCAB_FILES_NAMES
     __snake_case:  str  				=    PRETRAINED_VOCAB_FILES_MAP
     __snake_case:  List[Any]  				=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     __snake_case:  int  				=    ["""input_ids""", """attention_mask"""]



     def __init__(       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE="replace"    ,							__SCREAMING_SNAKE_CASE="<s>"    ,							__SCREAMING_SNAKE_CASE="</s>"    ,							__SCREAMING_SNAKE_CASE="</s>"    ,							__SCREAMING_SNAKE_CASE="<s>"    ,							__SCREAMING_SNAKE_CASE="<unk>"    ,							__SCREAMING_SNAKE_CASE="<pad>"    ,							__SCREAMING_SNAKE_CASE="<mask>"    ,							__SCREAMING_SNAKE_CASE=False    ,							**__SCREAMING_SNAKE_CASE    ,							)		->    Dict:







         """simple docstring"""



         a__    = AddedToken(_A    ,							lstrip=_A    ,							rstrip=_A      ) if isinstance(_A    ,							_A      ) else bos_token
         a__    = AddedToken(_A    ,							lstrip=_A    ,							rstrip=_A      ) if isinstance(_A    ,							_A      ) else eos_token
         a__    = AddedToken(_A    ,							lstrip=_A    ,							rstrip=_A      ) if isinstance(_A    ,							_A      ) else sep_token
         a__    = AddedToken(_A    ,							lstrip=_A    ,							rstrip=_A      ) if isinstance(_A    ,							_A      ) else cls_token
         a__    = AddedToken(_A    ,							lstrip=_A    ,							rstrip=_A      ) if isinstance(_A    ,							_A      ) else unk_token
         a__    = AddedToken(_A    ,							lstrip=_A    ,							rstrip=_A      ) if isinstance(_A    ,							_A      ) else pad_token

         # Mask token behave like a normal word, i.e. include the space before it
         a__    = AddedToken(_A    ,							lstrip=_A    ,							rstrip=_A      ) if isinstance(_A    ,							_A      ) else mask_token

         super().__init__(
             errors=_A    ,							bos_token=_A    ,							eos_token=_A    ,							unk_token=_A    ,							sep_token=_A    ,							cls_token=_A    ,							pad_token=_A    ,							mask_token=_A    ,							add_prefix_space=_A    ,							**_A    ,							)
         with open(_A    ,							encoding='utf-8'      ) as vocab_handle:
             a__    = json.load(_A      )
         a__    = {v: k for k, v in self.encoder.items()}
         a__    = errors  # how to handle errors in decoding
         a__    = bytes_to_unicode()
         a__    = {v: k for k, v in self.byte_encoder.items()}
         with open(_A    ,							encoding='utf-8'      ) as merges_handle:
             a__    = merges_handle.read().split('\n'      )[1:-1]
         a__    = [tuple(merge.split()      ) for merge in bpe_merges]
         a__    = dict(zip(_A    ,							range(len(_A      )      )      )      )
         a__    = {}
         a__    = add_prefix_space

         # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
         a__    = re.compile(R'\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'      )



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







         """simple docstring"""



         return len(self.encoder      )



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







         """simple docstring"""



         return dict(self.encoder    ,							**self.added_tokens_encoder      )



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE      )		->    Tuple:







         """simple docstring"""



         if token in self.cache:
             return self.cache[token]
         a__    = tuple(_A      )
         a__    = get_pairs(_A      )

         if not pairs:
             return token

         while True:
             a__    = min(_A    ,							key=lambda __SCREAMING_SNAKE_CASE      : self.bpe_ranks.get(_A    ,							float('inf'      )      )      )
             if bigram not in self.bpe_ranks:
                 break
             a__    = bigram
             a__    = []
             a__    = 0
             while i < len(_A      ):
                 try:
                     a__    = word.index(_A    ,							_A      )
                 except ValueError:
                     new_word.extend(word[i:]      )
                     break
                 else:
                     new_word.extend(word[i:j]      )
                     a__    = j

                 if word[i] == first and i < len(_A      ) - 1 and word[i + 1] == second:
                     new_word.append(first + second      )
                     i += 2
                 else:
                     new_word.append(word[i]      )
                     i += 1
             a__    = tuple(_A      )
             a__    = new_word
             if len(_A      ) == 1:
                 break
             else:
                 a__    = get_pairs(_A      )
         a__    = ' '.join(_A      )
         a__    = word
         return word



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE      )		->    int:







         """simple docstring"""



         a__    = []
         for token in re.findall(self.pat    ,							_A      ):
             a__    = ''.join(
                 self.byte_encoder[b] for b in token.encode('utf-8'      )      )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
             bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_A      ).split(' '      )      )
         return bpe_tokens



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE      )		->    Any:







         """simple docstring"""



         return self.encoder.get(_A    ,							self.encoder.get(self.unk_token      )      )



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE      )		->    str:







         """simple docstring"""



         return self.decoder.get(_A      )



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







         """simple docstring"""



         a__    = ''.join(_A      )
         a__    = bytearray([self.byte_decoder[c] for c in text]      ).decode('utf-8'    ,							errors=self.errors      )
         return text



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE = None      )		->    Tuple[str]:







         """simple docstring"""



         if not os.path.isdir(_A      ):
             logger.error(f'Vocabulary path ({save_directory}) should be a directory'      )
             return
         a__    = os.path.join(
             _A    ,							(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']      )
         a__    = os.path.join(
             _A    ,							(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']      )

         with open(_A    ,							'w'    ,							encoding='utf-8'      ) as f:
             f.write(json.dumps(self.encoder    ,							indent=2    ,							sort_keys=_A    ,							ensure_ascii=_A      ) + '\n'      )

         a__    = 0
         with open(_A    ,							'w'    ,							encoding='utf-8'      ) as writer:
             writer.write('#version: 0.2\n'      )
             for bpe_tokens, token_index in sorted(self.bpe_ranks.items()    ,							key=lambda __SCREAMING_SNAKE_CASE      : kv[1]      ):
                 if index != token_index:
                     logger.warning(
                         f'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
                         ' Please check that the tokenizer is not corrupted!'      )
                     a__    = token_index
                 writer.write(' '.join(_A      ) + '\n'      )
                 index += 1

         return vocab_file, merge_file



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE = None      )		->    List[int]:







         """simple docstring"""



         if token_ids_a is None:
             return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
         a__    = [self.cls_token_id]
         a__    = [self.sep_token_id]
         return cls + token_ids_a + sep + sep + token_ids_a + sep



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE = None    ,							__SCREAMING_SNAKE_CASE = False      )		->    List[int]:







         """simple docstring"""



         if already_has_special_tokens:
             return super().get_special_tokens_mask(
                 token_ids_a=_A    ,							token_ids_a=_A    ,							already_has_special_tokens=_A      )

         if token_ids_a is None:
             return [1] + ([0] * len(_A      )) + [1]
         return [1] + ([0] * len(_A      )) + [1, 1] + ([0] * len(_A      )) + [1]



     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE = None      )		->    List[int]:







         """simple docstring"""



         a__    = [self.sep_token_id]
         a__    = [self.cls_token_id]

         if token_ids_a is None:
             return len(cls + token_ids_a + sep      ) * [0]
         return len(cls + token_ids_a + sep + sep + token_ids_a + sep      ) * [0]




     def  lowercase__     (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE=False    ,							**__SCREAMING_SNAKE_CASE      )		->    List[str]:







         """simple docstring"""



         a__    = kwargs.pop('add_prefix_space'    ,							self.add_prefix_space      )
         if (is_split_into_words or add_prefix_space) and (len(_A      ) > 0 and not text[0].isspace()):
             a__    = ' ' + text
         return (text, kwargs)








"""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']      )      )


