



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


__SCREAMING_SNAKE_CASE     :  Dict										=					namedtuple('''ExampleNamedTuple''', '''a b c''')


class 							lowerCamelCase_(      unittest.TestCase			):





					'''simple docstring'''



					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = torch.randn(5 ,				2      )
									_lowerCamelCase					     = torch.device('''cuda'''      ) if torch.cuda.is_available() else torch.device('''cpu'''      )

									_lowerCamelCase					     = send_to_device(lowercase__ ,				lowercase__      )
									self.assertTrue(torch.equal(resulta.cpu() ,				lowercase__      )      )

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

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

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

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

					def 	snake_case__					(			self      ):






									with self.assertRaises(lowercase__      ) as cm:
													_lowerCamelCase					     = recursively_apply(torch.tensor ,				(torch.tensor(1      ), 1) ,				error_on_other_type=lowercase__      )
									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 	snake_case__					(			self      ):
									_lowerCamelCase					     = torch.tensor([1, 2, 3, 4, 5]      )
									self.assertEqual(listify(lowercase__      ) ,				[1, 2, 3, 4, 5]      )

									_lowerCamelCase					     = torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )
									self.assertEqual(listify(lowercase__      ) ,				[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )

									_lowerCamelCase					     = torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )
									self.assertEqual(
									    listify(lowercase__      ) ,				[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]      )

					def 	snake_case__					(			self      ):






									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 	snake_case__					(			self      ):
									_lowerCamelCase					     = RegressionModel()
									_lowerCamelCase					     = model.forward
									_lowerCamelCase					     = convert_outputs_to_fpaa(model.forward      )
									_lowerCamelCase					     = extract_model_from_parallel(lowercase__ ,				keep_fpaa_wrapper=lowercase__      )
									_lowerCamelCase					     = pickle.dumps(lowercase__      )

					@require_cuda
					def 	snake_case__					(			self      ):
									_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(lowercase__ ,				keep_fpaa_wrapper=lowercase__      )
									_lowerCamelCase					     = pickle.dumps(lowercase__      )

					@require_cuda
					@require_torch_min_version(version='''2.0'''      )
					def 	snake_case__					(			self      ):
									_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 ,				1_0      ).cuda()
									_lowerCamelCase					     = model(lowercase__      )

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = RegressionModel()
									# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
									_lowerCamelCase					     = torch.nn.parallel.DataParallel(lowercase__      )
									_lowerCamelCase					     = extract_model_from_parallel(lowercase__      )

									self.assertEqual(lowercase__ ,				lowercase__      )

					@require_torch_min_version(version='''2.0'''      )
					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = RegressionModel()
									_lowerCamelCase					     = torch.compile(lowercase__      )

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

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





					def 	snake_case__					(			self      ):
									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''']      )      )



'''simple docstring'''






import torch

from diffusers import EulerAncestralDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest
class 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase      ):
    _A			    =							(EulerAncestralDiscreteScheduler,)
    _A			    =							10




    def   __lowerCamelCase      (					self				,     **lowercase__			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			List[Any]     =       {
              "num_train_timesteps": 1100,
              "beta_start": 0.0001,
              "beta_end": 0.02,
              "beta_schedule": "linear",
          }

          config.update(**lowercase__			)
          return config




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for timesteps in [10, 50, 100, 1000]:
                self.check_over_configs(num_train_timesteps=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for beta_start, beta_end in zip([0.00001, 0.0001, 0.001]				,     [0.0002, 0.002, 0.02]			):
                self.check_over_configs(beta_start=lowercase__				,     beta_end=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for schedule in ["linear", "scaled_linear"]:
                self.check_over_configs(beta_schedule=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          for prediction_type in ["epsilon", "v_prediction"]:
                self.check_over_configs(prediction_type=lowercase__			)




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			int     =       self.scheduler_classes[0]
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.get_scheduler_config()
          SCREAMING_SNAKE_CASE_      :			Any     =       scheduler_class(**lowercase__			)

          scheduler.set_timesteps(self.num_inference_steps			)

          SCREAMING_SNAKE_CASE_      :			int     =       torch.manual_seed(0			)

          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       self.dummy_model()
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       self.dummy_sample_deter * scheduler.init_noise_sigma.cpu()
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       sample.to(lowercase__			)

          for i, t in enumerate(scheduler.timesteps			):
                SCREAMING_SNAKE_CASE_      :			Tuple     =       scheduler.scale_model_input(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			List[str]     =       model(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Dict     =       scheduler.step(lowercase__				,     lowercase__				,     lowercase__				,     generator=lowercase__			)
                SCREAMING_SNAKE_CASE_      :			Tuple     =       output.prev_sample

          SCREAMING_SNAKE_CASE_      :			List[str]     =       torch.sum(torch.abs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			Tuple     =       torch.mean(torch.abs(lowercase__			)			)

          assert abs(result_sum.item() - 152.3192			) < 1e-2
          assert abs(result_mean.item() - 0.1983			) < 1e-3




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			int     =       self.scheduler_classes[0]
          SCREAMING_SNAKE_CASE_      :			Tuple     =       self.get_scheduler_config(prediction_type="v_prediction"			)
          SCREAMING_SNAKE_CASE_      :			str     =       scheduler_class(**lowercase__			)

          scheduler.set_timesteps(self.num_inference_steps			)

          SCREAMING_SNAKE_CASE_      :			Dict     =       torch.manual_seed(0			)

          SCREAMING_SNAKE_CASE_      :			Tuple     =       self.dummy_model()
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       self.dummy_sample_deter * scheduler.init_noise_sigma
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       sample.to(lowercase__			)

          for i, t in enumerate(scheduler.timesteps			):
                SCREAMING_SNAKE_CASE_      :			List[str]     =       scheduler.scale_model_input(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			List[Any]     =       model(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       scheduler.step(lowercase__				,     lowercase__				,     lowercase__				,     generator=lowercase__			)
                SCREAMING_SNAKE_CASE_      :			List[Any]     =       output.prev_sample

          SCREAMING_SNAKE_CASE_      :			List[Any]     =       torch.sum(torch.abs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			Dict     =       torch.mean(torch.abs(lowercase__			)			)

          assert abs(result_sum.item() - 108.4439			) < 1e-2
          assert abs(result_mean.item() - 0.1412			) < 1e-3







    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       self.scheduler_classes[0]
          SCREAMING_SNAKE_CASE_      :			List[str]     =       self.get_scheduler_config()
          SCREAMING_SNAKE_CASE_      :			Tuple     =       scheduler_class(**lowercase__			)

          scheduler.set_timesteps(self.num_inference_steps				,     device=lowercase__			)
          SCREAMING_SNAKE_CASE_      :			Any     =       torch.manual_seed(0			)

          SCREAMING_SNAKE_CASE_      :			List[Any]     =       self.dummy_model()
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.dummy_sample_deter * scheduler.init_noise_sigma.cpu()
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       sample.to(lowercase__			)

          for t in scheduler.timesteps:
                SCREAMING_SNAKE_CASE_      :			Dict     =       scheduler.scale_model_input(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       model(lowercase__				,     lowercase__			)

                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       scheduler.step(lowercase__				,     lowercase__				,     lowercase__				,     generator=lowercase__			)
                SCREAMING_SNAKE_CASE_      :			List[str]     =       output.prev_sample

          SCREAMING_SNAKE_CASE_      :			Any     =       torch.sum(torch.abs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       torch.mean(torch.abs(lowercase__			)			)

          assert abs(result_sum.item() - 152.3192			) < 1e-2
          assert abs(result_mean.item() - 0.1983			) < 1e-3

