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


SCREAMING_SNAKE_CASE    =					logging.get_logger(__name__)


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

# See all MVP models at https://huggingface.co/models?filter=mvp
SCREAMING_SNAKE_CASE    =					{
    '''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''',
    },
}

SCREAMING_SNAKE_CASE    =					{
    '''RUCAIBox/mvp''': 1_0_2_4,
}







@lru_cache()
def snake_case_    (				):
  UpperCAmelCase__			:				List[str]												=       (
      list(range(ord("!"		)      ,					ord("~"		) + 1		)		) + list(range(ord("¡"		)      ,					ord("¬"		) + 1		)		) + list(range(ord("®"		)      ,					ord("ÿ"		) + 1		)		)
  )
  UpperCAmelCase__			:				List[Any]												=       bs[:]
  UpperCAmelCase__			:				Tuple												=       0
  for b in range(2**8		):
    if b not in bs:
      bs.append(lowercase__		)
      cs.append(2**8 + n		)
      n += 1
  UpperCAmelCase__			:				Dict												=       [chr(lowercase__		) for n in cs]
  return dict(zip(lowercase__      ,					lowercase__		)		)







def snake_case_    (				lowercase__		):
  UpperCAmelCase__			:				int												=       set()
  UpperCAmelCase__			:				Optional[int]												=       word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)		)
    UpperCAmelCase__			:				int												=       char
  return pairs






