


"""simple docstring"""






import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
 import torch

 from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
 import pyanvml.pyanvml as nvml


SCREAMING_SNAKE_CASE__				:  Union[str, Any]						= logging.get_logger(__name__)


class 	A_       (     _snake_case   ):





  """simple docstring"""

  lowercase		:	PyTorchBenchmarkArguments
  lowercase		:	PretrainedConfig
  lowercase		:	str   =						"PyTorch"
  @property
  def 							lowercase_     (						self						)				->					int:
     return torch.__version__
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					Tuple:
     a   :	Union[str, Any]						=    self._prepare_inference_func(snake_case_  ,					snake_case_  ,					snake_case_						)
     return self._measure_speed(_inference						)
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					Tuple:
     a   :	str						=    self._prepare_inference_func(snake_case_  ,					snake_case_  ,					snake_case_						)
     return self._measure_memory(_inference						)
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					str:
     a   :	Dict						=    self._prepare_train_func(snake_case_  ,					snake_case_  ,					snake_case_						)
     return self._measure_speed(_train						)
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					List[Any]:
     a   :	Optional[int]						=    self._prepare_train_func(snake_case_  ,					snake_case_  ,					snake_case_						)
     return self._measure_memory(_train						)
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					int:
     a   :	List[Any]						=    self.config_dict[model_name]

     if self.args.torchscript:
        a   :	List[Any]						=    True

     a   :	List[str]						=    (
         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:
           a   :	Optional[int]						=    config.architectures[0]
           a   :	List[Any]						=    __import__('transformers'  ,					fromlist=[model_class]						)
           a   :	List[str]						=    getattr(snake_case_  ,					snake_case_						)
           a   :	Tuple						=    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:
        a   :	Tuple						=    MODEL_MAPPING[config.__class__](snake_case_						)

     model.eval()
     model.to(self.args.device						)

     # encoder-decoder has vocab size saved differently
     a   :	Tuple						=    config.vocab_size if hasattr(snake_case_  ,					'vocab_size'						) else config.encoder.vocab_size
     a   :	int						=    torch.randint(snake_case_  ,					(batch_size, sequence_length)  ,					dtype=torch.long  ,					device=self.args.device						)

     if self.args.fpaa:
        logger.info('Running training in Mixed Precision...'						)
        if not self.args.is_gpu:
           raise ValueError('Mixed precision is possible only for GPU.'						)
        # amp seems to have memory leaks so that memory usage
        # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
        model.half()

     if self.args.torchscript:
        with torch.no_grad():
           a   :	str						=    torch.jit.trace(snake_case_  ,					snake_case_						)
     else:
        a   :	Optional[Any]						=    model

     def encoder_decoder_forward():
        with torch.no_grad():
           a   :	Dict						=    inference_model(snake_case_  ,					decoder_input_ids=snake_case_						)
        return outputs

     def encoder_forward():
        with torch.no_grad():
           a   :	Optional[Any]						=    inference_model(snake_case_						)
        return outputs

     a   :	str						=    encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
     return _forward
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase						)				->					Optional[Any]:
     a   :	Any						=    self.config_dict[model_name]

     a   :	Union[str, 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:
           a   :	Union[str, Any]						=    config.architectures[0]
           a   :	Optional[int]						=    __import__('transformers'  ,					fromlist=[model_class]						)
           a   :	int						=    getattr(snake_case_  ,					snake_case_						)
           a   :	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:
        a   :	Optional[int]						=    MODEL_WITH_LM_HEAD_MAPPING[config.__class__](snake_case_						)

     if self.args.torchscript:
        raise NotImplementedError('Training for torchscript is currently not implemented'						)
     else:
        a   :	Dict						=    model

     model.train()
     model.to(self.args.device						)

     # encoder-decoder has vocab size saved differently
     a   :	Dict						=    config.vocab_size if hasattr(snake_case_  ,					'vocab_size'						) else config.encoder.vocab_size
     a   :	Any						=    torch.randint(snake_case_  ,					(batch_size, sequence_length)  ,					dtype=torch.long  ,					device=self.args.device						)

     if self.args.fpaa:
        logger.info('Running training in Mixed Precision...'						)
        if not self.args.is_gpu:
           raise ValueError('Mixed precision is possible only for GPU.'						)

        # amp seems to have memory leaks so that memory usage
        # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
        model.half()

     def compute_loss_and_backprob_encoder():
        a   :	Optional[Any]						=    train_model(snake_case_  ,					labels=snake_case_						)[0]
        loss.backward()
        return loss

     def compute_loss_and_backprob_encoder_decoder():
        a   :	int						=    train_model(snake_case_  ,					decoder_input_ids=snake_case_  ,					labels=snake_case_						)[0]
        loss.backward()
        return loss

     a   :	Optional[int]						=    (
         compute_loss_and_backprob_encoder_decoder
         if config.is_encoder_decoder
         else compute_loss_and_backprob_encoder
     )
     return _train
  def 							lowercase_     (						self  ,					__UpperCAmelCase						)				->					Union[str, Any]:
     try:
        if self.args.is_tpu or self.args.torchscript:
           # run additional 10 times to stabilize compilation for tpu and torchscript
           logger.info('Do inference on TPU or torchscript. 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
        a   :	Tuple						=    timeit.repeat(
            snake_case_  ,					repeat=self.args.repeat  ,					number=10  ,					)

        if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
           import torch_xla.debug.metrics as met

           self.print_fn(met.metrics_report()						)

        return min(snake_case_						) / 10.0
     except RuntimeError as e:
        self.print_fn(f'Doesn\'t fit on GPU. {e}'						)
        return "N/A"






  def 							lowercase_     (						self  ,					__UpperCAmelCase						)				->					Optional[Any]:
     try:
        if self.args.trace_memory_line_by_line:
           a   :	Dict						=    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'
               ' `--no-memory` or `args.memory=False`'						)
        elif self.args.is_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.'						)
              a   :	Tuple						=    "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()
              a   :	Union[str, Any]						=    nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx						)
              a   :	int						=    nvml.nvmlDeviceGetMemoryInfo(snake_case_						)
              a   :	Optional[int]						=    meminfo.used
              a   :	Optional[Any]						=    Memory(snake_case_						)
              # shutdown nvml
              nvml.nvmlShutdown()
        else:
           # cpu
           a   :	str						=    measure_peak_memory_cpu(snake_case_						)
           a   :	Tuple						=    Memory(snake_case_						) if isinstance(snake_case_  ,					snake_case_						) else memory_bytes

        if self.args.trace_memory_line_by_line:
           a   :	List[str]						=    stop_memory_tracing(snake_case_						)
        else:
           a   :	List[Any]						=    None

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







