



import unittest

import numpy as np
import torch
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


class 						lowerCamelCase__			(		_a					,       unittest.TestCase ):
						_lowerCAmelCase             =			KandinskyPriorPipeline
						_lowerCAmelCase             =			['''prompt''']
						_lowerCAmelCase             =			['''prompt''', '''negative_prompt''']
						_lowerCAmelCase             =			[
						    '''num_images_per_prompt''',
						    '''generator''',
						    '''num_inference_steps''',
						    '''latents''',
						    '''negative_prompt''',
						    '''guidance_scale''',
						    '''output_type''',
						    '''return_dict''',
						]
						_lowerCAmelCase             =			False


						@property
						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													return 3_2


						@property
						def 	_lowerCamelCase      (			self    :	Optional[Any]      ):
													return 3_2


						@property
						def 	_lowerCamelCase      (			self    :	Tuple      ):
													return self.time_input_dim


						@property
						def 	_lowerCamelCase      (			self    :	Union[str, Any]      ):
													return self.time_input_dim * 4


						@property
						def 	_lowerCamelCase      (			self    :	str      ):
													return 1_0_0


						@property
						def 	_lowerCamelCase      (			self    :	Tuple      ):
													a__:							Union[str, Any]									=CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      )
													return tokenizer


						@property
						def 	_lowerCamelCase      (			self    :	Optional[Any]      ):
													torch.manual_seed(0      )
													a__:							List[Any]									=CLIPTextConfig(
													    bos_token_id=0   ,							eos_token_id=2   ,							hidden_size=self.text_embedder_hidden_size   ,							projection_dim=self.text_embedder_hidden_size   ,							intermediate_size=3_7   ,							layer_norm_eps=1e-05   ,							num_attention_heads=4   ,							num_hidden_layers=5   ,							pad_token_id=1   ,							vocab_size=1_0_0_0   ,							)
													return CLIPTextModelWithProjection(_a      )


						@property
						def 	_lowerCamelCase      (			self    :	str      ):
													torch.manual_seed(0      )

													a__:							Dict									={
													    "num_attention_heads": 2,
													    "attention_head_dim": 1_2,
													    "embedding_dim": self.text_embedder_hidden_size,
													    "num_layers": 1,
													}

													a__:							Tuple									=PriorTransformer(**_a      )
													# clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
													a__:							Union[str, Any]									=nn.Parameter(torch.ones(model.clip_std.shape      )      )
													return model


						@property
						def 	_lowerCamelCase      (			self    :	str      ):
													torch.manual_seed(0      )
													a__:							Any									=CLIPVisionConfig(
													    hidden_size=self.text_embedder_hidden_size   ,							image_size=2_2_4   ,							projection_dim=self.text_embedder_hidden_size   ,							intermediate_size=3_7   ,							num_attention_heads=4   ,							num_channels=3   ,							num_hidden_layers=5   ,							patch_size=1_4   ,							)

													a__:							Any									=CLIPVisionModelWithProjection(_a      )
													return model


						@property
						def 	_lowerCamelCase      (			self    :	str      ):
													a__:							Union[str, Any]									=CLIPImageProcessor(
													    crop_size=2_2_4   ,							do_center_crop=_a   ,							do_normalize=_a   ,							do_resize=_a   ,							image_mean=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]   ,							image_std=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]   ,							resample=3   ,							size=2_2_4   ,							)

													return image_processor


						def 	_lowerCamelCase      (			self    :	Union[str, Any]      ):
													a__:							Dict									=self.dummy_prior
													a__:							Any									=self.dummy_image_encoder
													a__:							int									=self.dummy_text_encoder
													a__:							str									=self.dummy_tokenizer
													a__:							List[str]									=self.dummy_image_processor

													a__:							int									=UnCLIPScheduler(
													    variance_type="fixed_small_log"   ,							prediction_type="sample"   ,							num_train_timesteps=1_0_0_0   ,							clip_sample=_a   ,							clip_sample_range=1_0.0   ,							)

													a__:							Any									={
													    "prior": prior,
													    "image_encoder": image_encoder,
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "scheduler": scheduler,
													    "image_processor": image_processor,
													}

													return components


						def 	_lowerCamelCase      (			self    :	Tuple   ,							_a    :	Optional[int]   ,							_a    :	Union[str, Any]=0      ):
													if str(_a      ).startswith("mps"      ):
																				a__:							List[Any]									=torch.manual_seed(_a      )
													else:
																				a__:							List[str]									=torch.Generator(device=_a      ).manual_seed(_a      )
													a__:							str									={
													    "prompt": "horse",
													    "generator": generator,
													    "guidance_scale": 4.0,
													    "num_inference_steps": 2,
													    "output_type": "np",
													}
													return inputs


						def 	_lowerCamelCase      (			self    :	str      ):
													a__:							Any									="cpu"

													a__:							str									=self.get_dummy_components()

													a__:							Union[str, Any]									=self.pipeline_class(**_a      )
													a__:							int									=pipe.to(_a      )

													pipe.set_progress_bar_config(disable=_a      )

													a__:							Union[str, Any]									=pipe(**self.get_dummy_inputs(_a      )      )
													a__:							Optional[int]									=output.image_embeds

													a__:							str									=pipe(
													    **self.get_dummy_inputs(_a      )   ,							return_dict=_a   ,							)[0]

													a__:							List[str]									=image[0, -1_0:]
													a__:							List[Any]									=image_from_tuple[0, -1_0:]

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

													a__:							Optional[int]									=np.array(
													    [-0.0_5_3_2, 1.7_1_2_0, 0.3_6_5_6, -1.0_8_5_2, -0.8_9_4_6, -1.1_7_5_6, 0.4_3_4_8, 0.2_4_8_2, 0.5_1_4_6, -0.1_1_5_6]      )

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


						@skip_mps
						def 	_lowerCamelCase      (			self    :	List[Any]      ):
													a__:							str									=torch_device == "cpu"
													a__:							Tuple									=True
													a__:							Optional[int]									=False

													self._test_inference_batch_single_identical(
													    test_max_difference=_a   ,							relax_max_difference=_a   ,							test_mean_pixel_difference=_a   ,							)







						@skip_mps
						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													a__:							str									=torch_device == "cpu"
													a__:							str									=False

													self._test_attention_slicing_forward_pass(
													    test_max_difference=_a   ,							test_mean_pixel_difference=_a   ,							)










