



'''simple docstring'''





import os
import re
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowercase_									=    logging.get_logger(__name__)

lowercase_									=    {
    "vocab_file": "vocab.txt",
    "merges_file": "bpe.codes",
}

lowercase_									=    {
    "vocab_file": {
        "vinai/phobert-base": "https://huggingface.co/vinai/phobert-base/resolve/main/vocab.txt",
        "vinai/phobert-large": "https://huggingface.co/vinai/phobert-large/resolve/main/vocab.txt",
    },
    "merges_file": {
        "vinai/phobert-base": "https://huggingface.co/vinai/phobert-base/resolve/main/bpe.codes",
        "vinai/phobert-large": "https://huggingface.co/vinai/phobert-large/resolve/main/bpe.codes",
    },
}

lowercase_									=    {
    "vinai/phobert-base": 256,
    "vinai/phobert-large": 256,
}




def lowerCAmelCase    (__A):



     """simple docstring"""



     _a           =    set()
     _a           =    word[0]
     for char in word[1:]:
          pairs.add((prev_char, char))
          _a           =    char

     _a           =    set(__A)
     return pairs






class __A     (    A		):



 '''simple docstring'''







 __lowerCamelCase       :			Optional[Any]		=			VOCAB_FILES_NAMES
 __lowerCamelCase       :			Dict		=			PRETRAINED_VOCAB_FILES_MAP
 __lowerCamelCase       :			List[str]		=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES



 def __init__(self					,  A					,  A					,  A="<s>"					,  A="</s>"					,  A="</s>"					,  A="<s>"					,  A="<unk>"					,  A="<pad>"					,  A="<mask>"					,  **A					,  )       ->	Optional[Any]:


      """simple docstring"""



      super().__init__(
          bos_token=A					,  eos_token=A					,  unk_token=A					,  sep_token=A					,  cls_token=A					,  pad_token=A					,  mask_token=A					,  **A					,  )

      _a           =    vocab_file
      _a           =    merges_file

      _a           =    {}
      _a           =    0
      _a           =    1
      _a           =    2
      _a           =    3

      self.add_from_file(A			)

      _a           =    {v: k for k, v in self.encoder.items()}

      with open(A					,  encoding='''utf-8'''			) as merges_handle:
           _a           =    merges_handle.read().split('''\n'''			)[:-1]
      _a           =    [tuple(merge.split()[:-1]			) for merge in merges]
      _a           =    dict(zip(A					,  range(len(A			)			)			)			)
      _a           =    {}



 def    a__    (self					,  A					,  A = 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    a__    (self					,  A					,  A = None					,  A = 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    a__    (self					,  A					,  A = 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]



 @property
 def    a__    (self			)       ->	Union[str, Any]:


      """simple docstring"""



      return len(self.encoder			)



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


      """simple docstring"""



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



 def    a__    (self					,  A			)       ->	Tuple:


      """simple docstring"""



      if token in self.cache:
           return self.cache[token]
      _a           =    tuple(A			)
      _a           =    tuple(list(word[:-1]			) + [word[-1] + '''</w>''']			)
      _a           =    get_pairs(A			)

      if not pairs:
           return token

      while True:
           _a           =    min(A					,  key=lambda A			: self.bpe_ranks.get(A					,  float('''inf'''			)			)			)
           if bigram not in self.bpe_ranks:
                break
           _a    ,       _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[:-4]
      _a           =    word
      return word



 def    a__    (self					,  A			)       ->	str:


      """simple docstring"""



      _a           =    []

      _a           =    re.findall(R'''\S+\n?'''					,  A			)

      for token in words:
           split_tokens.extend(list(self.bpe(A			).split(''' '''			)			)			)
      return split_tokens



 def    a__    (self					,  A			)       ->	Dict:


      """simple docstring"""



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



 def    a__    (self					,  A			)       ->	Any:


      """simple docstring"""



      return self.decoder.get(A					,  self.unk_token			)



 def    a__    (self					,  A			)       ->	int:


      """simple docstring"""



      _a           =    ''' '''.join(A			).replace('''@@ '''					,  ''''''			).strip()
      return out_string



 def    a__    (self					,  A					,  A = 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''']			)

      if os.path.abspath(self.vocab_file			) != os.path.abspath(A			):
           copyfile(self.vocab_file					,  A			)

      if os.path.abspath(self.merges_file			) != os.path.abspath(A			):
           copyfile(self.merges_file					,  A			)

      return out_vocab_file, out_merge_file



 def    a__    (self					,  A			)       ->	List[Any]:


      """simple docstring"""



      if isinstance(A					,  A			):
           try:
                with open(A					,  '''r'''					,  encoding='''utf-8'''			) as fd:
                     self.add_from_file(A			)
           except FileNotFoundError as fnfe:
                raise fnfe
           except UnicodeError:
                raise Exception(f'''Incorrect encoding detected in {f}, please rebuild the dataset'''			)
           return

      _a           =    f.readlines()
      for lineTmp in lines:
           _a           =    lineTmp.strip()
           _a           =    line.rfind(''' '''			)
           if idx == -1:
                raise ValueError('''Incorrect dictionary format, expected \'<token> <cnt>\''''			)
           _a           =    line[:idx]
           _a           =    len(self.encoder			)










import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 			__UpperCamelCase    (							A__	, unittest.TestCase    ):
     __A	:  Any							  =   DanceDiffusionPipeline
     __A	:  Any							  =   UNCONDITIONAL_AUDIO_GENERATION_PARAMS
     __A	:  Tuple							  =   PipelineTesterMixin.required_optional_params - {
         """callback""",
         """latents""",
         """callback_steps""",
         """output_type""",
         """num_images_per_prompt""",
     }
     __A	:  Tuple							  =   UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
     __A	:  List[str]							  =   False
     __A	:  str							  =   False





     def 		UpperCamelCase(     self					):
        torch.manual_seed(0					)
        _UpperCAmelCase           =   UNetaDModel(
            block_out_channels=(32, 32, 64)				,							extra_in_channels=16				,							sample_size=512				,							sample_rate=16000				,							in_channels=2				,							out_channels=2				,							flip_sin_to_cos=_UpperCamelCase				,							use_timestep_embedding=_UpperCamelCase				,							time_embedding_type='''fourier'''				,							mid_block_type='''UNetMidBlock1D'''				,							down_block_types=('''DownBlock1DNoSkip''', '''DownBlock1D''', '''AttnDownBlock1D''')				,							up_block_types=('''AttnUpBlock1D''', '''UpBlock1D''', '''UpBlock1DNoSkip''')				,							)
        _UpperCAmelCase           =   IPNDMScheduler()

        _UpperCAmelCase           =   {
            '''unet''': unet,
            '''scheduler''': scheduler,
        }
        return components





     def 		UpperCamelCase(     self				,							_UpperCamelCase				,							_UpperCamelCase=0					):
        if str(_UpperCamelCase					).startswith('''mps'''					):
           _UpperCAmelCase           =   torch.manual_seed(_UpperCamelCase					)
        else:
           _UpperCAmelCase           =   torch.Generator(device=_UpperCamelCase					).manual_seed(_UpperCamelCase					)
        _UpperCAmelCase           =   {
            '''batch_size''': 1,
            '''generator''': generator,
            '''num_inference_steps''': 4,
        }
        return inputs





     def 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        _UpperCAmelCase           =   self.get_dummy_components()
        _UpperCAmelCase           =   DanceDiffusionPipeline(**_UpperCamelCase					)
        _UpperCAmelCase           =   pipe.to(_UpperCamelCase					)
        pipe.set_progress_bar_config(disable=_UpperCamelCase					)

        _UpperCAmelCase           =   self.get_dummy_inputs(_UpperCamelCase					)
        _UpperCAmelCase           =   pipe(**_UpperCamelCase					)
        _UpperCAmelCase           =   output.audios

        _UpperCAmelCase           =   audio[0, -3:, -3:]

        assert audio.shape == (1, 2, components["unet"].sample_size)
        _UpperCAmelCase           =   np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]					)
        assert np.abs(audio_slice.flatten() - expected_slice					).max() < 1e-2





     @skip_mps
     def 		UpperCamelCase(     self					):
        return super().test_save_load_local()





     @skip_mps
     def 		UpperCamelCase(     self					):
        return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3					)





     @skip_mps
     def 		UpperCamelCase(     self					):
        return super().test_save_load_optional_components()





     @skip_mps
     def 		UpperCamelCase(     self					):
        return super().test_attention_slicing_forward_pass()





     def 		UpperCamelCase(     self					):
        super().test_inference_batch_single_identical(expected_max_diff=3e-3					)




