
'''simple docstring'''

from typing import SupportsIndex

import numpy as np
from scipy.ndimage import convolve







def __A  (    lowerCAmelCase_ ,       lowerCAmelCase_ ,       lowerCAmelCase_       ):
     _UpperCAmelCase					:  Dict         =    np.stack((horizontal_flow, vertical_flow) ,       2       )

     # Create a grid of all pixel coordinates and subtract the flow to get the
     # target pixels coordinates
     _UpperCAmelCase					:  int         =    np.stack(
         np.meshgrid(np.arange(0 ,       image.shape[1]       ) ,       np.arange(0 ,       image.shape[0]       )       ) ,       2       )
     _UpperCAmelCase					:  List[Any]         =    np.round(grid - flow       ).astype(np.intaa       )

     # Find the locations outside of the original image
     _UpperCAmelCase					:  List[str]         =    (grid < 0) | (grid >= np.array([image.shape[1], image.shape[0]]       ))
     _UpperCAmelCase					:  Optional[Any]         =    0

     _UpperCAmelCase					:  Optional[Any]         =    image[grid[:, :, 1], grid[:, :, 0]]

     # Set pixels at invalid locations to 0
     _UpperCAmelCase					:  Dict         =    0

     return warped







def __A  (    lowerCAmelCase_ ,       lowerCAmelCase_ ,       lowerCAmelCase_ ,       lowerCAmelCase_ = None ,       ):
     if alpha is None:
          _UpperCAmelCase					:  Any         =    0.1

     # Initialize flow
     _UpperCAmelCase					:  Optional[Any]         =    np.zeros_like(lowerCAmelCase_       )
     _UpperCAmelCase					:  str         =    np.zeros_like(lowerCAmelCase_       )

     # Prepare kernels for the calculation of the derivatives and the average velocity
     _UpperCAmelCase					:  str         =    np.array([[-1, 1], [-1, 1]]       ) * 0.25
     _UpperCAmelCase					:  Tuple         =    np.array([[-1, -1], [1, 1]]       ) * 0.25
     _UpperCAmelCase					:  Optional[int]         =    np.array([[1, 1], [1, 1]]       ) * 0.25
     _UpperCAmelCase					:  Union[str, Any]         =    np.array(
         [[1 / 12, 1 / 6, 1 / 12], [1 / 6, 0, 1 / 6], [1 / 12, 1 / 6, 1 / 12]]       )

     # Iteratively refine the flow
     for _ in range(lowerCAmelCase_       ):
          _UpperCAmelCase					:  List[Any]         =    warp(lowerCAmelCase_ ,       lowerCAmelCase_ ,       lowerCAmelCase_       )
          _UpperCAmelCase					:  List[Any]         =    convolve(lowerCAmelCase_ ,       lowerCAmelCase_       ) + convolve(lowerCAmelCase_ ,       lowerCAmelCase_       )
          _UpperCAmelCase					:  Any         =    convolve(lowerCAmelCase_ ,       lowerCAmelCase_       ) + convolve(lowerCAmelCase_ ,       lowerCAmelCase_       )
          _UpperCAmelCase					:  Optional[Any]         =    convolve(lowerCAmelCase_ ,       lowerCAmelCase_       ) + convolve(lowerCAmelCase_ ,       -kernel_t       )

          _UpperCAmelCase					:  Optional[int]         =    convolve(lowerCAmelCase_ ,       lowerCAmelCase_       )
          _UpperCAmelCase					:  Tuple         =    convolve(lowerCAmelCase_ ,       lowerCAmelCase_       )

          # This updates the flow as proposed in the paper (Step 12)
          _UpperCAmelCase					:  int         =    (
              derivative_x * avg_horizontal_velocity
              + derivative_y * avg_vertical_velocity
              + derivative_t
          )
          _UpperCAmelCase					:  Any         =    update / (alpha**2 + derivative_x**2 + derivative_y**2)

          _UpperCAmelCase					:  str         =    avg_horizontal_velocity - derivative_x * update
          _UpperCAmelCase					:  Dict         =    avg_vertical_velocity - derivative_y * update

     return horizontal_flow, vertical_flow


if __name__ == "__main__":
   import doctest

   doctest.testmod()




'''simple docstring'''

import torch

from diffusers import HeunDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest







class 	__lowerCAmelCase    (  __a				):
     snake_case   :    List[str]							      =      (HeunDiscreteScheduler,)
     snake_case   :    Optional[Any]							      =      1_0



     def     snake_case_ (self						,	**lowerCAmelCase__	):
          _UpperCAmelCase					:  Union[str, Any]         =    {
              """num_train_timesteps""": 1_1_0_0,
              """beta_start""": 0.0_0_0_1,
              """beta_end""": 0.0_2,
              """beta_schedule""": """linear""",
          }

          config.update(**lowerCAmelCase__	)
          return config



     def     snake_case_ (self	):
          for timesteps in [1_0, 5_0, 1_0_0, 1_0_0_0]:
               self.check_over_configs(num_train_timesteps=lowerCAmelCase__	)



     def     snake_case_ (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=lowerCAmelCase__						,	beta_end=lowerCAmelCase__	)



     def     snake_case_ (self	):
          for schedule in ["linear", "scaled_linear", "exp"]:
               self.check_over_configs(beta_schedule=lowerCAmelCase__	)



     def     snake_case_ (self	):
          for clip_sample_range in [1.0, 2.0, 3.0]:
               self.check_over_configs(clip_sample_range=lowerCAmelCase__						,	clip_sample=lowerCAmelCase__	)



     def     snake_case_ (self	):
          for prediction_type in ["epsilon", "v_prediction", "sample"]:
               self.check_over_configs(prediction_type=lowerCAmelCase__	)



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Tuple         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  Optional[int]         =    self.get_scheduler_config()
          _UpperCAmelCase					:  List[str]         =    scheduler_class(**lowerCAmelCase__	)

          scheduler.set_timesteps(self.num_inference_steps	)

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

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

               _UpperCAmelCase					:  Any         =    model(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Tuple         =    scheduler.step(lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  Optional[Any]         =    output.prev_sample

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

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



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Optional[int]         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  Any         =    self.get_scheduler_config(prediction_type="""v_prediction"""	)
          _UpperCAmelCase					:  Any         =    scheduler_class(**lowerCAmelCase__	)

          scheduler.set_timesteps(self.num_inference_steps	)

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

          for i, t in enumerate(scheduler.timesteps	):
               _UpperCAmelCase					:  List[str]         =    scheduler.scale_model_input(lowerCAmelCase__						,	lowerCAmelCase__	)

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

               _UpperCAmelCase					:  Optional[Any]         =    scheduler.step(lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  Union[str, Any]         =    output.prev_sample

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

          if torch_device in ["cpu", "mps"]:
               assert abs(result_sum.item() - 4.6_934e-07	) < 1e-2
               assert abs(result_mean.item() - 6.1_112e-10	) < 1e-3
          else:
               # CUDA
               assert abs(result_sum.item() - 4.693_428_650_170_972e-07	) < 1e-2
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-3



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Tuple         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  Any         =    self.get_scheduler_config()
          _UpperCAmelCase					:  Optional[int]         =    scheduler_class(**lowerCAmelCase__	)

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

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

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

               _UpperCAmelCase					:  Optional[Any]         =    model(lowerCAmelCase__						,	lowerCAmelCase__	)

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

          _UpperCAmelCase					:  Any         =    torch.sum(torch.abs(lowerCAmelCase__	)	)
          _UpperCAmelCase					:  Optional[Any]         =    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_2_3_3	) < 1e-2
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-3
          elif str(lowerCAmelCase__	).startswith("""mps"""	):
               # Larger tolerance on mps
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-2
          else:
               # CUDA
               assert abs(result_sum.item() - 0.1_2_3_3	) < 1e-2
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-3



     def     snake_case_ (self	):
          _UpperCAmelCase					:  int         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  str         =    self.get_scheduler_config()
          _UpperCAmelCase					:  Union[str, Any]         =    scheduler_class(**lowerCAmelCase__						,	use_karras_sigmas=lowerCAmelCase__	)

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

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

          for t in scheduler.timesteps:
               _UpperCAmelCase					:  Dict         =    scheduler.scale_model_input(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Optional[Any]         =    model(lowerCAmelCase__						,	lowerCAmelCase__	)

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

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

          assert abs(result_sum.item() - 0.0_0_0_1_5	) < 1e-2
          assert abs(result_mean.item() - 1.9_869_554_535_034_695e-07	) < 1e-2


