




"""simple docstring"""







import unittest

import torch

from diffusers import VQModel
from diffusers.utils import floats_tensor, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()






class       SCREAMING_SNAKE_CASE__					(					a__						,				a__						,				unittest.TestCase					):
						_a    						=						VQModel
						_a    						=						"""sample"""


						@property
						def   __lowercase (							self				:			Tuple						,						lowerCAmelCase				:			Dict=(32, 32)					):
										lowerCAmelCase  =				4
										lowerCAmelCase  =				3

										lowerCAmelCase  =				floats_tensor((batch_size, num_channels) + sizes					).to(lowerCAmelCase__					)

										return {"sample": image}


						@property
						def   __lowercase (							self				:			int					):
										return (3, 32, 32)


						@property
						def   __lowercase (							self				:			Dict					):
										return (3, 32, 32)


						def   __lowercase (							self				:			Any					):
										lowerCAmelCase  =				{
										    "block_out_channels": [32, 64],
										    "in_channels": 3,
										    "out_channels": 3,
										    "down_block_types": ["DownEncoderBlock2D", "DownEncoderBlock2D"],
										    "up_block_types": ["UpDecoderBlock2D", "UpDecoderBlock2D"],
										    "latent_channels": 3,
										}
										lowerCAmelCase  =				self.dummy_input
										return init_dict, inputs_dict


						def   __lowercase (							self				:			int					):
										pass


						def   __lowercase (							self				:			str					):
										pass


						def   __lowercase (							self				:			Tuple					):
										lowerCAmelCase  =				VQModel.from_pretrained("""fusing/vqgan-dummy"""						,						output_loading_info=lowerCAmelCase__					)
										self.assertIsNotNone(lowerCAmelCase__					)
										self.assertEqual(len(loading_info["""missing_keys"""]					)						,						0					)

										model.to(lowerCAmelCase__					)
										lowerCAmelCase  =				model(**self.dummy_input					)

										assert image is not None, "Make sure output is not None"




						def   __lowercase (							self				:			str					):
										lowerCAmelCase  =				VQModel.from_pretrained("""fusing/vqgan-dummy"""					)
										model.to(lowerCAmelCase__					).eval()

										torch.manual_seed(0					)
										if torch.cuda.is_available():
														torch.cuda.manual_seed_all(0					)

										lowerCAmelCase  =				torch.randn(1						,						model.config.in_channels						,						model.config.sample_size						,						model.config.sample_size					)
										lowerCAmelCase  =				image.to(lowerCAmelCase__					)
										with torch.no_grad():
														lowerCAmelCase  =				model(lowerCAmelCase__					).sample

										lowerCAmelCase  =				output[0, -1, -3:, -3:].flatten().cpu()
										# fmt: off
										lowerCAmelCase  =				torch.tensor([-0.0153, -0.4044, -0.1880, -0.5161, -0.2418, -0.4072, -0.1612, -0.0633, -0.0143]					)
										# fmt: on
										self.assertTrue(torch.allclose(lowerCAmelCase__						,						lowerCAmelCase__						,						atol=1e-3					)					)











"""simple docstring"""







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

a      =							logging.get_logger(__name__)





def 							lowercase					(snake_case__   : bool      ,       snake_case__   : bool )     ->  Tuple:


				'''simple docstring'''






				def run_func(snake_case__   : Any ):
								@wraps(snake_case__ )
								def run_in_eager_mode(*snake_case__   : Optional[Any]      ,       **snake_case__   : int ):
												return func(*snake_case__      ,       **snake_case__ )

								@wraps(snake_case__ )
								@tf.function(experimental_compile=snake_case__ )
								def run_in_graph_mode(*snake_case__   : int      ,       **snake_case__   : Tuple ):
												return func(*snake_case__      ,       **snake_case__ )

								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 							lowercase					(snake_case__   : int      ,       snake_case__   : int      ,       snake_case__   : int )     ->  ["tf.Tensor"]:


				'''simple docstring'''






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