import inspect
import unittest

from transformers import DPTConfig
from transformers.file_utils import is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch
		from torch import nn

		from transformers import MODEL_MAPPING, DPTForDepthEstimation, DPTForSemanticSegmentation, DPTModel
		from transformers.models.dpt.modeling_dpt import DPT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
		from PIL import Image

		from transformers import DPTImageProcessor


class 						lowerCamelCase__			:


						def __init__(			self    :	List[str]   ,							_a    :	Any   ,							_a    :	str=2   ,							_a    :	List[str]=3_2   ,							_a    :	str=1_6   ,							_a    :	Tuple=3   ,							_a    :	Any=True   ,							_a    :	Optional[Any]=True   ,							_a    :	Optional[Any]=3_2   ,							_a    :	Optional[Any]=4   ,							_a    :	str=[0, 1, 2, 3]   ,							_a    :	Any=4   ,							_a    :	List[Any]=3_7   ,							_a    :	Optional[Any]="gelu"   ,							_a    :	str=0.1   ,							_a    :	Optional[Any]=0.1   ,							_a    :	Optional[Any]=0.0_2   ,							_a    :	Optional[int]=3   ,							_a    :	List[str]=False   ,							_a    :	Dict=None   ,							):
													a__:							List[Any]									=parent
													a__:							List[str]									=batch_size
													a__:							Optional[Any]									=image_size
													a__:							Optional[Any]									=patch_size
													a__:							Dict									=num_channels
													a__:							Optional[int]									=is_training
													a__:							List[Any]									=use_labels
													a__:							List[str]									=hidden_size
													a__:							int									=num_hidden_layers
													a__:							List[Any]									=backbone_out_indices
													a__:							int									=num_attention_heads
													a__:							List[Any]									=intermediate_size
													a__:							int									=hidden_act
													a__:							int									=hidden_dropout_prob
													a__:							Union[str, Any]									=attention_probs_dropout_prob
													a__:							Tuple									=initializer_range
													a__:							Dict									=num_labels
													a__:							Tuple									=scope
													a__:							List[Any]									=is_hybrid
													# sequence length of DPT = num_patches + 1 (we add 1 for the [CLS] token)
													a__:							Any									=(image_size // patch_size) ** 2
													a__:							Tuple									=num_patches + 1


						def 	_lowerCamelCase      (			self    :	Union[str, Any]      ):
													a__:							str									=floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]      )

													a__:							Dict									=None
													if self.use_labels:
																				a__:							Tuple									=ids_tensor([self.batch_size, self.image_size, self.image_size]   ,							self.num_labels      )

													a__:							Union[str, Any]									=self.get_config()

													return config, pixel_values, labels


						def 	_lowerCamelCase      (			self    :	Union[str, Any]      ):
													return DPTConfig(
													    image_size=self.image_size   ,							patch_size=self.patch_size   ,							num_channels=self.num_channels   ,							hidden_size=self.hidden_size   ,							num_hidden_layers=self.num_hidden_layers   ,							backbone_out_indices=self.backbone_out_indices   ,							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   ,							is_decoder=_a   ,							initializer_range=self.initializer_range   ,							is_hybrid=self.is_hybrid   ,							)


						def 	_lowerCamelCase      (			self    :	Optional[int]   ,							_a    :	Dict   ,							_a    :	List[str]   ,							_a    :	int      ):
													a__:							Union[str, Any]									=DPTModel(config=_a      )
													model.to(_a      )
													model.eval()
													a__:							List[Any]									=model(_a      )
													self.parent.assertEqual(result.last_hidden_state.shape   ,							(self.batch_size, self.seq_length, self.hidden_size)      )


						def 	_lowerCamelCase      (			self    :	int   ,							_a    :	Tuple   ,							_a    :	Union[str, Any]   ,							_a    :	List[Any]      ):
													a__:							Union[str, Any]									=self.num_labels
													a__:							List[str]									=DPTForDepthEstimation(_a      )
													model.to(_a      )
													model.eval()
													a__:							Optional[int]									=model(_a      )
													self.parent.assertEqual(result.predicted_depth.shape   ,							(self.batch_size, self.image_size, self.image_size)      )


						def 	_lowerCamelCase      (			self    :	List[Any]   ,							_a    :	str   ,							_a    :	Optional[int]   ,							_a    :	int      ):
													a__:							Any									=self.num_labels
													a__:							List[str]									=DPTForSemanticSegmentation(_a      )
													model.to(_a      )
													model.eval()
													a__:							Union[str, Any]									=model(_a   ,							labels=_a      )
													self.parent.assertEqual(
													    result.logits.shape   ,							(self.batch_size, self.num_labels, self.image_size, self.image_size)      )







						def 	_lowerCamelCase      (			self    :	Optional[Any]      ):
													a__:							List[str]									=self.prepare_config_and_inputs()
													a__				,      a__				,      a__:							Union[str, Any]									=config_and_inputs
													a__:							Tuple									={"pixel_values": pixel_values}
													return config, inputs_dict


