




'''simple docstring'''


import tempfile

import torch

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

from .test_schedulers import SchedulerCommonTest







class     SCREAMING_SNAKE_CASE						(	snake_case_							):
      __magic_name__     :					List[Any]             =				(DPMSolverMultistepScheduler,)
      __magic_name__     :					Union[str, Any]             =				(('''num_inference_steps''', 25),)






      def        lowercase_	(   self      :     Tuple	,	**lowercase__      :     Tuple					):


           '''simple docstring'''

           a_    :       Optional[int]    	=     {
               """num_train_timesteps""": 1000,
               """beta_start""": 0.0001,
               """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""",
               """lower_order_final""": False,
               """lambda_min_clipped""": -float("""inf"""					),
               """variance_type""": None,
           }

           config.update(**lowercase__					)
           return config






      def        lowercase_	(   self      :     Optional[Any]	,	lowercase__      :     Optional[Any]=0	,	**lowercase__      :     int					):


           '''simple docstring'''

           a_    :       Union[str, Any]    	=     dict(self.forward_default_kwargs					)
           a_    :       Any    	=     kwargs.pop("""num_inference_steps"""	,	lowercase__					)
           a_    :       Optional[Any]    	=     self.dummy_sample
           a_    :       Any    	=     0.1 * sample
           a_    :       Tuple    	=     [residual + 0.2, residual + 0.15, residual + 0.10]

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

                with tempfile.TemporaryDirectory() as tmpdirname:
                     scheduler.save_config(lowercase__					)
                     a_    :       Dict    	=     scheduler_class.from_pretrained(lowercase__					)
                     new_scheduler.set_timesteps(lowercase__					)
                     # copy over dummy past residuals
                     a_    :       List[Any]    	=     dummy_past_residuals[: new_scheduler.config.solver_order]

                a_	,    a_    :       int    	=     sample, sample
                for t in range(lowercase__	,	time_step + scheduler.config.solver_order + 1					):
                     a_    :       Optional[Any]    	=     scheduler.step(lowercase__	,	lowercase__	,	lowercase__	,	**lowercase__					).prev_sample
                     a_    :       str    	=     new_scheduler.step(lowercase__	,	lowercase__	,	lowercase__	,	**lowercase__					).prev_sample

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






      def        lowercase_	(   self      :     Union[str, Any]					):


           '''simple docstring'''

           pass






      def        lowercase_	(   self      :     Dict	,	lowercase__      :     str=0	,	**lowercase__      :     Optional[int]					):


           '''simple docstring'''

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

           for scheduler_class in self.scheduler_classes:
                a_    :       str    	=     self.get_scheduler_config()
                a_    :       str    	=     scheduler_class(**lowercase__					)
                scheduler.set_timesteps(lowercase__					)

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

                with tempfile.TemporaryDirectory() as tmpdirname:
                     scheduler.save_config(lowercase__					)
                     a_    :       Optional[int]    	=     scheduler_class.from_pretrained(lowercase__					)
                     # copy over dummy past residuals
                     new_scheduler.set_timesteps(lowercase__					)

                     # copy over dummy past residual (must be after setting timesteps)
                     a_    :       Optional[int]    	=     dummy_past_residuals[: new_scheduler.config.solver_order]

                a_    :       List[str]    	=     scheduler.step(lowercase__	,	lowercase__	,	lowercase__	,	**lowercase__					).prev_sample
                a_    :       Optional[int]    	=     new_scheduler.step(lowercase__	,	lowercase__	,	lowercase__	,	**lowercase__					).prev_sample

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






      def        lowercase_	(   self      :     Optional[Any]	,	lowercase__      :     Optional[Any]=None	,	**lowercase__      :     Optional[Any]					):


           '''simple docstring'''

           if scheduler is None:
                a_    :       int    	=     self.scheduler_classes[0]
                a_    :       List[Any]    	=     self.get_scheduler_config(**lowercase__					)
                a_    :       Union[str, Any]    	=     scheduler_class(**lowercase__					)

           a_    :       Tuple    	=     10
           a_    :       str    	=     self.dummy_model()
           a_    :       List[str]    	=     self.dummy_sample_deter
           scheduler.set_timesteps(lowercase__					)

           for i, t in enumerate(scheduler.timesteps					):
                a_    :       Dict    	=     model(lowercase__	,	lowercase__					)
                a_    :       str    	=     scheduler.step(lowercase__	,	lowercase__	,	lowercase__					).prev_sample

           return sample






      def        lowercase_	(   self      :     List[str]					):


           '''simple docstring'''

           a_    :       Optional[int]    	=     dict(self.forward_default_kwargs					)

           a_    :       List[Any]    	=     kwargs.pop("""num_inference_steps"""	,	lowercase__					)

           for scheduler_class in self.scheduler_classes:
                a_    :       Optional[Any]    	=     self.get_scheduler_config()
                a_    :       Optional[Any]    	=     scheduler_class(**lowercase__					)

                a_    :       List[str]    	=     self.dummy_sample
                a_    :       Tuple    	=     0.1 * sample

                if num_inference_steps is not None and hasattr(lowercase__	,	"""set_timesteps"""					):
                     scheduler.set_timesteps(lowercase__					)
                elif num_inference_steps is not None and not hasattr(lowercase__	,	"""set_timesteps"""					):
                     a_    :       int    	=     num_inference_steps

                # copy over dummy past residuals (must be done after set_timesteps)
                a_    :       List[Any]    	=     [residual + 0.2, residual + 0.15, residual + 0.10]
                a_    :       int    	=     dummy_past_residuals[: scheduler.config.solver_order]

                a_    :       List[str]    	=     scheduler.timesteps[5]
                a_    :       str    	=     scheduler.timesteps[6]

                a_    :       Union[str, Any]    	=     scheduler.step(lowercase__	,	lowercase__	,	lowercase__	,	**lowercase__					).prev_sample
                a_    :       str    	=     scheduler.step(lowercase__	,	lowercase__	,	lowercase__	,	**lowercase__					).prev_sample

                self.assertEqual(output_a.shape	,	sample.shape					)
                self.assertEqual(output_a.shape	,	output_a.shape					)






      def        lowercase_	(   self      :     int					):


           '''simple docstring'''

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






      def        lowercase_	(   self      :     str					):


           '''simple docstring'''

           self.check_over_configs(thresholding=lowercase__					)
           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=lowercase__	,	prediction_type=lowercase__	,	sample_max_value=lowercase__	,	algorithm_type="""dpmsolver++"""	,	solver_order=lowercase__	,	solver_type=lowercase__	,	)






      def        lowercase_	(   self      :     Any					):


           '''simple docstring'''

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






      def        lowercase_	(   self      :     Optional[int]					):


           '''simple docstring'''

           for algorithm_type in ["dpmsolver", "dpmsolver++", "sde-dpmsolver", "sde-dpmsolver++"]:
                for solver_type in ["midpoint", "heun"]:
                     for order in [1, 2, 3]:
                          for prediction_type in ["epsilon", "sample"]:
                               if algorithm_type in ["sde-dpmsolver", "sde-dpmsolver++"]:
                                    if order == 3:
                                         continue
                               else:
                                    self.check_over_configs(
                                        solver_order=lowercase__	,	solver_type=lowercase__	,	prediction_type=lowercase__	,	algorithm_type=lowercase__	,	)
                               a_    :       List[str]    	=     self.full_loop(
                                   solver_order=lowercase__	,	solver_type=lowercase__	,	prediction_type=lowercase__	,	algorithm_type=lowercase__	,	)
                               assert not torch.isnan(lowercase__					).any(), "Samples have nan numbers"






      def        lowercase_	(   self      :     Optional[int]					):


           '''simple docstring'''

           self.check_over_configs(lower_order_final=lowercase__					)
           self.check_over_configs(lower_order_final=lowercase__					)






      def        lowercase_	(   self      :     Any					):


           '''simple docstring'''

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






      def        lowercase_	(   self      :     Optional[Any]					):


           '''simple docstring'''

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






      def        lowercase_	(   self      :     Tuple					):


           '''simple docstring'''

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






      def        lowercase_	(   self      :     Any					):


           '''simple docstring'''

           a_    :       str    	=     self.full_loop()
           a_    :       str    	=     torch.mean(torch.abs(lowercase__					)					)

           assert abs(result_mean.item() - 0.3301					) < 1e-3






      def        lowercase_	(   self      :     Optional[int]					):


           '''simple docstring'''

           a_    :       Optional[Any]    	=     self.full_loop(thresholding=lowercase__	,	dynamic_thresholding_ratio=0.87	,	sample_max_value=0.5					)
           a_    :       List[Any]    	=     torch.mean(torch.abs(lowercase__					)					)

           assert abs(result_mean.item() - 1.1364					) < 1e-3






      def        lowercase_	(   self      :     Optional[int]					):


           '''simple docstring'''

           a_    :       Tuple    	=     self.full_loop(prediction_type="""v_prediction"""					)
           a_    :       Any    	=     torch.mean(torch.abs(lowercase__					)					)

           assert abs(result_mean.item() - 0.2251					) < 1e-3






      def        lowercase_	(   self      :     Optional[Any]					):


           '''simple docstring'''

           a_    :       Union[str, Any]    	=     self.full_loop(prediction_type="""v_prediction"""	,	use_karras_sigmas=lowercase__					)
           a_    :       Any    	=     torch.mean(torch.abs(lowercase__					)					)

           assert abs(result_mean.item() - 0.2096					) < 1e-3






      def        lowercase_	(   self      :     Optional[int]					):


           '''simple docstring'''

           a_    :       Dict    	=     DPMSolverMultistepScheduler(**self.get_scheduler_config()					)
           a_    :       Tuple    	=     self.full_loop(scheduler=lowercase__					)
           a_    :       List[Any]    	=     torch.mean(torch.abs(lowercase__					)					)

           assert abs(result_mean.item() - 0.3301					) < 1e-3

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

           a_    :       List[str]    	=     self.full_loop(scheduler=lowercase__					)
           a_    :       Optional[int]    	=     torch.mean(torch.abs(lowercase__					)					)

           assert abs(result_mean.item() - 0.3301					) < 1e-3






      def        lowercase_	(   self      :     Any					):


           '''simple docstring'''

           a_    :       Optional[int]    	=     self.scheduler_classes[0]
           a_    :       Dict    	=     self.get_scheduler_config(thresholding=lowercase__	,	dynamic_thresholding_ratio=0					)
           a_    :       List[Any]    	=     scheduler_class(**lowercase__					)

           a_    :       List[str]    	=     10
           a_    :       Dict    	=     self.dummy_model()
           a_    :       Optional[int]    	=     self.dummy_sample_deter.half()
           scheduler.set_timesteps(lowercase__					)

           for i, t in enumerate(scheduler.timesteps					):
                a_    :       Dict    	=     model(lowercase__	,	lowercase__					)
                a_    :       Any    	=     scheduler.step(lowercase__	,	lowercase__	,	lowercase__					).prev_sample

           assert sample.dtype == torch.floataa






      def        lowercase_	(   self      :     Optional[Any]	,	**lowercase__      :     str					):


           '''simple docstring'''

           for scheduler_class in self.scheduler_classes:
                a_    :       Union[str, Any]    	=     self.get_scheduler_config(**lowercase__					)
                a_    :       List[str]    	=     scheduler_class(**lowercase__					)

                scheduler.set_timesteps(scheduler.config.num_train_timesteps					)
                assert len(scheduler.timesteps.unique()					) == scheduler.num_inference_steps






