




"""simple docstring"""


from dataclasses import dataclass
from typing import Optional

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .modeling_utils import ModelMixin






@dataclass
class       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      42







class       lowercase(  __a ,    __a				):







 '''simple docstring'''





 @register_to_config
 def __init__(       self:	List[Any],   a_:	int = 16,   a_:	int = 88,   a_:	Optional[int] = None,   a_:	Optional[int] = None,   a_:	int = 1,   a_:	float = 0.0,   a_:	int = 32,   a_:	Optional[int] = None,   a_:	bool = False,   a_:	Optional[int] = None,   a_:	str = "geglu",   a_:	bool = True,   a_:	bool = True,   ):




   '''simple docstring'''



   super().__init__()
   _snake_case							:	List[str]			=						num_attention_heads
   _snake_case							:	int			=						attention_head_dim
   _snake_case							:	Optional[int]			=						num_attention_heads * attention_head_dim

   _snake_case							:	Any			=						in_channels

   _snake_case							:	List[str]			=						torch.nn.GroupNorm(num_groups=a_,   num_channels=a_,   eps=1E-6,   affine=a_	)
   _snake_case							:	Dict			=						nn.Linear(a_,   a_	)

   # 3. Define transformers blocks
   _snake_case							:	List[Any]			=						nn.ModuleList(
       [
           BasicTransformerBlock(
               a_,   a_,   a_,   dropout=a_,   cross_attention_dim=a_,   activation_fn=a_,   attention_bias=a_,   double_self_attention=a_,   norm_elementwise_affine=a_,   )
           for d in range(a_	)
       ]	)

   _snake_case							:	Tuple			=						nn.Linear(a_,   a_	)


 def 				UpperCamelCase_     (       self:	Any,   a_:	List[str],   a_:	List[str]=None,   a_:	Union[str, Any]=None,   a_:	Tuple=None,   a_:	Any=1,   a_:	int=None,   a_:	bool = True,   ):




   '''simple docstring'''



   _snake_case	,      _snake_case	,      _snake_case	,      _snake_case							:	Any			=						hidden_states.shape
   _snake_case							:	List[str]			=						batch_frames // num_frames

   _snake_case							:	Tuple			=						hidden_states

   _snake_case							:	List[str]			=						hidden_states[None, :].reshape(a_,   a_,   a_,   a_,   a_	)
   _snake_case							:	Dict			=						hidden_states.permute(0,   2,   1,   3,   4	)

   _snake_case							:	Tuple			=						self.norm(a_	)
   _snake_case							:	Optional[int]			=						hidden_states.permute(0,   3,   4,   2,   1	).reshape(batch_size * height * width,   a_,   a_	)

   _snake_case							:	Any			=						self.proj_in(a_	)

   # 2. Blocks
   for block in self.transformer_blocks:
     _snake_case							:	List[str]			=						block(
         a_,   encoder_hidden_states=a_,   timestep=a_,   cross_attention_kwargs=a_,   class_labels=a_,   )

   # 3. Output
   _snake_case							:	Tuple			=						self.proj_out(a_	)
   _snake_case							:	List[str]			=						(
       hidden_states[None, None, :]
       .reshape(a_,   a_,   a_,   a_,   a_	)
       .permute(0,   3,   4,   1,   2	)
       .contiguous()
   )
   _snake_case							:	List[Any]			=						hidden_states.reshape(a_,   a_,   a_,   a_	)

   _snake_case							:	List[str]			=						hidden_states + residual

   if not return_dict:
     return (output,)

   return TransformerTemporalModelOutput(sample=a_	)








