


from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
   from ..models.auto import AutoTokenizer


class 			__lowerCAmelCase   :






  """simple docstring"""





  def     lowerCAmelCase__     (   self				:				List[Any]			, _lowerCAmelCase				:				str     )      ->       List[Any]:




     """simple docstring"""
     raise NotImplementedError()



  def     lowerCAmelCase__     (   self				:				Optional[int]     )      ->       Tuple:




     """simple docstring"""
     raise NotImplementedError()






class 			__lowerCAmelCase   (       snake_case_    ):






  """simple docstring"""





  def __init__(   self				:				Union[str, Any]			, _lowerCAmelCase				:				Union[str, Any]			, _lowerCAmelCase				:				Tuple = False			, **_lowerCAmelCase				:				Optional[Any]     )      ->       Tuple:




     """simple docstring"""
     snake_case_    			=    tokenizer
     snake_case_    			=    skip_prompt
     snake_case_    			=    decode_kwargs

     # variables used in the streaming process
     snake_case_    			=    []
     snake_case_    			=    0
     snake_case_    			=    True



  def     lowerCAmelCase__     (   self				:				int			, _lowerCAmelCase				:				List[str]     )      ->       Union[str, Any]:




     """simple docstring"""
     if len(value.shape     ) > 1 and value.shape[0] > 1:
        raise ValueError("TextStreamer only supports batch size 1"     )
     elif len(value.shape     ) > 1:
        snake_case_    			=    value[0]

     if self.skip_prompt and self.next_tokens_are_prompt:
        snake_case_    			=    False
        return

     # Add the new token to the cache and decodes the entire thing.
     self.token_cache.extend(value.tolist()     )
     snake_case_    			=    self.tokenizer.decode(self.token_cache			, **self.decode_kwargs     )

     # After the symbol for a new line, we flush the cache.
     if text.endswith("\n"     ):
        snake_case_    			=    text[self.print_len :]
        snake_case_    			=    []
        snake_case_    			=    0
     # If the last token is a CJK character, we print the characters.
     elif len(_lowerCAmelCase     ) > 0 and self._is_chinese_char(ord(text[-1]     )     ):
        snake_case_    			=    text[self.print_len :]
        self.print_len += len(_lowerCAmelCase     )
     # Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
     # which may change with the subsequent token -- there are probably smarter ways to do this!)
     else:
        snake_case_    			=    text[self.print_len : text.rfind(" "     ) + 1]
        self.print_len += len(_lowerCAmelCase     )

     self.on_finalized_text(_lowerCAmelCase     )



  def     lowerCAmelCase__     (   self				:				Tuple     )      ->       str:




     """simple docstring"""
     # Flush the cache, if it exists
     if len(self.token_cache     ) > 0:
        snake_case_    			=    self.tokenizer.decode(self.token_cache			, **self.decode_kwargs     )
        snake_case_    			=    text[self.print_len :]
        snake_case_    			=    []
        snake_case_    			=    0
     else:
        snake_case_    			=    ""

     snake_case_    			=    True
     self.on_finalized_text(_lowerCAmelCase			, stream_end=_lowerCAmelCase     )



  def     lowerCAmelCase__     (   self				:				int			, _lowerCAmelCase				:				int			, _lowerCAmelCase				:				Any = False     )      ->       List[str]:




     """simple docstring"""
     print(_lowerCAmelCase			, flush=_lowerCAmelCase			, end="" if not stream_end else None     )



  def     lowerCAmelCase__     (   self				:				Dict			, _lowerCAmelCase				:				int     )      ->       str:




     """simple docstring"""
     # This defines a "chinese character" as anything in the CJK Unicode block:
     #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
     #
     # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
     # despite its name. The modern Korean Hangul alphabet is a different block,
     # as is Japanese Hiragana and Katakana. Those alphabets are used to write
     # space-separated words, so they are not treated specially and handled
     # like the all of the other languages.
     if (
         (cp >= 0X4_E_0_0 and cp <= 0X9_F_F_F)
         or (cp >= 0X3_4_0_0 and cp <= 0X4_D_B_F)  #
         or (cp >= 0X2_0_0_0_0 and cp <= 0X2_A_6_D_F)  #
         or (cp >= 0X2_A_7_0_0 and cp <= 0X2_B_7_3_F)  #
         or (cp >= 0X2_B_7_4_0 and cp <= 0X2_B_8_1_F)  #
         or (cp >= 0X2_B_8_2_0 and cp <= 0X2_C_E_A_F)  #
         or (cp >= 0XF_9_0_0 and cp <= 0XF_A_F_F)
         or (cp >= 0X2_F_8_0_0 and cp <= 0X2_F_A_1_F)  #
     ):  #
        return True

     return False






