'''simple docstring'''






import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class    a (       __lowerCAmelCase						,  __lowerCAmelCase						,  unittest.TestCase				):



  """simple docstring"""
  __lowerCAmelCase				      =					StableDiffusionSAGPipeline
  __lowerCAmelCase				      =					TEXT_TO_IMAGE_PARAMS
  __lowerCAmelCase				      =					TEXT_TO_IMAGE_BATCH_PARAMS
  __lowerCAmelCase				      =					TEXT_TO_IMAGE_IMAGE_PARAMS
  __lowerCAmelCase				      =					TEXT_TO_IMAGE_IMAGE_PARAMS
  __lowerCAmelCase				      =					False







  def    lowercase_  (							self					):

    '''simple docstring'''







    torch.manual_seed(0					)
    __UpperCAmelCase:					Optional[Any]		= UNetaDConditionModel(
        block_out_channels=(32, 64)   ,    layers_per_block=2   ,    sample_size=32   ,    in_channels=4   ,    out_channels=4   ,    down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")   ,    up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")   ,    cross_attention_dim=32   ,    )
    __UpperCAmelCase:					Dict		= DDIMScheduler(
        beta_start=0.0_0_0_8_5   ,    beta_end=0.0_1_2   ,    beta_schedule="""scaled_linear"""   ,    clip_sample=snake_case_   ,    set_alpha_to_one=snake_case_   ,    )
    torch.manual_seed(0					)
    __UpperCAmelCase:					Any		= AutoencoderKL(
        block_out_channels=[32, 64]   ,    in_channels=3   ,    out_channels=3   ,    down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]   ,    up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]   ,    latent_channels=4   ,    )
    torch.manual_seed(0					)
    __UpperCAmelCase:					Dict		= CLIPTextConfig(
        bos_token_id=0   ,    eos_token_id=2   ,    hidden_size=32   ,    intermediate_size=37   ,    layer_norm_eps=1e-0_5   ,    num_attention_heads=4   ,    num_hidden_layers=5   ,    pad_token_id=1   ,    vocab_size=1000   ,    )
    __UpperCAmelCase:					Optional[int]		= CLIPTextModel(snake_case_					)
    __UpperCAmelCase:					Any		= CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""					)

    __UpperCAmelCase:					List[str]		= {
        """unet""": unet,
        """scheduler""": scheduler,
        """vae""": vae,
        """text_encoder""": text_encoder,
        """tokenizer""": tokenizer,
        """safety_checker""": None,
        """feature_extractor""": None,
    }
    return components







  def    lowercase_  (							self   ,    snake_case_   ,    snake_case_=0					):

    '''simple docstring'''







    if str(snake_case_					).startswith("""mps"""					):
      __UpperCAmelCase:					List[Any]		= torch.manual_seed(snake_case_					)
    else:
      __UpperCAmelCase:					Optional[int]		= torch.Generator(device=snake_case_					).manual_seed(snake_case_					)
    __UpperCAmelCase:					str		= {
        """prompt""": """.""",
        """generator""": generator,
        """num_inference_steps""": 2,
        """guidance_scale""": 1.0,
        """sag_scale""": 1.0,
        """output_type""": """numpy""",
    }
    return inputs







  def    lowercase_  (							self					):

    '''simple docstring'''







    super().test_inference_batch_single_identical(expected_max_diff=3e-3					)




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



  """simple docstring"""







  def    lowercase_  (							self					):

    '''simple docstring'''







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







  def    lowercase_  (							self					):

    '''simple docstring'''







    __UpperCAmelCase:					List[str]		= StableDiffusionSAGPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4"""					)
    __UpperCAmelCase:					Optional[int]		= sag_pipe.to(snake_case_					)
    sag_pipe.set_progress_bar_config(disable=snake_case_					)

    __UpperCAmelCase:					Union[str, Any]		= """."""
    __UpperCAmelCase:					int		= torch.manual_seed(0					)
    __UpperCAmelCase:					Tuple		= sag_pipe(
        [prompt]   ,    generator=snake_case_   ,    guidance_scale=7.5   ,    sag_scale=1.0   ,    num_inference_steps=20   ,    output_type="""np"""					)

    __UpperCAmelCase:					str		= output.images

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

    assert image.shape == (1, 512, 512, 3)
    __UpperCAmelCase:					List[Any]		= np.array([0.1_5_6_8, 0.1_7_3_8, 0.1_6_9_5, 0.1_6_9_3, 0.1_5_0_7, 0.1_7_0_5, 0.1_5_4_7, 0.1_7_5_1, 0.1_9_4_9]					)

    assert np.abs(image_slice.flatten() - expected_slice					).max() < 5e-2







  def    lowercase_  (							self					):

    '''simple docstring'''







    __UpperCAmelCase:					str		= StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""					)
    __UpperCAmelCase:					Any		= sag_pipe.to(snake_case_					)
    sag_pipe.set_progress_bar_config(disable=snake_case_					)

    __UpperCAmelCase:					Dict		= """."""
    __UpperCAmelCase:					Optional[Any]		= torch.manual_seed(0					)
    __UpperCAmelCase:					List[Any]		= sag_pipe(
        [prompt]   ,    generator=snake_case_   ,    guidance_scale=7.5   ,    sag_scale=1.0   ,    num_inference_steps=20   ,    output_type="""np"""					)

    __UpperCAmelCase:					int		= output.images

    __UpperCAmelCase:					List[str]		= image[0, -3:, -3:, -1]

    assert image.shape == (1, 512, 512, 3)
    __UpperCAmelCase:					Any		= np.array([0.3_4_5_9, 0.2_8_7_6, 0.2_5_3_7, 0.3_0_0_2, 0.2_6_7_1, 0.2_1_6_0, 0.3_0_2_6, 0.2_2_6_2, 0.2_3_7_1]					)

    assert np.abs(image_slice.flatten() - expected_slice					).max() < 5e-2







  def    lowercase_  (							self					):

    '''simple docstring'''







    __UpperCAmelCase:					Any		= StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""					)
    __UpperCAmelCase:					List[Any]		= sag_pipe.to(snake_case_					)
    sag_pipe.set_progress_bar_config(disable=snake_case_					)

    __UpperCAmelCase:					List[str]		= """."""
    __UpperCAmelCase:					Union[str, Any]		= torch.manual_seed(0					)
    __UpperCAmelCase:					Tuple		= sag_pipe(
        [prompt]   ,    width=768   ,    height=512   ,    generator=snake_case_   ,    guidance_scale=7.5   ,    sag_scale=1.0   ,    num_inference_steps=20   ,    output_type="""np"""   ,    )

    __UpperCAmelCase:					List[Any]		= output.images

    assert image.shape == (1, 512, 768, 3)
'''simple docstring'''






import os
import zipfile

import requests
from get_ci_error_statistics import download_artifact, get_artifacts_links






def  UpperCamelCase__					(					_lowercase    :				Union[str, Any]   ,    _lowercase    :				List[Any]=7     )							->	Union[str, Any]:
  __UpperCAmelCase:					Optional[int]		= None
  if token is not None:
    __UpperCAmelCase:					List[Any]		= {"""Accept""": """application/vnd.github+json""", """Authorization""": F'''Bearer {token}'''}

  # The id of a workflow (not of a workflow run)
  __UpperCAmelCase:					List[str]		= """636036"""

  __UpperCAmelCase:					Optional[int]		= F'''https://api.github.com/repos/huggingface/transformers/actions/workflows/{workflow_id}/runs'''
  # On `main` branch + event being `schedule` + not returning PRs + only `num_runs` results
  url += F'''?branch=main&event=schedule&exclude_pull_requests=true&per_page={num_runs}'''

  __UpperCAmelCase:					List[Any]		= requests.get(_lowercase   ,    headers=_lowercase     ).json()

  return result["workflow_runs"]






def  UpperCamelCase__					(					_lowercase    :				Optional[int]     )							->	List[str]:
  __UpperCAmelCase:					str		= get_daily_ci_runs(_lowercase     )
  __UpperCAmelCase:					str		= None
  for workflow_run in workflow_runs:
    if workflow_run["status"] == "completed":
      __UpperCAmelCase:					Optional[int]		= workflow_run["""id"""]
      break

  return workflow_run_id






def  UpperCamelCase__					(					_lowercase    :				Optional[int]   ,    _lowercase    :				List[str]   ,    _lowercase    :				List[str]     )							->	Dict:
  __UpperCAmelCase:					List[Any]		= get_last_daily_ci_runs(_lowercase     )
  if workflow_run_id is not None:
    __UpperCAmelCase:					List[Any]		= get_artifacts_links(worflow_run_id=_lowercase   ,    token=_lowercase     )
    for artifact_name in artifact_names:
      if artifact_name in artifacts_links:
        __UpperCAmelCase:					str		= artifacts_links[artifact_name]
        download_artifact(
            artifact_name=_lowercase   ,    artifact_url=_lowercase   ,    output_dir=_lowercase   ,    token=_lowercase     )







def  UpperCamelCase__					(					_lowercase    :				str   ,    _lowercase    :				Tuple   ,    _lowercase    :				Tuple     )							->	int:
  get_last_daily_ci_artifacts(_lowercase   ,    _lowercase   ,    _lowercase     )

  __UpperCAmelCase:					int		= {}
  for artifact_name in artifact_names:
    __UpperCAmelCase:					Optional[Any]		= os.path.join(_lowercase   ,    F'''{artifact_name}.zip'''     )
    if os.path.isfile(_lowercase     ):
      __UpperCAmelCase:					Any		= {}
      with zipfile.ZipFile(_lowercase     ) as z:
        for filename in z.namelist():
          if not os.path.isdir(_lowercase     ):
            # read the file
            with z.open(_lowercase     ) as f:
              __UpperCAmelCase:					Dict		= f.read().decode("""UTF-8"""     )

  return results