




import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest


class    SCREAMING_SNAKE_CASE_				(       _UpperCamelCase   ):



	"""simple docstring"""


	__magic_name__   :       Dict      	=		(DPMSolverSinglestepScheduler,)
	__magic_name__   :       int      	=		(('num_inference_steps', 25),)






	def 					lowerCamelCase__     (					self		:						int					,     **lowerCAmelCase		:						List[str]	)	->       Tuple:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[int]             =      {
						    """num_train_timesteps""": 1000,
						    """beta_start""": 0.00_01,
						    """beta_end""": 0.02,
						    """beta_schedule""": """linear""",
						    """solver_order""": 2,
						    """prediction_type""": """epsilon""",
						    """thresholding""": False,
						    """sample_max_value""": 1.0,
						    """algorithm_type""": """dpmsolver++""",
						    """solver_type""": """midpoint""",
						    """lambda_min_clipped""": -float("""inf"""	),
						    """variance_type""": None,
						}

						config.update(**lowerCAmelCase	)
						return config






	def 					lowerCamelCase__     (					self		:						List[str]					,     lowerCAmelCase		:						Tuple=0					,     **lowerCAmelCase		:						Optional[Any]	)	->       Union[str, Any]:







						"""simple docstring"""







						__UpperCamelCase					:	Dict             =      dict(self.forward_default_kwargs	)
						__UpperCamelCase					:	int             =      kwargs.pop("""num_inference_steps"""					,     lowerCAmelCase	)
						__UpperCamelCase					:	str             =      self.dummy_sample
						__UpperCamelCase					:	Optional[Any]             =      0.1 * sample
						__UpperCamelCase					:	Optional[int]             =      [residual + 0.2, residual + 0.15, residual + 0.10]

						for scheduler_class in self.scheduler_classes:
											__UpperCamelCase					:	Tuple             =      self.get_scheduler_config(**lowerCAmelCase	)
											__UpperCamelCase					:	Union[str, Any]             =      scheduler_class(**lowerCAmelCase	)
											scheduler.set_timesteps(lowerCAmelCase	)
											# copy over dummy past residuals
											__UpperCamelCase					:	Any             =      dummy_past_residuals[: scheduler.config.solver_order]

											with tempfile.TemporaryDirectory() as tmpdirname:
																scheduler.save_config(lowerCAmelCase	)
																__UpperCamelCase					:	List[Any]             =      scheduler_class.from_pretrained(lowerCAmelCase	)
																new_scheduler.set_timesteps(lowerCAmelCase	)
																# copy over dummy past residuals
																__UpperCamelCase					:	Union[str, Any]             =      dummy_past_residuals[: new_scheduler.config.solver_order]

											__UpperCamelCase      ,			__UpperCamelCase					:	List[str]             =      sample, sample
											for t in range(lowerCAmelCase					,     time_step + scheduler.config.solver_order + 1	):
																__UpperCamelCase					:	Tuple             =      scheduler.step(lowerCAmelCase					,     lowerCAmelCase					,     lowerCAmelCase					,     **lowerCAmelCase	).prev_sample
																__UpperCamelCase					:	List[Any]             =      new_scheduler.step(lowerCAmelCase					,     lowerCAmelCase					,     lowerCAmelCase					,     **lowerCAmelCase	).prev_sample

																assert torch.sum(torch.abs(output - new_output	)	) < 1E-5, "Scheduler outputs are not identical"






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







						"""simple docstring"""







						pass






	def 					lowerCamelCase__     (					self		:						str					,     lowerCAmelCase		:						Optional[Any]=0					,     **lowerCAmelCase		:						Optional[Any]	)	->       Any:







						"""simple docstring"""







						__UpperCamelCase					:	Dict             =      dict(self.forward_default_kwargs	)
						__UpperCamelCase					:	Dict             =      kwargs.pop("""num_inference_steps"""					,     lowerCAmelCase	)
						__UpperCamelCase					:	int             =      self.dummy_sample
						__UpperCamelCase					:	Any             =      0.1 * sample
						__UpperCamelCase					:	Optional[int]             =      [residual + 0.2, residual + 0.15, residual + 0.10]

						for scheduler_class in self.scheduler_classes:
											__UpperCamelCase					:	str             =      self.get_scheduler_config()
											__UpperCamelCase					:	Dict             =      scheduler_class(**lowerCAmelCase	)
											scheduler.set_timesteps(lowerCAmelCase	)

											# copy over dummy past residuals (must be after setting timesteps)
											__UpperCamelCase					:	Optional[int]             =      dummy_past_residuals[: scheduler.config.solver_order]

											with tempfile.TemporaryDirectory() as tmpdirname:
																scheduler.save_config(lowerCAmelCase	)
																__UpperCamelCase					:	Tuple             =      scheduler_class.from_pretrained(lowerCAmelCase	)
																# copy over dummy past residuals
																new_scheduler.set_timesteps(lowerCAmelCase	)

																# copy over dummy past residual (must be after setting timesteps)
																__UpperCamelCase					:	List[Any]             =      dummy_past_residuals[: new_scheduler.config.solver_order]

											__UpperCamelCase					:	int             =      scheduler.step(lowerCAmelCase					,     lowerCAmelCase					,     lowerCAmelCase					,     **lowerCAmelCase	).prev_sample
											__UpperCamelCase					:	List[str]             =      new_scheduler.step(lowerCAmelCase					,     lowerCAmelCase					,     lowerCAmelCase					,     **lowerCAmelCase	).prev_sample

											assert torch.sum(torch.abs(output - new_output	)	) < 1E-5, "Scheduler outputs are not identical"






	def 					lowerCamelCase__     (					self		:						Optional[Any]					,     lowerCAmelCase		:						Tuple=None					,     **lowerCAmelCase		:						Tuple	)	->       List[str]:







						"""simple docstring"""







						if scheduler is None:
											__UpperCamelCase					:	List[Any]             =      self.scheduler_classes[0]
											__UpperCamelCase					:	Any             =      self.get_scheduler_config(**lowerCAmelCase	)
											__UpperCamelCase					:	str             =      scheduler_class(**lowerCAmelCase	)

						__UpperCamelCase					:	Optional[Any]             =      self.scheduler_classes[0]
						__UpperCamelCase					:	Optional[int]             =      self.get_scheduler_config(**lowerCAmelCase	)
						__UpperCamelCase					:	Union[str, Any]             =      scheduler_class(**lowerCAmelCase	)

						__UpperCamelCase					:	Dict             =      10
						__UpperCamelCase					:	List[Any]             =      self.dummy_model()
						__UpperCamelCase					:	List[Any]             =      self.dummy_sample_deter
						scheduler.set_timesteps(lowerCAmelCase	)

						for i, t in enumerate(scheduler.timesteps	):
											__UpperCamelCase					:	Optional[int]             =      model(lowerCAmelCase					,     lowerCAmelCase	)
											__UpperCamelCase					:	Optional[Any]             =      scheduler.step(lowerCAmelCase					,     lowerCAmelCase					,     lowerCAmelCase	).prev_sample

						return sample






	def 					lowerCamelCase__     (					self		:						Dict	)	->       Optional[Any]:







						"""simple docstring"""







						__UpperCamelCase					:	str             =      DPMSolverSinglestepScheduler(**self.get_scheduler_config()	)
						__UpperCamelCase					:	str             =      50
						__UpperCamelCase					:	int             =      self.dummy_model()
						__UpperCamelCase					:	Any             =      self.dummy_sample_deter
						scheduler.set_timesteps(lowerCAmelCase	)

						# make sure that the first t is uneven
						for i, t in enumerate(scheduler.timesteps[3:]	):
											__UpperCamelCase					:	List[Any]             =      model(lowerCAmelCase					,     lowerCAmelCase	)
											__UpperCamelCase					:	Any             =      scheduler.step(lowerCAmelCase					,     lowerCAmelCase					,     lowerCAmelCase	).prev_sample

						__UpperCamelCase					:	List[str]             =      torch.mean(torch.abs(lowerCAmelCase	)	)

						assert abs(result_mean.item() - 0.25_74	) < 1E-3






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







						"""simple docstring"""







						for timesteps in [25, 50, 100, 999, 1000]:
											self.check_over_configs(num_train_timesteps=lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						Union[str, Any]	)	->       Tuple:







						"""simple docstring"""







						__UpperCamelCase					:	int             =      DPMSolverSinglestepScheduler(**self.get_scheduler_config()	)
						__UpperCamelCase					:	List[str]             =      self.full_loop(scheduler=lowerCAmelCase	)
						__UpperCamelCase					:	int             =      torch.mean(torch.abs(lowerCAmelCase	)	)

						assert abs(result_mean.item() - 0.27_91	) < 1E-3

						__UpperCamelCase					:	List[Any]             =      DEISMultistepScheduler.from_config(scheduler.config	)
						__UpperCamelCase					:	List[Any]             =      DPMSolverMultistepScheduler.from_config(scheduler.config	)
						__UpperCamelCase					:	Union[str, Any]             =      UniPCMultistepScheduler.from_config(scheduler.config	)
						__UpperCamelCase					:	List[Any]             =      DPMSolverSinglestepScheduler.from_config(scheduler.config	)

						__UpperCamelCase					:	Tuple             =      self.full_loop(scheduler=lowerCAmelCase	)
						__UpperCamelCase					:	Any             =      torch.mean(torch.abs(lowerCAmelCase	)	)

						assert abs(result_mean.item() - 0.27_91	) < 1E-3






	def 					lowerCamelCase__     (					self		:						int	)	->       List[Any]:







						"""simple docstring"""







						self.check_over_configs(thresholding=lowerCAmelCase	)
						for order in [1, 2, 3]:
											for solver_type in ["midpoint", "heun"]:
																for threshold in [0.5, 1.0, 2.0]:
																					for prediction_type in ["epsilon", "sample"]:
																										self.check_over_configs(
																										    thresholding=lowerCAmelCase					,     prediction_type=lowerCAmelCase					,     sample_max_value=lowerCAmelCase					,     algorithm_type="""dpmsolver++"""					,     solver_order=lowerCAmelCase					,     solver_type=lowerCAmelCase					,     )






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







						"""simple docstring"""







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






	def 					lowerCamelCase__     (					self		:						int	)	->       int:







						"""simple docstring"""







						for algorithm_type in ["dpmsolver", "dpmsolver++"]:
											for solver_type in ["midpoint", "heun"]:
																for order in [1, 2, 3]:
																					for prediction_type in ["epsilon", "sample"]:
																										self.check_over_configs(
																										    solver_order=lowerCAmelCase					,     solver_type=lowerCAmelCase					,     prediction_type=lowerCAmelCase					,     algorithm_type=lowerCAmelCase					,     )
																										__UpperCamelCase					:	Dict             =      self.full_loop(
																										    solver_order=lowerCAmelCase					,     solver_type=lowerCAmelCase					,     prediction_type=lowerCAmelCase					,     algorithm_type=lowerCAmelCase					,     )
																										assert not torch.isnan(lowerCAmelCase	).any(), "Samples have nan numbers"






	def 					lowerCamelCase__     (					self		:						Tuple	)	->       Dict:







						"""simple docstring"""







						self.check_over_configs(lower_order_final=lowerCAmelCase	)
						self.check_over_configs(lower_order_final=lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						Tuple	)	->       Tuple:







						"""simple docstring"""







						self.check_over_configs(lambda_min_clipped=-float("""inf"""	)	)
						self.check_over_configs(lambda_min_clipped=-5.1	)






	def 					lowerCamelCase__     (					self		:						Tuple	)	->       List[Any]:







						"""simple docstring"""







						self.check_over_configs(variance_type=lowerCAmelCase	)
						self.check_over_configs(variance_type="""learned_range"""	)






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







						"""simple docstring"""







						for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1000]:
											self.check_over_forward(num_inference_steps=lowerCAmelCase					,     time_step=0	)






	def 					lowerCamelCase__     (					self		:						Tuple	)	->       Dict:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[Any]             =      self.full_loop()
						__UpperCamelCase					:	Optional[Any]             =      torch.mean(torch.abs(lowerCAmelCase	)	)

						assert abs(result_mean.item() - 0.27_91	) < 1E-3






	def 					lowerCamelCase__     (					self		:						Optional[Any]	)	->       List[str]:







						"""simple docstring"""







						__UpperCamelCase					:	Any             =      self.full_loop(use_karras_sigmas=lowerCAmelCase	)
						__UpperCamelCase					:	List[str]             =      torch.mean(torch.abs(lowerCAmelCase	)	)

						assert abs(result_mean.item() - 0.22_48	) < 1E-3






	def 					lowerCamelCase__     (					self		:						Any	)	->       List[str]:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[int]             =      self.full_loop(prediction_type="""v_prediction"""	)
						__UpperCamelCase					:	Tuple             =      torch.mean(torch.abs(lowerCAmelCase	)	)

						assert abs(result_mean.item() - 0.14_53	) < 1E-3






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







						"""simple docstring"""







						__UpperCamelCase					:	Dict             =      self.full_loop(prediction_type="""v_prediction"""					,     use_karras_sigmas=lowerCAmelCase	)
						__UpperCamelCase					:	Optional[Any]             =      torch.mean(torch.abs(lowerCAmelCase	)	)

						assert abs(result_mean.item() - 0.06_49	) < 1E-3






	def 					lowerCamelCase__     (					self		:						Dict	)	->       str:







						"""simple docstring"""







						__UpperCamelCase					:	Tuple             =      self.scheduler_classes[0]
						__UpperCamelCase					:	int             =      self.get_scheduler_config(thresholding=lowerCAmelCase					,     dynamic_thresholding_ratio=0	)
						__UpperCamelCase					:	Tuple             =      scheduler_class(**lowerCAmelCase	)

						__UpperCamelCase					:	Dict             =      10
						__UpperCamelCase					:	Optional[Any]             =      self.dummy_model()
						__UpperCamelCase					:	Optional[Any]             =      self.dummy_sample_deter.half()
						scheduler.set_timesteps(lowerCAmelCase	)

						for i, t in enumerate(scheduler.timesteps	):
											__UpperCamelCase					:	int             =      model(lowerCAmelCase					,     lowerCAmelCase	)
											__UpperCamelCase					:	List[str]             =      scheduler.step(lowerCAmelCase					,     lowerCAmelCase					,     lowerCAmelCase	).prev_sample

						assert sample.dtype == torch.floataa












from tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch


@require_torch
@require_tf
class    SCREAMING_SNAKE_CASE_				(       _UpperCamelCase   ):



	"""simple docstring"""


	def 					lowerCamelCase__     (					self		:						Dict	)	->       List[Any]:







						"""simple docstring"""







						__UpperCamelCase					:	Optional[Any]             =      SMALL_MODEL_IDENTIFIER
						__UpperCamelCase					:	List[str]             =      """pt"""
						__UpperCamelCase					:	List[Any]             =      """tf"""






	def 					lowerCamelCase__     (					self		:						Any					,     lowerCAmelCase		:						List[Any]	)	->       List[Any]:







						"""simple docstring"""







						__UpperCamelCase					:	str             =      AutoModel.from_pretrained(self.test_model	)
						model_pt.save_pretrained(lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						List[Any]					,     lowerCAmelCase		:						Any	)	->       Optional[Any]:







						"""simple docstring"""







						__UpperCamelCase					:	List[str]             =      TFAutoModel.from_pretrained(self.test_model					,     from_pt=lowerCAmelCase	)
						model_tf.save_pretrained(lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						str	)	->       Tuple:







						"""simple docstring"""







						__UpperCamelCase					:	Any             =      """mock_framework"""

						# Framework provided - return whatever the user provides
						__UpperCamelCase					:	List[Any]             =      FeaturesManager.determine_framework(self.test_model					,     lowerCAmelCase	)
						self.assertEqual(lowerCAmelCase					,     lowerCAmelCase	)

						# Local checkpoint and framework provided - return provided framework
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
											self._setup_pt_ckpt(lowerCAmelCase	)
											__UpperCamelCase					:	Union[str, Any]             =      FeaturesManager.determine_framework(lowerCAmelCase					,     lowerCAmelCase	)
											self.assertEqual(lowerCAmelCase					,     lowerCAmelCase	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
											self._setup_tf_ckpt(lowerCAmelCase	)
											__UpperCamelCase					:	Any             =      FeaturesManager.determine_framework(lowerCAmelCase					,     lowerCAmelCase	)
											self.assertEqual(lowerCAmelCase					,     lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						Optional[Any]	)	->       str:







						"""simple docstring"""







						with TemporaryDirectory() as local_pt_ckpt:
											self._setup_pt_ckpt(lowerCAmelCase	)
											__UpperCamelCase					:	Optional[Any]             =      FeaturesManager.determine_framework(lowerCAmelCase	)
											self.assertEqual(lowerCAmelCase					,     self.framework_pt	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
											self._setup_tf_ckpt(lowerCAmelCase	)
											__UpperCamelCase					:	Union[str, Any]             =      FeaturesManager.determine_framework(lowerCAmelCase	)
											self.assertEqual(lowerCAmelCase					,     self.framework_tf	)

						# Invalid local checkpoint
						with TemporaryDirectory() as local_invalid_ckpt:
											with self.assertRaises(lowerCAmelCase	):
																__UpperCamelCase					:	int             =      FeaturesManager.determine_framework(lowerCAmelCase	)






	def 					lowerCamelCase__     (					self		:						Dict	)	->       Tuple:







						"""simple docstring"""







						__UpperCamelCase					:	str             =      MagicMock(return_value=lowerCAmelCase	)
						with patch("""transformers.onnx.features.is_tf_available"""					,     lowerCAmelCase	):
											__UpperCamelCase					:	Any             =      FeaturesManager.determine_framework(self.test_model	)
											self.assertEqual(lowerCAmelCase					,     self.framework_pt	)

						# PyTorch not in environment -> use TensorFlow
						__UpperCamelCase					:	Optional[int]             =      MagicMock(return_value=lowerCAmelCase	)
						with patch("""transformers.onnx.features.is_torch_available"""					,     lowerCAmelCase	):
											__UpperCamelCase					:	Any             =      FeaturesManager.determine_framework(self.test_model	)
											self.assertEqual(lowerCAmelCase					,     self.framework_tf	)

						# Both in environment -> use PyTorch
						__UpperCamelCase					:	Dict             =      MagicMock(return_value=lowerCAmelCase	)
						__UpperCamelCase					:	Any             =      MagicMock(return_value=lowerCAmelCase	)
						with patch("""transformers.onnx.features.is_tf_available"""					,     lowerCAmelCase	), patch(
						    """transformers.onnx.features.is_torch_available"""					,     lowerCAmelCase	):
											__UpperCamelCase					:	int             =      FeaturesManager.determine_framework(self.test_model	)
											self.assertEqual(lowerCAmelCase					,     self.framework_pt	)

						# Both not in environment -> raise error
						__UpperCamelCase					:	Union[str, Any]             =      MagicMock(return_value=lowerCAmelCase	)
						__UpperCamelCase					:	Tuple             =      MagicMock(return_value=lowerCAmelCase	)
						with patch("""transformers.onnx.features.is_tf_available"""					,     lowerCAmelCase	), patch(
						    """transformers.onnx.features.is_torch_available"""					,     lowerCAmelCase	):
											with self.assertRaises(lowerCAmelCase	):
																__UpperCamelCase					:	str             =      FeaturesManager.determine_framework(self.test_model	)