'''simple docstring'''


from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCAmelCase_			:     int							    =     {
    'configuration_dpr': ['DPR_PRETRAINED_CONFIG_ARCHIVE_MAP', 'DPRConfig'],
    'tokenization_dpr': [
        'DPRContextEncoderTokenizer',
        'DPRQuestionEncoderTokenizer',
        'DPRReaderOutput',
        'DPRReaderTokenizer',
    ],
}


try:
 if not is_tokenizers_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase_			:     Union[str, Any]							    =     [
     'DPRContextEncoderTokenizerFast',
     'DPRQuestionEncoderTokenizerFast',
     'DPRReaderTokenizerFast',
 ]

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase_			:     List[str]							    =     [
     'DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'DPRContextEncoder',
     'DPRPretrainedContextEncoder',
     'DPRPreTrainedModel',
     'DPRPretrainedQuestionEncoder',
     'DPRPretrainedReader',
     'DPRQuestionEncoder',
     'DPRReader',
 ]

try:
 if not is_tf_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase_			:     Dict							    =     [
     'TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST',
     'TFDPRContextEncoder',
     'TFDPRPretrainedContextEncoder',
     'TFDPRPretrainedQuestionEncoder',
     'TFDPRPretrainedReader',
     'TFDPRQuestionEncoder',
     'TFDPRReader',
 ]


