






"""simple docstring"""







import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

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


__a       :	Dict             = logging.get_logger(__name__)

__a       :	Dict             = '▁'

__a       :	int             = {
    'vocab_file': 'vocab.json',
    'spm_file': 'sentencepiece.bpe.model',
    'tokenizer_config_file': 'tokenizer_config.json',
}

__a       :	str             = {
    'vocab_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json',
    },
    'spm_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model',
    },
    'tokenizer_config_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json',
    },
}

__a       :	Any             = {
    'facebook/m2m100_418M': 1024,
}

# fmt: off
__a       :	Optional[Any]             = {
    'm2m100': ['af', 'am', 'ar', 'ast', 'az', 'ba', 'be', 'bg', 'bn', 'br', 'bs', 'ca', 'ceb', 'cs', 'cy', 'da', 'de', 'el', 'en', 'es', 'et', 'fa', 'ff', 'fi', 'fr', 'fy', 'ga', 'gd', 'gl', 'gu', 'ha', 'he', 'hi', 'hr', 'ht', 'hu', 'hy', 'id', 'ig', 'ilo', 'is', 'it', 'ja', 'jv', 'ka', 'kk', 'km', 'kn', 'ko', 'lb', 'lg', 'ln', 'lo', 'lt', 'lv', 'mg', 'mk', 'ml', 'mn', 'mr', 'ms', 'my', 'ne', 'nl', 'no', 'ns', 'oc', 'or', 'pa', 'pl', 'ps', 'pt', 'ro', 'ru', 'sd', 'si', 'sk', 'sl', 'so', 'sq', 'sr', 'ss', 'su', 'sv', 'sw', 'ta', 'th', 'tl', 'tn', 'tr', 'uk', 'ur', 'uz', 'vi', 'wo', 'xh', 'yi', 'yo', 'zh', 'zu'],
    'wmt21': ['en', 'ha', 'is', 'ja', 'cs', 'ru', 'zh', 'de']
}







