






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

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


if is_sentencepiece_available():
			from .tokenization_xlnet import XLNetTokenizer
else:
			UpperCAmelCase__      			=						None


UpperCAmelCase__      			=						logging.get_logger(__name__)

UpperCAmelCase__      			=						{'''vocab_file''': '''spiece.model''', '''tokenizer_file''': '''tokenizer.json'''}

UpperCAmelCase__      			=						{
    '''vocab_file''': {
        '''xlnet-base-cased''': '''https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model''',
        '''xlnet-large-cased''': '''https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model''',
    },
    '''tokenizer_file''': {
        '''xlnet-base-cased''': '''https://huggingface.co/xlnet-base-cased/resolve/main/tokenizer.json''',
        '''xlnet-large-cased''': '''https://huggingface.co/xlnet-large-cased/resolve/main/tokenizer.json''',
    },
}

UpperCAmelCase__      			=						{
    '''xlnet-base-cased''': None,
    '''xlnet-large-cased''': None,
}

UpperCAmelCase__      			=						'''▁'''

# Segments (not really needed)
UpperCAmelCase__      			=						0
UpperCAmelCase__      			=						1
UpperCAmelCase__      			=						2
UpperCAmelCase__      			=						3
UpperCAmelCase__      			=						4




class 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						snake_case__        =      VOCAB_FILES_NAMES
						snake_case__        =      PRETRAINED_VOCAB_FILES_MAP
						snake_case__        =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						snake_case__        =      """left"""
						snake_case__        =      XLNetTokenizer



						def __init__(self:					List[str] ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					int=None ,					__UpperCAmelCase:					Union[str, Any]=False ,					__UpperCAmelCase:					List[Any]=True ,					__UpperCAmelCase:					str=False ,					__UpperCAmelCase:					int="<s>" ,					__UpperCAmelCase:					int="</s>" ,					__UpperCAmelCase:					Optional[int]="<unk>" ,					__UpperCAmelCase:					Tuple="<sep>" ,					__UpperCAmelCase:					Any="<pad>" ,					__UpperCAmelCase:					Optional[int]="<cls>" ,					__UpperCAmelCase:					int="<mask>" ,					__UpperCAmelCase:					Any=["<eop>", "<eod>"] ,					**__UpperCAmelCase:					Dict ,					)    ->     List[Any]:




											'''simple docstring'''



											__a					:						str      =					AddedToken(__UpperCAmelCase ,					lstrip=__UpperCAmelCase ,					rstrip=__UpperCAmelCase   ) if isinstance(__UpperCAmelCase ,					__UpperCAmelCase   ) else mask_token

											super().__init__(
											    vocab_file=__UpperCAmelCase ,					tokenizer_file=__UpperCAmelCase ,					do_lower_case=__UpperCAmelCase ,					remove_space=__UpperCAmelCase ,					keep_accents=__UpperCAmelCase ,					bos_token=__UpperCAmelCase ,					eos_token=__UpperCAmelCase ,					unk_token=__UpperCAmelCase ,					sep_token=__UpperCAmelCase ,					pad_token=__UpperCAmelCase ,					cls_token=__UpperCAmelCase ,					mask_token=__UpperCAmelCase ,					additional_special_tokens=__UpperCAmelCase ,					**__UpperCAmelCase ,					)

											__a					:						Any      =					3
											__a					:						Optional[Any]      =					do_lower_case
											__a					:						Tuple      =					remove_space
											__a					:						Tuple      =					keep_accents
											__a					:						Dict      =					vocab_file
											__a					:						Tuple      =					False if not self.vocab_file else True



						def        UpperCAmelCase__					(self:					List[Any] ,					__UpperCAmelCase:					List[int] ,					__UpperCAmelCase:					Optional[List[int]] = None   )    ->     List[int]:




											'''simple docstring'''



											__a					:						int      =					[self.sep_token_id]
											__a					:						Any      =					[self.cls_token_id]
											if token_ids_a is None:
																return token_ids_a + sep + cls
											return token_ids_a + sep + token_ids_a + sep + cls



						def        UpperCAmelCase__					(self:					Union[str, Any] ,					__UpperCAmelCase:					List[int] ,					__UpperCAmelCase:					Optional[List[int]] = None   )    ->     List[int]:




											'''simple docstring'''



											__a					:						int      =					[self.sep_token_id]
											__a					:						Union[str, Any]      =					[2]

											if token_ids_a is None:
																return len(token_ids_a + sep   ) * [0] + cls_segment_id
											return len(token_ids_a + sep   ) * [0] + len(token_ids_a + sep   ) * [1] + cls_segment_id



						def        UpperCAmelCase__					(self:					int ,					__UpperCAmelCase:					str ,					__UpperCAmelCase:					Optional[str] = None   )    ->     Tuple[str]:




											'''simple docstring'''



											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(__UpperCAmelCase   ):
																logger.error(f'Vocabulary path ({save_directory}) should be a directory'   )
																return
											__a					:						Optional[int]      =					os.path.join(
											    __UpperCAmelCase ,					(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]   )

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

											return (out_vocab_file,)













