'''simple docstring'''


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      __SCREAMING_SNAKE_CASE						(       lowerCamelCase_						,	lowerCamelCase_						,	lowerCamelCase_						,	unittest.TestCase       ):







  '''simple docstring'''






  lowerCamelCase_		:Any						    =		StableDiffusionImageVariationPipeline
  lowerCamelCase_		:Union[str, Any]						    =		IMAGE_VARIATION_PARAMS
  lowerCamelCase_		:str						    =		IMAGE_VARIATION_BATCH_PARAMS
  lowerCamelCase_		:Dict						    =		frozenset([]       )
  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
  lowerCamelCase_		:List[str]						    =		frozenset([]       )


  def _UpperCamelCase     ( self						):




       '''simple docstring'''

       torch.manual_seed(0						)
       UpperCAmelCase_							:   Optional[int]   =      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      ,		)
       UpperCAmelCase_							:   Tuple   =      PNDMScheduler(skip_prk_steps=snake_case_						)
       torch.manual_seed(0						)
       UpperCAmelCase_							:   Dict   =      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      ,		)
       torch.manual_seed(0						)
       UpperCAmelCase_							:   List[Any]   =      CLIPVisionConfig(
           hidden_size=3_2      ,		projection_dim=3_2      ,		intermediate_size=3_7      ,		layer_norm_eps=1E-05      ,		num_attention_heads=4      ,		num_hidden_layers=5      ,		image_size=3_2      ,		patch_size=4      ,		)
       UpperCAmelCase_							:   str   =      CLIPVisionModelWithProjection(snake_case_						)
       UpperCAmelCase_							:   Union[str, Any]   =      CLIPImageProcessor(crop_size=3_2      ,		size=3_2						)

       UpperCAmelCase_							:   Optional[Any]   =      {
           'unet': unet,
           'scheduler': scheduler,
           'vae': vae,
           'image_encoder': image_encoder,
           'feature_extractor': feature_extractor,
           'safety_checker': None,
       }
       return components


  def _UpperCamelCase     ( self      ,		snake_case_      ,		snake_case_=0						):




       '''simple docstring'''

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


  def _UpperCamelCase     ( self						):




       '''simple docstring'''

       UpperCAmelCase_							:   Tuple   =      'cpu'  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase_							:   Union[str, Any]   =      self.get_dummy_components()
       UpperCAmelCase_							:   Dict   =      StableDiffusionImageVariationPipeline(**snake_case_						)
       UpperCAmelCase_							:   Union[str, Any]   =      sd_pipe.to(snake_case_						)
       sd_pipe.set_progress_bar_config(disable=snake_case_						)

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

       assert image.shape == (1, 6_4, 6_4, 3)
       UpperCAmelCase_							:   Dict   =      np.array([0.52_39, 0.57_23, 0.47_96, 0.50_49, 0.55_50, 0.46_85, 0.53_29, 0.48_91, 0.49_21]						)

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


  def _UpperCamelCase     ( self						):




       '''simple docstring'''

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

       UpperCAmelCase_							:   int   =      self.get_dummy_inputs(snake_case_						)
       UpperCAmelCase_							:   Optional[int]   =      2 * [inputs['image']]
       UpperCAmelCase_							:   Optional[Any]   =      sd_pipe(**snake_case_						)

       UpperCAmelCase_							:   Dict   =      output.images

       UpperCAmelCase_							:   str   =      image[-1, -3:, -3:, -1]

       assert image.shape == (2, 6_4, 6_4, 3)
       UpperCAmelCase_							:   Any   =      np.array([0.68_92, 0.56_37, 0.58_36, 0.57_71, 0.62_54, 0.64_09, 0.55_80, 0.55_69, 0.52_89]						)

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






  def _UpperCamelCase     ( self						):




       '''simple docstring'''

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







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







  '''simple docstring'''






  def _UpperCamelCase     ( self						):




       '''simple docstring'''

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


  def _UpperCamelCase     ( self      ,		snake_case_      ,		snake_case_="cpu"      ,		snake_case_=torch.floataa      ,		snake_case_=0						):




       '''simple docstring'''

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


  def _UpperCamelCase     ( self						):




       '''simple docstring'''

       UpperCAmelCase_							:   Dict   =      StableDiffusionImageVariationPipeline.from_pretrained(
           'lambdalabs/sd-image-variations-diffusers'      ,		safety_checker=snake_case_						)
       UpperCAmelCase_							:   int   =      sd_pipe.to(snake_case_						)
       sd_pipe.set_progress_bar_config(disable=snake_case_						)

       UpperCAmelCase_							:   Dict   =      self.get_inputs(snake_case_						)
       UpperCAmelCase_							:   int   =      sd_pipe(**snake_case_						).images
       UpperCAmelCase_							:   List[str]   =      image[0, -3:, -3:, -1].flatten()

       assert image.shape == (1, 5_1_2, 5_1_2, 3)
       UpperCAmelCase_							:   Optional[Any]   =      np.array([0.8_44_91, 0.9_07_89, 0.7_57_08, 0.7_87_34, 0.8_34_85, 0.7_00_99, 0.6_69_38, 0.6_87_27, 0.6_13_79]						)
       assert np.abs(image_slice - expected_slice						).max() < 6E-3


  def _UpperCamelCase     ( self						):




       '''simple docstring'''

       UpperCAmelCase_							:   Dict   =      0

       def callback_fn(snake_case_      ,		snake_case_      ,		snake_case_						) -> None:
            UpperCAmelCase_							:   str   =      True
            nonlocal number_of_steps
            number_of_steps += 1
            if step == 1:
                 UpperCAmelCase_							:   Union[str, Any]   =      latents.detach().cpu().numpy()
                 assert latents.shape == (1, 4, 6_4, 6_4)
                 UpperCAmelCase_							:   Optional[int]   =      latents[0, -3:, -3:, -1]
                 UpperCAmelCase_							:   int   =      np.array(
                     [-0.16_21, 0.28_37, -0.79_79, -0.12_21, -1.30_57, 0.76_81, -2.11_91, 0.04_64, 1.63_09]						)

                 assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5E-2
            elif step == 2:
                 UpperCAmelCase_							:   str   =      latents.detach().cpu().numpy()
                 assert latents.shape == (1, 4, 6_4, 6_4)
                 UpperCAmelCase_							:   List[str]   =      latents[0, -3:, -3:, -1]
                 UpperCAmelCase_							:   str   =      np.array([0.62_99, 1.75_00, 1.19_92, -2.15_82, -1.89_94, 0.73_34, -0.70_90, 1.01_37, 1.52_73]						)

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

       UpperCAmelCase_							:   Optional[Any]   =      False

       UpperCAmelCase_							:   List[Any]   =      StableDiffusionImageVariationPipeline.from_pretrained(
           'fusing/sd-image-variations-diffusers'      ,		safety_checker=snake_case_      ,		torch_dtype=torch.floataa      ,		)
       pipe.to(snake_case_						)
       pipe.set_progress_bar_config(disable=snake_case_						)
       pipe.enable_attention_slicing()

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






  def _UpperCamelCase     ( self						):




       '''simple docstring'''

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

       UpperCAmelCase_							:   Union[str, Any]   =      'fusing/sd-image-variations-diffusers'
       UpperCAmelCase_							:   Optional[int]   =      StableDiffusionImageVariationPipeline.from_pretrained(
           snake_case_      ,		safety_checker=snake_case_      ,		torch_dtype=torch.floataa						)
       UpperCAmelCase_							:   Optional[int]   =      pipe.to(snake_case_						)
       pipe.set_progress_bar_config(disable=snake_case_						)
       pipe.enable_attention_slicing(1						)
       pipe.enable_sequential_cpu_offload()

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

       UpperCAmelCase_							:   Dict   =      torch.cuda.max_memory_allocated()
       # make sure that less than 2.6 GB is allocated
       assert mem_bytes < 2.6 * 1_0**9







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







  '''simple docstring'''






  def _UpperCamelCase     ( self						):




       '''simple docstring'''

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


  def _UpperCamelCase     ( self      ,		snake_case_      ,		snake_case_="cpu"      ,		snake_case_=torch.floataa      ,		snake_case_=0						):




       '''simple docstring'''

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


  def _UpperCamelCase     ( self						):




       '''simple docstring'''

       UpperCAmelCase_							:   List[str]   =      StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'						)
       sd_pipe.to(snake_case_						)
       sd_pipe.set_progress_bar_config(disable=snake_case_						)

       UpperCAmelCase_							:   Optional[Any]   =      self.get_inputs(snake_case_						)
       UpperCAmelCase_							:   Optional[int]   =      sd_pipe(**snake_case_						).images[0]

       UpperCAmelCase_							:   List[Any]   =      load_numpy(
           'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
           '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'						)
       UpperCAmelCase_							:   Any   =      np.abs(expected_image - image						).max()
       assert max_diff < 1E-3






  def _UpperCamelCase     ( self						):




       '''simple docstring'''

       UpperCAmelCase_							:   Optional[int]   =      StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'						)
       UpperCAmelCase_							:   int   =      DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config						)
       sd_pipe.to(snake_case_						)
       sd_pipe.set_progress_bar_config(disable=snake_case_						)

       UpperCAmelCase_							:   Dict   =      self.get_inputs(snake_case_						)
       UpperCAmelCase_							:   Optional[Any]   =      2_5
       UpperCAmelCase_							:   List[Any]   =      sd_pipe(**snake_case_						).images[0]

       UpperCAmelCase_							:   List[str]   =      load_numpy(
           'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
           '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy'						)
       UpperCAmelCase_							:   List[Any]   =      np.abs(expected_image - image						).max()
       assert max_diff < 1E-3