class       SCREAMING_SNAKE_CASE__					(					_a					):
						_a    						=						42
						_a    						=						42
						_a    						=						"TensorFlow"


						@property
						def   __lowercase (							self				:			Optional[int]					):
										return tf.__version__


						def   __lowercase (							self				:			Optional[Any]						,						lowerCAmelCase				:			str						,						lowerCAmelCase				:			int						,						lowerCAmelCase				:			int					):
										# initialize GPU on separate process
										lowerCAmelCase  =				self.args.strategy
										if strategy is None:
														raise ValueError("""A device strategy has to be initialized before using TensorFlow."""					)
										lowerCAmelCase  =				self._prepare_inference_func(lowerCAmelCase						,						lowerCAmelCase						,						lowerCAmelCase					)
										return self._measure_speed(_inference					)


						def   __lowercase (							self				:			str						,						lowerCAmelCase				:			str						,						lowerCAmelCase				:			int						,						lowerCAmelCase				:			int					):
										lowerCAmelCase  =				self.args.strategy
										if strategy is None:
														raise ValueError("""A device strategy has to be initialized before using TensorFlow."""					)
										lowerCAmelCase  =				self._prepare_train_func(lowerCAmelCase						,						lowerCAmelCase						,						lowerCAmelCase					)
										return self._measure_speed(_train					)


						def   __lowercase (							self				:			str						,						lowerCAmelCase				:			str						,						lowerCAmelCase				:			int						,						lowerCAmelCase				:			int					):
										# initialize GPU on separate process
										if self.args.is_gpu:
														tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]						,						lowerCAmelCase					)
										lowerCAmelCase  =				self.args.strategy
										if strategy is None:
														raise ValueError("""A device strategy has to be initialized before using TensorFlow."""					)
										lowerCAmelCase  =				self._prepare_inference_func(lowerCAmelCase						,						lowerCAmelCase						,						lowerCAmelCase					)
										return self._measure_memory(_inference					)


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

										lowerCAmelCase  =				self._prepare_train_func(lowerCAmelCase						,						lowerCAmelCase						,						lowerCAmelCase					)
										return self._measure_memory(_train					)


						def   __lowercase (							self				:			List[str]						,						lowerCAmelCase				:			str						,						lowerCAmelCase				:			int						,						lowerCAmelCase				:			int					):
										lowerCAmelCase  =				self.config_dict[model_name]

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

										lowerCAmelCase  =				(
										    hasattr(lowerCAmelCase						,						"""architectures"""					)
										    and isinstance(config.architectures						,						lowerCAmelCase					)
										    and len(config.architectures					) > 0
										)
										if not self.args.only_pretrain_model and has_model_class_in_config:
														try:
																		lowerCAmelCase  =				"""TF""" + config.architectures[0]  # prepend 'TF' for tensorflow model
																		lowerCAmelCase  =				__import__("""transformers"""						,						fromlist=[model_class]					)
																		lowerCAmelCase  =				getattr(lowerCAmelCase						,						lowerCAmelCase					)
																		lowerCAmelCase  =				model_cls(lowerCAmelCase					)
														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:
														lowerCAmelCase  =				TF_MODEL_MAPPING[config.__class__](lowerCAmelCase					)

										# encoder-decoder has vocab size saved differently
										lowerCAmelCase  =				config.vocab_size if hasattr(lowerCAmelCase						,						"""vocab_size"""					) else config.encoder.vocab_size
										lowerCAmelCase  =				random_input_ids(lowerCAmelCase						,						lowerCAmelCase						,						lowerCAmelCase					)

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

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

										lowerCAmelCase  =				encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

										return _inference


						def   __lowercase (							self				:			Optional[int]						,						lowerCAmelCase				:			str						,						lowerCAmelCase				:			int						,						lowerCAmelCase				:			int					):
										lowerCAmelCase  =				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."""					)

										lowerCAmelCase  =				(
										    hasattr(lowerCAmelCase						,						"""architectures"""					)
										    and isinstance(config.architectures						,						lowerCAmelCase					)
										    and len(config.architectures					) > 0
										)
										if not self.args.only_pretrain_model and has_model_class_in_config:
														try:
																		lowerCAmelCase  =				"""TF""" + config.architectures[0]  # prepend 'TF' for tensorflow model
																		lowerCAmelCase  =				__import__("""transformers"""						,						fromlist=[model_class]					)
																		lowerCAmelCase  =				getattr(lowerCAmelCase						,						lowerCAmelCase					)
																		lowerCAmelCase  =				model_cls(lowerCAmelCase					)
														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:
														lowerCAmelCase  =				TF_MODEL_WITH_LM_HEAD_MAPPING[config.__class__](lowerCAmelCase					)

										# encoder-decoder has vocab size saved differently
										lowerCAmelCase  =				config.vocab_size if hasattr(lowerCAmelCase						,						"""vocab_size"""					) else config.encoder.vocab_size
										lowerCAmelCase  =				random_input_ids(lowerCAmelCase						,						lowerCAmelCase						,						lowerCAmelCase					)

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

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

										lowerCAmelCase  =				encoder_decoder_train if config.is_encoder_decoder else encoder_train

										return _train


						def   __lowercase (							self				:			Optional[int]						,						lowerCAmelCase				:			List[str]					):
										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(lowerCAmelCase						,						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
																		lowerCAmelCase  =				timeit.repeat(
																		    lowerCAmelCase						,						repeat=self.args.repeat						,						number=10						,						)

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




						def   __lowercase (							self				:			Optional[Any]						,						lowerCAmelCase				:			Callable[[], None]					):
										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."""					)
																						lowerCAmelCase  =				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."""					)
																										lowerCAmelCase  =				"""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()
																										lowerCAmelCase  =				nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx					)
																										lowerCAmelCase  =				nvml.nvmlDeviceGetMemoryInfo(lowerCAmelCase					)
																										lowerCAmelCase  =				meminfo.used
																										lowerCAmelCase  =				Memory(lowerCAmelCase					)
																										# 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."""					)
																										lowerCAmelCase  =				None
																						else:
																										lowerCAmelCase  =				measure_peak_memory_cpu(lowerCAmelCase					)
																										lowerCAmelCase  =				Memory(lowerCAmelCase					) if isinstance(lowerCAmelCase						,						lowerCAmelCase					) else memory_bytes
																		if self.args.trace_memory_line_by_line:
																						lowerCAmelCase  =				stop_memory_tracing(lowerCAmelCase					)
																						if memory is None:
																										lowerCAmelCase  =				summary.total
																		else:
																						lowerCAmelCase  =				None

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