@slow
@require_torch_gpu
class 			__UpperCamelCase    (							unittest.TestCase    ):





     def 		UpperCamelCase(     self					):
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()





     def 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   torch_device

        _UpperCAmelCase           =   DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''					)
        _UpperCAmelCase           =   pipe.to(_UpperCamelCase					)
        pipe.set_progress_bar_config(disable=_UpperCamelCase					)

        _UpperCAmelCase           =   torch.manual_seed(0					)
        _UpperCAmelCase           =   pipe(generator=_UpperCamelCase				,							num_inference_steps=100				,							audio_length_in_s=4.096					)
        _UpperCAmelCase           =   output.audios

        _UpperCAmelCase           =   audio[0, -3:, -3:]

        assert audio.shape == (1, 2, pipe.unet.sample_size)
        _UpperCAmelCase           =   np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]					)

        assert np.abs(audio_slice.flatten() - expected_slice					).max() < 1e-2





     def 		UpperCamelCase(     self					):
        _UpperCAmelCase           =   torch_device

        _UpperCAmelCase           =   DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''				,							torch_dtype=torch.floataa					)
        _UpperCAmelCase           =   pipe.to(_UpperCamelCase					)
        pipe.set_progress_bar_config(disable=_UpperCamelCase					)

        _UpperCAmelCase           =   torch.manual_seed(0					)
        _UpperCAmelCase           =   pipe(generator=_UpperCamelCase				,							num_inference_steps=100				,							audio_length_in_s=4.096					)
        _UpperCAmelCase           =   output.audios

        _UpperCAmelCase           =   audio[0, -3:, -3:]

        assert audio.shape == (1, 2, pipe.unet.sample_size)
        _UpperCAmelCase           =   np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]					)

        assert np.abs(audio_slice.flatten() - expected_slice					).max() < 1e-2