




import logging
import pathlib
import re
import sys
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Set, Union

import datasets
import librosa
import numpy as np
import torch
from lang_trans import arabic
from packaging import version
from torch import nn

from transformers import (
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaForCTC,
    WavaVecaProcessor,
    is_apex_available,
    trainer_utils,
)


if is_apex_available():
				from apex import amp

if version.parse(version.parse(torch.__version__).base_version) >= version.parse('''1.6'''):
				lowercase_		=							True
				from torch.cuda.amp import autocast


lowercase_		=							logging.getLogger(__name__)





@dataclass
class 		A__				:
	lowerCamelCase__		: str				    =field(
	    metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"} )
	lowerCamelCase__		: Optional[str]				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"}		,     )
	lowerCamelCase__		: Optional[bool]				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "Whether to freeze the feature extractor layers of the model."} )
	lowerCamelCase__		: Optional[bool]				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "Whether to log verbose messages or not."}		,     )







def  lowerCAmelCase  (	UpperCAmelCase,	UpperCAmelCase       )  ->Dict:






					"""simple docstring"""






					logging.basicConfig(
					    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s''',	datefmt='''%m/%d/%Y %H:%M:%S''',	handlers=[logging.StreamHandler(sys.stdout       )],	)
					__magic_name__					:							Optional[Any]		   =							logging.WARNING
					if model_args.verbose_logging:
										__magic_name__					:							str		   =							logging.DEBUG
					elif trainer_utils.is_main_process(training_args.local_rank       ):
										__magic_name__					:							Optional[Any]		   =							logging.INFO
					logger.setLevel(UpperCAmelCase       )





@dataclass
class 		A__				:
	lowerCamelCase__		: str				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "The name of the dataset to use (via the datasets library)."} )
	lowerCamelCase__		: Optional[str]				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "The configuration name of the dataset to use (via the datasets library)."} )
	lowerCamelCase__		: Optional[str]				    =field(
	    default="train"		,     metadata={
	        "help": "The name of the training data set split to use (via the datasets library). Defaults to 'train'"
	    }		,     )
	lowerCamelCase__		: Optional[str]				    =field(
	    default="validation"		,     metadata={
	        "help": (
	            "The name of the validation data set split to use (via the datasets library). Defaults to 'validation'"
	        )
	    }		,     )
	lowerCamelCase__		: Optional[str]				    =field(
	    default="text"		,     metadata={"help": "Column in the dataset that contains label (target text). Defaults to 'text'"}		,     )
	lowerCamelCase__		: Optional[str]				    =field(
	    default="file"		,     metadata={"help": "Column in the dataset that contains speech file path. Defaults to 'file'"}		,     )
	lowerCamelCase__		: Optional[bool]				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "Resample loaded audio to target feature extractor's sampling rate or not."}		,     )
	lowerCamelCase__		: Optional[float]				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "Filters out examples longer than specified. Defaults to no filtering."}		,     )
	lowerCamelCase__		: Optional[str]				    =field(
	    default="librispeech"		,     metadata={
	        "help": (
	            "Orthography used for normalization and tokenization: 'librispeech' (default), 'timit', or"
	            " 'buckwalter'."
	        )
	    }		,     )
	lowerCamelCase__		: bool				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "Overwrite the cached preprocessed datasets or not."} )
	lowerCamelCase__		: Optional[int]				    =field(
	    default=__SCREAMING_SNAKE_CASE		,     metadata={"help": "The number of processes to use for the preprocessing."}		,     )





@dataclass
class 		A__				:
	lowerCamelCase__		: bool				    =False
	lowerCamelCase__		: Optional[str]				    =None
	lowerCamelCase__		: Optional[str]				    ="|"
	lowerCamelCase__		: Optional[Dict[str, str]]				    =field(default_factory=__SCREAMING_SNAKE_CASE )
	lowerCamelCase__		: Optional[Set[str]]				    =field(default_factory=__SCREAMING_SNAKE_CASE )
	lowerCamelCase__		: Optional[Callable[[str], str]]				    =None






	@classmethod
	def  lowercase      (	cls				,     lowerCamelCase						)      ->   str:




						"""simple docstring"""






						if name == "librispeech":
											return cls()
						if name == "timit":
											return cls(
											    do_lower_case=lowerCamelCase				,     translation_table=str.maketrans({'''-''': ''' '''}						)				,     )
						if name == "buckwalter":
											__magic_name__					:							List[Any]		   =							{
											    '''-''': ''' ''',  # sometimes used to represent pauses
											    '''^''': '''v''',  # fixing "tha" in arabic_speech_corpus dataset
											}
											return cls(
											    vocab_file=pathlib.Path(lowerCamelCase						).parent.joinpath('''vocab/buckwalter.json'''						)				,     word_delimiter_token='''/'''				,     translation_table=str.maketrans(lowerCamelCase						)				,     words_to_remove={'''sil'''}				,     untransliterator=arabic.buckwalter.untransliterate				,     )
						raise ValueError(F'''Unsupported orthography: \'{name}\'.'''						)






	def  lowercase      (	self				,     lowerCamelCase						)      ->   str:




						"""simple docstring"""






						if len(self.translation_table						) > 0:
											__magic_name__					:							Any		   =							text.translate(self.translation_table						)
						if len(self.words_to_remove						) == 0:
											__magic_name__					:							int		   =							''' '''.join(text.split()						)  # clean up whitespaces
						else:
											__magic_name__					:							Optional[int]		   =							''' '''.join(w for w in text.split() if w not in self.words_to_remove						)  # and clean up whilespaces
						return text






	def  lowercase      (	self				,     lowerCamelCase						)      ->   WavaVecaProcessor:




						"""simple docstring"""






						__magic_name__					:							int		   =							WavaVecaFeatureExtractor.from_pretrained(
						    model_args.model_name_or_path				,     cache_dir=model_args.cache_dir						)
						if self.vocab_file:
											__magic_name__					:							Optional[Any]		   =							WavaVecaCTCTokenizer(
											    self.vocab_file				,     cache_dir=model_args.cache_dir				,     do_lower_case=self.do_lower_case				,     word_delimiter_token=self.word_delimiter_token				,     )
						else:
											__magic_name__					:							Tuple		   =							WavaVecaCTCTokenizer.from_pretrained(
											    model_args.model_name_or_path				,     cache_dir=model_args.cache_dir				,     do_lower_case=self.do_lower_case				,     word_delimiter_token=self.word_delimiter_token				,     )
						return WavaVecaProcessor(lowerCamelCase				,     lowerCamelCase						)





