





"""simple docstring"""






import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest




class 					__lowercase      (      _UpperCamelCase				):







					'''simple docstring'''




					__lowerCAmelCase	=	(KDPMaDiscreteScheduler,)
					__lowerCAmelCase	=	10







					def 			_lowerCamelCase		(    self						,      **_UpperCAmelCase  ):
									__a			:    Dict      							=	{
									    '''num_train_timesteps''': 1100,
									    '''beta_start''': 0.0_0_0_1,
									    '''beta_end''': 0.0_2,
									    '''beta_schedule''': '''linear''',
									}

									config.update(**_UpperCAmelCase  )
									return config







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







					def 			_lowerCamelCase		(    self  ):
									for beta_start, beta_end in zip([0.0_0_0_0_1, 0.0_0_0_1, 0.0_0_1]						,      [0.0_0_0_2, 0.0_0_2, 0.0_2]  ):
													self.check_over_configs(beta_start=_UpperCAmelCase						,      beta_end=_UpperCAmelCase  )







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







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







					def 			_lowerCamelCase		(    self  ):
									__a			:    List[Any]      							=	self.scheduler_classes[0]
									__a			:    Union[str, Any]      							=	self.get_scheduler_config(prediction_type='''v_prediction'''  )
									__a			:    str      							=	scheduler_class(**_UpperCAmelCase  )

									scheduler.set_timesteps(self.num_inference_steps  )

									__a			:    Optional[int]      							=	self.dummy_model()
									__a			:    Any      							=	self.dummy_sample_deter * scheduler.init_noise_sigma
									__a			:    Tuple      							=	sample.to(_UpperCAmelCase  )

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

													__a			:    int      							=	model(_UpperCAmelCase						,      _UpperCAmelCase  )

													__a			:    List[Any]      							=	scheduler.step(_UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  )
													__a			:    str      							=	output.prev_sample

									__a			:    Union[str, Any]      							=	torch.sum(torch.abs(_UpperCAmelCase  )  )
									__a			:    Any      							=	torch.mean(torch.abs(_UpperCAmelCase  )  )

									if torch_device in ["cpu", "mps"]:
													assert abs(result_sum.item() - 4.6_9_3_4e-0_7  ) < 1e-2
													assert abs(result_mean.item() - 6.1_1_1_2e-1_0  ) < 1e-3
									else:
													# CUDA
													assert abs(result_sum.item() - 4.6_9_3_4_2_8_6_5_0_1_7_0_9_7_2e-0_7  ) < 1e-2
													assert abs(result_mean.item() - 0.0_0_0_2  ) < 1e-3







					def 			_lowerCamelCase		(    self  ):
									if torch_device == "mps":
													return
									__a			:    Tuple      							=	self.scheduler_classes[0]
									__a			:    Union[str, Any]      							=	self.get_scheduler_config()
									__a			:    Optional[Any]      							=	scheduler_class(**_UpperCAmelCase  )

									scheduler.set_timesteps(self.num_inference_steps  )

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

									for i, t in enumerate(scheduler.timesteps  ):
													__a			:    Union[str, Any]      							=	scheduler.scale_model_input(_UpperCAmelCase						,      _UpperCAmelCase  )

													__a			:    str      							=	model(_UpperCAmelCase						,      _UpperCAmelCase  )

													__a			:    Optional[int]      							=	scheduler.step(_UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  )
													__a			:    Tuple      							=	output.prev_sample

									__a			:    str      							=	torch.sum(torch.abs(_UpperCAmelCase  )  )
									__a			:    int      							=	torch.mean(torch.abs(_UpperCAmelCase  )  )

									if torch_device in ["cpu", "mps"]:
													assert abs(result_sum.item() - 2_0.4_1_2_5  ) < 1e-2
													assert abs(result_mean.item() - 0.0_2_6_6  ) < 1e-3
									else:
													# CUDA
													assert abs(result_sum.item() - 2_0.4_1_2_5  ) < 1e-2
													assert abs(result_mean.item() - 0.0_2_6_6  ) < 1e-3







					def 			_lowerCamelCase		(    self  ):
									if torch_device == "mps":
													return
									__a			:    Optional[int]      							=	self.scheduler_classes[0]
									__a			:    Optional[int]      							=	self.get_scheduler_config()
									__a			:    int      							=	scheduler_class(**_UpperCAmelCase  )

									scheduler.set_timesteps(self.num_inference_steps						,      device=_UpperCAmelCase  )

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

									for t in scheduler.timesteps:
													__a			:    Optional[int]      							=	scheduler.scale_model_input(_UpperCAmelCase						,      _UpperCAmelCase  )

													__a			:    int      							=	model(_UpperCAmelCase						,      _UpperCAmelCase  )

													__a			:    Optional[Any]      							=	scheduler.step(_UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  )
													__a			:    int      							=	output.prev_sample

									__a			:    Optional[int]      							=	torch.sum(torch.abs(_UpperCAmelCase  )  )
									__a			:    Union[str, Any]      							=	torch.mean(torch.abs(_UpperCAmelCase  )  )

									if str(_UpperCAmelCase  ).startswith('''cpu'''  ):
													# The following sum varies between 148 and 156 on mps. Why?
													assert abs(result_sum.item() - 2_0.4_1_2_5  ) < 1e-2
													assert abs(result_mean.item() - 0.0_2_6_6  ) < 1e-3
									else:
													# CUDA
													assert abs(result_sum.item() - 2_0.4_1_2_5  ) < 1e-2
													assert abs(result_mean.item() - 0.0_2_6_6  ) < 1e-3


"""simple docstring"""
import requests




def 					a__						(						lowerCAmelCase			,	lowerCAmelCase    )	-> None:
      UpperCAmelCase__			:							List[str]							=    {"""Content-Type""": """application/json"""}
      UpperCAmelCase__			:							List[str]							=    requests.post(lowerCAmelCase			,	json={"""text""": message_body}			,	headers=lowerCAmelCase    )
      if response.status_code != 2_00:
            UpperCAmelCase__			:							str							=    (
                """Request to slack returned an error """
                F"""{response.status_code}, the response is:\n{response.text}"""
            )
            raise ValueError(lowerCAmelCase    )


if __name__ == "__main__":
    # Set the slack url to the one provided by Slack when you create the webhook at
    # https://my.slack.com/services/new/incoming-webhook/
    send_slack_message("""<YOUR MESSAGE BODY>""", """<SLACK CHANNEL URL>""")

