
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     __A(				a						,     a						,     a						,     unittest.TestCase    ):
				snake_case_		     =					StableDiffusionImageVariationPipeline
				snake_case_		     =					IMAGE_VARIATION_PARAMS
				snake_case_		     =					IMAGE_VARIATION_BATCH_PARAMS
				snake_case_		     =					frozenset([]    )
				# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
				snake_case_		     =					frozenset([]    )




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   List[Any]:


								'''simple docstring'''




								torch.manual_seed(0       )
								__a    							=					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	,  )
								__a    							=					PNDMScheduler(skip_prk_steps=_snake_case       )
								torch.manual_seed(0       )
								__a    							=					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       )
								__a    							=					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	,  )
								__a    							=					CLIPVisionModelWithProjection(_snake_case       )
								__a    							=					CLIPImageProcessor(crop_size=32	,  size=32       )

								__a    							=					{
								    '''unet''': unet,
								    '''scheduler''': scheduler,
								    '''vae''': vae,
								    '''image_encoder''': image_encoder,
								    '''feature_extractor''': feature_extractor,
								    '''safety_checker''': None,
								}
								return components




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case=0       )					->   int:


								'''simple docstring'''




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




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   List[str]:


								'''simple docstring'''




								__a    							=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
								__a    							=					self.get_dummy_components()
								__a    							=					StableDiffusionImageVariationPipeline(**_snake_case       )
								__a    							=					sd_pipe.to(_snake_case       )
								sd_pipe.set_progress_bar_config(disable=_snake_case       )

								__a    							=					self.get_dummy_inputs(_snake_case       )
								__a    							=					sd_pipe(**_snake_case       ).images
								__a    							=					image[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)
								__a    							=					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        SCREAMING_SNAKE_CASE_	(  self       )					->   Any:


								'''simple docstring'''




								__a    							=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
								__a    							=					self.get_dummy_components()
								__a    							=					StableDiffusionImageVariationPipeline(**_snake_case       )
								__a    							=					sd_pipe.to(_snake_case       )
								sd_pipe.set_progress_bar_config(disable=_snake_case       )

								__a    							=					self.get_dummy_inputs(_snake_case       )
								__a    							=					2 * [inputs['''image''']]
								__a    							=					sd_pipe(**_snake_case       )

								__a    							=					output.images

								__a    							=					image[-1, -3:, -3:, -1]

								assert image.shape == (2, 64, 64, 3)
								__a    							=					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        SCREAMING_SNAKE_CASE_	(  self       )					->   List[str]:


								'''simple docstring'''




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







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




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   List[Any]:


								'''simple docstring'''




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




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case="cpu"	,  _snake_case=torch.floataa	,  _snake_case=0       )					->   List[str]:


								'''simple docstring'''




								__a    							=					torch.Generator(device=_snake_case       ).manual_seed(_snake_case       )
								__a    							=					load_image(
								    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
								    '''/stable_diffusion_imgvar/input_image_vermeer.png'''       )
								__a    							=					np.random.RandomState(_snake_case       ).standard_normal((1, 4, 64, 64)       )
								__a    							=					torch.from_numpy(_snake_case       ).to(device=_snake_case	,  dtype=_snake_case       )
								__a    							=					{
								    '''image''': init_image,
								    '''latents''': latents,
								    '''generator''': generator,
								    '''num_inference_steps''': 3,
								    '''guidance_scale''': 7.5,
								    '''output_type''': '''numpy''',
								}
								return inputs




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Optional[Any]:


								'''simple docstring'''




								__a    							=					StableDiffusionImageVariationPipeline.from_pretrained(
								    '''lambdalabs/sd-image-variations-diffusers'''	,  safety_checker=_snake_case       )
								__a    							=					sd_pipe.to(_snake_case       )
								sd_pipe.set_progress_bar_config(disable=_snake_case       )

								__a    							=					self.get_inputs(_snake_case       )
								__a    							=					sd_pipe(**_snake_case       ).images
								__a    							=					image[0, -3:, -3:, -1].flatten()

								assert image.shape == (1, 512, 512, 3)
								__a    							=					np.array([0.8_4491, 0.9_0789, 0.7_5708, 0.7_8734, 0.8_3485, 0.7_0099, 0.6_6938, 0.6_8727, 0.6_1379]       )
								assert np.abs(image_slice - expected_slice       ).max() < 6E-3




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Dict:


								'''simple docstring'''




								__a    							=					0

								def callback_fn(_snake_case	,  _snake_case	,  _snake_case       ) -> None:
												__a    							=					True
												nonlocal number_of_steps
												number_of_steps += 1
												if step == 1:
																__a    							=					latents.detach().cpu().numpy()
																assert latents.shape == (1, 4, 64, 64)
																__a    							=					latents[0, -3:, -3:, -1]
																__a    							=					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:
																__a    							=					latents.detach().cpu().numpy()
																assert latents.shape == (1, 4, 64, 64)
																__a    							=					latents[0, -3:, -3:, -1]
																__a    							=					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

								__a    							=					False

								__a    							=					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()

								__a    							=					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        SCREAMING_SNAKE_CASE_	(  self       )					->   Tuple:


								'''simple docstring'''




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

								__a    							=					'''fusing/sd-image-variations-diffusers'''
								__a    							=					StableDiffusionImageVariationPipeline.from_pretrained(
								    _snake_case	,  safety_checker=_snake_case	,  torch_dtype=torch.floataa       )
								__a    							=					pipe.to(_snake_case       )
								pipe.set_progress_bar_config(disable=_snake_case       )
								pipe.enable_attention_slicing(1       )
								pipe.enable_sequential_cpu_offload()

								__a    							=					self.get_inputs(_snake_case	,  dtype=torch.floataa       )
								__a    							=					pipe(**_snake_case       )

								__a    							=					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     __A(				unittest.TestCase    ):




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Any:


								'''simple docstring'''




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




				def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case="cpu"	,  _snake_case=torch.floataa	,  _snake_case=0       )					->   List[Any]:


								'''simple docstring'''




								__a    							=					torch.Generator(device=_snake_case       ).manual_seed(_snake_case       )
								__a    							=					load_image(
								    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
								    '''/stable_diffusion_imgvar/input_image_vermeer.png'''       )
								__a    							=					np.random.RandomState(_snake_case       ).standard_normal((1, 4, 64, 64)       )
								__a    							=					torch.from_numpy(_snake_case       ).to(device=_snake_case	,  dtype=_snake_case       )
								__a    							=					{
								    '''image''': init_image,
								    '''latents''': latents,
								    '''generator''': generator,
								    '''num_inference_steps''': 50,
								    '''guidance_scale''': 7.5,
								    '''output_type''': '''numpy''',
								}
								return inputs




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Any:


								'''simple docstring'''




								__a    							=					StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''       )
								sd_pipe.to(_snake_case       )
								sd_pipe.set_progress_bar_config(disable=_snake_case       )

								__a    							=					self.get_inputs(_snake_case       )
								__a    							=					sd_pipe(**_snake_case       ).images[0]

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




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Tuple:


								'''simple docstring'''




								__a    							=					StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''       )
								__a    							=					DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config       )
								sd_pipe.to(_snake_case       )
								sd_pipe.set_progress_bar_config(disable=_snake_case       )

								__a    							=					self.get_inputs(_snake_case       )
								__a    							=					25
								__a    							=					sd_pipe(**_snake_case       ).images[0]

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