class        _SCREAMING_SNAKE_CASE						(  __snake_case							):





							"""simple docstring"""





							_SCREAMING_SNAKE_CASE    =VOCAB_FILES_NAMES
							_SCREAMING_SNAKE_CASE    =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							_SCREAMING_SNAKE_CASE    =PRETRAINED_VOCAB_FILES_MAP
							_SCREAMING_SNAKE_CASE    =['input_ids', 'attention_mask']

							_SCREAMING_SNAKE_CASE    =[]
							_SCREAMING_SNAKE_CASE    =[]




							def __init__(   self:     List[Any]				,		__A:     Optional[Any]				,		__A:     List[Any]				,		__A:     Tuple=None				,		__A:     Tuple=None				,		__A:     Any="<s>"				,		__A:     str="</s>"				,		__A:     List[Any]="</s>"				,		__A:     Union[str, Any]="<pad>"				,		__A:     Optional[int]="<unk>"				,		__A:     Union[str, Any]="m2m100"				,		__A:     Optional[Dict[str, Any]] = None				,		__A:     Union[str, Any]=8				,		**__A:     List[str]				,		):

												'''simple docstring'''

												a__          =  {} if sp_model_kwargs is None else sp_model_kwargs

												a__          =  language_codes
												a__          =  FAIRSEQ_LANGUAGE_CODES[language_codes]
												a__          =  {lang_code: F'__{lang_code}__' for lang_code in fairseq_language_code}

												a__          =  kwargs.get('''additional_special_tokens'''				,		[]    )
												kwargs["additional_special_tokens"] += [
												    self.get_lang_token(__A    )
												    for lang_code in fairseq_language_code
												    if self.get_lang_token(__A    ) not in kwargs["additional_special_tokens"]
												]

												super().__init__(
												    src_lang=__A				,		tgt_lang=__A				,		bos_token=__A				,		eos_token=__A				,		sep_token=__A				,		unk_token=__A				,		pad_token=__A				,		language_codes=__A				,		sp_model_kwargs=self.sp_model_kwargs				,		num_madeup_words=__A				,		**__A				,		)

												a__          =  vocab_file
												a__          =  load_json(__A    )
												a__          =  {v: k for k, v in self.encoder.items()}
												a__          =  spm_file
												a__          =  load_spm(__A				,		self.sp_model_kwargs    )

												a__          =  len(self.encoder    )

												a__          =  {
												    self.get_lang_token(__A    ): self.encoder_size + i for i, lang_code in enumerate(__A    )
												}
												a__          =  {lang_code: self.encoder_size + i for i, lang_code in enumerate(__A    )}
												a__          =  {v: k for k, v in self.lang_token_to_id.items()}

												a__          =  src_lang if src_lang is not None else '''en'''
												a__          =  tgt_lang
												a__          =  self.get_lang_id(self._src_lang    )
												self.set_src_lang_special_tokens(self._src_lang    )

												a__          =  num_madeup_words




							@property
							def  lowercase		(   self:     Optional[Any]    ):

												'''simple docstring'''

												return len(self.encoder    ) + len(self.lang_token_to_id    )




							@property
							def  lowercase		(   self:     Optional[Any]    ):

												'''simple docstring'''

												return self._src_lang




							@src_lang.setter
							def  lowercase		(   self:     Any				,		__A:     str    ):

												'''simple docstring'''

												a__          =  new_src_lang
												self.set_src_lang_special_tokens(self._src_lang    )




							def  lowercase		(   self:     Dict				,		__A:     str    ):

												'''simple docstring'''

												return self.sp_model.encode(__A				,		out_type=__A    )




							def  lowercase		(   self:     Any				,		__A:     List[str]    ):

												'''simple docstring'''

												if token in self.lang_token_to_id:
																	return self.lang_token_to_id[token]
												return self.encoder.get(__A				,		self.encoder[self.unk_token]    )




							def  lowercase		(   self:     Optional[Any]				,		__A:     int    ):

												'''simple docstring'''

												if index in self.id_to_lang_token:
																	return self.id_to_lang_token[index]
												return self.decoder.get(__A				,		self.unk_token    )




							def  lowercase		(   self:     Optional[int]				,		__A:     str    ):

												'''simple docstring'''

												a__          =  []
												a__          =  ''''''
												for token in tokens:
																	# make sure that special tokens are not decoded using sentencepiece model
																	if token in self.all_special_tokens:
																						out_string += self.sp_model.decode(__A    ) + token
																						a__          =  []
																	else:
																						current_sub_tokens.append(__A    )
												out_string += self.sp_model.decode(__A    )
												return out_string.strip()




							def  lowercase		(   self:     List[str]				,		__A:     List[int]				,		__A:     Optional[List[int]] = None				,		__A:     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    )

												a__          =  [1] * len(self.prefix_tokens    )
												a__          =  [1] * len(self.suffix_tokens    )
												if token_ids_a is None:
																	return prefix_ones + ([0] * len(__A    )) + suffix_ones
												return prefix_ones + ([0] * len(__A    )) + ([0] * len(__A    )) + suffix_ones




							def  lowercase		(   self:     Tuple				,		__A:     List[int]				,		__A:     Optional[List[int]] = None    ):

												'''simple docstring'''

												if token_ids_a is None:
																	return self.prefix_tokens + token_ids_a + self.suffix_tokens
												# We don't expect to process pairs, but leave the pair logic for API consistency
												return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens




							def  lowercase		(   self:     Any    ):

												'''simple docstring'''

												a__          =  {self.convert_ids_to_tokens(__A    ): i for i in range(self.vocab_size    )}
												vocab.update(self.added_tokens_encoder    )
												return vocab




							def __getstate__(   self:     Optional[Any]    ):

												'''simple docstring'''

												a__          =  self.__dict__.copy()
												a__          =  None
												return state




							def __setstate__(   self:     Dict				,		__A:     Dict    ):

												'''simple docstring'''

												a__          =  d

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

												a__          =  load_spm(self.spm_file				,		self.sp_model_kwargs    )




							def  lowercase		(   self:     Optional[int]				,		__A:     str				,		__A:     Optional[str] = None    ):

												'''simple docstring'''

												a__          =  Path(__A    )
												if not save_dir.is_dir():
																	raise OSError(F'{save_directory} should be a directory'    )
												a__          =  save_dir / (
												    (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''vocab_file''']
												)
												a__          =  save_dir / (
												    (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''spm_file''']
												)

												save_json(self.encoder				,		__A    )

												if os.path.abspath(self.spm_file    ) != os.path.abspath(__A    ) and os.path.isfile(self.spm_file    ):
																	copyfile(self.spm_file				,		__A    )
												elif not os.path.isfile(self.spm_file    ):
																	with open(__A				,		'''wb'''    ) as fi:
																						a__          =  self.sp_model.serialized_model_proto()
																						fi.write(__A    )

												return (str(__A    ), str(__A    ))




							def  lowercase		(   self:     Optional[int]				,		__A:     List[str]				,		__A:     str = "en"				,		__A:     Optional[List[str]] = None				,		__A:     str = "ro"				,		**__A:     str				,		):

												'''simple docstring'''

												a__          =  src_lang
												a__          =  tgt_lang
												self.set_src_lang_special_tokens(self.src_lang    )
												return super().prepare_seqaseq_batch(__A				,		__A				,		**__A    )




							def  lowercase		(   self:     Union[str, Any]				,		__A:     Tuple				,		__A:     Optional[str]				,		__A:     Optional[str]				,		**__A:     List[str]    ):

												'''simple docstring'''

												if src_lang is None or tgt_lang is None:
																	raise ValueError('''Translation requires a `src_lang` and a `tgt_lang` for this model'''    )
												a__          =  src_lang
												a__          =  self(__A				,		add_special_tokens=__A				,		**__A    )
												a__          =  self.get_lang_id(__A    )
												a__          =  tgt_lang_id
												return inputs




							def  lowercase		(   self:     Dict    ):

												'''simple docstring'''

												self.set_src_lang_special_tokens(self.src_lang    )




							def  lowercase		(   self:     int    ):

												'''simple docstring'''

												self.set_tgt_lang_special_tokens(self.tgt_lang    )




							def  lowercase		(   self:     Any				,		__A:     str    ):

												'''simple docstring'''

												a__          =  self.get_lang_token(__A    )
												a__          =  self.lang_token_to_id[lang_token]
												a__          =  [self.cur_lang_id]
												a__          =  [self.eos_token_id]




							def  lowercase		(   self:     Tuple				,		__A:     str    ):

												'''simple docstring'''

												a__          =  self.get_lang_token(__A    )
												a__          =  self.lang_token_to_id[lang_token]
												a__          =  [self.cur_lang_id]
												a__          =  [self.eos_token_id]




							def  lowercase		(   self:     Any				,		__A:     str    ):

												'''simple docstring'''

												return self.lang_code_to_token[lang]





							def  lowercase		(   self:     List[Any]				,		__A:     str    ):

												'''simple docstring'''

												a__          =  self.get_lang_token(__A    )
												return self.lang_token_to_id[lang_token]





