import random
import timeit
from functools import wraps
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_tf_auto import TF_MODEL_MAPPING, TF_MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_tf_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_tf_available():
		import tensorflow as tf
		from tensorflow.python.framework.errors_impl import ResourceExhaustedError

		from .benchmark_args_tf import TensorFlowBenchmarkArguments

if is_pyanvml_available():
		import pyanvml.pyanvml as nvml

__lowerCamelCase					:	Any       =  logging.get_logger(__name__)




def       _snake_case	(   lowerCAmelCase							:   bool   , lowerCAmelCase							:   bool	):
		"""simple docstring"""







		def run_func(lowerCAmelCase							:   int	):
				@wraps(lowerCAmelCase	)
				def run_in_eager_mode(*lowerCAmelCase							:   Tuple   , **lowerCAmelCase							:   Any	):
						return func(*lowerCAmelCase   , **lowerCAmelCase	)

				@wraps(lowerCAmelCase	)
				@tf.function(experimental_compile=lowerCAmelCase	)
				def run_in_graph_mode(*lowerCAmelCase							:   Union[str, Any]   , **lowerCAmelCase							:   List[str]	):
						return func(*lowerCAmelCase   , **lowerCAmelCase	)

				if do_eager_mode is True:
						if use_xla is not False:
								raise ValueError(
								    "Cannot run model in XLA, if `args.eager_mode` is set to `True`. Please set `args.eager_mode=False`."	)
						return run_in_eager_mode
				else:
						return run_in_graph_mode

		return run_func




def       _snake_case	(   lowerCAmelCase							:   int   , lowerCAmelCase							:   int   , lowerCAmelCase							:   int	):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		List[Any]				      =    random.Random()
		SCREAMING_SNAKE_CASE_  :		str				      =    [rng.randint(0   , vocab_size - 1	) for i in range(batch_size * sequence_length	)]
		return tf.constant(lowerCAmelCase   , shape=(batch_size, sequence_length)   , dtype=tf.intaa	)







