


import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class      UpperCAmelCase_ ( A__		,			A__		,			A__		,			unittest.TestCase   ):

   '''simple docstring'''




   __A			:				str							=		StableDiffusionImageVariationPipeline
   __A			:				Any							=		IMAGE_VARIATION_PARAMS
   __A			:				Optional[int]							=		IMAGE_VARIATION_BATCH_PARAMS
   __A			:				Optional[int]							=		frozenset([]   )
   # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
   __A			:				Dict							=		frozenset([]   )






   def _snake_case   (		self			):






     """simple docstring"""





     torch.manual_seed(0			)
     lowerCamelCase			: Optional[Any]   =    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")						,    cross_attention_dim=32						,    )
     lowerCamelCase			: Optional[int]   =    PNDMScheduler(skip_prk_steps=_A			)
     torch.manual_seed(0			)
     lowerCamelCase			: Any   =    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			)
     lowerCamelCase			: Tuple   =    CLIPVisionConfig(
         hidden_size=32						,    projection_dim=32						,    intermediate_size=37						,    layer_norm_eps=1e-05						,    num_attention_heads=4						,    num_hidden_layers=5						,    image_size=32						,    patch_size=4						,    )
     lowerCamelCase			: Any   =    CLIPVisionModelWithProjection(_A			)
     lowerCamelCase			: Optional[Any]   =    CLIPImageProcessor(crop_size=32						,    size=32			)

     lowerCamelCase			: int   =    {
         "unet": unet,
         "scheduler": scheduler,
         "vae": vae,
         "image_encoder": image_encoder,
         "feature_extractor": feature_extractor,
         "safety_checker": None,
     }
     return components






   def _snake_case   (		self						,    __A						,    __A=0			):






     """simple docstring"""





     lowerCamelCase			: List[Any]   =    floats_tensor((1, 3, 32, 32)						,    rng=random.Random(_A			)			)
     lowerCamelCase			: List[Any]   =    image.cpu().permute(0						,    2						,    3						,    1			)[0]
     lowerCamelCase			: str   =    Image.fromarray(np.uinta(_A			)			).convert("RGB"			).resize((32, 32)			)
     if str(_A			).startswith("mps"			):
       lowerCamelCase			: Union[str, Any]   =    torch.manual_seed(_A			)
     else:
       lowerCamelCase			: Dict   =    torch.Generator(device=_A			).manual_seed(_A			)
     lowerCamelCase			: Union[str, Any]   =    {
         "image": image,
         "generator": generator,
         "num_inference_steps": 2,
         "guidance_scale": 6.0,
         "output_type": "numpy",
     }
     return inputs






   def _snake_case   (		self			):






     """simple docstring"""





     lowerCamelCase			: Any   =    "cpu"  # ensure determinism for the device-dependent torch.Generator
     lowerCamelCase			: Optional[Any]   =    self.get_dummy_components()
     lowerCamelCase			: int   =    StableDiffusionImageVariationPipeline(**_A			)
     lowerCamelCase			: List[Any]   =    sd_pipe.to(_A			)
     sd_pipe.set_progress_bar_config(disable=_A			)

     lowerCamelCase			: Tuple   =    self.get_dummy_inputs(_A			)
     lowerCamelCase			: int   =    sd_pipe(**_A			).images
     lowerCamelCase			: Tuple   =    image[0, -3:, -3:, -1]

     assert image.shape == (1, 64, 64, 3)
     lowerCamelCase			: Optional[int]   =    np.array([0.5239, 0.5723, 0.4796, 0.5049, 0.5550, 0.4685, 0.5329, 0.4891, 0.4921]			)

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






   def _snake_case   (		self			):






     """simple docstring"""





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

     lowerCamelCase			: List[str]   =    self.get_dummy_inputs(_A			)
     lowerCamelCase			: List[Any]   =    2 * [inputs["image"]]
     lowerCamelCase			: Union[str, Any]   =    sd_pipe(**_A			)

     lowerCamelCase			: List[Any]   =    output.images

     lowerCamelCase			: Dict   =    image[-1, -3:, -3:, -1]

     assert image.shape == (2, 64, 64, 3)
     lowerCamelCase			: List[Any]   =    np.array([0.6892, 0.5637, 0.5836, 0.5771, 0.6254, 0.6409, 0.5580, 0.5569, 0.5289]			)

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






   def _snake_case   (		self			):






     """simple docstring"""





     super().test_inference_batch_single_identical(expected_max_diff=3e-3			)



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

   '''simple docstring'''




   def _snake_case   (		self			):






     """simple docstring"""





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






   def _snake_case   (		self						,    __A						,    __A="cpu"						,    __A=torch.floataa						,    __A=0			):






     """simple docstring"""





     lowerCamelCase			: Union[str, Any]   =    torch.Generator(device=_A			).manual_seed(_A			)
     lowerCamelCase			: str   =    load_image(
         "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
         "/stable_diffusion_imgvar/input_image_vermeer.png"			)
     lowerCamelCase			: Union[str, Any]   =    np.random.RandomState(_A			).standard_normal((1, 4, 64, 64)			)
     lowerCamelCase			: int   =    torch.from_numpy(_A			).to(device=_A						,    dtype=_A			)
     lowerCamelCase			: List[str]   =    {
         "image": init_image,
         "latents": latents,
         "generator": generator,
         "num_inference_steps": 3,
         "guidance_scale": 7.5,
         "output_type": "numpy",
     }
     return inputs






   def _snake_case   (		self			):






     """simple docstring"""





     lowerCamelCase			: List[str]   =    StableDiffusionImageVariationPipeline.from_pretrained(
         "lambdalabs/sd-image-variations-diffusers"						,    safety_checker=_A			)
     lowerCamelCase			: Dict   =    sd_pipe.to(_A			)
     sd_pipe.set_progress_bar_config(disable=_A			)

     lowerCamelCase			: Dict   =    self.get_inputs(_A			)
     lowerCamelCase			: Union[str, Any]   =    sd_pipe(**_A			).images
     lowerCamelCase			: List[Any]   =    image[0, -3:, -3:, -1].flatten()

     assert image.shape == (1, 512, 512, 3)
     lowerCamelCase			: Any   =    np.array([0.84491, 0.90789, 0.75708, 0.78734, 0.83485, 0.70099, 0.66938, 0.68727, 0.61379]			)
     assert np.abs(image_slice - expected_slice			).max() < 6e-3






   def _snake_case   (		self			):






     """simple docstring"""





     lowerCamelCase			: Dict   =    0

     def callback_fn(__A						,    __A						,    __A			) -> None:
       lowerCamelCase			: List[str]   =    True
       nonlocal number_of_steps
       number_of_steps += 1
       if step == 1:
         lowerCamelCase			: Tuple   =    latents.detach().cpu().numpy()
         assert latents.shape == (1, 4, 64, 64)
         lowerCamelCase			: List[Any]   =    latents[0, -3:, -3:, -1]
         lowerCamelCase			: int   =    np.array(
             [-0.1621, 0.2837, -0.7979, -0.1221, -1.3057, 0.7681, -2.1191, 0.0464, 1.6309]			)

         assert np.abs(latents_slice.flatten() - expected_slice			).max() < 5e-2
       elif step == 2:
         lowerCamelCase			: Any   =    latents.detach().cpu().numpy()
         assert latents.shape == (1, 4, 64, 64)
         lowerCamelCase			: Optional[int]   =    latents[0, -3:, -3:, -1]
         lowerCamelCase			: Tuple   =    np.array([0.6299, 1.7500, 1.1992, -2.1582, -1.8994, 0.7334, -0.7090, 1.0137, 1.5273]			)

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

     lowerCamelCase			: Tuple   =    False

     lowerCamelCase			: Union[str, Any]   =    StableDiffusionImageVariationPipeline.from_pretrained(
         "fusing/sd-image-variations-diffusers"						,    safety_checker=_A						,    torch_dtype=torch.floataa						,    )
     pipe.to(_A			)
     pipe.set_progress_bar_config(disable=_A			)
     pipe.enable_attention_slicing()

     lowerCamelCase			: Optional[int]   =    self.get_inputs(_A						,    dtype=torch.floataa			)
     pipe(**_A						,    callback=_A						,    callback_steps=1			)
     assert callback_fn.has_been_called
     assert number_of_steps == inputs["num_inference_steps"]






   def _snake_case   (		self			):






     """simple docstring"""





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

     lowerCamelCase			: Optional[Any]   =    "fusing/sd-image-variations-diffusers"
     lowerCamelCase			: Dict   =    StableDiffusionImageVariationPipeline.from_pretrained(
         _A						,    safety_checker=_A						,    torch_dtype=torch.floataa			)
     lowerCamelCase			: Any   =    pipe.to(_A			)
     pipe.set_progress_bar_config(disable=_A			)
     pipe.enable_attention_slicing(1			)
     pipe.enable_sequential_cpu_offload()

     lowerCamelCase			: List[str]   =    self.get_inputs(_A						,    dtype=torch.floataa			)
     lowerCamelCase			: Optional[Any]   =    pipe(**_A			)

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



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

   '''simple docstring'''




   def _snake_case   (		self			):






     """simple docstring"""





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






   def _snake_case   (		self						,    __A						,    __A="cpu"						,    __A=torch.floataa						,    __A=0			):






     """simple docstring"""





     lowerCamelCase			: List[str]   =    torch.Generator(device=_A			).manual_seed(_A			)
     lowerCamelCase			: Optional[int]   =    load_image(
         "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
         "/stable_diffusion_imgvar/input_image_vermeer.png"			)
     lowerCamelCase			: Union[str, Any]   =    np.random.RandomState(_A			).standard_normal((1, 4, 64, 64)			)
     lowerCamelCase			: int   =    torch.from_numpy(_A			).to(device=_A						,    dtype=_A			)
     lowerCamelCase			: Any   =    {
         "image": init_image,
         "latents": latents,
         "generator": generator,
         "num_inference_steps": 50,
         "guidance_scale": 7.5,
         "output_type": "numpy",
     }
     return inputs






   def _snake_case   (		self			):






     """simple docstring"""





     lowerCamelCase			: Optional[Any]   =    StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"			)
     sd_pipe.to(_A			)
     sd_pipe.set_progress_bar_config(disable=_A			)

     lowerCamelCase			: Union[str, Any]   =    self.get_inputs(_A			)
     lowerCamelCase			: List[Any]   =    sd_pipe(**_A			).images[0]

     lowerCamelCase			: List[str]   =    load_numpy(
         "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
         "/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy"			)
     lowerCamelCase			: Dict   =    np.abs(expected_image - image			).max()
     assert max_diff < 1e-3






   def _snake_case   (		self			):






     """simple docstring"""





     lowerCamelCase			: Tuple   =    StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"			)
     lowerCamelCase			: int   =    DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config			)
     sd_pipe.to(_A			)
     sd_pipe.set_progress_bar_config(disable=_A			)

     lowerCamelCase			: List[str]   =    self.get_inputs(_A			)
     lowerCamelCase			: Dict   =    25
     lowerCamelCase			: str   =    sd_pipe(**_A			).images[0]

     lowerCamelCase			: Optional[int]   =    load_numpy(
         "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
         "/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy"			)
     lowerCamelCase			: Union[str, Any]   =    np.abs(expected_image - image			).max()
     assert max_diff < 1e-3







