




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				):
 def __init__(       self    ,			*lowerCAmelCase__    ,			lowerCAmelCase__=None    ,			lowerCAmelCase__=None    ,			**lowerCAmelCase__			) ->    int:




        '''simple docstring'''
        super().__init__(*lowerCAmelCase__    ,			**lowerCAmelCase__			)
        lowercase__:       List[Any]        =       eval_examples
        lowercase__:       List[Any]        =       post_process_function
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__ = None    ,			lowerCAmelCase__=None    ,			lowerCAmelCase__ = None    ,			lowerCAmelCase__ = "eval"    ,			**lowerCAmelCase__    ,			) ->    Dict[str, float]:




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

        lowercase__:       Any        =       self.eval_dataset if eval_dataset is None else eval_dataset
        lowercase__:       Any        =       self.get_eval_dataloader(lowerCAmelCase__			)
        lowercase__:       Tuple        =       self.eval_examples if eval_examples is None else eval_examples

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

               # Prefix all keys with metric_key_prefix + '_'
               for key in list(metrics.keys()			):
                      if not key.startswith(F'{metric_key_prefix}_'			):
                             lowercase__:       List[str]        =       metrics.pop(lowerCAmelCase__			)

               metrics.update(output.metrics			)
        else:
               lowercase__:       List[Any]        =       output.metrics

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

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

        lowercase__:       Union[str, Any]        =       self.callback_handler.on_evaluate(self.args    ,			self.state    ,			self.control    ,			lowerCAmelCase__			)
        return metrics
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__=None    ,			lowerCAmelCase__ = "test"    ,			**lowerCAmelCase__			) ->    List[Any]:




        '''simple docstring'''
        lowercase__:       Dict        =       gen_kwargs.copy()

        lowercase__:       Any        =       self.get_test_dataloader(lowerCAmelCase__			)

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

        lowercase__:       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(
                lowerCAmelCase__    ,			lowerCAmelCase__    ,			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

        lowercase__:       Dict        =       self.post_process_function(lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			'predict'			)
        lowercase__:       Union[str, Any]        =       self.compute_metrics(lowerCAmelCase__			)

        # Prefix all keys with metric_key_prefix + '_'
        for key in list(metrics.keys()			):
               if not key.startswith(F'{metric_key_prefix}_'			):
                      lowercase__:       Tuple        =       metrics.pop(lowerCAmelCase__			)
        metrics.update(output.metrics			)
        return PredictionOutput(predictions=predictions.predictions    ,			label_ids=predictions.label_ids    ,			metrics=lowerCAmelCase__			)






import os
from typing import Dict, List, Union

import tensorflow as tf
from keras_nlp.tokenizers import BytePairTokenizer
from tensorflow_text import pad_model_inputs

from .tokenization_gpta import GPTaTokenizer





class   __a      (   tf.keras.layers.Layer				):
 def __init__(       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__ = None    ,			lowerCAmelCase__ = None			) ->    int:




        '''simple docstring'''
        super().__init__()
        lowercase__:       Union[str, Any]        =       pad_token_id
        lowercase__:       List[str]        =       max_length
        lowercase__:       int        =       vocab
        lowercase__:       List[Any]        =       merges
        lowercase__:       str        =       BytePairTokenizer(lowerCAmelCase__    ,			lowerCAmelCase__    ,			sequence_length=lowerCAmelCase__			)
 @classmethod
 def 		SCREAMING_SNAKE_CASE__       (       cls    ,			lowerCAmelCase__    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			) ->    Any:




        '''simple docstring'''
        lowercase__:       Tuple        =       [' '.join(lowerCAmelCase__			) for m in tokenizer.bpe_ranks.keys()]
        lowercase__:       List[Any]        =       tokenizer.get_vocab()
        return cls(lowerCAmelCase__    ,			lowerCAmelCase__    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			)
 @classmethod
 def 		SCREAMING_SNAKE_CASE__       (       cls    ,			lowerCAmelCase__    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			) ->    List[Any]:




        '''simple docstring'''
        lowercase__:       int        =       GPTaTokenizer.from_pretrained(lowerCAmelCase__    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			)
        return cls.from_tokenizer(lowerCAmelCase__    ,			*lowerCAmelCase__    ,			**lowerCAmelCase__			)
 @classmethod
 def 		SCREAMING_SNAKE_CASE__       (       cls    ,			lowerCAmelCase__			) ->    Dict:




        '''simple docstring'''
        return cls(**lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    List[Any]:




        '''simple docstring'''
        return {
            "vocab": self.vocab,
            "merges": self.merges,
            "max_length": self.max_length,
            "pad_token_id": self.pad_token_id,
        }
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__ = None			) ->    Optional[Any]:




        '''simple docstring'''
        lowercase__:       Optional[Any]        =       self.tf_tokenizer(lowerCAmelCase__			)
        lowercase__:       List[Any]        =       tf.ones_like(lowerCAmelCase__			)

        if self.pad_token_id is not None:
               # pad the tokens up to max length
               lowercase__:       int        =       max_length if max_length is not None else self.max_length

               if max_length is not None:
                      lowercase__      ,						lowercase__:       List[Any]        =       pad_model_inputs(
                          lowerCAmelCase__    ,			max_seq_length=lowerCAmelCase__    ,			pad_value=self.pad_token_id			)

        return {"attention_mask": attention_mask, "input_ids": input_ids}


