

'''simple docstring'''


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


lowerCAmelCase__										= logging.get_logger(__name__)

lowerCAmelCase__										= "▁"

lowerCAmelCase__										= {"vocab_file": "sentencepiece.bpe.model"}

lowerCAmelCase__										= {
    "vocab_file": {
        "xlm-roberta-base": "https://huggingface.co/xlm-roberta-base/resolve/main/sentencepiece.bpe.model",
        "xlm-roberta-large": "https://huggingface.co/xlm-roberta-large/resolve/main/sentencepiece.bpe.model",
        "xlm-roberta-large-finetuned-conll02-dutch": (
            "https://huggingface.co/xlm-roberta-large-finetuned-conll02-dutch/resolve/main/sentencepiece.bpe.model"
        ),
        "xlm-roberta-large-finetuned-conll02-spanish": (
            "https://huggingface.co/xlm-roberta-large-finetuned-conll02-spanish/resolve/main/sentencepiece.bpe.model"
        ),
        "xlm-roberta-large-finetuned-conll03-english": (
            "https://huggingface.co/xlm-roberta-large-finetuned-conll03-english/resolve/main/sentencepiece.bpe.model"
        ),
        "xlm-roberta-large-finetuned-conll03-german": (
            "https://huggingface.co/xlm-roberta-large-finetuned-conll03-german/resolve/main/sentencepiece.bpe.model"
        ),
    }
}

lowerCAmelCase__										= {
    "xlm-roberta-base": 512,
    "xlm-roberta-large": 512,
    "xlm-roberta-large-finetuned-conll02-dutch": 512,
    "xlm-roberta-large-finetuned-conll02-spanish": 512,
    "xlm-roberta-large-finetuned-conll03-english": 512,
    "xlm-roberta-large-finetuned-conll03-german": 512,
}



