


import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


snake_case				:    List[Any]       				=     logging.get_logger(__name__)

snake_case				:    Dict       				=     '''▁'''

snake_case				:    List[Any]       				=     {'''vocab_file''': '''sentencepiece.bpe.model''', '''monolingual_vocab_file''': '''dict.txt'''}

snake_case				:    Any       				=     {
    '''vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/sentencepiece.bpe.model''',
    },
    '''monolingual_vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/dict.txt''',
    },
}

snake_case				:    int       				=     {'''vinai/bartpho-syllable''': 10_24}







class 		_snake_case  (      _snake_case							):
 SCREAMING_SNAKE_CASE__      =       VOCAB_FILES_NAMES
 SCREAMING_SNAKE_CASE__      =       PRETRAINED_VOCAB_FILES_MAP
 SCREAMING_SNAKE_CASE__      =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 SCREAMING_SNAKE_CASE__      =       ['input_ids', 'attention_mask']




 def __init__(		self					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase="<s>"					,  _lowerCamelCase="</s>"					,  _lowerCamelCase="</s>"					,  _lowerCamelCase="<s>"					,  _lowerCamelCase="<unk>"					,  _lowerCamelCase="<pad>"					,  _lowerCamelCase="<mask>"					,  _lowerCamelCase = None					,  **_lowerCamelCase					,  ):
       # Mask token behave like a normal word, i.e. include the space before it
       a						:List[str] 							=       AddedToken(_lowerCamelCase					,  lstrip=_lowerCamelCase					,  rstrip=_lowerCamelCase    ) if isinstance(_lowerCamelCase					,  _lowerCamelCase    ) else mask_token

       a						:Tuple 							=       {} if sp_model_kwargs is None else sp_model_kwargs

       super().__init__(
           bos_token=_lowerCamelCase					,  eos_token=_lowerCamelCase					,  unk_token=_lowerCamelCase					,  sep_token=_lowerCamelCase					,  cls_token=_lowerCamelCase					,  pad_token=_lowerCamelCase					,  mask_token=_lowerCamelCase					,  sp_model_kwargs=self.sp_model_kwargs					,  **_lowerCamelCase					,  )

       a						:Optional[int] 							=       vocab_file
       a						:List[Any] 							=       monolingual_vocab_file
       a						:Tuple 							=       spm.SentencePieceProcessor(**self.sp_model_kwargs    )
       self.sp_model.Load(str(_lowerCamelCase    )    )

       # Load the reduced vocab

       # Keep order of special tokens for backward compatibility
       a						:List[str] 							=       {}
       a						:Optional[int] 							=       0
       for token in [bos_token, pad_token, eos_token, unk_token, sep_token, cls_token]:
             if str(_lowerCamelCase    ) not in self.fairseq_tokens_to_ids:
                   a						:List[str] 							=       cnt
                   cnt += 1
       with open(_lowerCamelCase					,  '''r'''					,  encoding='''utf-8'''    ) as f:
             for line in f.readlines():
                   a						:Union[str, Any] 							=       line.strip().split()[0]
                   a						:Union[str, Any] 							=       len(self.fairseq_tokens_to_ids    )
       if str(_lowerCamelCase    ) not in self.fairseq_tokens_to_ids:
             a						:Optional[int] 							=       len(self.fairseq_tokens_to_ids    )

       a						:Dict 							=       {v: k for k, v in self.fairseq_tokens_to_ids.items()}




 def __getstate__(		self    ):
       a						:Union[str, Any] 							=       self.__dict__.copy()
       a						:List[Any] 							=       None
       a						:Dict 							=       self.sp_model.serialized_model_proto()
       return state




 def __setstate__(		self					,  _lowerCamelCase    ):
       a						:Tuple 							=       d

       # for backward compatibility
       if not hasattr(self					,  '''sp_model_kwargs'''    ):
             a						:int 							=       {}

       a						:str 							=       spm.SentencePieceProcessor(**self.sp_model_kwargs    )
       self.sp_model.LoadFromSerializedProto(self.sp_model_proto    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = None    ):

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




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = None					,  _lowerCamelCase = False    ):

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

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




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = None    ):
       a						:Tuple 							=       [self.sep_token_id]
       a						:Optional[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]




 @property
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       return len(self.fairseq_ids_to_tokens    )




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:Optional[Any] 							=       {self.convert_ids_to_tokens(_lowerCamelCase    ): i for i in range(self.vocab_size    )}
       vocab.update(self.added_tokens_encoder    )
       return vocab




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase    ):
       return self.sp_model.encode(_lowerCamelCase					,  out_type=_lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase    ):
       if token in self.fairseq_tokens_to_ids:
             return self.fairseq_tokens_to_ids[token]
       else:
             return self.unk_token_id




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase    ):
       return self.fairseq_ids_to_tokens[index]




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase    ):
       a						:Union[str, Any] 							=       ''''''.join(_lowerCamelCase    ).replace(_lowerCamelCase					,  ''' '''    ).strip()
       return out_string




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = None    ):
       if not os.path.isdir(_lowerCamelCase    ):
             logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''    )
             return
       a						:int 							=       os.path.join(
           _lowerCamelCase					,  (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']    )
       a						:Optional[int] 							=       os.path.join(
           _lowerCamelCase					,  (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''monolingual_vocab_file''']					,  )

       if os.path.abspath(self.vocab_file    ) != os.path.abspath(_lowerCamelCase    ) and os.path.isfile(self.vocab_file    ):
             copyfile(self.vocab_file					,  _lowerCamelCase    )
       elif not os.path.isfile(self.vocab_file    ):
             with open(_lowerCamelCase					,  '''wb'''    ) as fi:
                   a						:List[Any] 							=       self.sp_model.serialized_model_proto()
                   fi.write(_lowerCamelCase    )

       if os.path.abspath(self.monolingual_vocab_file    ) != os.path.abspath(
           _lowerCamelCase    ) and os.path.isfile(self.monolingual_vocab_file    ):
             copyfile(self.monolingual_vocab_file					,  _lowerCamelCase    )
       elif not os.path.isfile(self.monolingual_vocab_file    ):
             with open(_lowerCamelCase					,  '''w'''					,  encoding='''utf-8'''    ) as fp:
                   for token in self.fairseq_tokens_to_ids:
                         if token not in self.all_special_tokens:
                               fp.write(F'''{str(_lowerCamelCase    )} \n'''    )

       return out_vocab_file, out_monolingual_vocab_file













import json
import os
from typing import Optional

import numpy as np

from ...feature_extraction_utils import BatchFeature
from ...processing_utils import ProcessorMixin
from ...utils import logging
from ...utils.hub import get_file_from_repo
from ..auto import AutoTokenizer


_a      	=  logging.get_logger(__name__)





class __lowerCamelCase (      snake_case__):





	"""simple docstring"""






	UpperCamelCase__								=      "AutoTokenizer"
	UpperCamelCase__								=      ["tokenizer"]

	UpperCamelCase__								=      {
	    "semantic_prompt": 1,
	    "coarse_prompt": 2,
	    "fine_prompt": 2,
	}

	def __init__(							self       ,    UpperCAmelCase       ,    UpperCAmelCase=None    ):
							"""simple docstring"""







							super().__init__(UpperCAmelCase    )

							_UpperCAmelCase									=			speaker_embeddings

	@classmethod
	def        UpperCamelCase    (							cls       ,    UpperCAmelCase       ,    UpperCAmelCase="speaker_embeddings_path.json"       ,    **UpperCAmelCase    ):
							"""simple docstring"""







							if speaker_embeddings_dict_path is not None:
													_UpperCAmelCase									=			get_file_from_repo(
													    UpperCAmelCase       ,    UpperCAmelCase       ,    subfolder=kwargs.pop('subfolder'       ,    UpperCAmelCase    )       ,    cache_dir=kwargs.pop('cache_dir'       ,    UpperCAmelCase    )       ,    force_download=kwargs.pop('force_download'       ,    UpperCAmelCase    )       ,    proxies=kwargs.pop('proxies'       ,    UpperCAmelCase    )       ,    resume_download=kwargs.pop('resume_download'       ,    UpperCAmelCase    )       ,    local_files_only=kwargs.pop('local_files_only'       ,    UpperCAmelCase    )       ,    use_auth_token=kwargs.pop('use_auth_token'       ,    UpperCAmelCase    )       ,    revision=kwargs.pop('revision'       ,    UpperCAmelCase    )       ,    )
													if speaker_embeddings_path is None:
																			logger.warning(
																			    F"""`{os.path.join(UpperCAmelCase       ,    UpperCAmelCase    )}` does not exists
                    , no preloaded speaker embeddings will be used - Make sure to provide a correct path to the json
                    dictionnary if wanted, otherwise set `speaker_embeddings_dict_path=None`."""    )
																			_UpperCAmelCase									=			None
													else:
																			with open(UpperCAmelCase    ) as speaker_embeddings_json:
																									_UpperCAmelCase									=			json.load(UpperCAmelCase    )
							else:
													_UpperCAmelCase									=			None

							_UpperCAmelCase									=			AutoTokenizer.from_pretrained(UpperCAmelCase       ,    **UpperCAmelCase    )

							return cls(tokenizer=UpperCAmelCase       ,    speaker_embeddings=UpperCAmelCase    )

	def        UpperCamelCase    (							self       ,    UpperCAmelCase       ,    UpperCAmelCase="speaker_embeddings_path.json"       ,    UpperCAmelCase="speaker_embeddings"       ,    UpperCAmelCase = False       ,    **UpperCAmelCase       ,    ):
							"""simple docstring"""







							if self.speaker_embeddings is not None:
													os.makedirs(os.path.join(UpperCAmelCase       ,    UpperCAmelCase       ,    'v2'    )       ,    exist_ok=UpperCAmelCase    )

													_UpperCAmelCase									=			{}

													_UpperCAmelCase									=			save_directory

													for prompt_key in self.speaker_embeddings:
																			if prompt_key != "repo_or_path":
																									_UpperCAmelCase									=			self._load_voice_preset(UpperCAmelCase    )

																									_UpperCAmelCase									=			{}
																									for key in self.speaker_embeddings[prompt_key]:
																															np.save(
																															    os.path.join(
																															        embeddings_dict['repo_or_path']       ,    UpperCAmelCase       ,    F"""{prompt_key}_{key}"""    )       ,    voice_preset[key]       ,    allow_pickle=UpperCAmelCase       ,    )
																															_UpperCAmelCase									=			os.path.join(UpperCAmelCase       ,    F"""{prompt_key}_{key}.npy"""    )

																									_UpperCAmelCase									=			tmp_dict

													with open(os.path.join(UpperCAmelCase       ,    UpperCAmelCase    )       ,    'w'    ) as fp:
																			json.dump(UpperCAmelCase       ,    UpperCAmelCase    )

							super().save_pretrained(UpperCAmelCase       ,    UpperCAmelCase       ,    **UpperCAmelCase    )

	def        UpperCamelCase    (							self       ,    UpperCAmelCase = None       ,    **UpperCAmelCase    ):
							"""simple docstring"""







							_UpperCAmelCase									=			self.speaker_embeddings[voice_preset]

							_UpperCAmelCase									=			{}
							for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
													if key not in voice_preset_paths:
																			raise ValueError(
																			    F"""Voice preset unrecognized, missing {key} as a key in self.speaker_embeddings[{voice_preset}]."""    )

													_UpperCAmelCase									=			get_file_from_repo(
													    self.speaker_embeddings.get('repo_or_path'       ,    '/'    )       ,    voice_preset_paths[key]       ,    subfolder=kwargs.pop('subfolder'       ,    UpperCAmelCase    )       ,    cache_dir=kwargs.pop('cache_dir'       ,    UpperCAmelCase    )       ,    force_download=kwargs.pop('force_download'       ,    UpperCAmelCase    )       ,    proxies=kwargs.pop('proxies'       ,    UpperCAmelCase    )       ,    resume_download=kwargs.pop('resume_download'       ,    UpperCAmelCase    )       ,    local_files_only=kwargs.pop('local_files_only'       ,    UpperCAmelCase    )       ,    use_auth_token=kwargs.pop('use_auth_token'       ,    UpperCAmelCase    )       ,    revision=kwargs.pop('revision'       ,    UpperCAmelCase    )       ,    )
													if path is None:
																			raise ValueError(
																			    F"""`{os.path.join(self.speaker_embeddings.get("repo_or_path"       ,    "/"    )       ,    voice_preset_paths[key]    )}` does not exists
                    , no preloaded voice preset will be used - Make sure to provide correct paths to the {voice_preset}
                    embeddings."""    )

													_UpperCAmelCase									=			np.load(UpperCAmelCase    )

							return voice_preset_dict

	def        UpperCamelCase    (							self       ,    UpperCAmelCase = None    ):
							"""simple docstring"""







							for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
													if key not in voice_preset:
																			raise ValueError(F"""Voice preset unrecognized, missing {key} as a key."""    )

													if not isinstance(voice_preset[key]       ,    np.ndarray    ):
																			raise ValueError(F"""{key} voice preset must be a {str(self.preset_shape[key]    )}D ndarray."""    )

													if len(voice_preset[key].shape    ) != self.preset_shape[key]:
																			raise ValueError(F"""{key} voice preset must be a {str(self.preset_shape[key]    )}D ndarray."""    )

	def __call__(							self       ,    UpperCAmelCase=None       ,    UpperCAmelCase=None       ,    UpperCAmelCase="pt"       ,    UpperCAmelCase=256       ,    UpperCAmelCase=False       ,    UpperCAmelCase=True       ,    UpperCAmelCase=False       ,    **UpperCAmelCase       ,    ):
							"""simple docstring"""







							if voice_preset is not None and not isinstance(UpperCAmelCase       ,    UpperCAmelCase    ):
													if (
													    isinstance(UpperCAmelCase       ,    UpperCAmelCase    )
													    and self.speaker_embeddings is not None
													    and voice_preset in self.speaker_embeddings
													):
																			_UpperCAmelCase									=			self._load_voice_preset(UpperCAmelCase    )

													else:
																			if isinstance(UpperCAmelCase       ,    UpperCAmelCase    ) and not voice_preset.endswith('.npz'    ):
																									_UpperCAmelCase									=			voice_preset + '.npz'

																			_UpperCAmelCase									=			np.load(UpperCAmelCase    )

							if voice_preset is not None:
													self._validate_voice_preset_dict(UpperCAmelCase       ,    **UpperCAmelCase    )
													_UpperCAmelCase									=			BatchFeature(data=UpperCAmelCase       ,    tensor_type=UpperCAmelCase    )

							_UpperCAmelCase									=			self.tokenizer(
							    UpperCAmelCase       ,    return_tensors=UpperCAmelCase       ,    padding='max_length'       ,    max_length=UpperCAmelCase       ,    return_attention_mask=UpperCAmelCase       ,    return_token_type_ids=UpperCAmelCase       ,    add_special_tokens=UpperCAmelCase       ,    **UpperCAmelCase       ,    )

							if voice_preset is not None:
													_UpperCAmelCase									=			voice_preset

							return encoded_text