@dataclass
class 		A__				:
	lowerCamelCase__		: WavaVecaProcessor
	lowerCamelCase__		: Union[bool, str]				    =True
	lowerCamelCase__		: Optional[int]				    =None
	lowerCamelCase__		: Optional[int]				    =None
	lowerCamelCase__		: Optional[int]				    =None
	lowerCamelCase__		: Optional[int]				    =None






	def __call__(	self				,     lowerCamelCase						)      ->   Dict[str, torch.Tensor]:




						"""simple docstring"""






						__magic_name__					:							Dict		   =							[{'''input_values''': feature['''input_values''']} for feature in features]
						__magic_name__					:							str		   =							[{'''input_ids''': feature['''labels''']} for feature in features]

						__magic_name__					:							List[Any]		   =							self.processor.pad(
						    lowerCamelCase				,     padding=self.padding				,     max_length=self.max_length				,     pad_to_multiple_of=self.pad_to_multiple_of				,     return_tensors='''pt'''				,     )
						__magic_name__					:							Optional[int]		   =							self.processor.pad(
						    labels=lowerCamelCase				,     padding=self.padding				,     max_length=self.max_length_labels				,     pad_to_multiple_of=self.pad_to_multiple_of_labels				,     return_tensors='''pt'''				,     )

						# replace padding with -100 to ignore loss correctly
						__magic_name__					:							Dict		   =							labels_batch['''input_ids'''].masked_fill(labels_batch.attention_mask.ne(1						)				,     -100						)

						__magic_name__					:							Optional[int]		   =							labels

						return batch





