

import logging
import os
from dataclasses import dataclass
from typing import List, Optional, Union

import tqdm
from filelock import FileLock

from transformers import (
    BartTokenizer,
    BartTokenizerFast,
    DataProcessor,
    PreTrainedTokenizer,
    RobertaTokenizer,
    RobertaTokenizerFast,
    XLMRobertaTokenizer,
    is_tf_available,
    is_torch_available,
)


UpperCAmelCase__ :     Dict				    =    logging.getLogger(__name__)







@dataclass(frozen=UpperCAmelCase       )
class        a__       :







  """simple docstring"""




  UpperCAmelCase__      : str
  UpperCAmelCase__      : str
  UpperCAmelCase__      : Optional[str]			       =None
  UpperCAmelCase__      : Optional[str]			       =None
  UpperCAmelCase__      : Optional[str]			       =None







@dataclass(frozen=UpperCAmelCase       )
class        a__       :







  """simple docstring"""




  UpperCAmelCase__      : List[int]
  UpperCAmelCase__      : Optional[List[int]]			       =None
  UpperCAmelCase__      : Optional[List[int]]			       =None
  UpperCAmelCase__      : Optional[Union[int, float]]			       =None
  UpperCAmelCase__      : Optional[int]			       =None


if is_torch_available():
  import torch
  from torch.utils.data import Dataset







  class        a__       (   UpperCAmelCase       ):







    """simple docstring"""




    UpperCAmelCase__      : List[InputFeatures]


    def __init__(			self    : int				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : PreTrainedTokenizer				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : Optional[int] = None				,				UpperCAmelCase__    : Dict=False				,				UpperCAmelCase__    : bool = False				,				)					->int:
          """simple docstring"""







          SCREAMING_SNAKE_CASE :      List[Any]       =   hans_processors[task]()

          SCREAMING_SNAKE_CASE :      Optional[Any]       =   os.path.join(
              UpperCAmelCase__				,				"""cached_{}_{}_{}_{}""".format(
                  """dev""" if evaluate else """train"""				,				tokenizer.__class__.__name__				,				str(UpperCAmelCase__		)				,				UpperCAmelCase__				,				)				,				)
          SCREAMING_SNAKE_CASE :      Tuple       =   processor.get_labels()
          if tokenizer.__class__ in (
              RobertaTokenizer,
              RobertaTokenizerFast,
              XLMRobertaTokenizer,
              BartTokenizer,
              BartTokenizerFast,
          ):
                # HACK(label indices are swapped in RoBERTa pretrained model)
                SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      List[str]       =   label_list[2], label_list[1]
          SCREAMING_SNAKE_CASE :      List[str]       =   label_list

          # Make sure only the first process in distributed training processes the dataset,
          # and the others will use the cache.
          SCREAMING_SNAKE_CASE :      Optional[int]       =   cached_features_file + """.lock"""
          with FileLock(UpperCAmelCase__		):
                if os.path.exists(UpperCAmelCase__		) and not overwrite_cache:
                      logger.info(f"Loading features from cached file {cached_features_file}"		)
                      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.load(UpperCAmelCase__		)
                else:
                      logger.info(f"Creating features from dataset file at {data_dir}"		)

                      SCREAMING_SNAKE_CASE :      Optional[Any]       =   (
                          processor.get_dev_examples(UpperCAmelCase__		) if evaluate else processor.get_train_examples(UpperCAmelCase__		)
                      )

                      logger.info("""Training examples: %s"""				,				len(UpperCAmelCase__		)		)
                      SCREAMING_SNAKE_CASE :      str       =   hans_convert_examples_to_features(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)
                      logger.info("""Saving features into cached file %s"""				,				UpperCAmelCase__		)
                      torch.save(self.features				,				UpperCAmelCase__		)


    def __len__(			self    : str		)					->Dict:
          """simple docstring"""







          return len(self.features		)


    def __getitem__(			self    : List[str]				,				UpperCAmelCase__    : Union[str, Any]		)					->InputFeatures:
          """simple docstring"""







          return self.features[i]




    def  _lowercase      (			self    : List[Any]		)					->Optional[int]:
          """simple docstring"""







          return self.label_list


