




import math

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from .attention_processor import Attention
from .embeddings import get_timestep_embedding
from .modeling_utils import ModelMixin




class 		__A      (			UpperCamelCase__					,						UpperCamelCase__			):

						@register_to_config
						def __init__(		self   :List[str]    , __snake_case   :int = 1_28    , __snake_case   :int = 2_56    , __snake_case   :float = 2000.0    , __snake_case   :int = 7_68    , __snake_case   :int = 12    , __snake_case   :int = 12    , __snake_case   :int = 64    , __snake_case   :int = 20_48    , __snake_case   :float = 0.1    , ):
										'''simple docstring'''

										super().__init__()

										__magic_name__	:	int								=nn.Sequential(
										    nn.Linear(__snake_case    , d_model * 4    , bias=__snake_case )    , nn.SiLU()    , nn.Linear(d_model * 4    , d_model * 4    , bias=__snake_case )    , nn.SiLU()    , )

										__magic_name__	:	Optional[Any]								=nn.Embedding(__snake_case    , __snake_case )
										__magic_name__	:	List[Any]								=False

										__magic_name__	:	Any								=nn.Linear(__snake_case    , __snake_case    , bias=__snake_case )

										__magic_name__	:	Union[str, Any]								=nn.Dropout(p=__snake_case )

										__magic_name__	:	List[str]								=nn.ModuleList()
										for lyr_num in range(__snake_case ):
														# FiLM conditional T5 decoder
														__magic_name__	:	List[str]								=DecoderLayer(d_model=__snake_case    , d_kv=__snake_case    , num_heads=__snake_case    , d_ff=__snake_case    , dropout_rate=__snake_case )
														self.decoders.append(__snake_case )

										__magic_name__	:	Tuple								=TaLayerNorm(__snake_case )

										__magic_name__	:	Optional[int]								=nn.Dropout(p=__snake_case )
										__magic_name__	:	List[Any]								=nn.Linear(__snake_case    , __snake_case    , bias=__snake_case )

						def     A__						(		self   :int    , __snake_case   :Any    , __snake_case   :Any ):
										'''simple docstring'''

										__magic_name__	:	int								=torch.mul(query_input.unsqueeze(-1 )    , key_input.unsqueeze(-2 ) )
										return mask.unsqueeze(-3 )


						def     A__						(		self   :Dict    , __snake_case   :Optional[int]    , __snake_case   :Dict    , __snake_case   :Any ):
										'''simple docstring'''

										__magic_name__				, __magic_name__				, __magic_name__	:	str								=decoder_input_tokens.shape
										assert decoder_noise_time.shape == (batch,)

										# decoder_noise_time is in [0, 1), so rescale to expected timing range.
										__magic_name__	:	Tuple								=get_timestep_embedding(
										    decoder_noise_time * self.config.max_decoder_noise_time    , embedding_dim=self.config.d_model    , max_period=self.config.max_decoder_noise_time    , ).to(dtype=self.dtype )

										__magic_name__	:	Tuple								=self.conditioning_emb(__snake_case ).unsqueeze(1 )

										assert conditioning_emb.shape == (batch, 1, self.config.d_model * 4)

										__magic_name__	:	int								=decoder_input_tokens.shape[1]

										# If we want to use relative positions for audio context, we can just offset
										# this sequence by the length of encodings_and_masks.
										__magic_name__	:	Any								=torch.broadcast_to(
										    torch.arange(__snake_case    , device=decoder_input_tokens.device )    , (batch, seq_length)    , )

										__magic_name__	:	List[Any]								=self.position_encoding(__snake_case )

										__magic_name__	:	Any								=self.continuous_inputs_projection(__snake_case )
										inputs += position_encodings
										__magic_name__	:	List[Any]								=self.dropout(__snake_case )

										# decoder: No padding present.
										__magic_name__	:	Any								=torch.ones(
										    decoder_input_tokens.shape[:2]    , device=decoder_input_tokens.device    , dtype=inputs.dtype )

										# Translate encoding masks to encoder-decoder masks.
										__magic_name__	:	Dict								=[(x, self.encoder_decoder_mask(__snake_case    , __snake_case )) for x, y in encodings_and_masks]

										# cross attend style: concat encodings
										__magic_name__	:	Optional[Any]								=torch.cat([x[0] for x in encodings_and_encdec_masks]    , dim=1 )
										__magic_name__	:	Optional[int]								=torch.cat([x[1] for x in encodings_and_encdec_masks]    , dim=-1 )

										for lyr in self.decoders:
														__magic_name__	:	Optional[Any]								=lyr(
														    __snake_case    , conditioning_emb=__snake_case    , encoder_hidden_states=__snake_case    , encoder_attention_mask=__snake_case    , )[0]

										__magic_name__	:	Any								=self.decoder_norm(__snake_case )
										__magic_name__	:	Dict								=self.post_dropout(__snake_case )

										__magic_name__	:	int								=self.spec_out(__snake_case )
										return spec_out




