'''simple docstring'''




import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    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 			lowerCamelCase_			(A__							,				A__							,				unittest.TestCase					):
      '''simple docstring'''

      __UpperCamelCase:						Tuple      			=		StableDiffusionXLPipeline
      __UpperCamelCase:						str      			=		TEXT_TO_IMAGE_PARAMS
      __UpperCamelCase:						Tuple      			=		TEXT_TO_IMAGE_BATCH_PARAMS
      __UpperCamelCase:						Optional[Any]      			=		TEXT_TO_IMAGE_IMAGE_PARAMS
      __UpperCamelCase:						int      			=		TEXT_TO_IMAGE_IMAGE_PARAMS




      def       _A			( self		:       Dict ):
             torch.manual_seed(0 )
             _UpperCAmelCase  :						Dict							   =					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=_lowerCAmelCase		,      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		,      )
             _UpperCAmelCase  :						Optional[int]							   =					EulerDiscreteScheduler(
                 beta_start=0.00_085		,      beta_end=0.012		,      steps_offset=1		,      beta_schedule="scaled_linear"		,      timestep_spacing="leading"		,      )
             torch.manual_seed(0 )
             _UpperCAmelCase  :						List[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 )
             _UpperCAmelCase  :						Tuple							   =					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		,      )
             _UpperCAmelCase  :						Tuple							   =					CLIPTextModel(_lowerCAmelCase )
             _UpperCAmelCase  :						Optional[Any]							   =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		,      local_files_only=_lowerCAmelCase )

             _UpperCAmelCase  :						Optional[Any]							   =					CLIPTextModelWithProjection(_lowerCAmelCase )
             _UpperCAmelCase  :						Any							   =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		,      local_files_only=_lowerCAmelCase )

             _UpperCAmelCase  :						str							   =					{
                 "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       _A			( self		:       Optional[Any]		,      A		:       int		,      A		:       List[Any]=0 ):
             if str(_lowerCAmelCase ).startswith("mps" ):
                    _UpperCAmelCase  :						Union[str, Any]							   =					torch.manual_seed(_lowerCAmelCase )
             else:
                    _UpperCAmelCase  :						List[str]							   =					torch.Generator(device=_lowerCAmelCase ).manual_seed(_lowerCAmelCase )
             _UpperCAmelCase  :						Any							   =					{
                 "prompt": "A painting of a squirrel eating a burger",
                 "generator": generator,
                 "num_inference_steps": 2,
                 "guidance_scale": 5.0,
                 "output_type": "numpy",
             }
             return inputs




      def       _A			( self		:       Optional[Any] ):
             _UpperCAmelCase  :						List[str]							   =					"cpu"  # ensure determinism for the device-dependent torch.Generator
             _UpperCAmelCase  :						Optional[Any]							   =					self.get_dummy_components()
             _UpperCAmelCase  :						Any							   =					StableDiffusionXLPipeline(**_lowerCAmelCase )
             _UpperCAmelCase  :						Optional[Any]							   =					sd_pipe.to(_lowerCAmelCase )
             sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase )

             _UpperCAmelCase  :						Any							   =					self.get_dummy_inputs(_lowerCAmelCase )
             _UpperCAmelCase  :						List[str]							   =					sd_pipe(**_lowerCAmelCase ).images
             _UpperCAmelCase  :						int							   =					image[0, -3:, -3:, -1]

             assert image.shape == (1, 64, 64, 3)
             _UpperCAmelCase  :						Any							   =					np.array([0.5_873, 0.6_128, 0.4_797, 0.5_122, 0.5_674, 0.4_639, 0.5_227, 0.5_149, 0.4_747] )

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




      def       _A			( self		:       List[Any] ):
             _UpperCAmelCase  :						Union[str, Any]							   =					self.get_dummy_components()
             _UpperCAmelCase  :						Union[str, Any]							   =					StableDiffusionXLPipeline(**_lowerCAmelCase )
             _UpperCAmelCase  :						Dict							   =					sd_pipe.to(_lowerCAmelCase )
             _UpperCAmelCase  :						int							   =					sd_pipe.to(_lowerCAmelCase )
             sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase )

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

             _UpperCAmelCase  :						List[Any]							   =					sd_pipe(**_lowerCAmelCase )
             _UpperCAmelCase  :						List[Any]							   =					output.images[0, -3:, -3:, -1]

             # forward with prompt embeds
             _UpperCAmelCase  :						Any							   =					self.get_dummy_inputs(_lowerCAmelCase )
             _UpperCAmelCase  :						Tuple							   =					3 * ["this is a negative prompt"]
             _UpperCAmelCase  :						Tuple							   =					3 * [inputs.pop("prompt" )]

             (
                 (
                 _UpperCAmelCase    
             )   ,						(
                 _UpperCAmelCase    
             )   ,						(
                 _UpperCAmelCase    
             )   ,						(
                 _UpperCAmelCase    
             )   ,						    
             )  :						List[str]							   =					sd_pipe.encode_prompt(_lowerCAmelCase		,      negative_prompt=_lowerCAmelCase )

             _UpperCAmelCase  :						List[Any]							   =					sd_pipe(
                 **_lowerCAmelCase		,      prompt_embeds=_lowerCAmelCase		,      negative_prompt_embeds=_lowerCAmelCase		,      pooled_prompt_embeds=_lowerCAmelCase		,      negative_pooled_prompt_embeds=_lowerCAmelCase		,      )
             _UpperCAmelCase  :						Tuple							   =					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




      def       _A			( self		:       List[Any] ):
             super().test_attention_slicing_forward_pass(expected_max_diff=3E-3 )





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