class 		a__						(					A__					):
					A	       =							42
					A	       =							42
					A	       =							"TensorFlow"

					@property
					def 				__UpperCamelCase   (    self      :       str    ):







							"""simple docstring"""






							return tf.__version__

					def 				__UpperCamelCase   (    self      :       int,_A      :       str,_A      :       int,_A      :       int    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		List[str]				      =    self.args.strategy
							if strategy is None:
									raise ValueError("A device strategy has to be initialized before using TensorFlow."    )
							SCREAMING_SNAKE_CASE_  :		Any				      =    self._prepare_inference_func(_A,_A,_A    )
							return self._measure_speed(_inference    )

					def 				__UpperCamelCase   (    self      :       str,_A      :       str,_A      :       int,_A      :       int    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		List[Any]				      =    self.args.strategy
							if strategy is None:
									raise ValueError("A device strategy has to be initialized before using TensorFlow."    )
							SCREAMING_SNAKE_CASE_  :		List[Any]				      =    self._prepare_train_func(_A,_A,_A    )
							return self._measure_speed(_train    )

					def 				__UpperCamelCase   (    self      :       Dict,_A      :       str,_A      :       int,_A      :       int    ):







							"""simple docstring"""






							if self.args.is_gpu:
									tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx],_A    )
							SCREAMING_SNAKE_CASE_  :		Tuple				      =    self.args.strategy
							if strategy is None:
									raise ValueError("A device strategy has to be initialized before using TensorFlow."    )
							SCREAMING_SNAKE_CASE_  :		str				      =    self._prepare_inference_func(_A,_A,_A    )
							return self._measure_memory(_inference    )

					def 				__UpperCamelCase   (    self      :       Optional[Any],_A      :       str,_A      :       int,_A      :       int    ):







							"""simple docstring"""






							if self.args.is_gpu:
									tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx],_A    )
							SCREAMING_SNAKE_CASE_  :		Any				      =    self.args.strategy
							if strategy is None:
									raise ValueError("A device strategy has to be initialized before using TensorFlow."    )

							SCREAMING_SNAKE_CASE_  :		str				      =    self._prepare_train_func(_A,_A,_A    )
							return self._measure_memory(_train    )

					def 				__UpperCamelCase   (    self      :       int,_A      :       str,_A      :       int,_A      :       int    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    self.config_dict[model_name]

							if self.args.fpaa:
									raise NotImplementedError("Mixed precision is currently not supported."    )

							SCREAMING_SNAKE_CASE_  :		Dict				      =    (
							    hasattr(_A,"architectures"    )
							    and isinstance(config.architectures,_A    )
							    and len(config.architectures    ) > 0
							)
							if not self.args.only_pretrain_model and has_model_class_in_config:
									try:
											SCREAMING_SNAKE_CASE_  :		List[Any]				      =    "TF" + config.architectures[0]  # prepend 'TF' for tensorflow model
											SCREAMING_SNAKE_CASE_  :		List[Any]				      =    __import__("transformers",fromlist=[model_class]    )
											SCREAMING_SNAKE_CASE_  :		Tuple				      =    getattr(_A,_A    )
											SCREAMING_SNAKE_CASE_  :		List[str]				      =    model_cls(_A    )
									except ImportError:
											raise ImportError(
											    F'{model_class} does not exist. If you just want to test the pretrained model, you might want to'
											    " set `--only_pretrain_model` or `args.only_pretrain_model=True`."    )
							else:
									SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    TF_MODEL_MAPPING[config.__class__](_A    )

							# encoder-decoder has vocab size saved differently
							SCREAMING_SNAKE_CASE_  :		Tuple				      =    config.vocab_size if hasattr(_A,"vocab_size"    ) else config.encoder.vocab_size
							SCREAMING_SNAKE_CASE_  :		str				      =    random_input_ids(_A,_A,_A    )

							@run_with_tf_optimizations(self.args.eager_mode,self.args.use_xla    )
							def encoder_decoder_forward():
									return model(_A,decoder_input_ids=_A,training=_A    )

							@run_with_tf_optimizations(self.args.eager_mode,self.args.use_xla    )
							def encoder_forward():
									return model(_A,training=_A    )

							SCREAMING_SNAKE_CASE_  :		Tuple				      =    encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

							return _inference

					def 				__UpperCamelCase   (    self      :       Dict,_A      :       str,_A      :       int,_A      :       int    ):







							"""simple docstring"""






							SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    self.config_dict[model_name]

							if self.args.eager_mode is not False:
									raise ValueError("Training cannot be done in eager mode. Please make sure that `args.eager_mode = False`."    )

							if self.args.fpaa:
									raise NotImplementedError("Mixed precision is currently not supported."    )

							SCREAMING_SNAKE_CASE_  :		Dict				      =    (
							    hasattr(_A,"architectures"    )
							    and isinstance(config.architectures,_A    )
							    and len(config.architectures    ) > 0
							)
							if not self.args.only_pretrain_model and has_model_class_in_config:
									try:
											SCREAMING_SNAKE_CASE_  :		Dict				      =    "TF" + config.architectures[0]  # prepend 'TF' for tensorflow model
											SCREAMING_SNAKE_CASE_  :		Tuple				      =    __import__("transformers",fromlist=[model_class]    )
											SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    getattr(_A,_A    )
											SCREAMING_SNAKE_CASE_  :		List[Any]				      =    model_cls(_A    )
									except ImportError:
											raise ImportError(
											    F'{model_class} does not exist. If you just want to test the pretrained model, you might want to'
											    " set `--only_pretrain_model` or `args.only_pretrain_model=True`."    )
							else:
									SCREAMING_SNAKE_CASE_  :		Any				      =    TF_MODEL_WITH_LM_HEAD_MAPPING[config.__class__](_A    )

							# encoder-decoder has vocab size saved differently
							SCREAMING_SNAKE_CASE_  :		Any				      =    config.vocab_size if hasattr(_A,"vocab_size"    ) else config.encoder.vocab_size
							SCREAMING_SNAKE_CASE_  :		int				      =    random_input_ids(_A,_A,_A    )

							@run_with_tf_optimizations(self.args.eager_mode,self.args.use_xla    )
							def encoder_decoder_train():
									SCREAMING_SNAKE_CASE_  :		Dict				      =    model(_A,decoder_input_ids=_A,labels=_A,training=_A    )[0]
									SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    tf.gradients(_A,model.trainable_variables    )
									return gradients

							@run_with_tf_optimizations(self.args.eager_mode,self.args.use_xla    )
							def encoder_train():
									SCREAMING_SNAKE_CASE_  :		Tuple				      =    model(_A,labels=_A,training=_A    )[0]
									SCREAMING_SNAKE_CASE_  :		Tuple				      =    tf.gradients(_A,model.trainable_variables    )
									return gradients

							SCREAMING_SNAKE_CASE_  :		List[Any]				      =    encoder_decoder_train if config.is_encoder_decoder else encoder_train

							return _train

					def 				__UpperCamelCase   (    self      :       int,_A      :       Any    ):







							"""simple docstring"""






							with self.args.strategy.scope():
									try:
											if self.args.is_tpu or self.args.use_xla:
													# run additional 10 times to stabilize compilation for tpu
													logger.info("Do inference on TPU. Running model 5 times to stabilize compilation"    )
													timeit.repeat(_A,repeat=1,number=5    )

											# as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
											SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    timeit.repeat(
											    _A,repeat=self.args.repeat,number=10,)

											return min(_A    ) / 10.0
									except ResourceExhaustedError as e:
											self.print_fn(F'Doesn\'t fit on GPU. {e}'    )






					def 				__UpperCamelCase   (    self      :       List[str],_A      :       Callable[[], None]    ):







							"""simple docstring"""






							logger.info(
							    "Note that TensorFlow allocates more memory than "
							    "it might need to speed up computation. "
							    "The memory reported here corresponds to the memory "
							    "reported by `nvidia-smi`, which can vary depending "
							    "on total available memory on the GPU that is used."    )
							with self.args.strategy.scope():
									try:
											if self.args.trace_memory_line_by_line:
													if not self.args.eager_mode:
															raise ValueError(
															    "`args.eager_mode` is set to `False`. Make sure to run model in eager mode to measure memory"
															    " consumption line by line."    )
													SCREAMING_SNAKE_CASE_  :		List[str]				      =    start_memory_tracing("transformers"    )

											if self.args.is_tpu:
													# tpu
													raise NotImplementedError(
													    "Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking"
													    " with `args.memory=False`"    )
											elif self.args.is_gpu:
													# gpu
													if not is_pyanvml_available():
															logger.warning(
															    "py3nvml not installed, we won't log GPU memory usage. "
															    "Install py3nvml (pip install py3nvml) to log information about GPU."    )
															SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    "N/A"
													else:
															logger.info(
															    "Measuring total GPU usage on GPU device. Make sure to not have additional processes"
															    " running on the same GPU."    )
															# init nvml
															nvml.nvmlInit()
															func()
															SCREAMING_SNAKE_CASE_  :		str				      =    nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx    )
															SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    nvml.nvmlDeviceGetMemoryInfo(_A    )
															SCREAMING_SNAKE_CASE_  :		Any				      =    meminfo.used
															SCREAMING_SNAKE_CASE_  :		List[Any]				      =    Memory(_A    )
															# shutdown nvml
															nvml.nvmlShutdown()
											else:
													# cpu
													if self.args.trace_memory_line_by_line:
															logger.info(
															    "When enabling line by line tracing, the max peak memory for CPU is inaccurate in"
															    " TensorFlow."    )
															SCREAMING_SNAKE_CASE_  :		Union[str, Any]				      =    None
													else:
															SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    measure_peak_memory_cpu(_A    )
															SCREAMING_SNAKE_CASE_  :		str				      =    Memory(_A    ) if isinstance(_A,_A    ) else memory_bytes
											if self.args.trace_memory_line_by_line:
													SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    stop_memory_tracing(_A    )
													if memory is None:
															SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    summary.total
											else:
													SCREAMING_SNAKE_CASE_  :		Optional[int]				      =    None

											return memory, summary
									except ResourceExhaustedError as e:
											self.print_fn(F'Doesn\'t fit on GPU. {e}'    )
											return "N/A", None




