"""simple docstring"""



from __future__ import annotations

import collections
import tempfile
import unittest

import numpy as np

from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import is_tf_available, is_vision_available

from ...test_modeling_tf_common import floats_tensor, ids_tensor, random_attention_mask
from ..bert.test_modeling_tf_bert import TFBertModelTester
from ..clip.test_modeling_tf_clip import TFCLIPVisionModelTester
from ..deit.test_modeling_tf_deit import TFDeiTModelTester
from ..roberta.test_modeling_tf_roberta import TFRobertaModelTester
from ..vit.test_modeling_tf_vit import TFViTModelTester


if is_tf_available():
    from transformers import (
        TFBertModel,
        TFCLIPVisionModel,
        TFDeiTModel,
        TFRobertaModel,
        TFVisionTextDualEncoderModel,
        TFViTModel,
        VisionTextDualEncoderConfig,
    )

if is_vision_available():
    from PIL import Image

    from transformers import VisionTextDualEncoderProcessor



def 					_lowerCamelCase(					a							):
       if isinstance(a ,  collections.abc.Iterable							):
              return x
       return (x, x)






@require_tf
class 				snake_case__     :


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase		):
        pass


 def 		a__ (				self		):
        pass


 def 		a__ (				self		):
        pass


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase=None  ,						**lowerCamelCase		):
        __a     				=  VisionTextDualEncoderConfig.from_vision_text_configs(lowerCamelCase  ,						lowerCamelCase		)

        __a     				=  TFVisionTextDualEncoderModel(lowerCamelCase		)

        __a     				=  model(input_ids=lowerCamelCase  ,						pixel_values=lowerCamelCase  ,						attention_mask=lowerCamelCase		)

        self.assertEqual(output["text_embeds"].shape  ,						(input_ids.shape[0], config.projection_dim)		)
        self.assertEqual(output["image_embeds"].shape  ,						(pixel_values.shape[0], config.projection_dim)		)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase=None  ,						**lowerCamelCase		):
        __a				, __a     				=  self.get_vision_text_model(lowerCamelCase  ,						lowerCamelCase		)
        __a     				=  TFVisionTextDualEncoderModel(vision_model=lowerCamelCase  ,						text_model=lowerCamelCase		)

        __a     				=  model(input_ids=lowerCamelCase  ,						pixel_values=lowerCamelCase  ,						attention_mask=lowerCamelCase		)

        self.assertEqual(output["text_embeds"].shape  ,						(input_ids.shape[0], model.config.projection_dim)		)
        self.assertEqual(output["image_embeds"].shape  ,						(pixel_values.shape[0], model.config.projection_dim)		)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase=None  ,						**lowerCamelCase		):
        __a				, __a     				=  self.get_vision_text_model(lowerCamelCase  ,						lowerCamelCase		)
        __a     				=  {"vision_model": vision_model, "text_model": text_model}
        __a     				=  TFVisionTextDualEncoderModel.from_vision_text_pretrained(**lowerCamelCase		)

        __a     				=  model(input_ids=lowerCamelCase  ,						pixel_values=lowerCamelCase  ,						attention_mask=lowerCamelCase		)

        self.assertEqual(output["text_embeds"].shape  ,						(input_ids.shape[0], model.config.projection_dim)		)
        self.assertEqual(output["image_embeds"].shape  ,						(pixel_values.shape[0], model.config.projection_dim)		)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase=None  ,						**lowerCamelCase		):
        __a				, __a     				=  self.get_vision_text_model(lowerCamelCase  ,						lowerCamelCase		)
        __a     				=  TFVisionTextDualEncoderModel(vision_model=lowerCamelCase  ,						text_model=lowerCamelCase		)

        __a     				=  model(input_ids=lowerCamelCase  ,						pixel_values=lowerCamelCase  ,						attention_mask=lowerCamelCase		)
        __a     				=  output[0].numpy()

        with tempfile.TemporaryDirectory() as tmpdirname:
               model.save_pretrained(lowerCamelCase		)
               __a     				=  TFVisionTextDualEncoderModel.from_pretrained(lowerCamelCase		)

               __a     				=  model(input_ids=lowerCamelCase  ,						pixel_values=lowerCamelCase  ,						attention_mask=lowerCamelCase		)
               __a     				=  after_output[0].numpy()
               __a     				=  np.amax(np.abs(out_a - out_a		)		)
               self.assertLessEqual(lowerCamelCase  ,						1E-5		)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase=None  ,						**lowerCamelCase		):
        __a				, __a     				=  self.get_vision_text_model(lowerCamelCase  ,						lowerCamelCase		)
        __a     				=  TFVisionTextDualEncoderModel(vision_model=lowerCamelCase  ,						text_model=lowerCamelCase		)

        __a     				=  model(
            input_ids=lowerCamelCase  ,						pixel_values=lowerCamelCase  ,						attention_mask=lowerCamelCase  ,						output_attentions=lowerCamelCase		)

        __a     				=  output.vision_model_output.attentions
        self.assertEqual(len(lowerCamelCase		)  ,						vision_config.num_hidden_layers		)

        # in ViT, the seq_len equals the number of patches + 1 (we add 1 for the [CLS] token)
        __a     				=  to_atuple(vision_model.config.image_size		)
        __a     				=  to_atuple(vision_model.config.patch_size		)
        __a     				=  (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])
        __a     				=  num_patches + 1
        self.assertEqual(vision_attentions[0].shape[-3:]  ,						(vision_config.num_attention_heads, seq_len, seq_len)		)

        __a     				=  output.text_model_output.attentions
        self.assertEqual(len(lowerCamelCase		)  ,						text_config.num_hidden_layers		)

        self.assertEqual(
            text_attentions[0].shape[-3:]  ,						(text_config.num_attention_heads, input_ids.shape[-1], input_ids.shape[-1])  ,						)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase		):
        __a     				=  np.abs((a - b)		).max()
        self.assertLessEqual(lowerCamelCase  ,						lowerCamelCase  ,						F"Difference between torch and flax is {diff} (>= {tol})."		)


 def 		a__ (				self		):
        __a     				=  self.prepare_config_and_inputs()
        self.check_vision_text_dual_encoder_model(**lowerCamelCase		)


 def 		a__ (				self		):
        __a     				=  self.prepare_config_and_inputs()
        self.check_model_from_pretrained_configs(**lowerCamelCase		)


 def 		a__ (				self		):
        __a     				=  self.prepare_config_and_inputs()
        self.check_vision_text_dual_encoder_from_pretrained(**lowerCamelCase		)


 def 		a__ (				self		):
        __a     				=  self.prepare_config_and_inputs()
        self.check_save_load(**lowerCamelCase		)


 def 		a__ (				self		):
        __a     				=  self.prepare_config_and_inputs()
        self.check_vision_text_output_attention(**lowerCamelCase		)






 @slow
 def 		a__ (				self		):
        __a				, __a     				=  self.get_pretrained_model_and_inputs()

        __a     				=  model_a(**lowerCamelCase		)
        __a     				=  outputs[0].numpy()

        with tempfile.TemporaryDirectory() as tmp_dirname:
               model_a.save_pretrained(lowerCamelCase		)
               __a     				=  TFVisionTextDualEncoderModel.from_pretrained(lowerCamelCase		)

               __a     				=  model_a(**lowerCamelCase		)
               __a     				=  after_outputs[0].numpy()
               __a     				=  np.amax(np.abs(out_a - out_a		)		)
               self.assertLessEqual(lowerCamelCase  ,						1E-5		)






