





import torch

from diffusers import HeunDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest

class      A_	(a_      ):
	UpperCAmelCase__ =      (HeunDiscreteScheduler,)
	UpperCAmelCase__ =      1_0
	def 					_lowercase    (   self							, **_A		):






				'''simple docstring'''







				UpperCAmelCase       		=    {
				    '''num_train_timesteps''': 1_1_0_0,
				    '''beta_start''': 0.00_01,
				    '''beta_end''': 0.02,
				    '''beta_schedule''': '''linear''',
				}

				config.update(**_A		)
				return config
	def 					_lowercase    (   self		):






				'''simple docstring'''







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






				'''simple docstring'''







				for beta_start, beta_end in zip([0.0_00_01, 0.00_01, 0.0_01]							, [0.00_02, 0.0_02, 0.02]		):
							self.check_over_configs(beta_start=_A							, beta_end=_A		)
	def 					_lowercase    (   self		):






				'''simple docstring'''







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






				'''simple docstring'''







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






				'''simple docstring'''







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






				'''simple docstring'''







				UpperCAmelCase       		=    self.scheduler_classes[0]
				UpperCAmelCase       		=    self.get_scheduler_config()
				UpperCAmelCase       		=    scheduler_class(**_A		)

				scheduler.set_timesteps(self.num_inference_steps		)

				UpperCAmelCase       		=    self.dummy_model()
				UpperCAmelCase       		=    self.dummy_sample_deter * scheduler.init_noise_sigma
				UpperCAmelCase       		=    sample.to(_A		)

				for i, t in enumerate(scheduler.timesteps		):
							UpperCAmelCase       		=    scheduler.scale_model_input(_A							, _A		)

							UpperCAmelCase       		=    model(_A							, _A		)

							UpperCAmelCase       		=    scheduler.step(_A							, _A							, _A		)
							UpperCAmelCase       		=    output.prev_sample

				UpperCAmelCase       		=    torch.sum(torch.abs(_A		)		)
				UpperCAmelCase       		=    torch.mean(torch.abs(_A		)		)

				if torch_device in ["cpu", "mps"]:
							assert abs(result_sum.item() - 0.12_33		) < 1E-2
							assert abs(result_mean.item() - 0.00_02		) < 1E-3
				else:
							# CUDA
							assert abs(result_sum.item() - 0.12_33		) < 1E-2
							assert abs(result_mean.item() - 0.00_02		) < 1E-3
	def 					_lowercase    (   self		):






				'''simple docstring'''







				UpperCAmelCase       		=    self.scheduler_classes[0]
				UpperCAmelCase       		=    self.get_scheduler_config(prediction_type='''v_prediction'''		)
				UpperCAmelCase       		=    scheduler_class(**_A		)

				scheduler.set_timesteps(self.num_inference_steps		)

				UpperCAmelCase       		=    self.dummy_model()
				UpperCAmelCase       		=    self.dummy_sample_deter * scheduler.init_noise_sigma
				UpperCAmelCase       		=    sample.to(_A		)

				for i, t in enumerate(scheduler.timesteps		):
							UpperCAmelCase       		=    scheduler.scale_model_input(_A							, _A		)

							UpperCAmelCase       		=    model(_A							, _A		)

							UpperCAmelCase       		=    scheduler.step(_A							, _A							, _A		)
							UpperCAmelCase       		=    output.prev_sample

				UpperCAmelCase       		=    torch.sum(torch.abs(_A		)		)
				UpperCAmelCase       		=    torch.mean(torch.abs(_A		)		)

				if torch_device in ["cpu", "mps"]:
							assert abs(result_sum.item() - 4.6934E-07		) < 1E-2
							assert abs(result_mean.item() - 6.1112E-10		) < 1E-3
				else:
							# CUDA
							assert abs(result_sum.item() - 4.693_4286_5017_0972E-07		) < 1E-2
							assert abs(result_mean.item() - 0.00_02		) < 1E-3
	def 					_lowercase    (   self		):






				'''simple docstring'''







				UpperCAmelCase       		=    self.scheduler_classes[0]
				UpperCAmelCase       		=    self.get_scheduler_config()
				UpperCAmelCase       		=    scheduler_class(**_A		)

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

				UpperCAmelCase       		=    self.dummy_model()
				UpperCAmelCase       		=    self.dummy_sample_deter.to(_A		) * scheduler.init_noise_sigma

				for t in scheduler.timesteps:
							UpperCAmelCase       		=    scheduler.scale_model_input(_A							, _A		)

							UpperCAmelCase       		=    model(_A							, _A		)

							UpperCAmelCase       		=    scheduler.step(_A							, _A							, _A		)
							UpperCAmelCase       		=    output.prev_sample

				UpperCAmelCase       		=    torch.sum(torch.abs(_A		)		)
				UpperCAmelCase       		=    torch.mean(torch.abs(_A		)		)

				if str(_A		).startswith('''cpu'''		):
							# The following sum varies between 148 and 156 on mps. Why?
							assert abs(result_sum.item() - 0.12_33		) < 1E-2
							assert abs(result_mean.item() - 0.00_02		) < 1E-3
				elif str(_A		).startswith('''mps'''		):
							# Larger tolerance on mps
							assert abs(result_mean.item() - 0.00_02		) < 1E-2
				else:
							# CUDA
							assert abs(result_sum.item() - 0.12_33		) < 1E-2
							assert abs(result_mean.item() - 0.00_02		) < 1E-3







	def 					_lowercase    (   self		):






				'''simple docstring'''







				UpperCAmelCase       		=    self.scheduler_classes[0]
				UpperCAmelCase       		=    self.get_scheduler_config()
				UpperCAmelCase       		=    scheduler_class(**_A							, use_karras_sigmas=_A		)

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

				UpperCAmelCase       		=    self.dummy_model()
				UpperCAmelCase       		=    self.dummy_sample_deter.to(_A		) * scheduler.init_noise_sigma
				UpperCAmelCase       		=    sample.to(_A		)

				for t in scheduler.timesteps:
							UpperCAmelCase       		=    scheduler.scale_model_input(_A							, _A		)

							UpperCAmelCase       		=    model(_A							, _A		)

							UpperCAmelCase       		=    scheduler.step(_A							, _A							, _A		)
							UpperCAmelCase       		=    output.prev_sample

				UpperCAmelCase       		=    torch.sum(torch.abs(_A		)		)
				UpperCAmelCase       		=    torch.mean(torch.abs(_A		)		)

				assert abs(result_sum.item() - 0.0_00_15		) < 1E-2
				assert abs(result_mean.item() - 1.9869_5545_3503_4695E-07		) < 1E-2