import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


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




class 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						def __init__(self:					str ,					*__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Dict=None ,					**__UpperCAmelCase:					Tuple   )    ->     Dict:




											'''simple docstring'''



											super().__init__(*__UpperCAmelCase ,					**__UpperCAmelCase   )
											__a					:						str      =					eval_examples
											__a					:						List[Any]      =					post_process_function



						def        UpperCAmelCase__					(self:					Optional[int] ,					__UpperCAmelCase:					Optional[Dataset] = None ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Optional[List[str]] = None ,					__UpperCAmelCase:					str = "eval" ,					**__UpperCAmelCase:					Any ,					)    ->     Dict[str, float]:




											'''simple docstring'''



											__a					:						Optional[int]      =					gen_kwargs.copy()
											__a					:						Union[str, Any]      =					(
											    gen_kwargs["max_length"] if gen_kwargs.get("max_length"   ) is not None else self.args.generation_max_length
											)
											__a					:						Dict      =					(
											    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"   ) is not None else self.args.generation_num_beams
											)
											__a					:						Optional[Any]      =					gen_kwargs

											__a					:						Union[str, Any]      =					self.eval_dataset if eval_dataset is None else eval_dataset
											__a					:						Optional[int]      =					self.get_eval_dataloader(__UpperCAmelCase   )
											__a					:						Tuple      =					self.eval_examples if eval_examples is None else eval_examples

											# Temporarily disable metric computation, we will do it in the loop here.
											__a					:						int      =					self.compute_metrics
											__a					:						Dict      =					None
											__a					:						List[Any]      =					time.time()
											__a					:						str      =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																__a					:						str      =					eval_loop(
																    __UpperCAmelCase ,					description="Evaluation" ,					prediction_loss_only=True if compute_metrics is None else None ,					ignore_keys=__UpperCAmelCase ,					metric_key_prefix=__UpperCAmelCase ,					)
											finally:
																__a					:						Union[str, Any]      =					compute_metrics
											__a					:						str      =					self.args.eval_batch_size * self.args.world_size
											if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
											output.metrics.update(
											    speed_metrics(
											        __UpperCAmelCase ,					__UpperCAmelCase ,					num_samples=output.num_samples ,					num_steps=math.ceil(output.num_samples / total_batch_size   ) ,					)   )

											if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
																# Only the main node write the results by default
																__a					:						Union[str, Any]      =					self.post_process_function(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase   )
																__a					:						Dict      =					self.compute_metrics(__UpperCAmelCase   )

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

																metrics.update(output.metrics   )
											else:
																__a					:						Optional[int]      =					output.metrics

											if self.args.should_log:
																# Only the main node log the results by default
																self.log(__UpperCAmelCase   )

											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()   )

											__a					:						Tuple      =					self.callback_handler.on_evaluate(self.args ,					self.state ,					self.control ,					__UpperCAmelCase   )
											return metrics



						def        UpperCAmelCase__					(self:					Dict ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					Union[str, Any]=None ,					__UpperCAmelCase:					str = "test" ,					**__UpperCAmelCase:					List[Any]   )    ->     Union[str, Any]:




											'''simple docstring'''



											__a					:						Any      =					gen_kwargs.copy()

											__a					:						Tuple      =					self.get_test_dataloader(__UpperCAmelCase   )

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

											__a					:						Optional[Any]      =					self.args.eval_batch_size * self.args.world_size
											if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
											output.metrics.update(
											    speed_metrics(
											        __UpperCAmelCase ,					__UpperCAmelCase ,					num_samples=output.num_samples ,					num_steps=math.ceil(output.num_samples / total_batch_size   ) ,					)   )
											if self.post_process_function is None or self.compute_metrics is None:
																return output

											__a					:						Optional[int]      =					self.post_process_function(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					"predict"   )
											__a					:						Dict      =					self.compute_metrics(__UpperCAmelCase   )

											# Prefix all keys with metric_key_prefix + '_'
											for key in list(metrics.keys()   ):
																if not key.startswith(f'{metric_key_prefix}_'   ):
																					__a					:						List[str]      =					metrics.pop(__UpperCAmelCase   )
											metrics.update(output.metrics   )
											return PredictionOutput(predictions=predictions.predictions ,					label_ids=predictions.label_ids ,					metrics=__UpperCAmelCase   )