class 		__A      (			nn.Module			):

						def __init__(		self   :str    , __snake_case   :int    , __snake_case   :List[str]    , __snake_case   :Optional[int]    , __snake_case   :List[str]    , __snake_case   :Tuple    , __snake_case   :List[str]=1E-6 ):
										'''simple docstring'''

										super().__init__()
										__magic_name__	:	Optional[int]								=nn.ModuleList()

										# cond self attention: layer 0
										self.layer.append(
										    TaLayerSelfAttentionCond(d_model=__snake_case    , d_kv=__snake_case    , num_heads=__snake_case    , dropout_rate=__snake_case ) )

										# cross attention: layer 1
										self.layer.append(
										    TaLayerCrossAttention(
										        d_model=__snake_case    , d_kv=__snake_case    , num_heads=__snake_case    , dropout_rate=__snake_case    , layer_norm_epsilon=__snake_case    , ) )

										# Film Cond MLP + dropout: last layer
										self.layer.append(
										    TaLayerFFCond(d_model=__snake_case    , d_ff=__snake_case    , dropout_rate=__snake_case    , layer_norm_epsilon=__snake_case ) )


						def     A__						(		self   :str    , __snake_case   :int    , __snake_case   :Dict=None    , __snake_case   :str=None    , __snake_case   :Optional[Any]=None    , __snake_case   :Any=None    , __snake_case   :str=None    , ):
										'''simple docstring'''

										__magic_name__	:	List[str]								=self.layer[0](
										    __snake_case    , conditioning_emb=__snake_case    , attention_mask=__snake_case    , )

										if encoder_hidden_states is not None:
														__magic_name__	:	Any								=torch.where(encoder_attention_mask > 0    , 0    , -1E10 ).to(
														    encoder_hidden_states.dtype )

														__magic_name__	:	List[str]								=self.layer[1](
														    __snake_case    , key_value_states=__snake_case    , attention_mask=__snake_case    , )

										# Apply Film Conditional Feed Forward layer
										__magic_name__	:	Optional[Any]								=self.layer[-1](__snake_case    , __snake_case )

										return (hidden_states,)




class 		__A      (			nn.Module			):

						def __init__(		self   :Optional[int]    , __snake_case   :Any    , __snake_case   :int    , __snake_case   :List[str]    , __snake_case   :str ):
										'''simple docstring'''

										super().__init__()
										__magic_name__	:	Dict								=TaLayerNorm(__snake_case )
										__magic_name__	:	Optional[int]								=TaFiLMLayer(in_features=d_model * 4    , out_features=__snake_case )
										__magic_name__	:	Union[str, Any]								=Attention(query_dim=__snake_case    , heads=__snake_case    , dim_head=__snake_case    , out_bias=__snake_case    , scale_qk=__snake_case )
										__magic_name__	:	Optional[int]								=nn.Dropout(__snake_case )


						def     A__						(		self   :Tuple    , __snake_case   :Optional[int]    , __snake_case   :List[Any]=None    , __snake_case   :List[Any]=None    , ):
										'''simple docstring'''

										__magic_name__	:	List[str]								=self.layer_norm(__snake_case )

										if conditioning_emb is not None:
														__magic_name__	:	Optional[Any]								=self.FiLMLayer(__snake_case    , __snake_case )

										# Self-attention block
										__magic_name__	:	str								=self.attention(__snake_case )

										__magic_name__	:	Tuple								=hidden_states + self.dropout(__snake_case )

										return hidden_states