class 			__lowerCAmelCase   (       snake_case_    ):






  """simple docstring"""





  def __init__(   self				:				int			, _lowerCAmelCase				:				Optional[int]			, _lowerCAmelCase				:				int = False			, _lowerCAmelCase				:				str = None			, **_lowerCAmelCase				:				Optional[Any]     )      ->       List[str]:




     """simple docstring"""
     super().__init__(_lowerCAmelCase			, _lowerCAmelCase			, **_lowerCAmelCase     )
     snake_case_    			=    Queue()
     snake_case_    			=    None
     snake_case_    			=    timeout



  def     lowerCAmelCase__     (   self				:				Any			, _lowerCAmelCase				:				Any			, _lowerCAmelCase				:				List[str] = False     )      ->       Dict:




     """simple docstring"""
     self.text_queue.put(_lowerCAmelCase			, timeout=self.timeout     )
     if stream_end:
        self.text_queue.put(self.stop_signal			, timeout=self.timeout     )



  def __iter__(   self				:				Any     )      ->       str:




     """simple docstring"""
     return self



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




     """simple docstring"""
     snake_case_    			=    self.text_queue.get(timeout=self.timeout     )
     if value == self.stop_signal:
        raise StopIteration()
     else:
        return value






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     _SCREAMING_SNAKE_CASE     (      snake_case_						):



	def __init__(    self						,  *lowercase						,  lowercase=None						,  lowercase=None						,  **lowercase )				->       List[str]:
					super().__init__(*lowercase						,  **lowercase )
					lowerCamelCase_						=			eval_examples
					lowerCamelCase_						=			post_process_function



	def     SCREAMING_SNAKE_CASE_(    self						,  lowercase = None						,  lowercase=None						,  lowercase = None						,  lowercase = "eval"						,  **lowercase						,  )				->       Dict[str, float]:
					lowerCamelCase_						=			gen_kwargs.copy()
					lowerCamelCase_						=			(
					    gen_kwargs["max_length"] if gen_kwargs.get("max_length" ) is not None else self.args.generation_max_length
					)
					lowerCamelCase_						=			(
					    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams" ) is not None else self.args.generation_num_beams
					)
					lowerCamelCase_						=			gen_kwargs

					lowerCamelCase_						=			self.eval_dataset if eval_dataset is None else eval_dataset
					lowerCamelCase_						=			self.get_eval_dataloader(lowercase )
					lowerCamelCase_						=			self.eval_examples if eval_examples is None else eval_examples

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

									metrics.update(output.metrics )
					else:
									lowerCamelCase_						=			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() )

					lowerCamelCase_						=			self.callback_handler.on_evaluate(self.args						,  self.state						,  self.control						,  lowercase )
					return metrics



	def     SCREAMING_SNAKE_CASE_(    self						,  lowercase						,  lowercase						,  lowercase=None						,  lowercase = "test"						,  **lowercase )				->       Union[str, Any]:
					lowerCamelCase_						=			gen_kwargs.copy()

					lowerCamelCase_						=			self.get_test_dataloader(lowercase )

					# Temporarily disable metric computation, we will do it in the loop here.
					lowerCamelCase_						=			self.compute_metrics
					lowerCamelCase_						=			None
					lowerCamelCase_						=			time.time()
					lowerCamelCase_						=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					try:
									lowerCamelCase_						=			eval_loop(
									    lowercase						,  description="Prediction"						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase						,  metric_key_prefix=lowercase						,  )
					finally:
									lowerCamelCase_						=			compute_metrics

					lowerCamelCase_						=			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

					lowerCamelCase_						=			self.post_process_function(lowercase						,  lowercase						,  lowercase						,  "predict" )
					lowerCamelCase_						=			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}_' ):
													lowerCamelCase_						=			metrics.pop(lowercase )
					metrics.update(output.metrics )
					return PredictionOutput(predictions=predictions.predictions						,  label_ids=predictions.label_ids						,  metrics=lowercase )


