



'''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          =		WebClient(token=os.environ["CI_SLACK_BOT_TOKEN"])





def 						__snake_case(				_lowerCAmelCase		)		-> Optional[Any]:
      snake_case__  :    List[Any]    						=      test_results.split(""" """		)

      snake_case__  :    Tuple    						=      0
      snake_case__  :    Any    						=      0

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

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

      return failed, success, time_spent





def 						__snake_case(				_lowerCAmelCase		)		-> str:
      snake_case__  :    Optional[Any]    						=      {}
      snake_case__  :    List[str]    						=      None
      snake_case__  :    List[str]    						=      False
      for line in failures_short_lines.split("""\n"""		):
            if re.search(r"""_ \[doctest\]"""     ,   lowerCAmelCase__		):
                  snake_case__  :    Optional[Any]    						=      True
                  snake_case__  :    Union[str, Any]    						=      line.split(""" """		)[2]
            elif in_error and not line.split(""" """		)[0].isdigit():
                  snake_case__  :    Union[str, Any]    						=      line
                  snake_case__  :    int    						=      False

      return failures







class     UpperCAmelCase_       :
    """simple docstring"""







    def __init__(    self					:     Optional[Any] ,     snake_case_					:     str ,     snake_case_					:     Dict						):
          snake_case__  :    str    						=      title

          snake_case__  :    Optional[int]    						=      doc_test_results['time_spent'].split(""","""						)[0]
          snake_case__  :    int    						=      doc_test_results['success']
          snake_case__  :    Dict    						=      doc_test_results['failures']
          snake_case__  :    Optional[int]    						=      self.n_success + self.n_failures

          # Failures and success of the modeling tests
          snake_case__  :    Any    						=      doc_test_results




    @property
    def lowerCamelCase       (    self					:     List[str]						):
          snake_case__  :    int    						=      [self._time_spent]
          snake_case__  :    Any    						=      0

          for time in time_spent:
                snake_case__  :    Optional[Any]    						=      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(SCREAMING_SNAKE_CASE_						) == 1:
                      snake_case__  :    Any    						=      [0, 0, time_parts[0]]

                snake_case__  :    List[str]    						=      int(time_parts[0]						), int(time_parts[1]						), float(time_parts[2]						)
                total_secs += hours * 3_600 + minutes * 60 + seconds

          snake_case__  :    Union[str, Any]    						=      total_secs // 3_600, (total_secs % 3_600) // 60, total_secs % 60
          return f"{int(SCREAMING_SNAKE_CASE_						)}h{int(SCREAMING_SNAKE_CASE_						)}m{int(SCREAMING_SNAKE_CASE_						)}s"




    @property
    def lowerCamelCase       (    self					:     int						):
          return {"type": "header", "text": {"type": "plain_text", "text": self.title}}




    @property
    def lowerCamelCase       (    self					:     str						):
          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 lowerCamelCase       (    self					:     List[Any]						):
          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 lowerCamelCase       (    self					:     Tuple						):
          snake_case__  :    int    						=      40
          snake_case__  :    List[Any]    						=      {k: v['failed'] for k, v in doc_test_results.items() if isinstance(SCREAMING_SNAKE_CASE_ ,     SCREAMING_SNAKE_CASE_						)}

          snake_case__  :    List[str]    						=      ''
          for category, failures in category_failures.items():
                if len(SCREAMING_SNAKE_CASE_						) == 0:
                      continue

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

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

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




    @property
    def lowerCamelCase       (    self					:     int						):
          snake_case__  :    int    						=      [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(SCREAMING_SNAKE_CASE_						)




    @staticmethod
    def lowerCamelCase       (    ):
          snake_case__  :    Tuple    						=      [
              {
                  '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(SCREAMING_SNAKE_CASE_						)}						)						)

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




    def lowerCamelCase       (    self					:     Union[str, Any]						):
          print("""Sending the following payload"""						)
          print(json.dumps({"""blocks""": json.loads(self.payload						)}						)						)

          snake_case__  :    Any    						=      f"{self.n_failures} failures out of {self.n_tests} tests," if self.n_failures else 'All tests passed.'

          snake_case__  :    str    						=      client.chat_postMessage(
              channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""] ,     blocks=self.payload ,     text=SCREAMING_SNAKE_CASE_ ,     )




    def lowerCamelCase       (    self					:     str ,     snake_case_					:     int ,     snake_case_					:     Any ,     snake_case_					:     Any ,     snake_case_					:     Dict						):
          snake_case__  :    List[Any]    						=      ''
          for key, value in failures.items():
                snake_case__  :    List[Any]    						=      value[:200] + ' [Truncated]' if len(SCREAMING_SNAKE_CASE_						) > 250 else value
                failures_text += f"*{key}*\n_{value}_\n\n"

          snake_case__  :    int    						=      job_name
          snake_case__  :    Dict    						=      {'type': 'section', 'text': {'type': 'mrkdwn', 'text': text}}

          if job_link is not None:
                snake_case__  :    str    						=      {
                    '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 lowerCamelCase       (    self					:     Union[str, Any]						):
          if self.thread_ts is None:
                raise ValueError("""Can only post reply if a post has been made."""						)

          snake_case__  :    Dict    						=      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"""						)

          snake_case__  :    Dict    						=      sorted(self.doc_test_results.items() ,     key=lambda snake_case_						: t[0]						)
          for job, job_result in sorted_dict:
                if len(job_result["""failures"""]						):
                      snake_case__  :    Tuple    						=      f"*Num failures* :{len(job_result['failed']						)} \n"
                      snake_case__  :    List[str]    						=      job_result['failures']
                      snake_case__  :    Union[str, Any]    						=      self.get_reply_blocks(SCREAMING_SNAKE_CASE_ ,     SCREAMING_SNAKE_CASE_ ,     SCREAMING_SNAKE_CASE_ ,     text=SCREAMING_SNAKE_CASE_						)

                      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=SCREAMING_SNAKE_CASE_ ,     thread_ts=self.thread_ts["""ts"""] ,     )

                      time.sleep(1						)





def 						__snake_case(				)		-> str:
      snake_case__  :    Union[str, Any]    						=      os.environ['GITHUB_RUN_ID']
      snake_case__  :    int    						=      f"https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"
      snake_case__  :    str    						=      requests.get(lowerCAmelCase__		).json()
      snake_case__  :    List[str]    						=      {}

      try:
            jobs.update({job["""name"""]: job["""html_url"""] for job in result["""jobs"""]}		)
            snake_case__  :    Optional[Any]    						=      math.ceil((result["""total_count"""] - 100) / 100		)

            for i in range(lowerCAmelCase__		):
                  snake_case__  :    int    						=      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."""     ,   lowerCAmelCase__		)

      return {}





def 						__snake_case(				_lowerCAmelCase		)		-> List[Any]:
      snake_case__  :    int    						=      {}

      if os.path.exists(lowerCAmelCase__		):
            snake_case__  :    Tuple    						=      os.listdir(lowerCAmelCase__		)
            for file in files:
                  try:
                        with open(os.path.join(lowerCAmelCase__     ,   lowerCAmelCase__		)     ,   encoding="""utf-8"""		) as f:
                              snake_case__  :    List[str]    						=      f.read()
                  except UnicodeDecodeError as e:
                        raise ValueError(f"Could not open {os.path.join(lowerCAmelCase__     ,   lowerCAmelCase__		)}."		) from e

      return _artifact





def 						__snake_case(				)		-> Dict:







      class     UpperCAmelCase_       :
          """simple docstring"""







          def __init__(    self					:     Dict ,     snake_case_					:     str						):
                snake_case__  :    List[Any]    						=      name
                snake_case__  :    Tuple    						=      []




          def __str__(    self					:     Optional[Any]						):
                return self.name




          def lowerCamelCase       (    self					:     List[Any] ,     snake_case_					:     str						):
                self.paths.append({"""name""": self.name, """path""": path}						)
      snake_case__  :    Dict[str, Artifact]    						=      {}

      snake_case__  :    Any    						=      filter(os.path.isdir     ,   os.listdir()		)
      for directory in directories:
            snake_case__  :    int    						=      directory
            if artifact_name not in _available_artifacts:
                  snake_case__  :    Optional[Any]    						=      Artifact(lowerCAmelCase__		)

                  _available_artifacts[artifact_name].add_path(lowerCAmelCase__		)

      return _available_artifacts


if __name__ == "__main__":
       __a          =		get_job_links()
       __a          =		retrieve_available_artifacts()

       __a          =		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          =		{
           v: {
               """failed""": [],
               """failures""": {},
           }
           for v in docs.values()
       }

       # Link to the GitHub Action job
       __a          =		github_actions_job_links.get("run_doctests")

       __a          =		available_artifacts["""doc_tests_gpu_test_reports"""].paths[0]
       __a          =		retrieve_artifact(artifact_path["name"])
       if "stats" in artifact:
              __a          =		handle_test_results(artifact["stats"])
              __a          =		failed
              __a          =		success
              __a          =		time_spent[1:-1] + """, """

              __a          =		extract_first_line_failure(artifact["failures_short"])
              for line in artifact["summary_short"].split("\n"):
                     if re.search("FAILED", line):
                            __a          =		line.replace("FAILED ", "")
                            __a          =		line.split()[0].replace("\n", "")

                            if "::" in line:
                                   __a          =		line.split("::")
                            else:
                                   __a          =		line, line

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

                                          __a          =		all_failures[test] if test in all_failures else """N/A"""
                                          __a          =		failure
                                          break

       __a          =		Message("🤗 Results of the doc tests.", doc_test_results)
       message.post()
       message.post_reply()





"""simple docstring"""







import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


lowercase__		:				Optional[Any]          =	logging.get_logger(__name__)







class       UpperCamelCase__   (	lowercase_  ):




      """simple docstring"""




      _SCREAMING_SNAKE_CASE  				=     """linear"""
      _SCREAMING_SNAKE_CASE  				=     """cosine"""
      _SCREAMING_SNAKE_CASE  				=     """cosine_with_restarts"""
      _SCREAMING_SNAKE_CASE  				=     """polynomial"""
      _SCREAMING_SNAKE_CASE  				=     """constant"""
      _SCREAMING_SNAKE_CASE  				=     """constant_with_warmup"""
      _SCREAMING_SNAKE_CASE  				=     """piecewise_constant"""




def 				UpperCamelCase_    (    lowerCAmelCase__      :				Optimizer    ,   lowerCAmelCase__      :				int = -1						)						->   Tuple:

  """simple docstring"""




  return LambdaLR(lowerCAmelCase__    ,   lambda lowerCAmelCase__						: 1    ,   last_epoch=lowerCAmelCase__						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Optimizer    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				int = -1						)						->   str:

  """simple docstring"""




  def lr_lambda(lowerCAmelCase__      :				int						):
    if current_step < num_warmup_steps:
      return float(lowerCAmelCase__						) / float(max(1.0    ,   lowerCAmelCase__						)						)
    return 1.0

  return LambdaLR(lowerCAmelCase__    ,   lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Optimizer    ,   lowerCAmelCase__      :				str    ,   lowerCAmelCase__      :				int = -1						)						->   int:

  """simple docstring"""




  lowerCAmelCase_   :				str								= {}
  lowerCAmelCase_   :				Optional[Any]								= step_rules.split(','						)
  for rule_str in rule_list[:-1]:
    lowerCAmelCase_						,lowerCAmelCase_   :				Optional[int]								= rule_str.split(':'						)
    lowerCAmelCase_   :				List[str]								= int(lowerCAmelCase__						)
    lowerCAmelCase_   :				Optional[int]								= float(lowerCAmelCase__						)
    lowerCAmelCase_   :				Optional[Any]								= value
  lowerCAmelCase_   :				str								= float(rule_list[-1]						)

  def create_rules_function(lowerCAmelCase__      :				Optional[Any]    ,   lowerCAmelCase__      :				int						):
    def rule_func(lowerCAmelCase__      :				int						) -> float:
      lowerCAmelCase_   :				str								= sorted(rules_dict.keys()						)
      for i, sorted_step in enumerate(lowerCAmelCase__						):
        if steps < sorted_step:
          return rules_dict[sorted_steps[i]]
      return last_lr_multiple

    return rule_func

  lowerCAmelCase_   :				Dict								= create_rules_function(lowerCAmelCase__    ,   lowerCAmelCase__						)

  return LambdaLR(lowerCAmelCase__    ,   lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Optional[int]    ,   lowerCAmelCase__      :				Optional[int]    ,   lowerCAmelCase__      :				List[str]    ,   lowerCAmelCase__      :				Union[str, Any]=-1						)						->   Any:

  """simple docstring"""




  def lr_lambda(lowerCAmelCase__      :				int						):
    if current_step < num_warmup_steps:
      return float(lowerCAmelCase__						) / float(max(1    ,   lowerCAmelCase__						)						)
    return max(
        0.0    ,   float(num_training_steps - current_step						) / float(max(1    ,   num_training_steps - num_warmup_steps						)						)						)

  return LambdaLR(lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Optimizer    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				float = 0.5    ,   lowerCAmelCase__      :				int = -1						)						->   Union[str, Any]:

  """simple docstring"""




  def lr_lambda(lowerCAmelCase__      :				Any						):
    if current_step < num_warmup_steps:
      return float(lowerCAmelCase__						) / float(max(1    ,   lowerCAmelCase__						)						)
    lowerCAmelCase_   :				List[Any]								= float(current_step - num_warmup_steps						) / float(max(1    ,   num_training_steps - num_warmup_steps						)						)
    return max(0.0    ,   0.5 * (1.0 + math.cos(math.pi * float(lowerCAmelCase__						) * 2.0 * progress						))						)

  return LambdaLR(lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Optimizer    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				int    ,   lowerCAmelCase__      :				int = 1    ,   lowerCAmelCase__      :				int = -1						)						->   int:

  """simple docstring"""




  def lr_lambda(lowerCAmelCase__      :				str						):
    if current_step < num_warmup_steps:
      return float(lowerCAmelCase__						) / float(max(1    ,   lowerCAmelCase__						)						)
    lowerCAmelCase_   :				Dict								= float(current_step - num_warmup_steps						) / float(max(1    ,   num_training_steps - num_warmup_steps						)						)
    if progress >= 1.0:
      return 0.0
    return max(0.0    ,   0.5 * (1.0 + math.cos(math.pi * ((float(lowerCAmelCase__						) * progress) % 1.0)						))						)

  return LambdaLR(lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__						)

def 				UpperCamelCase_    (    lowerCAmelCase__      :				str    ,   lowerCAmelCase__      :				List[Any]    ,   lowerCAmelCase__      :				List[str]    ,   lowerCAmelCase__      :				Tuple=1e-7    ,   lowerCAmelCase__      :				Union[str, Any]=1.0    ,   lowerCAmelCase__      :				int=-1						)						->   Optional[Any]:

  """simple docstring"""




  lowerCAmelCase_   :				Any								= optimizer.defaults['lr']
  if not (lr_init > lr_end):
    raise ValueError(f"lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})"						)

  def lr_lambda(lowerCAmelCase__      :				int						):
    if current_step < num_warmup_steps:
      return float(lowerCAmelCase__						) / float(max(1    ,   lowerCAmelCase__						)						)
    elif current_step > num_training_steps:
      return lr_end / lr_init  # as LambdaLR multiplies by lr_init
    else:
      lowerCAmelCase_   :				Any								= lr_init - lr_end
      lowerCAmelCase_   :				int								= num_training_steps - num_warmup_steps
      lowerCAmelCase_   :				Dict								= 1 - (current_step - num_warmup_steps) / decay_steps
      lowerCAmelCase_   :				Dict								= lr_range * pct_remaining**power + lr_end
      return decay / lr_init  # as LambdaLR multiplies by lr_init

  return LambdaLR(lowerCAmelCase__    ,   lowerCAmelCase__    ,   lowerCAmelCase__						)


lowercase__		:				Any          =	{
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}

def 				UpperCamelCase_    (    lowerCAmelCase__      :				Union[str, SchedulerType]    ,   lowerCAmelCase__      :				Optimizer    ,   lowerCAmelCase__      :				Optional[str] = None    ,   lowerCAmelCase__      :				Optional[int] = None    ,   lowerCAmelCase__      :				Optional[int] = None    ,   lowerCAmelCase__      :				int = 1    ,   lowerCAmelCase__      :				float = 1.0    ,   lowerCAmelCase__      :				int = -1    ,   )						->   int:

  """simple docstring"""




  lowerCAmelCase_   :				List[str]								= SchedulerType(lowerCAmelCase__						)
  lowerCAmelCase_   :				List[Any]								= TYPE_TO_SCHEDULER_FUNCTION[name]
  if name == SchedulerType.CONSTANT:
    return schedule_func(lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__						)

  if name == SchedulerType.PIECEWISE_CONSTANT:
    return schedule_func(lowerCAmelCase__    ,   step_rules=lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__						)

  # All other schedulers require `num_warmup_steps`
  if num_warmup_steps is None:
    raise ValueError(f"{name} requires `num_warmup_steps`, please provide that argument."						)

  if name == SchedulerType.CONSTANT_WITH_WARMUP:
    return schedule_func(lowerCAmelCase__    ,   num_warmup_steps=lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__						)

  # All other schedulers require `num_training_steps`
  if num_training_steps is None:
    raise ValueError(f"{name} requires `num_training_steps`, please provide that argument."						)

  if name == SchedulerType.COSINE_WITH_RESTARTS:
    return schedule_func(
        lowerCAmelCase__    ,   num_warmup_steps=lowerCAmelCase__    ,   num_training_steps=lowerCAmelCase__    ,   num_cycles=lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__    ,   )

  if name == SchedulerType.POLYNOMIAL:
    return schedule_func(
        lowerCAmelCase__    ,   num_warmup_steps=lowerCAmelCase__    ,   num_training_steps=lowerCAmelCase__    ,   power=lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__    ,   )

  return schedule_func(
      lowerCAmelCase__    ,   num_warmup_steps=lowerCAmelCase__    ,   num_training_steps=lowerCAmelCase__    ,   last_epoch=lowerCAmelCase__						)





