
'''simple docstring'''



import json
import os
import random
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import BertTokenizer, BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
     from PIL import Image

     from transformers import FlavaImageProcessor, FlavaProcessor
     from transformers.models.flava.image_processing_flava import (
         FLAVA_CODEBOOK_MEAN,
         FLAVA_CODEBOOK_STD,
         FLAVA_IMAGE_MEAN,
         FLAVA_IMAGE_STD,
     )

@require_vision
class       _a (unittest.TestCase):





       """simple docstring"""




       def 				UpperCamelCase (							self					)			->	str:
          _SCREAMING_SNAKE_CASE          = tempfile.mkdtemp()

          # fmt: off
          _SCREAMING_SNAKE_CASE          = ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "want", "##want", "##ed", "wa", "un", "runn", "##ing", ",", "low", "lowest"]
          # fmt: on
          _SCREAMING_SNAKE_CASE          = os.path.join(self.tmpdirname				,							VOCAB_FILES_NAMES["""vocab_file"""]					)

          with open(self.vocab_file				,							"""w"""				,							encoding="""utf-8"""					) as fp:
             fp.write("""""".join([x + """\n""" for x in vocab_tokens]					)					)

          _SCREAMING_SNAKE_CASE          = {
              "image_mean": FLAVA_IMAGE_MEAN,
              "image_std": FLAVA_IMAGE_STD,
              "do_normalize": True,
              "do_resize": True,
              "size": 2_24,
              "do_center_crop": True,
              "crop_size": 2_24,
              "input_size_patches": 14,
              "total_mask_patches": 75,
              "mask_group_max_patches": None,
              "mask_group_min_patches": 16,
              "mask_group_min_aspect_ratio": 0.3,
              "mask_group_max_aspect_ratio": None,
              "codebook_do_resize": True,
              "codebook_size": 1_12,
              "codebook_do_center_crop": True,
              "codebook_crop_size": 1_12,
              "codebook_do_map_pixels": True,
              "codebook_do_normalize": True,
              "codebook_image_mean": FLAVA_CODEBOOK_MEAN,
              "codebook_image_std": FLAVA_CODEBOOK_STD,
          }

          _SCREAMING_SNAKE_CASE          = os.path.join(self.tmpdirname				,							__lowerCamelCase					)
          with open(self.image_processor_file				,							"""w"""				,							encoding="""utf-8"""					) as fp:
             json.dump(__lowerCamelCase				,							__lowerCamelCase					)






       def 				UpperCamelCase (							self				,							**A__					)			->	List[str]:
          return BertTokenizer.from_pretrained(self.tmpdirname				,							**__lowerCamelCase					)






       def 				UpperCamelCase (							self				,							**A__					)			->	Optional[Any]:
          return BertTokenizerFast.from_pretrained(self.tmpdirname				,							**__lowerCamelCase					)






       def 				UpperCamelCase (							self				,							**A__					)			->	Tuple:
          return FlavaImageProcessor.from_pretrained(self.tmpdirname				,							**__lowerCamelCase					)






       def 				UpperCamelCase (							self					)			->	int:
          shutil.rmtree(self.tmpdirname					)






       def 				UpperCamelCase (							self					)			->	Optional[int]:
          _SCREAMING_SNAKE_CASE          = [np.random.randint(2_55				,							size=(3, 30, 4_00)				,							dtype=np.uinta					)]

          _SCREAMING_SNAKE_CASE          = [Image.fromarray(np.moveaxis(__lowerCamelCase				,							0				,							-1					)					) for x in image_inputs]

          return image_inputs






       def 				UpperCamelCase (							self					)			->	Any:
          _SCREAMING_SNAKE_CASE          = self.get_tokenizer()
          _SCREAMING_SNAKE_CASE          = self.get_rust_tokenizer()
          _SCREAMING_SNAKE_CASE          = self.get_image_processor()

          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=__lowerCamelCase				,							image_processor=__lowerCamelCase					)
          processor_slow.save_pretrained(self.tmpdirname					)
          _SCREAMING_SNAKE_CASE          = FlavaProcessor.from_pretrained(self.tmpdirname				,							use_fast=__lowerCamelCase					)

          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=__lowerCamelCase				,							image_processor=__lowerCamelCase					)
          processor_fast.save_pretrained(self.tmpdirname					)
          _SCREAMING_SNAKE_CASE          = FlavaProcessor.from_pretrained(self.tmpdirname					)

          self.assertEqual(processor_slow.tokenizer.get_vocab()				,							tokenizer_slow.get_vocab()					)
          self.assertEqual(processor_fast.tokenizer.get_vocab()				,							tokenizer_fast.get_vocab()					)
          self.assertEqual(tokenizer_slow.get_vocab()				,							tokenizer_fast.get_vocab()					)
          self.assertIsInstance(processor_slow.tokenizer				,							__lowerCamelCase					)
          self.assertIsInstance(processor_fast.tokenizer				,							__lowerCamelCase					)

          self.assertEqual(processor_slow.image_processor.to_json_string()				,							image_processor.to_json_string()					)
          self.assertEqual(processor_fast.image_processor.to_json_string()				,							image_processor.to_json_string()					)
          self.assertIsInstance(processor_slow.image_processor				,							__lowerCamelCase					)
          self.assertIsInstance(processor_fast.image_processor				,							__lowerCamelCase					)






       def 				UpperCamelCase (							self					)			->	Optional[int]:
          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=self.get_tokenizer()				,							image_processor=self.get_image_processor()					)
          processor.save_pretrained(self.tmpdirname					)

          _SCREAMING_SNAKE_CASE          = self.get_tokenizer(bos_token="""(BOS)"""				,							eos_token="""(EOS)"""					)
          _SCREAMING_SNAKE_CASE          = self.get_image_processor(do_normalize=__lowerCamelCase				,							padding_value=1.0					)

          _SCREAMING_SNAKE_CASE          = FlavaProcessor.from_pretrained(
              self.tmpdirname				,							bos_token="""(BOS)"""				,							eos_token="""(EOS)"""				,							do_normalize=__lowerCamelCase				,							padding_value=1.0					)

          self.assertEqual(processor.tokenizer.get_vocab()				,							tokenizer_add_kwargs.get_vocab()					)
          self.assertIsInstance(processor.tokenizer				,							__lowerCamelCase					)

          self.assertEqual(processor.image_processor.to_json_string()				,							image_processor_add_kwargs.to_json_string()					)
          self.assertIsInstance(processor.image_processor				,							__lowerCamelCase					)






       def 				UpperCamelCase (							self					)			->	Any:
          _SCREAMING_SNAKE_CASE          = self.get_image_processor()
          _SCREAMING_SNAKE_CASE          = self.get_tokenizer()

          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=__lowerCamelCase				,							image_processor=__lowerCamelCase					)

          _SCREAMING_SNAKE_CASE          = self.prepare_image_inputs()

          _SCREAMING_SNAKE_CASE          = image_processor(__lowerCamelCase				,							return_tensors="""np"""					)
          _SCREAMING_SNAKE_CASE          = processor(images=__lowerCamelCase				,							return_tensors="""np"""					)

          for key in input_feat_extract.keys():
             self.assertAlmostEqual(input_feat_extract[key].sum()				,							input_processor[key].sum()				,							delta=1E-2					)

          # With rest of the args
          random.seed(12_34					)
          _SCREAMING_SNAKE_CASE          = image_processor(
              __lowerCamelCase				,							return_image_mask=__lowerCamelCase				,							return_codebook_pixels=__lowerCamelCase				,							return_tensors="""np"""					)
          random.seed(12_34					)
          _SCREAMING_SNAKE_CASE          = processor(
              images=__lowerCamelCase				,							return_image_mask=__lowerCamelCase				,							return_codebook_pixels=__lowerCamelCase				,							return_tensors="""np"""					)

          for key in input_feat_extract.keys():
             self.assertAlmostEqual(input_feat_extract[key].sum()				,							input_processor[key].sum()				,							delta=1E-2					)






       def 				UpperCamelCase (							self					)			->	int:
          _SCREAMING_SNAKE_CASE          = self.get_image_processor()
          _SCREAMING_SNAKE_CASE          = self.get_tokenizer()

          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=__lowerCamelCase				,							image_processor=__lowerCamelCase					)

          _SCREAMING_SNAKE_CASE          = "lower newer"

          _SCREAMING_SNAKE_CASE          = processor(text=__lowerCamelCase					)

          _SCREAMING_SNAKE_CASE          = tokenizer(__lowerCamelCase					)

          for key in encoded_tok.keys():
             self.assertListEqual(encoded_tok[key]				,							encoded_processor[key]					)






       def 				UpperCamelCase (							self					)			->	Tuple:
          _SCREAMING_SNAKE_CASE          = self.get_image_processor()
          _SCREAMING_SNAKE_CASE          = self.get_tokenizer()

          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=__lowerCamelCase				,							image_processor=__lowerCamelCase					)

          _SCREAMING_SNAKE_CASE          = "lower newer"
          _SCREAMING_SNAKE_CASE          = self.prepare_image_inputs()

          _SCREAMING_SNAKE_CASE          = processor(text=__lowerCamelCase				,							images=__lowerCamelCase					)

          self.assertListEqual(list(inputs.keys()					)				,							["""input_ids""", """token_type_ids""", """attention_mask""", """pixel_values"""]					)

          # add extra args
          _SCREAMING_SNAKE_CASE          = processor(text=__lowerCamelCase				,							images=__lowerCamelCase				,							return_codebook_pixels=__lowerCamelCase				,							return_image_mask=__lowerCamelCase					)

          self.assertListEqual(
              list(inputs.keys()					)				,							[
                  """input_ids""",
                  """token_type_ids""",
                  """attention_mask""",
                  """pixel_values""",
                  """codebook_pixel_values""",
                  """bool_masked_pos""",
              ]				,							)

          # test if it raises when no input is passed
          with pytest.raises(__lowerCamelCase					):
             processor()






       def 				UpperCamelCase (							self					)			->	Dict:
          _SCREAMING_SNAKE_CASE          = self.get_image_processor()
          _SCREAMING_SNAKE_CASE          = self.get_tokenizer()

          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=__lowerCamelCase				,							image_processor=__lowerCamelCase					)

          _SCREAMING_SNAKE_CASE          = [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

          _SCREAMING_SNAKE_CASE          = processor.batch_decode(__lowerCamelCase					)
          _SCREAMING_SNAKE_CASE          = tokenizer.batch_decode(__lowerCamelCase					)

          self.assertListEqual(__lowerCamelCase				,							__lowerCamelCase					)






       def 				UpperCamelCase (							self					)			->	Any:
          _SCREAMING_SNAKE_CASE          = self.get_image_processor()
          _SCREAMING_SNAKE_CASE          = self.get_tokenizer()

          _SCREAMING_SNAKE_CASE          = FlavaProcessor(tokenizer=__lowerCamelCase				,							image_processor=__lowerCamelCase					)

          _SCREAMING_SNAKE_CASE          = "lower newer"
          _SCREAMING_SNAKE_CASE          = self.prepare_image_inputs()

          _SCREAMING_SNAKE_CASE          = processor(text=__lowerCamelCase				,							images=__lowerCamelCase					)

          self.assertListEqual(list(inputs.keys()					)				,							processor.model_input_names					)













import warnings
from dataclasses import dataclass, field
from typing import Optional, Tuple

from .training_args import TrainingArguments
from .utils import cached_property, is_tf_available, logging, requires_backends


UpperCamelCase__          =logging.get_logger(__name__)

if is_tf_available():
       import tensorflow as tf





@dataclass
class 	lowerCAmelCase__(       __lowercase			):






       '''simple docstring'''


       __snake_case       	=			'tf'
       __snake_case       	=			field(
           default=__lowercase     ,   metadata={'help': 'Name of TPU'}     ,   )

       __snake_case       	=			field(
           default=__lowercase     ,   metadata={'help': 'Zone of TPU'}     ,   )

       __snake_case       	=			field(
           default=__lowercase     ,   metadata={'help': 'Name of Cloud TPU-enabled project'}     ,   )

       __snake_case       	=			field(
           default=1.0     ,   metadata={'help': 'Power for the Polynomial decay LR scheduler.'}     ,   )

       __snake_case       	=			field(default=__lowercase     ,   metadata={'help': 'Whether to activate the XLA compilation or not'}			)




       @cached_property
       def 			UpperCamelCase_	(					self							)	->   Tuple["tf.distribute.Strategy", int]:
            requires_backends(self			,  ["tf"]							)
            logger.info("Tensorflow: setting up strategy"							)

            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							tf.config.list_physical_devices("GPU"							)

            # Set to float16 at first
            if self.fpaa:
                 tf.keras.mixed_precision.set_global_policy("mixed_float16"							)

            if self.no_cuda:
                 _SCREAMING_SNAKE_CASE				:		List[Any]  			=							tf.distribute.OneDeviceStrategy(device="/cpu:0"							)
            else:
                 try:
                      if self.tpu_name:
                           _SCREAMING_SNAKE_CASE				:		Any  			=							tf.distribute.cluster_resolver.TPUClusterResolver(
                               self.tpu_name			,  zone=self.tpu_zone			,  project=self.gcp_project							)
                      else:
                           _SCREAMING_SNAKE_CASE				:		str  			=							tf.distribute.cluster_resolver.TPUClusterResolver()
                 except ValueError:
                      if self.tpu_name:
                           raise RuntimeError(F"""Couldn't connect to TPU {self.tpu_name}!"""							)
                      else:
                           _SCREAMING_SNAKE_CASE				:		Any  			=							None

                 if tpu:
                      # Set to bfloat16 in case of TPU
                      if self.fpaa:
                           tf.keras.mixed_precision.set_global_policy("mixed_bfloat16"							)

                      tf.config.experimental_connect_to_cluster(__lowerCamelCase							)
                      tf.tpu.experimental.initialize_tpu_system(__lowerCamelCase							)

                      _SCREAMING_SNAKE_CASE				:		List[str]  			=							tf.distribute.TPUStrategy(__lowerCamelCase							)

                 elif len(__lowerCamelCase							) == 0:
                      _SCREAMING_SNAKE_CASE				:		Dict  			=							tf.distribute.OneDeviceStrategy(device="/cpu:0"							)
                 elif len(__lowerCamelCase							) == 1:
                      _SCREAMING_SNAKE_CASE				:		Dict  			=							tf.distribute.OneDeviceStrategy(device="/gpu:0"							)
                 elif len(__lowerCamelCase							) > 1:
                      # If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
                      _SCREAMING_SNAKE_CASE				:		Any  			=							tf.distribute.MirroredStrategy()
                 else:
                      raise ValueError("Cannot find the proper strategy, please check your environment properties."							)

            return strategy




       @property
       def 			UpperCamelCase_	(					self							)	->   "tf.distribute.Strategy":
            requires_backends(self			,  ["tf"]							)
            return self._setup_strategy




       @property
       def 			UpperCamelCase_	(					self							)	->   int:
            requires_backends(self			,  ["tf"]							)
            return self._setup_strategy.num_replicas_in_sync




       @property
       def 			UpperCamelCase_	(					self							)	->   List[Any]:
            return False  # TF Logging is handled by Keras not the Trainer




       @property
       def 			UpperCamelCase_	(					self							)	->   int:
            if self.per_gpu_train_batch_size:
                 logger.warning(
                     "Using deprecated `--per_gpu_train_batch_size` argument which will be removed in a future "
                     "version. Using `--per_device_train_batch_size` is preferred."							)
            _SCREAMING_SNAKE_CASE				:		Tuple  			=							self.per_gpu_train_batch_size or self.per_device_train_batch_size
            return per_device_batch_size * self.n_replicas




       @property
       def 			UpperCamelCase_	(					self							)	->   int:
            if self.per_gpu_eval_batch_size:
                 logger.warning(
                     "Using deprecated `--per_gpu_eval_batch_size` argument which will be removed in a future "
                     "version. Using `--per_device_eval_batch_size` is preferred."							)
            _SCREAMING_SNAKE_CASE				:		int  			=							self.per_gpu_eval_batch_size or self.per_device_eval_batch_size
            return per_device_batch_size * self.n_replicas







       @property
       def 			UpperCamelCase_	(					self							)	->   int:
            requires_backends(self			,  ["tf"]							)
            warnings.warn(
                "The n_gpu argument is deprecated and will be removed in a future version, use n_replicas instead."			,  __lowerCamelCase			,  )
            return self._setup_strategy.num_replicas_in_sync