class 		A__				(   __SCREAMING_SNAKE_CASE ):






	def  lowercase      (	self				,     lowerCamelCase				,     lowerCamelCase						)      ->   torch.Tensor:




						"""simple docstring"""






						model.train()
						__magic_name__					:							List[str]		   =							self._prepare_inputs(lowerCamelCase						)

						if self.use_amp:
											with autocast():
																__magic_name__					:							Optional[Any]		   =							self.compute_loss(lowerCamelCase				,     lowerCamelCase						)
						else:
											__magic_name__					:							Any		   =							self.compute_loss(lowerCamelCase				,     lowerCamelCase						)

						if self.args.n_gpu > 1:
											if model.module.config.ctc_loss_reduction == "mean":
																__magic_name__					:							str		   =							loss.mean()
											elif model.module.config.ctc_loss_reduction == "sum":
																__magic_name__					:							Optional[Any]		   =							loss.sum() / (inputs['''labels'''] >= 0).sum()
											else:
																raise ValueError(F'''{model.config.ctc_loss_reduction} is not valid. Choose one of [\'mean\', \'sum\']'''						)

						if self.args.gradient_accumulation_steps > 1:
											__magic_name__					:							str		   =							loss / self.args.gradient_accumulation_steps

						if self.use_amp:
											self.scaler.scale(lowerCamelCase						).backward()
						elif self.use_apex:
											with amp.scale_loss(lowerCamelCase				,     self.optimizer						) as scaled_loss:
																scaled_loss.backward()
						elif self.deepspeed:
											self.deepspeed.backward(lowerCamelCase						)
						else:
											loss.backward()

						return loss.detach()







