


import numpy as np
import torch
from torch.nn import CrossEntropyLoss
from transformers import AutoModelForCausalLM, AutoTokenizer

import datasets
from datasets import logging


UpperCamelCase__												=      """\

"""

UpperCamelCase__												=      """
Perplexity (PPL) is one of the most common metrics for evaluating language models.
It is defined as the exponentiated average negative log-likelihood of a sequence.

For more information, see https://huggingface.co/docs/transformers/perplexity
"""

UpperCamelCase__												=      """
Args:
    model_id (str): model used for calculating Perplexity
            NOTE: Perplexity can only be calculated for causal language models.
                    This includes models such as gpt2, causal variations of bert,
                    causal versions of t5, and more (the full list can be found
                    in the AutoModelForCausalLM documentation here:
                    https://huggingface.co/docs/transformers/master/en/model_doc/auto#transformers.AutoModelForCausalLM )

    input_texts (list of str): input text, each separate text snippet
        is one list entry.
    batch_size (int): the batch size to run texts through the model. Defaults to 16.
    add_start_token (bool): whether to add the start token to the texts,
        so the perplexity can include the probability of the first word. Defaults to True.
    device (str): device to run on, defaults to 'cuda' when available
Returns:
    perplexity: dictionary containing the perplexity scores for the texts
        in the input list, as well as the mean perplexity. If one of the input texts is
        longer than the max input length of the model, then it is truncated to the
        max length for the perplexity computation.
Examples:
    Example 1:
        >>> perplexity = datasets.load_metric(\"perplexity\")
        >>> input_texts = [\"lorem ipsum\", \"Happy Birthday!\", \"Bienvenue\"]
        >>> results = perplexity.compute(model_id='gpt2',
        ...                              add_start_token=False,
        ...                              input_texts=input_texts) # doctest:+ELLIPSIS
        >>> print(list(results.keys()))
        ['perplexities', 'mean_perplexity']
        >>> print(round(results[\"mean_perplexity\"], 2))
        78.22
        >>> print(round(results[\"perplexities\"][0], 2))
        11.11

    Example 2:
        >>> perplexity = datasets.load_metric(\"perplexity\")
        >>> input_texts = datasets.load_dataset(\"wikitext\",
        ...                                     \"wikitext-2-raw-v1\",
        ...                                     split=\"test\")[\"text\"][:50] # doctest:+ELLIPSIS
        [...]
        >>> input_texts = [s for s in input_texts if s!='']
        >>> results = perplexity.compute(model_id='gpt2',
        ...                              input_texts=input_texts) # doctest:+ELLIPSIS
        >>> print(list(results.keys()))
        ['perplexities', 'mean_perplexity']
        >>> print(round(results[\"mean_perplexity\"], 2))
        60.35
        >>> print(round(results[\"perplexities\"][0], 2))
        81.12
"""