@slow
@require_torch_gpu
class 			lowerCamelCase_			(unittest.TestCase					):
      '''simple docstring'''

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




      def       _A			( self		:       Optional[int]		,      A		:       Dict		,      A		:       List[str]="cpu"		,      A		:       Union[str, Any]=torch.floataa		,      A		:       Optional[int]=0 ):
             _UpperCAmelCase  :						str							   =					torch.Generator(device=_lowerCAmelCase ).manual_seed(_lowerCAmelCase )
             _UpperCAmelCase  :						str							   =					np.random.RandomState(_lowerCAmelCase ).standard_normal((1, 4, 64, 64) )
             _UpperCAmelCase  :						Dict							   =					torch.from_numpy(_lowerCAmelCase ).to(device=_lowerCAmelCase		,      dtype=_lowerCAmelCase )
             _UpperCAmelCase  :						Dict							   =					{
                 "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       _A			( self		:       Dict ):
             _UpperCAmelCase  :						int							   =					DiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2-base" )
             pipe.to(_lowerCAmelCase )
             pipe.set_progress_bar_config(disable=_lowerCAmelCase )

             _UpperCAmelCase  :						Any							   =					self.get_inputs(_lowerCAmelCase )
             _UpperCAmelCase  :						Union[str, Any]							   =					pipe(**_lowerCAmelCase ).images
             _UpperCAmelCase  :						Any							   =					image[0, -3:, -3:, -1].flatten()

             assert image.shape == (1, 512, 512, 3)
             _UpperCAmelCase  :						Optional[Any]							   =					np.array([0.49_493, 0.47_896, 0.40_798, 0.54_214, 0.53_212, 0.48_202, 0.47_656, 0.46_329, 0.48_506] )
             assert np.abs(image_slice - expected_slice ).max() < 7E-3


'''simple docstring'''




import unittest

import requests

from transformers import MODEL_FOR_VISION_2_SEQ_MAPPING, TF_MODEL_FOR_VISION_2_SEQ_MAPPING, is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    is_torch_available,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_torch_available():
  from transformers.pytorch_utils import is_torch_greater_or_equal_than_1_11
else:
  __SCREAMING_SNAKE_CASE :  Any  =						False


if is_vision_available():
  from PIL import Image
else:

  class 			lowerCamelCase_			:
        '''simple docstring'''

        @staticmethod
        def   _A			( *A		:       str		,      **A		:       Optional[Any] ):
               pass

@is_pipeline_test
@require_vision
class 			lowerCamelCase_			(unittest.TestCase					):
      '''simple docstring'''

      __UpperCamelCase:						Union[str, Any]      			=		MODEL_FOR_VISION_2_SEQ_MAPPING
      __UpperCamelCase:						int      			=		TF_MODEL_FOR_VISION_2_SEQ_MAPPING




      def   _A			( self		:       Any		,      A		:       Any		,      A		:       str		,      A		:       Optional[int] ):
             _UpperCAmelCase  :						Dict							   =					pipeline("image-to-text"		,      model=A		,      tokenizer=A		,      image_processor=A )
             _UpperCAmelCase  :						Union[str, Any]							   =					[
                 Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png" ),
                 "./tests/fixtures/tests_samples/COCO/000000039769.png",
             ]
             return pipe, examples




      def   _A			( self		:       Tuple		,      A		:       Union[str, Any]		,      A		:       Optional[Any] ):
             _UpperCAmelCase  :						List[str]							   =					pipe(A )
             self.assertEqual(
                 A		,      [
                     [{"generated_text": ANY(A )}],
                     [{"generated_text": ANY(A )}],
                 ]		,      )




      @require_tf
      def   _A			( self		:       Any ):
             _UpperCAmelCase  :						Any							   =					pipeline("image-to-text"		,      model="hf-internal-testing/tiny-random-vit-gpt2"		,      framework="tf" )
             _UpperCAmelCase  :						int							   =					"./tests/fixtures/tests_samples/COCO/000000039769.png"

             _UpperCAmelCase  :						List[str]							   =					pipe(A )
             self.assertEqual(
                 A		,      [
                     {
                         "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                     },
                 ]		,      )

             _UpperCAmelCase  :						Tuple							   =					pipe([image, image] )
             self.assertEqual(
                 A		,      [
                     [
                         {
                             "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                         }
                     ],
                     [
                         {
                             "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                         }
                     ],
                 ]		,      )

             _UpperCAmelCase  :						Dict							   =					pipe(A		,      max_new_tokens=1 )
             self.assertEqual(
                 A		,      [{"generated_text": "growth"}]		,      )




      @require_torch
      def   _A			( self		:       Union[str, Any] ):
             _UpperCAmelCase  :						List[str]							   =					pipeline("image-to-text"		,      model="hf-internal-testing/tiny-random-vit-gpt2" )
             _UpperCAmelCase  :						Optional[int]							   =					"./tests/fixtures/tests_samples/COCO/000000039769.png"

             _UpperCAmelCase  :						Optional[Any]							   =					pipe(A )
             self.assertEqual(
                 A		,      [
                     {
                         "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                     },
                 ]		,      )

             _UpperCAmelCase  :						str							   =					pipe([image, image] )
             self.assertEqual(
                 A		,      [
                     [
                         {
                             "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                         }
                     ],
                     [
                         {
                             "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                         }
                     ],
                 ]		,      )




      @require_torch
      def   _A			( self		:       Union[str, Any] ):
             _UpperCAmelCase  :						Any							   =					pipeline("image-to-text"		,      model="hf-internal-testing/tiny-random-BlipForConditionalGeneration" )
             _UpperCAmelCase  :						Any							   =					"./tests/fixtures/tests_samples/COCO/000000039769.png"
             _UpperCAmelCase  :						int							   =					"a photo of"

             _UpperCAmelCase  :						str							   =					pipe(A		,      prompt=A )
             self.assertTrue(outputs[0]["generated_text"].startswith(A ) )




      @slow
      @require_torch
      def   _A			( self		:       List[Any] ):
             _UpperCAmelCase  :						Dict							   =					pipeline("image-to-text"		,      model="ydshieh/vit-gpt2-coco-en" )
             _UpperCAmelCase  :						int							   =					"./tests/fixtures/tests_samples/COCO/000000039769.png"

             _UpperCAmelCase  :						List[str]							   =					pipe(A )
             self.assertEqual(A		,      [{"generated_text": "a cat laying on a blanket next to a cat laying on a bed "}] )

             _UpperCAmelCase  :						int							   =					pipe([image, image] )
             self.assertEqual(
                 A		,      [
                     [{"generated_text": "a cat laying on a blanket next to a cat laying on a bed "}],
                     [{"generated_text": "a cat laying on a blanket next to a cat laying on a bed "}],
                 ]		,      )




      @slow
      @require_torch
      def   _A			( self		:       Optional[Any] ):
             _UpperCAmelCase  :						Tuple							   =					pipeline("image-to-text"		,      model="Salesforce/blip-image-captioning-base" )
             _UpperCAmelCase  :						int							   =					"https://huggingface.co/datasets/sayakpaul/sample-datasets/resolve/main/pokemon.png"
             _UpperCAmelCase  :						Optional[int]							   =					Image.open(requests.get(A		,      stream=A ).raw )

             _UpperCAmelCase  :						Any							   =					pipe(A )
             self.assertEqual(A		,      [{"generated_text": "a pink pokemon pokemon with a blue shirt and a blue shirt"}] )




      @slow
      @require_torch
      def   _A			( self		:       List[Any] ):
             _UpperCAmelCase  :						List[str]							   =					pipeline("image-to-text"		,      model="microsoft/git-base-coco" )
             _UpperCAmelCase  :						Tuple							   =					"https://huggingface.co/datasets/sayakpaul/sample-datasets/resolve/main/pokemon.png"
             _UpperCAmelCase  :						Dict							   =					Image.open(requests.get(A		,      stream=A ).raw )

             _UpperCAmelCase  :						List[str]							   =					pipe(A )
             self.assertEqual(A		,      [{"generated_text": "a cartoon of a purple character."}] )




      @slow
      @require_torch
      def   _A			( self		:       int ):
             _UpperCAmelCase  :						Any							   =					pipeline("image-to-text"		,      model="Salesforce/blip-image-captioning-base" )
             _UpperCAmelCase  :						List[str]							   =					"https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"
             _UpperCAmelCase  :						int							   =					Image.open(requests.get(A		,      stream=A ).raw )

             _UpperCAmelCase  :						Dict							   =					"a photography of"

             _UpperCAmelCase  :						Optional[int]							   =					pipe(A		,      prompt=A )
             self.assertEqual(A		,      [{"generated_text": "a photography of a volcano"}] )

             with self.assertRaises(A ):
                    _UpperCAmelCase  :						List[Any]							   =					pipe([image, image]		,      prompt=[prompt, prompt] )




      @slow
      @require_torch
      def   _A			( self		:       Union[str, Any] ):
             _UpperCAmelCase  :						Optional[int]							   =					pipeline("image-to-text"		,      model="microsoft/git-base-coco" )
             _UpperCAmelCase  :						Optional[Any]							   =					"https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"
             _UpperCAmelCase  :						List[str]							   =					Image.open(requests.get(A		,      stream=A ).raw )

             _UpperCAmelCase  :						List[Any]							   =					"a photo of a"

             _UpperCAmelCase  :						Tuple							   =					pipe(A		,      prompt=A )
             self.assertEqual(A		,      [{"generated_text": "a photo of a tent with a tent and a tent in the background."}] )

             with self.assertRaises(A ):
                    _UpperCAmelCase  :						List[str]							   =					pipe([image, image]		,      prompt=[prompt, prompt] )




      @unittest.skipIf(
          not is_torch_greater_or_equal_than_1_11		,      reason="`Pix2StructImageProcessor` requires `torch>=1.11.0`." )
      @slow
      @require_torch
      def   _A			( self		:       Union[str, Any] ):
             _UpperCAmelCase  :						str							   =					pipeline("image-to-text"		,      model="google/pix2struct-ai2d-base" )
             _UpperCAmelCase  :						List[Any]							   =					"https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"
             _UpperCAmelCase  :						List[Any]							   =					Image.open(requests.get(A		,      stream=A ).raw )

             _UpperCAmelCase  :						Optional[int]							   =					"What does the label 15 represent? (1) lava (2) core (3) tunnel (4) ash cloud"

             _UpperCAmelCase  :						str							   =					pipe(A		,      prompt=A )
             self.assertEqual(A		,      [{"generated_text": "ash cloud"}] )

             with self.assertRaises(A ):
                    _UpperCAmelCase  :						List[str]							   =					pipe([image, image]		,      prompt=[prompt, prompt] )





      @slow
      @require_tf
      def   _A			( self		:       List[Any] ):
             _UpperCAmelCase  :						str							   =					pipeline("image-to-text"		,      model="ydshieh/vit-gpt2-coco-en"		,      framework="tf" )
             _UpperCAmelCase  :						str							   =					"./tests/fixtures/tests_samples/COCO/000000039769.png"

             _UpperCAmelCase  :						Any							   =					pipe(A )
             self.assertEqual(A		,      [{"generated_text": "a cat laying on a blanket next to a cat laying on a bed "}] )

             _UpperCAmelCase  :						Optional[Any]							   =					pipe([image, image] )
             self.assertEqual(
                 A		,      [
                     [{"generated_text": "a cat laying on a blanket next to a cat laying on a bed "}],
                     [{"generated_text": "a cat laying on a blanket next to a cat laying on a bed "}],
                 ]		,      )

