


"""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


lowercase  		=			logging.get_logger(__name__)


class 			SCREAMING_SNAKE_CASE_  (   lowerCamelCase_):




							'''simple docstring'''





							__magic_name__					:     PyTorchBenchmarkArguments
							__magic_name__					:     PretrainedConfig
							__magic_name__					:     str				  =  "PyTorch"




							@property
							def        UpperCAmelCase (						self)   ->	Dict:







								'''simple docstring'''




								return torch.__version__




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)   ->	Optional[Any]:







								'''simple docstring'''




								snake_case__  :						str		=      self._prepare_inference_func(lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)
								return self._measure_speed(_inference)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)   ->	Optional[Any]:







								'''simple docstring'''




								snake_case__  :						int		=      self._prepare_inference_func(lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)
								return self._measure_memory(_inference)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)   ->	Optional[int]:







								'''simple docstring'''




								snake_case__  :						Union[str, Any]		=      self._prepare_train_func(lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)
								return self._measure_speed(_train)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)   ->	Any:







								'''simple docstring'''




								snake_case__  :						Any		=      self._prepare_train_func(lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)
								return self._measure_memory(_train)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)   ->	int:







								'''simple docstring'''




								snake_case__  :						Any		=      self.config_dict[model_name]

								if self.args.torchscript:
									snake_case__  :						Any		=      True

								snake_case__  :						Union[str, Any]		=      (
								    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:
										snake_case__  :						Any		=      config.architectures[0]
										snake_case__  :						Dict		=      __import__("transformers"	,					fromlist=[model_class])
										snake_case__  :						List[str]		=      getattr(lowerCamelCase__	,					lowerCamelCase__)
										snake_case__  :						Dict		=      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:
									snake_case__  :						Optional[int]		=      MODEL_MAPPING[config.__class__](lowerCamelCase__)

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

								# encoder-decoder has vocab size saved differently
								snake_case__  :						str		=      config.vocab_size if hasattr(lowerCamelCase__	,					"vocab_size") else config.encoder.vocab_size
								snake_case__  :						Union[str, Any]		=      torch.randint(lowerCamelCase__	,					(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():
										snake_case__  :						int		=      torch.jit.trace(lowerCamelCase__	,					lowerCamelCase__)
								else:
									snake_case__  :						Any		=      model

								def encoder_decoder_forward():
									with torch.no_grad():
										snake_case__  :						int		=      inference_model(lowerCamelCase__	,					decoder_input_ids=lowerCamelCase__)
									return outputs

								def encoder_forward():
									with torch.no_grad():
										snake_case__  :						Tuple		=      inference_model(lowerCamelCase__)
									return outputs

								snake_case__  :						int		=      encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
								return _forward




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)   ->	Union[str, Any]:







								'''simple docstring'''




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

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

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

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

								# encoder-decoder has vocab size saved differently
								snake_case__  :						Union[str, Any]		=      config.vocab_size if hasattr(lowerCamelCase__	,					"vocab_size") else config.encoder.vocab_size
								snake_case__  :						List[Any]		=      torch.randint(lowerCamelCase__	,					(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():
									snake_case__  :						Optional[Any]		=      train_model(lowerCamelCase__	,					labels=lowerCamelCase__)[0]
									loss.backward()
									return loss

								def compute_loss_and_backprob_encoder_decoder():
									snake_case__  :						List[Any]		=      train_model(lowerCamelCase__	,					decoder_input_ids=lowerCamelCase__	,					labels=lowerCamelCase__)[0]
									loss.backward()
									return loss

								snake_case__  :						Any		=      (
								    compute_loss_and_backprob_encoder_decoder
								    if config.is_encoder_decoder
								    else compute_loss_and_backprob_encoder
								)
								return _train




							def        UpperCAmelCase (						self	,					lowerCamelCase__)   ->	Tuple:







								'''simple docstring'''




								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(
										    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
									snake_case__  :						str		=      timeit.repeat(
									    lowerCamelCase__	,					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(lowerCamelCase__) / 10.0
								except RuntimeError as e:
									self.print_fn(f"""Doesn't fit on GPU. {e}""")
									return "N/A"







							def        UpperCAmelCase (						self	,					lowerCamelCase__)   ->	Dict:







								'''simple docstring'''




								try:
									if self.args.trace_memory_line_by_line:
										snake_case__  :						int		=      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.")
											snake_case__  :						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()
											snake_case__  :						List[Any]		=      nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx)
											snake_case__  :						List[Any]		=      nvml.nvmlDeviceGetMemoryInfo(lowerCamelCase__)
											snake_case__  :						int		=      meminfo.used
											snake_case__  :						Dict		=      Memory(lowerCamelCase__)
											# shutdown nvml
											nvml.nvmlShutdown()
									else:
										# cpu
										snake_case__  :						Optional[Any]		=      measure_peak_memory_cpu(lowerCamelCase__)
										snake_case__  :						Union[str, Any]		=      Memory(lowerCamelCase__) if isinstance(lowerCamelCase__	,					lowerCamelCase__) else memory_bytes

									if self.args.trace_memory_line_by_line:
										snake_case__  :						Tuple		=      stop_memory_tracing(lowerCamelCase__)
									else:
										snake_case__  :						int		=      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 os