"""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       UpperCAmelCase__ (snake_case__							:      bool	,     snake_case__							:      bool					):
  """simple docstring"""

  def run_func(snake_case__							:      Tuple					):
    @wraps(snake_case__					)
    def run_in_eager_mode(*snake_case__							:      str	,     **snake_case__							:      Any					):
      return func(*snake_case__	,     **snake_case__					)

    @wraps(snake_case__					)
    @tf.function(experimental_compile=snake_case__					)
    def run_in_graph_mode(*snake_case__							:      Any	,     **snake_case__							:      Optional[int]					):
      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       UpperCAmelCase__ (snake_case__							:      int	,     snake_case__							:      int	,     snake_case__							:      int					):
  """simple docstring"""

  _snake_case							:	List[str]			=						random.Random()
  _snake_case							:	Optional[int]			=						[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       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      42
 lowercase__				      =      42
 lowercase__				      =      "TensorFlow"


 @property
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   return tf.__version__


 def 				UpperCamelCase_     (       self:	List[str],   a_:	str,   a_:	int,   a_:	int	):




   '''simple docstring'''



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


 def 				UpperCamelCase_     (       self:	int,   a_:	str,   a_:	int,   a_:	int	):




   '''simple docstring'''



   _snake_case							:	Tuple			=						self.args.strategy
   if strategy is None:
     raise ValueError("""A device strategy has to be initialized before using TensorFlow."""	)
   _snake_case							:	Optional[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_	)
   _snake_case							:	str			=						self.args.strategy
   if strategy is None:
     raise ValueError("""A device strategy has to be initialized before using TensorFlow."""	)
   _snake_case							:	List[str]			=						self._prepare_inference_func(a_,   a_,   a_	)
   return self._measure_memory(_inference	)


 def 				UpperCamelCase_     (       self:	Tuple,   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_	)
   _snake_case							:	Dict			=						self.args.strategy
   if strategy is None:
     raise ValueError("""A device strategy has to be initialized before using TensorFlow."""	)

   _snake_case							:	Optional[int]			=						self._prepare_train_func(a_,   a_,   a_	)
   return self._measure_memory(_train	)


 def 				UpperCamelCase_     (       self:	Optional[Any],   a_:	str,   a_:	int,   a_:	int	):




   '''simple docstring'''



   _snake_case							:	List[Any]			=						self.config_dict[model_name]

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

   _snake_case							:	List[Any]			=						(
       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:
       _snake_case							:	str			=						"""TF""" + config.architectures[0]  # prepend 'TF' for tensorflow model
       _snake_case							:	List[Any]			=						__import__("""transformers""",   fromlist=[model_class]	)
       _snake_case							:	Dict			=						getattr(a_,   a_	)
       _snake_case							:	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:
     _snake_case							:	Any			=						TF_MODEL_MAPPING[config.__class__](a_	)

   # encoder-decoder has vocab size saved differently
   _snake_case							:	List[Any]			=						config.vocab_size if hasattr(a_,   """vocab_size"""	) else config.encoder.vocab_size
   _snake_case							:	List[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_	)

   _snake_case							:	Optional[int]			=						encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

   return _inference


 def 				UpperCamelCase_     (       self:	Optional[int],   a_:	str,   a_:	int,   a_:	int	):




   '''simple docstring'''



   _snake_case							:	str			=						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."""	)

   _snake_case							:	Tuple			=						(
       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:
       _snake_case							:	List[str]			=						"""TF""" + config.architectures[0]  # prepend 'TF' for tensorflow model
       _snake_case							:	str			=						__import__("""transformers""",   fromlist=[model_class]	)
       _snake_case							:	Tuple			=						getattr(a_,   a_	)
       _snake_case							:	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:
     _snake_case							:	Optional[Any]			=						TF_MODEL_WITH_LM_HEAD_MAPPING[config.__class__](a_	)

   # encoder-decoder has vocab size saved differently
   _snake_case							:	List[Any]			=						config.vocab_size if hasattr(a_,   """vocab_size"""	) else config.encoder.vocab_size
   _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():
     _snake_case							:	Dict			=						model(a_,   decoder_input_ids=a_,   labels=a_,   training=a_	)[0]
     _snake_case							:	str			=						tf.gradients(a_,   model.trainable_variables	)
     return gradients

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

   _snake_case							:	int			=						encoder_decoder_train if config.is_encoder_decoder else encoder_train

   return _train


 def 				UpperCamelCase_     (       self:	Union[str, Any],   a_:	str	):




   '''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
       _snake_case							:	Dict			=						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:	Optional[Any],   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."""	)
         _snake_case							:	List[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."""	)
           _snake_case							:	Optional[Any]			=						"""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()
           _snake_case							:	List[str]			=						nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx	)
           _snake_case							:	Tuple			=						nvml.nvmlDeviceGetMemoryInfo(a_	)
           _snake_case							:	List[str]			=						meminfo.used
           _snake_case							:	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."""	)
           _snake_case							:	List[Any]			=						None
         else:
           _snake_case							:	int			=						measure_peak_memory_cpu(a_	)
           _snake_case							:	List[str]			=						Memory(a_	) if isinstance(a_,   a_	) else memory_bytes
       if self.args.trace_memory_line_by_line:
         _snake_case							:	Tuple			=						stop_memory_tracing(a_	)
         if memory is None:
           _snake_case							:	int			=						summary.total
       else:
         _snake_case							:	int			=						None

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


