




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


lowerCamelCase__			:       Any         =							WebClient(token=os.environ['CI_SLACK_BOT_TOKEN'])

def  UpperCAmelCase_							(				__UpperCAmelCase						:     Dict							) ->			Optional[int]:
      SCREAMING_SNAKE_CASE_     						=      test_results.split(' '							)

      SCREAMING_SNAKE_CASE_     						=      0
      SCREAMING_SNAKE_CASE_     						=      0

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

      for i, expression in enumerate(__UpperCAmelCase							):
            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_							(				__UpperCAmelCase						:     Tuple							) ->			str:
      SCREAMING_SNAKE_CASE_     						=      {}
      SCREAMING_SNAKE_CASE_     						=      None
      SCREAMING_SNAKE_CASE_     						=      False
      for line in failures_short_lines.split('\n'							):
            if re.search(r'_ \[doctest\]'      ,      __UpperCAmelCase							):
                  SCREAMING_SNAKE_CASE_     						=      True
                  SCREAMING_SNAKE_CASE_     						=      line.split(' '							)[2]
            elif in_error and not line.split(' '							)[0].isdigit():
                  SCREAMING_SNAKE_CASE_     						=      line
                  SCREAMING_SNAKE_CASE_     						=      False

      return failures





class 	lowerCamelCase_							:


 '''simple docstring'''







 def __init__(							self  :		List[Any]    ,    _lowerCAmelCase  :		str    ,    _lowerCAmelCase  :		Dict						):
       SCREAMING_SNAKE_CASE_     						=      title

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

       # Failures and success of the modeling tests
       SCREAMING_SNAKE_CASE_     						=      doc_test_results



 @property
 def 				lowerCAmelCase_  (							self  :		int						):
       SCREAMING_SNAKE_CASE_     						=      [self._time_spent]
       SCREAMING_SNAKE_CASE_     						=      0

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

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

       SCREAMING_SNAKE_CASE_						,			SCREAMING_SNAKE_CASE_						,			SCREAMING_SNAKE_CASE_     						=      total_secs // 3_600, (total_secs % 3_600) // 60, total_secs % 60
       return F"{int(_lowerCAmelCase						)}h{int(_lowerCAmelCase						)}m{int(_lowerCAmelCase						)}s"



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



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

       SCREAMING_SNAKE_CASE_     						=      ''
       for category, failures in category_failures.items():
             if len(_lowerCAmelCase						) == 0:
                   continue

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

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

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



 @property
 def 				lowerCAmelCase_  (							self  :		List[Any]						):
       SCREAMING_SNAKE_CASE_     						=      [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(_lowerCAmelCase						)



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

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



 def 				lowerCAmelCase_  (							self  :		List[Any]						):
       print('Sending the following payload'						)
       print(json.dumps({'blocks': json.loads(self.payload						)}						)						)

       SCREAMING_SNAKE_CASE_     						=      F"{self.n_failures} failures out of {self.n_tests} tests," if self.n_failures else 'All tests passed.'

       SCREAMING_SNAKE_CASE_     						=      client.chat_postMessage(
           channel=os.environ['CI_SLACK_CHANNEL_ID_DAILY']    ,    blocks=self.payload    ,    text=_lowerCAmelCase    ,    )



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

       SCREAMING_SNAKE_CASE_     						=      job_name
       SCREAMING_SNAKE_CASE_     						=      {'type': 'section', 'text': {'type': 'mrkdwn', 'text': text}}

       if job_link is not None:
             SCREAMING_SNAKE_CASE_     						=      {
                 '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  :		Any						):
       if self.thread_ts is None:
             raise ValueError('Can only post reply if a post has been made.'						)

       SCREAMING_SNAKE_CASE_     						=      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'						)

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

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

                   time.sleep(1						)


def  UpperCAmelCase_							(				) ->			List[str]:
      SCREAMING_SNAKE_CASE_     						=      os.environ['GITHUB_RUN_ID']
      SCREAMING_SNAKE_CASE_     						=      f"https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"
      SCREAMING_SNAKE_CASE_     						=      requests.get(__UpperCAmelCase							).json()
      SCREAMING_SNAKE_CASE_     						=      {}

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

            for i in range(__UpperCAmelCase							):
                  SCREAMING_SNAKE_CASE_     						=      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.'      ,      __UpperCAmelCase							)

      return {}

def  UpperCAmelCase_							(				__UpperCAmelCase						:     str							) ->			Tuple:
      SCREAMING_SNAKE_CASE_     						=      {}

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

      return _artifact

def  UpperCAmelCase_							(				) ->			Tuple:





      class 	lowerCamelCase_							:


       '''simple docstring'''







       def __init__(							self  :		List[str]    ,    _lowerCAmelCase  :		str						):
             SCREAMING_SNAKE_CASE_     						=      name
             SCREAMING_SNAKE_CASE_     						=      []



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




       def 				lowerCAmelCase_  (							self  :		List[str]    ,    _lowerCAmelCase  :		str						):
             self.paths.append({'name': self.name, 'path': path}						)


      SCREAMING_SNAKE_CASE_     						=      {}

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

                  _available_artifacts[artifact_name].add_path(__UpperCAmelCase							)

      return _available_artifacts


if __name__ == "__main__":
   lowerCamelCase__			:       Any         =							get_job_links()
   lowerCamelCase__			:       List[Any]         =							retrieve_available_artifacts()

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

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

   lowerCamelCase__			:       List[Any]         =							available_artifacts['doc_tests_gpu_test_reports'].paths[0]
   lowerCamelCase__			:       List[str]         =							retrieve_artifact(artifact_path['name'])
   if "stats" in artifact:
      lowerCamelCase__    ,							lowerCamelCase__    ,							lowerCamelCase__			:       str         =							handle_test_results(artifact['stats'])
      lowerCamelCase__			:       str         =							failed
      lowerCamelCase__			:       Union[str, Any]         =							success
      lowerCamelCase__			:       Dict         =							time_spent[1:-1] + ', '

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

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

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

                  lowerCamelCase__			:       int         =							all_failures[test] if test in all_failures else 'N/A'
                  lowerCamelCase__			:       Optional[Any]         =							failure
                  break

   lowerCamelCase__			:       int         =							Message('🤗 Results of the doc tests.', doc_test_results)
   message.post()
   message.post_reply()





import gc
import threading
import time

import psutil
import torch





class 	lowerCamelCase_							:


 '''simple docstring'''







 def __init__(							self  :		Optional[Any]						):
       SCREAMING_SNAKE_CASE_     						=      psutil.Process()
       SCREAMING_SNAKE_CASE_     						=      False



 def 				lowerCAmelCase_  (							self  :		Dict						):
       SCREAMING_SNAKE_CASE_     						=      -1

       while True:
             SCREAMING_SNAKE_CASE_     						=      max(self.process.memory_info().rss    ,    self.cpu_memory_peak						)

             # can't sleep or will not catch the peak right (this comment is here on purpose)
             if not self.peak_monitoring:
                   break



 def 				lowerCAmelCase_  (							self  :		List[str]						):
       SCREAMING_SNAKE_CASE_     						=      True
       SCREAMING_SNAKE_CASE_     						=      threading.Thread(target=self.peak_monitor						)
       SCREAMING_SNAKE_CASE_     						=      True
       self.thread.start()




 def 				lowerCAmelCase_  (							self  :		List[str]						):
       SCREAMING_SNAKE_CASE_     						=      False
       self.thread.join()
       return self.cpu_memory_peak


lowerCamelCase__			:       List[str]         =							PeakCPUMemory()

def  UpperCAmelCase_							(				) ->			Tuple:
      # Time
      SCREAMING_SNAKE_CASE_     						=      {'time': time.time()}

      gc.collect()
      torch.cuda.empty_cache()

      # CPU mem
      SCREAMING_SNAKE_CASE_     						=      psutil.Process().memory_info().rss
      cpu_peak_tracker.start()

      # GPU mem
      for i in range(torch.cuda.device_count()							):
            SCREAMING_SNAKE_CASE_     						=      torch.cuda.memory_allocated(__UpperCAmelCase							)
      torch.cuda.reset_peak_memory_stats()

      return measures

def  UpperCAmelCase_							(				__UpperCAmelCase						:     Optional[int]							) ->			Optional[Any]:
      # Time
      SCREAMING_SNAKE_CASE_     						=      {'time': time.time() - start_measures['time']}

      gc.collect()
      torch.cuda.empty_cache()

      # CPU mem
      SCREAMING_SNAKE_CASE_     						=      (psutil.Process().memory_info().rss - start_measures['cpu']) / 2**20
      SCREAMING_SNAKE_CASE_     						=      (cpu_peak_tracker.stop() - start_measures['cpu']) / 2**20

      # GPU mem
      for i in range(torch.cuda.device_count()							):
            SCREAMING_SNAKE_CASE_     						=      (torch.cuda.memory_allocated(__UpperCAmelCase							) - start_measures[str(__UpperCAmelCase							)]) / 2**20
            SCREAMING_SNAKE_CASE_     						=      (torch.cuda.max_memory_allocated(__UpperCAmelCase							) - start_measures[str(__UpperCAmelCase							)]) / 2**20

      return measures





def  UpperCAmelCase_							(				__UpperCAmelCase						:     Optional[int]      ,      __UpperCAmelCase						:     Any							) ->			Optional[Any]:
      print(f"{description}:"							)
      print(f"- Time: {measures['time']:.2f}s"							)
      for i in range(torch.cuda.device_count()							):
            print(f"- GPU {i} allocated: {measures[str(__UpperCAmelCase							)]:.2f}MiB"							)
            SCREAMING_SNAKE_CASE_     						=      measures[f"{i}-peak"]
            print(f"- GPU {i} peak: {peak:.2f}MiB"							)
      print(f"- CPU RAM allocated: {measures['cpu']:.2f}MiB"							)
      print(f"- CPU RAM peak: {measures['cpu-peak']:.2f}MiB"							)