class 		__A      (			nn.Module			):

						def __init__(		self   :List[str]    , __snake_case   :Any    , __snake_case   :str    , __snake_case   :Union[str, Any]    , __snake_case   :Optional[Any]    , __snake_case   :Optional[int] ):
										'''simple docstring'''

										super().__init__()
										__magic_name__	:	Any								=Attention(query_dim=__snake_case    , heads=__snake_case    , dim_head=__snake_case    , out_bias=__snake_case    , scale_qk=__snake_case )
										__magic_name__	:	List[str]								=TaLayerNorm(__snake_case    , eps=__snake_case )
										__magic_name__	:	Any								=nn.Dropout(__snake_case )


						def     A__						(		self   :Tuple    , __snake_case   :List[str]    , __snake_case   :str=None    , __snake_case   :Optional[Any]=None    , ):
										'''simple docstring'''

										__magic_name__	:	Dict								=self.layer_norm(__snake_case )
										__magic_name__	:	List[Any]								=self.attention(
										    __snake_case    , encoder_hidden_states=__snake_case    , attention_mask=attention_mask.squeeze(1 )    , )
										__magic_name__	:	List[Any]								=hidden_states + self.dropout(__snake_case )
										return layer_output




class 		__A      (			nn.Module			):

						def __init__(		self   :int    , __snake_case   :List[str]    , __snake_case   :Optional[Any]    , __snake_case   :Optional[int]    , __snake_case   :Tuple ):
										'''simple docstring'''

										super().__init__()
										__magic_name__	:	str								=TaDenseGatedActDense(d_model=__snake_case    , d_ff=__snake_case    , dropout_rate=__snake_case )
										__magic_name__	:	Union[str, Any]								=TaFiLMLayer(in_features=d_model * 4    , out_features=__snake_case )
										__magic_name__	:	Dict								=TaLayerNorm(__snake_case    , eps=__snake_case )
										__magic_name__	:	Any								=nn.Dropout(__snake_case )


						def     A__						(		self   :Optional[Any]    , __snake_case   :List[Any]    , __snake_case   :str=None ):
										'''simple docstring'''

										__magic_name__	:	Optional[Any]								=self.layer_norm(__snake_case )
										if conditioning_emb is not None:
														__magic_name__	:	Any								=self.film(__snake_case    , __snake_case )

										__magic_name__	:	Tuple								=self.DenseReluDense(__snake_case )
										__magic_name__	:	Tuple								=hidden_states + self.dropout(__snake_case )
										return hidden_states




class 		__A      (			nn.Module			):

						def __init__(		self   :List[Any]    , __snake_case   :int    , __snake_case   :int    , __snake_case   :Tuple ):
										'''simple docstring'''

										super().__init__()
										__magic_name__	:	Any								=nn.Linear(__snake_case    , __snake_case    , bias=__snake_case )
										__magic_name__	:	Any								=nn.Linear(__snake_case    , __snake_case    , bias=__snake_case )
										__magic_name__	:	List[str]								=nn.Linear(__snake_case    , __snake_case    , bias=__snake_case )
										__magic_name__	:	str								=nn.Dropout(__snake_case )
										__magic_name__	:	List[str]								=NewGELUActivation()


						def     A__						(		self   :Optional[int]    , __snake_case   :Dict ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=self.act(self.wi_a(__snake_case ) )
										__magic_name__	:	Optional[Any]								=self.wi_a(__snake_case )
										__magic_name__	:	Optional[Any]								=hidden_gelu * hidden_linear
										__magic_name__	:	List[str]								=self.dropout(__snake_case )

										__magic_name__	:	Dict								=self.wo(__snake_case )
										return hidden_states




class 		__A      (			nn.Module			):

						def __init__(		self   :Tuple    , __snake_case   :str    , __snake_case   :List[Any]=1E-6 ):
										'''simple docstring'''

										super().__init__()
										__magic_name__	:	Dict								=nn.Parameter(torch.ones(__snake_case ) )
										__magic_name__	:	Optional[Any]								=eps


						def     A__						(		self   :Tuple    , __snake_case   :Union[str, Any] ):
										'''simple docstring'''

										__magic_name__	:	List[Any]								=hidden_states.to(torch.floataa ).pow(2 ).mean(-1    , keepdim=__snake_case )
										__magic_name__	:	List[Any]								=hidden_states * torch.rsqrt(variance + self.variance_epsilon )

										# convert into half-precision if necessary
										if self.weight.dtype in [torch.floataa, torch.bfloataa]:
														__magic_name__	:	Union[str, Any]								=hidden_states.to(self.weight.dtype )

										return self.weight * hidden_states




class 		__A      (			nn.Module			):

						def     A__						(		self   :List[str]    , __snake_case   :torch.Tensor ):
										'''simple docstring'''

										return 0.5 * input * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi ) * (input + 0.044715 * torch.pow(__snake_case    , 3.0 )) ))




