



import torch

from diffusers import EulerAncestralDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest





class 							lowerCAmelCase_		(		UpperCAmelCase__     ):
		SCREAMING_SNAKE_CASE_       :			Any					       =	(EulerAncestralDiscreteScheduler,)
		SCREAMING_SNAKE_CASE_       :			str					       =	1_0







		def 					a_  (    self  :   Union[str, Any]       ,		**UpperCAmelCase_  :   Any					)       ->		Tuple:


						'''simple docstring'''







						_UpperCAmelCase						:  Dict     					=	{
						    '''num_train_timesteps''': 1100,
						    '''beta_start''': 0.0_001,
						    '''beta_end''': 0.02,
						    '''beta_schedule''': '''linear''',
						}

						config.update(**__lowercase					)
						return config







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


						'''simple docstring'''







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







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


						'''simple docstring'''







						for beta_start, beta_end in zip([0.00_001, 0.0_001, 0.001]       ,		[0.0_002, 0.002, 0.02]					):
										self.check_over_configs(beta_start=__lowercase       ,		beta_end=__lowercase					)







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


						'''simple docstring'''







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







		def 					a_  (    self  :   Tuple					)       ->		Tuple:


						'''simple docstring'''







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







		def 					a_  (    self  :   Tuple					)       ->		Tuple:


						'''simple docstring'''







						_UpperCAmelCase						:  int     					=	self.scheduler_classes[0]
						_UpperCAmelCase						:  int     					=	self.get_scheduler_config()
						_UpperCAmelCase						:  List[Any]     					=	scheduler_class(**__lowercase					)

						scheduler.set_timesteps(self.num_inference_steps					)

						_UpperCAmelCase						:  str     					=	torch.manual_seed(0					)

						_UpperCAmelCase						:  Tuple     					=	self.dummy_model()
						_UpperCAmelCase						:  Dict     					=	self.dummy_sample_deter * scheduler.init_noise_sigma.cpu()
						_UpperCAmelCase						:  List[str]     					=	sample.to(__lowercase					)

						for i, t in enumerate(scheduler.timesteps					):
										_UpperCAmelCase						:  Tuple     					=	scheduler.scale_model_input(__lowercase       ,		__lowercase					)

										_UpperCAmelCase						:  str     					=	model(__lowercase       ,		__lowercase					)

										_UpperCAmelCase						:  List[Any]     					=	scheduler.step(__lowercase       ,		__lowercase       ,		__lowercase       ,		generator=__lowercase					)
										_UpperCAmelCase						:  Any     					=	output.prev_sample

						_UpperCAmelCase						:  int     					=	torch.sum(torch.abs(__lowercase					)					)
						_UpperCAmelCase						:  Dict     					=	torch.mean(torch.abs(__lowercase					)					)

						assert abs(result_sum.item() - 152.3192					) < 1E-2
						assert abs(result_mean.item() - 0.1_983					) < 1E-3







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


						'''simple docstring'''







						_UpperCAmelCase						:  int     					=	self.scheduler_classes[0]
						_UpperCAmelCase						:  Tuple     					=	self.get_scheduler_config(prediction_type='''v_prediction'''					)
						_UpperCAmelCase						:  Dict     					=	scheduler_class(**__lowercase					)

						scheduler.set_timesteps(self.num_inference_steps					)

						_UpperCAmelCase						:  List[str]     					=	torch.manual_seed(0					)

						_UpperCAmelCase						:  List[str]     					=	self.dummy_model()
						_UpperCAmelCase						:  List[Any]     					=	self.dummy_sample_deter * scheduler.init_noise_sigma
						_UpperCAmelCase						:  List[str]     					=	sample.to(__lowercase					)

						for i, t in enumerate(scheduler.timesteps					):
										_UpperCAmelCase						:  int     					=	scheduler.scale_model_input(__lowercase       ,		__lowercase					)

										_UpperCAmelCase						:  List[Any]     					=	model(__lowercase       ,		__lowercase					)

										_UpperCAmelCase						:  Any     					=	scheduler.step(__lowercase       ,		__lowercase       ,		__lowercase       ,		generator=__lowercase					)
										_UpperCAmelCase						:  int     					=	output.prev_sample

						_UpperCAmelCase						:  int     					=	torch.sum(torch.abs(__lowercase					)					)
						_UpperCAmelCase						:  int     					=	torch.mean(torch.abs(__lowercase					)					)

						assert abs(result_sum.item() - 108.4439					) < 1E-2
						assert abs(result_mean.item() - 0.1_412					) < 1E-3







		def 					a_  (    self  :   Tuple					)       ->		str:


						'''simple docstring'''







						_UpperCAmelCase						:  int     					=	self.scheduler_classes[0]
						_UpperCAmelCase						:  int     					=	self.get_scheduler_config()
						_UpperCAmelCase						:  Optional[int]     					=	scheduler_class(**__lowercase					)

						scheduler.set_timesteps(self.num_inference_steps       ,		device=__lowercase					)
						_UpperCAmelCase						:  int     					=	torch.manual_seed(0					)

						_UpperCAmelCase						:  Union[str, Any]     					=	self.dummy_model()
						_UpperCAmelCase						:  Optional[Any]     					=	self.dummy_sample_deter * scheduler.init_noise_sigma.cpu()
						_UpperCAmelCase						:  Optional[int]     					=	sample.to(__lowercase					)

						for t in scheduler.timesteps:
										_UpperCAmelCase						:  str     					=	scheduler.scale_model_input(__lowercase       ,		__lowercase					)

										_UpperCAmelCase						:  Dict     					=	model(__lowercase       ,		__lowercase					)

										_UpperCAmelCase						:  List[str]     					=	scheduler.step(__lowercase       ,		__lowercase       ,		__lowercase       ,		generator=__lowercase					)
										_UpperCAmelCase						:  int     					=	output.prev_sample

						_UpperCAmelCase						:  Any     					=	torch.sum(torch.abs(__lowercase					)					)
						_UpperCAmelCase						:  Optional[Any]     					=	torch.mean(torch.abs(__lowercase					)					)

						assert abs(result_sum.item() - 152.3192					) < 1E-2
						assert abs(result_mean.item() - 0.1_983					) < 1E-3




