
"""simple docstring"""



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

import datasets
from datasets import logging


a					:	List[Any]							=		'''\

'''

a					:	List[Any]							=		'''
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
'''

a					:	Union[str, Any]							=		'''
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 		a_						( datasets.Metric	):







    def      _snake_case					(			self			:       Any     )		->List[str]:






        '''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      _snake_case					(			self			:       Optional[int]     ,					__UpperCamelCase			:       Tuple     ,					__UpperCamelCase			:       List[Any]     ,					__UpperCamelCase			:       int = 16     ,					__UpperCamelCase			:       bool = True     ,					__UpperCamelCase			:       Any=None     )		->Union[str, Any]:






        '''simple docstring'''






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

        _UpperCAmelCase          =   AutoModelForCausalLM.from_pretrained(__UpperCamelCase     )
        _UpperCAmelCase          =   model.to(__UpperCamelCase     )

        _UpperCAmelCase          =   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:
            _UpperCAmelCase          =   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"
            _UpperCAmelCase          =   model.config.max_length - 1
        else:
            _UpperCAmelCase          =   model.config.max_length

        _UpperCAmelCase          =   tokenizer(
            __UpperCamelCase     ,					add_special_tokens=__UpperCamelCase     ,					padding=__UpperCamelCase     ,					truncation=__UpperCamelCase     ,					max_length=__UpperCamelCase     ,					return_tensors="""pt"""     ,					return_attention_mask=__UpperCamelCase     ,					).to(__UpperCamelCase     )

        _UpperCAmelCase          =   encodings["""input_ids"""]
        _UpperCAmelCase          =   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."

        _UpperCAmelCase          =   []
        _UpperCAmelCase          =   CrossEntropyLoss(reduction="""none"""     )

        for start_index in logging.tqdm(range(0     ,					len(__UpperCamelCase     )     ,					__UpperCamelCase     )     ):
            _UpperCAmelCase          =   min(start_index + batch_size     ,					len(__UpperCamelCase     )     )
            _UpperCAmelCase          =   encoded_texts[start_index:end_index]
            _UpperCAmelCase          =   attn_masks[start_index:end_index]

            if add_start_token:
                _UpperCAmelCase          =   torch.tensor([[tokenizer.bos_token_id]] * encoded_batch.size(dim=0     )     ).to(__UpperCamelCase     )
                _UpperCAmelCase          =   torch.cat([bos_tokens_tensor, encoded_batch]     ,					dim=1     )
                _UpperCAmelCase          =   torch.cat(
                    [torch.ones(bos_tokens_tensor.size()     ,					dtype=torch.intaa     ).to(__UpperCamelCase     ), attn_mask]     ,					dim=1     )

            _UpperCAmelCase          =   encoded_batch

            with torch.no_grad():
                _UpperCAmelCase          =   model(__UpperCamelCase     ,					attention_mask=__UpperCamelCase     ).logits

            _UpperCAmelCase          =   out_logits[..., :-1, :].contiguous()
            _UpperCAmelCase          =   labels[..., 1:].contiguous()
            _UpperCAmelCase          =   attn_mask[..., 1:].contiguous()

            _UpperCAmelCase          =   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     )}

"""simple docstring"""



import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


a					:	int							=		WebClient(token=os.environ['''CI_SLACK_BOT_TOKEN'''])


def  _UpperCamelCase	(						_A      )    ->  List[str]:





    """simple docstring"""

    _UpperCAmelCase          =   test_results.split(""" """      )

    _UpperCAmelCase          =   0
    _UpperCAmelCase          =   0

    # When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
    # When it is too long, those signs are not present.
    _UpperCAmelCase          =   expressions[-2] if """=""" in expressions[-1] else expressions[-1]

    for i, expression in enumerate(_A      ):
        if "failed" in expression:
            failed += int(expressions[i - 1]      )
        if "passed" in expression:
            success += int(expressions[i - 1]      )

    return failed, success, time_spent


