



'''simple docstring'''

import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

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


enable_full_determinism()

class __snake_case(				_lowerCAmelCase		,							_lowerCAmelCase		,							unittest.TestCase ):


						'''simple docstring'''



						UpperCAmelCase							:   Union[str, Any]									=		StableDiffusionXLImgaImgPipeline
						UpperCAmelCase							:   Any									=		TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
						UpperCAmelCase							:   Dict									=		PipelineTesterMixin.required_optional_params - {"latents"}
						UpperCAmelCase							:   List[str]									=		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
						UpperCAmelCase							:   Tuple									=		IMAGE_TO_IMAGE_IMAGE_PARAMS
						UpperCAmelCase							:   Optional[int]									=		IMAGE_TO_IMAGE_IMAGE_PARAMS


						def     __snake_case   ( self				)     ->     Union[str, Any]:
							torch.manual_seed(0				)
							lowerCAmelCase   =     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=A_     ,	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     ,	)
							lowerCAmelCase   =     EulerDiscreteScheduler(
							    beta_start=0.0_0_0_8_5     ,	beta_end=0.0_1_2     ,	steps_offset=1     ,	beta_schedule="""scaled_linear"""     ,	timestep_spacing="""leading"""     ,	)
							torch.manual_seed(0				)
							lowerCAmelCase   =     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				)
							lowerCAmelCase   =     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     ,	)
							lowerCAmelCase   =     CLIPTextModel(A_				)
							lowerCAmelCase   =     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     ,	local_files_only=A_				)

							lowerCAmelCase   =     CLIPTextModelWithProjection(A_				)
							lowerCAmelCase   =     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     ,	local_files_only=A_				)

							lowerCAmelCase   =     {
							    """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     __snake_case   ( self     ,	A_     ,	A_=0				)     ->     List[str]:
							lowerCAmelCase   =     floats_tensor((1, 3, 32, 32)     ,	rng=random.Random(A_				)				).to(A_				)
							lowerCAmelCase   =     image / 2 + 0.5
							if str(A_				).startswith("""mps"""				):
								lowerCAmelCase   =     torch.manual_seed(A_				)
							else:
								lowerCAmelCase   =     torch.Generator(device=A_				).manual_seed(A_				)
							lowerCAmelCase   =     {
							    """prompt""": """A painting of a squirrel eating a burger""",
							    """image""": image,
							    """generator""": generator,
							    """num_inference_steps""": 2,
							    """guidance_scale""": 5.0,
							    """output_type""": """numpy""",
							    """strength""": 0.7_5,
							}
							return inputs


						def     __snake_case   ( self				)     ->     str:
							lowerCAmelCase   =     """cpu"""  # ensure determinism for the device-dependent torch.Generator
							lowerCAmelCase   =     self.get_dummy_components()
							lowerCAmelCase   =     StableDiffusionXLImgaImgPipeline(**A_				)
							lowerCAmelCase   =     sd_pipe.to(A_				)
							sd_pipe.set_progress_bar_config(disable=A_				)

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

							assert image.shape == (1, 32, 32, 3)

							lowerCAmelCase   =     np.array([0.4_6_5_6, 0.4_8_4_0, 0.4_4_3_9, 0.6_6_9_8, 0.5_5_7_4, 0.4_5_2_4, 0.5_7_9_9, 0.5_9_4_3, 0.5_1_6_5]				)

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


						def     __snake_case   ( self				)     ->     Union[str, Any]:
							super().test_attention_slicing_forward_pass(expected_max_diff=3e-3				)


						def     __snake_case   ( self				)     ->     Dict:
							super().test_inference_batch_single_identical(expected_max_diff=3e-3				)


						def     __snake_case   ( self				)     ->     Optional[int]:
							pass




						def     __snake_case   ( self				)     ->     List[str]:
							lowerCAmelCase   =     self.get_dummy_components()
							lowerCAmelCase   =     StableDiffusionXLImgaImgPipeline(**A_				)
							lowerCAmelCase   =     sd_pipe.to(A_				)
							lowerCAmelCase   =     sd_pipe.to(A_				)
							sd_pipe.set_progress_bar_config(disable=A_				)

							# forward without prompt embeds
							lowerCAmelCase   =     self.get_dummy_inputs(A_				)
							lowerCAmelCase   =     3 * ["""this is a negative prompt"""]
							lowerCAmelCase   =     negative_prompt
							lowerCAmelCase   =     3 * [inputs["""prompt"""]]

							lowerCAmelCase   =     sd_pipe(**A_				)
							lowerCAmelCase   =     output.images[0, -3:, -3:, -1]

							# forward with prompt embeds
							lowerCAmelCase   =     self.get_dummy_inputs(A_				)
							lowerCAmelCase   =     3 * ["""this is a negative prompt"""]
							lowerCAmelCase   =     3 * [inputs.pop("""prompt"""				)]

							(
							    (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       
							)   =     sd_pipe.encode_prompt(A_     ,	negative_prompt=A_				)

							lowerCAmelCase   =     sd_pipe(
							    **A_     ,	prompt_embeds=A_     ,	negative_prompt_embeds=A_     ,	pooled_prompt_embeds=A_     ,	negative_pooled_prompt_embeds=A_     ,	)
							lowerCAmelCase   =     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







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


						'''simple docstring'''



						def     __snake_case   ( self				)     ->     Optional[Any]:
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()


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

							lowerCAmelCase   =     self.get_inputs(A_				)
							lowerCAmelCase   =     pipe(**A_				).images
							lowerCAmelCase   =     image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 512, 512, 3)
							lowerCAmelCase   =     np.array([0.4_9_4_9_3, 0.4_7_8_9_6, 0.4_0_7_9_8, 0.5_4_2_1_4, 0.5_3_2_1_2, 0.4_8_2_0_2, 0.4_7_6_5_6, 0.4_6_3_2_9, 0.4_8_5_0_6]				)
							assert np.abs(image_slice - expected_slice				).max() < 7e-3




'''simple docstring'''

import unittest

from transformers import LiltConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
				import torch

				from transformers import (
				    LiltForQuestionAnswering,
				    LiltForSequenceClassification,
				    LiltForTokenClassification,
				    LiltModel,
				)
				from transformers.models.lilt.modeling_lilt import LILT_PRETRAINED_MODEL_ARCHIVE_LIST

class __snake_case:


						'''simple docstring'''



						def __init__( self     ,	A_     ,	A_=13     ,	A_=7     ,	A_=True     ,	A_=True     ,	A_=True     ,	A_=True     ,	A_=99     ,	A_=24     ,	A_=2     ,	A_=6     ,	A_=37     ,	A_="gelu"     ,	A_=0.1     ,	A_=0.1     ,	A_=512     ,	A_=16     ,	A_=2     ,	A_=0.0_2     ,	A_=3     ,	A_=None     ,	A_=1000     ,	)     ->     Union[str, Any]:
							lowerCAmelCase   =     parent
							lowerCAmelCase   =     batch_size
							lowerCAmelCase   =     seq_length
							lowerCAmelCase   =     is_training
							lowerCAmelCase   =     use_input_mask
							lowerCAmelCase   =     use_token_type_ids
							lowerCAmelCase   =     use_labels
							lowerCAmelCase   =     vocab_size
							lowerCAmelCase   =     hidden_size
							lowerCAmelCase   =     num_hidden_layers
							lowerCAmelCase   =     num_attention_heads
							lowerCAmelCase   =     intermediate_size
							lowerCAmelCase   =     hidden_act
							lowerCAmelCase   =     hidden_dropout_prob
							lowerCAmelCase   =     attention_probs_dropout_prob
							lowerCAmelCase   =     max_position_embeddings
							lowerCAmelCase   =     type_vocab_size
							lowerCAmelCase   =     type_sequence_label_size
							lowerCAmelCase   =     initializer_range
							lowerCAmelCase   =     num_labels
							lowerCAmelCase   =     scope
							lowerCAmelCase   =     range_bbox


						def     __snake_case   ( self				)     ->     List[Any]:
							lowerCAmelCase   =     ids_tensor([self.batch_size, self.seq_length]     ,	self.vocab_size				)

							lowerCAmelCase   =     ids_tensor([self.batch_size, self.seq_length, 4]     ,	self.range_bbox				)
							# Ensure that bbox is legal
							for i in range(bbox.shape[0]				):
								for j in range(bbox.shape[1]				):
									if bbox[i, j, 3] < bbox[i, j, 1]:
										lowerCAmelCase   =     bbox[i, j, 3]
										lowerCAmelCase   =     bbox[i, j, 1]
										lowerCAmelCase   =     t
									if bbox[i, j, 2] < bbox[i, j, 0]:
										lowerCAmelCase   =     bbox[i, j, 2]
										lowerCAmelCase   =     bbox[i, j, 0]
										lowerCAmelCase   =     t

							lowerCAmelCase   =     None
							if self.use_input_mask:
								lowerCAmelCase   =     ids_tensor([self.batch_size, self.seq_length]     ,	vocab_size=2				)

							lowerCAmelCase   =     None
							if self.use_token_type_ids:
								lowerCAmelCase   =     ids_tensor([self.batch_size, self.seq_length]     ,	self.type_vocab_size				)

							lowerCAmelCase   =     None
							lowerCAmelCase   =     None
							if self.use_labels:
								lowerCAmelCase   =     ids_tensor([self.batch_size]     ,	self.type_sequence_label_size				)
								lowerCAmelCase   =     ids_tensor([self.batch_size, self.seq_length]     ,	self.num_labels				)

							lowerCAmelCase   =     self.get_config()

							return config, input_ids, bbox, token_type_ids, input_mask, sequence_labels, token_labels


						def     __snake_case   ( self				)     ->     int:
							return LiltConfig(
							    vocab_size=self.vocab_size     ,	hidden_size=self.hidden_size     ,	num_hidden_layers=self.num_hidden_layers     ,	num_attention_heads=self.num_attention_heads     ,	intermediate_size=self.intermediate_size     ,	hidden_act=self.hidden_act     ,	hidden_dropout_prob=self.hidden_dropout_prob     ,	attention_probs_dropout_prob=self.attention_probs_dropout_prob     ,	max_position_embeddings=self.max_position_embeddings     ,	type_vocab_size=self.type_vocab_size     ,	initializer_range=self.initializer_range     ,	)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	)     ->     int:
							lowerCAmelCase   =     LiltModel(config=A_				)
							model.to(A_				)
							model.eval()
							lowerCAmelCase   =     model(A_     ,	bbox=A_     ,	attention_mask=A_     ,	token_type_ids=A_				)
							lowerCAmelCase   =     model(A_     ,	bbox=A_     ,	token_type_ids=A_				)
							lowerCAmelCase   =     model(A_     ,	bbox=A_				)

							self.parent.assertEqual(result.last_hidden_state.shape     ,	(self.batch_size, self.seq_length, self.hidden_size)				)
							self.parent.assertEqual(result.pooler_output.shape     ,	(self.batch_size, self.hidden_size)				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	)     ->     Union[str, Any]:
							lowerCAmelCase   =     self.num_labels
							lowerCAmelCase   =     LiltForTokenClassification(config=A_				)
							model.to(A_				)
							model.eval()
							lowerCAmelCase   =     model(
							    A_     ,	bbox=A_     ,	attention_mask=A_     ,	token_type_ids=A_     ,	labels=A_				)
							self.parent.assertEqual(result.logits.shape     ,	(self.batch_size, self.seq_length, self.num_labels)				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_     ,	)     ->     Optional[Any]:
							lowerCAmelCase   =     LiltForQuestionAnswering(config=A_				)
							model.to(A_				)
							model.eval()
							lowerCAmelCase   =     model(
							    A_     ,	bbox=A_     ,	attention_mask=A_     ,	token_type_ids=A_     ,	start_positions=A_     ,	end_positions=A_     ,	)
							self.parent.assertEqual(result.start_logits.shape     ,	(self.batch_size, self.seq_length)				)
							self.parent.assertEqual(result.end_logits.shape     ,	(self.batch_size, self.seq_length)				)




						def     __snake_case   ( self				)     ->     Optional[int]:
							lowerCAmelCase   =     self.prepare_config_and_inputs()
							(
							    (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       (
							    lowerCAmelCase
							),       
							)   =     config_and_inputs
							lowerCAmelCase   =     {
							    """input_ids""": input_ids,
							    """bbox""": bbox,
							    """token_type_ids""": token_type_ids,
							    """attention_mask""": input_mask,
							}
							return config, inputs_dict







@require_torch
class __snake_case(				_lowerCAmelCase		,							_lowerCAmelCase		,							_lowerCAmelCase		,							unittest.TestCase ):


						'''simple docstring'''



						UpperCAmelCase							:   int									=		(
						    (
						        LiltModel,
						        LiltForSequenceClassification,
						        LiltForTokenClassification,
						        LiltForQuestionAnswering,
						    )
						    if is_torch_available()
						    else ()
						)
						UpperCAmelCase							:   Union[str, Any]									=		(
						    {
						        "feature-extraction": LiltModel,
						        "question-answering": LiltForQuestionAnswering,
						        "text-classification": LiltForSequenceClassification,
						        "token-classification": LiltForTokenClassification,
						        "zero-shot": LiltForSequenceClassification,
						    }
						    if is_torch_available()
						    else {}
						)
						UpperCAmelCase							:   Optional[int]									=		False
						UpperCAmelCase							:   Dict									=		False


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	A_     ,	A_				)     ->     Union[str, Any]:
							return True


						def     __snake_case   ( self				)     ->     int:
							lowerCAmelCase   =     LiltModelTester(self				)
							lowerCAmelCase   =     ConfigTester(self     ,	config_class=A_     ,	hidden_size=37				)


						def     __snake_case   ( self				)     ->     Any:
							self.config_tester.run_common_tests()


						def     __snake_case   ( self				)     ->     str:
							lowerCAmelCase   =     self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*A_				)


						def     __snake_case   ( self				)     ->     int:
							lowerCAmelCase   =     self.model_tester.prepare_config_and_inputs()
							for type in ["absolute", "relative_key", "relative_key_query"]:
								lowerCAmelCase   =     type
								self.model_tester.create_and_check_model(*A_				)


						def     __snake_case   ( self				)     ->     List[Any]:
							lowerCAmelCase   =     self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_token_classification(*A_				)


						def     __snake_case   ( self				)     ->     str:
							lowerCAmelCase   =     self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_question_answering(*A_				)




						@slow
						def     __snake_case   ( self				)     ->     Union[str, Any]:
							for model_name in LILT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
								lowerCAmelCase   =     LiltModel.from_pretrained(A_				)
								self.assertIsNotNone(A_				)







@require_torch
@slow
class __snake_case(				unittest.TestCase ):


						'''simple docstring'''



						def     __snake_case   ( self				)     ->     Tuple:
							lowerCAmelCase   =     LiltModel.from_pretrained("""SCUT-DLVCLab/lilt-roberta-en-base"""				).to(A_				)

							lowerCAmelCase   =     torch.tensor([[1, 2]]     ,	device=A_				)
							lowerCAmelCase   =     torch.tensor([[[1, 2, 3, 4], [5, 6, 7, 8]]]     ,	device=A_				)

							# forward pass
							with torch.no_grad():
								lowerCAmelCase   =     model(input_ids=A_     ,	bbox=A_				)

							lowerCAmelCase   =     torch.Size([1, 2, 768]				)
							lowerCAmelCase   =     torch.tensor(
							    [[-0.0_6_5_3, 0.0_9_5_0, -0.0_0_6_1], [-0.0_5_4_5, 0.0_9_2_6, -0.0_3_2_4]]     ,	device=A_     ,	)

							self.assertTrue(outputs.last_hidden_state.shape     ,	A_				)
							self.assertTrue(torch.allclose(outputs.last_hidden_state[0, :, :3]     ,	A_     ,	atol=1e-3				)				)