
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     __A:




			def __init__(  self	,  _snake_case	,  _snake_case=2	,  _snake_case=32	,  _snake_case=16	,  _snake_case=3	,  _snake_case=True	,  _snake_case=True	,  _snake_case=32	,  _snake_case=4	,  _snake_case=[0, 1, 2, 3]	,  _snake_case=4	,  _snake_case=37	,  _snake_case="gelu"	,  _snake_case=0.1	,  _snake_case=0.1	,  _snake_case=0.02	,  _snake_case=3	,  _snake_case=False	,  _snake_case=None	,  )					->   Any:


							'''simple docstring'''




							__a    							=					parent
							__a    							=					batch_size
							__a    							=					image_size
							__a    							=					patch_size
							__a    							=					num_channels
							__a    							=					is_training
							__a    							=					use_labels
							__a    							=					hidden_size
							__a    							=					num_hidden_layers
							__a    							=					backbone_out_indices
							__a    							=					num_attention_heads
							__a    							=					intermediate_size
							__a    							=					hidden_act
							__a    							=					hidden_dropout_prob
							__a    							=					attention_probs_dropout_prob
							__a    							=					initializer_range
							__a    							=					num_labels
							__a    							=					scope
							__a    							=					is_hybrid
							# sequence length of DPT = num_patches + 1 (we add 1 for the [CLS] token)
							__a    							=					(image_size // patch_size) ** 2
							__a    							=					num_patches + 1




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   Union[str, Any]:


							'''simple docstring'''




							__a    							=					floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]       )

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

							__a    							=					self.get_config()

							return config, pixel_values, labels




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   Union[str, Any]:


							'''simple docstring'''




							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=_snake_case	,  initializer_range=self.initializer_range	,  is_hybrid=self.is_hybrid	,  )




			def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case	,  _snake_case       )					->   Optional[int]:


							'''simple docstring'''




							__a    							=					DPTModel(config=_snake_case       )
							model.to(_snake_case       )
							model.eval()
							__a    							=					model(_snake_case       )
							self.parent.assertEqual(result.last_hidden_state.shape	,  (self.batch_size, self.seq_length, self.hidden_size)       )




			def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case	,  _snake_case       )					->   str:


							'''simple docstring'''




							__a    							=					self.num_labels
							__a    							=					DPTForDepthEstimation(_snake_case       )
							model.to(_snake_case       )
							model.eval()
							__a    							=					model(_snake_case       )
							self.parent.assertEqual(result.predicted_depth.shape	,  (self.batch_size, self.image_size, self.image_size)       )




			def        SCREAMING_SNAKE_CASE_	(  self	,  _snake_case	,  _snake_case	,  _snake_case       )					->   Any:


							'''simple docstring'''




							__a    							=					self.num_labels
							__a    							=					DPTForSemanticSegmentation(_snake_case       )
							model.to(_snake_case       )
							model.eval()
							__a    							=					model(_snake_case	,  labels=_snake_case       )
							self.parent.assertEqual(
							    result.logits.shape	,  (self.batch_size, self.num_labels, self.image_size, self.image_size)       )




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   Dict:


							'''simple docstring'''




							__a    							=					self.prepare_config_and_inputs()
							__a							,  __a							,  __a    							=					config_and_inputs
							__a    							=					{'''pixel_values''': pixel_values}
							return config, inputs_dict







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

			snake_case_		     =					False
			snake_case_		     =					False
			snake_case_		     =					False




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


							'''simple docstring'''




							__a    							=					DPTModelTester(self       )
							__a    							=					ConfigTester(self	,  config_class=_snake_case	,  has_text_modality=_snake_case	,  hidden_size=37       )




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   Tuple:


							'''simple docstring'''




							self.config_tester.run_common_tests()




			@unittest.skip(reason='''DPT does not use inputs_embeds'''       )
			def        SCREAMING_SNAKE_CASE_	(  self       )					->   str:


							'''simple docstring'''




							pass




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


							'''simple docstring'''




							__a							,  __a    							=					self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
											__a    							=					model_class(_snake_case       )
											self.assertIsInstance(model.get_input_embeddings()	,  (nn.Module)       )
											__a    							=					model.get_output_embeddings()
											self.assertTrue(x is None or isinstance(_snake_case	,  nn.Linear       )       )




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


							'''simple docstring'''




							__a							,  __a    							=					self.model_tester.prepare_config_and_inputs_for_common()

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

											__a    							=					['''pixel_values''']
											self.assertListEqual(arg_names[:1]	,  _snake_case       )




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


							'''simple docstring'''




							__a    							=					self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*_snake_case       )




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


							'''simple docstring'''




							__a    							=					self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_depth_estimation(*_snake_case       )




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   int:


							'''simple docstring'''




							__a    							=					self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_semantic_segmentation(*_snake_case       )




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


							'''simple docstring'''




							for model_class in self.all_model_classes:
											if model_class.__name__ == "DPTForDepthEstimation":
															continue

											__a							,  __a    							=					self.model_tester.prepare_config_and_inputs_for_common()
											__a    							=					True

											if model_class in get_values(_snake_case       ):
															continue

											__a    							=					model_class(_snake_case       )
											model.to(_snake_case       )
											model.train()
											__a    							=					self._prepare_for_class(_snake_case	,  _snake_case	,  return_labels=_snake_case       )
											__a    							=					model(**_snake_case       ).loss
											loss.backward()




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   Any:


							'''simple docstring'''




							for model_class in self.all_model_classes:
											if model_class.__name__ == "DPTForDepthEstimation":
															continue

											__a							,  __a    							=					self.model_tester.prepare_config_and_inputs_for_common()
											__a    							=					False
											__a    							=					True

											if model_class in get_values(_snake_case       ) or not model_class.supports_gradient_checkpointing:
															continue
											__a    							=					model_class(_snake_case       )
											model.to(_snake_case       )
											model.gradient_checkpointing_enable()
											model.train()
											__a    							=					self._prepare_for_class(_snake_case	,  _snake_case	,  return_labels=_snake_case       )
											__a    							=					model(**_snake_case       ).loss
											loss.backward()




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


							'''simple docstring'''




							__a							,  __a    							=					self.model_tester.prepare_config_and_inputs_for_common()

							__a    							=					_config_zero_init(_snake_case       )
							for model_class in self.all_model_classes:
											__a    							=					model_class(config=_snake_case       )
											# Skip the check for the backbone
											__a    							=					[]
											for name, module in model.named_modules():
															if module.__class__.__name__ == "DPTViTHybridEmbeddings":
																			__a    							=					[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        SCREAMING_SNAKE_CASE_	(  self       )					->   List[str]:


							'''simple docstring'''




							pass




			@slow
			def        SCREAMING_SNAKE_CASE_	(  self       )					->   Union[str, Any]:


							'''simple docstring'''




							for model_name in DPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											__a    							=					DPTModel.from_pretrained(_snake_case       )
											self.assertIsNotNone(_snake_case       )



def 					__lowerCAmelCase    (    ) ->						Tuple:
				__a    							=					Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''						)
				return image







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




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   Dict:


							'''simple docstring'''




							__a    							=					DPTImageProcessor.from_pretrained('''Intel/dpt-large'''       )
							__a    							=					DPTForDepthEstimation.from_pretrained('''Intel/dpt-large'''       ).to(_snake_case       )

							__a    							=					prepare_img()
							__a    							=					image_processor(images=_snake_case	,  return_tensors='''pt'''       ).to(_snake_case       )

							# forward pass
							with torch.no_grad():
											__a    							=					model(**_snake_case       )
											__a    							=					outputs.predicted_depth

							# verify the predicted depth
							__a    							=					torch.Size((1, 384, 384)       )
							self.assertEqual(predicted_depth.shape	,  _snake_case       )

							__a    							=					torch.tensor(
							    [[6.3199, 6.3629, 6.4148], [6.3850, 6.3615, 6.4166], [6.3519, 6.3176, 6.3575]]       ).to(_snake_case       )

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




			def        SCREAMING_SNAKE_CASE_	(  self       )					->   int:


							'''simple docstring'''




							__a    							=					DPTImageProcessor.from_pretrained('''Intel/dpt-large-ade'''       )
							__a    							=					DPTForSemanticSegmentation.from_pretrained('''Intel/dpt-large-ade'''       ).to(_snake_case       )

							__a    							=					prepare_img()
							__a    							=					image_processor(images=_snake_case	,  return_tensors='''pt'''       ).to(_snake_case       )

							# forward pass
							with torch.no_grad():
											__a    							=					model(**_snake_case       )

							# verify the logits
							__a    							=					torch.Size((1, 150, 480, 480)       )
							self.assertEqual(outputs.logits.shape	,  _snake_case       )

							__a    							=					torch.tensor(
							    [[4.0480, 4.2420, 4.4360], [4.3124, 4.5693, 4.8261], [4.5768, 4.8965, 5.2163]]       ).to(_snake_case       )

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




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


							'''simple docstring'''




							__a    							=					DPTImageProcessor.from_pretrained('''Intel/dpt-large-ade'''       )
							__a    							=					DPTForSemanticSegmentation.from_pretrained('''Intel/dpt-large-ade'''       ).to(_snake_case       )

							__a    							=					prepare_img()
							__a    							=					image_processor(images=_snake_case	,  return_tensors='''pt'''       ).to(_snake_case       )

							# forward pass
							with torch.no_grad():
											__a    							=					model(**_snake_case       )

							__a    							=					outputs.logits.detach().cpu()

							__a    							=					image_processor.post_process_semantic_segmentation(outputs=_snake_case	,  target_sizes=[(500, 300)]       )
							__a    							=					torch.Size((500, 300)       )
							self.assertEqual(segmentation[0].shape	,  _snake_case       )

							__a    							=					image_processor.post_process_semantic_segmentation(outputs=_snake_case       )
							__a    							=					torch.Size((480, 480)       )
							self.assertEqual(segmentation[0].shape	,  _snake_case       )