if TYPE_CHECKING:
 from .configuration_dpr import DPR_PRETRAINED_CONFIG_ARCHIVE_MAP, DPRConfig
 from .tokenization_dpr import (
     DPRContextEncoderTokenizer,
     DPRQuestionEncoderTokenizer,
     DPRReaderOutput,
     DPRReaderTokenizer,
 )

 try:
  if not is_tokenizers_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .tokenization_dpr_fast import (
      DPRContextEncoderTokenizerFast,
      DPRQuestionEncoderTokenizerFast,
      DPRReaderTokenizerFast,
  )

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_dpr import (
      DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
      DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
      DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
      DPRContextEncoder,
      DPRPretrainedContextEncoder,
      DPRPreTrainedModel,
      DPRPretrainedQuestionEncoder,
      DPRPretrainedReader,
      DPRQuestionEncoder,
      DPRReader,
  )

 try:
  if not is_tf_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_tf_dpr import (
      TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
      TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
      TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
      TFDPRContextEncoder,
      TFDPRPretrainedContextEncoder,
      TFDPRPretrainedQuestionEncoder,
      TFDPRPretrainedReader,
      TFDPRQuestionEncoder,
      TFDPRReader,
  )

else:
 import sys

 lowerCAmelCase_			:     int							    =     _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