from ...configuration_utils import PretrainedConfig
from ...utils import is_detectrona_available, logging


A  :			int     				=				logging.get_logger(__name__)

A  :			Dict     				=				{
    'layoutlmv2-base-uncased': 'https://huggingface.co/microsoft/layoutlmv2-base-uncased/resolve/main/config.json',
    'layoutlmv2-large-uncased': 'https://huggingface.co/microsoft/layoutlmv2-large-uncased/resolve/main/config.json',
    # See all LayoutLMv2 models at https://huggingface.co/models?filter=layoutlmv2
}

# soft dependency
if is_detectrona_available():
					import detectrona







class     __A(				a    ):
				snake_case_		     =					'''layoutlmv2'''




				def __init__(  self	,  _snake_case=30_522	,  _snake_case=768	,  _snake_case=12	,  _snake_case=12	,  _snake_case=3_072	,  _snake_case="gelu"	,  _snake_case=0.1	,  _snake_case=0.1	,  _snake_case=512	,  _snake_case=2	,  _snake_case=0.02	,  _snake_case=1E-12	,  _snake_case=0	,  _snake_case=1_024	,  _snake_case=128	,  _snake_case=32	,  _snake_case=True	,  _snake_case=256	,  _snake_case=64	,  _snake_case=True	,  _snake_case=[7, 7, 256]	,  _snake_case=128	,  _snake_case=128	,  _snake_case=True	,  _snake_case=True	,  _snake_case=False	,  _snake_case=None	,  **_snake_case	,  )					->   Optional[int]:


								'''simple docstring'''




								super().__init__(
								    vocab_size=_snake_case	,  hidden_size=_snake_case	,  num_hidden_layers=_snake_case	,  num_attention_heads=_snake_case	,  intermediate_size=_snake_case	,  hidden_act=_snake_case	,  hidden_dropout_prob=_snake_case	,  attention_probs_dropout_prob=_snake_case	,  max_position_embeddings=_snake_case	,  type_vocab_size=_snake_case	,  initializer_range=_snake_case	,  layer_norm_eps=_snake_case	,  pad_token_id=_snake_case	,  **_snake_case	,  )
								__a    							=					max_ad_position_embeddings
								__a    							=					max_rel_pos
								__a    							=					rel_pos_bins
								__a    							=					fast_qkv
								__a    							=					max_rel_ad_pos
								__a    							=					rel_ad_pos_bins
								__a    							=					convert_sync_batchnorm
								__a    							=					image_feature_pool_shape
								__a    							=					coordinate_size
								__a    							=					shape_size
								__a    							=					has_relative_attention_bias
								__a    							=					has_spatial_attention_bias
								__a    							=					has_visual_segment_embedding
								__a    							=					(
								    detectrona_config_args if detectrona_config_args is not None else self.get_default_detectrona_config()
								)




				@classmethod
				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Dict:


								'''simple docstring'''




								return {
								    "MODEL.MASK_ON": True,
								    "MODEL.PIXEL_STD": [57.375, 57.120, 58.395],
								    "MODEL.BACKBONE.NAME": "build_resnet_fpn_backbone",
								    "MODEL.FPN.IN_FEATURES": ["res2", "res3", "res4", "res5"],
								    "MODEL.ANCHOR_GENERATOR.SIZES": [[32], [64], [128], [256], [512]],
								    "MODEL.RPN.IN_FEATURES": ["p2", "p3", "p4", "p5", "p6"],
								    "MODEL.RPN.PRE_NMS_TOPK_TRAIN": 2_000,
								    "MODEL.RPN.PRE_NMS_TOPK_TEST": 1_000,
								    "MODEL.RPN.POST_NMS_TOPK_TRAIN": 1_000,
								    "MODEL.POST_NMS_TOPK_TEST": 1_000,
								    "MODEL.ROI_HEADS.NAME": "StandardROIHeads",
								    "MODEL.ROI_HEADS.NUM_CLASSES": 5,
								    "MODEL.ROI_HEADS.IN_FEATURES": ["p2", "p3", "p4", "p5"],
								    "MODEL.ROI_BOX_HEAD.NAME": "FastRCNNConvFCHead",
								    "MODEL.ROI_BOX_HEAD.NUM_FC": 2,
								    "MODEL.ROI_BOX_HEAD.POOLER_RESOLUTION": 14,
								    "MODEL.ROI_MASK_HEAD.NAME": "MaskRCNNConvUpsampleHead",
								    "MODEL.ROI_MASK_HEAD.NUM_CONV": 4,
								    "MODEL.ROI_MASK_HEAD.POOLER_RESOLUTION": 7,
								    "MODEL.RESNETS.DEPTH": 101,
								    "MODEL.RESNETS.SIZES": [[32], [64], [128], [256], [512]],
								    "MODEL.RESNETS.ASPECT_RATIOS": [[0.5, 1.0, 2.0]],
								    "MODEL.RESNETS.OUT_FEATURES": ["res2", "res3", "res4", "res5"],
								    "MODEL.RESNETS.NUM_GROUPS": 32,
								    "MODEL.RESNETS.WIDTH_PER_GROUP": 8,
								    "MODEL.RESNETS.STRIDE_IN_1X1": False,
								}




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Optional[Any]:


								'''simple docstring'''




								__a    							=					detectrona.config.get_cfg()
								for k, v in self.detectrona_config_args.items():
												__a    							=					k.split('''.'''       )
												__a    							=					detectrona_config
												for attribute in attributes[:-1]:
																__a    							=					getattr(_snake_case	,  _snake_case       )
												setattr(_snake_case	,  attributes[-1]	,  _snake_case       )

								return detectrona_config