



"""simple docstring"""

import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


a_			   =					logging.getLogger(__name__)

if is_torch_tpu_available(check_device=False):
			import torch_xla.core.xla_model as xm
			import torch_xla.debug.metrics as met







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						def __init__(self   ,		*lowercase__   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		**lowercase__       ):
											super().__init__(*lowercase__   ,		**lowercase__       )
											snake_case_					:		List[Any] 					=					eval_examples
											snake_case_					:		Tuple 					=					post_process_function
											snake_case_					:		str 					=					quant_trainer_args
											snake_case_					:		int 					=					1_28  # default number of calibration samples







						def 						__UpperCamelCase							(self   ,		lowercase__=None       ):
											if calib_dataset is None and self.calib_dataset is None:
																raise ValueError("""Trainer: calibration requires an calib_dataset."""       )
											snake_case_					:		List[Any] 					=					calib_dataset if calib_dataset is not None else self.calib_dataset

											snake_case_					:		str 					=					self._remove_unused_columns(lowercase__   ,		description="""Calibration"""       )

											return DataLoader(
											    lowercase__   ,		batch_size=self.args.eval_batch_size   ,		collate_fn=self.data_collator   ,		drop_last=self.args.dataloader_drop_last   ,		num_workers=self.args.dataloader_num_workers   ,		pin_memory=self.args.dataloader_pin_memory   ,		shuffle=lowercase__   ,		)







						def 						__UpperCamelCase							(self   ,		lowercase__=None       ):
											snake_case_					:		List[Any] 					=					self.train_dataset if calib_dataset is None else calib_dataset
											snake_case_					:		Optional[Any] 					=					self.get_calib_dataloader(lowercase__       )

											snake_case_					:		List[str] 					=					self.model
											quant_trainer.configure_model(lowercase__   ,		self.quant_trainer_args   ,		calib=lowercase__       )
											model.eval()
											quant_trainer.enable_calibration(lowercase__       )

											logger.info("""***** Running calibration *****"""       )
											logger.info(f'  Num examples = {self.calib_num}'       )
											logger.info(f'  Batch size = {calib_dataloader.batch_size}'       )

											for step, inputs in enumerate(lowercase__       ):
																# Prediction step
																snake_case_					,			snake_case_					,			snake_case_					:		Union[str, Any] 					=					self.prediction_step(lowercase__   ,		lowercase__   ,		prediction_loss_only=lowercase__       )
																if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
																					break

											quant_trainer.finish_calibration(lowercase__   ,		self.quant_trainer_args       )
											snake_case_					:		int 					=					model







						def 						__UpperCamelCase							(self   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__ = "eval"       ):
											snake_case_					:		Optional[Any] 					=					self.eval_dataset if eval_dataset is None else eval_dataset
											snake_case_					:		Any 					=					self.get_eval_dataloader(lowercase__       )
											snake_case_					:		str 					=					self.eval_examples if eval_examples is None else eval_examples

											# Temporarily disable metric computation, we will do it in the loop here.
											snake_case_					:		Tuple 					=					self.compute_metrics
											snake_case_					:		Union[str, Any] 					=					None
											snake_case_					:		Optional[int] 					=					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																snake_case_					:		List[Any] 					=					eval_loop(
																    lowercase__   ,		description="""Evaluation"""   ,		prediction_loss_only=True if compute_metrics is None else None   ,		ignore_keys=lowercase__   ,		)
											finally:
																snake_case_					:		Optional[int] 					=					compute_metrics

											if self.post_process_function is not None and self.compute_metrics is not None:
																snake_case_					:		Dict 					=					self.post_process_function(lowercase__   ,		lowercase__   ,		output.predictions       )
																snake_case_					:		int 					=					self.compute_metrics(lowercase__       )

																# Prefix all keys with metric_key_prefix + '_'
																for key in list(metrics.keys()       ):
																					if not key.startswith(f'{metric_key_prefix}_'       ):
																										snake_case_					:		int 					=					metrics.pop(lowercase__       )

																self.log(lowercase__       )
											else:
																snake_case_					:		Union[str, Any] 					=					{}

											if self.args.tpu_metrics_debug or self.args.debug:
																# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
																xm.master_print(met.metrics_report()       )

											snake_case_					:		Union[str, Any] 					=					self.callback_handler.on_evaluate(self.args   ,		self.state   ,		self.control   ,		lowercase__       )
											return metrics







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__   ,		lowercase__=None   ,		lowercase__ = "test"       ):
											snake_case_					:		str 					=					self.get_test_dataloader(lowercase__       )

											# Temporarily disable metric computation, we will do it in the loop here.
											snake_case_					:		Optional[Any] 					=					self.compute_metrics
											snake_case_					:		Any 					=					None
											snake_case_					:		List[str] 					=					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																snake_case_					:		Any 					=					eval_loop(
																    lowercase__   ,		description="""Prediction"""   ,		prediction_loss_only=True if compute_metrics is None else None   ,		ignore_keys=lowercase__   ,		)
											finally:
																snake_case_					:		Optional[Any] 					=					compute_metrics

											if self.post_process_function is None or self.compute_metrics is None:
																return output

											snake_case_					:		Optional[Any] 					=					self.post_process_function(lowercase__   ,		lowercase__   ,		output.predictions   ,		"""predict"""       )
											snake_case_					:		Optional[int] 					=					self.compute_metrics(lowercase__       )

											# Prefix all keys with metric_key_prefix + '_'
											for key in list(metrics.keys()       ):
																if not key.startswith(f'{metric_key_prefix}_'       ):
																					snake_case_					:		Optional[int] 					=					metrics.pop(lowercase__       )

											return PredictionOutput(predictions=predictions.predictions   ,		label_ids=predictions.label_ids   ,		metrics=lowercase__       )







						def 						__UpperCamelCase							(self   ,		lowercase__="./"       ):
											snake_case_					:		Any 					=					self.eval_dataset
											snake_case_					:		Dict 					=					self.get_eval_dataloader(lowercase__       )

											snake_case_					:		str 					=					next(iter(lowercase__       )       )

											# saving device - to make it consistent
											snake_case_					:		List[str] 					=					torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""       )

											# convert to tuple
											snake_case_					:		Any 					=					tuple(v.to(lowercase__       ) for k, v in batch.items()       )

											logger.info("""Converting model to be onnx compatible"""       )
											from pytorch_quantization.nn import TensorQuantizer

											snake_case_					:		Optional[Any] 					=					True

											snake_case_					:		Union[str, Any] 					=					self.model.to(lowercase__       )

											model.eval()
											model.float()

											snake_case_					:		List[str] 					=					model.module if hasattr(lowercase__   ,		"""module"""       ) else model
											quant_trainer.configure_model(lowercase__   ,		self.quant_trainer_args       )

											snake_case_					:		Dict 					=					os.path.join(lowercase__   ,		"""model.onnx"""       )
											logger.info(f'exporting model to {output_model_file}'       )

											snake_case_					:		Union[str, Any] 					=					{0: """batch_size""", 1: """seq_len"""}

											torch.onnx.export(
											    lowercase__   ,		lowercase__   ,		lowercase__   ,		export_params=lowercase__   ,		opset_version=13   ,		do_constant_folding=lowercase__   ,		input_names=["""input_ids""", """attention_mask""", """token_type_ids"""]   ,		output_names=["""output_start_logits""", """output_end_logits"""]   ,		dynamic_axes={
											        """input_ids""": axes,
											        """attention_mask""": axes,
											        """token_type_ids""": axes,
											        """output_start_logits""": axes,
											        """output_end_logits""": axes,
											    }   ,		verbose=lowercase__   ,		)
											logger.info("""onnx export finished"""       )









