

"""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 				__a    (   __a					):



							'''simple docstring'''







							_lowerCamelCase				: PyTorchBenchmarkArguments
							_lowerCamelCase				: PretrainedConfig
							_lowerCamelCase				: str					      =      "PyTorch"
							@property
							def      SCREAMING_SNAKE_CASE	(    self					) ->       Optional[int]:




										'''simple docstring'''
										return torch.__version__
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					) ->       float:




										'''simple docstring'''
										__lowercase						      =					self._prepare_inference_func(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					)
										return self._measure_speed(_inference					)
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					) ->       [Memory, Optional[MemorySummary]]:




										'''simple docstring'''
										__lowercase						      =					self._prepare_inference_func(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					)
										return self._measure_memory(_inference					)
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					) ->       float:




										'''simple docstring'''
										__lowercase						      =					self._prepare_train_func(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					)
										return self._measure_speed(_train					)
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					) ->       [Memory, Optional[MemorySummary]]:




										'''simple docstring'''
										__lowercase						      =					self._prepare_train_func(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					)
										return self._measure_memory(_train					)
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					) ->       Callable[[], None]:




										'''simple docstring'''
										__lowercase						      =					self.config_dict[model_name]

										if self.args.torchscript:
													__lowercase						      =					True

										__lowercase						      =					(
										    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:
																__lowercase						      =					config.architectures[0]
																__lowercase						      =					__import__("transformers"     ,						fromlist=[model_class]					)
																__lowercase						      =					getattr(_lowerCamelCase     ,						_lowerCamelCase					)
																__lowercase						      =					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:
													__lowercase						      =					MODEL_MAPPING[config.__class__](_lowerCamelCase					)

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

										# encoder-decoder has vocab size saved differently
										__lowercase						      =					config.vocab_size if hasattr(_lowerCamelCase     ,						"vocab_size"					) else config.encoder.vocab_size
										__lowercase						      =					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():
																__lowercase						      =					torch.jit.trace(_lowerCamelCase     ,						_lowerCamelCase					)
										else:
													__lowercase						      =					model

										def encoder_decoder_forward():
													with torch.no_grad():
																__lowercase						      =					inference_model(_lowerCamelCase     ,						decoder_input_ids=_lowerCamelCase					)
													return outputs

										def encoder_forward():
													with torch.no_grad():
																__lowercase						      =					inference_model(_lowerCamelCase					)
													return outputs

										__lowercase						      =					encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
										return _forward
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					) ->       Callable[[], None]:




										'''simple docstring'''
										__lowercase						      =					self.config_dict[model_name]

										__lowercase						      =					(
										    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:
																__lowercase						      =					config.architectures[0]
																__lowercase						      =					__import__("transformers"     ,						fromlist=[model_class]					)
																__lowercase						      =					getattr(_lowerCamelCase     ,						_lowerCamelCase					)
																__lowercase						      =					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:
													__lowercase						      =					MODEL_WITH_LM_HEAD_MAPPING[config.__class__](_lowerCamelCase					)

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

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

										# encoder-decoder has vocab size saved differently
										__lowercase						      =					config.vocab_size if hasattr(_lowerCamelCase     ,						"vocab_size"					) else config.encoder.vocab_size
										__lowercase						      =					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():
													__lowercase						      =					train_model(_lowerCamelCase     ,						labels=_lowerCamelCase					)[0]
													loss.backward()
													return loss

										def compute_loss_and_backprob_encoder_decoder():
													__lowercase						      =					train_model(_lowerCamelCase     ,						decoder_input_ids=_lowerCamelCase     ,						labels=_lowerCamelCase					)[0]
													loss.backward()
													return loss

										__lowercase						      =					(
										    compute_loss_and_backprob_encoder_decoder
										    if config.is_encoder_decoder
										    else compute_loss_and_backprob_encoder
										)
										return _train
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase					) ->       float:




										'''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
													__lowercase						      =					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      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase					) ->       [Memory, MemorySummary]:




										'''simple docstring'''
										try:
													if self.args.trace_memory_line_by_line:
																__lowercase						      =					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."					)
																			__lowercase						      =					"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()
																			__lowercase						      =					nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx					)
																			__lowercase						      =					nvml.nvmlDeviceGetMemoryInfo(_lowerCamelCase					)
																			__lowercase						      =					meminfo.used
																			__lowercase						      =					Memory(_lowerCamelCase					)
																			# shutdown nvml
																			nvml.nvmlShutdown()
													else:
																# cpu
																__lowercase						      =					measure_peak_memory_cpu(_lowerCamelCase					)
																__lowercase						      =					Memory(_lowerCamelCase					) if isinstance(_lowerCamelCase     ,						_lowerCamelCase					) else memory_bytes

													if self.args.trace_memory_line_by_line:
																__lowercase						      =					stop_memory_tracing(_lowerCamelCase					)
													else:
																__lowercase						      =					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							 =							get_logger(__name__)



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





