'''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 	SCREAMING_SNAKE_CASE      (    unittest.TestCase):







		"""simple docstring"""
		lowercase :   int							=				'hf-internal-testing/tiny-stable-diffusion-pipe'
		lowercase :   Optional[Any]							=				1
		lowercase :   Any							=				77
		lowercase :   Tuple							=				32
		lowercase :   str							=				4
		lowercase :   List[Any]							=				64
		lowercase :   Union[str, Any]							=				torch.manual_seed(0)






		def 		UpperCamelCase__							( self		,    __A=0.9_999					)			->  Optional[Any]:
								_lowerCAmelCase					    =UNetaDConditionModel.from_pretrained(self.model_id		,    subfolder='unet'					)
								_lowerCAmelCase					    =unet.to(__A					)
								_lowerCAmelCase					    =EMAModel(unet.parameters()		,    decay=__A		,    model_cls=__A		,    model_config=unet.config					)
								return unet, ema_unet






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






		def 		UpperCamelCase__							( self					)			->  Any:
								_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					)			->  List[Any]:
								_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		,    __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 		UpperCamelCase__							( self					)			->  Any:
								_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(__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.
								_lowerCAmelCase					    =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 		UpperCamelCase__							( self					)			->  str:
								# 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(__A					)
								ema_unet.step(unet_step_one.parameters()					)
								_lowerCAmelCase					    =ema_unet.shadow_params

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

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






		def 		UpperCamelCase__							( self					)			->  Tuple:
								# 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(__A					)
								ema_unet.step(unet_step_one.parameters()					)
								_lowerCAmelCase					    =ema_unet.shadow_params

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

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






		@skip_mps
		def 		UpperCamelCase__							( self					)			->  int:
								_lowerCAmelCase    ,					_lowerCAmelCase					    =self.get_models()
								_lowerCAmelCase    ,					_lowerCAmelCase    ,					_lowerCAmelCase					    =self.get_dummy_inputs()

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

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

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






'''simple docstring'''

import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


lowercase_  		=			namedtuple('''ExampleNamedTuple''', '''a b c''')
class 	SCREAMING_SNAKE_CASE      (    unittest.TestCase):







		"""simple docstring"""






		def 		UpperCamelCase__							( self					)			->  int:
								_lowerCAmelCase					    =torch.randn(5		,    2					)
								_lowerCAmelCase					    =torch.device('cuda'					) if torch.cuda.is_available() else torch.device('cpu'					)

								_lowerCAmelCase					    =send_to_device(__A		,    __A					)
								self.assertTrue(torch.equal(resulta.cpu()		,    __A					)					)

								_lowerCAmelCase					    =send_to_device((tensor, [tensor, tensor], 1)		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta[0].cpu()		,    __A					)					)
								self.assertIsInstance(resulta[1]		,    __A					)
								self.assertTrue(torch.equal(resulta[1][0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta[1][1].cpu()		,    __A					)					)
								self.assertEqual(resulta[2]		,    1					)

								_lowerCAmelCase					    =send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta['a'].cpu()		,    __A					)					)
								self.assertIsInstance(resulta['b']		,    __A					)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta['b'][1].cpu()		,    __A					)					)
								self.assertEqual(resulta['c']		,    1					)

								_lowerCAmelCase					    =send_to_device(ExampleNamedTuple(a=__A		,    b=[tensor, tensor]		,    c=1					)		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta.a.cpu()		,    __A					)					)
								self.assertIsInstance(resulta.b		,    __A					)
								self.assertTrue(torch.equal(resulta.b[0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta.b[1].cpu()		,    __A					)					)
								self.assertEqual(resulta.c		,    1					)

								_lowerCAmelCase					    =send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}					)		,    __A					)
								self.assertIsInstance(__A		,    __A					)
								self.assertTrue(torch.equal(resulta['a'].cpu()		,    __A					)					)
								self.assertIsInstance(resulta['b']		,    __A					)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()		,    __A					)					)
								self.assertTrue(torch.equal(resulta['b'][1].cpu()		,    __A					)					)
								self.assertEqual(resulta['c']		,    1					)






		def 		UpperCamelCase__							( self					)			->  Optional[int]:
								with self.assertRaises(__A					) as cm:
														_lowerCAmelCase					    =recursively_apply(torch.tensor		,    (torch.tensor(1					), 1)		,    error_on_other_type=__A					)
								self.assertEqual(
								    str(cm.exception					)		,    'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'		,    )






		def 		UpperCamelCase__							( self					)			->  int:
								_lowerCAmelCase					    =torch.tensor([1, 2, 3, 4, 5]					)
								self.assertEqual(listify(__A					)		,    [1, 2, 3, 4, 5]					)

								_lowerCAmelCase					    =torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)
								self.assertEqual(listify(__A					)		,    [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)

								_lowerCAmelCase					    =torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]					)
								self.assertEqual(
								    listify(__A					)		,    [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]					)






		def 		UpperCamelCase__							( self					)			->  List[Any]:
								with patch_environment(aa=1		,    BB=2					):
														self.assertEqual(os.environ.get('AA'					)		,    '1'					)
														self.assertEqual(os.environ.get('BB'					)		,    '2'					)

								self.assertNotIn('AA'		,    os.environ					)
								self.assertNotIn('BB'		,    os.environ					)






		def 		UpperCamelCase__							( self					)			->  Union[str, Any]:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =model.forward
								_lowerCAmelCase					    =convert_outputs_to_fpaa(model.forward					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A		,    keep_fpaa_wrapper=__A					)
								_lowerCAmelCase					    =pickle.dumps(__A					)






		@require_cuda
		def 		UpperCamelCase__							( self					)			->  Dict:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =model.forward
								_lowerCAmelCase					    =torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
								_lowerCAmelCase					    =convert_outputs_to_fpaa(model.forward					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A		,    keep_fpaa_wrapper=__A					)
								_lowerCAmelCase					    =pickle.dumps(__A					)






		@require_cuda
		@require_torch_min_version(version='2.0'					)
		def 		UpperCamelCase__							( self					)			->  int:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =model.forward
								_lowerCAmelCase					    =torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
								_lowerCAmelCase					    =convert_outputs_to_fpaa(model.forward					)
								_lowerCAmelCase					    =torch.compile(model.forward		,    backend='inductor'					)
								_lowerCAmelCase					    =torch.randn(4		,    10					).cuda()
								_lowerCAmelCase					    =model(__A					)






		def 		UpperCamelCase__							( self					)			->  Optional[int]:
								_lowerCAmelCase					    =RegressionModel()
								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								_lowerCAmelCase					    =torch.nn.parallel.DataParallel(__A					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A					)

								self.assertEqual(__A		,    __A					)






		@require_torch_min_version(version='2.0'					)
		def 		UpperCamelCase__							( self					)			->  str:
								_lowerCAmelCase					    =RegressionModel()
								_lowerCAmelCase					    =torch.compile(__A					)

								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								_lowerCAmelCase					    =torch.nn.parallel.DataParallel(__A					)
								_lowerCAmelCase					    =torch.compile(__A					)
								_lowerCAmelCase					    =extract_model_from_parallel(__A					)

								self.assertEqual(compiled_model._orig_mod		,    compiled_model_unwrapped._orig_mod					)






		def 		UpperCamelCase__							( self					)			->  Tuple:
								self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3]					)]					)		,    torch.device('cpu'					)					)
								self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3]					)}					)		,    torch.device('cpu'					)					)
								self.assertIsNone(find_device([1, 'a']					)					)