"""simple docstring"""

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

from tokenizers import processors

from ...tokenization_utils import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
			from .tokenization_mbart import MBartTokenizer
else:
			a_			   =					None


a_			   =					logging.get_logger(__name__)


a_			   =					{'''vocab_file''': '''sentencepiece.bpe.model''', '''tokenizer_file''': '''tokenizer.json'''}

a_			   =					{
    '''vocab_file''': {
        '''facebook/mbart-large-en-ro''': (
            '''https://huggingface.co/facebook/mbart-large-en-ro/resolve/main/sentencepiece.bpe.model'''
        ),
        '''facebook/mbart-large-cc25''': (
            '''https://huggingface.co/facebook/mbart-large-cc25/resolve/main/sentencepiece.bpe.model'''
        ),
    },
    '''tokenizer_file''': {
        '''facebook/mbart-large-en-ro''': '''https://huggingface.co/facebook/mbart-large-en-ro/resolve/main/tokenizer.json''',
        '''facebook/mbart-large-cc25''': '''https://huggingface.co/facebook/mbart-large-cc25/resolve/main/tokenizer.json''',
    },
}

a_			   =					{
    '''facebook/mbart-large-en-ro''': 1024,
    '''facebook/mbart-large-cc25''': 1024,
}

# fmt: off
a_			   =					['''ar_AR''', '''cs_CZ''', '''de_DE''', '''en_XX''', '''es_XX''', '''et_EE''', '''fi_FI''', '''fr_XX''', '''gu_IN''', '''hi_IN''', '''it_IT''', '''ja_XX''', '''kk_KZ''', '''ko_KR''', '''lt_LT''', '''lv_LV''', '''my_MM''', '''ne_NP''', '''nl_XX''', '''ro_RO''', '''ru_RU''', '''si_LK''', '''tr_TR''', '''vi_VN''', '''zh_CN''']







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						_A		:							Dict							       =		VOCAB_FILES_NAMES
						_A		:							Union[str, Any]							       =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						_A		:							Optional[int]							       =		PRETRAINED_VOCAB_FILES_MAP
						_A		:							str							       =		["""input_ids""", """attention_mask"""]
						_A		:							Tuple							       =		MBartTokenizer

						_A		:							List[int]							       =		[]
						_A		:							List[int]							       =		[]







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

											super().__init__(
											    vocab_file=lowercase__   ,		tokenizer_file=lowercase__   ,		bos_token=lowercase__   ,		eos_token=lowercase__   ,		sep_token=lowercase__   ,		cls_token=lowercase__   ,		unk_token=lowercase__   ,		pad_token=lowercase__   ,		mask_token=lowercase__   ,		src_lang=lowercase__   ,		tgt_lang=lowercase__   ,		additional_special_tokens=lowercase__   ,		**lowercase__   ,		)

											snake_case_					:		Dict 					=					vocab_file
											snake_case_					:		Optional[int] 					=					False if not self.vocab_file else True

											snake_case_					:		Optional[Any] 					=					FAIRSEQ_LANGUAGE_CODES.copy()

											if additional_special_tokens is not None:
																# Only add those special tokens if they are not already there.
																_additional_special_tokens.extend(
																    [t for t in additional_special_tokens if t not in _additional_special_tokens]       )

											self.add_special_tokens({"""additional_special_tokens""": _additional_special_tokens}       )
											snake_case_					:		Any 					=					{
											    lang_code: self.convert_tokens_to_ids(lowercase__       ) for lang_code in FAIRSEQ_LANGUAGE_CODES
											}

											snake_case_					:		Tuple 					=					src_lang if src_lang is not None else """en_XX"""
											snake_case_					:		Tuple 					=					self.convert_tokens_to_ids(self._src_lang       )
											snake_case_					:		Tuple 					=					tgt_lang
											self.set_src_lang_special_tokens(self._src_lang       )







						@property
						def 						__UpperCamelCase							(self       ):
											return self._src_lang







						@src_lang.setter
						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Tuple 					=					new_src_lang
											self.set_src_lang_special_tokens(self._src_lang       )







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = None       ):
											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 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = None       ):
											snake_case_					:		List[Any] 					=					[self.sep_token_id]
											snake_case_					:		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]







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		**lowercase__       ):
											if src_lang is None or tgt_lang is None:
																raise ValueError("""Translation requires a `src_lang` and a `tgt_lang` for this model"""       )
											snake_case_					:		int 					=					src_lang
											snake_case_					:		List[str] 					=					self(lowercase__   ,		add_special_tokens=lowercase__   ,		return_tensors=lowercase__   ,		**lowercase__       )
											snake_case_					:		List[str] 					=					self.convert_tokens_to_ids(lowercase__       )
											snake_case_					:		Union[str, Any] 					=					tgt_lang_id
											return inputs







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = "en_XX"   ,		lowercase__ = None   ,		lowercase__ = "ro_RO"   ,		**lowercase__   ,		):
											snake_case_					:		List[str] 					=					src_lang
											snake_case_					:		int 					=					tgt_lang
											return super().prepare_seqaseq_batch(lowercase__   ,		lowercase__   ,		**lowercase__       )







						def 						__UpperCamelCase							(self       ):
											return self.set_src_lang_special_tokens(self.src_lang       )







						def 						__UpperCamelCase							(self       ):
											return self.set_tgt_lang_special_tokens(self.tgt_lang       )







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		int 					=					self.convert_tokens_to_ids(lowercase__       )
											snake_case_					:		Tuple 					=					[]
											snake_case_					:		List[Any] 					=					[self.eos_token_id, self.cur_lang_code]

											snake_case_					:		List[Any] 					=					self.convert_ids_to_tokens(self.prefix_tokens       )
											snake_case_					:		Union[str, Any] 					=					self.convert_ids_to_tokens(self.suffix_tokens       )

											snake_case_					:		Optional[int] 					=					processors.TemplateProcessing(
											    single=prefix_tokens_str + ["""$A"""] + suffix_tokens_str   ,		pair=prefix_tokens_str + ["""$A""", """$B"""] + suffix_tokens_str   ,		special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str   ,		self.prefix_tokens + self.suffix_tokens       )       )   ,		)







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Tuple 					=					self.convert_tokens_to_ids(lowercase__       )
											snake_case_					:		Optional[int] 					=					[]
											snake_case_					:		Optional[Any] 					=					[self.eos_token_id, self.cur_lang_code]

											snake_case_					:		Union[str, Any] 					=					self.convert_ids_to_tokens(self.prefix_tokens       )
											snake_case_					:		int 					=					self.convert_ids_to_tokens(self.suffix_tokens       )

											snake_case_					:		List[Any] 					=					processors.TemplateProcessing(
											    single=prefix_tokens_str + ["""$A"""] + suffix_tokens_str   ,		pair=prefix_tokens_str + ["""$A""", """$B"""] + suffix_tokens_str   ,		special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str   ,		self.prefix_tokens + self.suffix_tokens       )       )   ,		)







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = None       ):
											if not self.can_save_slow_tokenizer:
																raise ValueError(
																    """Your fast tokenizer does not have the necessary information to save the vocabulary for a slow """
																    """tokenizer."""       )

											if not os.path.isdir(lowercase__       ):
																logger.error(f'Vocabulary path ({save_directory}) should be a directory.'       )
																return
											snake_case_					:		List[str] 					=					os.path.join(
											    lowercase__   ,		(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]       )

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

											return (out_vocab_file,)