class 		__A      (			nn.Module			):

						def __init__(		self   :str    , __snake_case   :Any    , __snake_case   :Optional[int] ):
										'''simple docstring'''

										super().__init__()
										__magic_name__	:	Optional[int]								=nn.Linear(__snake_case    , out_features * 2    , bias=__snake_case )


						def     A__						(		self   :int    , __snake_case   :Union[str, Any]    , __snake_case   :Union[str, Any] ):
										'''simple docstring'''

										__magic_name__	:	Dict								=self.scale_bias(__snake_case )
										__magic_name__				, __magic_name__	:	Any								=torch.chunk(__snake_case    , 2    , -1 )
										__magic_name__	:	Optional[int]								=x * (1 + scale) + shift
										return x










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

UpperCAmelCase_		:				Optional[Any]									=     logging.get_logger(__name__)




def 		lowerCAmelCase_   (	lowerCamelCase	,     lowerCamelCase   ):
				def run_func(lowerCamelCase   ):
								@wraps(lowerCamelCase   )
								def run_in_eager_mode(*lowerCamelCase	,     **lowerCamelCase   ):
												return func(*lowerCamelCase	,     **lowerCamelCase   )

								@wraps(lowerCamelCase   )
								@tf.function(experimental_compile=lowerCamelCase   )
								def run_in_graph_mode(*lowerCamelCase	,     **lowerCamelCase   ):
												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 		lowerCAmelCase_   (	lowerCamelCase	,     lowerCamelCase	,     lowerCamelCase   ):
				__magic_name__	:	Dict								=random.Random()
				__magic_name__	:	Union[str, Any]								=[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      (			UpperCamelCase__			):
						UpperCamelCase     						=					42
						UpperCamelCase     						=					42
						UpperCamelCase     						=					"TensorFlow"

						@property
						def     A__						(		self   :str ):
										'''simple docstring'''

										return tf.__version__

						def     A__						(		self   :str    , __snake_case   :str    , __snake_case   :int    , __snake_case   :int ):
										'''simple docstring'''

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

						def     A__						(		self   :int    , __snake_case   :str    , __snake_case   :int    , __snake_case   :int ):
										'''simple docstring'''

										__magic_name__	:	Tuple								=self.args.strategy
										if strategy is None:
														raise ValueError("""A device strategy has to be initialized before using TensorFlow.""" )
										__magic_name__	:	Any								=self._prepare_train_func(__snake_case    , __snake_case    , __snake_case )
										return self._measure_speed(_train )

						def     A__						(		self   :str    , __snake_case   :str    , __snake_case   :int    , __snake_case   :int ):
										'''simple docstring'''

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

						def     A__						(		self   :str    , __snake_case   :str    , __snake_case   :int    , __snake_case   :int ):
										'''simple docstring'''

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

										__magic_name__	:	Any								=self._prepare_train_func(__snake_case    , __snake_case    , __snake_case )
										return self._measure_memory(_train )

						def     A__						(		self   :int    , __snake_case   :str    , __snake_case   :int    , __snake_case   :int ):
										'''simple docstring'''

										__magic_name__	:	int								=self.config_dict[model_name]

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

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

										# encoder-decoder has vocab size saved differently
										__magic_name__	:	List[str]								=config.vocab_size if hasattr(__snake_case    , """vocab_size""" ) else config.encoder.vocab_size
										__magic_name__	:	Any								=random_input_ids(__snake_case    , __snake_case    , __snake_case )

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

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

										__magic_name__	:	Tuple								=encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

										return _inference

						def     A__						(		self   :int    , __snake_case   :str    , __snake_case   :int    , __snake_case   :int ):
										'''simple docstring'''

										__magic_name__	:	Tuple								=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.""" )

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

										# encoder-decoder has vocab size saved differently
										__magic_name__	:	int								=config.vocab_size if hasattr(__snake_case    , """vocab_size""" ) else config.encoder.vocab_size
										__magic_name__	:	Optional[Any]								=random_input_ids(__snake_case    , __snake_case    , __snake_case )

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

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

										__magic_name__	:	Union[str, Any]								=encoder_decoder_train if config.is_encoder_decoder else encoder_train

										return _train

						def     A__						(		self   :Any    , __snake_case   :Union[str, 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(__snake_case    , 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
																		__magic_name__	:	Union[str, Any]								=timeit.repeat(
																		    __snake_case    , repeat=self.args.repeat    , number=10    , )

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


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

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




