


"""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 snake_case   (	unittest.TestCase):
		__UpperCamelCase							     =      'hf-internal-testing/tiny-stable-diffusion-pipe'
		__UpperCamelCase							     =      1
		__UpperCamelCase							     =      77
		__UpperCamelCase							     =      32
		__UpperCamelCase							     =      4
		__UpperCamelCase							     =      __UpperCamelCase							     =      64
		__UpperCamelCase							     =      torch.manual_seed(0)







		def 			a_			(       self   :							Optional[int]     , a__   :							Optional[int]=0.9_9_9_9		)							->    Any:





						'''simple docstring'''







						_A       	=      UNetaDConditionModel.from_pretrained(self.model_id     , subfolder="unet"		)
						_A       	=      unet.to(__A		)
						_A       	=      EMAModel(unet.parameters()     , decay=__A     , model_cls=__A     , model_config=unet.config		)
						return unet, ema_unet







		def 			a_			(       self   :							Dict		)							->    List[Any]:





						'''simple docstring'''







						_A       	=      torch.randn(
						    self.batch_size     , self.num_in_channels     , self.latent_height     , self.latent_width     , generator=self.generator		).to(__A		)
						_A       	=      torch.randint(0     , 10_00     , size=(self.batch_size,)     , generator=self.generator		).to(__A		)
						_A       	=      torch.randn(
						    self.batch_size     , self.prompt_length     , self.text_encoder_hidden_dim     , generator=self.generator		).to(__A		)
						return noisy_latents, timesteps, encoder_hidden_states







		def 			a_			(       self   :							Dict     , a__   :							Optional[Any]		)							->    Any:





						'''simple docstring'''







						_A       	=      {}
						for k, param in unet.state_dict().items():
										_A       	=      torch.randn_like(__A		) + (param * torch.randn_like(__A		))
										updated_state_dict.update({k: updated_param}		)
						unet.load_state_dict(__A		)
						return unet







		def 			a_			(       self   :							Tuple		)							->    List[Any]:





						'''simple docstring'''







						_A       	=      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 			a_			(       self   :							List[Any]		)							->    Dict:





						'''simple docstring'''







						_A       	=      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()		)
						_A       	=      list(unet.parameters()		)
						for s_param, param in zip(ema_unet.shadow_params     , __A		):
										assert torch.allclose(__A     , __A		)

						# 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     , __A		):
										assert torch.allclose(__A     , __A		)







		def 			a_			(       self   :							Optional[Any]		)							->    Any:





						'''simple docstring'''







						_A       	=      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.
						_A       	=      self.simulate_backprop(__A		)

						# 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.
						_A       	=      list(unet_pseudo_updated_step_one.parameters()		)
						for s_param, param in zip(ema_unet.shadow_params     , __A		):
										assert ~torch.allclose(__A     , __A		)

						# 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     , __A		):
										assert ~torch.allclose(__A     , __A		)







		def 			a_			(       self   :							List[Any]		)							->    int:





						'''simple docstring'''







						_A       	=      self.get_models()

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

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

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







		def 			a_			(       self   :							Union[str, Any]		)							->    List[Any]:





						'''simple docstring'''







						_A       	=      self.get_models(decay=0.0		)
						_A       	=      self.simulate_backprop(__A		)
						ema_unet.step(unet_step_one.parameters()		)
						_A       	=      ema_unet.shadow_params

						_A       	=      self.simulate_backprop(__A		)
						ema_unet.step(unet_step_two.parameters()		)
						_A       	=      ema_unet.shadow_params

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







		@skip_mps
		def 			a_			(       self   :							List[str]		)							->    Tuple:





						'''simple docstring'''







						_A       	=      self.get_models()
						_A       	=      self.get_dummy_inputs()

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

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

						assert torch.allclose(__A     , __A     , atol=1E-4		)


from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


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

_lowerCamelCase							:  Optional[int]     =				{
    "facebook/m2m100_418M": "https://huggingface.co/facebook/m2m100_418M/resolve/main/config.json",
    # See all M2M100 models at https://huggingface.co/models?filter=m2m_100
}





class    __UpperCAmelCase	(			lowerCamelCase__		):
   UpperCamelCase  	=	"""m2m_100"""
   UpperCamelCase  	=	["""past_key_values"""]
   UpperCamelCase  	=	{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}



   def __init__(				self				:	int,						__A				:	Optional[Any]=1_2_8_1_1_2,						__A				:	str=1_0_2_4,						__A				:	Tuple=1_2,						__A				:	Any=4_0_9_6,						__A				:	str=1_6,						__A				:	List[str]=1_2,						__A				:	int=4_0_9_6,						__A				:	List[Any]=1_6,						__A				:	Tuple=0.0_5,						__A				:	Tuple=0.0_5,						__A				:	List[Any]=True,						__A				:	int=True,						__A				:	str="relu",						__A				:	str=1_0_2_4,						__A				:	str=0.1,						__A				:	str=0.1,						__A				:	List[str]=0.0,						__A				:	List[str]=0.0_2,						__A				:	Any=2,						__A				:	List[str]=True,						__A				:	Tuple=1,						__A				:	Optional[Any]=0,						__A				:	Optional[Any]=2,						**__A				:	Optional[Any],						):
     UpperCAmelCase			:				str										=					vocab_size
     UpperCAmelCase			:				Optional[Any]										=					max_position_embeddings
     UpperCAmelCase			:				List[str]										=					d_model
     UpperCAmelCase			:				Any										=					encoder_ffn_dim
     UpperCAmelCase			:				Optional[int]										=					encoder_layers
     UpperCAmelCase			:				List[Any]										=					encoder_attention_heads
     UpperCAmelCase			:				str										=					decoder_ffn_dim
     UpperCAmelCase			:				Any										=					decoder_layers
     UpperCAmelCase			:				List[str]										=					decoder_attention_heads
     UpperCAmelCase			:				str										=					dropout
     UpperCAmelCase			:				int										=					attention_dropout
     UpperCAmelCase			:				Dict										=					activation_dropout
     UpperCAmelCase			:				Tuple										=					activation_function
     UpperCAmelCase			:				Any										=					init_std
     UpperCAmelCase			:				Union[str, Any]										=					encoder_layerdrop
     UpperCAmelCase			:				List[Any]										=					decoder_layerdrop
     UpperCAmelCase			:				Optional[Any]										=					use_cache
     UpperCAmelCase			:				str										=					encoder_layers
     UpperCAmelCase			:				List[Any]										=					scale_embedding  # scale factor will be sqrt(d_model) if True

     super().__init__(
         pad_token_id=__A,						bos_token_id=__A,						eos_token_id=__A,						is_encoder_decoder=__A,						decoder_start_token_id=__A,						**__A,						)






class    __UpperCAmelCase	(			lowerCamelCase__		):



   @property
   def 			__magic_name__       (				self				:	List[str]		):
     UpperCAmelCase			:				List[Any]										=					OrderedDict(
         [
             ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
             ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
         ]		)

     if self.use_past:
       UpperCAmelCase			:				List[str]										=					{0: '''batch'''}
       UpperCAmelCase			:				Any										=					{0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
     else:
       UpperCAmelCase			:				Any										=					{0: '''batch''', 1: '''decoder_sequence'''}
       UpperCAmelCase			:				List[Any]										=					{0: '''batch''', 1: '''decoder_sequence'''}

     if self.use_past:
       self.fill_with_past_key_values_(__A,						direction='''inputs'''		)
     return common_inputs



   def 			__magic_name__       (				self				:	Dict,						__A				:	PreTrainedTokenizer,						__A				:	int = -1,						__A				:	int = -1,						__A				:	bool = False,						__A				:	Optional[TensorType] = None,						):
     # Copied from OnnxConfig.generate_dummy_inputs
     # Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
     # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
     UpperCAmelCase			:				List[str]										=					compute_effective_axis_dimension(
         __A,						fixed_dimension=OnnxConfig.default_fixed_batch,						num_token_to_add=0		)

     # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
     UpperCAmelCase			:				Optional[int]										=					tokenizer.num_special_tokens_to_add(__A		)
     UpperCAmelCase			:				int										=					compute_effective_axis_dimension(
         __A,						fixed_dimension=OnnxConfig.default_fixed_sequence,						num_token_to_add=__A		)

     # Generate dummy inputs according to compute batch and sequence
     UpperCAmelCase			:				List[str]										=					[''' '''.join([tokenizer.unk_token]		) * seq_length] * batch_size
     UpperCAmelCase			:				Tuple										=					dict(tokenizer(__A,						return_tensors=__A		)		)
     return common_inputs



   def 			__magic_name__       (				self				:	Any,						__A				:	PreTrainedTokenizer,						__A				:	int = -1,						__A				:	int = -1,						__A				:	bool = False,						__A				:	Optional[TensorType] = None,						):
     UpperCAmelCase			:				int										=					self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
         __A,						__A,						__A,						__A,						__A		)

     # Generate decoder inputs
     UpperCAmelCase			:				Union[str, Any]										=					seq_length if not self.use_past else 1
     UpperCAmelCase			:				Optional[int]										=					self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
         __A,						__A,						__A,						__A,						__A		)
     UpperCAmelCase			:				Dict										=					{F'''decoder_{name}''': tensor for name, tensor in decoder_inputs.items()}
     UpperCAmelCase			:				Optional[int]										=					dict(**__A,						**__A		)

     if self.use_past:
       if not is_torch_available():
         raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''		)
       else:
         import torch
       UpperCAmelCase					,		UpperCAmelCase			:				List[str]										=					common_inputs['''input_ids'''].shape
       UpperCAmelCase			:				Dict										=					common_inputs['''decoder_input_ids'''].shape[1]
       UpperCAmelCase					,		UpperCAmelCase			:				Dict										=					self.num_attention_heads
       UpperCAmelCase			:				Union[str, Any]										=					(
           batch,
           num_encoder_attention_heads,
           encoder_seq_length,
           self._config.hidden_size // num_encoder_attention_heads,
       )
       UpperCAmelCase			:				str										=					decoder_seq_length + 3
       UpperCAmelCase			:				Union[str, Any]										=					(
           batch,
           num_decoder_attention_heads,
           decoder_past_length,
           self._config.hidden_size // num_decoder_attention_heads,
       )

       UpperCAmelCase			:				Union[str, Any]										=					torch.cat(
           [common_inputs['''decoder_attention_mask'''], torch.ones(__A,						__A		)],						dim=1		)

       UpperCAmelCase			:				Any										=					[]
       # If the number of encoder and decoder layers are present in the model configuration, both are considered
       UpperCAmelCase					,		UpperCAmelCase			:				str										=					self.num_layers
       UpperCAmelCase			:				Any										=					min(__A,						__A		)
       UpperCAmelCase			:				Dict										=					max(__A,						__A		) - min_num_layers
       UpperCAmelCase			:				Tuple										=					'''encoder''' if num_encoder_layers > num_decoder_layers else '''decoder'''

       for _ in range(__A		):
         common_inputs["past_key_values"].append(
             (
                 torch.zeros(__A		),
                 torch.zeros(__A		),
                 torch.zeros(__A		),
                 torch.zeros(__A		),
             )		)
       # TODO: test this.
       UpperCAmelCase			:				List[str]										=					encoder_shape if remaining_side_name == '''encoder''' else decoder_shape
       for _ in range(__A,						__A		):
         common_inputs["past_key_values"].append((torch.zeros(__A		), torch.zeros(__A		))		)
     return common_inputs





   UpperCamelCase  	=	_generate_dummy_inputs_for_default_and_seqaseq_lm