class        UpperCAmelCase_       (						lowercase__							):

 '''simple docstring'''
 lowercase_				:						str				=       VOCAB_FILES_NAMES
 lowercase_				:						str				=       PRETRAINED_VOCAB_FILES_MAP
 lowercase_				:						List[Any]				=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 lowercase_				:						int				=       ["""input_ids""", """attention_mask"""]







 def __init__(				self		:     Optional[int] ,							snake_case__		:     Optional[int] ,							snake_case__		:     Union[str, Any] ,							snake_case__		:     Union[str, Any]="replace" ,							snake_case__		:     Tuple="<s>" ,							snake_case__		:     str="</s>" ,							snake_case__		:     List[Any]="</s>" ,							snake_case__		:     List[str]="<s>" ,							snake_case__		:     Dict="<unk>" ,							snake_case__		:     List[Any]="<pad>" ,							snake_case__		:     List[str]="<mask>" ,							snake_case__		:     Optional[Any]=False ,							**snake_case__		:     Union[str, Any] ,							):



   '''simple docstring'''




   UpperCAmelCase__			:				List[Any]												=       AddedToken(_A ,							lstrip=_A ,							rstrip=_A   ) if isinstance(_A ,							_A   ) else bos_token
   UpperCAmelCase__			:				Tuple												=       AddedToken(_A ,							lstrip=_A ,							rstrip=_A   ) if isinstance(_A ,							_A   ) else eos_token
   UpperCAmelCase__			:				List[Any]												=       AddedToken(_A ,							lstrip=_A ,							rstrip=_A   ) if isinstance(_A ,							_A   ) else sep_token
   UpperCAmelCase__			:				Optional[Any]												=       AddedToken(_A ,							lstrip=_A ,							rstrip=_A   ) if isinstance(_A ,							_A   ) else cls_token
   UpperCAmelCase__			:				Optional[int]												=       AddedToken(_A ,							lstrip=_A ,							rstrip=_A   ) if isinstance(_A ,							_A   ) else unk_token
   UpperCAmelCase__			:				str												=       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
   UpperCAmelCase__			:				List[str]												=       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:
     UpperCAmelCase__			:				int												=       json.load(_A   )
   UpperCAmelCase__			:				str												=       {v: k for k, v in self.encoder.items()}
   UpperCAmelCase__			:				Optional[Any]												=       errors  # how to handle errors in decoding
   UpperCAmelCase__			:				int												=       bytes_to_unicode()
   UpperCAmelCase__			:				Union[str, Any]												=       {v: k for k, v in self.byte_encoder.items()}
   with open(_A ,							encoding="utf-8"   ) as merges_handle:
     UpperCAmelCase__			:				Dict												=       merges_handle.read().split("\n"   )[1:-1]
   UpperCAmelCase__			:				Any												=       [tuple(merge.split()   ) for merge in bpe_merges]
   UpperCAmelCase__			:				List[Any]												=       dict(zip(_A ,							range(len(_A   )   )   )   )
   UpperCAmelCase__			:				Optional[int]												=       {}
   UpperCAmelCase__			:				Optional[int]												=       add_prefix_space

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







 @property
 def        UpperCamelCase     (				self		:     Any   ):



   '''simple docstring'''




   return len(self.encoder   )







 def        UpperCamelCase     (				self		:     List[Any]   ):



   '''simple docstring'''




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







 def        UpperCamelCase     (				self		:     str ,							snake_case__		:     Optional[Any]   ):



   '''simple docstring'''




   if token in self.cache:
     return self.cache[token]
   UpperCAmelCase__			:				Any												=       tuple(_A   )
   UpperCAmelCase__			:				List[str]												=       get_pairs(_A   )

   if not pairs:
     return token

   while True:
     UpperCAmelCase__			:				List[str]												=       min(_A ,							key=lambda snake_case__   : self.bpe_ranks.get(_A ,							float("inf"   )   )   )
     if bigram not in self.bpe_ranks:
       break
     UpperCAmelCase__			:				List[str]												=       bigram
     UpperCAmelCase__			:				List[Any]												=       []
     UpperCAmelCase__			:				str												=       0
     while i < len(_A   ):
       try:
         UpperCAmelCase__			:				Optional[int]												=       word.index(_A ,							_A   )
       except ValueError:
         new_word.extend(word[i:]   )
         break
       else:
         new_word.extend(word[i:j]   )
         UpperCAmelCase__			:				str												=       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
     UpperCAmelCase__			:				Union[str, Any]												=       tuple(_A   )
     UpperCAmelCase__			:				List[Any]												=       new_word
     if len(_A   ) == 1:
       break
     else:
       UpperCAmelCase__			:				Tuple												=       get_pairs(_A   )
   UpperCAmelCase__			:				Union[str, Any]												=       ' '.join(_A   )
   UpperCAmelCase__			:				str												=       word
   return word







 def        UpperCamelCase     (				self		:     Dict ,							snake_case__		:     List[Any]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				int												=       []
   for token in re.findall(self.pat ,							_A   ):
     UpperCAmelCase__			:				Union[str, Any]												=       ''.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        UpperCamelCase     (				self		:     Optional[Any] ,							snake_case__		:     Any   ):



   '''simple docstring'''




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







 def        UpperCamelCase     (				self		:     Optional[int] ,							snake_case__		:     List[str]   ):



   '''simple docstring'''




   return self.decoder.get(_A   )







 def        UpperCamelCase     (				self		:     Tuple ,							snake_case__		:     Any   ):



   '''simple docstring'''




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







 def        UpperCamelCase     (				self		:     Optional[int] ,							snake_case__		:     str ,							snake_case__		:     Optional[str] = None   ):



   '''simple docstring'''




   if not os.path.isdir(_A   ):
     logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""   )
     return
   UpperCAmelCase__			:				Optional[int]												=       os.path.join(
       _A ,							(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]   )
   UpperCAmelCase__			:				Dict												=       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"   )

   UpperCAmelCase__			:				Dict												=       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 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!"   )
         UpperCAmelCase__			:				Any												=       token_index
       writer.write(" ".join(_A   ) + "\n"   )
       index += 1

   return vocab_file, merge_file







 def        UpperCamelCase     (				self		:     Union[str, Any] ,							snake_case__		:     List[int] ,							snake_case__		:     Optional[List[int]] = None   ):



   '''simple docstring'''




   if token_ids_a is None:
     return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
   UpperCAmelCase__			:				Union[str, Any]												=       [self.cls_token_id]
   UpperCAmelCase__			:				Tuple												=       [self.sep_token_id]
   return cls + token_ids_a + sep + sep + token_ids_a + sep







 def        UpperCamelCase     (				self		:     Dict ,							snake_case__		:     List[int] ,							snake_case__		:     Optional[List[int]] = None ,							snake_case__		:     bool = False   ):



   '''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        UpperCamelCase     (				self		:     str ,							snake_case__		:     List[int] ,							snake_case__		:     Optional[List[int]] = None   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Optional[Any]												=       [self.sep_token_id]
   UpperCAmelCase__			:				int												=       [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        UpperCamelCase     (				self		:     str ,							snake_case__		:     List[Any] ,							snake_case__		:     str=False ,							**snake_case__		:     List[Any]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				List[Any]												=       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()):
     UpperCAmelCase__			:				Tuple												=       ' ' + text
   return (text, kwargs)





'''simple docstring'''





import random
import shutil
import tempfile
import unittest

import numpy as np

from transformers import TaTokenizer, TaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torch
from transformers.utils.import_utils import is_speech_available, is_torch_available


if is_torch_available():
      pass

if is_speech_available():
      from transformers import EncodecFeatureExtractor, MusicgenProcessor


SCREAMING_SNAKE_CASE    =					random.Random()







def snake_case_    (				lowercase__      ,					lowercase__=1.0      ,					lowercase__=None      ,					lowercase__=None		):
  if rng is None:
    UpperCAmelCase__			:				Optional[int]												=       global_rng

  UpperCAmelCase__			:				List[str]												=       []
  for batch_idx in range(shape[0]		):
    values.append([]		)
    for _ in range(shape[1]		):
      values[-1].append(rng.random() * scale		)

  return values






@require_torch
@require_sentencepiece
class        UpperCAmelCase_       (						unittest.TestCase							):

 '''simple docstring'''







 def        UpperCamelCase     (				self		:     Optional[int]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Tuple												=       "facebook/musicgen-small"
   UpperCAmelCase__			:				Union[str, Any]												=       tempfile.mkdtemp()







 def        UpperCamelCase     (				self		:     Union[str, Any] ,							**snake_case__		:     Dict   ):



   '''simple docstring'''




   return TaTokenizer.from_pretrained(self.checkpoint ,							**snake_case__   )







 def        UpperCamelCase     (				self		:     Optional[int] ,							**snake_case__		:     str   ):



   '''simple docstring'''




   return EncodecFeatureExtractor.from_pretrained(self.checkpoint ,							**snake_case__   )







 def        UpperCamelCase     (				self		:     List[str]   ):



   '''simple docstring'''




   shutil.rmtree(self.tmpdirname   )







 def        UpperCamelCase     (				self		:     Any   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Any												=       self.get_tokenizer()
   UpperCAmelCase__			:				Tuple												=       self.get_feature_extractor()

   UpperCAmelCase__			:				int												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   processor.save_pretrained(self.tmpdirname   )
   UpperCAmelCase__			:				Dict												=       MusicgenProcessor.from_pretrained(self.tmpdirname   )

   self.assertEqual(processor.tokenizer.get_vocab() ,							tokenizer.get_vocab()   )
   self.assertIsInstance(processor.tokenizer ,							snake_case__   )

   self.assertEqual(processor.feature_extractor.to_json_string() ,							feature_extractor.to_json_string()   )
   self.assertIsInstance(processor.feature_extractor ,							snake_case__   )







 def        UpperCamelCase     (				self		:     Tuple   ):



   '''simple docstring'''




   UpperCAmelCase__			:				int												=       MusicgenProcessor(tokenizer=self.get_tokenizer() ,							feature_extractor=self.get_feature_extractor()   )
   processor.save_pretrained(self.tmpdirname   )

   UpperCAmelCase__			:				List[Any]												=       self.get_tokenizer(bos_token="(BOS)" ,							eos_token="(EOS)"   )
   UpperCAmelCase__			:				int												=       self.get_feature_extractor(do_normalize=snake_case__ ,							padding_value=1.0   )

   UpperCAmelCase__			:				Optional[Any]												=       MusicgenProcessor.from_pretrained(
       self.tmpdirname ,							bos_token="(BOS)" ,							eos_token="(EOS)" ,							do_normalize=snake_case__ ,							padding_value=1.0   )

   self.assertEqual(processor.tokenizer.get_vocab() ,							tokenizer_add_kwargs.get_vocab()   )
   self.assertIsInstance(processor.tokenizer ,							snake_case__   )

   self.assertEqual(processor.feature_extractor.to_json_string() ,							feature_extractor_add_kwargs.to_json_string()   )
   self.assertIsInstance(processor.feature_extractor ,							snake_case__   )







 def        UpperCamelCase     (				self		:     List[Any]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				int												=       self.get_feature_extractor()
   UpperCAmelCase__			:				int												=       self.get_tokenizer()

   UpperCAmelCase__			:				List[str]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				Optional[Any]												=       floats_list((3, 10_00)   )

   UpperCAmelCase__			:				List[Any]												=       feature_extractor(snake_case__ ,							return_tensors="np"   )
   UpperCAmelCase__			:				List[Any]												=       processor(snake_case__ ,							return_tensors="np"   )

   for key in input_feat_extract.keys():
     self.assertAlmostEqual(input_feat_extract[key].sum() ,							input_processor[key].sum() ,							delta=1e-2   )







 def        UpperCamelCase     (				self		:     List[str]   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Union[str, Any]												=       self.get_feature_extractor()
   UpperCAmelCase__			:				Union[str, Any]												=       self.get_tokenizer()

   UpperCAmelCase__			:				Optional[int]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				List[Any]												=       "This is a test string"

   UpperCAmelCase__			:				Dict												=       processor(text=snake_case__   )

   UpperCAmelCase__			:				Any												=       tokenizer(snake_case__   )

   for key in encoded_tok.keys():
     self.assertListEqual(encoded_tok[key] ,							encoded_processor[key]   )







 def        UpperCamelCase     (				self		:     int   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Union[str, Any]												=       self.get_feature_extractor()
   UpperCAmelCase__			:				str												=       self.get_tokenizer()

   UpperCAmelCase__			:				List[Any]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				Optional[int]												=       [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

   UpperCAmelCase__			:				Any												=       processor.batch_decode(sequences=snake_case__   )
   UpperCAmelCase__			:				Optional[Any]												=       tokenizer.batch_decode(snake_case__   )

   self.assertListEqual(snake_case__ ,							snake_case__   )







 def        UpperCamelCase     (				self		:     Tuple   ):



   '''simple docstring'''




   UpperCAmelCase__			:				Any												=       self.get_feature_extractor()
   UpperCAmelCase__			:				List[Any]												=       self.get_tokenizer()

   UpperCAmelCase__			:				Union[str, Any]												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   self.assertListEqual(
       processor.model_input_names ,							feature_extractor.model_input_names ,							msg="`processor` and `feature_extractor` model input names do not match" ,							)







 def        UpperCamelCase     (				self		:     Dict   ):



   '''simple docstring'''




   UpperCAmelCase__			:				int												=       self.get_feature_extractor(padding_side="left"   )
   UpperCAmelCase__			:				str												=       self.get_tokenizer()

   UpperCAmelCase__			:				int												=       MusicgenProcessor(tokenizer=snake_case__ ,							feature_extractor=snake_case__   )

   UpperCAmelCase__			:				List[Any]												=       [floats_list((1, x)   )[0] for x in range(5 ,							20 ,							5   )]
   UpperCAmelCase__			:				List[Any]												=       processor(snake_case__   ).padding_mask

   UpperCAmelCase__			:				Dict												=       np.asarray(floats_list((3, 20)   )   )[:, None, :]
   UpperCAmelCase__			:				Optional[Any]												=       processor.batch_decode(snake_case__ ,							padding_mask=snake_case__   )

   self.assertIsInstance(snake_case__ ,							snake_case__   )

   for audio in decoded_audios:
     self.assertIsInstance(snake_case__ ,							np.ndarray   )

   self.assertTrue(decoded_audios[0].shape == (1, 10)   )
   self.assertTrue(decoded_audios[1].shape == (1, 15)   )
   self.assertTrue(decoded_audios[2].shape == (1, 20)   )