class 					lowercase						(  a_ ):
							_lowerCamelCase		:      Tuple=   VOCAB_FILES_NAMES
							_lowerCamelCase		:      Tuple=   PRETRAINED_VOCAB_FILES_MAP
							_lowerCamelCase		:      Dict=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							_lowerCamelCase		:      str=   ["input_ids", "attention_mask"]


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

								UpperCAmelCase_		:      str       	=					{} if sp_model_kwargs is None else sp_model_kwargs

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

								UpperCAmelCase_		:      str       	=					spm.SentencePieceProcessor(**self.sp_model_kwargs)
								self.sp_model.Load(str(_snake_case))
								UpperCAmelCase_		:      Dict       	=					vocab_file

								# Original fairseq vocab and spm vocab must be "aligned":
								# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
								# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
								# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
								# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

								# Mimic fairseq token-to-id alignment for the first 4 token
								UpperCAmelCase_		:      Dict       	=					{'<s>': 0, '<pad>': 1, '</s>': 2, '<unk>': 3}

								# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
								UpperCAmelCase_		:      int       	=					1

								UpperCAmelCase_		:      Dict       	=					len(self.sp_model) + self.fairseq_offset
								UpperCAmelCase_		:      Dict       	=					{v: k for k, v in self.fairseq_tokens_to_ids.items()}


							def __getstate__(			self)							->				Any:
								UpperCAmelCase_		:      Any       	=					self.__dict__.copy()
								UpperCAmelCase_		:      Tuple       	=					None
								UpperCAmelCase_		:      str       	=					self.sp_model.serialized_model_proto()
								return state


							def __setstate__(			self    ,			_snake_case)							->				Union[str, Any]:
								UpperCAmelCase_		:      Optional[int]       	=					d

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

								UpperCAmelCase_		:      int       	=					spm.SentencePieceProcessor(**self.sp_model_kwargs)
								self.sp_model.LoadFromSerializedProto(self.sp_model_proto)


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

								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_		:      List[str]       	=					[self.sep_token_id]
								return cls + token_ids_a + sep + sep + token_ids_a + sep


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

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

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


							def 						_snake_case							(			self    ,			_snake_case    ,			_snake_case = None)							->				List[int]:
								UpperCAmelCase_		:      Any       	=					[self.sep_token_id]
								UpperCAmelCase_		:      Dict       	=					[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 						_snake_case							(			self)							->				List[Any]:
								return len(self.sp_model) + self.fairseq_offset + 1  # Add the <mask> token


							def 						_snake_case							(			self)							->				Dict:
								UpperCAmelCase_		:      Union[str, Any]       	=					{self.convert_ids_to_tokens(_snake_case): i for i in range(self.vocab_size)}
								vocab.update(self.added_tokens_encoder)
								return vocab


							def 						_snake_case							(			self    ,			_snake_case)							->				List[str]:
								return self.sp_model.encode(_snake_case    ,			out_type=_snake_case)


							def 						_snake_case							(			self    ,			_snake_case)							->				Tuple:
								if token in self.fairseq_tokens_to_ids:
									return self.fairseq_tokens_to_ids[token]
								UpperCAmelCase_		:      Any       	=					self.sp_model.PieceToId(_snake_case)

								# Need to return unknown token if the SP model returned 0
								return spm_id + self.fairseq_offset if spm_id else self.unk_token_id


							def 						_snake_case							(			self    ,			_snake_case)							->				Any:
								if index in self.fairseq_ids_to_tokens:
									return self.fairseq_ids_to_tokens[index]
								return self.sp_model.IdToPiece(index - self.fairseq_offset)


							def 						_snake_case							(			self    ,			_snake_case)							->				Tuple:
								UpperCAmelCase_		:      Optional[Any]       	=					''.join(_snake_case).replace(_snake_case    ,			' ').strip()
								return out_string





							def 						_snake_case							(			self    ,			_snake_case    ,			_snake_case = None)							->				Tuple[str]:
								if not os.path.isdir(_snake_case):
									logger.error(F"""Vocabulary path ({save_directory}) should be a directory""")
									return
								UpperCAmelCase_		:      Union[str, Any]       	=					os.path.join(
								    _snake_case    ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file'])

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

								return (out_vocab_file,)









'''simple docstring'''


import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
from .tokenization_wavaveca import WavaVecaCTCTokenizer



class 					lowercase						(  a_ ):
							_lowerCamelCase		:      List[str]=   "Wav2Vec2FeatureExtractor"
							_lowerCamelCase		:      Any=   "AutoTokenizer"


							def __init__(			self    ,			_snake_case    ,			_snake_case)							->				List[str]:
								super().__init__(_snake_case    ,			_snake_case)
								UpperCAmelCase_		:      int       	=					self.feature_extractor
								UpperCAmelCase_		:      str       	=					False


							@classmethod
							def 						_snake_case							(			cls    ,			_snake_case    ,			**_snake_case)							->				List[Any]:
								try:
									return super().from_pretrained(_snake_case    ,			**_snake_case)
								except OSError:
									warnings.warn(
									    F"""Loading a tokenizer inside {cls.__name__} from a config that does not"""
									    ' include a `tokenizer_class` attribute is deprecated and will be '
									    'removed in v5. Please add `\'tokenizer_class\': \'Wav2Vec2CTCTokenizer\'`'
									    ' attribute to either your `config.json` or `tokenizer_config.json` '
									    'file to suppress this warning: '    ,			_snake_case    ,			)

									UpperCAmelCase_		:      Any       	=					WavaVecaFeatureExtractor.from_pretrained(_snake_case    ,			**_snake_case)
									UpperCAmelCase_		:      List[str]       	=					WavaVecaCTCTokenizer.from_pretrained(_snake_case    ,			**_snake_case)

									return cls(feature_extractor=_snake_case    ,			tokenizer=_snake_case)


							def __call__(			self    ,			*_snake_case    ,			**_snake_case)							->				Optional[Any]:
								# For backward compatibility
								if self._in_target_context_manager:
									return self.current_processor(*_snake_case    ,			**_snake_case)

								if "raw_speech" in kwargs:
									warnings.warn('Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.')
									UpperCAmelCase_		:      List[str]       	=					kwargs.pop('raw_speech')
								else:
									UpperCAmelCase_		:      Dict       	=					kwargs.pop('audio'    ,			_snake_case)
								UpperCAmelCase_		:      str       	=					kwargs.pop('sampling_rate'    ,			_snake_case)
								UpperCAmelCase_		:      List[str]       	=					kwargs.pop('text'    ,			_snake_case)
								if len(_snake_case) > 0:
									UpperCAmelCase_		:      List[str]       	=					args[0]
									UpperCAmelCase_		:      Tuple       	=					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 audio is not None:
									UpperCAmelCase_		:      Union[str, Any]       	=					self.feature_extractor(_snake_case    ,			*_snake_case    ,			sampling_rate=_snake_case    ,			**_snake_case)
								if text is not None:
									UpperCAmelCase_		:      Optional[int]       	=					self.tokenizer(_snake_case    ,			**_snake_case)

								if text is None:
									return inputs
								elif audio is None:
									return encodings
								else:
									UpperCAmelCase_		:      List[str]       	=					encodings['input_ids']
									return inputs


							def 						_snake_case							(			self    ,			*_snake_case    ,			**_snake_case)							->				str:
								# For backward compatibility
								if self._in_target_context_manager:
									return self.current_processor.pad(*_snake_case    ,			**_snake_case)

								UpperCAmelCase_		:      Union[str, Any]       	=					kwargs.pop('input_features'    ,			_snake_case)
								UpperCAmelCase_		:      Dict       	=					kwargs.pop('labels'    ,			_snake_case)
								if len(_snake_case) > 0:
									UpperCAmelCase_		:      Optional[Any]       	=					args[0]
									UpperCAmelCase_		:      int       	=					args[1:]

								if input_features is not None:
									UpperCAmelCase_		:      Dict       	=					self.feature_extractor.pad(_snake_case    ,			*_snake_case    ,			**_snake_case)
								if labels is not None:
									UpperCAmelCase_		:      Union[str, Any]       	=					self.tokenizer.pad(_snake_case    ,			**_snake_case)

								if labels is None:
									return input_features
								elif input_features is None:
									return labels
								else:
									UpperCAmelCase_		:      Union[str, Any]       	=					labels['input_ids']
									return input_features


							def 						_snake_case							(			self    ,			*_snake_case    ,			**_snake_case)							->				Dict:
								return self.tokenizer.batch_decode(*_snake_case    ,			**_snake_case)


							def 						_snake_case							(			self    ,			*_snake_case    ,			**_snake_case)							->				Tuple:
								return self.tokenizer.decode(*_snake_case    ,			**_snake_case)





							@contextmanager
							def 						_snake_case							(			self)							->				Tuple:
								warnings.warn(
								    '`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '
								    'labels by using the argument `text` of the regular `__call__` method (either in the same call as '
								    'your audio inputs, or in a separate call.')
								UpperCAmelCase_		:      Optional[Any]       	=					True
								UpperCAmelCase_		:      Tuple       	=					self.tokenizer
								yield
								UpperCAmelCase_		:      Tuple       	=					self.feature_extractor
								UpperCAmelCase_		:      Tuple       	=					False