if is_tf_available():
  import tensorflow as tf







  class        a__       :







    """simple docstring"""




    UpperCAmelCase__      : List[InputFeatures]


    def __init__(			self    : List[Any]				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : PreTrainedTokenizer				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : Optional[int] = 1_2_8				,				UpperCAmelCase__    : str=False				,				UpperCAmelCase__    : bool = False				,				)					->str:
          """simple docstring"""







          SCREAMING_SNAKE_CASE :      str       =   hans_processors[task]()
          SCREAMING_SNAKE_CASE :      Tuple       =   processor.get_labels()
          if tokenizer.__class__ in (
              RobertaTokenizer,
              RobertaTokenizerFast,
              XLMRobertaTokenizer,
              BartTokenizer,
              BartTokenizerFast,
          ):
                # HACK(label indices are swapped in RoBERTa pretrained model)
                SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      str       =   label_list[2], label_list[1]
          SCREAMING_SNAKE_CASE :      Dict       =   label_list

          SCREAMING_SNAKE_CASE :      List[str]       =   processor.get_dev_examples(UpperCAmelCase__		) if evaluate else processor.get_train_examples(UpperCAmelCase__		)
          SCREAMING_SNAKE_CASE :      Any       =   hans_convert_examples_to_features(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)

          def gen():
                for ex_index, ex in tqdm.tqdm(enumerate(self.features		)				,				desc="""convert examples to features"""		):
                      if ex_index % 1_0_0_0_0 == 0:
                            logger.info("""Writing example %d of %d""" % (ex_index, len(UpperCAmelCase__		))		)

                      yield (
                          {
                              "example_id": 0,
                              "input_ids": ex.input_ids,
                              "attention_mask": ex.attention_mask,
                              "token_type_ids": ex.token_type_ids,
                          },
                          ex.label,
                      )

          SCREAMING_SNAKE_CASE :      Union[str, Any]       =   tf.data.Dataset.from_generator(
              UpperCAmelCase__				,				(
                  {
                      """example_id""": tf.intaa,
                      """input_ids""": tf.intaa,
                      """attention_mask""": tf.intaa,
                      """token_type_ids""": tf.intaa,
                  },
                  tf.intaa,
              )				,				(
                  {
                      """example_id""": tf.TensorShape([]		),
                      """input_ids""": tf.TensorShape([None, None]		),
                      """attention_mask""": tf.TensorShape([None, None]		),
                      """token_type_ids""": tf.TensorShape([None, None]		),
                  },
                  tf.TensorShape([]		),
              )				,				)


    def  _lowercase      (			self    : List[Any]		)					->Any:
          """simple docstring"""







          return self.dataset


    def __len__(			self    : List[Any]		)					->List[str]:
          """simple docstring"""







          return len(self.features		)


    def __getitem__(			self    : Optional[int]				,				UpperCAmelCase__    : Any		)					->InputFeatures:
          """simple docstring"""







          return self.features[i]




    def  _lowercase      (			self    : List[str]		)					->Optional[int]:
          """simple docstring"""







          return self.label_list