"""simple docstring"""


import json
import os
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 FEATURE_EXTRACTOR_NAME, is_vision_available


if is_vision_available():
	from PIL import Image

	from transformers import ChineseCLIPImageProcessor, ChineseCLIPProcessor
@require_vision
class   _UpperCamelCase  (   unittest.TestCase		):
							'''simple docstring'''







							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			tempfile.mkdtemp()

										__lowerCAmelCase				    =			[
										    "[UNK]",
										    "[CLS]",
										    "[SEP]",
										    "[PAD]",
										    "[MASK]",
										    "的",
										    "价",
										    "格",
										    "是",
										    "15",
										    "便",
										    "alex",
										    "##andra",
										    "，",
										    "。",
										    "-",
										    "t",
										    "shirt",
										]
										__lowerCAmelCase				    =			os.path.join(self.tmpdirname    ,   VOCAB_FILES_NAMES["vocab_file"]	)
										with open(self.vocab_file    ,   "w"    ,   encoding="utf-8"	) as vocab_writer:
													vocab_writer.write("".join([x + "\n" for x in vocab_tokens]	)	)

										__lowerCAmelCase				    =			{
										    "do_resize": True,
										    "size": {"height": 2_24, "width": 2_24},
										    "do_center_crop": True,
										    "crop_size": {"height": 18, "width": 18},
										    "do_normalize": True,
										    "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],
										    "do_convert_rgb": True,
										}
										__lowerCAmelCase				    =			os.path.join(self.tmpdirname    ,   __a	)
										with open(self.image_processor_file    ,   "w"    ,   encoding="utf-8"	) as fp:
													json.dump(__a    ,   __a	)

							def 			snake_case				(     self    ,   **__a	):
										return BertTokenizer.from_pretrained(self.tmpdirname    ,   **__a	)

							def 			snake_case				(     self    ,   **__a	):
										return BertTokenizerFast.from_pretrained(self.tmpdirname    ,   **__a	)

							def 			snake_case				(     self    ,   **__a	):
										return ChineseCLIPImageProcessor.from_pretrained(self.tmpdirname    ,   **__a	)

							def 			snake_case				(     self	):
										shutil.rmtree(self.tmpdirname	)

							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			[np.random.randint(2_55    ,   size=(3, 30, 4_00)    ,   dtype=np.uinta	)]

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

										return image_inputs

							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			self.get_tokenizer()
										__lowerCAmelCase				    =			self.get_rust_tokenizer()
										__lowerCAmelCase				    =			self.get_image_processor()

										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=__a    ,   image_processor=__a	)
										processor_slow.save_pretrained(self.tmpdirname	)
										__lowerCAmelCase				    =			ChineseCLIPProcessor.from_pretrained(self.tmpdirname    ,   use_fast=__a	)

										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=__a    ,   image_processor=__a	)
										processor_fast.save_pretrained(self.tmpdirname	)
										__lowerCAmelCase				    =			ChineseCLIPProcessor.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    ,   __a	)
										self.assertIsInstance(processor_fast.tokenizer    ,   __a	)

										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    ,   __a	)
										self.assertIsInstance(processor_fast.image_processor    ,   __a	)

							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=self.get_tokenizer()    ,   image_processor=self.get_image_processor()	)
										processor.save_pretrained(self.tmpdirname	)

										__lowerCAmelCase				    =			self.get_tokenizer(cls_token="(CLS)"    ,   sep_token="(SEP)"	)
										__lowerCAmelCase				    =			self.get_image_processor(do_normalize=__a	)

										__lowerCAmelCase				    =			ChineseCLIPProcessor.from_pretrained(
										    self.tmpdirname    ,   cls_token="(CLS)"    ,   sep_token="(SEP)"    ,   do_normalize=__a	)

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

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

							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			self.get_image_processor()
										__lowerCAmelCase				    =			self.get_tokenizer()

										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=__a    ,   image_processor=__a	)

										__lowerCAmelCase				    =			self.prepare_image_inputs()

										__lowerCAmelCase				    =			image_processor(__a    ,   return_tensors="np"	)
										__lowerCAmelCase				    =			processor(images=__a    ,   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 			snake_case				(     self	):
										__lowerCAmelCase				    =			self.get_image_processor()
										__lowerCAmelCase				    =			self.get_tokenizer()

										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=__a    ,   image_processor=__a	)

										__lowerCAmelCase				    =			"Alexandra，T-shirt的价格是15便士。"

										__lowerCAmelCase				    =			processor(text=__a	)

										__lowerCAmelCase				    =			tokenizer(__a	)

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

							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			self.get_image_processor()
										__lowerCAmelCase				    =			self.get_tokenizer()

										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=__a    ,   image_processor=__a	)

										__lowerCAmelCase				    =			"Alexandra，T-shirt的价格是15便士。"
										__lowerCAmelCase				    =			self.prepare_image_inputs()

										__lowerCAmelCase				    =			processor(text=__a    ,   images=__a	)

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

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

							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			self.get_image_processor()
										__lowerCAmelCase				    =			self.get_tokenizer()

										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=__a    ,   image_processor=__a	)

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

										__lowerCAmelCase				    =			processor.batch_decode(__a	)
										__lowerCAmelCase				    =			tokenizer.batch_decode(__a	)

										self.assertListEqual(__a    ,   __a	)



							def 			snake_case				(     self	):
										__lowerCAmelCase				    =			self.get_image_processor()
										__lowerCAmelCase				    =			self.get_tokenizer()

										__lowerCAmelCase				    =			ChineseCLIPProcessor(tokenizer=__a    ,   image_processor=__a	)

										__lowerCAmelCase				    =			"Alexandra，T-shirt的价格是15便士。"
										__lowerCAmelCase				    =			self.prepare_image_inputs()

										__lowerCAmelCase				    =			processor(text=__a    ,   images=__a	)

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




