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


A_						       =				logging.get_logger(__name__)




class    __lowerCamelCase     (   lowerCAmelCase ):
				a__:			PyTorchBenchmarkArguments
				a__:			PretrainedConfig
				a__:			str									=						"PyTorch"




				@property
				def 					UpperCAmelCase__	(							self      ):
							return torch.__version__




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ):
							lowerCamelCase_ 		=     self._prepare_inference_func(UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      )
							return self._measure_speed(_inference      )




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ):
							lowerCamelCase_ 		=     self._prepare_inference_func(UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      )
							return self._measure_memory(_inference      )




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ):
							lowerCamelCase_ 		=     self._prepare_train_func(UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      )
							return self._measure_speed(_train      )




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ):
							lowerCamelCase_ 		=     self._prepare_train_func(UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      )
							return self._measure_memory(_train      )




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ):
							lowerCamelCase_ 		=     self.config_dict[model_name]

							if self.args.torchscript:
										lowerCamelCase_ 		=     True

							lowerCamelCase_ 		=     (
							    hasattr(UpperCAmelCase       ,	'''architectures'''      )
							    and isinstance(config.architectures       ,	UpperCAmelCase      )
							    and len(config.architectures      ) > 0
							)
							if not self.args.only_pretrain_model and has_model_class_in_config:
										try:
													lowerCamelCase_ 		=     config.architectures[0]
													lowerCamelCase_ 		=     __import__('''transformers'''       ,	fromlist=[model_class]      )
													lowerCamelCase_ 		=     getattr(UpperCAmelCase       ,	UpperCAmelCase      )
													lowerCamelCase_ 		=     model_cls(UpperCAmelCase      )
										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:
										lowerCamelCase_ 		=     MODEL_MAPPING[config.__class__](UpperCAmelCase      )

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

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

							def encoder_decoder_forward():
										with torch.no_grad():
													lowerCamelCase_ 		=     inference_model(UpperCAmelCase       ,	decoder_input_ids=UpperCAmelCase      )
										return outputs

							def encoder_forward():
										with torch.no_grad():
													lowerCamelCase_ 		=     inference_model(UpperCAmelCase      )
										return outputs

							lowerCamelCase_ 		=     encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
							return _forward




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ):
							lowerCamelCase_ 		=     self.config_dict[model_name]

							lowerCamelCase_ 		=     (
							    hasattr(UpperCAmelCase       ,	'''architectures'''      )
							    and isinstance(config.architectures       ,	UpperCAmelCase      )
							    and len(config.architectures      ) > 0
							)
							if not self.args.only_pretrain_model and has_model_class_in_config:
										try:
													lowerCamelCase_ 		=     config.architectures[0]
													lowerCamelCase_ 		=     __import__('''transformers'''       ,	fromlist=[model_class]      )
													lowerCamelCase_ 		=     getattr(UpperCAmelCase       ,	UpperCAmelCase      )
													lowerCamelCase_ 		=     model_cls(UpperCAmelCase      )
										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:
										lowerCamelCase_ 		=     MODEL_WITH_LM_HEAD_MAPPING[config.__class__](UpperCAmelCase      )

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

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

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

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

							lowerCamelCase_ 		=     (
							    compute_loss_and_backprob_encoder_decoder
							    if config.is_encoder_decoder
							    else compute_loss_and_backprob_encoder
							)
							return _train




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase      ):
							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(
													    UpperCAmelCase       ,	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
										lowerCamelCase_ 		=     timeit.repeat(
										    UpperCAmelCase       ,	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(UpperCAmelCase      ) / 1_0.0
							except RuntimeError as e:
										self.print_fn(f"Doesn't fit on GPU. {e}"      )
										return "N/A"





				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase      ):
							try:
										if self.args.trace_memory_line_by_line:
													lowerCamelCase_ 		=     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.'''      )
																lowerCamelCase_ 		=     '''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()
																lowerCamelCase_ 		=     nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx      )
																lowerCamelCase_ 		=     nvml.nvmlDeviceGetMemoryInfo(UpperCAmelCase      )
																lowerCamelCase_ 		=     meminfo.used
																lowerCamelCase_ 		=     Memory(UpperCAmelCase      )
																# shutdown nvml
																nvml.nvmlShutdown()
										else:
													# cpu
													lowerCamelCase_ 		=     measure_peak_memory_cpu(UpperCAmelCase      )
													lowerCamelCase_ 		=     Memory(UpperCAmelCase      ) if isinstance(UpperCAmelCase       ,	UpperCAmelCase      ) else memory_bytes

										if self.args.trace_memory_line_by_line:
													lowerCamelCase_ 		=     stop_memory_tracing(UpperCAmelCase      )
										else:
													lowerCamelCase_ 		=     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 tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()




