

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				,    lowerCamelCase				,    lowerCamelCase=2				,    lowerCamelCase=32				,    lowerCamelCase=16				,    lowerCamelCase=3				,    lowerCamelCase=True				,    lowerCamelCase=True				,    lowerCamelCase=32				,    lowerCamelCase=4				,    lowerCamelCase=[0, 1, 2, 3]				,    lowerCamelCase=4				,    lowerCamelCase=37				,    lowerCamelCase="gelu"				,    lowerCamelCase=0.1				,    lowerCamelCase=0.1				,    lowerCamelCase=0.02				,    lowerCamelCase=3				,    lowerCamelCase=False				,    lowerCamelCase=None				,    )		->			Any:
        snake_case_					= parent
        snake_case_					= batch_size
        snake_case_					= image_size
        snake_case_					= patch_size
        snake_case_					= num_channels
        snake_case_					= is_training
        snake_case_					= use_labels
        snake_case_					= hidden_size
        snake_case_					= num_hidden_layers
        snake_case_					= backbone_out_indices
        snake_case_					= num_attention_heads
        snake_case_					= intermediate_size
        snake_case_					= hidden_act
        snake_case_					= hidden_dropout_prob
        snake_case_					= attention_probs_dropout_prob
        snake_case_					= initializer_range
        snake_case_					= num_labels
        snake_case_					= scope
        snake_case_					= is_hybrid
        # sequence length of DPT = num_patches + 1 (we add 1 for the [CLS] token)
        snake_case_					= (image_size // patch_size) ** 2
        snake_case_					= num_patches + 1







      def       lowerCAmelCase_       (	self  )		->			int:
        snake_case_					= floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]  )

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

        snake_case_					= self.get_config()

        return config, pixel_values, labels







      def       lowerCAmelCase_       (	self  )		->			Optional[int]:
        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=lowerCamelCase				,    initializer_range=self.initializer_range				,    is_hybrid=self.is_hybrid				,    )







      def       lowerCAmelCase_       (	self				,    lowerCamelCase				,    lowerCamelCase				,    lowerCamelCase  )		->			Optional[int]:
        snake_case_					= DPTModel(config=lowerCamelCase  )
        model.to(lowerCamelCase  )
        model.eval()
        snake_case_					= model(lowerCamelCase  )
        self.parent.assertEqual(result.last_hidden_state.shape				,    (self.batch_size, self.seq_length, self.hidden_size)  )







      def       lowerCAmelCase_       (	self				,    lowerCamelCase				,    lowerCamelCase				,    lowerCamelCase  )		->			List[Any]:
        snake_case_					= self.num_labels
        snake_case_					= DPTForDepthEstimation(lowerCamelCase  )
        model.to(lowerCamelCase  )
        model.eval()
        snake_case_					= model(lowerCamelCase  )
        self.parent.assertEqual(result.predicted_depth.shape				,    (self.batch_size, self.image_size, self.image_size)  )







      def       lowerCAmelCase_       (	self				,    lowerCamelCase				,    lowerCamelCase				,    lowerCamelCase  )		->			Optional[Any]:
        snake_case_					= self.num_labels
        snake_case_					= DPTForSemanticSegmentation(lowerCamelCase  )
        model.to(lowerCamelCase  )
        model.eval()
        snake_case_					= model(lowerCamelCase				,    labels=lowerCamelCase  )
        self.parent.assertEqual(
            result.logits.shape				,    (self.batch_size, self.num_labels, self.image_size, self.image_size)  )







      def       lowerCAmelCase_       (	self  )		->			List[Any]:
        snake_case_					= self.prepare_config_and_inputs()
        snake_case_		,							snake_case_		,							snake_case_					= config_and_inputs
        snake_case_					= {"""pixel_values""": pixel_values}
        return config, inputs_dict







@require_torch
class 					__lowerCamelCase							(      __snake_case    ,				__snake_case    ,				unittest.TestCase      ):
      lowerCamelCase_      : Union[str, Any]  			=   (DPTModel, DPTForDepthEstimation, DPTForSemanticSegmentation) if is_torch_available() else ()
      lowerCamelCase_      : Union[str, Any]  			=   (
          {
              'depth-estimation': DPTForDepthEstimation,
              'feature-extraction': DPTModel,
              'image-segmentation': DPTForSemanticSegmentation,
          }
          if is_torch_available()
          else {}
      )

      lowerCamelCase_      : Optional[Any]  			=   False
      lowerCamelCase_      : List[Any]  			=   False
      lowerCamelCase_      : Optional[Any]  			=   False







      def       lowerCAmelCase_       (	self  )		->			str:
        snake_case_					= DPTModelTester(self  )
        snake_case_					= ConfigTester(self				,    config_class=lowerCamelCase				,    has_text_modality=lowerCamelCase				,    hidden_size=37  )







      def       lowerCAmelCase_       (	self  )		->			Dict:
        self.config_tester.run_common_tests()







      @unittest.skip(reason="""DPT does not use inputs_embeds"""  )
      def       lowerCAmelCase_       (	self  )		->			Dict:
        pass







      def       lowerCAmelCase_       (	self  )		->			str:
        snake_case_		,							snake_case_					= self.model_tester.prepare_config_and_inputs_for_common()

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







      def       lowerCAmelCase_       (	self  )		->			Optional[Any]:
        snake_case_		,							snake_case_					= self.model_tester.prepare_config_and_inputs_for_common()

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

          snake_case_					= ["""pixel_values"""]
          self.assertListEqual(arg_names[:1]				,    lowerCamelCase  )







      def       lowerCAmelCase_       (	self  )		->			List[Any]:
        snake_case_					= self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*lowerCamelCase  )







      def       lowerCAmelCase_       (	self  )		->			Dict:
        snake_case_					= self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_depth_estimation(*lowerCamelCase  )







      def       lowerCAmelCase_       (	self  )		->			List[Any]:
        snake_case_					= self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_semantic_segmentation(*lowerCamelCase  )







      def       lowerCAmelCase_       (	self  )		->			Optional[Any]:
        for model_class in self.all_model_classes:
          if model_class.__name__ == "DPTForDepthEstimation":
            continue

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

          if model_class in get_values(lowerCamelCase  ):
            continue

          snake_case_					= model_class(lowerCamelCase  )
          model.to(lowerCamelCase  )
          model.train()
          snake_case_					= self._prepare_for_class(lowerCamelCase				,    lowerCamelCase				,    return_labels=lowerCamelCase  )
          snake_case_					= model(**lowerCamelCase  ).loss
          loss.backward()







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

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

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







      def       lowerCAmelCase_       (	self  )		->			Union[str, Any]:
        snake_case_		,							snake_case_					= self.model_tester.prepare_config_and_inputs_for_common()

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







      @slow
      def       lowerCAmelCase_       (	self  )		->			Any:
        for model_name in DPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
          snake_case_					= DPTModel.from_pretrained(lowerCamelCase  )
          self.assertIsNotNone(lowerCamelCase  )



