


'''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				):
	_lowerCamelCase :       List[Any]         =			"""hf-internal-testing/tiny-stable-diffusion-pipe"""
	_lowerCamelCase :       Optional[int]         =			1
	_lowerCamelCase :       List[str]         =			77
	_lowerCamelCase :       Any         =			32
	_lowerCamelCase :       str         =			4
	_lowerCamelCase :       Union[str, Any]         =			64
	_lowerCamelCase :       Optional[Any]         =			torch.manual_seed(0				)



	def 					lowercase  (       self			,       snake_case__=0.9_9_9_9     ):
		lowerCAmelCase       :     str   				=			UNetaDConditionModel.from_pretrained(self.model_id			,       subfolder='unet'     )
		lowerCAmelCase       :     int   				=			unet.to(__UpperCAmelCase     )
		lowerCAmelCase       :     Union[str, Any]   				=			EMAModel(unet.parameters()			,       decay=__UpperCAmelCase			,       model_cls=__UpperCAmelCase			,       model_config=unet.config     )
		return unet, ema_unet



	def 					lowercase  (       self     ):
		lowerCAmelCase       :     Tuple   				=			torch.randn(
		    self.batch_size			,       self.num_in_channels			,       self.latent_height			,       self.latent_width			,       generator=self.generator     ).to(__UpperCAmelCase     )
		lowerCAmelCase       :     Optional[int]   				=			torch.randint(0			,       1000			,       size=(self.batch_size,)			,       generator=self.generator     ).to(__UpperCAmelCase     )
		lowerCAmelCase       :     Tuple   				=			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 					lowercase  (       self			,       snake_case__     ):
		lowerCAmelCase       :     str   				=			{}
		for k, param in unet.state_dict().items():
			lowerCAmelCase       :     Optional[int]   				=			torch.randn_like(__UpperCAmelCase     ) + (param * torch.randn_like(__UpperCAmelCase     ))
			updated_state_dict.update({k: updated_param}     )
		unet.load_state_dict(__UpperCAmelCase     )
		return unet



	def 					lowercase  (       self     ):
		lowerCAmelCase       :     Any   				=			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 					lowercase  (       self     ):
		lowerCAmelCase       :     Any   				=			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       :     Optional[int]   				=			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 					lowercase  (       self     ):
		lowerCAmelCase       :     Any   				=			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       :     str   				=			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[Any]   				=			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 					lowercase  (       self     ):
		# If we call EMA step after a backpropagation consecutively for two times,
		# the shadow params from those two steps should be different.
		lowerCAmelCase       :     Any   				=			self.get_models()

		# First backprop + EMA
		lowerCAmelCase       :     List[str]   				=			self.simulate_backprop(__UpperCAmelCase     )
		ema_unet.step(unet_step_one.parameters()     )
		lowerCAmelCase       :     Dict   				=			ema_unet.shadow_params

		# Second backprop + EMA
		lowerCAmelCase       :     Tuple   				=			self.simulate_backprop(__UpperCAmelCase     )
		ema_unet.step(unet_step_two.parameters()     )
		lowerCAmelCase       :     Optional[Any]   				=			ema_unet.shadow_params

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



	def 					lowercase  (       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       :     Dict   				=			self.get_models(decay=0.0     )
		lowerCAmelCase       :     Dict   				=			self.simulate_backprop(__UpperCAmelCase     )
		ema_unet.step(unet_step_one.parameters()     )
		lowerCAmelCase       :     str   				=			ema_unet.shadow_params

		lowerCAmelCase       :     Optional[Any]   				=			self.simulate_backprop(__UpperCAmelCase     )
		ema_unet.step(unet_step_two.parameters()     )
		lowerCAmelCase       :     Dict   				=			ema_unet.shadow_params

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



	@skip_mps
	def 					lowercase  (       self     ):
		lowerCAmelCase       :     Dict   				=			self.get_models()
		lowerCAmelCase       :     Any   				=			self.get_dummy_inputs()

		with tempfile.TemporaryDirectory() as tmpdir:
			ema_unet.save_pretrained(__UpperCAmelCase     )
			lowerCAmelCase       :     Union[str, Any]   				=			UNetaDConditionModel.from_pretrained(__UpperCAmelCase			,       model_cls=__UpperCAmelCase     )
			lowerCAmelCase       :     Any   				=			loaded_unet.to(unet.device     )

		# Since no EMA step has been performed the outputs should match.
		lowerCAmelCase       :     Any   				=			unet(__UpperCAmelCase			,       __UpperCAmelCase			,       __UpperCAmelCase     ).sample
		lowerCAmelCase       :     Tuple   				=			loaded_unet(__UpperCAmelCase			,       __UpperCAmelCase			,       __UpperCAmelCase     ).sample

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




"""simple docstring"""







from ..utils import DummyObject, requires_backends






class 		_A    (    metaclass=_a    ):







    """simple docstring"""




    UpperCAmelCase			:				int      =			["""torch""", """transformers""", """invisible_watermark"""]






    def __init__(   self							:       int							,					*__UpperCAmelCase							:       List[Any]							,					**__UpperCAmelCase							:       Optional[int]):
      requires_backends(self							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       Optional[Any]							,					*__UpperCAmelCase							:       Any							,					**__UpperCAmelCase							:       int):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       List[Any]							,					*__UpperCAmelCase							:       Optional[Any]							,					**__UpperCAmelCase							:       Tuple):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






class 		_A    (    metaclass=_a    ):







    """simple docstring"""




    UpperCAmelCase			:				Optional[int]      =			["""torch""", """transformers""", """invisible_watermark"""]






    def __init__(   self							:       int							,					*__UpperCAmelCase							:       Tuple							,					**__UpperCAmelCase							:       Tuple):
      requires_backends(self							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       int							,					*__UpperCAmelCase							:       str							,					**__UpperCAmelCase							:       Optional[int]):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       Any							,					*__UpperCAmelCase							:       Optional[int]							,					**__UpperCAmelCase							:       Union[str, Any]):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