@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION   ,    _KWARGS_DESCRIPTION		)
class     UpperCAmelCase__ (   datasets.Metric		):



     '''simple docstring'''




     def 	lowerCAmelCase_ (			self							:							int		):





            """simple docstring"""



            return datasets.MetricInfo(
                description=_DESCRIPTION						,   citation=_CITATION						,   inputs_description=_KWARGS_DESCRIPTION						,   features=datasets.Features(
                    {
                        '''input_texts''': datasets.Value('''string'''		),
                    }		)						,   reference_urls=['''https://huggingface.co/docs/transformers/perplexity''']						,   )




     def 	lowerCAmelCase_ (			self							:							Optional[Any]						,   UpperCamelCase							:							List[str]						,   UpperCamelCase							:							str						,   UpperCamelCase							:							int = 16						,   UpperCamelCase							:							bool = True						,   UpperCamelCase							:							Union[str, Any]=None		):





            """simple docstring"""



            if device is not None:
                   assert device in ["gpu", "cpu", "cuda"], "device should be either gpu or cpu."
                   if device == "gpu":
                          _lowercase       :					str     							=      '''cuda'''
            else:
                   _lowercase       :					Any     							=      '''cuda''' if torch.cuda.is_available() else '''cpu'''

            _lowercase       :					List[str]     							=      AutoModelForCausalLM.from_pretrained(UpperCAmelCase__		)
            _lowercase       :					str     							=      model.to(UpperCAmelCase__		)

            _lowercase       :					str     							=      AutoTokenizer.from_pretrained(UpperCAmelCase__		)

            # if batch_size > 1 (which generally leads to padding being required), and
            # if there is not an already assigned pad_token, assign an existing
            # special token to also be the padding token
            if tokenizer.pad_token is None and batch_size > 1:
                   _lowercase       :					Optional[int]     							=      list(tokenizer.special_tokens_map_extended.values()		)
                   # check that the model already has at least one special token defined
                   assert (
                       len(UpperCAmelCase__		) > 0
                   ), "If batch_size > 1, model must have at least one special token to use for padding. Please use a different model or set batch_size=1."
                   # assign one of the special tokens to also be the pad token
                   tokenizer.add_special_tokens({'''pad_token''': existing_special_tokens[0]}		)

            if add_start_token:
                   # leave room for <BOS> token to be added:
                   assert (
                       tokenizer.bos_token is not None
                   ), "Input model must already have a BOS token if using add_start_token=True. Please use a different model, or set add_start_token=False"
                   _lowercase       :					Tuple     							=      model.config.max_length - 1
            else:
                   _lowercase       :					Optional[int]     							=      model.config.max_length

            _lowercase       :					Tuple     							=      tokenizer(
                UpperCAmelCase__						,   add_special_tokens=UpperCAmelCase__						,   padding=UpperCAmelCase__						,   truncation=UpperCAmelCase__						,   max_length=UpperCAmelCase__						,   return_tensors='''pt'''						,   return_attention_mask=UpperCAmelCase__						,   ).to(UpperCAmelCase__		)

            _lowercase       :					Dict     							=      encodings['''input_ids''']
            _lowercase       :					Dict     							=      encodings['''attention_mask''']

            # check that each input is long enough:
            if add_start_token:
                   assert torch.all(torch.ge(attn_masks.sum(1		)						,   1		)		), "Each input text must be at least one token long."
            else:
                   assert torch.all(
                       torch.ge(attn_masks.sum(1		)						,   2		)		), "When add_start_token=False, each input text must be at least two tokens long. Run with add_start_token=True if inputting strings of only one token, and remove all empty input strings."

            _lowercase       :					Any     							=      []
            _lowercase       :					Any     							=      CrossEntropyLoss(reduction='''none'''		)

            for start_index in logging.tqdm(range(0						,   len(UpperCAmelCase__		)						,   UpperCAmelCase__		)		):
                   _lowercase       :					List[Any]     							=      min(start_index + batch_size						,   len(UpperCAmelCase__		)		)
                   _lowercase       :					Tuple     							=      encoded_texts[start_index:end_index]
                   _lowercase       :					int     							=      attn_masks[start_index:end_index]

                   if add_start_token:
                          _lowercase       :					str     							=      torch.tensor([[tokenizer.bos_token_id]] * encoded_batch.size(dim=0		)		).to(UpperCAmelCase__		)
                          _lowercase       :					Optional[Any]     							=      torch.cat([bos_tokens_tensor, encoded_batch]						,   dim=1		)
                          _lowercase       :					str     							=      torch.cat(
                              [torch.ones(bos_tokens_tensor.size()						,   dtype=torch.intaa		).to(UpperCAmelCase__		), attn_mask]						,   dim=1		)

                   _lowercase       :					Any     							=      encoded_batch

                   with torch.no_grad():
                          _lowercase       :					Dict     							=      model(UpperCAmelCase__						,   attention_mask=UpperCAmelCase__		).logits

                   _lowercase       :					Optional[Any]     							=      out_logits[..., :-1, :].contiguous()
                   _lowercase       :					Optional[int]     							=      labels[..., 1:].contiguous()
                   _lowercase       :					Dict     							=      attn_mask[..., 1:].contiguous()

                   _lowercase       :					List[str]     							=      torch.expa(
                       (loss_fct(shift_logits.transpose(1						,   2		)						,   UpperCAmelCase__		) * shift_attention_mask_batch).sum(1		)
                       / shift_attention_mask_batch.sum(1		)		)

                   ppls += perplexity_batch.tolist()

            return {"perplexities": ppls, "mean_perplexity": np.mean(UpperCAmelCase__		)}





