




import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, 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_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class  _lowerCAmelCase     (  _lowercase    ,			_lowercase    ,			unittest.TestCase	):
      A__		     =		StableDiffusionXLImgaImgPipeline
      A__		     =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
      A__		     =		PipelineTesterMixin.required_optional_params - {'latents'}
      A__		     =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
      A__		     =		IMAGE_TO_IMAGE_IMAGE_PARAMS
      A__		     =		IMAGE_TO_IMAGE_IMAGE_PARAMS

      def 	__magic_name__(   self		):
        torch.manual_seed(0		)
        lowerCAmelCase__				:  List[str]		       =     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''')      ,				attention_head_dim=(2, 4)      ,				use_linear_projection=__UpperCAmelCase      ,				addition_embed_type='''text_time'''      ,				addition_time_embed_dim=8      ,				transformer_layers_per_block=(1, 2)      ,				projection_class_embeddings_input_dim=80      ,				cross_attention_dim=64      ,				)
        lowerCAmelCase__				:  str		       =     EulerDiscreteScheduler(
            beta_start=0.00085      ,				beta_end=0.012      ,				steps_offset=1      ,				beta_schedule='''scaled_linear'''      ,				timestep_spacing='''leading'''      ,				)
        torch.manual_seed(0		)
        lowerCAmelCase__				:  str		       =     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      ,				sample_size=128      ,				)
        torch.manual_seed(0		)
        lowerCAmelCase__				:  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=1000      ,				hidden_act='''gelu'''      ,				projection_dim=32      ,				)
        lowerCAmelCase__				:  Dict		       =     CLIPTextModel(__UpperCAmelCase		)
        lowerCAmelCase__				:  Dict		       =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      ,				local_files_only=__UpperCAmelCase		)

        lowerCAmelCase__				:  Tuple		       =     CLIPTextModelWithProjection(__UpperCAmelCase		)
        lowerCAmelCase__				:  Optional[Any]		       =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      ,				local_files_only=__UpperCAmelCase		)

        lowerCAmelCase__				:  Dict		       =     {
            '''unet''': unet,
            '''scheduler''': scheduler,
            '''vae''': vae,
            '''text_encoder''': text_encoder,
            '''tokenizer''': tokenizer,
            '''text_encoder_2''': text_encoder_a,
            '''tokenizer_2''': tokenizer_a,
            # "safety_checker": None,
            # "feature_extractor": None,
        }
        return components

      def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase=0		):
        lowerCAmelCase__				:  Any		       =     floats_tensor((1, 3, 32, 32)      ,				rng=random.Random(__UpperCAmelCase		)		).to(__UpperCAmelCase		)
        lowerCAmelCase__				:  Optional[Any]		       =     image / 2 + 0.5
        if str(__UpperCAmelCase		).startswith('''mps'''		):
          lowerCAmelCase__				:  List[Any]		       =     torch.manual_seed(__UpperCAmelCase		)
        else:
          lowerCAmelCase__				:  Any		       =     torch.Generator(device=__UpperCAmelCase		).manual_seed(__UpperCAmelCase		)
        lowerCAmelCase__				:  List[Any]		       =     {
            '''prompt''': '''A painting of a squirrel eating a burger''',
            '''image''': image,
            '''generator''': generator,
            '''num_inference_steps''': 2,
            '''guidance_scale''': 5.0,
            '''output_type''': '''numpy''',
            '''strength''': 0.75,
        }
        return inputs

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Union[str, Any]		       =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__				:  Tuple		       =     self.get_dummy_components()
        lowerCAmelCase__				:  Any		       =     StableDiffusionXLImgaImgPipeline(**__UpperCAmelCase		)
        lowerCAmelCase__				:  Tuple		       =     sd_pipe.to(__UpperCAmelCase		)
        sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

        lowerCAmelCase__				:  Dict		       =     self.get_dummy_inputs(__UpperCAmelCase		)
        lowerCAmelCase__				:  Optional[Any]		       =     sd_pipe(**__UpperCAmelCase		).images
        lowerCAmelCase__				:  Optional[int]		       =     image[0, -3:, -3:, -1]

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

        lowerCAmelCase__				:  List[str]		       =     np.array([0.4656, 0.4840, 0.4439, 0.6698, 0.5574, 0.4524, 0.5799, 0.5943, 0.5165]		)

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

      def 	__magic_name__(   self		):
        super().test_attention_slicing_forward_pass(expected_max_diff=3e-3		)

      def 	__magic_name__(   self		):
        super().test_inference_batch_single_identical(expected_max_diff=3e-3		)

      def 	__magic_name__(   self		):
        pass







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  str		       =     self.get_dummy_components()
        lowerCAmelCase__				:  Dict		       =     StableDiffusionXLImgaImgPipeline(**__UpperCAmelCase		)
        lowerCAmelCase__				:  Optional[Any]		       =     sd_pipe.to(__UpperCAmelCase		)
        lowerCAmelCase__				:  int		       =     sd_pipe.to(__UpperCAmelCase		)
        sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

        # forward without prompt embeds
        lowerCAmelCase__				:  Optional[Any]		       =     self.get_dummy_inputs(__UpperCAmelCase		)
        lowerCAmelCase__				:  List[str]		       =     3 * ['''this is a negative prompt''']
        lowerCAmelCase__				:  int		       =     negative_prompt
        lowerCAmelCase__				:  List[Any]		       =     3 * [inputs['''prompt''']]

        lowerCAmelCase__				:  Optional[Any]		       =     sd_pipe(**__UpperCAmelCase		)
        lowerCAmelCase__				:  Dict		       =     output.images[0, -3:, -3:, -1]

        # forward with prompt embeds
        lowerCAmelCase__				:  str		       =     self.get_dummy_inputs(__UpperCAmelCase		)
        lowerCAmelCase__				:  List[str]		       =     3 * ['''this is a negative prompt''']
        lowerCAmelCase__				:  Dict		       =     3 * [inputs.pop('''prompt'''		)]

        (
            (
            lowerCAmelCase__
        )       ,			(
            lowerCAmelCase__
        )       ,			(
            lowerCAmelCase__
        )       ,			(
            lowerCAmelCase__
        )       ,			
        )				:  Optional[int]		       =     sd_pipe.encode_prompt(__UpperCAmelCase      ,				negative_prompt=__UpperCAmelCase		)

        lowerCAmelCase__				:  Union[str, Any]		       =     sd_pipe(
            **__UpperCAmelCase      ,				prompt_embeds=__UpperCAmelCase      ,				negative_prompt_embeds=__UpperCAmelCase      ,				pooled_prompt_embeds=__UpperCAmelCase      ,				negative_pooled_prompt_embeds=__UpperCAmelCase      ,				)
        lowerCAmelCase__				:  int		       =     output.images[0, -3:, -3:, -1]

        # make sure that it's equal
        assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()		).max() < 1e-4






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

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

      def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase="cpu"      ,				__UpperCAmelCase=torch.floataa      ,				__UpperCAmelCase=0		):
        lowerCAmelCase__				:  Any		       =     torch.Generator(device=__UpperCAmelCase		).manual_seed(__UpperCAmelCase		)
        lowerCAmelCase__				:  Union[str, Any]		       =     np.random.RandomState(__UpperCAmelCase		).standard_normal((1, 4, 64, 64)		)
        lowerCAmelCase__				:  List[str]		       =     torch.from_numpy(__UpperCAmelCase		).to(device=__UpperCAmelCase      ,				dtype=__UpperCAmelCase		)
        lowerCAmelCase__				:  Tuple		       =     {
            '''prompt''': '''a photograph of an astronaut riding a horse''',
            '''latents''': latents,
            '''generator''': generator,
            '''num_inference_steps''': 3,
            '''guidance_scale''': 7.5,
            '''output_type''': '''numpy''',
        }
        return inputs







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Any		       =     DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base'''		)
        pipe.to(__UpperCAmelCase		)
        pipe.set_progress_bar_config(disable=__UpperCAmelCase		)

        lowerCAmelCase__				:  Optional[Any]		       =     self.get_inputs(__UpperCAmelCase		)
        lowerCAmelCase__				:  int		       =     pipe(**__UpperCAmelCase		).images
        lowerCAmelCase__				:  Union[str, Any]		       =     image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 512, 512, 3)
        lowerCAmelCase__				:  str		       =     np.array([0.49493, 0.47896, 0.40798, 0.54214, 0.53212, 0.48202, 0.47656, 0.46329, 0.48506]		)
        assert np.abs(image_slice - expected_slice		).max() < 7e-3







from tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch






@require_torch
@require_tf
class  _lowerCAmelCase     (  _lowercase	):

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  Union[str, Any]		       =     SMALL_MODEL_IDENTIFIER
        lowerCAmelCase__				:  List[str]		       =     '''pt'''
        lowerCAmelCase__				:  int		       =     '''tf'''

      def 	__magic_name__(   self      ,				__UpperCAmelCase		):
        lowerCAmelCase__				:  Optional[Any]		       =     AutoModel.from_pretrained(self.test_model		)
        model_pt.save_pretrained(__UpperCAmelCase		)

      def 	__magic_name__(   self      ,				__UpperCAmelCase		):
        lowerCAmelCase__				:  str		       =     TFAutoModel.from_pretrained(self.test_model      ,				from_pt=__UpperCAmelCase		)
        model_tf.save_pretrained(__UpperCAmelCase		)

      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  int		       =     '''mock_framework'''

        # Framework provided - return whatever the user provides
        lowerCAmelCase__				:  Tuple		       =     FeaturesManager.determine_framework(self.test_model      ,				__UpperCAmelCase		)
        self.assertEqual(__UpperCAmelCase      ,				__UpperCAmelCase		)

        # Local checkpoint and framework provided - return provided framework
        # PyTorch checkpoint
        with TemporaryDirectory() as local_pt_ckpt:
          self._setup_pt_ckpt(__UpperCAmelCase		)
          lowerCAmelCase__				:  int		       =     FeaturesManager.determine_framework(__UpperCAmelCase      ,				__UpperCAmelCase		)
          self.assertEqual(__UpperCAmelCase      ,				__UpperCAmelCase		)

        # TensorFlow checkpoint
        with TemporaryDirectory() as local_tf_ckpt:
          self._setup_tf_ckpt(__UpperCAmelCase		)
          lowerCAmelCase__				:  List[str]		       =     FeaturesManager.determine_framework(__UpperCAmelCase      ,				__UpperCAmelCase		)
          self.assertEqual(__UpperCAmelCase      ,				__UpperCAmelCase		)

      def 	__magic_name__(   self		):
        # PyTorch checkpoint
        with TemporaryDirectory() as local_pt_ckpt:
          self._setup_pt_ckpt(__UpperCAmelCase		)
          lowerCAmelCase__				:  Tuple		       =     FeaturesManager.determine_framework(__UpperCAmelCase		)
          self.assertEqual(__UpperCAmelCase      ,				self.framework_pt		)

        # TensorFlow checkpoint
        with TemporaryDirectory() as local_tf_ckpt:
          self._setup_tf_ckpt(__UpperCAmelCase		)
          lowerCAmelCase__				:  int		       =     FeaturesManager.determine_framework(__UpperCAmelCase		)
          self.assertEqual(__UpperCAmelCase      ,				self.framework_tf		)

        # Invalid local checkpoint
        with TemporaryDirectory() as local_invalid_ckpt:
          with self.assertRaises(__UpperCAmelCase		):
            lowerCAmelCase__				:  Tuple		       =     FeaturesManager.determine_framework(__UpperCAmelCase		)







      def 	__magic_name__(   self		):
        lowerCAmelCase__				:  List[Any]		       =     MagicMock(return_value=__UpperCAmelCase		)
        with patch('''transformers.onnx.features.is_tf_available'''      ,				__UpperCAmelCase		):
          lowerCAmelCase__				:  int		       =     FeaturesManager.determine_framework(self.test_model		)
          self.assertEqual(__UpperCAmelCase      ,				self.framework_pt		)

        # PyTorch not in environment -> use TensorFlow
        lowerCAmelCase__				:  Union[str, Any]		       =     MagicMock(return_value=__UpperCAmelCase		)
        with patch('''transformers.onnx.features.is_torch_available'''      ,				__UpperCAmelCase		):
          lowerCAmelCase__				:  Optional[int]		       =     FeaturesManager.determine_framework(self.test_model		)
          self.assertEqual(__UpperCAmelCase      ,				self.framework_tf		)

        # Both in environment -> use PyTorch
        lowerCAmelCase__				:  Union[str, Any]		       =     MagicMock(return_value=__UpperCAmelCase		)
        lowerCAmelCase__				:  List[Any]		       =     MagicMock(return_value=__UpperCAmelCase		)
        with patch('''transformers.onnx.features.is_tf_available'''      ,				__UpperCAmelCase		), patch(
            '''transformers.onnx.features.is_torch_available'''      ,				__UpperCAmelCase		):
          lowerCAmelCase__				:  List[Any]		       =     FeaturesManager.determine_framework(self.test_model		)
          self.assertEqual(__UpperCAmelCase      ,				self.framework_pt		)

        # Both not in environment -> raise error
        lowerCAmelCase__				:  Optional[int]		       =     MagicMock(return_value=__UpperCAmelCase		)
        lowerCAmelCase__				:  Dict		       =     MagicMock(return_value=__UpperCAmelCase		)
        with patch('''transformers.onnx.features.is_tf_available'''      ,				__UpperCAmelCase		), patch(
            '''transformers.onnx.features.is_torch_available'''      ,				__UpperCAmelCase		):
          with self.assertRaises(__UpperCAmelCase		):
            lowerCAmelCase__				:  Tuple		       =     FeaturesManager.determine_framework(self.test_model		)