from typing import SupportsIndex

import numpy as np
from scipy.ndimage import convolve





def 						__SCREAMING_SNAKE_CASE  (    UpperCamelCase__      ,      UpperCamelCase__      ,      UpperCamelCase__			)  ->    np.ndarray:







			'''simple docstring'''

			UpperCAmelCase       		=    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       		=    np.stack(
			    np.meshgrid(np.arange(0      ,      image.shape[1]			)      ,      np.arange(0      ,      image.shape[0]			)			)      ,      2			)
			UpperCAmelCase       		=    np.round(grid - flow			).astype(np.intaa			)

			# Find the locations outside of the original image
			UpperCAmelCase       		=    (grid < 0) | (grid >= np.array([image.shape[1], image.shape[0]]			))
			UpperCAmelCase       		=    0

			UpperCAmelCase       		=    image[grid[:, :, 1], grid[:, :, 0]]

			# Set pixels at invalid locations to 0
			UpperCAmelCase       		=    0

			return warped





def 						__SCREAMING_SNAKE_CASE  (    UpperCamelCase__      ,      UpperCamelCase__      ,      UpperCamelCase__      ,      UpperCamelCase__ = None      ,      )  ->    tuple[np.ndarray, np.ndarray]:







			'''simple docstring'''

			if alpha is None:
						UpperCAmelCase       		=    0.1

			# Initialize flow
			UpperCAmelCase       		=    np.zeros_like(UpperCamelCase__			)
			UpperCAmelCase       		=    np.zeros_like(UpperCamelCase__			)

			# Prepare kernels for the calculation of the derivatives and the average velocity
			UpperCAmelCase       		=    np.array([[-1, 1], [-1, 1]]			) * 0.25
			UpperCAmelCase       		=    np.array([[-1, -1], [1, 1]]			) * 0.25
			UpperCAmelCase       		=    np.array([[1, 1], [1, 1]]			) * 0.25
			UpperCAmelCase       		=    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(UpperCamelCase__			):
						UpperCAmelCase       		=    warp(UpperCamelCase__      ,      UpperCamelCase__      ,      UpperCamelCase__			)
						UpperCAmelCase       		=    convolve(UpperCamelCase__      ,      UpperCamelCase__			) + convolve(UpperCamelCase__      ,      UpperCamelCase__			)
						UpperCAmelCase       		=    convolve(UpperCamelCase__      ,      UpperCamelCase__			) + convolve(UpperCamelCase__      ,      UpperCamelCase__			)
						UpperCAmelCase       		=    convolve(UpperCamelCase__      ,      UpperCamelCase__			) + convolve(UpperCamelCase__      ,      -kernel_t			)

						UpperCAmelCase       		=    convolve(UpperCamelCase__      ,      UpperCamelCase__			)
						UpperCAmelCase       		=    convolve(UpperCamelCase__      ,      UpperCamelCase__			)

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

						UpperCAmelCase       		=    avg_horizontal_velocity - derivative_x * update
						UpperCAmelCase       		=    avg_vertical_velocity - derivative_y * update

			return horizontal_flow, vertical_flow


if __name__ == "__main__":
				import doctest

				doctest.testmod()