import warnings
from typing import Dict

import numpy as np

from ..utils import ExplicitEnum, add_end_docstrings, is_tf_available, is_torch_available
from .base import PIPELINE_INIT_ARGS, GenericTensor, Pipeline


if is_tf_available():
  from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

if is_torch_available():
  from ..models.auto.modeling_auto import MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING







def SCREAMING_SNAKE_CASE							(			lowercase_		)  ->			int:



       """simple docstring"""

       return 1.0 / (1.0 + np.exp(-_outputs		))







def SCREAMING_SNAKE_CASE							(			lowercase_		)  ->			Union[str, Any]:



       """simple docstring"""

       A__ 					=       np.max(_outputs					,							axis=-1					,							keepdims=lowercase_		)
       A__ 					=       np.exp(_outputs - maxes		)
       return shifted_exp / shifted_exp.sum(axis=-1					,							keepdims=lowercase_		)


class 							UpperCamelCase_       (					UpperCAmelCase__	):

    '''simple docstring'''






    UpperCAmelCase__				    = '''sigmoid'''
    UpperCAmelCase__				    = '''softmax'''
    UpperCAmelCase__				    = '''none'''


@add_end_docstrings(
    UpperCAmelCase__						,   R'''
        return_all_scores (`bool`, *optional*, defaults to `False`):
            Whether to return all prediction scores or just the one of the predicted class.
        function_to_apply (`str`, *optional*, defaults to `"default"`):
            The function to apply to the model outputs in order to retrieve the scores. Accepts four different values:

            - `"default"`: if the model has a single label, will apply the sigmoid function on the output. If the model
              has several labels, will apply the softmax function on the output.
            - `"sigmoid"`: Applies the sigmoid function on the output.
            - `"softmax"`: Applies the softmax function on the output.
            - `"none"`: Does not apply any function on the output.
    '''						,   )
