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


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

def 						UpperCamelCase_     (	__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    snake_case_ :    Any     			=						test_results.split(" "       )

    snake_case_ :    List[str]     			=						0
    snake_case_ :    Dict     			=						0

    # When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
    # When it is too long, those signs are not present.
    snake_case_ :    Tuple     			=						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_     (	__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    snake_case_ :    Tuple     			=						{}
    snake_case_ :    str     			=						None
    snake_case_ :    Any     			=						False
    for line in failures_short_lines.split("\n"       ):
        if re.search(r"_ \[doctest\]",		_A       ):
            snake_case_ :    Optional[int]     			=						True
            snake_case_ :    Tuple     			=						line.split(" "       )[2]
        elif in_error and not line.split(" "       )[0].isdigit():
            snake_case_ :    List[Any]     			=						line
            snake_case_ :    Any     			=						False

    return failures






class    UpperCAmelCase_				:





 def __init__(    self ,       lowercase_ ,       lowercase_):
     snake_case_ :    int     			=						title

     snake_case_ :    List[str]     			=						doc_test_results["time_spent"].split(",")[0]
     snake_case_ :    Tuple     			=						doc_test_results["success"]
     snake_case_ :    Tuple     			=						doc_test_results["failures"]
     snake_case_ :    int     			=						self.n_success + self.n_failures

     # Failures and success of the modeling tests
     snake_case_ :    List[Any]     			=						doc_test_results





 @property
 def        snake_case__		(    self):
     snake_case_ :    Dict     			=						[self._time_spent]
     snake_case_ :    Dict     			=						0

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

         snake_case_			, snake_case_			, snake_case_ :    Optional[int]     			=						int(time_parts[0]), int(time_parts[1]), float(time_parts[2])
         total_secs += hours * 36_00 + minutes * 60 + seconds

     snake_case_			, snake_case_			, snake_case_ :    Any     			=						total_secs // 36_00, (total_secs % 36_00) // 60, total_secs % 60
     return F'{int(__UpperCamelCase)}h{int(__UpperCamelCase)}m{int(__UpperCamelCase)}s'





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





 @property
 def        snake_case__		(    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        snake_case__		(    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        snake_case__		(    self):
     snake_case_ :    List[Any]     			=						40
     snake_case_ :    Optional[int]     			=						{k: v["failed"] for k, v in doc_test_results.items() if isinstance(__UpperCamelCase ,       __UpperCamelCase)}

     snake_case_ :    Optional[int]     			=						""
     for category, failures in category_failures.items():
         if len(__UpperCamelCase) == 0:
             continue

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

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

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





 @property
 def        snake_case__		(    self):
     snake_case_ :    str     			=						[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(__UpperCamelCase)





 @staticmethod
 def        snake_case__		(    ):
     snake_case_ :    str     			=						[
         {
             "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(__UpperCamelCase)}))

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





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

     snake_case_ :    List[Any]     			=						F'{self.n_failures} failures out of {self.n_tests} tests,' if self.n_failures else "All tests passed."

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





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_ ,       lowercase_ ,       lowercase_):
     snake_case_ :    List[str]     			=						""
     for key, value in failures.items():
         snake_case_ :    Tuple     			=						value[:2_00] + " [Truncated]" if len(__UpperCamelCase) > 2_50 else value
         failures_text += F'*{key}*\n_{value}_\n\n'

     snake_case_ :    List[Any]     			=						job_name
     snake_case_ :    int     			=						{"type": "section", "text": {"type": "mrkdwn", "text": text}}

     if job_link is not None:
         snake_case_ :    Union[str, Any]     			=						{
             "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        snake_case__		(    self):






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

     snake_case_ :    Optional[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")

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

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

             time.sleep(1)


def 						UpperCamelCase_     (	):
    """simple docstring"""






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

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

        for i in range(_A       ):
            snake_case_ :    List[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_     (	__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    snake_case_ :    Dict     			=						{}

    if os.path.exists(_A       ):
        snake_case_ :    Union[str, Any]     			=						os.listdir(_A       )
        for file in files:
            try:
                with open(os.path.join(_A,		_A       ),		encoding="utf-8"       ) as f:
                    snake_case_ :    Optional[Any]     			=						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_				:





     def __init__(    self ,       lowercase_):
         snake_case_ :    str     			=						name
         snake_case_ :    Optional[int]     			=						[]





     def __str__(    self):
         return self.name





     def        snake_case__		(    self ,       lowercase_):
         self.paths.append({"name": self.name, "path": path})


    snake_case_ :    Optional[int]     			=						{}

    snake_case_ :    List[Any]     			=						filter(os.path.isdir,		os.listdir()       )
    for directory in directories:
        snake_case_ :    int     			=						directory
        if artifact_name not in _available_artifacts:
            snake_case_ :    Optional[Any]     			=						Artifact(_A       )

            _available_artifacts[artifact_name].add_path(_A       )

    return _available_artifacts


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

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

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

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

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

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

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

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

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






'''simple docstring'''
import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionInstructPixaPixPipeline,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.utils import floats_tensor, load_image, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class    UpperCAmelCase_				(	snake_case__    ,						snake_case__    ,						snake_case__    ,						unittest.TestCase					):
 UpperCAmelCase_		    =						StableDiffusionInstructPixaPixPipeline
 UpperCAmelCase_		    =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width""", """cross_attention_kwargs"""}
 UpperCAmelCase_		    =						TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
 UpperCAmelCase_		    =						IMAGE_TO_IMAGE_IMAGE_PARAMS
 UpperCAmelCase_		    =						IMAGE_TO_IMAGE_IMAGE_PARAMS





 def        snake_case__		(    self):
     torch.manual_seed(0)
     snake_case_ :    List[str]     			=						UNetaDConditionModel(
         block_out_channels=(32, 64) ,       layers_per_block=2 ,       sample_size=32 ,       in_channels=8 ,       out_channels=4 ,       down_block_types=("DownBlock2D", "CrossAttnDownBlock2D") ,       up_block_types=("CrossAttnUpBlock2D", "UpBlock2D") ,       cross_attention_dim=32 ,       )
     snake_case_ :    Any     			=						PNDMScheduler(skip_prk_steps=lowercase_)
     torch.manual_seed(0)
     snake_case_ :    Tuple     			=						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)
     snake_case_ :    Union[str, 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=10_00 ,       )
     snake_case_ :    Union[str, Any]     			=						CLIPTextModel(lowercase_)
     snake_case_ :    Optional[Any]     			=						CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

     snake_case_ :    str     			=						{
         "unet": unet,
         "scheduler": scheduler,
         "vae": vae,
         "text_encoder": text_encoder,
         "tokenizer": tokenizer,
         "safety_checker": None,
         "feature_extractor": None,
     }
     return components





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_=0):
     snake_case_ :    int     			=						floats_tensor((1, 3, 32, 32) ,       rng=random.Random(lowercase_)).to(lowercase_)
     snake_case_ :    Any     			=						image.cpu().permute(0 ,       2 ,       3 ,       1)[0]
     snake_case_ :    str     			=						Image.fromarray(np.uinta(lowercase_)).convert("RGB")
     if str(lowercase_).startswith("mps"):
         snake_case_ :    Union[str, Any]     			=						torch.manual_seed(lowercase_)
     else:
         snake_case_ :    Any     			=						torch.Generator(device=lowercase_).manual_seed(lowercase_)
     snake_case_ :    List[str]     			=						{
         "prompt": "A painting of a squirrel eating a burger",
         "image": image,
         "generator": generator,
         "num_inference_steps": 2,
         "guidance_scale": 6.0,
         "image_guidance_scale": 1,
         "output_type": "numpy",
     }
     return inputs





 def        snake_case__		(    self):
     snake_case_ :    Union[str, Any]     			=						"cpu"  # ensure determinism for the device-dependent torch.Generator
     snake_case_ :    Tuple     			=						self.get_dummy_components()
     snake_case_ :    Any     			=						StableDiffusionInstructPixaPixPipeline(**lowercase_)
     snake_case_ :    Union[str, Any]     			=						sd_pipe.to(lowercase_)
     sd_pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_ :    Optional[Any]     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    List[Any]     			=						sd_pipe(**lowercase_).images
     snake_case_ :    List[Any]     			=						image[0, -3:, -3:, -1]
     assert image.shape == (1, 32, 32, 3)
     snake_case_ :    Optional[int]     			=						np.array([0.7_526, 0.3_750, 0.4_547, 0.6_117, 0.5_866, 0.5_016, 0.4_327, 0.5_642, 0.4_815])

     assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3





 def        snake_case__		(    self):
     snake_case_ :    Optional[int]     			=						"cpu"  # ensure determinism for the device-dependent torch.Generator
     snake_case_ :    Union[str, Any]     			=						self.get_dummy_components()
     snake_case_ :    str     			=						StableDiffusionInstructPixaPixPipeline(**lowercase_)
     snake_case_ :    List[str]     			=						sd_pipe.to(lowercase_)
     sd_pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_ :    List[Any]     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    List[Any]     			=						"french fries"
     snake_case_ :    Union[str, Any]     			=						sd_pipe(**lowercase_ ,       negative_prompt=lowercase_)
     snake_case_ :    Optional[int]     			=						output.images
     snake_case_ :    Optional[Any]     			=						image[0, -3:, -3:, -1]

     assert image.shape == (1, 32, 32, 3)
     snake_case_ :    Tuple     			=						np.array([0.7_511, 0.3_642, 0.4_553, 0.6_236, 0.5_797, 0.5_013, 0.4_343, 0.5_611, 0.4_831])

     assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3





 def        snake_case__		(    self):
     snake_case_ :    List[str]     			=						"cpu"  # ensure determinism for the device-dependent torch.Generator
     snake_case_ :    Union[str, Any]     			=						self.get_dummy_components()
     snake_case_ :    Tuple     			=						StableDiffusionInstructPixaPixPipeline(**lowercase_)
     snake_case_ :    Tuple     			=						sd_pipe.to(lowercase_)
     sd_pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_ :    List[Any]     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    str     			=						[inputs["prompt"]] * 2

     snake_case_ :    Optional[Any]     			=						np.array(inputs["image"]).astype(np.floataa) / 255.0
     snake_case_ :    List[str]     			=						torch.from_numpy(lowercase_).unsqueeze(0).to(lowercase_)
     snake_case_ :    Any     			=						image / 2 + 0.5
     snake_case_ :    str     			=						image.permute(0 ,       3 ,       1 ,       2)
     snake_case_ :    Dict     			=						image.repeat(2 ,       1 ,       1 ,       1)

     snake_case_ :    List[str]     			=						sd_pipe(**lowercase_).images
     snake_case_ :    Dict     			=						image[-1, -3:, -3:, -1]

     assert image.shape == (2, 32, 32, 3)
     snake_case_ :    List[Any]     			=						np.array([0.5_812, 0.5_748, 0.5_222, 0.5_908, 0.5_695, 0.7_174, 0.6_804, 0.5_523, 0.5_579])

     assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3





 def        snake_case__		(    self):
     snake_case_ :    List[str]     			=						"cpu"  # ensure determinism for the device-dependent torch.Generator
     snake_case_ :    Tuple     			=						self.get_dummy_components()
     snake_case_ :    Tuple     			=						EulerAncestralDiscreteScheduler(
         beta_start=0.00_085 ,       beta_end=0.012 ,       beta_schedule="scaled_linear")
     snake_case_ :    Dict     			=						StableDiffusionInstructPixaPixPipeline(**lowercase_)
     snake_case_ :    Union[str, Any]     			=						sd_pipe.to(lowercase_)
     sd_pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_ :    Tuple     			=						self.get_dummy_inputs(lowercase_)
     snake_case_ :    Optional[int]     			=						sd_pipe(**lowercase_).images
     snake_case_ :    Union[str, Any]     			=						image[0, -3:, -3:, -1]

     snake_case_ :    Optional[Any]     			=						[round(lowercase_ ,       4) for x in image_slice.flatten().tolist()]
     print(",".join([str(lowercase_) for x in slice]))

     assert image.shape == (1, 32, 32, 3)
     snake_case_ :    Tuple     			=						np.array([0.7_417, 0.3_842, 0.4_732, 0.5_776, 0.5_891, 0.5_139, 0.4_052, 0.5_673, 0.4_986])

     assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3





 def        snake_case__		(    self):
     super().test_inference_batch_single_identical(expected_max_diff=3E-3)





 def        snake_case__		(    self):
     snake_case_ :    Optional[Any]     			=						self.get_dummy_components()
     snake_case_ :    Union[str, Any]     			=						StableDiffusionInstructPixaPixPipeline(**lowercase_)
     snake_case_ :    List[str]     			=						VaeImageProcessor(do_resize=lowercase_ ,       do_normalize=lowercase_)
     snake_case_ :    List[Any]     			=						pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_ :    str     			=						pipe(**self.get_dummy_inputs_by_type(lowercase_ ,       input_image_type="pt"))[0]

     snake_case_ :    str     			=						components["vae"]
     snake_case_ :    str     			=						self.get_dummy_inputs_by_type(lowercase_ ,       input_image_type="pt")

     for image_param in self.image_latents_params:
         if image_param in inputs.keys():
             snake_case_ :    str     			=						vae.encode(inputs[image_param]).latent_dist.mode()

     snake_case_ :    List[str]     			=						pipe(**lowercase_)[0]

     snake_case_ :    Any     			=						np.abs(out - out_latents_inputs).max()
     self.assertLess(lowercase_ ,       1E-4 ,       "passing latents as image input generate different result from passing image")






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





 def        snake_case__		(    self):
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()





 def        snake_case__		(    self ,       lowercase_=0):
     snake_case_ :    int     			=						torch.manual_seed(lowercase_)
     snake_case_ :    Union[str, Any]     			=						load_image(
         "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg")
     snake_case_ :    Tuple     			=						{
         "prompt": "turn him into a cyborg",
         "image": image,
         "generator": generator,
         "num_inference_steps": 3,
         "guidance_scale": 7.5,
         "image_guidance_scale": 1.0,
         "output_type": "numpy",
     }
     return inputs





 def        snake_case__		(    self):
     snake_case_ :    str     			=						StableDiffusionInstructPixaPixPipeline.from_pretrained(
         "timbrooks/instruct-pix2pix" ,       safety_checker=lowercase_)
     pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     pipe.enable_attention_slicing()

     snake_case_ :    int     			=						self.get_inputs()
     snake_case_ :    Dict     			=						pipe(**lowercase_).images
     snake_case_ :    str     			=						image[0, -3:, -3:, -1].flatten()

     assert image.shape == (1, 5_12, 5_12, 3)
     snake_case_ :    Dict     			=						np.array([0.5_902, 0.6_015, 0.6_027, 0.5_983, 0.6_092, 0.6_061, 0.5_765, 0.5_785, 0.5_555])

     assert np.abs(expected_slice - image_slice).max() < 1E-3





 def        snake_case__		(    self):
     snake_case_ :    int     			=						StableDiffusionInstructPixaPixPipeline.from_pretrained(
         "timbrooks/instruct-pix2pix" ,       safety_checker=lowercase_)
     snake_case_ :    Dict     			=						LMSDiscreteScheduler.from_config(pipe.scheduler.config)
     pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     pipe.enable_attention_slicing()

     snake_case_ :    int     			=						self.get_inputs()
     snake_case_ :    Dict     			=						pipe(**lowercase_).images
     snake_case_ :    str     			=						image[0, -3:, -3:, -1].flatten()

     assert image.shape == (1, 5_12, 5_12, 3)
     snake_case_ :    str     			=						np.array([0.6_578, 0.6_817, 0.6_972, 0.6_761, 0.6_856, 0.6_916, 0.6_428, 0.6_516, 0.6_301])

     assert np.abs(expected_slice - image_slice).max() < 1E-3





 def        snake_case__		(    self):
     snake_case_ :    Optional[int]     			=						StableDiffusionInstructPixaPixPipeline.from_pretrained(
         "timbrooks/instruct-pix2pix" ,       safety_checker=lowercase_)
     snake_case_ :    List[Any]     			=						DDIMScheduler.from_config(pipe.scheduler.config)
     pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     pipe.enable_attention_slicing()

     snake_case_ :    str     			=						self.get_inputs()
     snake_case_ :    Optional[Any]     			=						pipe(**lowercase_).images
     snake_case_ :    Dict     			=						image[0, -3:, -3:, -1].flatten()

     assert image.shape == (1, 5_12, 5_12, 3)
     snake_case_ :    List[str]     			=						np.array([0.3_828, 0.3_834, 0.3_818, 0.3_792, 0.3_865, 0.3_752, 0.3_792, 0.3_847, 0.3_753])

     assert np.abs(expected_slice - image_slice).max() < 1E-3





 def        snake_case__		(    self):
     snake_case_ :    Union[str, Any]     			=						0

     def callback_fn(lowercase_ ,       lowercase_ ,       lowercase_) -> None:
         snake_case_ :    Optional[Any]     			=						True
         nonlocal number_of_steps
         number_of_steps += 1
         if step == 1:
             snake_case_ :    Optional[Any]     			=						latents.detach().cpu().numpy()
             assert latents.shape == (1, 4, 64, 64)
             snake_case_ :    Optional[int]     			=						latents[0, -3:, -3:, -1]
             snake_case_ :    Optional[int]     			=						np.array([-0.2_463, -0.4_644, -0.9_756, 1.5_176, 1.4_414, 0.7_866, 0.9_897, 0.8_521, 0.7_983])

             assert np.abs(latents_slice.flatten() - expected_slice).max() < 5E-2
         elif step == 2:
             snake_case_ :    List[str]     			=						latents.detach().cpu().numpy()
             assert latents.shape == (1, 4, 64, 64)
             snake_case_ :    Union[str, Any]     			=						latents[0, -3:, -3:, -1]
             snake_case_ :    Any     			=						np.array([-0.2_644, -0.4_626, -0.9_653, 1.5_176, 1.4_551, 0.7_686, 0.9_805, 0.8_452, 0.8_115])

             assert np.abs(latents_slice.flatten() - expected_slice).max() < 5E-2

     snake_case_ :    Optional[int]     			=						False

     snake_case_ :    Union[str, Any]     			=						StableDiffusionInstructPixaPixPipeline.from_pretrained(
         "timbrooks/instruct-pix2pix" ,       safety_checker=lowercase_ ,       torch_dtype=torch.floataa)
     snake_case_ :    Dict     			=						pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     pipe.enable_attention_slicing()

     snake_case_ :    Any     			=						self.get_inputs()
     pipe(**lowercase_ ,       callback=lowercase_ ,       callback_steps=1)
     assert callback_fn.has_been_called
     assert number_of_steps == 3





 def        snake_case__		(    self):
     torch.cuda.empty_cache()
     torch.cuda.reset_max_memory_allocated()
     torch.cuda.reset_peak_memory_stats()

     snake_case_ :    int     			=						StableDiffusionInstructPixaPixPipeline.from_pretrained(
         "timbrooks/instruct-pix2pix" ,       safety_checker=lowercase_ ,       torch_dtype=torch.floataa)
     snake_case_ :    str     			=						pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     pipe.enable_attention_slicing(1)
     pipe.enable_sequential_cpu_offload()

     snake_case_ :    Dict     			=						self.get_inputs()
     snake_case_ :    int     			=						pipe(**lowercase_)

     snake_case_ :    Dict     			=						torch.cuda.max_memory_allocated()
     # make sure that less than 2.2 GB is allocated
     assert mem_bytes < 2.2 * 10**9





 def        snake_case__		(    self):
     snake_case_ :    List[Any]     			=						self.get_inputs()
     # resize to resolution that is divisible by 8 but not 16 or 32
     snake_case_ :    Union[str, Any]     			=						inputs["image"].resize((5_04, 5_04))

     snake_case_ :    str     			=						"timbrooks/instruct-pix2pix"
     snake_case_ :    Any     			=						StableDiffusionInstructPixaPixPipeline.from_pretrained(
         lowercase_ ,       safety_checker=lowercase_ ,       )
     pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     pipe.enable_attention_slicing()

     snake_case_ :    Any     			=						pipe(**lowercase_)
     snake_case_ :    int     			=						output.images[0]

     snake_case_ :    Tuple     			=						image[2_55:2_58, 3_83:3_86, -1]

     assert image.shape == (5_04, 5_04, 3)
     snake_case_ :    Optional[int]     			=						np.array([0.2_726, 0.2_529, 0.2_664, 0.2_655, 0.2_641, 0.2_642, 0.2_591, 0.2_649, 0.2_590])

     assert np.abs(image_slice.flatten() - expected_slice).max() < 5E-3