@require_torch
class 						lowerCamelCase__			(		_a					,       _a					,       unittest.TestCase ):
						_lowerCAmelCase             =			(DPTModel, DPTForDepthEstimation, DPTForSemanticSegmentation) if is_torch_available() else ()
						_lowerCAmelCase             =			(
						    {
						        '''depth-estimation''': DPTForDepthEstimation,
						        '''feature-extraction''': DPTModel,
						        '''image-segmentation''': DPTForSemanticSegmentation,
						    }
						    if is_torch_available()
						    else {}
						)

						_lowerCAmelCase             =			False
						_lowerCAmelCase             =			False
						_lowerCAmelCase             =			False


						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													a__:							Dict									=DPTModelTester(self      )
													a__:							int									=ConfigTester(self   ,							config_class=_a   ,							has_text_modality=_a   ,							hidden_size=3_7      )


						def 	_lowerCamelCase      (			self    :	List[str]      ):
													self.config_tester.run_common_tests()


						@unittest.skip(reason="DPT does not use inputs_embeds"      )
						def 	_lowerCamelCase      (			self    :	List[Any]      ):
													pass


						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													a__				,      a__:							Optional[Any]									=self.model_tester.prepare_config_and_inputs_for_common()

													for model_class in self.all_model_classes:
																				a__:							Optional[Any]									=model_class(_a      )
																				self.assertIsInstance(model.get_input_embeddings()   ,							(nn.Module)      )
																				a__:							Optional[Any]									=model.get_output_embeddings()
																				self.assertTrue(x is None or isinstance(_a   ,							nn.Linear      )      )


						def 	_lowerCamelCase      (			self    :	List[Any]      ):
													a__				,      a__:							int									=self.model_tester.prepare_config_and_inputs_for_common()

													for model_class in self.all_model_classes:
																				a__:							Any									=model_class(_a      )
																				a__:							Any									=inspect.signature(model.forward      )
																				# signature.parameters is an OrderedDict => so arg_names order is deterministic
																				a__:							List[str]									=[*signature.parameters.keys()]

																				a__:							List[str]									=["pixel_values"]
																				self.assertListEqual(arg_names[:1]   ,							_a      )


						def 	_lowerCamelCase      (			self    :	Any      ):
													a__:							List[str]									=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_model(*_a      )


						def 	_lowerCamelCase      (			self    :	Dict      ):
													a__:							Any									=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_for_depth_estimation(*_a      )


						def 	_lowerCamelCase      (			self    :	Dict      ):
													a__:							Any									=self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_for_semantic_segmentation(*_a      )


						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													for model_class in self.all_model_classes:
																				if model_class.__name__ == "DPTForDepthEstimation":
																											continue

																				a__				,      a__:							List[Any]									=self.model_tester.prepare_config_and_inputs_for_common()
																				a__:							Optional[int]									=True

																				if model_class in get_values(_a      ):
																											continue

																				a__:							Union[str, Any]									=model_class(_a      )
																				model.to(_a      )
																				model.train()
																				a__:							Optional[int]									=self._prepare_for_class(_a   ,							_a   ,							return_labels=_a      )
																				a__:							List[Any]									=model(**_a      ).loss
																				loss.backward()


						def 	_lowerCamelCase      (			self    :	Union[str, Any]      ):
													for model_class in self.all_model_classes:
																				if model_class.__name__ == "DPTForDepthEstimation":
																											continue

																				a__				,      a__:							Optional[Any]									=self.model_tester.prepare_config_and_inputs_for_common()
																				a__:							Union[str, Any]									=False
																				a__:							Any									=True

																				if model_class in get_values(_a      ) or not model_class.supports_gradient_checkpointing:
																											continue
																				a__:							Tuple									=model_class(_a      )
																				model.to(_a      )
																				model.gradient_checkpointing_enable()
																				model.train()
																				a__:							Dict									=self._prepare_for_class(_a   ,							_a   ,							return_labels=_a      )
																				a__:							Optional[Any]									=model(**_a      ).loss
																				loss.backward()


						def 	_lowerCamelCase      (			self    :	Tuple      ):
													a__				,      a__:							str									=self.model_tester.prepare_config_and_inputs_for_common()

													a__:							Tuple									=_config_zero_init(_a      )
													for model_class in self.all_model_classes:
																				a__:							List[Any]									=model_class(config=_a      )
																				# Skip the check for the backbone
																				a__:							List[Any]									=[]
																				for name, module in model.named_modules():
																											if module.__class__.__name__ == "DPTViTHybridEmbeddings":
																																		a__:							Optional[Any]									=[F"{name}.{key}" for key in module.state_dict().keys()]
																																		break

																				for name, param in model.named_parameters():
																											if param.requires_grad:
																																		if name in backbone_params:
																																									continue
																																		self.assertIn(
																																		    ((param.data.mean() * 1e9).round() / 1e9).item()   ,							[0.0, 1.0]   ,							msg=F"Parameter {name} of model {model_class} seems not properly initialized"   ,							)


						@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."      )
						def 	_lowerCamelCase      (			self    :	Dict      ):
													pass







						@slow
						def 	_lowerCamelCase      (			self    :	str      ):
													for model_name in DPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																				a__:							List[Any]									=DPTModel.from_pretrained(_a      )
																				self.assertIsNotNone(_a      )





