

from __future__ import annotations

import copy
import inspect
import unittest

import numpy as np

from transformers import is_tf_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_tf, slow
from transformers.utils import cached_property

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
		import tensorflow as tf

		from transformers import (
		    TF_LAYOUTLMV3_PRETRAINED_MODEL_ARCHIVE_LIST,
		    TF_MODEL_FOR_MULTIPLE_CHOICE_MAPPING,
		    TF_MODEL_FOR_QUESTION_ANSWERING_MAPPING,
		    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
		    TF_MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING,
		    LayoutLMvaConfig,
		    TFLayoutLMvaForQuestionAnswering,
		    TFLayoutLMvaForSequenceClassification,
		    TFLayoutLMvaForTokenClassification,
		    TFLayoutLMvaModel,
		)

if is_vision_available():
		from PIL import Image

		from transformers import LayoutLMvaImageProcessor







class 					__lowercase  :





							"""simple docstring"""

							def __init__(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase=2    ,						__UpperCAmelCase=3    ,						__UpperCAmelCase=4    ,						__UpperCAmelCase=2    ,						__UpperCAmelCase=7    ,						__UpperCAmelCase=True    ,						__UpperCAmelCase=True    ,						__UpperCAmelCase=True    ,						__UpperCAmelCase=True    ,						__UpperCAmelCase=99    ,						__UpperCAmelCase=36    ,						__UpperCAmelCase=2    ,						__UpperCAmelCase=4    ,						__UpperCAmelCase=37    ,						__UpperCAmelCase="gelu"    ,						__UpperCAmelCase=0.1    ,						__UpperCAmelCase=0.1    ,						__UpperCAmelCase=5_12    ,						__UpperCAmelCase=16    ,						__UpperCAmelCase=2    ,						__UpperCAmelCase=0.0_2    ,						__UpperCAmelCase=6    ,						__UpperCAmelCase=6    ,						__UpperCAmelCase=3    ,						__UpperCAmelCase=4    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=10_00    ,						)		->       Optional[int]:
														A      :					int      		=					parent
														A      :					int      		=					batch_size
														A      :					Optional[Any]      		=					num_channels
														A      :					int      		=					image_size
														A      :					Any      		=					patch_size
														A      :					Any      		=					is_training
														A      :					int      		=					use_input_mask
														A      :					Union[str, Any]      		=					use_token_type_ids
														A      :					Optional[Any]      		=					use_labels
														A      :					str      		=					vocab_size
														A      :					Any      		=					hidden_size
														A      :					str      		=					num_hidden_layers
														A      :					Union[str, Any]      		=					num_attention_heads
														A      :					Tuple      		=					intermediate_size
														A      :					Optional[Any]      		=					hidden_act
														A      :					Union[str, Any]      		=					hidden_dropout_prob
														A      :					Tuple      		=					attention_probs_dropout_prob
														A      :					Optional[Any]      		=					max_position_embeddings
														A      :					List[str]      		=					type_vocab_size
														A      :					Tuple      		=					type_sequence_label_size
														A      :					Optional[Any]      		=					initializer_range
														A      :					Tuple      		=					coordinate_size
														A      :					Tuple      		=					shape_size
														A      :					List[str]      		=					num_labels
														A      :					Optional[Any]      		=					num_choices
														A      :					Optional[int]      		=					scope
														A      :					List[str]      		=					range_bbox

														# LayoutLMv3's sequence length equals the number of text tokens + number of patches + 1 (we add 1 for the CLS token)
														A      :					Optional[Any]      		=					text_seq_length
														A      :					int      		=					(image_size // patch_size) ** 2 + 1
														A      :					Any      		=					self.text_seq_length + self.image_seq_length




							def       snake_case			(					self						)		->       Dict:
														A      :					Optional[int]      		=					ids_tensor([self.batch_size, self.text_seq_length]    ,						self.vocab_size						)

														A      :					Union[str, Any]      		=					ids_tensor([self.batch_size, self.text_seq_length, 4]    ,						self.range_bbox						)
														A      :					Tuple      		=					bbox.numpy()
														# 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]:
																																			A      :					Any      		=					bbox[i, j, 3]
																																			A      :					Tuple      		=					bbox[i, j, 1]
																																			A      :					int      		=					tmp_coordinate
																												if bbox[i, j, 2] < bbox[i, j, 0]:
																																			A      :					str      		=					bbox[i, j, 2]
																																			A      :					int      		=					bbox[i, j, 0]
																																			A      :					Union[str, Any]      		=					tmp_coordinate
														A      :					Tuple      		=					tf.constant(__UpperCAmelCase						)

														A      :					Tuple      		=					floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]						)

														A      :					Union[str, Any]      		=					None
														if self.use_input_mask:
																					A      :					Dict      		=					random_attention_mask([self.batch_size, self.text_seq_length]						)

														A      :					List[Any]      		=					None
														if self.use_token_type_ids:
																					A      :					List[Any]      		=					ids_tensor([self.batch_size, self.text_seq_length]    ,						self.type_vocab_size						)

														A      :					Any      		=					None
														A      :					str      		=					None
														if self.use_labels:
																					A      :					List[Any]      		=					ids_tensor([self.batch_size]    ,						self.type_sequence_label_size						)
																					A      :					List[str]      		=					ids_tensor([self.batch_size, self.text_seq_length]    ,						self.num_labels						)

														A      :					Dict      		=					LayoutLMvaConfig(
														    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    ,						coordinate_size=self.coordinate_size    ,						shape_size=self.shape_size    ,						input_size=self.image_size    ,						patch_size=self.patch_size    ,						)

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




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       Union[str, Any]:
														A      :					int      		=					TFLayoutLMvaModel(config=__UpperCAmelCase						)

														# text + image
														A      :					Dict      		=					model(__UpperCAmelCase    ,						pixel_values=__UpperCAmelCase    ,						training=__UpperCAmelCase						)
														A      :					List[Any]      		=					model(
														    __UpperCAmelCase    ,						bbox=__UpperCAmelCase    ,						pixel_values=__UpperCAmelCase    ,						attention_mask=__UpperCAmelCase    ,						token_type_ids=__UpperCAmelCase    ,						training=__UpperCAmelCase    ,						)
														A      :					Any      		=					model(__UpperCAmelCase    ,						bbox=__UpperCAmelCase    ,						pixel_values=__UpperCAmelCase    ,						training=__UpperCAmelCase						)

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

														# text only
														A      :					Union[str, Any]      		=					model(__UpperCAmelCase    ,						training=__UpperCAmelCase						)
														self.parent.assertEqual(
														    result.last_hidden_state.shape    ,						(self.batch_size, self.text_seq_length, self.hidden_size)						)

														# image only
														A      :					Tuple      		=					model({'''pixel_values''': pixel_values}    ,						training=__UpperCAmelCase						)
														self.parent.assertEqual(
														    result.last_hidden_state.shape    ,						(self.batch_size, self.image_seq_length, self.hidden_size)						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       Union[str, Any]:
														A      :					str      		=					self.num_labels
														A      :					Any      		=					TFLayoutLMvaForSequenceClassification(config=__UpperCAmelCase						)
														A      :					str      		=					model(
														    __UpperCAmelCase    ,						bbox=__UpperCAmelCase    ,						pixel_values=__UpperCAmelCase    ,						attention_mask=__UpperCAmelCase    ,						token_type_ids=__UpperCAmelCase    ,						labels=__UpperCAmelCase    ,						training=__UpperCAmelCase    ,						)
														self.parent.assertEqual(result.logits.shape    ,						(self.batch_size, self.num_labels)						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       Tuple:
														A      :					Optional[int]      		=					self.num_labels
														A      :					int      		=					TFLayoutLMvaForTokenClassification(config=__UpperCAmelCase						)
														A      :					Optional[Any]      		=					model(
														    __UpperCAmelCase    ,						bbox=__UpperCAmelCase    ,						pixel_values=__UpperCAmelCase    ,						attention_mask=__UpperCAmelCase    ,						token_type_ids=__UpperCAmelCase    ,						labels=__UpperCAmelCase    ,						training=__UpperCAmelCase    ,						)
														self.parent.assertEqual(result.logits.shape    ,						(self.batch_size, self.text_seq_length, self.num_labels)						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       Optional[Any]:
														A      :					Tuple      		=					2
														A      :					List[Any]      		=					TFLayoutLMvaForQuestionAnswering(config=__UpperCAmelCase						)
														A      :					str      		=					model(
														    __UpperCAmelCase    ,						bbox=__UpperCAmelCase    ,						pixel_values=__UpperCAmelCase    ,						attention_mask=__UpperCAmelCase    ,						token_type_ids=__UpperCAmelCase    ,						start_positions=__UpperCAmelCase    ,						end_positions=__UpperCAmelCase    ,						training=__UpperCAmelCase    ,						)
														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[Any]:
														A      :					Tuple      		=					self.prepare_config_and_inputs()
														((A)		,		(A)		,		(A)		,		(A)		,		(A)		,		(A)		,		(A)		,		(A))      :					int      		=					config_and_inputs
														A      :					Union[str, Any]      		=					{
														    '''input_ids''': input_ids,
														    '''bbox''': bbox,
														    '''pixel_values''': pixel_values,
														    '''token_type_ids''': token_type_ids,
														    '''attention_mask''': input_mask,
														}
														return config, inputs_dict







@require_tf
class 					__lowercase  (						_SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE		,       unittest.TestCase				):





							"""simple docstring"""

							UpperCAmelCase_						:		Optional[int]								=	(
							    (
							        TFLayoutLMvaModel,
							        TFLayoutLMvaForQuestionAnswering,
							        TFLayoutLMvaForSequenceClassification,
							        TFLayoutLMvaForTokenClassification,
							    )
							    if is_tf_available()
							    else ()
							)
							UpperCAmelCase_						:		List[str]								=	(
							    {'''document-question-answering''': TFLayoutLMvaForQuestionAnswering, '''feature-extraction''': TFLayoutLMvaModel}
							    if is_tf_available()
							    else {}
							)

							UpperCAmelCase_						:		Dict								=	False
							UpperCAmelCase_						:		int								=	False
							UpperCAmelCase_						:		str								=	False




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       Union[str, Any]:
														return True




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase=False						)		->       dict:
														A      :					Dict      		=					copy.deepcopy(__UpperCAmelCase						)

														if model_class in get_values(__UpperCAmelCase						):
																					A      :					Any      		=					{
																					    k: tf.tile(tf.expand_dims(__UpperCAmelCase    ,						1						)    ,						(1, self.model_tester.num_choices) + (1,) * (v.ndim - 1)						)
																					    if isinstance(__UpperCAmelCase    ,						tf.Tensor						) and v.ndim > 0
																					    else v
																					    for k, v in inputs_dict.items()
																					}

														if return_labels:
																					if model_class in get_values(__UpperCAmelCase						):
																												A      :					Optional[Any]      		=					tf.ones(self.model_tester.batch_size    ,						dtype=tf.intaa						)
																					elif model_class in get_values(__UpperCAmelCase						):
																												A      :					Optional[Any]      		=					tf.zeros(self.model_tester.batch_size    ,						dtype=tf.intaa						)
																												A      :					Optional[Any]      		=					tf.zeros(self.model_tester.batch_size    ,						dtype=tf.intaa						)
																					elif model_class in get_values(__UpperCAmelCase						):
																												A      :					List[str]      		=					tf.zeros(self.model_tester.batch_size    ,						dtype=tf.intaa						)
																					elif model_class in get_values(__UpperCAmelCase						):
																												A      :					Optional[int]      		=					tf.zeros(
																												    (self.model_tester.batch_size, self.model_tester.text_seq_length)    ,						dtype=tf.intaa						)

														return inputs_dict




							def       snake_case			(					self						)		->       Optional[int]:
														A      :					str      		=					TFLayoutLMvaModelTester(self						)
														A      :					Optional[int]      		=					ConfigTester(self    ,						config_class=__UpperCAmelCase    ,						hidden_size=37						)




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




							def       snake_case			(					self						)		->       Union[str, Any]:
														A		,		A      :					int      		=					self.model_tester.prepare_config_and_inputs_for_common()
														for model_class in self.all_model_classes:
																					A      :					int      		=					model_class(__UpperCAmelCase						)
																					if getattr(__UpperCAmelCase    ,						'''hf_compute_loss'''    ,						__UpperCAmelCase						):
																												# The number of elements in the loss should be the same as the number of elements in the label
																												A      :					Optional[Any]      		=					self._prepare_for_class(inputs_dict.copy()    ,						__UpperCAmelCase    ,						return_labels=__UpperCAmelCase						)
																												A      :					int      		=					prepared_for_class[
																												    sorted(prepared_for_class.keys() - inputs_dict.keys()    ,						reverse=__UpperCAmelCase						)[0]
																												]
																												A      :					Optional[int]      		=					added_label.shape.as_list()[:1]

																												# Test that model correctly compute the loss with kwargs
																												A      :					Optional[Any]      		=					self._prepare_for_class(inputs_dict.copy()    ,						__UpperCAmelCase    ,						return_labels=__UpperCAmelCase						)
																												A      :					Optional[int]      		=					prepared_for_class.pop('''input_ids'''						)

																												A      :					Any      		=					model(__UpperCAmelCase    ,						**__UpperCAmelCase						)[0]
																												self.assertTrue(loss.shape.as_list() == expected_loss_size or loss.shape.as_list() == [1]						)

																												# Test that model correctly compute the loss when we mask some positions
																												A      :					Dict      		=					self._prepare_for_class(inputs_dict.copy()    ,						__UpperCAmelCase    ,						return_labels=__UpperCAmelCase						)
																												A      :					Optional[Any]      		=					prepared_for_class.pop('''input_ids'''						)
																												if "labels" in prepared_for_class:
																																			A      :					Optional[int]      		=					prepared_for_class['''labels'''].numpy()
																																			if len(labels.shape						) > 1 and labels.shape[1] != 1:
																																										A      :					List[Any]      		=					-1_00
																																										A      :					Any      		=					tf.convert_to_tensor(__UpperCAmelCase						)
																																										A      :					Any      		=					model(__UpperCAmelCase    ,						**__UpperCAmelCase						)[0]
																																										self.assertTrue(loss.shape.as_list() == expected_loss_size or loss.shape.as_list() == [1]						)
																																										self.assertTrue(not np.any(np.isnan(loss.numpy()						)						)						)

                # Test that model correctly compute the loss with a dict
																												A      :					Any      		=					self._prepare_for_class(inputs_dict.copy()    ,						__UpperCAmelCase    ,						return_labels=__UpperCAmelCase						)
																												A      :					List[str]      		=					model(__UpperCAmelCase						)[0]
																												self.assertTrue(loss.shape.as_list() == expected_loss_size or loss.shape.as_list() == [1]						)

																												# Test that model correctly compute the loss with a tuple
																												A      :					str      		=					self._prepare_for_class(inputs_dict.copy()    ,						__UpperCAmelCase    ,						return_labels=__UpperCAmelCase						)

																												# Get keys that were added with the _prepare_for_class function
																												A      :					Optional[Any]      		=					prepared_for_class.keys() - inputs_dict.keys()
																												A      :					List[str]      		=					inspect.signature(model.call						).parameters
																												A      :					List[Any]      		=					list(signature.keys()						)

																												# Create a dictionary holding the location of the tensors in the tuple
																												A      :					Dict      		=					{0: '''input_ids'''}
																												for label_key in label_keys:
																																			A      :					Optional[int]      		=					signature_names.index(__UpperCAmelCase						)
																																			A      :					int      		=					label_key
																												A      :					Optional[int]      		=					sorted(tuple_index_mapping.items()						)
																												# Initialize a list with their default values, update the values and convert to a tuple
																												A      :					Tuple      		=					[]

																												for name in signature_names:
																																			if name != "kwargs":
																																										list_input.append(signature[name].default						)

																												for index, value in sorted_tuple_index_mapping:
																																			A      :					int      		=					prepared_for_class[value]

																												A      :					Union[str, Any]      		=					tuple(__UpperCAmelCase						)

																												# Send to model
																												A      :					Any      		=					model(tuple_input[:-1]						)[0]

																												self.assertTrue(loss.shape.as_list() == expected_loss_size or loss.shape.as_list() == [1]						)




							def       snake_case			(					self						)		->       Optional[int]:
														(
														    (
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		
														)      :					Union[str, Any]      		=					self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_model(__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)




							def       snake_case			(					self						)		->       List[Any]:
														(
														    (
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		
														)      :					List[Any]      		=					self.model_tester.prepare_config_and_inputs()
														for type in ["absolute", "relative_key", "relative_key_query"]:
																					A      :					Dict      		=					type
																					self.model_tester.create_and_check_model(__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)




							def       snake_case			(					self						)		->       Any:
														(
														    (
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		
														)      :					Any      		=					self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_for_sequence_classification(
														    __UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)




							def       snake_case			(					self						)		->       Any:
														(
														    (
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		
														)      :					Union[str, Any]      		=					self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_for_token_classification(
														    __UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)




							def       snake_case			(					self						)		->       Optional[int]:
														(
														    (
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		(
														    A
														)		,		
														)      :					List[Any]      		=					self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_for_question_answering(
														    __UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)






							@slow
							def       snake_case			(					self						)		->       Optional[Any]:
														for model_name in TF_LAYOUTLMV3_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																					A      :					Tuple      		=					TFLayoutLMvaModel.from_pretrained(__UpperCAmelCase						)
																					self.assertIsNotNone(__UpperCAmelCase						)






def        snake_case__		(    ):
							A      :					Dict      		=					Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''		)
							return image







@require_tf
class 					__lowercase  (						unittest.TestCase				):





							"""simple docstring"""

							@cached_property
							def       snake_case			(					self						)		->       Tuple:
														return LayoutLMvaImageProcessor(apply_ocr=__UpperCAmelCase						) if is_vision_available() else None






							@slow
							def       snake_case			(					self						)		->       Tuple:
														A      :					Dict      		=					TFLayoutLMvaModel.from_pretrained('''microsoft/layoutlmv3-base'''						)

														A      :					List[str]      		=					self.default_image_processor
														A      :					str      		=					prepare_img()
														A      :					List[str]      		=					image_processor(images=__UpperCAmelCase    ,						return_tensors='''tf'''						).pixel_values

														A      :					Any      		=					tf.constant([[1, 2]]						)
														A      :					int      		=					tf.expand_dims(tf.constant([[1, 2, 3, 4], [5, 6, 7, 8]]						)    ,						axis=0						)

														# forward pass
														A      :					Tuple      		=					model(input_ids=__UpperCAmelCase    ,						bbox=__UpperCAmelCase    ,						pixel_values=__UpperCAmelCase    ,						training=__UpperCAmelCase						)

														# verify the logits
														A      :					List[str]      		=					(1, 1_99, 7_68)
														self.assertEqual(outputs.last_hidden_state.shape    ,						__UpperCAmelCase						)

														A      :					str      		=					tf.constant(
														    [[-0.0_5_2_9, 0.3_6_1_8, 0.1_6_3_2], [-0.1_5_8_7, -0.1_6_6_7, -0.0_4_0_0], [-0.1_5_5_7, -0.1_6_7_1, -0.0_5_0_5]]						)

														self.assertTrue(np.allclose(outputs.last_hidden_state[0, :3, :3]    ,						__UpperCAmelCase    ,						atol=1E-4						)						)









import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionInstructPixaPixPipeline,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.utils import floats_tensor, load_image, 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_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







class 					__lowercase  (						_SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE		,       unittest.TestCase				):





							"""simple docstring"""

							UpperCAmelCase_						:		Optional[int]								=	StableDiffusionInstructPixaPixPipeline
							UpperCAmelCase_						:		str								=	TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width''', '''cross_attention_kwargs'''}
							UpperCAmelCase_						:		List[Any]								=	TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
							UpperCAmelCase_						:		Optional[int]								=	IMAGE_TO_IMAGE_IMAGE_PARAMS
							UpperCAmelCase_						:		Any								=	IMAGE_TO_IMAGE_IMAGE_PARAMS




							def       snake_case			(					self						)		->       List[Any]:
														torch.manual_seed(0						)
														A      :					Tuple      		=					UNetaDConditionModel(
														    block_out_channels=(32, 64)    ,						layers_per_block=2    ,						sample_size=32    ,						in_channels=8    ,						out_channels=4    ,						down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')    ,						up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')    ,						cross_attention_dim=32    ,						)
														A      :					Any      		=					PNDMScheduler(skip_prk_steps=__UpperCAmelCase						)
														torch.manual_seed(0						)
														A      :					Dict      		=					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      :					str      		=					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=10_00    ,						)
														A      :					Dict      		=					CLIPTextModel(__UpperCAmelCase						)
														A      :					Dict      		=					CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

														A      :					int      		=					{
														    '''unet''': unet,
														    '''scheduler''': scheduler,
														    '''vae''': vae,
														    '''text_encoder''': text_encoder,
														    '''tokenizer''': tokenizer,
														    '''safety_checker''': None,
														    '''feature_extractor''': None,
														}
														return components




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase=0						)		->       Dict:
														A      :					Any      		=					floats_tensor((1, 3, 32, 32)    ,						rng=random.Random(__UpperCAmelCase						)						).to(__UpperCAmelCase						)
														A      :					Dict      		=					image.cpu().permute(0    ,						2    ,						3    ,						1						)[0]
														A      :					Optional[Any]      		=					Image.fromarray(np.uinta(__UpperCAmelCase						)						).convert('''RGB'''						)
														if str(__UpperCAmelCase						).startswith('''mps'''						):
																					A      :					str      		=					torch.manual_seed(__UpperCAmelCase						)
														else:
																					A      :					Optional[Any]      		=					torch.Generator(device=__UpperCAmelCase						).manual_seed(__UpperCAmelCase						)
														A      :					List[Any]      		=					{
														    '''prompt''': '''A painting of a squirrel eating a burger''',
														    '''image''': image,
														    '''generator''': generator,
														    '''num_inference_steps''': 2,
														    '''guidance_scale''': 6.0,
														    '''image_guidance_scale''': 1,
														    '''output_type''': '''numpy''',
														}
														return inputs




							def       snake_case			(					self						)		->       Union[str, Any]:
														A      :					Optional[Any]      		=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
														A      :					Dict      		=					self.get_dummy_components()
														A      :					str      		=					StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase						)
														A      :					Any      		=					sd_pipe.to(__UpperCAmelCase						)
														sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase						)

														A      :					Optional[Any]      		=					self.get_dummy_inputs(__UpperCAmelCase						)
														A      :					int      		=					sd_pipe(**__UpperCAmelCase						).images
														A      :					List[str]      		=					image[0, -3:, -3:, -1]
														assert image.shape == (1, 32, 32, 3)
														A      :					Optional[Any]      		=					np.array([0.7_5_2_6, 0.3_7_5_0, 0.4_5_4_7, 0.6_1_1_7, 0.5_8_6_6, 0.5_0_1_6, 0.4_3_2_7, 0.5_6_4_2, 0.4_8_1_5]						)

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




							def       snake_case			(					self						)		->       int:
														A      :					Any      		=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
														A      :					Optional[Any]      		=					self.get_dummy_components()
														A      :					Any      		=					StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase						)
														A      :					str      		=					sd_pipe.to(__UpperCAmelCase						)
														sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase						)

														A      :					str      		=					self.get_dummy_inputs(__UpperCAmelCase						)
														A      :					Optional[int]      		=					'''french fries'''
														A      :					Any      		=					sd_pipe(**__UpperCAmelCase    ,						negative_prompt=__UpperCAmelCase						)
														A      :					Optional[int]      		=					output.images
														A      :					str      		=					image[0, -3:, -3:, -1]

														assert image.shape == (1, 32, 32, 3)
														A      :					Any      		=					np.array([0.7_5_1_1, 0.3_6_4_2, 0.4_5_5_3, 0.6_2_3_6, 0.5_7_9_7, 0.5_0_1_3, 0.4_3_4_3, 0.5_6_1_1, 0.4_8_3_1]						)

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




							def       snake_case			(					self						)		->       Tuple:
														A      :					List[Any]      		=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
														A      :					List[Any]      		=					self.get_dummy_components()
														A      :					Optional[int]      		=					StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase						)
														A      :					Tuple      		=					sd_pipe.to(__UpperCAmelCase						)
														sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase						)

														A      :					str      		=					self.get_dummy_inputs(__UpperCAmelCase						)
														A      :					int      		=					[inputs['''prompt''']] * 2

														A      :					List[str]      		=					np.array(inputs['''image''']						).astype(np.floataa						) / 2_5_5.0
														A      :					Any      		=					torch.from_numpy(__UpperCAmelCase						).unsqueeze(0						).to(__UpperCAmelCase						)
														A      :					Union[str, Any]      		=					image / 2 + 0.5
														A      :					str      		=					image.permute(0    ,						3    ,						1    ,						2						)
														A      :					List[str]      		=					image.repeat(2    ,						1    ,						1    ,						1						)

														A      :					Union[str, Any]      		=					sd_pipe(**__UpperCAmelCase						).images
														A      :					int      		=					image[-1, -3:, -3:, -1]

														assert image.shape == (2, 32, 32, 3)
														A      :					Any      		=					np.array([0.5_8_1_2, 0.5_7_4_8, 0.5_2_2_2, 0.5_9_0_8, 0.5_6_9_5, 0.7_1_7_4, 0.6_8_0_4, 0.5_5_2_3, 0.5_5_7_9]						)

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




							def       snake_case			(					self						)		->       Dict:
														A      :					int      		=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
														A      :					List[Any]      		=					self.get_dummy_components()
														A      :					Union[str, Any]      		=					EulerAncestralDiscreteScheduler(
														    beta_start=0.0_0_0_8_5    ,						beta_end=0.0_1_2    ,						beta_schedule='''scaled_linear'''						)
														A      :					int      		=					StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase						)
														A      :					Tuple      		=					sd_pipe.to(__UpperCAmelCase						)
														sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase						)

														A      :					Tuple      		=					self.get_dummy_inputs(__UpperCAmelCase						)
														A      :					Optional[Any]      		=					sd_pipe(**__UpperCAmelCase						).images
														A      :					Dict      		=					image[0, -3:, -3:, -1]

														A      :					Optional[int]      		=					[round(__UpperCAmelCase    ,						4						) for x in image_slice.flatten().tolist()]
														print(''','''.join([str(__UpperCAmelCase						) for x in slice]						)						)

														assert image.shape == (1, 32, 32, 3)
														A      :					Dict      		=					np.array([0.7_4_1_7, 0.3_8_4_2, 0.4_7_3_2, 0.5_7_7_6, 0.5_8_9_1, 0.5_1_3_9, 0.4_0_5_2, 0.5_6_7_3, 0.4_9_8_6]						)

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




							def       snake_case			(					self						)		->       Union[str, Any]:
														super().test_inference_batch_single_identical(expected_max_diff=3E-3						)






							def       snake_case			(					self						)		->       Union[str, Any]:
														A      :					int      		=					self.get_dummy_components()
														A      :					int      		=					StableDiffusionInstructPixaPixPipeline(**__UpperCAmelCase						)
														A      :					Any      		=					VaeImageProcessor(do_resize=__UpperCAmelCase    ,						do_normalize=__UpperCAmelCase						)
														A      :					Optional[Any]      		=					pipe.to(__UpperCAmelCase						)
														pipe.set_progress_bar_config(disable=__UpperCAmelCase						)

														A      :					Union[str, Any]      		=					pipe(**self.get_dummy_inputs_by_type(__UpperCAmelCase    ,						input_image_type='''pt'''						)						)[0]

														A      :					Optional[Any]      		=					components['''vae''']
														A      :					Dict      		=					self.get_dummy_inputs_by_type(__UpperCAmelCase    ,						input_image_type='''pt'''						)

														for image_param in self.image_latents_params:
																					if image_param in inputs.keys():
																												A      :					Optional[Any]      		=					vae.encode(inputs[image_param]						).latent_dist.mode()

														A      :					List[str]      		=					pipe(**__UpperCAmelCase						)[0]

														A      :					Any      		=					np.abs(out - out_latents_inputs						).max()
														self.assertLess(__UpperCAmelCase    ,						1E-4    ,						'''passing latents as image input generate different result from passing image'''						)







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





							"""simple docstring"""

							def       snake_case			(					self						)		->       str:
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()




							def       snake_case			(					self    ,						__UpperCAmelCase=0						)		->       Tuple:
														A      :					List[str]      		=					torch.manual_seed(__UpperCAmelCase						)
														A      :					List[str]      		=					load_image(
														    '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg'''						)
														A      :					Dict      		=					{
														    '''prompt''': '''turn him into a cyborg''',
														    '''image''': image,
														    '''generator''': generator,
														    '''num_inference_steps''': 3,
														    '''guidance_scale''': 7.5,
														    '''image_guidance_scale''': 1.0,
														    '''output_type''': '''numpy''',
														}
														return inputs




							def       snake_case			(					self						)		->       Optional[int]:
														A      :					Union[str, Any]      		=					StableDiffusionInstructPixaPixPipeline.from_pretrained(
														    '''timbrooks/instruct-pix2pix'''    ,						safety_checker=__UpperCAmelCase						)
														pipe.to(__UpperCAmelCase						)
														pipe.set_progress_bar_config(disable=__UpperCAmelCase						)
														pipe.enable_attention_slicing()

														A      :					Optional[int]      		=					self.get_inputs()
														A      :					Any      		=					pipe(**__UpperCAmelCase						).images
														A      :					Union[str, Any]      		=					image[0, -3:, -3:, -1].flatten()

														assert image.shape == (1, 5_12, 5_12, 3)
														A      :					Dict      		=					np.array([0.5_9_0_2, 0.6_0_1_5, 0.6_0_2_7, 0.5_9_8_3, 0.6_0_9_2, 0.6_0_6_1, 0.5_7_6_5, 0.5_7_8_5, 0.5_5_5_5]						)

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




							def       snake_case			(					self						)		->       Dict:
														A      :					str      		=					StableDiffusionInstructPixaPixPipeline.from_pretrained(
														    '''timbrooks/instruct-pix2pix'''    ,						safety_checker=__UpperCAmelCase						)
														A      :					List[str]      		=					LMSDiscreteScheduler.from_config(pipe.scheduler.config						)
														pipe.to(__UpperCAmelCase						)
														pipe.set_progress_bar_config(disable=__UpperCAmelCase						)
														pipe.enable_attention_slicing()

														A      :					Union[str, Any]      		=					self.get_inputs()
														A      :					Union[str, Any]      		=					pipe(**__UpperCAmelCase						).images
														A      :					str      		=					image[0, -3:, -3:, -1].flatten()

														assert image.shape == (1, 5_12, 5_12, 3)
														A      :					Dict      		=					np.array([0.6_5_7_8, 0.6_8_1_7, 0.6_9_7_2, 0.6_7_6_1, 0.6_8_5_6, 0.6_9_1_6, 0.6_4_2_8, 0.6_5_1_6, 0.6_3_0_1]						)

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




							def       snake_case			(					self						)		->       List[Any]:
														A      :					List[Any]      		=					StableDiffusionInstructPixaPixPipeline.from_pretrained(
														    '''timbrooks/instruct-pix2pix'''    ,						safety_checker=__UpperCAmelCase						)
														A      :					List[str]      		=					DDIMScheduler.from_config(pipe.scheduler.config						)
														pipe.to(__UpperCAmelCase						)
														pipe.set_progress_bar_config(disable=__UpperCAmelCase						)
														pipe.enable_attention_slicing()

														A      :					str      		=					self.get_inputs()
														A      :					List[Any]      		=					pipe(**__UpperCAmelCase						).images
														A      :					str      		=					image[0, -3:, -3:, -1].flatten()

														assert image.shape == (1, 5_12, 5_12, 3)
														A      :					Any      		=					np.array([0.3_8_2_8, 0.3_8_3_4, 0.3_8_1_8, 0.3_7_9_2, 0.3_8_6_5, 0.3_7_5_2, 0.3_7_9_2, 0.3_8_4_7, 0.3_7_5_3]						)

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




							def       snake_case			(					self						)		->       Optional[Any]:
														A      :					Optional[Any]      		=					0

														def callback_fn(__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						) -> None:
																					A      :					List[Any]      		=					True
																					nonlocal number_of_steps
																					number_of_steps += 1
																					if step == 1:
																												A      :					Optional[Any]      		=					latents.detach().cpu().numpy()
																												assert latents.shape == (1, 4, 64, 64)
																												A      :					Union[str, Any]      		=					latents[0, -3:, -3:, -1]
																												A      :					List[str]      		=					np.array([-0.2_4_6_3, -0.4_6_4_4, -0.9_7_5_6, 1.5_1_7_6, 1.4_4_1_4, 0.7_8_6_6, 0.9_8_9_7, 0.8_5_2_1, 0.7_9_8_3]						)

																												assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5E-2
																					elif step == 2:
																												A      :					int      		=					latents.detach().cpu().numpy()
																												assert latents.shape == (1, 4, 64, 64)
																												A      :					Optional[int]      		=					latents[0, -3:, -3:, -1]
																												A      :					str      		=					np.array([-0.2_6_4_4, -0.4_6_2_6, -0.9_6_5_3, 1.5_1_7_6, 1.4_5_5_1, 0.7_6_8_6, 0.9_8_0_5, 0.8_4_5_2, 0.8_1_1_5]						)

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

														A      :					Optional[Any]      		=					False

														A      :					str      		=					StableDiffusionInstructPixaPixPipeline.from_pretrained(
														    '''timbrooks/instruct-pix2pix'''    ,						safety_checker=__UpperCAmelCase    ,						torch_dtype=torch.floataa						)
														A      :					List[str]      		=					pipe.to(__UpperCAmelCase						)
														pipe.set_progress_bar_config(disable=__UpperCAmelCase						)
														pipe.enable_attention_slicing()

														A      :					List[Any]      		=					self.get_inputs()
														pipe(**__UpperCAmelCase    ,						callback=__UpperCAmelCase    ,						callback_steps=1						)
														assert callback_fn.has_been_called
														assert number_of_steps == 3




							def       snake_case			(					self						)		->       Tuple:
														torch.cuda.empty_cache()
														torch.cuda.reset_max_memory_allocated()
														torch.cuda.reset_peak_memory_stats()

														A      :					Optional[Any]      		=					StableDiffusionInstructPixaPixPipeline.from_pretrained(
														    '''timbrooks/instruct-pix2pix'''    ,						safety_checker=__UpperCAmelCase    ,						torch_dtype=torch.floataa						)
														A      :					Dict      		=					pipe.to(__UpperCAmelCase						)
														pipe.set_progress_bar_config(disable=__UpperCAmelCase						)
														pipe.enable_attention_slicing(1						)
														pipe.enable_sequential_cpu_offload()

														A      :					Optional[Any]      		=					self.get_inputs()
														A      :					str      		=					pipe(**__UpperCAmelCase						)

														A      :					Dict      		=					torch.cuda.max_memory_allocated()
														# make sure that less than 2.2 GB is allocated
														assert mem_bytes < 2.2 * 10**9






							def       snake_case			(					self						)		->       int:
														A      :					Union[str, Any]      		=					self.get_inputs()
														# resize to resolution that is divisible by 8 but not 16 or 32
														A      :					Union[str, Any]      		=					inputs['''image'''].resize((5_04, 5_04)						)

														A      :					str      		=					'''timbrooks/instruct-pix2pix'''
														A      :					Tuple      		=					StableDiffusionInstructPixaPixPipeline.from_pretrained(
														    __UpperCAmelCase    ,						safety_checker=__UpperCAmelCase    ,						)
														pipe.to(__UpperCAmelCase						)
														pipe.set_progress_bar_config(disable=__UpperCAmelCase						)
														pipe.enable_attention_slicing()

														A      :					Tuple      		=					pipe(**__UpperCAmelCase						)
														A      :					List[str]      		=					output.images[0]

														A      :					List[Any]      		=					image[2_55:2_58, 3_83:3_86, -1]

														assert image.shape == (5_04, 5_04, 3)
														A      :					Any      		=					np.array([0.2_7_2_6, 0.2_5_2_9, 0.2_6_6_4, 0.2_6_5_5, 0.2_6_4_1, 0.2_6_4_2, 0.2_5_9_1, 0.2_6_4_9, 0.2_5_9_0]						)

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






