






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 							__a							(  _UpperCAmelCase						):





	'''simple docstring'''



	def __init__(						self ,    *UpperCamelCase__ ,    UpperCamelCase__=None ,    UpperCamelCase__=None ,    **UpperCamelCase__       ):
					super().__init__(*lowercase__ ,    **lowercase__       )
					SCREAMING_SNAKE_CASE_   :	List[str]													=   eval_examples
					SCREAMING_SNAKE_CASE_   :	int													=   post_process_function


	def   __snake_case			(						self ,    UpperCamelCase__ = None ,    UpperCamelCase__=None ,    UpperCamelCase__ = None ,    UpperCamelCase__ = "eval" ,    **UpperCamelCase__ ,    ):
					SCREAMING_SNAKE_CASE_   :	Any													=   gen_kwargs.copy()
					SCREAMING_SNAKE_CASE_   :	List[str]													=   (
					    gen_kwargs["max_length"] if gen_kwargs.get('max_length'       ) is not None else self.args.generation_max_length
					)
					SCREAMING_SNAKE_CASE_   :	Dict													=   (
					    gen_kwargs["num_beams"] if gen_kwargs.get('num_beams'       ) is not None else self.args.generation_num_beams
					)
					SCREAMING_SNAKE_CASE_   :	Optional[int]													=   gen_kwargs

					SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   self.eval_dataset if eval_dataset is None else eval_dataset
					SCREAMING_SNAKE_CASE_   :	List[str]													=   self.get_eval_dataloader(lowercase__       )
					SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   self.eval_examples if eval_examples is None else eval_examples

					# Temporarily disable metric computation, we will do it in the loop here.
					SCREAMING_SNAKE_CASE_   :	int													=   self.compute_metrics
					SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   None
					SCREAMING_SNAKE_CASE_   :	List[str]													=   time.time()
					SCREAMING_SNAKE_CASE_   :	str													=   self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					try:
									SCREAMING_SNAKE_CASE_   :	str													=   eval_loop(
									    lowercase__ ,    description='Evaluation' ,    prediction_loss_only=True if compute_metrics is None else None ,    ignore_keys=lowercase__ ,    metric_key_prefix=lowercase__ ,    )
					finally:
									SCREAMING_SNAKE_CASE_   :	List[str]													=   compute_metrics
					SCREAMING_SNAKE_CASE_   :	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(
					        lowercase__ ,    lowercase__ ,    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
									SCREAMING_SNAKE_CASE_   :	Any													=   self.post_process_function(lowercase__ ,    lowercase__ ,    lowercase__       )
									SCREAMING_SNAKE_CASE_   :	str													=   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}_'''       ):
																	SCREAMING_SNAKE_CASE_   :	Tuple													=   metrics.pop(lowercase__       )

									metrics.update(output.metrics       )
					else:
									SCREAMING_SNAKE_CASE_   :	Tuple													=   output.metrics

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

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

					SCREAMING_SNAKE_CASE_   :	Optional[int]													=   self.callback_handler.on_evaluate(self.args ,    self.state ,    self.control ,    lowercase__       )
					return metrics






	def   __snake_case			(						self ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__=None ,    UpperCamelCase__ = "test" ,    **UpperCamelCase__       ):
					SCREAMING_SNAKE_CASE_   :	Dict													=   gen_kwargs.copy()

					SCREAMING_SNAKE_CASE_   :	List[str]													=   self.get_test_dataloader(lowercase__       )

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

					SCREAMING_SNAKE_CASE_   :	Tuple													=   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(
					        lowercase__ ,    lowercase__ ,    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

					SCREAMING_SNAKE_CASE_   :	List[Any]													=   self.post_process_function(lowercase__ ,    lowercase__ ,    lowercase__ ,    'predict'       )
					SCREAMING_SNAKE_CASE_   :	Dict													=   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}_'''       ):
													SCREAMING_SNAKE_CASE_   :	int													=   metrics.pop(lowercase__       )
					metrics.update(output.metrics       )
					return PredictionOutput(predictions=predictions.predictions ,    label_ids=predictions.label_ids ,    metrics=lowercase__       )







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

import sentencepiece as spm

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


A        =			logging.get_logger(__name__)

A        =			{'vocab_file': 'spm_char.model'}

A        =			{
    'vocab_file': {
        'microsoft/speecht5_asr': 'https://huggingface.co/microsoft/speecht5_asr/resolve/main/spm_char.model',
        'microsoft/speecht5_tts': 'https://huggingface.co/microsoft/speecht5_tts/resolve/main/spm_char.model',
        'microsoft/speecht5_vc': 'https://huggingface.co/microsoft/speecht5_vc/resolve/main/spm_char.model',
    }
}

