
'''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


__lowerCAmelCase         =    WebClient(token=os.environ["""CI_SLACK_BOT_TOKEN"""])





def 					UpperCAmelCase_		(__a      :    Optional[int] ):






   """simple docstring"""

   _a				:	List[str]			=   test_results.split(' ' )

   _a				:	Any			=   0
   _a				:	Union[str, 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.
   _a				:	Any			=   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      :    Any ):






   """simple docstring"""

   _a				:	Optional[Any]			=   {}
   _a				:	Tuple			=   None
   _a				:	str			=   False
   for line in failures_short_lines.split('\n' ):
      if re.search(R'_ \[doctest\]' ,	__a ):
         _a				:	Tuple			=   True
         _a				:	str			=   line.split(' ' )[2]
      elif in_error and not line.split(' ' )[0].isdigit():
         _a				:	Dict			=   line
         _a				:	List[Any]			=   False

   return failures







class UpperCAmelCase__	:





  """simple docstring"""

  def __init__(		self      :     Tuple							,_a      :     str							,_a      :     Dict       ):


     '''simple docstring'''






     _a				:	Tuple			=   title

     _a				:	Tuple			=   doc_test_results['time_spent'].split(','       )[0]
     _a				:	List[str]			=   doc_test_results['success']
     _a				:	List[Any]			=   doc_test_results['failures']
     _a				:	str			=   self.n_success + self.n_failures

     # Failures and success of the modeling tests
     _a				:	Union[str, Any]			=   doc_test_results


  @property
  def   __lowercase		(		self      :     str       ):


     '''simple docstring'''






     _a				:	Any			=   [self._time_spent]
     _a				:	Union[str, Any]			=   0

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

        _a,  _a,  _a				:	Union[str, Any]			=   int(time_parts[0]       ), int(time_parts[1]       ), float(time_parts[2]       )
        total_secs += hours * 3600 + minutes * 60 + seconds

     _a,  _a,  _a				:	List[str]			=   total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
     return F"""{int(_a       )}h{int(_a       )}m{int(_a       )}s"""


  @property
  def   __lowercase		(		self      :     Any       ):


     '''simple docstring'''






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


  @property
  def   __lowercase		(		self      :     List[str]       ):


     '''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   __lowercase		(		self      :     int       ):


     '''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   __lowercase		(		self      :     Tuple       ):


     '''simple docstring'''






     _a				:	Any			=   40
     _a				:	List[Any]			=   {k: v['failed'] for k, v in doc_test_results.items() if isinstance(_a							,_a       )}

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

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

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

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


  @property
  def   __lowercase		(		self      :     Any       ):


     '''simple docstring'''






     _a				:	Tuple			=   [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(_a       )


  @staticmethod
  def   __lowercase		(		):


     '''simple docstring'''






     _a				:	Optional[int]			=   [
         {
             '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(_a       )}       )       )

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


  def   __lowercase		(		self      :     Union[str, Any]       ):


     '''simple docstring'''






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

     _a				:	Optional[int]			=   F"""{self.n_failures} failures out of {self.n_tests} tests,""" if self.n_failures else 'All tests passed.'

     _a				:	Union[str, Any]			=   client.chat_postMessage(
         channel=os.environ['CI_SLACK_CHANNEL_ID_DAILY']							,blocks=self.payload							,text=_a							,)


  def   __lowercase		(		self      :     Dict							,_a      :     List[Any]							,_a      :     Any							,_a      :     List[Any]							,_a      :     Tuple       ):


     '''simple docstring'''






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

     _a				:	Optional[int]			=   job_name
     _a				:	Optional[Any]			=   {'type': 'section', 'text': {'type': 'mrkdwn', 'text': text}}

     if job_link is not None:
        _a				:	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   __lowercase		(		self      :     Any       ):


     '''simple docstring'''






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

     _a				:	int			=   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'       )

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

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

           time.sleep(1       )





def 					UpperCAmelCase_		():






   """simple docstring"""

   _a				:	Dict			=   os.environ['GITHUB_RUN_ID']
   _a				:	List[Any]			=   f"""https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"""
   _a				:	Union[str, Any]			=   requests.get(__a ).json()
   _a				:	Union[str, Any]			=   {}

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

      for i in range(__a ):
         _a				:	Optional[Any]			=   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      :    str ):






   """simple docstring"""

   _a				:	Optional[int]			=   {}

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

   return _artifact





def 					UpperCAmelCase_		():






   """simple docstring"""

   class UpperCAmelCase__	:





     """simple docstring"""

     def __init__(		self      :     Optional[Any]							,_a      :     str       ):


        '''simple docstring'''






        _a				:	Union[str, Any]			=   name
        _a				:	str			=   []


     def __str__(		self      :     Tuple       ):


        '''simple docstring'''






        return self.name


     def   __lowercase		(		self      :     Dict							,_a      :     str       ):


        '''simple docstring'''






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


   _a				:	Dict[str, Artifact]			=   {}

   _a				:	Any			=   filter(os.path.isdir ,	os.listdir() )
   for directory in directories:
      _a				:	Dict			=   directory
      if artifact_name not in _available_artifacts:
         _a				:	int			=   Artifact(__a )

         _available_artifacts[artifact_name].add_path(__a )

   return _available_artifacts


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

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

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

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

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

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

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

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

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





'''simple docstring'''







import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel

from diffusers import DDIMScheduler, LDMPipeline, UNetaDModel, VQModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()







class UpperCAmelCase__	( unittest.TestCase	):





  """simple docstring"""

  @property
  def   __lowercase		(		self      :     Union[str, Any]       ):


     '''simple docstring'''






     torch.manual_seed(0       )
     _a				:	Any			=   UNetaDModel(
         block_out_channels=(32, 64)							,layers_per_block=2							,sample_size=32							,in_channels=3							,out_channels=3							,down_block_types=('DownBlock2D', 'AttnDownBlock2D')							,up_block_types=('AttnUpBlock2D', 'UpBlock2D')							,)
     return model


  @property
  def   __lowercase		(		self      :     Optional[int]       ):


     '''simple docstring'''






     torch.manual_seed(0       )
     _a				:	Union[str, Any]			=   VQModel(
         block_out_channels=[32, 64]							,in_channels=3							,out_channels=3							,down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']							,up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']							,latent_channels=3							,)
     return model


  @property
  def   __lowercase		(		self      :     Optional[int]       ):


     '''simple docstring'''






     torch.manual_seed(0       )
     _a				:	Any			=   CLIPTextConfig(
         bos_token_id=0							,eos_token_id=2							,hidden_size=32							,intermediate_size=37							,layer_norm_eps=1E-05							,num_attention_heads=4							,num_hidden_layers=5							,pad_token_id=1							,vocab_size=1000							,)
     return CLIPTextModel(_a       )


  def   __lowercase		(		self      :     Tuple       ):


     '''simple docstring'''






     _a				:	Dict			=   self.dummy_uncond_unet
     _a				:	List[Any]			=   DDIMScheduler()
     _a				:	List[Any]			=   self.dummy_vq_model

     _a				:	str			=   LDMPipeline(unet=_a							,vqvae=_a							,scheduler=_a       )
     ldm.to(_a       )
     ldm.set_progress_bar_config(disable=_a       )

     _a				:	List[str]			=   torch.manual_seed(0       )
     _a				:	List[str]			=   ldm(generator=_a							,num_inference_steps=2							,output_type='numpy'       ).images

     _a				:	List[str]			=   torch.manual_seed(0       )
     _a				:	Union[str, Any]			=   ldm(generator=_a							,num_inference_steps=2							,output_type='numpy'							,return_dict=_a       )[0]

     _a				:	Tuple			=   image[0, -3:, -3:, -1]
     _a				:	Optional[Any]			=   image_from_tuple[0, -3:, -3:, -1]

     assert image.shape == (1, 64, 64, 3)
     _a				:	int			=   np.array([0.8512, 0.818, 0.6411, 0.6808, 0.4465, 0.5618, 0.46, 0.6231, 0.5172]       )
     _a				:	Any			=   1E-2 if torch_device != 'mps' else 3E-2

     assert np.abs(image_slice.flatten() - expected_slice       ).max() < tolerance
     assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < tolerance







@slow
@require_torch
class UpperCAmelCase__	( unittest.TestCase	):





  """simple docstring"""

  def   __lowercase		(		self      :     Optional[Any]       ):


     '''simple docstring'''






     _a				:	List[str]			=   LDMPipeline.from_pretrained('CompVis/ldm-celebahq-256'       )
     ldm.to(_a       )
     ldm.set_progress_bar_config(disable=_a       )

     _a				:	Optional[int]			=   torch.manual_seed(0       )
     _a				:	Dict			=   ldm(generator=_a							,num_inference_steps=5							,output_type='numpy'       ).images

     _a				:	str			=   image[0, -3:, -3:, -1]

     assert image.shape == (1, 256, 256, 3)
     _a				:	Optional[int]			=   np.array([0.4399, 0.4_4975, 0.4_6825, 0.474, 0.4359, 0.4581, 0.4_5095, 0.4341, 0.4447]       )
     _a				:	int			=   1E-2 if torch_device != 'mps' else 3E-2

     assert np.abs(image_slice.flatten() - expected_slice       ).max() < tolerance