class 							UpperCamelCase_       (					UpperCAmelCase__	):

    '''simple docstring'''






    UpperCAmelCase__				    = False
    UpperCAmelCase__				    = ClassificationFunction.NONE


    def __init__(     self :      Any   ,					**UpperCAmelCase__ :      Optional[Any])		->str:





           '''simple docstring'''

           super().__init__(**UpperCAmelCase__)

           self.check_model_type(
               TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
               if self.framework == '''tf'''
               else MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING)


    def 			SCREAMING_SNAKE_CASE				(     self :      List[str]   ,					UpperCAmelCase__ :      Tuple=None   ,					UpperCAmelCase__ :      Any=None   ,					UpperCAmelCase__ :      int=""   ,					**UpperCAmelCase__ :      Any)		->int:





           '''simple docstring'''

           A__ 					=       tokenizer_kwargs

           A__ 					=       {}
           if hasattr(self.model.config   ,					'''return_all_scores''') and return_all_scores is None:
                  A__ 					=       self.model.config.return_all_scores

           if isinstance(UpperCAmelCase__   ,					UpperCAmelCase__) or top_k is None:
                  A__ 					=       top_k
                  A__ 					=       False
           elif return_all_scores is not None:
                  warnings.warn(
                      '''`return_all_scores` is now deprecated,  if want a similar functionality use `top_k=None` instead of'''
                      ''' `return_all_scores=True` or `top_k=1` instead of `return_all_scores=False`.'''   ,					UpperCAmelCase__   ,					)
                  if return_all_scores:
                         A__ 					=       None
                  else:
                         A__ 					=       1

           if isinstance(UpperCAmelCase__   ,					UpperCAmelCase__):
                  A__ 					=       ClassificationFunction[function_to_apply.upper()]

           if function_to_apply is not None:
                  A__ 					=       function_to_apply
           return preprocess_params, {}, postprocess_params


    def __call__(     self :      str   ,					*UpperCAmelCase__ :      List[Any]   ,					**UpperCAmelCase__ :      Optional[int])		->Union[str, Any]:





           '''simple docstring'''

           A__ 					=       super().__call__(*UpperCAmelCase__   ,					**UpperCAmelCase__)
           # TODO try and retrieve it in a nicer way from _sanitize_parameters.
           A__ 					=       '''top_k''' not in kwargs
           if isinstance(args[0]   ,					UpperCAmelCase__) and _legacy:
                  # This pipeline is odd, and return a list when single item is run
                  return [result]
           else:
                  return result


    def 			SCREAMING_SNAKE_CASE				(     self :      Tuple   ,					UpperCAmelCase__ :      Any   ,					**UpperCAmelCase__ :      str)		->Dict[str, GenericTensor]:





           '''simple docstring'''

           A__ 					=       self.framework
           if isinstance(UpperCAmelCase__   ,					UpperCAmelCase__):
                  return self.tokenizer(**UpperCAmelCase__   ,					return_tensors=UpperCAmelCase__   ,					**UpperCAmelCase__)
           elif isinstance(UpperCAmelCase__   ,					UpperCAmelCase__) and len(UpperCAmelCase__) == 1 and isinstance(inputs[0]   ,					UpperCAmelCase__) and len(inputs[0]) == 2:
                  # It used to be valid to use a list of list of list for text pairs, keeping this path for BC
                  return self.tokenizer(
                      text=inputs[0][0]   ,					text_pair=inputs[0][1]   ,					return_tensors=UpperCAmelCase__   ,					**UpperCAmelCase__)
           elif isinstance(UpperCAmelCase__   ,					UpperCAmelCase__):
                  # This is likely an invalid usage of the pipeline attempting to pass text pairs.
                  raise ValueError(
                      '''The pipeline received invalid inputs, if you are trying to send text pairs, you can try to send a'''
                      ''' dictionary `{"text": "My text", "text_pair": "My pair"}` in order to send a text pair.''')
           return self.tokenizer(UpperCAmelCase__   ,					return_tensors=UpperCAmelCase__   ,					**UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      List[str]   ,					UpperCAmelCase__ :      Tuple)		->Tuple:





           '''simple docstring'''

           return self.model(**UpperCAmelCase__)





    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[Any]   ,					UpperCAmelCase__ :      int   ,					UpperCAmelCase__ :      Optional[int]=None   ,					UpperCAmelCase__ :      List[Any]=1   ,					UpperCAmelCase__ :      str=True)		->Dict:





           '''simple docstring'''

           if function_to_apply is None:
                  if self.model.config.problem_type == "multi_label_classification" or self.model.config.num_labels == 1:
                         A__ 					=       ClassificationFunction.SIGMOID
                  elif self.model.config.problem_type == "single_label_classification" or self.model.config.num_labels > 1:
                         A__ 					=       ClassificationFunction.SOFTMAX
                  elif hasattr(self.model.config   ,					'''function_to_apply''') and function_to_apply is None:
                         A__ 					=       self.model.config.function_to_apply
                  else:
                         A__ 					=       ClassificationFunction.NONE

           A__ 					=       model_outputs['''logits'''][0]
           A__ 					=       outputs.numpy()

           if function_to_apply == ClassificationFunction.SIGMOID:
                  A__ 					=       sigmoid(UpperCAmelCase__)
           elif function_to_apply == ClassificationFunction.SOFTMAX:
                  A__ 					=       softmax(UpperCAmelCase__)
           elif function_to_apply == ClassificationFunction.NONE:
                  A__ 					=       outputs
           else:
                  raise ValueError(f"""Unrecognized `function_to_apply` argument: {function_to_apply}""")

           if top_k == 1 and _legacy:
                  return {"label": self.model.config.idalabel[scores.argmax().item()], "score": scores.max().item()}

           A__ 					=       [
               {'''label''': self.model.config.idalabel[i], '''score''': score.item()} for i, score in enumerate(UpperCAmelCase__)
           ]
           if not _legacy:
                  dict_scores.sort(key=lambda UpperCAmelCase__: x["score"]   ,					reverse=UpperCAmelCase__)
                  if top_k is not None:
                         A__ 					=       dict_scores[:top_k]
           return dict_scores