class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  def  _lowercase      (			self    : List[Any]				,				UpperCAmelCase__    : Optional[Any]		)					->Dict:
        """simple docstring"""







        return self._create_examples(self._read_tsv(os.path.join(UpperCAmelCase__				,				"""heuristics_train_set.txt"""		)		)				,				"""train"""		)


  def  _lowercase      (			self    : Union[str, Any]				,				UpperCAmelCase__    : List[str]		)					->Dict:
        """simple docstring"""







        return self._create_examples(self._read_tsv(os.path.join(UpperCAmelCase__				,				"""heuristics_evaluation_set.txt"""		)		)				,				"""dev"""		)


  def  _lowercase      (			self    : Tuple		)					->Tuple:
        """simple docstring"""







        return ["contradiction", "entailment", "neutral"]




  def  _lowercase      (			self    : str				,				UpperCAmelCase__    : List[str]				,				UpperCAmelCase__    : List[Any]		)					->Union[str, Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   []
        for i, line in enumerate(UpperCAmelCase__		):
              if i == 0:
                    continue
              SCREAMING_SNAKE_CASE :      List[Any]       =   """%s-%s""" % (set_type, line[0])
              SCREAMING_SNAKE_CASE :      Dict       =   line[5]
              SCREAMING_SNAKE_CASE :      Dict       =   line[6]
              SCREAMING_SNAKE_CASE :      str       =   line[7][2:] if line[7].startswith("""ex"""		) else line[7]
              SCREAMING_SNAKE_CASE :      Dict       =   line[0]
              examples.append(InputExample(guid=UpperCAmelCase__				,				text_a=UpperCAmelCase__				,				text_b=UpperCAmelCase__				,				label=UpperCAmelCase__				,				pairID=UpperCAmelCase__		)		)
        return examples






def 					__lowercase      (     _A						,    _A						,    _A						,    _A						,    )     ->						int:
      SCREAMING_SNAKE_CASE :      Tuple       =   {label: i for i, label in enumerate(_A			)}

      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   []
      for ex_index, example in tqdm.tqdm(enumerate(_A			)						,    desc="""convert examples to features"""			):
            if ex_index % 10000 == 0:
                  logger.info("""Writing example %d""" % (ex_index)			)

            SCREAMING_SNAKE_CASE :      Any       =   tokenizer(
                example.text_a						,    example.text_b						,    add_special_tokens=_A						,    max_length=_A						,    padding="""max_length"""						,    truncation=_A						,    return_overflowing_tokens=_A						,    )

            SCREAMING_SNAKE_CASE :      Any       =   label_map[example.label] if example.label in label_map else 0

            SCREAMING_SNAKE_CASE :      Union[str, Any]       =   int(example.pairID			)

            features.append(InputFeatures(**_A						,    label=_A						,    pairID=_A			)			)

      for i, example in enumerate(examples[:5]			):
            logger.info("""*** Example ***"""			)
            logger.info(F"guid: {example}"			)
            logger.info(F"features: {features[i]}"			)

      return features


UpperCAmelCase__ :     Dict				    =    {
    """hans""": 3,
}







UpperCAmelCase__ :     Any				    =    {
    """hans""": HansProcessor,
}







import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizer

from diffusers import (
    AltDiffusionImgaImgPipeline,
    AutoencoderKL,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()







class        a__       (   unittest.TestCase       ):







  """simple docstring"""




  def  _lowercase      (			self    : Union[str, Any]		)					->List[Any]:
        """simple docstring"""







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


  @property
  def  _lowercase      (			self    : str		)					->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[int]       =   1
        SCREAMING_SNAKE_CASE :      Dict       =   3
        SCREAMING_SNAKE_CASE :      List[Any]       =   (3_2, 3_2)

        SCREAMING_SNAKE_CASE :      Any       =   floats_tensor((batch_size, num_channels) + sizes				,				rng=random.Random(0		)		).to(UpperCAmelCase__		)
        return image


  @property
  def  _lowercase      (			self    : Tuple		)					->List[Any]:
        """simple docstring"""







        torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE :      str       =   UNetaDConditionModel(
            block_out_channels=(3_2, 6_4)				,				layers_per_block=2				,				sample_size=3_2				,				in_channels=4				,				out_channels=4				,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")				,				up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")				,				cross_attention_dim=3_2				,				)
        return model


  @property
  def  _lowercase      (			self    : str		)					->int:
        """simple docstring"""







        torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE :      Tuple       =   AutoencoderKL(
            block_out_channels=[3_2, 6_4]				,				in_channels=3				,				out_channels=3				,				down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]				,				up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]				,				latent_channels=4				,				)
        return model


  @property
  def  _lowercase      (			self    : Tuple		)					->Union[str, Any]:
        """simple docstring"""







        torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE :      str       =   RobertaSeriesConfig(
            hidden_size=3_2				,				project_dim=3_2				,				intermediate_size=3_7				,				layer_norm_eps=1e-05				,				num_attention_heads=4				,				num_hidden_layers=5				,				pad_token_id=1				,				vocab_size=5_0_0_6				,				)
        return RobertaSeriesModelWithTransformation(UpperCAmelCase__		)


  @property
  def  _lowercase      (			self    : Union[str, Any]		)					->List[str]:
        """simple docstring"""







        def extract(*UpperCAmelCase__    : List[Any]				,				**UpperCAmelCase__    : Tuple		):







              class        a__       :







                """simple docstring"""




                def __init__(			self    : Tuple		)					->Dict:
                      """simple docstring"""







                      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.ones([0]		)




                def  _lowercase      (			self    : Optional[int]				,				UpperCAmelCase__    : Dict		)					->Optional[Any]:
                      """simple docstring"""







                      self.pixel_values.to(UpperCAmelCase__		)
                      return self


              return Out()

        return extract


  def  _lowercase      (			self    : List[str]		)					->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[int]       =   """cpu"""  # ensure determinism for the device-dependent torch.Generator
        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.dummy_cond_unet
        SCREAMING_SNAKE_CASE :      str       =   PNDMScheduler(skip_prk_steps=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      List[Any]       =   self.dummy_vae
        SCREAMING_SNAKE_CASE :      str       =   self.dummy_text_encoder
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   XLMRobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-xlm-roberta"""		)
        SCREAMING_SNAKE_CASE :      List[Any]       =   7_7

        SCREAMING_SNAKE_CASE :      str       =   self.dummy_image.to(UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   init_image / 2 + 0.5

        # make sure here that pndm scheduler skips prk
        SCREAMING_SNAKE_CASE :      List[str]       =   AltDiffusionImgaImgPipeline(
            unet=UpperCAmelCase__				,				scheduler=UpperCAmelCase__				,				vae=UpperCAmelCase__				,				text_encoder=UpperCAmelCase__				,				tokenizer=UpperCAmelCase__				,				safety_checker=UpperCAmelCase__				,				feature_extractor=self.dummy_extractor				,				)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor				,				do_normalize=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Tuple       =   alt_pipe.to(UpperCAmelCase__		)
        alt_pipe.set_progress_bar_config(disable=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      str       =   """A painting of a squirrel eating a burger"""
        SCREAMING_SNAKE_CASE :      Tuple       =   torch.Generator(device=UpperCAmelCase__		).manual_seed(0		)
        SCREAMING_SNAKE_CASE :      List[Any]       =   alt_pipe(
            [prompt]				,				generator=UpperCAmelCase__				,				guidance_scale=6.0				,				num_inference_steps=2				,				output_type="""np"""				,				image=UpperCAmelCase__				,				)

        SCREAMING_SNAKE_CASE :      List[str]       =   output.images

        SCREAMING_SNAKE_CASE :      List[Any]       =   torch.Generator(device=UpperCAmelCase__		).manual_seed(0		)
        SCREAMING_SNAKE_CASE :      Any       =   alt_pipe(
            [prompt]				,				generator=UpperCAmelCase__				,				guidance_scale=6.0				,				num_inference_steps=2				,				output_type="""np"""				,				image=UpperCAmelCase__				,				return_dict=UpperCAmelCase__				,				)[0]

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   image[0, -3:, -3:, -1]
        SCREAMING_SNAKE_CASE :      int       =   image_from_tuple[0, -3:, -3:, -1]

        assert image.shape == (1, 3_2, 3_2, 3)
        SCREAMING_SNAKE_CASE :      str       =   np.array([0.44_27, 0.37_31, 0.42_49, 0.49_41, 0.45_46, 0.41_48, 0.41_93, 0.46_66, 0.44_99]		)

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


  @unittest.skipIf(torch_device != """cuda"""				,				"""This test requires a GPU"""		)
  def  _lowercase      (			self    : Optional[int]		)					->List[str]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.dummy_cond_unet
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   PNDMScheduler(skip_prk_steps=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      List[str]       =   self.dummy_vae
        SCREAMING_SNAKE_CASE :      int       =   self.dummy_text_encoder
        SCREAMING_SNAKE_CASE :      List[Any]       =   XLMRobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-xlm-roberta"""		)
        SCREAMING_SNAKE_CASE :      Tuple       =   7_7

        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.dummy_image.to(UpperCAmelCase__		)

        # put models in fp16
        SCREAMING_SNAKE_CASE :      List[Any]       =   unet.half()
        SCREAMING_SNAKE_CASE :      Tuple       =   vae.half()
        SCREAMING_SNAKE_CASE :      str       =   bert.half()

        # make sure here that pndm scheduler skips prk
        SCREAMING_SNAKE_CASE :      str       =   AltDiffusionImgaImgPipeline(
            unet=UpperCAmelCase__				,				scheduler=UpperCAmelCase__				,				vae=UpperCAmelCase__				,				text_encoder=UpperCAmelCase__				,				tokenizer=UpperCAmelCase__				,				safety_checker=UpperCAmelCase__				,				feature_extractor=self.dummy_extractor				,				)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor				,				do_normalize=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      str       =   alt_pipe.to(UpperCAmelCase__		)
        alt_pipe.set_progress_bar_config(disable=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      int       =   """A painting of a squirrel eating a burger"""
        SCREAMING_SNAKE_CASE :      List[str]       =   torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE :      Any       =   alt_pipe(
            [prompt]				,				generator=UpperCAmelCase__				,				num_inference_steps=2				,				output_type="""np"""				,				image=UpperCAmelCase__				,				).images

        assert image.shape == (1, 3_2, 3_2, 3)




  @unittest.skipIf(torch_device != """cuda"""				,				"""This test requires a GPU"""		)
  def  _lowercase      (			self    : Dict		)					->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Dict       =   load_image(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/img2img/sketch-mountains-input.jpg"""		)
        # resize to resolution that is divisible by 8 but not 16 or 32
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   init_image.resize((7_6_0, 5_0_4)		)

        SCREAMING_SNAKE_CASE :      int       =   """BAAI/AltDiffusion"""
        SCREAMING_SNAKE_CASE :      List[str]       =   AltDiffusionImgaImgPipeline.from_pretrained(
            UpperCAmelCase__				,				safety_checker=UpperCAmelCase__				,				)
        pipe.to(UpperCAmelCase__		)
        pipe.set_progress_bar_config(disable=UpperCAmelCase__		)
        pipe.enable_attention_slicing()

        SCREAMING_SNAKE_CASE :      int       =   """A fantasy landscape, trending on artstation"""

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   pipe(
            prompt=UpperCAmelCase__				,				image=UpperCAmelCase__				,				strength=0.75				,				guidance_scale=7.5				,				generator=UpperCAmelCase__				,				output_type="""np"""				,				)
        SCREAMING_SNAKE_CASE :      Tuple       =   output.images[0]

        SCREAMING_SNAKE_CASE :      Tuple       =   image[2_5_5:2_5_8, 3_8_3:3_8_6, -1]

        assert image.shape == (5_0_4, 7_6_0, 3)
        SCREAMING_SNAKE_CASE :      List[str]       =   np.array([0.93_58, 0.93_97, 0.95_99, 0.99_01, 1.00_00, 1.00_00, 0.98_82, 1.00_00, 1.00_00]		)

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







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







  """simple docstring"""




  def  _lowercase      (			self    : Any		)					->List[str]:
        """simple docstring"""







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




  def  _lowercase      (			self    : Optional[Any]		)					->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   load_image(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/img2img/sketch-mountains-input.jpg"""		)
        SCREAMING_SNAKE_CASE :      str       =   init_image.resize((7_6_8, 5_1_2)		)
        SCREAMING_SNAKE_CASE :      int       =   load_numpy(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/img2img/fantasy_landscape_alt.npy"""		)

        SCREAMING_SNAKE_CASE :      Tuple       =   """BAAI/AltDiffusion"""
        SCREAMING_SNAKE_CASE :      str       =   AltDiffusionImgaImgPipeline.from_pretrained(
            UpperCAmelCase__				,				safety_checker=UpperCAmelCase__				,				)
        pipe.to(UpperCAmelCase__		)
        pipe.set_progress_bar_config(disable=UpperCAmelCase__		)
        pipe.enable_attention_slicing()

        SCREAMING_SNAKE_CASE :      Dict       =   """A fantasy landscape, trending on artstation"""

        SCREAMING_SNAKE_CASE :      Optional[int]       =   torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   pipe(
            prompt=UpperCAmelCase__				,				image=UpperCAmelCase__				,				strength=0.75				,				guidance_scale=7.5				,				generator=UpperCAmelCase__				,				output_type="""np"""				,				)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   output.images[0]

        assert image.shape == (5_1_2, 7_6_8, 3)
        # img2img is flaky across GPUs even in fp32, so using MAE here
        assert np.abs(expected_image - image		).max() < 1e-2




