code: import torch
  
  from diffusers import HeunDiscreteScheduler
  from diffusers.utils import torch_device
  
  from .test_schedulers import SchedulerCommonTest
  
  
  
  
  
  
  
  class 					UpperCAmelCase_				(							_a    ):
      UpperCAmelCase_			:       Any						 =							(HeunDiscreteScheduler,)
      UpperCAmelCase_			:       Optional[int]						 =							10
  
  
  
  
      def 						__lowerCamelCase (					self  :   Optional[Any]      ,      **lowerCAmelCase  :   Tuple   )	-> List[str]:
        a__     		=     {
            "num_train_timesteps": 1100,
            "beta_start": 0.0_001,
            "beta_end": 0.02,
            "beta_schedule": "linear",
        }
  
        config.update(**lowerCAmelCase   )
        return config
  
  
  
  
      def 						__lowerCamelCase (					self  :   int   )	-> str:
        for timesteps in [10, 50, 100, 1000]:
          self.check_over_configs(num_train_timesteps=lowerCAmelCase   )
  
  
  
  
      def 						__lowerCamelCase (					self  :   str   )	-> List[Any]:
        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=lowerCAmelCase      ,      beta_end=lowerCAmelCase   )
  
  
  
  
      def 						__lowerCamelCase (					self  :   Any   )	-> Union[str, Any]:
        for schedule in ["linear", "scaled_linear", "exp"]:
          self.check_over_configs(beta_schedule=lowerCAmelCase   )
  
  
  
  
      def 						__lowerCamelCase (					self  :   Optional[int]   )	-> Any:
        for clip_sample_range in [1.0, 2.0, 3.0]:
          self.check_over_configs(clip_sample_range=lowerCAmelCase      ,      clip_sample=lowerCAmelCase   )
  
  
  
  
      def 						__lowerCamelCase (					self  :   List[str]   )	-> str:
        for prediction_type in ["epsilon", "v_prediction", "sample"]:
          self.check_over_configs(prediction_type=lowerCAmelCase   )
  
  
  
  
      def 						__lowerCamelCase (					self  :   Union[str, Any]   )	-> Union[str, Any]:
        a__     		=     self.scheduler_classes[0]
        a__     		=     self.get_scheduler_config()
        a__     		=     scheduler_class(**lowerCAmelCase   )
  
        scheduler.set_timesteps(self.num_inference_steps   )
  
        a__     		=     self.dummy_model()
        a__     		=     self.dummy_sample_deter * scheduler.init_noise_sigma
        a__     		=     sample.to(lowerCAmelCase   )
  
        for i, t in enumerate(scheduler.timesteps   ):
          a__     		=     scheduler.scale_model_input(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     model(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     scheduler.step(lowerCAmelCase      ,      lowerCAmelCase      ,      lowerCAmelCase   )
          a__     		=     output.prev_sample
  
        a__     		=     torch.sum(torch.abs(lowerCAmelCase   )   )
        a__     		=     torch.mean(torch.abs(lowerCAmelCase   )   )
  
        if torch_device in ["cpu", "mps"]:
          assert abs(result_sum.item() - 0.1_233   ) < 1e-2
          assert abs(result_mean.item() - 0.0_002   ) < 1e-3
        else:
          # CUDA
          assert abs(result_sum.item() - 0.1_233   ) < 1e-2
          assert abs(result_mean.item() - 0.0_002   ) < 1e-3
  
  
  
  
      def 						__lowerCamelCase (					self  :   List[str]   )	-> Optional[Any]:
        a__     		=     self.scheduler_classes[0]
        a__     		=     self.get_scheduler_config(prediction_type="v_prediction"   )
        a__     		=     scheduler_class(**lowerCAmelCase   )
  
        scheduler.set_timesteps(self.num_inference_steps   )
  
        a__     		=     self.dummy_model()
        a__     		=     self.dummy_sample_deter * scheduler.init_noise_sigma
        a__     		=     sample.to(lowerCAmelCase   )
  
        for i, t in enumerate(scheduler.timesteps   ):
          a__     		=     scheduler.scale_model_input(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     model(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     scheduler.step(lowerCAmelCase      ,      lowerCAmelCase      ,      lowerCAmelCase   )
          a__     		=     output.prev_sample
  
        a__     		=     torch.sum(torch.abs(lowerCAmelCase   )   )
        a__     		=     torch.mean(torch.abs(lowerCAmelCase   )   )
  
        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_002   ) < 1e-3
  
  
  
  
      def 						__lowerCamelCase (					self  :   str   )	-> Union[str, Any]:
        a__     		=     self.scheduler_classes[0]
        a__     		=     self.get_scheduler_config()
        a__     		=     scheduler_class(**lowerCAmelCase   )
  
        scheduler.set_timesteps(self.num_inference_steps      ,      device=lowerCAmelCase   )
  
        a__     		=     self.dummy_model()
        a__     		=     self.dummy_sample_deter.to(lowerCAmelCase   ) * scheduler.init_noise_sigma
  
        for t in scheduler.timesteps:
          a__     		=     scheduler.scale_model_input(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     model(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     scheduler.step(lowerCAmelCase      ,      lowerCAmelCase      ,      lowerCAmelCase   )
          a__     		=     output.prev_sample
  
        a__     		=     torch.sum(torch.abs(lowerCAmelCase   )   )
        a__     		=     torch.mean(torch.abs(lowerCAmelCase   )   )
  
        if str(lowerCAmelCase   ).startswith("cpu"   ):
          # The following sum varies between 148 and 156 on mps. Why?
          assert abs(result_sum.item() - 0.1_233   ) < 1e-2
          assert abs(result_mean.item() - 0.0_002   ) < 1e-3
        elif str(lowerCAmelCase   ).startswith("mps"   ):
          # Larger tolerance on mps
          assert abs(result_mean.item() - 0.0_002   ) < 1e-2
        else:
          # CUDA
          assert abs(result_sum.item() - 0.1_233   ) < 1e-2
          assert abs(result_mean.item() - 0.0_002   ) < 1e-3
  
  
  
  
      def 						__lowerCamelCase (					self  :   Dict   )	-> int:
        a__     		=     self.scheduler_classes[0]
        a__     		=     self.get_scheduler_config()
        a__     		=     scheduler_class(**lowerCAmelCase      ,      use_karras_sigmas=lowerCAmelCase   )
  
        scheduler.set_timesteps(self.num_inference_steps      ,      device=lowerCAmelCase   )
  
        a__     		=     self.dummy_model()
        a__     		=     self.dummy_sample_deter.to(lowerCAmelCase   ) * scheduler.init_noise_sigma
        a__     		=     sample.to(lowerCAmelCase   )
  
        for t in scheduler.timesteps:
          a__     		=     scheduler.scale_model_input(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     model(lowerCAmelCase      ,      lowerCAmelCase   )
  
          a__     		=     scheduler.step(lowerCAmelCase      ,      lowerCAmelCase      ,      lowerCAmelCase   )
          a__     		=     output.prev_sample
  
        a__     		=     torch.sum(torch.abs(lowerCAmelCase   )   )
        a__     		=     torch.mean(torch.abs(lowerCAmelCase   )   )
  
        assert abs(result_sum.item() - 0.00_015   ) < 1e-2
        assert abs(result_mean.item() - 1.9_8_6_9_5_5_4_5_3_5_0_3_4_6_9_5e-0_7   ) < 1e-2
  
  
style_context: from unittest.mock import patch
  
  import datasets
  from datasets import Dataset
  def 				__lowercase	(						)				->Union[str, Any]:
  
    '''simple docstring'''
  
  
  
  
    a__     		=     Dataset.from_dict({"col_1": [3, 2, 0, 1]}					)
  
    with patch("tqdm.auto.tqdm"					) as mock_tqdm:
      datasets.disable_progress_bar()
      dset.map(lambda lowerCamelCase__					: {"col_2": x["col_1"] + 1}					)
      mock_tqdm.assert_not_called()
  
      mock_tqdm.reset_mock()
  
      datasets.enable_progress_bar()
      dset.map(lambda lowerCamelCase__					: {"col_2": x["col_1"] + 1}					)
      mock_tqdm.assert_called()
  
  