






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


_UpperCamelCase		:  List[Any]						=		WebClient(token=os.environ['CI_SLACK_BOT_TOKEN'])
def 						_SCREAMING_SNAKE_CASE (    __snake_case			:  Dict    ):





  '''simple docstring'''





  lowercase							    =       test_results.split(' '    )

  lowercase							    =       0
  lowercase							    =       0

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

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

  return failed, success, time_spent
def 						_SCREAMING_SNAKE_CASE (    __snake_case			:  Tuple    ):





  '''simple docstring'''





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

  return failures

class        a     :







    def __init__(			self							,		_lowerCamelCase							,		_lowerCamelCase       ):
      lowercase							    =       title

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

      # Failures and success of the modeling tests
      lowercase							    =       doc_test_results







    @property
    def 			UpperCamelCase_      (			self       ):
      lowercase							    =       [self._time_spent]
      lowercase							    =       0

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

        lowercase							    =       int(time_parts[0]       ), int(time_parts[1]       ), float(time_parts[2]       )
        total_secs += hours * 3_6_0_0 + minutes * 6_0 + seconds

      lowercase							    =       total_secs // 3_6_0_0, (total_secs % 3_6_0_0) // 6_0, total_secs % 6_0
      return F'{int(__A       )}h{int(__A       )}m{int(__A       )}s'







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







    @property
    def 			UpperCamelCase_      (			self       ):
      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 			UpperCamelCase_      (			self       ):
      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 			UpperCamelCase_      (			self       ):
      lowercase							    =       4_0
      lowercase							    =       {k: v["failed"] for k, v in doc_test_results.items() if isinstance(__A							,		__A       )}

      lowercase							    =       ""
      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 			UpperCamelCase_      (			self       ):
      lowercase							    =       [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 			UpperCamelCase_      (			):
      lowercase							    =       [
          {
              "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 			UpperCamelCase_      (			self       ):
      print('Sending the following payload'       )
      print(json.dumps({'blocks': json.loads(self.payload       )}       )       )

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

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







    def 			UpperCamelCase_      (			self							,		_lowerCamelCase							,		_lowerCamelCase							,		_lowerCamelCase							,		_lowerCamelCase       ):
      lowercase							    =       ""
      for key, value in failures.items():
        lowercase							    =       value[:2_0_0] + " [Truncated]" if len(__A       ) > 2_5_0 else value
        failures_text += F'*{key}*\n_{value}_\n\n'

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

      if job_link is not None:
        lowercase							    =       {
            "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 			UpperCamelCase_      (			self       ):





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

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

      lowercase							    =       sorted(self.doc_test_results.items()							,		key=lambda _lowerCamelCase       : t[0]       )
      for job, job_result in sorted_dict:
        if len(job_result['failures']       ):
          lowercase							    =       F'*Num failures* :{len(job_result["failed"]       )} \n'
          lowercase							    =       job_result["failures"]
          lowercase							    =       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 						_SCREAMING_SNAKE_CASE (    ):





  '''simple docstring'''





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

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

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

  return {}
def 						_SCREAMING_SNAKE_CASE (    __snake_case			:  Any    ):





  '''simple docstring'''





  lowercase							    =       {}

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

  return _artifact
def 						_SCREAMING_SNAKE_CASE (    ):





  '''simple docstring'''





  class        a     :







      def __init__(			self							,		_lowerCamelCase       ):
        lowercase							    =       name
        lowercase							    =       []







      def __str__(			self       ):
        return self.name







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


  lowercase							    =       {}

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

      _available_artifacts[artifact_name].add_path(SCREAMING_SNAKE_CASE_    )

  return _available_artifacts


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

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

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

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

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

                if "::" in line:
                    _UpperCamelCase,		_UpperCamelCase		:  Tuple						=		line.split('::')
                else:
                    _UpperCamelCase,		_UpperCamelCase		:  str						=		line, line

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

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

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






import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()
class 	UpperCAmelCase_					(					UpperCamelCase		, UpperCamelCase		, UpperCamelCase		, unittest.TestCase    ):

   '''simple docstring'''




   __A :      Tuple     		=		StableUnCLIPPipeline
   __A :      Optional[int]     		=		TEXT_TO_IMAGE_PARAMS
   __A :      str     		=		TEXT_TO_IMAGE_BATCH_PARAMS
   __A :      int     		=		TEXT_TO_IMAGE_IMAGE_PARAMS
   __A :      Tuple     		=		TEXT_TO_IMAGE_IMAGE_PARAMS

   # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
   __A :      Union[str, Any]     		=		False






   def _snake_case    (		self ):






     """simple docstring"""





     lowerCamelCase :					List[str]    =				32
     lowerCamelCase :					Dict    =				embedder_hidden_size

     # prior components

     torch.manual_seed(0 )
     lowerCamelCase :					Optional[int]    =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip" )

     torch.manual_seed(0 )
     lowerCamelCase :					Optional[int]    =				CLIPTextModelWithProjection(
         CLIPTextConfig(
             bos_token_id=0						,				eos_token_id=2						,				hidden_size=__A						,				projection_dim=__A						,				intermediate_size=37						,				layer_norm_eps=1e-05						,				num_attention_heads=4						,				num_hidden_layers=5						,				pad_token_id=1						,				vocab_size=1000						,				) )

     torch.manual_seed(0 )
     lowerCamelCase :					List[Any]    =				PriorTransformer(
         num_attention_heads=2						,				attention_head_dim=12						,				embedding_dim=__A						,				num_layers=1						,				)

     torch.manual_seed(0 )
     lowerCamelCase :					Dict    =				DDPMScheduler(
         variance_type="fixed_small_log"						,				prediction_type="sample"						,				num_train_timesteps=1000						,				clip_sample=__A						,				clip_sample_range=5.0						,				beta_schedule="squaredcos_cap_v2"						,				)

     # regular denoising components

     torch.manual_seed(0 )
     lowerCamelCase :					Optional[int]    =				StableUnCLIPImageNormalizer(embedding_dim=__A )
     lowerCamelCase :					Tuple    =				DDPMScheduler(beta_schedule="squaredcos_cap_v2" )

     torch.manual_seed(0 )
     lowerCamelCase :					List[Any]    =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip" )

     torch.manual_seed(0 )
     lowerCamelCase :					str    =				CLIPTextModel(
         CLIPTextConfig(
             bos_token_id=0						,				eos_token_id=2						,				hidden_size=__A						,				projection_dim=32						,				intermediate_size=37						,				layer_norm_eps=1e-05						,				num_attention_heads=4						,				num_hidden_layers=5						,				pad_token_id=1						,				vocab_size=1000						,				) )

     torch.manual_seed(0 )
     lowerCamelCase :					Any    =				UNetaDConditionModel(
         sample_size=32						,				in_channels=4						,				out_channels=4						,				down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")						,				up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")						,				block_out_channels=(32, 64)						,				attention_head_dim=(2, 4)						,				class_embed_type="projection"						,				projection_class_embeddings_input_dim=embedder_projection_dim * 2						,				cross_attention_dim=__A						,				layers_per_block=1						,				upcast_attention=__A						,				use_linear_projection=__A						,				)

     torch.manual_seed(0 )
     lowerCamelCase :					int    =				DDIMScheduler(
         beta_schedule="scaled_linear"						,				beta_start=0.00085						,				beta_end=0.012						,				prediction_type="v_prediction"						,				set_alpha_to_one=__A						,				steps_offset=1						,				)

     torch.manual_seed(0 )
     lowerCamelCase :					Optional[Any]    =				AutoencoderKL()

     lowerCamelCase :					Optional[int]    =				{
         # prior components
         "prior_tokenizer": prior_tokenizer,
         "prior_text_encoder": prior_text_encoder,
         "prior": prior,
         "prior_scheduler": prior_scheduler,
         # image noising components
         "image_normalizer": image_normalizer,
         "image_noising_scheduler": image_noising_scheduler,
         # regular denoising components
         "tokenizer": tokenizer,
         "text_encoder": text_encoder,
         "unet": unet,
         "scheduler": scheduler,
         "vae": vae,
     }

     return components






   def _snake_case    (		self						,				__A						,				__A=0 ):






     """simple docstring"""





     if str(__A ).startswith("mps" ):
       lowerCamelCase :					Optional[int]    =				torch.manual_seed(__A )
     else:
       lowerCamelCase :					Optional[Any]    =				torch.Generator(device=__A ).manual_seed(__A )
     lowerCamelCase :					Tuple    =				{
         "prompt": "A painting of a squirrel eating a burger",
         "generator": generator,
         "num_inference_steps": 2,
         "prior_num_inference_steps": 2,
         "output_type": "numpy",
     }
     return inputs






   def _snake_case    (		self ):






     """simple docstring"""





     lowerCamelCase :					Dict    =				torch_device == "cpu"

     self._test_attention_slicing_forward_pass(test_max_difference=__A )






   def _snake_case    (		self ):






     """simple docstring"""





     lowerCamelCase :					List[Any]    =				torch_device in ["cpu", "mps"]

     self._test_inference_batch_single_identical(test_max_difference=__A )



@slow
@require_torch_gpu
class 	UpperCAmelCase_					(					unittest.TestCase    ):

   '''simple docstring'''




   def _snake_case    (		self ):






     """simple docstring"""





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






   def _snake_case    (		self ):






     """simple docstring"""





     lowerCamelCase :					Optional[Any]    =				load_numpy(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy" )

     lowerCamelCase :					str    =				StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"						,				torch_dtype=torch.floataa )
     pipe.to(__A )
     pipe.set_progress_bar_config(disable=__A )
     # stable unclip will oom when integration tests are run on a V100,
     # so turn on memory savings
     pipe.enable_attention_slicing()
     pipe.enable_sequential_cpu_offload()

     lowerCamelCase :					List[Any]    =				torch.Generator(device="cpu" ).manual_seed(0 )
     lowerCamelCase :					Dict    =				pipe("anime turle"						,				generator=__A						,				output_type="np" )

     lowerCamelCase :					Dict    =				output.images[0]

     assert image.shape == (768, 768, 3)

     assert_mean_pixel_difference(__A						,				__A )






   def _snake_case    (		self ):






     """simple docstring"""





     torch.cuda.empty_cache()
     torch.cuda.reset_max_memory_allocated()
     torch.cuda.reset_peak_memory_stats()

     lowerCamelCase :					int    =				StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"						,				torch_dtype=torch.floataa )
     lowerCamelCase :					Union[str, Any]    =				pipe.to(__A )
     pipe.set_progress_bar_config(disable=__A )
     pipe.enable_attention_slicing()
     pipe.enable_sequential_cpu_offload()

     lowerCamelCase :					Any    =				pipe(
         "anime turtle"						,				prior_num_inference_steps=2						,				num_inference_steps=2						,				output_type="np"						,				)

     lowerCamelCase :					List[str]    =				torch.cuda.max_memory_allocated()
     # make sure that less than 7 GB is allocated
     assert mem_bytes < 7 * 10**9