@require_tf
class 				snake_case__     (   snake_case_,		unittest.TestCase   ):


 def 		a__ (				self		):
        __a     				=  TFVisionTextDualEncoderModel.from_vision_text_pretrained(
            "hf-internal-testing/tiny-random-vit"  ,						"hf-internal-testing/tiny-random-bert"		)
        __a     				=  13
        __a     				=  floats_tensor(
            [
                batch_size,
                model.vision_model.config.num_channels,
                model.vision_model.config.image_size,
                model.vision_model.config.image_size,
            ]		)
        __a     				=  ids_tensor([batch_size, 4]  ,						model.text_model.config.vocab_size		)
        __a     				=  random_attention_mask([batch_size, 4]		)
        __a     				=  {"pixel_values": pixel_values, "input_ids": input_ids, "attention_mask": attention_mask}

        return model, inputs


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase		):
        __a     				=  TFViTModel(lowerCamelCase  ,						name="vision_model"		)
        __a     				=  TFBertModel(lowerCamelCase  ,						name="text_model"		)
        return vision_model, text_model






 def 		a__ (				self		):
        __a     				=  TFViTModelTester(self		)
        __a     				=  TFBertModelTester(self		)
        __a     				=  vit_model_tester.prepare_config_and_inputs()
        __a     				=  bert_model_tester.prepare_config_and_inputs()

        __a				, __a				, __a     				=  vision_config_and_inputs

        (
            (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				,  
        )     				=  text_config_and_inputs

        return {
            "text_config": text_config,
            "vision_config": vision_config,
            "pixel_values": pixel_values,
            "attention_mask": input_mask,
            "input_ids": input_ids,
            "text_token_type_ids": token_type_ids,
            "text_sequence_labels": sequence_labels,
            "text_token_labels": token_labels,
            "text_choice_labels": choice_labels,
        }






@require_tf
class 				snake_case__     (   snake_case_,		unittest.TestCase   ):


 def 		a__ (				self		):
        # DeiT repo doesn't have TF weights, but we don't actually use the weights at all so let's
        # just reinitialize it.
        __a     				=  TFVisionTextDualEncoderModel.from_vision_text_pretrained(
            "Rocketknight1/tiny-random-deit-tf"  ,						"hf-internal-testing/tiny-random-roberta"		)
        __a     				=  13
        __a     				=  floats_tensor(
            [
                batch_size,
                model.vision_model.config.num_channels,
                model.vision_model.config.image_size,
                model.vision_model.config.image_size,
            ]		)
        __a     				=  ids_tensor([batch_size, 4]  ,						model.text_model.config.vocab_size		)
        __a     				=  random_attention_mask([batch_size, 4]		)
        __a     				=  {"pixel_values": pixel_values, "input_ids": input_ids, "attention_mask": attention_mask}

        return model, inputs


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase=None  ,						**lowerCamelCase		):
        __a				, __a     				=  self.get_vision_text_model(lowerCamelCase  ,						lowerCamelCase		)
        __a     				=  TFVisionTextDualEncoderModel(vision_model=lowerCamelCase  ,						text_model=lowerCamelCase		)

        __a     				=  model(
            input_ids=lowerCamelCase  ,						pixel_values=lowerCamelCase  ,						attention_mask=lowerCamelCase  ,						output_attentions=lowerCamelCase		)

        __a     				=  output.vision_model_output.attentions
        self.assertEqual(len(lowerCamelCase		)  ,						vision_config.num_hidden_layers		)

        # in DEiT, the seq_len equals the number of patches + 2 (we add 2 for the [CLS] and distillation tokens)
        __a     				=  to_atuple(vision_model.config.image_size		)
        __a     				=  to_atuple(vision_model.config.patch_size		)
        __a     				=  (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])
        __a     				=  num_patches + 2
        self.assertEqual(vision_attentions[0].shape[-3:]  ,						(vision_config.num_attention_heads, seq_len, seq_len)		)

        __a     				=  output.text_model_output.attentions
        self.assertEqual(len(lowerCamelCase		)  ,						text_config.num_hidden_layers		)

        self.assertEqual(
            text_attentions[0].shape[-3:]  ,						(text_config.num_attention_heads, input_ids.shape[-1], input_ids.shape[-1])  ,						)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase		):
        __a     				=  TFDeiTModel(lowerCamelCase  ,						name="vision_model"		)
        __a     				=  TFRobertaModel(lowerCamelCase  ,						name="text_model"		)
        return vision_model, text_model






 def 		a__ (				self		):
        __a     				=  TFDeiTModelTester(self		)
        __a     				=  TFRobertaModelTester(self		)
        __a     				=  vit_model_tester.prepare_config_and_inputs()
        __a     				=  bert_model_tester.prepare_config_and_inputs()

        __a				, __a				, __a     				=  vision_config_and_inputs

        (
            (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				,  
        )     				=  text_config_and_inputs

        return {
            "text_config": text_config,
            "vision_config": vision_config,
            "pixel_values": pixel_values,
            "attention_mask": input_mask,
            "input_ids": input_ids,
            "text_token_type_ids": token_type_ids,
            "text_sequence_labels": sequence_labels,
            "text_token_labels": token_labels,
            "text_choice_labels": choice_labels,
        }






@require_tf
class 				snake_case__     (   snake_case_,		unittest.TestCase   ):


 def 		a__ (				self		):
        __a     				=  TFVisionTextDualEncoderModel.from_vision_text_pretrained(
            "Rocketknight1/tiny-random-clip-tf"  ,						"hf-internal-testing/tiny-random-bert"		)
        __a     				=  13
        __a     				=  floats_tensor(
            [
                batch_size,
                model.vision_model.config.num_channels,
                model.vision_model.config.image_size,
                model.vision_model.config.image_size,
            ]		)
        __a     				=  ids_tensor([batch_size, 4]  ,						model.text_model.config.vocab_size		)
        __a     				=  random_attention_mask([batch_size, 4]		)
        __a     				=  {"pixel_values": pixel_values, "input_ids": input_ids, "attention_mask": attention_mask}

        return model, inputs


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase		):
        __a     				=  TFCLIPVisionModel(lowerCamelCase  ,						name="vision_model"		)
        __a     				=  TFBertModel(lowerCamelCase  ,						name="text_model"		)
        return vision_model, text_model






 def 		a__ (				self		):
        __a     				=  TFCLIPVisionModelTester(self		)
        __a     				=  TFBertModelTester(self		)
        __a     				=  clip_model_tester.prepare_config_and_inputs()
        __a     				=  bert_model_tester.prepare_config_and_inputs()

        __a				, __a     				=  vision_config_and_inputs

        (
            (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				, (
            __a 
        )				,  
        )     				=  text_config_and_inputs

        return {
            "text_config": text_config,
            "vision_config": vision_config,
            "pixel_values": pixel_values,
            "attention_mask": input_mask,
            "input_ids": input_ids,
            "text_token_type_ids": token_type_ids,
            "text_sequence_labels": sequence_labels,
            "text_token_labels": token_labels,
            "text_choice_labels": choice_labels,
        }






@require_vision
@require_tf
class 				snake_case__     (   unittest.TestCase   ):


 @slow
 def 		a__ (				self		):
        __a     				=  TFVisionTextDualEncoderModel.from_pretrained(
            "clip-italian/clip-italian"  ,						logit_scale_init_value=1.0  ,						from_pt=lowerCamelCase		)
        __a     				=  VisionTextDualEncoderProcessor.from_pretrained("clip-italian/clip-italian"		)

        __a     				=  Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"		)
        __a     				=  processor(
            text=["una foto di un gatto", "una foto di un cane"]  ,						images=lowerCamelCase  ,						padding=lowerCamelCase  ,						return_tensors="np"		)

        __a     				=  model(**lowerCamelCase		)

        # verify the logits
        self.assertEqual(outputs.logits_per_image.shape  ,						(inputs.pixel_values.shape[0], inputs.input_ids.shape[0])		)
        self.assertEqual(
            outputs.logits_per_text.shape  ,						(inputs.input_ids.shape[0], inputs.pixel_values.shape[0])  ,						)

        __a     				=  np.array([[1.228_4727, 0.310_4122]]		)

        self.assertTrue(np.allclose(outputs.logits_per_image.numpy()  ,						lowerCamelCase  ,						atol=1E-3		)		)

"""simple docstring"""



from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
   import PIL


SCREAMING_SNAKE_CASE__:Tuple			=	logging.get_logger(__name__)






class 				snake_case__     (   snake_case_   ):
 _snake_case   :			Optional[Any]           =		["""pixel_values"""]


 def __init__(				self  ,						lowerCamelCase = True  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = PILImageResampling.BILINEAR  ,						lowerCamelCase = True  ,						lowerCamelCase = 1 / 255  ,						lowerCamelCase = True  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						**lowerCamelCase  ,						):
        super().__init__(**lowerCamelCase		)
        __a     				=  size if size is not None else {"shortest_edge": 384}
        __a     				=  get_size_dict(lowerCamelCase  ,						default_to_square=lowerCamelCase		)

        __a     				=  do_resize
        __a     				=  size
        # Default value set here for backwards compatibility where the value in config is None
        __a     				=  crop_pct if crop_pct is not None else 224 / 256
        __a     				=  resample
        __a     				=  do_rescale
        __a     				=  rescale_factor
        __a     				=  do_normalize
        __a     				=  image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
        __a     				=  image_std if image_std is not None else IMAGENET_STANDARD_STD


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase = PILImageResampling.BICUBIC  ,						lowerCamelCase = None  ,						**lowerCamelCase  ,						):
        __a     				=  get_size_dict(lowerCamelCase  ,						default_to_square=lowerCamelCase		)
        if "shortest_edge" not in size:
               raise ValueError(F"Size dictionary must contain 'shortest_edge' key. Got {size.keys()}"		)
        __a     				=  size["shortest_edge"]

        if shortest_edge < 384:
               # maintain same ratio, resizing shortest edge to shortest_edge/crop_pct
               __a     				=  int(shortest_edge / crop_pct		)
               __a     				=  get_resize_output_image_size(lowerCamelCase  ,						size=lowerCamelCase  ,						default_to_square=lowerCamelCase		)
               __a     				=  resize(image=lowerCamelCase  ,						size=lowerCamelCase  ,						resample=lowerCamelCase  ,						data_format=lowerCamelCase  ,						**lowerCamelCase		)
               # then crop to (shortest_edge, shortest_edge)
               return center_crop(image=lowerCamelCase  ,						size=(shortest_edge, shortest_edge)  ,						data_format=lowerCamelCase  ,						**lowerCamelCase		)
        else:
               # warping (no cropping) when evaluated at 384 or larger
               return resize(
                   lowerCamelCase  ,						size=(shortest_edge, shortest_edge)  ,						resample=lowerCamelCase  ,						data_format=lowerCamelCase  ,						**lowerCamelCase		)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase = None  ,						**lowerCamelCase  ,						):
        return rescale(lowerCamelCase  ,						scale=lowerCamelCase  ,						data_format=lowerCamelCase  ,						**lowerCamelCase		)


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase  ,						lowerCamelCase = None  ,						**lowerCamelCase  ,						):
        return normalize(lowerCamelCase  ,						mean=lowerCamelCase  ,						std=lowerCamelCase  ,						data_format=lowerCamelCase  ,						**lowerCamelCase		)






 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = ChannelDimension.FIRST  ,						**lowerCamelCase  ,						):
        __a     				=  do_resize if do_resize is not None else self.do_resize
        __a     				=  crop_pct if crop_pct is not None else self.crop_pct
        __a     				=  resample if resample is not None else self.resample
        __a     				=  do_rescale if do_rescale is not None else self.do_rescale
        __a     				=  rescale_factor if rescale_factor is not None else self.rescale_factor
        __a     				=  do_normalize if do_normalize is not None else self.do_normalize
        __a     				=  image_mean if image_mean is not None else self.image_mean
        __a     				=  image_std if image_std is not None else self.image_std

        __a     				=  size if size is not None else self.size
        __a     				=  get_size_dict(lowerCamelCase  ,						default_to_square=lowerCamelCase		)

        __a     				=  make_list_of_images(lowerCamelCase		)

        if not valid_images(lowerCamelCase		):
               raise ValueError(
                   "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
                   "torch.Tensor, tf.Tensor or jax.ndarray."		)

        if do_resize and size is None or resample is None:
               raise ValueError("Size and resample must be specified if do_resize is True."		)

        if do_resize and size["shortest_edge"] < 384 and crop_pct is None:
               raise ValueError("crop_pct must be specified if size < 384."		)

        if do_rescale and rescale_factor is None:
               raise ValueError("Rescale factor must be specified if do_rescale is True."		)

        if do_normalize and (image_mean is None or image_std is None):
               raise ValueError("Image mean and std must be specified if do_normalize is True."		)

        # All transformations expect numpy arrays.
        __a     				=  [to_numpy_array(lowerCamelCase		) for image in images]

        if do_resize:
               __a     				=  [self.resize(image=lowerCamelCase  ,						size=lowerCamelCase  ,						crop_pct=lowerCamelCase  ,						resample=lowerCamelCase		) for image in images]

        if do_rescale:
               __a     				=  [self.rescale(image=lowerCamelCase  ,						scale=lowerCamelCase		) for image in images]

        if do_normalize:
               __a     				=  [self.normalize(image=lowerCamelCase  ,						mean=lowerCamelCase  ,						std=lowerCamelCase		) for image in images]

        __a     				=  [to_channel_dimension_format(lowerCamelCase  ,						lowerCamelCase		) for image in images]

        __a     				=  {"pixel_values": images}
        return BatchFeature(data=lowerCamelCase  ,						tensor_type=lowerCamelCase		)
