



import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()


class   snake_case_ (lowercase__ , lowercase__ , lowercase__ , unittest.TestCase  ):




       """simple docstring"""






       _lowerCamelCase			    =				StableUnCLIPImgaImgPipeline
       _lowerCamelCase			    =				TEXT_GUIDED_IMAGE_VARIATION_PARAMS
       _lowerCamelCase			    =				TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
       _lowerCamelCase			    =				frozenset(
           []  )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
       _lowerCamelCase			    =				frozenset([]  )



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	Optional[Any]				      =		32
             UpperCAmelCase_  :	List[Any]				      =		embedder_hidden_size

             # image encoding components

             UpperCAmelCase_  :	str				      =		CLIPImageProcessor(crop_size=32     ,size=32)

             torch.manual_seed(0)
             UpperCAmelCase_  :	Union[str, Any]				      =		CLIPVisionModelWithProjection(
                 CLIPVisionConfig(
                     hidden_size=lowercase     ,projection_dim=lowercase     ,num_hidden_layers=5     ,num_attention_heads=4     ,image_size=32     ,intermediate_size=37     ,patch_size=1     ,))

             # regular denoising components

             torch.manual_seed(0)
             UpperCAmelCase_  :	str				      =		StableUnCLIPImageNormalizer(embedding_dim=lowercase)
             UpperCAmelCase_  :	Tuple				      =		DDPMScheduler(beta_schedule="squaredcos_cap_v2")

             torch.manual_seed(0)
             UpperCAmelCase_  :	List[Any]				      =		CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

             torch.manual_seed(0)
             UpperCAmelCase_  :	Tuple				      =		CLIPTextModel(
                 CLIPTextConfig(
                     bos_token_id=0     ,eos_token_id=2     ,hidden_size=lowercase     ,projection_dim=32     ,intermediate_size=37     ,layer_norm_eps=1E-05     ,num_attention_heads=4     ,num_hidden_layers=5     ,pad_token_id=1     ,vocab_size=1000     ,))

             torch.manual_seed(0)
             UpperCAmelCase_  :	Optional[Any]				      =		UNetaDConditionModel(
                 sample_size=32     ,in_channels=4     ,out_channels=4     ,down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")     ,up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")     ,block_out_channels=(32, 64)     ,attention_head_dim=(2, 4)     ,class_embed_type="projection"     ,projection_class_embeddings_input_dim=embedder_projection_dim * 2     ,cross_attention_dim=lowercase     ,layers_per_block=1     ,upcast_attention=lowercase     ,use_linear_projection=lowercase     ,)

             torch.manual_seed(0)
             UpperCAmelCase_  :	Dict				      =		DDIMScheduler(
                 beta_schedule="scaled_linear"     ,beta_start=0.0_0085     ,beta_end=0.012     ,prediction_type="v_prediction"     ,set_alpha_to_one=lowercase     ,steps_offset=1     ,)

             torch.manual_seed(0)
             UpperCAmelCase_  :	Tuple				      =		AutoencoderKL()

             UpperCAmelCase_  :	str				      =		{
                 # image encoding components
                 "feature_extractor": feature_extractor,
                 "image_encoder": image_encoder.eval(),
                 # image noising components
                 "image_normalizer": image_normalizer.eval(),
                 "image_noising_scheduler": image_noising_scheduler,
                 # regular denoising components
                 "tokenizer": tokenizer,
                 "text_encoder": text_encoder.eval(),
                 "unet": unet.eval(),
                 "scheduler": scheduler,
                 "vae": vae.eval(),
             }

             return components



       def  A_   (     self     ,lowercase     ,lowercase=0     ,lowercase=True):

             """simple docstring"""





             if str(lowercase).startswith("mps"):
                   UpperCAmelCase_  :	Optional[int]				      =		torch.manual_seed(lowercase)
             else:
                   UpperCAmelCase_  :	int				      =		torch.Generator(device=lowercase).manual_seed(lowercase)

             UpperCAmelCase_  :	Dict				      =		floats_tensor((1, 3, 32, 32)     ,rng=random.Random(lowercase)).to(lowercase)

             if pil_image:
                   UpperCAmelCase_  :	Union[str, Any]				      =		input_image * 0.5 + 0.5
                   UpperCAmelCase_  :	Optional[int]				      =		input_image.clamp(0     ,1)
                   UpperCAmelCase_  :	List[Any]				      =		input_image.cpu().permute(0     ,2     ,3     ,1).float().numpy()
                   UpperCAmelCase_  :	Tuple				      =		DiffusionPipeline.numpy_to_pil(lowercase)[0]

             return {
                 "prompt": "An anime racoon running a marathon",
                 "image": input_image,
                 "generator": generator,
                 "num_inference_steps": 2,
                 "output_type": "np",
             }



       @skip_mps
       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	Optional[int]				      =		"cpu"  # ensure determinism for the device-dependent torch.Generator
             UpperCAmelCase_  :	str				      =		self.get_dummy_components()
             UpperCAmelCase_  :	Dict				      =		StableUnCLIPImgaImgPipeline(**lowercase)
             UpperCAmelCase_  :	Tuple				      =		sd_pipe.to(lowercase)
             sd_pipe.set_progress_bar_config(disable=lowercase)

             UpperCAmelCase_  :	List[str]				      =		self.get_dummy_inputs(lowercase)
             inputs.update({"image_embeds": None})
             UpperCAmelCase_  :	Optional[Any]				      =		sd_pipe(**lowercase).images
             UpperCAmelCase_  :	Optional[Any]				      =		image[0, -3:, -3:, -1]

             assert image.shape == (1, 32, 32, 3)
             UpperCAmelCase_  :	Tuple				      =		np.array([0.3872, 0.7224, 0.5601, 0.4741, 0.6872, 0.5814, 0.4636, 0.3867, 0.5078])

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



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	str				      =		torch_device in ["cpu", "mps"]

             self._test_attention_slicing_forward_pass(test_max_difference=lowercase)



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	Dict				      =		torch_device in ["cpu", "mps"]

             self._test_inference_batch_single_identical(test_max_difference=lowercase)



       @unittest.skipIf(
           torch_device != "cuda" or not is_xformers_available()     ,reason="XFormers attention is only available with CUDA and `xformers` installed"     ,)
       def  A_   (     self):

             """simple docstring"""





             self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowercase)


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




       """simple docstring"""






       def  A_   (     self):

             """simple docstring"""





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



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	str				      =		load_image(
                 "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png")

             UpperCAmelCase_  :	Dict				      =		load_numpy(
                 "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy")

             UpperCAmelCase_  :	Optional[int]				      =		StableUnCLIPImgaImgPipeline.from_pretrained(
                 "fusing/stable-unclip-2-1-l-img2img"     ,torch_dtype=torch.floataa)
             pipe.to(lowercase)
             pipe.set_progress_bar_config(disable=lowercase)
             # stable unclip will oom when integration tests are run on a V100,
             # so turn on memory savings
             pipe.enable_attention_slicing()
             pipe.enable_sequential_cpu_offload()

             UpperCAmelCase_  :	str				      =		torch.Generator(device="cpu").manual_seed(0)
             UpperCAmelCase_  :	Dict				      =		pipe(lowercase     ,"anime turle"     ,generator=lowercase     ,output_type="np")

             UpperCAmelCase_  :	List[str]				      =		output.images[0]

             assert image.shape == (768, 768, 3)

             assert_mean_pixel_difference(lowercase     ,lowercase)



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	int				      =		load_image(
                 "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png")

             UpperCAmelCase_  :	Any				      =		load_numpy(
                 "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy")

             UpperCAmelCase_  :	Dict				      =		StableUnCLIPImgaImgPipeline.from_pretrained(
                 "fusing/stable-unclip-2-1-h-img2img"     ,torch_dtype=torch.floataa)
             pipe.to(lowercase)
             pipe.set_progress_bar_config(disable=lowercase)
             # stable unclip will oom when integration tests are run on a V100,
             # so turn on memory savings
             pipe.enable_attention_slicing()
             pipe.enable_sequential_cpu_offload()

             UpperCAmelCase_  :	Optional[Any]				      =		torch.Generator(device="cpu").manual_seed(0)
             UpperCAmelCase_  :	List[Any]				      =		pipe(lowercase     ,"anime turle"     ,generator=lowercase     ,output_type="np")

             UpperCAmelCase_  :	Optional[int]				      =		output.images[0]

             assert image.shape == (768, 768, 3)

             assert_mean_pixel_difference(lowercase     ,lowercase)



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	Union[str, Any]				      =		load_image(
                 "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png")

             torch.cuda.empty_cache()
             torch.cuda.reset_max_memory_allocated()
             torch.cuda.reset_peak_memory_stats()

             UpperCAmelCase_  :	Any				      =		StableUnCLIPImgaImgPipeline.from_pretrained(
                 "fusing/stable-unclip-2-1-h-img2img"     ,torch_dtype=torch.floataa)
             UpperCAmelCase_  :	int				      =		pipe.to(lowercase)
             pipe.set_progress_bar_config(disable=lowercase)
             pipe.enable_attention_slicing()
             pipe.enable_sequential_cpu_offload()

             UpperCAmelCase_  :	int				      =		pipe(
                 lowercase     ,"anime turtle"     ,num_inference_steps=2     ,output_type="np"     ,)

             UpperCAmelCase_  :	Tuple				      =		torch.cuda.max_memory_allocated()
             # make sure that less than 7 GB is allocated
             assert mem_bytes < 7 * 10**9





# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
   import torch


class   snake_case_ (TensorFormatter[Mapping, """torch.Tensor""", Mapping]  ):




       """simple docstring"""






       def __init__(     self     ,lowercase=None     ,**lowercase):

             """simple docstring"""





             super().__init__(features=lowercase)
             UpperCAmelCase_  :	Union[str, Any]				      =		torch_tensor_kwargs
             import torch  # noqa import torch at initialization



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             import torch

             if isinstance(lowercase     ,lowercase) and column:
                   if all(
                       isinstance(lowercase     ,torch.Tensor) and x.shape == column[0].shape and x.dtype == column[0].dtype
                       for x in column):
                         return torch.stack(lowercase)
             return column



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             import torch

             if isinstance(lowercase     ,(str, bytes, type(lowercase))):
                   return value
             elif isinstance(lowercase     ,(np.character, np.ndarray)) and np.issubdtype(value.dtype     ,np.character):
                   return value.tolist()

             UpperCAmelCase_  :	List[str]				      =		{}

             if isinstance(lowercase     ,(np.number, np.ndarray)) and np.issubdtype(value.dtype     ,np.integer):
                   UpperCAmelCase_  :	Dict				      =		{"dtype": torch.intaa}
             elif isinstance(lowercase     ,(np.number, np.ndarray)) and np.issubdtype(value.dtype     ,np.floating):
                   UpperCAmelCase_  :	Optional[Any]				      =		{"dtype": torch.floataa}
             elif config.PIL_AVAILABLE and "PIL" in sys.modules:
                   import PIL.Image

                   if isinstance(lowercase     ,PIL.Image.Image):
                         UpperCAmelCase_  :	List[str]				      =		np.asarray(lowercase)
             return torch.tensor(lowercase     ,**{**default_dtype, **self.torch_tensor_kwargs})



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             import torch

             # support for torch, tf, jax etc.
             if hasattr(lowercase     ,"__array__") and not isinstance(lowercase     ,torch.Tensor):
                   UpperCAmelCase_  :	int				      =		data_struct.__array__()
             # support for nested types like struct of list of struct
             if isinstance(lowercase     ,np.ndarray):
                   if data_struct.dtype == object:  # torch tensors cannot be instantied from an array of objects
                         return self._consolidate([self.recursive_tensorize(lowercase) for substruct in data_struct])
             elif isinstance(lowercase     ,(list, tuple)):
                   return self._consolidate([self.recursive_tensorize(lowercase) for substruct in data_struct])
             return self._tensorize(lowercase)



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             return map_nested(self._recursive_tensorize     ,lowercase     ,map_list=lowercase)



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             UpperCAmelCase_  :	List[Any]				      =		self.numpy_arrow_extractor().extract_row(lowercase)
             UpperCAmelCase_  :	Tuple				      =		self.python_features_decoder.decode_row(lowercase)
             return self.recursive_tensorize(lowercase)



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             UpperCAmelCase_  :	str				      =		self.numpy_arrow_extractor().extract_column(lowercase)
             UpperCAmelCase_  :	List[Any]				      =		self.python_features_decoder.decode_column(lowercase     ,pa_table.column_names[0])
             UpperCAmelCase_  :	List[Any]				      =		self.recursive_tensorize(lowercase)
             UpperCAmelCase_  :	List[str]				      =		self._consolidate(lowercase)
             return column



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             UpperCAmelCase_  :	Any				      =		self.numpy_arrow_extractor().extract_batch(lowercase)
             UpperCAmelCase_  :	Optional[int]				      =		self.python_features_decoder.decode_batch(lowercase)
             UpperCAmelCase_  :	List[Any]				      =		self.recursive_tensorize(lowercase)
             for column_name in batch:
                   UpperCAmelCase_  :	Optional[int]				      =		self._consolidate(batch[column_name])
             return batch