"""simple docstring"""
import random
import unittest

import torch

from diffusers import IFSuperResolutionPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin






@skip_mps
class   _UpperCAmelCase					(       _snake_case		,       _snake_case		,       unittest.TestCase):
    __lowercase				:			List[str]	=       IFSuperResolutionPipeline
    __lowercase				:			Optional[Any]	=       TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""width""", """height"""}
    __lowercase				:			List[Any]	=       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    __lowercase				:			Optional[Any]	=       PipelineTesterMixin.required_optional_params - {"""latents"""}
    def        lowerCamelCase__						(     self  ):
      return self._get_superresolution_dummy_components()
    def        lowerCamelCase__						(     self    ,     snake_case_    ,     snake_case_=0  ):
      if str(snake_case_  ).startswith("mps"  ):
        _snake_case  :    Tuple            = torch.manual_seed(snake_case_  )
      else:
        _snake_case  :    Tuple            = torch.Generator(device=snake_case_  ).manual_seed(snake_case_  )

      _snake_case  :    List[str]            = floats_tensor((1, 3, 32, 32)    ,     rng=random.Random(snake_case_  )  ).to(snake_case_  )

      _snake_case  :    List[str]            = {
          "prompt": "A painting of a squirrel eating a burger",
          "image": image,
          "generator": generator,
          "num_inference_steps": 2,
          "output_type": "numpy",
      }

      return inputs
    @unittest.skipIf(
        torch_device != "cuda" or not is_xformers_available()    ,     reason="XFormers attention is only available with CUDA and `xformers` installed"    ,     )
    def        lowerCamelCase__						(     self  ):
      self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3  )
    def        lowerCamelCase__						(     self  ):
      self._test_save_load_optional_components()
    @unittest.skipIf(torch_device != "cuda"    ,     reason="float16 requires CUDA"  )
    def        lowerCamelCase__						(     self  ):
      # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
      super().test_save_load_floataa(expected_max_diff=1E-1  )
    def        lowerCamelCase__						(     self  ):
      self._test_attention_slicing_forward_pass(expected_max_diff=1E-2  )
    def        lowerCamelCase__						(     self  ):
      self._test_save_load_local()


    def        lowerCamelCase__						(     self  ):
      self._test_inference_batch_single_identical(
          expected_max_diff=1E-2    ,     )