def  lowerCAmelCase  (	)  ->Tuple:






					"""simple docstring"""






					__magic_name__					:							List[str]		   =							HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)       )

					__magic_name__   , __magic_name__   , __magic_name__					:							Optional[int]		   =							parser.parse_args_into_dataclasses()
					configure_logger(UpperCAmelCase,	UpperCAmelCase       )

					__magic_name__					:							int		   =							Orthography.from_name(data_args.orthography.lower()       )
					__magic_name__					:							Tuple		   =							orthography.create_processor(UpperCAmelCase       )
					__magic_name__					:							int		   =							WavaVecaForCTC.from_pretrained(
					    model_args.model_name_or_path,	cache_dir=model_args.cache_dir,	gradient_checkpointing=training_args.gradient_checkpointing,	vocab_size=len(processor.tokenizer       ),	)

					__magic_name__					:							Dict		   =							datasets.load_dataset(
					    data_args.dataset_name,	data_args.dataset_config_name,	split=data_args.train_split_name       )
					__magic_name__					:							int		   =							datasets.load_dataset(
					    data_args.dataset_name,	data_args.dataset_config_name,	split=data_args.validation_split_name       )

					__magic_name__					:							List[str]		   =							datasets.load_metric('''wer'''       )
					__magic_name__					:							Any		   =							processor.feature_extractor.sampling_rate if data_args.target_feature_extractor_sampling_rate else None
					__magic_name__					:							str		   =							''''''.join(t for t in processor.tokenizer.get_vocab().keys() if len(UpperCAmelCase       ) == 1       )
					__magic_name__					:							Optional[int]		   =							re.compile(  # remove characters not in vocabulary
					    rF'''[^\s{re.escape(UpperCAmelCase       )}]''',	flags=re.IGNORECASE if processor.tokenizer.do_lower_case else 0,	)
					__magic_name__					:							Dict		   =							[]

					def prepare_example(UpperCAmelCase       ):  # TODO(elgeish) make use of multiprocessing?
										__magic_name__   , __magic_name__					:							Tuple		   =							librosa.load(example[data_args.speech_file_column],	sr=UpperCAmelCase       )
										if data_args.max_duration_in_seconds is not None:
															__magic_name__					:							Any		   =							len(example['''speech''']       ) / example['''sampling_rate''']
										# Normalize and clean up text; order matters!
										__magic_name__					:							List[Any]		   =							orthography.preprocess_for_training(example[data_args.target_text_column]       )
										__magic_name__					:							str		   =							vocabulary_text_cleaner.sub('''''',	UpperCAmelCase       )
										if updated_text != example[data_args.target_text_column]:
															text_updates.append((example[data_args.target_text_column], updated_text)       )
															__magic_name__					:							List[str]		   =							updated_text
										return example

					__magic_name__					:							Optional[int]		   =							train_dataset.map(UpperCAmelCase,	remove_columns=[data_args.speech_file_column]       )
					__magic_name__					:							Optional[Any]		   =							val_dataset.map(UpperCAmelCase,	remove_columns=[data_args.speech_file_column]       )

					if data_args.max_duration_in_seconds is not None:

										def filter_by_max_duration(UpperCAmelCase       ):
															return example["duration_in_seconds"] <= data_args.max_duration_in_seconds

										__magic_name__					:							Optional[int]		   =							len(UpperCAmelCase       )
										__magic_name__					:							Optional[int]		   =							len(UpperCAmelCase       )
										__magic_name__					:							Tuple		   =							train_dataset.filter(UpperCAmelCase,	remove_columns=['''duration_in_seconds''']       )
										__magic_name__					:							List[str]		   =							val_dataset.filter(UpperCAmelCase,	remove_columns=['''duration_in_seconds''']       )
										if len(UpperCAmelCase       ) > old_train_size:
															logger.warning(
															    F'''Filtered out {len(UpperCAmelCase       ) - old_train_size} train example(s) longer than'''
															    F''' {data_args.max_duration_in_seconds} second(s).'''       )
										if len(UpperCAmelCase       ) > old_val_size:
															logger.warning(
															    F'''Filtered out {len(UpperCAmelCase       ) - old_val_size} validation example(s) longer than'''
															    F''' {data_args.max_duration_in_seconds} second(s).'''       )
					logger.info(F'''Split sizes: {len(UpperCAmelCase       )} train and {len(UpperCAmelCase       )} validation.'''       )

					logger.warning(F'''Updated {len(UpperCAmelCase       )} transcript(s) using \'{data_args.orthography}\' orthography rules.'''       )
					if logger.isEnabledFor(logging.DEBUG       ):
										for original_text, updated_text in text_updates:
															logger.debug(F'''Updated text: "{original_text}" -> "{updated_text}"'''       )
					__magic_name__					:							List[str]		   =							None

					def prepare_dataset(UpperCAmelCase       ):
										# check that all files have the correct sampling rate
										assert (
										    len(set(batch['''sampling_rate''']       )       ) == 1
										), F'''Make sure all inputs have the same sampling rate of {processor.feature_extractor.sampling_rate}.'''

										__magic_name__					:							Optional[int]		   =							processor(
										    audio=batch['''speech'''],	text=batch[data_args.target_text_column],	sampling_rate=batch['''sampling_rate'''][0]       )
										batch.update(UpperCAmelCase       )
										return batch

					__magic_name__					:							List[Any]		   =							train_dataset.map(
					    UpperCAmelCase,	batch_size=training_args.per_device_train_batch_size,	batched=UpperCAmelCase,	num_proc=data_args.preprocessing_num_workers,	)
					__magic_name__					:							Union[str, Any]		   =							val_dataset.map(
					    UpperCAmelCase,	batch_size=training_args.per_device_train_batch_size,	batched=UpperCAmelCase,	num_proc=data_args.preprocessing_num_workers,	)

					__magic_name__					:							str		   =							DataCollatorCTCWithPadding(processor=UpperCAmelCase,	padding=UpperCAmelCase       )

					def compute_metrics(UpperCAmelCase       ):
										__magic_name__					:							Union[str, Any]		   =							pred.predictions
										__magic_name__					:							List[Any]		   =							np.argmax(UpperCAmelCase,	axis=-1       )

										__magic_name__					:							int		   =							processor.tokenizer.pad_token_id

										__magic_name__					:							Optional[Any]		   =							processor.batch_decode(UpperCAmelCase       )
										# we do not want to group tokens when computing the metrics
										__magic_name__					:							Dict		   =							processor.batch_decode(pred.label_ids,	group_tokens=UpperCAmelCase       )
										if logger.isEnabledFor(logging.DEBUG       ):
															for reference, predicted in zip(UpperCAmelCase,	UpperCAmelCase       ):
																				logger.debug(F'''reference: "{reference}"'''       )
																				logger.debug(F'''predicted: "{predicted}"'''       )
																				if orthography.untransliterator is not None:
																									logger.debug(F'''reference (untransliterated): "{orthography.untransliterator(UpperCAmelCase       )}"'''       )
																									logger.debug(F'''predicted (untransliterated): "{orthography.untransliterator(UpperCAmelCase       )}"'''       )

										__magic_name__					:							Dict		   =							wer_metric.compute(predictions=UpperCAmelCase,	references=UpperCAmelCase       )

										return {"wer": wer}

					if model_args.freeze_feature_extractor:
										model.freeze_feature_extractor()

					__magic_name__					:							Optional[int]		   =							CTCTrainer(
					    model=UpperCAmelCase,	data_collator=UpperCAmelCase,	args=UpperCAmelCase,	compute_metrics=UpperCAmelCase,	train_dataset=UpperCAmelCase,	eval_dataset=UpperCAmelCase,	tokenizer=processor.feature_extractor,	)

					trainer.train()


