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


snake_case	=      WebClient(token=os.environ['''CI_SLACK_BOT_TOKEN'''])







def       A_ (			_lowerCamelCase      :				str     ):
      _lowerCAmelCase		 = test_results.split(' '     )

      _lowerCAmelCase		 = 0
      _lowerCAmelCase		 = 0

      # When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
      # When it is too long, those signs are not present.
      _lowerCAmelCase		 = 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       A_ (			_lowerCamelCase      :				Dict     ):
      _lowerCAmelCase		 = {}
      _lowerCAmelCase		 = None
      _lowerCAmelCase		 = False
      for line in failures_short_lines.split('\n'     ):
            if re.search(r'_ \[doctest\]'	,	_lowerCamelCase     ):
                  _lowerCAmelCase		 = True
                  _lowerCAmelCase		 = line.split(' '     )[2]
            elif in_error and not line.split(' '     )[0].isdigit():
                  _lowerCAmelCase		 = line
                  _lowerCAmelCase		 = False

      return failures





class 							SCREAMING_SNAKE_CASE :
  """simple docstring"""







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


        """simple docstring"""



        _lowerCAmelCase		 = title

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

        # Failures and success of the modeling tests
        _lowerCAmelCase		 = doc_test_results



  @property
  def 	a (	self :       Optional[Any]       ):


        """simple docstring"""



        _lowerCAmelCase		 = [self._time_spent]
        _lowerCAmelCase		 = 0

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

              _lowerCAmelCase ,	_lowerCAmelCase ,	_lowerCAmelCase		 = int(time_parts[0]       ), int(time_parts[1]       ), float(time_parts[2]       )
              total_secs += hours * 3600 + minutes * 60 + seconds

        _lowerCAmelCase ,	_lowerCAmelCase ,	_lowerCAmelCase		 = total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
        return F"{int(__lowerCAmelCase       )}h{int(__lowerCAmelCase       )}m{int(__lowerCAmelCase       )}s"



  @property
  def 	a (	self :       List[Any]       ):


        """simple docstring"""



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



  @property
  def 	a (	self :       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 	a (	self :       List[str]       ):


        """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 	a (	self :       int       ):


        """simple docstring"""



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

        _lowerCAmelCase		 = ''
        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 	a (	self :       int       ):


        """simple docstring"""



        _lowerCAmelCase		 = [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 	a (	):


        """simple docstring"""



        _lowerCAmelCase		 = [
            {
                '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 	a (	self :       List[str]       ):


        """simple docstring"""



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

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

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



  def 	a (	self :       List[str]       ,       __lowerCAmelCase :       str       ,       __lowerCAmelCase :       List[str]       ,       __lowerCAmelCase :       List[str]       ,       __lowerCAmelCase :       Union[str, Any]       ):


        """simple docstring"""



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

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

        if job_link is not None:
              _lowerCAmelCase		 = {
                  '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 	a (	self :       List[str]       ):


        """simple docstring"""



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

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

        _lowerCAmelCase		 = sorted(self.doc_test_results.items()       ,       key=lambda __lowerCAmelCase       : t[0]       )
        for job, job_result in sorted_dict:
              if len(job_result['failures']       ):
                    _lowerCAmelCase		 = F"*Num failures* :{len(job_result['failed']       )} \n"
                    _lowerCAmelCase		 = job_result['failures']
                    _lowerCAmelCase		 = 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       A_ (			):
      _lowerCAmelCase		 = os.environ['GITHUB_RUN_ID']
      _lowerCAmelCase		 = F"https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"
      _lowerCAmelCase		 = requests.get(_lowerCamelCase     ).json()
      _lowerCAmelCase		 = {}

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

            for i in range(_lowerCamelCase     ):
                  _lowerCAmelCase		 = 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       A_ (			_lowerCamelCase      :				int     ):
      _lowerCAmelCase		 = {}

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

      return _artifact







def       A_ (			):





      class 							SCREAMING_SNAKE_CASE :
        """simple docstring"""







        def __init__(	self :       List[str]       ,       __lowerCAmelCase :       str       ):


              """simple docstring"""



              _lowerCAmelCase		 = name
              _lowerCAmelCase		 = []



        def __str__(	self :       List[str]       ):


              """simple docstring"""



              return self.name



        def 	a (	self :       List[Any]       ,       __lowerCAmelCase :       str       ):


              """simple docstring"""



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



      _lowerCAmelCase		 = {}

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

                  _available_artifacts[artifact_name].add_path(_lowerCamelCase     )

      return _available_artifacts


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

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

 # Link to the GitHub Action job
 snake_case	=      github_actions_job_links.get('''run_doctests''')

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

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

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

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

      snake_case	=      all_failures[test] if test in all_failures else 'N/A'
      snake_case	=      failure
      break

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









'''simple docstring'''

import argparse

import torch

from diffusers.pipelines.stable_diffusion.convert_from_ckpt import download_from_original_stable_diffusion_ckpt


if __name__ == "__main__":
 lowerCamelCase				:  int    		=      argparse.ArgumentParser()

 parser.add_argument(
     '--checkpoint_path', default=None, type=str, required=True, help='Path to the checkpoint to convert.'
 )
 # !wget https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml
 parser.add_argument(
     '--original_config_file',
     default=None,
     type=str,
     help='The YAML config file corresponding to the original architecture.',
 )
 parser.add_argument(
     '--num_in_channels',
     default=None,
     type=int,
     help='The number of input channels. If `None` number of input channels will be automatically inferred.',
 )
 parser.add_argument(
     '--scheduler_type',
     default='pndm',
     type=str,
     help='Type of scheduler to use. Should be one of [\'pndm\', \'lms\', \'ddim\', \'euler\', \'euler-ancestral\', \'dpm\']',
 )
 parser.add_argument(
     '--pipeline_type',
     default=None,
     type=str,
     help=(
         'The pipeline type. One of \'FrozenOpenCLIPEmbedder\', \'FrozenCLIPEmbedder\', \'PaintByExample\''
         '. If `None` pipeline will be automatically inferred.'
     ),
 )
 parser.add_argument(
     '--image_size',
     default=None,
     type=int,
     help=(
         'The image size that the model was trained on. Use 512 for Stable Diffusion v1.X and Stable Siffusion v2'
         ' Base. Use 768 for Stable Diffusion v2.'
     ),
 )
 parser.add_argument(
     '--prediction_type',
     default=None,
     type=str,
     help=(
         'The prediction type that the model was trained on. Use \'epsilon\' for Stable Diffusion v1.X and Stable'
         ' Diffusion v2 Base. Use \'v_prediction\' for Stable Diffusion v2.'
     ),
 )
 parser.add_argument(
     '--extract_ema',
     action='store_true',
     help=(
         'Only relevant for checkpoints that have both EMA and non-EMA weights. Whether to extract the EMA weights'
         ' or not. Defaults to `False`. Add `--extract_ema` to extract the EMA weights. EMA weights usually yield'
         ' higher quality images for inference. Non-EMA weights are usually better to continue fine-tuning.'
     ),
 )
 parser.add_argument(
     '--upcast_attention',
     action='store_true',
     help=(
         'Whether the attention computation should always be upcasted. This is necessary when running stable'
         ' diffusion 2.1.'
     ),
 )
 parser.add_argument(
     '--from_safetensors',
     action='store_true',
     help='If `--checkpoint_path` is in `safetensors` format, load checkpoint with safetensors instead of PyTorch.',
 )
 parser.add_argument(
     '--to_safetensors',
     action='store_true',
     help='Whether to store pipeline in safetensors format or not.',
 )
 parser.add_argument('--dump_path', default=None, type=str, required=True, help='Path to the output model.')
 parser.add_argument('--device', type=str, help='Device to use (e.g. cpu, cuda:0, cuda:1, etc.)')
 parser.add_argument(
     '--stable_unclip',
     type=str,
     default=None,
     required=False,
     help='Set if this is a stable unCLIP model. One of \'txt2img\' or \'img2img\'.',
 )
 parser.add_argument(
     '--stable_unclip_prior',
     type=str,
     default=None,
     required=False,
     help='Set if this is a stable unCLIP txt2img model. Selects which prior to use. If `--stable_unclip` is set to `txt2img`, the karlo prior (https://huggingface.co/kakaobrain/karlo-v1-alpha/tree/main/prior) is selected by default.',
 )
 parser.add_argument(
     '--clip_stats_path',
     type=str,
     help='Path to the clip stats file. Only required if the stable unclip model\'s config specifies `model.params.noise_aug_config.params.clip_stats_path`.',
     required=False,
 )
 parser.add_argument(
     '--controlnet', action='store_true', default=None, help='Set flag if this is a controlnet checkpoint.'
 )
 parser.add_argument('--half', action='store_true', help='Save weights in half precision.')
 parser.add_argument(
     '--vae_path',
     type=str,
     default=None,
     required=False,
     help='Set to a path, hub id to an already converted vae to not convert it again.',
 )
 lowerCamelCase				:  str    		=      parser.parse_args()

 lowerCamelCase				:  Tuple    		=      download_from_original_stable_diffusion_ckpt(
     checkpoint_path=args.checkpoint_path,
     original_config_file=args.original_config_file,
     image_size=args.image_size,
     prediction_type=args.prediction_type,
     model_type=args.pipeline_type,
     extract_ema=args.extract_ema,
     scheduler_type=args.scheduler_type,
     num_in_channels=args.num_in_channels,
     upcast_attention=args.upcast_attention,
     from_safetensors=args.from_safetensors,
     device=args.device,
     stable_unclip=args.stable_unclip,
     stable_unclip_prior=args.stable_unclip_prior,
     clip_stats_path=args.clip_stats_path,
     controlnet=args.controlnet,
     vae_path=args.vae_path,
 )

 if args.half:
  pipe.to(torch_dtype=torch.floataa)

 if args.controlnet:
  # only save the controlnet model
  pipe.controlnet.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)
 else:
  pipe.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)