'''simple docstring'''


from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


snake_case__     :			int      			=  {
    '''configuration_markuplm''': ['''MARKUPLM_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''MarkupLMConfig'''],
    '''feature_extraction_markuplm''': ['''MarkupLMFeatureExtractor'''],
    '''processing_markuplm''': ['''MarkupLMProcessor'''],
    '''tokenization_markuplm''': ['''MarkupLMTokenizer'''],
}

try:
 if not is_tokenizers_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 snake_case__     :			List[str]      			=  ['''MarkupLMTokenizerFast''']

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 snake_case__     :			Dict      			=  [
     '''MARKUPLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
     '''MarkupLMForQuestionAnswering''',
     '''MarkupLMForSequenceClassification''',
     '''MarkupLMForTokenClassification''',
     '''MarkupLMModel''',
     '''MarkupLMPreTrainedModel''',
 ]


if TYPE_CHECKING:
 from .configuration_markuplm import MARKUPLM_PRETRAINED_CONFIG_ARCHIVE_MAP, MarkupLMConfig
 from .feature_extraction_markuplm import MarkupLMFeatureExtractor
 from .processing_markuplm import MarkupLMProcessor
 from .tokenization_markuplm import MarkupLMTokenizer

 try:
  if not is_tokenizers_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .tokenization_markuplm_fast import MarkupLMTokenizerFast

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_markuplm import (
      MARKUPLM_PRETRAINED_MODEL_ARCHIVE_LIST,
      MarkupLMForQuestionAnswering,
      MarkupLMForSequenceClassification,
      MarkupLMForTokenClassification,
      MarkupLMModel,
      MarkupLMPreTrainedModel,
  )


else:
 import sys

 snake_case__     :			Optional[Any]      			=  _LazyModule(__name__, globals()['''__file__'''], _import_structure)