class    __lowerCamelCase     (   unittest.TestCase ):
				a__:			Union[str, Any]									=						'hf-internal-testing/tiny-stable-diffusion-pipe'
				a__:			Tuple									=						1
				a__:			int									=						77
				a__:			List[str]									=						32
				a__:			List[Any]									=						4
				a__:			str									=						64
				a__:			Optional[Any]									=						torch.manual_seed(0 )




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase=0.9_9_9_9      ):
							lowerCamelCase_ 		=     UNetaDConditionModel.from_pretrained(self.model_id       ,	subfolder='''unet'''      )
							lowerCamelCase_ 		=     unet.to(UpperCAmelCase      )
							lowerCamelCase_ 		=     EMAModel(unet.parameters()       ,	decay=UpperCAmelCase       ,	model_cls=UpperCAmelCase       ,	model_config=unet.config      )
							return unet, ema_unet




				def 					UpperCAmelCase__	(							self      ):
							lowerCamelCase_ 		=     torch.randn(
							    self.batch_size       ,	self.num_in_channels       ,	self.latent_height       ,	self.latent_width       ,	generator=self.generator      ).to(UpperCAmelCase      )
							lowerCamelCase_ 		=     torch.randint(0       ,	1000       ,	size=(self.batch_size,)       ,	generator=self.generator      ).to(UpperCAmelCase      )
							lowerCamelCase_ 		=     torch.randn(
							    self.batch_size       ,	self.prompt_length       ,	self.text_encoder_hidden_dim       ,	generator=self.generator      ).to(UpperCAmelCase      )
							return noisy_latents, timesteps, encoder_hidden_states




				def 					UpperCAmelCase__	(							self       ,	UpperCAmelCase      ):
							lowerCamelCase_ 		=     {}
							for k, param in unet.state_dict().items():
										lowerCamelCase_ 		=     torch.randn_like(UpperCAmelCase      ) + (param * torch.randn_like(UpperCAmelCase      ))
										updated_state_dict.update({k: updated_param}      )
							unet.load_state_dict(UpperCAmelCase      )
							return unet




				def 					UpperCAmelCase__	(							self      ):
							lowerCamelCase_		,   lowerCamelCase_ 		=     self.get_models()
							# Take the first (hypothetical) EMA step.
							ema_unet.step(unet.parameters()      )
							assert ema_unet.optimization_step == 1

							# Take two more.
							for _ in range(2      ):
										ema_unet.step(unet.parameters()      )
							assert ema_unet.optimization_step == 3




				def 					UpperCAmelCase__	(							self      ):
							lowerCamelCase_		,   lowerCamelCase_ 		=     self.get_models()
							# Since the `unet` is not being updated (i.e., backprop'd)
							# there won't be any difference between the `params` of `unet`
							# and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
							ema_unet.step(unet.parameters()      )
							lowerCamelCase_ 		=     list(unet.parameters()      )
							for s_param, param in zip(ema_unet.shadow_params       ,	UpperCAmelCase      ):
										assert torch.allclose(UpperCAmelCase       ,	UpperCAmelCase      )

							# The above holds true even if we call `ema.step()` multiple times since
							# `unet` params are still not being updated.
							for _ in range(4      ):
										ema_unet.step(unet.parameters()      )
							for s_param, param in zip(ema_unet.shadow_params       ,	UpperCAmelCase      ):
										assert torch.allclose(UpperCAmelCase       ,	UpperCAmelCase      )




				def 					UpperCAmelCase__	(							self      ):
							lowerCamelCase_		,   lowerCamelCase_ 		=     self.get_models()
							# Here we simulate the parameter updates for `unet`. Since there might
							# be some parameters which are initialized to zero we take extra care to
							# initialize their values to something non-zero before the multiplication.
							lowerCamelCase_ 		=     self.simulate_backprop(UpperCAmelCase      )

							# Take the EMA step.
							ema_unet.step(unet_pseudo_updated_step_one.parameters()      )

							# Now the EMA'd parameters won't be equal to the original model parameters.
							lowerCamelCase_ 		=     list(unet_pseudo_updated_step_one.parameters()      )
							for s_param, param in zip(ema_unet.shadow_params       ,	UpperCAmelCase      ):
										assert ~torch.allclose(UpperCAmelCase       ,	UpperCAmelCase      )

							# Ensure this is the case when we take multiple EMA steps.
							for _ in range(4      ):
										ema_unet.step(unet.parameters()      )
							for s_param, param in zip(ema_unet.shadow_params       ,	UpperCAmelCase      ):
										assert ~torch.allclose(UpperCAmelCase       ,	UpperCAmelCase      )




				def 					UpperCAmelCase__	(							self      ):
							# If we call EMA step after a backpropagation consecutively for two times,
							# the shadow params from those two steps should be different.
							lowerCamelCase_		,   lowerCamelCase_ 		=     self.get_models()

							# First backprop + EMA
							lowerCamelCase_ 		=     self.simulate_backprop(UpperCAmelCase      )
							ema_unet.step(unet_step_one.parameters()      )
							lowerCamelCase_ 		=     ema_unet.shadow_params

							# Second backprop + EMA
							lowerCamelCase_ 		=     self.simulate_backprop(UpperCAmelCase      )
							ema_unet.step(unet_step_two.parameters()      )
							lowerCamelCase_ 		=     ema_unet.shadow_params

							for step_one, step_two in zip(UpperCAmelCase       ,	UpperCAmelCase      ):
										assert ~torch.allclose(UpperCAmelCase       ,	UpperCAmelCase      )




				def 					UpperCAmelCase__	(							self      ):
							# If there's no decay even if there are backprops, EMA steps
							# won't take any effect i.e., the shadow params would remain the
							# same.
							lowerCamelCase_		,   lowerCamelCase_ 		=     self.get_models(decay=0.0      )
							lowerCamelCase_ 		=     self.simulate_backprop(UpperCAmelCase      )
							ema_unet.step(unet_step_one.parameters()      )
							lowerCamelCase_ 		=     ema_unet.shadow_params

							lowerCamelCase_ 		=     self.simulate_backprop(UpperCAmelCase      )
							ema_unet.step(unet_step_two.parameters()      )
							lowerCamelCase_ 		=     ema_unet.shadow_params

							for step_one, step_two in zip(UpperCAmelCase       ,	UpperCAmelCase      ):
										assert torch.allclose(UpperCAmelCase       ,	UpperCAmelCase      )





				@skip_mps
				def 					UpperCAmelCase__	(							self      ):
							lowerCamelCase_		,   lowerCamelCase_ 		=     self.get_models()
							lowerCamelCase_		,   lowerCamelCase_		,   lowerCamelCase_ 		=     self.get_dummy_inputs()

							with tempfile.TemporaryDirectory() as tmpdir:
										ema_unet.save_pretrained(UpperCAmelCase      )
										lowerCamelCase_ 		=     UNetaDConditionModel.from_pretrained(UpperCAmelCase       ,	model_cls=UpperCAmelCase      )
										lowerCamelCase_ 		=     loaded_unet.to(unet.device      )

							# Since no EMA step has been performed the outputs should match.
							lowerCamelCase_ 		=     unet(UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ).sample
							lowerCamelCase_ 		=     loaded_unet(UpperCAmelCase       ,	UpperCAmelCase       ,	UpperCAmelCase      ).sample

							assert torch.allclose(UpperCAmelCase       ,	UpperCAmelCase       ,	atol=1e-4      )