from typing import Optional, Union

from ...file_utils import TensorType
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, TruncationStrategy
class      UpperCAmelCase_ ( UpperCamelCase   ):

   '''simple docstring'''




   __A			:				Union[str, Any]							=		"MarkupLMFeatureExtractor"
   __A			:				List[str]							=		("MarkupLMTokenizer", "MarkupLMTokenizerFast")
   __A			:				str							=		True






   def __call__(		self						,    __A=None						,    __A=None						,    __A=None						,    __A=None						,    __A=None						,    __A = True						,    __A = False						,    __A = None						,    __A = None						,    __A = 0						,    __A = None						,    __A = None						,    __A = None						,    __A = False						,    __A = False						,    __A = False						,    __A = False						,    __A = True						,    __A = None						,    **__A						,    ):






     """simple docstring"""





     if self.parse_html:
       if html_strings is None:
         raise ValueError("Make sure to pass HTML strings in case `parse_html` is set to `True`"			)

       if nodes is not None or xpaths is not None or node_labels is not None:
         raise ValueError(
             "Please don't pass nodes, xpaths nor node labels in case `parse_html` is set to `True`"			)

       lowerCamelCase			: Any   =    self.feature_extractor(__A			)
       lowerCamelCase			: Optional[int]   =    features["nodes"]
       lowerCamelCase			: Any   =    features["xpaths"]
     else:
       if html_strings is not None:
         raise ValueError("You have passed HTML strings but `parse_html` is set to `False`."			)
       if nodes is None or xpaths is None:
         raise ValueError("Make sure to pass nodes and xpaths in case `parse_html` is set to `False`"			)

        # # second, apply the tokenizer
     if questions is not None and self.parse_html:
       if isinstance(__A						,    __A			):
         lowerCamelCase			: Tuple   =    [questions]  # add batch dimension (as the feature extractor always adds a batch dimension)

     lowerCamelCase			: Tuple   =    self.tokenizer(
         text=questions if questions is not None else nodes						,    text_pair=nodes if questions is not None else None						,    xpaths=__A						,    node_labels=__A						,    add_special_tokens=__A						,    padding=__A						,    truncation=__A						,    max_length=__A						,    stride=__A						,    pad_to_multiple_of=__A						,    return_token_type_ids=__A						,    return_attention_mask=__A						,    return_overflowing_tokens=__A						,    return_special_tokens_mask=__A						,    return_offsets_mapping=__A						,    return_length=__A						,    verbose=__A						,    return_tensors=__A						,    **__A						,    )

     return encoded_inputs






   def _snake_case   (		self						,    *__A						,    **__A			):






     """simple docstring"""





     return self.tokenizer.batch_decode(*__A						,    **__A			)






   def _snake_case   (		self						,    *__A						,    **__A			):






     """simple docstring"""





     return self.tokenizer.decode(*__A						,    **__A			)






   @property
   def _snake_case   (		self			):






     """simple docstring"""





     lowerCamelCase			: Dict   =    self.tokenizer.model_input_names
     return tokenizer_input_names