def   UpperCamelCase(   )      ->							Optional[int]:





  '''simple docstring'''





  snake_case_					= Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""				)
  return image







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







      def       lowerCAmelCase_       (	self  )		->			str:
        snake_case_					= DPTImageProcessor.from_pretrained("""Intel/dpt-large"""  )
        snake_case_					= DPTForDepthEstimation.from_pretrained("""Intel/dpt-large"""  ).to(lowerCamelCase  )

        snake_case_					= prepare_img()
        snake_case_					= image_processor(images=lowerCamelCase				,    return_tensors="""pt"""  ).to(lowerCamelCase  )

        # forward pass
        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )
          snake_case_					= outputs.predicted_depth

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

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

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







      def       lowerCAmelCase_       (	self  )		->			List[Any]:
        snake_case_					= DPTImageProcessor.from_pretrained("""Intel/dpt-large-ade"""  )
        snake_case_					= DPTForSemanticSegmentation.from_pretrained("""Intel/dpt-large-ade"""  ).to(lowerCamelCase  )

        snake_case_					= prepare_img()
        snake_case_					= image_processor(images=lowerCamelCase				,    return_tensors="""pt"""  ).to(lowerCamelCase  )

        # forward pass
        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )

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

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

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







      def       lowerCAmelCase_       (	self  )		->			int:
        snake_case_					= DPTImageProcessor.from_pretrained("""Intel/dpt-large-ade"""  )
        snake_case_					= DPTForSemanticSegmentation.from_pretrained("""Intel/dpt-large-ade"""  ).to(lowerCamelCase  )

        snake_case_					= prepare_img()
        snake_case_					= image_processor(images=lowerCamelCase				,    return_tensors="""pt"""  ).to(lowerCamelCase  )

        # forward pass
        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )

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

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

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


import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin







class 					__lowerCamelCase							(      __snake_case      ):
      lowerCamelCase_      : Dict  			=   'AutoFeatureExtractor'
      lowerCamelCase_      : Optional[int]  			=   'Speech2Text2Tokenizer'







      def __init__(	self				,    lowerCamelCase				,    lowerCamelCase  )		->			List[str]:
        super().__init__(lowerCamelCase				,    lowerCamelCase  )
        snake_case_					= self.feature_extractor
        snake_case_					= False







      def __call__(	self				,    *lowerCamelCase				,    **lowerCamelCase  )		->			Dict:
        # For backward compatibility
        if self._in_target_context_manager:
          return self.current_processor(*lowerCamelCase				,    **lowerCamelCase  )

        if "raw_speech" in kwargs:
          warnings.warn("""Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."""  )
          snake_case_					= kwargs.pop("""raw_speech"""  )
        else:
          snake_case_					= kwargs.pop("""audio"""				,    lowerCamelCase  )
        snake_case_					= kwargs.pop("""sampling_rate"""				,    lowerCamelCase  )
        snake_case_					= kwargs.pop("""text"""				,    lowerCamelCase  )
        if len(lowerCamelCase  ) > 0:
          snake_case_					= args[0]
          snake_case_					= args[1:]

        if audio is None and text is None:
          raise ValueError("""You need to specify either an `audio` or `text` input to process."""  )

        if audio is not None:
          snake_case_					= self.feature_extractor(lowerCamelCase				,    *lowerCamelCase				,    sampling_rate=lowerCamelCase				,    **lowerCamelCase  )
        if text is not None:
          snake_case_					= self.tokenizer(lowerCamelCase				,    **lowerCamelCase  )

        if text is None:
          return inputs
        elif audio is None:
          return encodings
        else:
          snake_case_					= encodings["""input_ids"""]
          return inputs







      def       lowerCAmelCase_       (	self				,    *lowerCamelCase				,    **lowerCamelCase  )		->			int:
        return self.tokenizer.batch_decode(*lowerCamelCase				,    **lowerCamelCase  )







      def       lowerCAmelCase_       (	self				,    *lowerCamelCase				,    **lowerCamelCase  )		->			Union[str, Any]:
        return self.tokenizer.decode(*lowerCamelCase				,    **lowerCamelCase  )







      @contextmanager
      def       lowerCAmelCase_       (	self  )		->			Any:
        warnings.warn(
            """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
            """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
            """your audio inputs, or in a separate call."""  )
        snake_case_					= True
        snake_case_					= self.tokenizer
        yield
        snake_case_					= self.feature_extractor
        snake_case_					= False