def        SCREAMING_SNAKE_CASE       (      lowerCamelCase_ ,     lowerCamelCase_):
					a__          =  sentencepiece.SentencePieceProcessor(**lowerCamelCase_)
					spm.Load(str(lowerCamelCase_))
					return spm





def        SCREAMING_SNAKE_CASE       (      lowerCamelCase_):
					with open(lowerCamelCase_ ,     '''r''') as f:
										return json.load(lowerCamelCase_)





def        SCREAMING_SNAKE_CASE       (      lowerCamelCase_ ,     lowerCamelCase_):
					with open(lowerCamelCase_ ,     '''w''') as f:
										json.dump(lowerCamelCase_ ,     lowerCamelCase_ ,     indent=2)










"""simple docstring"""







from typing import List, Optional

import numpy as np

from ...processing_utils import ProcessorMixin
from ...utils import to_numpy







class        _SCREAMING_SNAKE_CASE						(  __snake_case							):





							"""simple docstring"""





							_SCREAMING_SNAKE_CASE    ='EncodecFeatureExtractor'
							_SCREAMING_SNAKE_CASE    =('T5Tokenizer', 'T5TokenizerFast')




							def __init__(   self:     List[Any]				,		__A:     Any				,		__A:     Dict    ):

												'''simple docstring'''

												super().__init__(__A				,		__A    )
												a__          =  self.feature_extractor
												a__          =  False




							def  lowercase		(   self:     Union[str, Any]				,		__A:     List[Any]=None				,		__A:     Optional[Any]=None				,		__A:     List[Any]=True    ):

												'''simple docstring'''

												return self.tokenizer.get_decoder_prompt_ids(task=__A				,		language=__A				,		no_timestamps=__A    )




							def __call__(   self:     Union[str, Any]				,		*__A:     int				,		**__A:     Dict    ):

												'''simple docstring'''

												if self._in_target_context_manager:
																	return self.current_processor(*__A				,		**__A    )

												a__          =  kwargs.pop('''audio'''				,		__A    )
												a__          =  kwargs.pop('''sampling_rate'''				,		__A    )
												a__          =  kwargs.pop('''text'''				,		__A    )
												if len(__A    ) > 0:
																	a__          =  args[0]
																	a__          =  args[1:]

												if audio is None and text is None:
																	raise ValueError('''You need to specify either an `audio` or `text` input to process.'''    )

												if text is not None:
																	a__          =  self.tokenizer(__A				,		**__A    )

												if audio is not None:
																	a__          =  self.feature_extractor(__A				,		*__A				,		sampling_rate=__A				,		**__A    )

												if audio is None:
																	return inputs

												elif text is None:
																	return audio_inputs

												else:
																	a__          =  audio_inputs['''input_values''']
																	if "padding_mask" in audio_inputs:
																						a__          =  audio_inputs['''padding_mask''']
																	return inputs




							def  lowercase		(   self:     Union[str, Any]				,		*__A:     List[str]				,		**__A:     Tuple    ):

												'''simple docstring'''

												a__          =  kwargs.pop('''audio'''				,		__A    )
												a__          =  kwargs.pop('''padding_mask'''				,		__A    )

												if len(__A    ) > 0:
																	a__          =  args[0]
																	a__          =  args[1:]

												if audio_values is not None:
																	return self._decode_audio(__A				,		padding_mask=__A    )
												else:
																	return self.tokenizer.batch_decode(*__A				,		**__A    )




							def  lowercase		(   self:     Union[str, Any]				,		*__A:     Optional[int]				,		**__A:     Optional[Any]    ):

												'''simple docstring'''

												return self.tokenizer.decode(*__A				,		**__A    )





							def  lowercase		(   self:     Union[str, Any]				,		__A:     Dict				,		__A:     Optional = None    ):

												'''simple docstring'''

												a__          =  to_numpy(__A    )
												a__       ,a__       ,a__          =  audio_values.shape

												if padding_mask is None:
																	return list(__A    )

												a__          =  to_numpy(__A    )

												# match the sequence length of the padding mask to the generated audio arrays by padding with the **non-padding**
												# token (so that the generated audio values are **not** treated as padded tokens)
												a__          =  seq_len - padding_mask.shape[-1]
												a__          =  1 - self.feature_extractor.padding_value
												a__          =  np.pad(__A				,		((0, 0), (0, difference))				,		'''constant'''				,		constant_values=__A    )

												a__          =  audio_values.tolist()
												for i in range(__A    ):
																	a__          =  np.asarray(audio_values[i]    )[
																	    padding_mask[i][None, :] != self.feature_extractor.padding_value
																	]
																	a__          =  sliced_audio.reshape(__A				,		-1    )

												return audio_values