def  _UpperCamelCase	(						_A      )    ->  int:





    """simple docstring"""

    _UpperCAmelCase          =   {}
    _UpperCAmelCase          =   None
    _UpperCAmelCase          =   False
    for line in failures_short_lines.split("""\n"""      ):
        if re.search(R"""_ \[doctest\]"""					,      _A      ):
            _UpperCAmelCase          =   True
            _UpperCAmelCase          =   line.split(""" """      )[2]
        elif in_error and not line.split(""" """      )[0].isdigit():
            _UpperCAmelCase          =   line
            _UpperCAmelCase          =   False

    return failures




class 		a_						:







    def __init__(			self			:       Union[str, Any]     ,					__UpperCamelCase			:       str     ,					__UpperCamelCase			:       Dict     )		->Union[str, Any]:






        '''simple docstring'''






        _UpperCAmelCase          =   title

        _UpperCAmelCase          =   doc_test_results["""time_spent"""].split(""","""     )[0]
        _UpperCAmelCase          =   doc_test_results["""success"""]
        _UpperCAmelCase          =   doc_test_results["""failures"""]
        _UpperCAmelCase          =   self.n_success + self.n_failures

        # Failures and success of the modeling tests
        _UpperCAmelCase          =   doc_test_results







    @property
    def      _snake_case					(			self			:       int     )		->str:






        '''simple docstring'''






        _UpperCAmelCase          =   [self._time_spent]
        _UpperCAmelCase          =   0

        for time in time_spent:
            _UpperCAmelCase          =   time.split(""":"""     )

            # Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
            if len(__UpperCamelCase     ) == 1:
                _UpperCAmelCase          =   [0, 0, time_parts[0]]

            _UpperCAmelCase    ,_UpperCAmelCase    ,_UpperCAmelCase          =   int(time_parts[0]     ), int(time_parts[1]     ), float(time_parts[2]     )
            total_secs += hours * 36_00 + minutes * 60 + seconds

        _UpperCAmelCase    ,_UpperCAmelCase    ,_UpperCAmelCase          =   total_secs // 36_00, (total_secs % 36_00) // 60, total_secs % 60
        return f"""{int(__UpperCamelCase     )}h{int(__UpperCamelCase     )}m{int(__UpperCamelCase     )}s"""







    @property
    def      _snake_case					(			self			:       List[Any]     )		->Dict:






        '''simple docstring'''






        return {"type": "header", "text": {"type": "plain_text", "text": self.title}}







    @property
    def      _snake_case					(			self			:       Optional[Any]     )		->Dict:






        '''simple docstring'''






        return {
            "type": "section",
            "text": {
                "type": "plain_text",
                "text": f"""🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.""",
                "emoji": True,
            },
            "accessory": {
                "type": "button",
                "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                "url": f"""https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
            },
        }







    @property
    def      _snake_case					(			self			:       Union[str, Any]     )		->Dict:






        '''simple docstring'''






        return {
            "type": "section",
            "text": {
                "type": "plain_text",
                "text": (
                    f"""There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in"""
                    f""" {self.time}."""
                ),
                "emoji": True,
            },
            "accessory": {
                "type": "button",
                "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                "url": f"""https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
            },
        }







    @property
    def      _snake_case					(			self			:       List[str]     )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase          =   40
        _UpperCAmelCase          =   {k: v["""failed"""] for k, v in doc_test_results.items() if isinstance(__UpperCamelCase     ,					__UpperCamelCase     )}

        _UpperCAmelCase          =   """"""
        for category, failures in category_failures.items():
            if len(__UpperCamelCase     ) == 0:
                continue

            if report != "":
                report += "\n\n"

            report += f"""*{category} failures*:""".ljust(line_length // 2     ).rjust(line_length // 2     ) + "\n"
            report += "`"
            report += "`\n`".join(__UpperCamelCase     )
            report += "`"

        return {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"""The following examples had failures:\n\n\n{report}\n""",
            },
        }







    @property
    def      _snake_case					(			self			:       Union[str, Any]     )		->str:






        '''simple docstring'''






        _UpperCAmelCase          =   [self.header]

        if self.n_failures > 0:
            blocks.append(self.failures     )

        if self.n_failures > 0:
            blocks.extend([self.category_failures]     )

        if self.n_failures == 0:
            blocks.append(self.no_failures     )

        return json.dumps(__UpperCamelCase     )







    @staticmethod
    def      _snake_case					(			)		->Any:






        '''simple docstring'''






        _UpperCAmelCase          =   [
            {
                """type""": """section""",
                """text""": {
                    """type""": """plain_text""",
                    """text""": """There was an issue running the tests.""",
                },
                """accessory""": {
                    """type""": """button""",
                    """text""": {"""type""": """plain_text""", """text""": """Check Action results""", """emoji""": True},
                    """url""": f"""https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
                },
            }
        ]

        print("""Sending the following payload"""     )
        print(json.dumps({"""blocks""": json.loads(__UpperCamelCase     )}     )     )

        client.chat_postMessage(
            channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]     ,					text="""There was an issue running the tests."""     ,					blocks=__UpperCamelCase     ,					)







    def      _snake_case					(			self			:       Tuple     )		->Optional[Any]:






        '''simple docstring'''






        print("""Sending the following payload"""     )
        print(json.dumps({"""blocks""": json.loads(self.payload     )}     )     )

        _UpperCAmelCase          =   f"""{self.n_failures} failures out of {self.n_tests} tests,""" if self.n_failures else """All tests passed."""

        _UpperCAmelCase          =   client.chat_postMessage(
            channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]     ,					blocks=self.payload     ,					text=__UpperCamelCase     ,					)







    def      _snake_case					(			self			:       List[Any]     ,					__UpperCamelCase			:       Optional[int]     ,					__UpperCamelCase			:       Any     ,					__UpperCamelCase			:       Any     ,					__UpperCamelCase			:       List[str]     )		->str:






        '''simple docstring'''






        _UpperCAmelCase          =   """"""
        for key, value in failures.items():
            _UpperCAmelCase          =   value[:2_00] + """ [Truncated]""" if len(__UpperCamelCase     ) > 2_50 else value
            failures_text += f"""*{key}*\n_{value}_\n\n"""

        _UpperCAmelCase          =   job_name
        _UpperCAmelCase          =   {"""type""": """section""", """text""": {"""type""": """mrkdwn""", """text""": text}}

        if job_link is not None:
            _UpperCAmelCase          =   {
                """type""": """button""",
                """text""": {"""type""": """plain_text""", """text""": """GitHub Action job""", """emoji""": True},
                """url""": job_link,
            }

        return [
            {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
            content,
            {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
        ]







    def      _snake_case					(			self			:       int     )		->Optional[Any]:






        '''simple docstring'''






        if self.thread_ts is None:
            raise ValueError("""Can only post reply if a post has been made."""     )

        _UpperCAmelCase          =   self.doc_test_results.pop("""job_link"""     )
        self.doc_test_results.pop("""failures"""     )
        self.doc_test_results.pop("""success"""     )
        self.doc_test_results.pop("""time_spent"""     )

        _UpperCAmelCase          =   sorted(self.doc_test_results.items()     ,					key=lambda __UpperCamelCase     : t[0]     )
        for job, job_result in sorted_dict:
            if len(job_result["""failures"""]     ):
                _UpperCAmelCase          =   f"""*Num failures* :{len(job_result["failed"]     )} \n"""
                _UpperCAmelCase          =   job_result["""failures"""]
                _UpperCAmelCase          =   self.get_reply_blocks(__UpperCamelCase     ,					__UpperCamelCase     ,					__UpperCamelCase     ,					text=__UpperCamelCase     )

                print("""Sending the following reply"""     )
                print(json.dumps({"""blocks""": blocks}     )     )

                client.chat_postMessage(
                    channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]     ,					text=f"""Results for {job}"""     ,					blocks=__UpperCamelCase     ,					thread_ts=self.thread_ts["""ts"""]     ,					)

                time.sleep(1     )







def  _UpperCamelCase	(						)    ->  List[str]:





    """simple docstring"""

    _UpperCAmelCase          =   os.environ["""GITHUB_RUN_ID"""]
    _UpperCAmelCase          =   F"""https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"""
    _UpperCAmelCase          =   requests.get(_A      ).json()
    _UpperCAmelCase          =   {}

    try:
        jobs.update({job["""name"""]: job["""html_url"""] for job in result["""jobs"""]}      )
        _UpperCAmelCase          =   math.ceil((result["""total_count"""] - 1_0_0) / 1_0_0      )

        for i in range(_A      ):
            _UpperCAmelCase          =   requests.get(url + F"""&page={i + 2}"""      ).json()
            jobs.update({job["""name"""]: job["""html_url"""] for job in result["""jobs"""]}      )

        return jobs
    except Exception as e:
        print("""Unknown error, could not fetch links."""					,      _A      )

    return {}


def  _UpperCamelCase	(						_A      )    ->  Optional[int]:





    """simple docstring"""

    _UpperCAmelCase          =   {}

    if os.path.exists(_A      ):
        _UpperCAmelCase          =   os.listdir(_A      )
        for file in files:
            try:
                with open(os.path.join(_A					,      _A      )					,      encoding="""utf-8"""      ) as f:
                    _UpperCAmelCase          =   f.read()
            except UnicodeDecodeError as e:
                raise ValueError(F"""Could not open {os.path.join(_A					,      _A      )}."""      ) from e

    return _artifact


def  _UpperCamelCase	(						)    ->  int:





    """simple docstring"""

    class 		a_						:







        def __init__(			self			:       List[Any]     ,					__UpperCamelCase			:       str     )		->Tuple:






            '''simple docstring'''






            _UpperCAmelCase          =   name
            _UpperCAmelCase          =   []







        def __str__(			self			:       int     )		->Optional[Any]:






            '''simple docstring'''






            return self.name







        def      _snake_case					(			self			:       Dict     ,					__UpperCamelCase			:       str     )		->int:






            '''simple docstring'''






            self.paths.append({"""name""": self.name, """path""": path}     )







    _UpperCAmelCase          =   {}

    _UpperCAmelCase          =   filter(os.path.isdir					,      os.listdir()      )
    for directory in directories:
        _UpperCAmelCase          =   directory
        if artifact_name not in _available_artifacts:
            _UpperCAmelCase          =   Artifact(_A      )

            _available_artifacts[artifact_name].add_path(_A      )

    return _available_artifacts


if __name__ == "__main__":
 a					:	Dict							=		get_job_links()
 a					:	Dict							=		retrieve_available_artifacts()

 a					:	Optional[int]							=		collections.OrderedDict(
     [
         ('''*.py''', '''API Examples'''),
         ('''*.md''', '''MD Examples'''),
     ]
 )

 # This dict will contain all the information relative to each doc test category:
 # - failed: list of failed tests
 # - failures: dict in the format 'test': 'error_message'
 a					:	Dict							=		{
     v: {
         '''failed''': [],
         '''failures''': {},
     }
     for v in docs.values()
 }

 # Link to the GitHub Action job
 a					:	int							=		github_actions_job_links.get('''run_doctests''')

 a					:	Tuple							=		available_artifacts['''doc_tests_gpu_test_reports'''].paths[0]
 a					:	Optional[Any]							=		retrieve_artifact(artifact_path['''name'''])
 if "stats" in artifact:
  a		, a		, a					:	str							=		handle_test_results(artifact['''stats'''])
  a					:	Tuple							=		failed
  a					:	int							=		success
  a					:	Any							=		time_spent[1:-1] + ''', '''

  a					:	Dict							=		extract_first_line_failure(artifact['''failures_short'''])
  for line in artifact["summary_short"].split('''\n'''):
   if re.search('''FAILED''', line):
    a					:	List[Any]							=		line.replace('''FAILED ''', '''''')
    a					:	Tuple							=		line.split()[0].replace('''\n''', '''''')

    if "::" in line:
     a		, a					:	Union[str, Any]							=		line.split('''::''')
    else:
     a		, a					:	Optional[Any]							=		line, line

    for file_regex in docs.keys():
     if fnmatch(file_path, file_regex):
      a					:	List[Any]							=		docs[file_regex]
      doc_test_results[category]["failed"].append(test)

      a					:	Optional[Any]							=		all_failures[test] if test in all_failures else '''N/A'''
      a					:	List[str]							=		failure
      break

 a					:	List[Any]							=		Message('''🤗 Results of the doc tests.''', doc_test_results)
 message.post()
 message.post_reply()