import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


A  :			Optional[int]     				=				2 << 2_0  # 2mb
A  :			Optional[int]     				=				get_logger(__name__)







class     __A(				a    ):




				def        SCREAMING_SNAKE_CASE_	(  self       )					->   Union[str, Any]:


								'''simple docstring'''




								__a    							=					self._options.get_all_options().get('''runner'''       )
								return runner in [None, "DirectRunner", "PortableRunner"]



def 					__lowerCAmelCase    (    a__		,					a__		,					a__=False						) ->						Union[str, Any]:
				__a    							=					FileSystems
				if fs.exists(a__						):
								if force_upload:
												logger.info(F"""Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True."""						)
								else:
												logger.info(F"""Remote path already exist: {remote_file_path}. Skipping it as force_upload=False."""						)
												return
				with fs.create(a__						) as remote_file:
								with open(a__		,					'''rb'''						) as local_file:
												__a    							=					local_file.read(a__						)
												while chunk:
																remote_file.write(a__						)
																__a    							=					local_file.read(a__						)






def 					__lowerCAmelCase    (    a__		,					a__		,					a__=False						) ->						Dict:
				__a    							=					FileSystems
				if os.path.exists(a__						):
								if force_download:
												logger.info(F"""Local path already exist: {remote_file_path}. Overwriting it as force_upload=True."""						)
								else:
												logger.info(F"""Local path already exist: {remote_file_path}. Skipping it as force_upload=False."""						)
												return
				with fs.open(a__						) as remote_file:
								with open(a__		,					'''wb'''						) as local_file:
												__a    							=					remote_file.read(a__						)
												while chunk:
																local_file.write(a__						)
																__a    							=					remote_file.read(a__						)