from packaging import version

from .. import __version__
from .accelerate_utils import apply_forward_hook
from .constants import (
    CONFIG_NAME,
    DEPRECATED_REVISION_ARGS,
    DIFFUSERS_CACHE,
    DIFFUSERS_DYNAMIC_MODULE_NAME,
    FLAX_WEIGHTS_NAME,
    HF_MODULES_CACHE,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    ONNX_EXTERNAL_WEIGHTS_NAME,
    ONNX_WEIGHTS_NAME,
    SAFETENSORS_WEIGHTS_NAME,
    TEXT_ENCODER_ATTN_MODULE,
    WEIGHTS_NAME,
)
from .deprecation_utils import deprecate
from .doc_utils import replace_example_docstring
from .dynamic_modules_utils import get_class_from_dynamic_module
from .hub_utils import (
    HF_HUB_OFFLINE,
    _add_variant,
    _get_model_file,
    extract_commit_hash,
    http_user_agent,
)
from .import_utils import (
    BACKENDS_MAPPING,
    ENV_VARS_TRUE_AND_AUTO_VALUES,
    ENV_VARS_TRUE_VALUES,
    USE_JAX,
    USE_TF,
    USE_TORCH,
    DummyObject,
    OptionalDependencyNotAvailable,
    is_accelerate_available,
    is_accelerate_version,
    is_bsa_available,
    is_flax_available,
    is_ftfy_available,
    is_inflect_available,
    is_invisible_watermark_available,
    is_k_diffusion_available,
    is_k_diffusion_version,
    is_librosa_available,
    is_note_seq_available,
    is_omegaconf_available,
    is_onnx_available,
    is_safetensors_available,
    is_scipy_available,
    is_tensorboard_available,
    is_tf_available,
    is_torch_available,
    is_torch_version,
    is_torchsde_available,
    is_transformers_available,
    is_transformers_version,
    is_unidecode_available,
    is_wandb_available,
    is_xformers_available,
    requires_backends,
)
from .logging import get_logger
from .outputs import BaseOutput
from .pil_utils import PIL_INTERPOLATION, numpy_to_pil, pt_to_pil
from .torch_utils import is_compiled_module, randn_tensor


if is_torch_available():
      from .testing_utils import (
          floats_tensor,
          load_hf_numpy,
          load_image,
          load_numpy,
          load_pt,
          nightly,
          parse_flag_from_env,
          print_tensor_test,
          require_torch_a,
          require_torch_gpu,
          skip_mps,
          slow,
          torch_all_close,
          torch_device,
      )
      from .torch_utils import maybe_allow_in_graph

from .testing_utils import export_to_gif, export_to_video


lowercase						:   Tuple      						=							get_logger(__name__)




def 		a_(			lowerCamelCase   )					->Any:






       '''simple docstring'''




       if version.parse(lowerCamelCase   ) < version.parse(lowerCamelCase   ):
              if "dev" in min_version:
                     lowercase	:		str    =   (
                         '''This example requires a source install from HuggingFace diffusers (see '''
                         '''`https://huggingface.co/docs/diffusers/installation#install-from-source`),'''
                     )
              else:
                     lowercase	:		List[str]    =   f"""This example requires a minimum version of {min_version},"""
              error_message += f""" but the version found is {__version__}.\n"""
              raise ImportError(lowerCamelCase   )