if __name__ == "__main__":
				main()









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

import sentencepiece

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


lowercase_		=							logging.get_logger(__name__)

lowercase_		=							{
    '''source_spm''': '''source.spm''',
    '''target_spm''': '''target.spm''',
    '''vocab''': '''vocab.json''',
    '''target_vocab_file''': '''target_vocab.json''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

lowercase_		=							{
    '''source_spm''': {
        '''Helsinki-NLP/opus-mt-en-de''': '''https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/source.spm'''
    },
    '''target_spm''': {
        '''Helsinki-NLP/opus-mt-en-de''': '''https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/target.spm'''
    },
    '''vocab''': {
        '''Helsinki-NLP/opus-mt-en-de''': '''https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/vocab.json'''
    },
    '''tokenizer_config_file''': {
        '''Helsinki-NLP/opus-mt-en-de''': (
            '''https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/tokenizer_config.json'''
        )
    },
}

lowercase_		=							{'''Helsinki-NLP/opus-mt-en-de''': 512}
lowercase_		=							{}





class 		A__				(   __SCREAMING_SNAKE_CASE ):
	lowerCamelCase__		: int				    =VOCAB_FILES_NAMES
	lowerCamelCase__		: Any				    =PRETRAINED_VOCAB_FILES_MAP
	lowerCamelCase__		: Optional[Any]				    =PRETRAINED_INIT_CONFIGURATION
	lowerCamelCase__		: Any				    =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowerCamelCase__		: Optional[int]				    =["input_ids", "attention_mask"]
	lowerCamelCase__		: Tuple				    =re.compile(">>.+<<" )  # type: re.Pattern






	def __init__(	self				,     lowerCamelCase				,     lowerCamelCase				,     lowerCamelCase				,     lowerCamelCase=None				,     lowerCamelCase=None				,     lowerCamelCase=None				,     lowerCamelCase="<unk>"				,     lowerCamelCase="</s>"				,     lowerCamelCase="<pad>"				,     lowerCamelCase=512				,     lowerCamelCase = None				,     lowerCamelCase=False				,     **lowerCamelCase				,     )      ->   None:




						"""simple docstring"""






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

						super().__init__(
						    # bos_token=bos_token,  unused. Start decoding with config.decoder_start_token_id
						    source_lang=lowerCamelCase				,     target_lang=lowerCamelCase				,     unk_token=lowerCamelCase				,     eos_token=lowerCamelCase				,     pad_token=lowerCamelCase				,     model_max_length=lowerCamelCase				,     sp_model_kwargs=self.sp_model_kwargs				,     target_vocab_file=lowerCamelCase				,     separate_vocabs=lowerCamelCase				,     **lowerCamelCase				,     )
						assert Path(lowerCamelCase						).exists(), F'''cannot find spm source {source_spm}'''

						__magic_name__					:							Optional[Any]		   =							separate_vocabs
						__magic_name__					:							int		   =							load_json(lowerCamelCase						)
						if self.unk_token not in self.encoder:
											raise KeyError('''<unk> token must be in vocab'''						)
						assert self.pad_token in self.encoder

						if separate_vocabs:
											__magic_name__					:							int		   =							load_json(lowerCamelCase						)
											__magic_name__					:							Tuple		   =							{v: k for k, v in self.target_encoder.items()}
											__magic_name__					:							Optional[Any]		   =							[]
						else:
											__magic_name__					:							Dict		   =							{v: k for k, v in self.encoder.items()}
											__magic_name__					:							list		   =							[k for k in self.encoder if k.startswith('''>>'''						) and k.endswith('''<<'''						)]

						__magic_name__					:							Optional[int]		   =							source_lang
						__magic_name__					:							List[Any]		   =							target_lang
						__magic_name__					:							Tuple		   =							[source_spm, target_spm]

						# load SentencePiece model for pre-processing
						__magic_name__					:							str		   =							load_spm(lowerCamelCase				,     self.sp_model_kwargs						)
						__magic_name__					:							int		   =							load_spm(lowerCamelCase				,     self.sp_model_kwargs						)
						__magic_name__					:							List[str]		   =							self.spm_source
						__magic_name__					:							int		   =							self.encoder

						# Multilingual target side: default to using first supported language code.

						self._setup_normalizer()






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




						"""simple docstring"""






						try:
											from sacremoses import MosesPunctNormalizer

											__magic_name__					:							List[str]		   =							MosesPunctNormalizer(self.source_lang						).normalize
						except (ImportError, FileNotFoundError):
											warnings.warn('''Recommended: pip install sacremoses.'''						)
											__magic_name__					:							Dict		   =							lambda lowerCamelCase						: x






	def  lowercase      (	self				,     lowerCamelCase						)      ->   str:




						"""simple docstring"""






						return self.punc_normalizer(lowerCamelCase						) if x else ""






	def  lowercase      (	self				,     lowerCamelCase						)      ->   str:




						"""simple docstring"""






						return self.current_encoder.get(lowerCamelCase				,     self.current_encoder[self.unk_token]						)






	def  lowercase      (	self				,     lowerCamelCase						)      ->   List[str]:




						"""simple docstring"""






						__magic_name__					:							int		   =							self.language_code_re.match(lowerCamelCase						)
						__magic_name__					:							list		   =							[match.group(0						)] if match else []
						return code, self.language_code_re.sub(''''''				,     lowerCamelCase						)






	def  lowercase      (	self				,     lowerCamelCase						)      ->   List[str]:




						"""simple docstring"""






						__magic_name__   , __magic_name__					:							Tuple		   =							self.remove_language_code(lowerCamelCase						)
						__magic_name__					:							Any		   =							self.current_spm.encode(lowerCamelCase				,     out_type=lowerCamelCase						)
						return code + pieces






	def  lowercase      (	self				,     lowerCamelCase						)      ->   str:




						"""simple docstring"""






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






	def  lowercase      (	self				,     lowerCamelCase				,     **lowerCamelCase						)      ->   Dict:




						"""simple docstring"""






						return super().batch_decode(lowerCamelCase				,     **lowerCamelCase						)






	def  lowercase      (	self				,     lowerCamelCase				,     **lowerCamelCase						)      ->   Any:




						"""simple docstring"""






						return super().decode(lowerCamelCase				,     **lowerCamelCase						)






	def  lowercase      (	self				,     lowerCamelCase						)      ->   str:




						"""simple docstring"""






						__magic_name__					:							int		   =							self.spm_source if self._decode_use_source_tokenizer else self.spm_target
						__magic_name__					:							List[str]		   =							[]
						__magic_name__					:							str		   =							''''''
						for token in tokens:
											# make sure that special tokens are not decoded using sentencepiece model
											if token in self.all_special_tokens:
																out_string += sp_model.decode_pieces(lowerCamelCase						) + token + " "
																__magic_name__					:							Dict		   =							[]
											else:
																current_sub_tokens.append(lowerCamelCase						)
						out_string += sp_model.decode_pieces(lowerCamelCase						)
						return out_string.strip()






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




						"""simple docstring"""






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






	def  lowercase      (	self						)      ->   List[Any]:




						"""simple docstring"""






						__magic_name__					:							List[Any]		   =							self.spm_source
						__magic_name__					:							Tuple		   =							self.encoder






	def  lowercase      (	self						)      ->   Optional[int]:




						"""simple docstring"""






						__magic_name__					:							int		   =							self.spm_target
						if self.separate_vocabs:
											__magic_name__					:							List[str]		   =							self.target_encoder






	@property
	def  lowercase      (	self						)      ->   int:




						"""simple docstring"""






						return len(self.encoder						)






	def  lowercase      (	self				,     lowerCamelCase				,     lowerCamelCase = None						)      ->   Tuple[str]:




						"""simple docstring"""






						if not os.path.isdir(lowerCamelCase						):
											logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''						)
											return
						__magic_name__					:							Any		   =							[]

						if self.separate_vocabs:
											__magic_name__					:							Optional[Any]		   =							os.path.join(
											    lowerCamelCase				,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab''']				,     )
											__magic_name__					:							Dict		   =							os.path.join(
											    lowerCamelCase				,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''target_vocab_file''']				,     )
											save_json(self.encoder				,     lowerCamelCase						)
											save_json(self.target_encoder				,     lowerCamelCase						)
											saved_files.append(lowerCamelCase						)
											saved_files.append(lowerCamelCase						)
						else:
											__magic_name__					:							List[str]		   =							os.path.join(
											    lowerCamelCase				,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab''']						)
											save_json(self.encoder				,     lowerCamelCase						)
											saved_files.append(lowerCamelCase						)

						for spm_save_filename, spm_orig_path, spm_model in zip(
						    [VOCAB_FILES_NAMES['''source_spm'''], VOCAB_FILES_NAMES['''target_spm''']]				,     self.spm_files				,     [self.spm_source, self.spm_target]				,     ):
											__magic_name__					:							List[Any]		   =							os.path.join(
											    lowerCamelCase				,     (filename_prefix + '''-''' if filename_prefix else '''''') + spm_save_filename						)
											if os.path.abspath(lowerCamelCase						) != os.path.abspath(lowerCamelCase						) and os.path.isfile(lowerCamelCase						):
																copyfile(lowerCamelCase				,     lowerCamelCase						)
																saved_files.append(lowerCamelCase						)
											elif not os.path.isfile(lowerCamelCase						):
																with open(lowerCamelCase				,     '''wb'''						) as fi:
																					__magic_name__					:							int		   =							spm_model.serialized_model_proto()
																					fi.write(lowerCamelCase						)
																saved_files.append(lowerCamelCase						)

						return tuple(lowerCamelCase						)






	def  lowercase      (	self						)      ->   Dict:




						"""simple docstring"""






						return self.get_src_vocab()






	def  lowercase      (	self						)      ->   Dict:




						"""simple docstring"""






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






	def  lowercase      (	self						)      ->   Optional[int]:




						"""simple docstring"""






						return dict(self.target_encoder				,     **self.added_tokens_decoder						)






	def __getstate__(	self						)      ->   Dict:




						"""simple docstring"""






						__magic_name__					:							Union[str, Any]		   =							self.__dict__.copy()
						state.update(
						    {k: None for k in ['''spm_source''', '''spm_target''', '''current_spm''', '''punc_normalizer''', '''target_vocab_file''']}						)
						return state






	def __setstate__(	self				,     lowerCamelCase						)      ->   None:




						"""simple docstring"""






						__magic_name__					:							List[str]		   =							d

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

						__magic_name__   , __magic_name__					:							Optional[int]		   =							(load_spm(lowerCamelCase				,     self.sp_model_kwargs						) for f in self.spm_files)
						__magic_name__					:							str		   =							self.spm_source
						self._setup_normalizer()






	def  lowercase      (	self				,     *lowerCamelCase				,     **lowerCamelCase						)      ->   Optional[Any]:




						"""simple docstring"""






						return 1






	def  lowercase      (	self				,     lowerCamelCase						)      ->   int:




						"""simple docstring"""






						__magic_name__					:							Optional[int]		   =							set(self.all_special_ids						)  # call it once instead of inside list comp
						all_special_ids.remove(self.unk_token_id						)  # <unk> is only sometimes special
						return [1 if x in all_special_ids else 0 for x in seq]






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




						"""simple docstring"""






						if already_has_special_tokens:
											return self._special_token_mask(lowerCamelCase						)
						elif token_ids_a is None:
											return self._special_token_mask(lowerCamelCase						) + [1]
						else:
											return self._special_token_mask(token_ids_a + token_ids_a						) + [1]







def  lowerCAmelCase  (	UpperCAmelCase,	UpperCAmelCase       )  ->sentencepiece.SentencePieceProcessor:






					"""simple docstring"""






					__magic_name__					:							int		   =							sentencepiece.SentencePieceProcessor(**UpperCAmelCase       )
					spm.Load(UpperCAmelCase       )
					return spm







def  lowerCAmelCase  (	UpperCAmelCase,	UpperCAmelCase       )  ->None:






					"""simple docstring"""






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







def  lowerCAmelCase  (	UpperCAmelCase       )  ->Union[Dict, List]:






					"""simple docstring"""






					with open(UpperCAmelCase,	'''r'''       ) as f:
										return json.load(UpperCAmelCase       )