A        =			{
    'microsoft/speecht5_asr': 1024,
    'microsoft/speecht5_tts': 1024,
    'microsoft/speecht5_vc': 1024,
}


class 							__a							(  __A						):





		'''simple docstring'''



		UpperCAmelCase__      :				str				    =						VOCAB_FILES_NAMES
		UpperCAmelCase__      :				List[str]				    =						PRETRAINED_VOCAB_FILES_MAP
		UpperCAmelCase__      :				Optional[int]				    =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		UpperCAmelCase__      :				Dict				    =						["""input_ids""", """attention_mask"""]


		def __init__(						self ,    UpperCamelCase__ ,    UpperCamelCase__="<s>" ,    UpperCamelCase__="</s>" ,    UpperCamelCase__="<unk>" ,    UpperCamelCase__="<pad>" ,    UpperCamelCase__ = None ,    **UpperCamelCase__ ,    ):
						SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   {} if sp_model_kwargs is None else sp_model_kwargs

						super().__init__(
						    bos_token=UpperCamelCase__ ,    eos_token=UpperCamelCase__ ,    unk_token=UpperCamelCase__ ,    pad_token=UpperCamelCase__ ,    sp_model_kwargs=self.sp_model_kwargs ,    **UpperCamelCase__ ,    )

						SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   vocab_file

						SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   spm.SentencePieceProcessor(**self.sp_model_kwargs       )
						self.sp_model.Load(UpperCamelCase__       )


		@property
		def   __snake_case			(						self       ):
						return self.sp_model.get_piece_size()


		def   __snake_case			(						self       ):
						SCREAMING_SNAKE_CASE_   :	List[str]													=   {self.convert_ids_to_tokens(UpperCamelCase__       ): i for i in range(self.vocab_size       )}
						vocab.update(self.added_tokens_encoder       )
						return vocab


		def __getstate__(						self       ):
						SCREAMING_SNAKE_CASE_   :	str													=   self.__dict__.copy()
						SCREAMING_SNAKE_CASE_   :	List[str]													=   None
						return state


		def __setstate__(						self ,    UpperCamelCase__       ):
						SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   d

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

						SCREAMING_SNAKE_CASE_   :	List[Any]													=   spm.SentencePieceProcessor(**self.sp_model_kwargs       )
						self.sp_model.Load(self.vocab_file       )


		def   __snake_case			(						self ,    UpperCamelCase__       ):
						return self.sp_model.encode(UpperCamelCase__ ,    out_type=UpperCamelCase__       )


		def   __snake_case			(						self ,    UpperCamelCase__       ):
						return self.sp_model.piece_to_id(UpperCamelCase__       )


		def   __snake_case			(						self ,    UpperCamelCase__       ):
						SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   self.sp_model.IdToPiece(UpperCamelCase__       )
						return token


		def   __snake_case			(						self ,    UpperCamelCase__       ):
						SCREAMING_SNAKE_CASE_   :	int													=   []
						SCREAMING_SNAKE_CASE_   :	int													=   ''
						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(UpperCamelCase__       ) + token
														SCREAMING_SNAKE_CASE_   :	Any													=   []
										else:
														current_sub_tokens.append(UpperCamelCase__       )
						out_string += self.sp_model.decode(UpperCamelCase__       )
						return out_string.strip()


		def   __snake_case			(						self ,    UpperCamelCase__ ,    UpperCamelCase__=None       ):
						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   __snake_case			(						self ,    UpperCamelCase__ ,    UpperCamelCase__ = None ,    UpperCamelCase__ = False       ):
						if already_has_special_tokens:
										return super().get_special_tokens_mask(
										    token_ids_a=UpperCamelCase__ ,    token_ids_a=UpperCamelCase__ ,    already_has_special_tokens=UpperCamelCase__       )

						SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   [1]
						if token_ids_a is None:
										return ([0] * len(UpperCamelCase__       )) + suffix_ones
						return ([0] * len(UpperCamelCase__       )) + ([0] * len(UpperCamelCase__       )) + suffix_ones






		def   __snake_case			(						self ,    UpperCamelCase__ ,    UpperCamelCase__ = None       ):
						if not os.path.isdir(UpperCamelCase__       ):
										logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''       )
										return
						SCREAMING_SNAKE_CASE_   :	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__       ) and os.path.isfile(self.vocab_file       ):
										copyfile(self.vocab_file ,    UpperCamelCase__       )
						elif not os.path.isfile(self.vocab_file       ):
										with open(UpperCamelCase__ ,    'wb'       ) as fi:
														SCREAMING_SNAKE_CASE_   :	Tuple													=   self.sp_model.serialized_model_proto()
														fi.write(UpperCamelCase__       )

						return (out_vocab_file,)