import torch.nn as nn

from .imports import is_fpa_available


if is_fpa_available():
						import transformer_engine.pytorch as te



def       lowerCAmelCase			(		lowerCAmelCase_			, lowerCAmelCase_=True			, lowerCAmelCase_=True			, lowerCAmelCase_=True			)->			Union[str, Any]:
				if not is_fpa_available():
								raise ImportError('''Using `convert_model` requires transformer_engine to be installed.'''			)
				for name, module in model.named_children():
								if isinstance(lowerCAmelCase_			, nn.Linear			) and to_transformer_engine and _convert_linear:
												# Return early if the linear layer weights are not multiples of 16
												if any(p % 16 != 0 for p in module.weight.shape			):
																return
												lowerCAmelCase_			:   Optional[int]     					=					module.bias is not None
												lowerCAmelCase_			:   Any     					=					te.Linear(
												    module.in_features			, module.out_features			, bias=lowerCAmelCase_			, params_dtype=module.weight.dtype			)
												lowerCAmelCase_			:   Dict     					=					module.weight.data.clone()
												if has_bias:
																lowerCAmelCase_			:   Any     					=					module.bias.data.clone()

												setattr(lowerCAmelCase_			, lowerCAmelCase_			, lowerCAmelCase_			)
								elif isinstance(lowerCAmelCase_			, nn.LayerNorm			) and to_transformer_engine and _convert_ln:
												lowerCAmelCase_			:   List[str]     					=					te.LayerNorm(module.normalized_shape[0]			, eps=module.eps			, params_dtype=module.weight.dtype			)
												lowerCAmelCase_			:   Union[str, Any]     					=					module.weight.data.clone()
												lowerCAmelCase_			:   List[Any]     					=					module.bias.data.clone()

												setattr(lowerCAmelCase_			, lowerCAmelCase_			, lowerCAmelCase_			)
								elif isinstance(lowerCAmelCase_			, te.Linear			) and not to_transformer_engine and _convert_linear:
												lowerCAmelCase_			:   Dict     					=					module.bias is not None
												lowerCAmelCase_			:   Union[str, Any]     					=					nn.Linear(
												    module.in_features			, module.out_features			, bias=lowerCAmelCase_			, params_dtype=module.weight.dtype			)
												lowerCAmelCase_			:   Optional[Any]     					=					module.weight.data.clone()
												if has_bias:
																lowerCAmelCase_			:   Union[str, Any]     					=					module.bias.data.clone()

												setattr(lowerCAmelCase_			, lowerCAmelCase_			, lowerCAmelCase_			)
								elif isinstance(lowerCAmelCase_			, te.LayerNorm			) and not to_transformer_engine and _convert_ln:
												lowerCAmelCase_			:   Union[str, Any]     					=					nn.LayerNorm(module.normalized_shape[0]			, eps=module.eps			, params_dtype=module.weight.dtype			)
												lowerCAmelCase_			:   List[str]     					=					module.weight.data.clone()
												lowerCAmelCase_			:   List[Any]     					=					module.bias.data.clone()

												setattr(lowerCAmelCase_			, lowerCAmelCase_			, lowerCAmelCase_			)
								else:
												convert_model(
												    lowerCAmelCase_			, to_transformer_engine=lowerCAmelCase_			, _convert_linear=_convert_linear			, _convert_ln=_convert_ln			, )




def       lowerCAmelCase			(		lowerCAmelCase_			)->			str:
				if not is_fpa_available():
								raise ImportError('''Using `has_transformer_engine_layers` requires transformer_engine to be installed.'''			)
				for m in model.modules():
								if isinstance(lowerCAmelCase_			, (te.LayerNorm, te.Linear)			):
												return True
				return False