def     __lowerCamelCase							(   ):
							a__:							Optional[int]									=Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"    )
							return image







@require_torch
@require_vision
@slow
class 						lowerCamelCase__			(		unittest.TestCase ):


						def 	_lowerCamelCase      (			self    :	List[Any]      ):
													a__:							Any									=DPTImageProcessor.from_pretrained("Intel/dpt-large"      )
													a__:							str									=DPTForDepthEstimation.from_pretrained("Intel/dpt-large"      ).to(_a      )

													a__:							Optional[int]									=prepare_img()
													a__:							List[Any]									=image_processor(images=_a   ,							return_tensors="pt"      ).to(_a      )

													# forward pass
													with torch.no_grad():
																				a__:							int									=model(**_a      )
																				a__:							List[str]									=outputs.predicted_depth

													# verify the predicted depth
													a__:							Optional[int]									=torch.Size((1, 3_8_4, 3_8_4)      )
													self.assertEqual(predicted_depth.shape   ,							_a      )

													a__:							Union[str, Any]									=torch.tensor(
													    [[6.3_1_9_9, 6.3_6_2_9, 6.4_1_4_8], [6.3_8_5_0, 6.3_6_1_5, 6.4_1_6_6], [6.3_5_1_9, 6.3_1_7_6, 6.3_5_7_5]]      ).to(_a      )

													self.assertTrue(torch.allclose(outputs.predicted_depth[0, :3, :3]   ,							_a   ,							atol=1e-4      )      )


						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													a__:							Optional[Any]									=DPTImageProcessor.from_pretrained("Intel/dpt-large-ade"      )
													a__:							Union[str, Any]									=DPTForSemanticSegmentation.from_pretrained("Intel/dpt-large-ade"      ).to(_a      )

													a__:							Dict									=prepare_img()
													a__:							Optional[int]									=image_processor(images=_a   ,							return_tensors="pt"      ).to(_a      )

													# forward pass
													with torch.no_grad():
																				a__:							int									=model(**_a      )

													# verify the logits
													a__:							int									=torch.Size((1, 1_5_0, 4_8_0, 4_8_0)      )
													self.assertEqual(outputs.logits.shape   ,							_a      )

													a__:							Any									=torch.tensor(
													    [[4.0_4_8_0, 4.2_4_2_0, 4.4_3_6_0], [4.3_1_2_4, 4.5_6_9_3, 4.8_2_6_1], [4.5_7_6_8, 4.8_9_6_5, 5.2_1_6_3]]      ).to(_a      )

													self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]   ,							_a   ,							atol=1e-4      )      )







						def 	_lowerCamelCase      (			self    :	int      ):
													a__:							Dict									=DPTImageProcessor.from_pretrained("Intel/dpt-large-ade"      )
													a__:							List[Any]									=DPTForSemanticSegmentation.from_pretrained("Intel/dpt-large-ade"      ).to(_a      )

													a__:							int									=prepare_img()
													a__:							Any									=image_processor(images=_a   ,							return_tensors="pt"      ).to(_a      )

													# forward pass
													with torch.no_grad():
																				a__:							Optional[int]									=model(**_a      )

													a__:							Union[str, Any]									=outputs.logits.detach().cpu()

													a__:							List[Any]									=image_processor.post_process_semantic_segmentation(outputs=_a   ,							target_sizes=[(5_0_0, 3_0_0)]      )
													a__:							Dict									=torch.Size((5_0_0, 3_0_0)      )
													self.assertEqual(segmentation[0].shape   ,							_a      )

													a__:							Optional[Any]									=image_processor.post_process_semantic_segmentation(outputs=_a      )
													a__:							List[str]									=torch.Size((4_8_0, 4_8_0)      )
													self.assertEqual(segmentation[0].shape   ,							_a      )





