






import math
import warnings
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch
from scipy import integrate

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin
@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->LMSDiscrete
class        __snake_case(					__UpperCamelCase      ):
 lowercase :       Optional[int]		      =		42
 lowercase :       int		      =		None
def       a__						(  A		,				A=0.9_9_9		,				A="cosine"		,				):
   if alpha_transform_type == "cosine":

      def alpha_bar_fn(A					):
         return math.cos((t + 0.0_0_8) / 1.0_0_8 * math.pi / 2					) ** 2

   elif alpha_transform_type == "exp":

      def alpha_bar_fn(A					):
         return math.exp(t * -1_2.0					)

   else:
      raise ValueError(F'''Unsupported alpha_tranform_type: {alpha_transform_type}'''					)

   lowercase__						:	int         = []
   for i in range(UpperCamelCase__					):
      lowercase__						:	str         = i / num_diffusion_timesteps
      lowercase__						:	Dict         = (i + 1) / num_diffusion_timesteps
      betas.append(min(1 - alpha_bar_fn(UpperCamelCase__					) / alpha_bar_fn(UpperCamelCase__					)		,				UpperCamelCase__					)					)
   return torch.tensor(UpperCamelCase__		,				dtype=torch.floataa					)






class        __snake_case(					__UpperCamelCase     ,       __UpperCamelCase      ):
 lowercase :       Optional[Any]		      =		[e.name for e in KarrasDiffusionSchedulers]
 lowercase :       List[str]		      =		1



 @register_to_config
 def __init__(	self   :   Dict ,		UpperCAmelCase__   :   Tuple = 1_000 ,		UpperCAmelCase__   :   Any = 0.0_001 ,		UpperCAmelCase__   :   Tuple = 0.02 ,		UpperCAmelCase__   :   str = "linear" ,		UpperCAmelCase__   :   Optional[int] = None ,		UpperCAmelCase__   :   Dict = False ,		UpperCAmelCase__   :   List[str] = "epsilon" ,		UpperCAmelCase__   :   str = "linspace" ,		UpperCAmelCase__   :   Dict = 0 ,		):

    """simple docstring"""






    if trained_betas is not None:
       lowercase__						:	int         = torch.tensor(__lowerCamelCase ,		dtype=torch.floataa	)
    elif beta_schedule == "linear":
       lowercase__						:	str         = torch.linspace(__lowerCamelCase ,		__lowerCamelCase ,		__lowerCamelCase ,		dtype=torch.floataa	)
    elif beta_schedule == "scaled_linear":
       # this schedule is very specific to the latent diffusion model.
       lowercase__						:	int         = (
           torch.linspace(beta_start**0.5 ,		beta_end**0.5 ,		__lowerCamelCase ,		dtype=torch.floataa	) ** 2
       )
    elif beta_schedule == "squaredcos_cap_v2":
       # Glide cosine schedule
       lowercase__						:	List[str]         = betas_for_alpha_bar(__lowerCamelCase	)
    else:
       raise NotImplementedError(f'''{beta_schedule} does is not implemented for {self.__class__}'''	)

    lowercase__						:	Dict         = 1.0 - self.betas
    lowercase__						:	List[str]         = torch.cumprod(self.alphas ,		dim=0	)

    lowercase__						:	Optional[Any]         = np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5	)
    lowercase__						:	str         = np.concatenate([sigmas[::-1], [0.0]]	).astype(np.floataa	)
    lowercase__						:	str         = torch.from_numpy(__lowerCamelCase	)

    # setable values
    lowercase__						:	Union[str, Any]         = None
    lowercase__						:	Dict         = use_karras_sigmas
    self.set_timesteps(__lowerCamelCase ,		__lowerCamelCase	)
    lowercase__						:	Tuple         = []
    lowercase__						:	Union[str, Any]         = False



 @property
 def 						UpperCAmelCase    (	self   :   str	):

    """simple docstring"""






    if self.config.timestep_spacing in ["linspace", "trailing"]:
       return self.sigmas.max()

    return (self.sigmas.max() ** 2 + 1) ** 0.5



 def 						UpperCAmelCase    (	self   :   Union[str, Any] ,		UpperCAmelCase__   :   List[str] ,		UpperCAmelCase__   :   Optional[Any]	):

    """simple docstring"""






    if isinstance(__lowerCamelCase ,		torch.Tensor	):
       lowercase__						:	Tuple         = timestep.to(self.timesteps.device	)
    lowercase__						:	Any         = (self.timesteps == timestep).nonzero().item()
    lowercase__						:	List[str]         = self.sigmas[step_index]
    lowercase__						:	List[Any]         = sample / ((sigma**2 + 1) ** 0.5)
    lowercase__						:	Union[str, Any]         = True
    return sample



 def 						UpperCAmelCase    (	self   :   Tuple ,		UpperCAmelCase__   :   Tuple ,		UpperCAmelCase__   :   int ,		UpperCAmelCase__   :   str	):

    """simple docstring"""






    def lms_derivative(UpperCAmelCase__   :   List[str]	):
       lowercase__						:	Optional[Any]         = 1.0
       for k in range(__lowerCamelCase	):
          if current_order == k:
             continue
          prod *= (tau - self.sigmas[t - k]) / (self.sigmas[t - current_order] - self.sigmas[t - k])
       return prod

    lowercase__						:	Optional[Any]         = integrate.quad(__lowerCamelCase ,		self.sigmas[t] ,		self.sigmas[t + 1] ,		epsrel=1e-4	)[0]

    return integrated_coeff



 def 						UpperCAmelCase    (	self   :   int ,		UpperCAmelCase__   :   List[str] ,		UpperCAmelCase__   :   str = None	):

    """simple docstring"""






    lowercase__						:	Dict         = num_inference_steps

    # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
    if self.config.timestep_spacing == "linspace":
       lowercase__						:	Optional[int]         = np.linspace(0 ,		self.config.num_train_timesteps - 1 ,		__lowerCamelCase ,		dtype=__lowerCamelCase	)[
           ::-1
       ].copy()
    elif self.config.timestep_spacing == "leading":
       lowercase__						:	Optional[int]         = self.config.num_train_timesteps // self.num_inference_steps
       # creates integer timesteps by multiplying by ratio
       # casting to int to avoid issues when num_inference_step is power of 3
       lowercase__						:	Optional[Any]         = (np.arange(0 ,		__lowerCamelCase	) * step_ratio).round()[::-1].copy().astype(__lowerCamelCase	)
       timesteps += self.config.steps_offset
    elif self.config.timestep_spacing == "trailing":
       lowercase__						:	Union[str, Any]         = self.config.num_train_timesteps / self.num_inference_steps
       # creates integer timesteps by multiplying by ratio
       # casting to int to avoid issues when num_inference_step is power of 3
       lowercase__						:	Union[str, Any]         = (np.arange(self.config.num_train_timesteps ,		0 ,		-step_ratio	)).round().copy().astype(__lowerCamelCase	)
       timesteps -= 1
    else:
       raise ValueError(
           f'''{self.config.timestep_spacing} is not supported. Please make sure to choose one of \'linspace\', \'leading\' or \'trailing\'.'''	)

    lowercase__						:	List[Any]         = np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5	)
    lowercase__						:	List[Any]         = np.log(__lowerCamelCase	)
    lowercase__						:	Optional[Any]         = np.interp(__lowerCamelCase ,		np.arange(0 ,		len(__lowerCamelCase	)	) ,		__lowerCamelCase	)

    if self.use_karras_sigmas:
       lowercase__						:	Union[str, Any]         = self._convert_to_karras(in_sigmas=__lowerCamelCase	)
       lowercase__						:	List[Any]         = np.array([self._sigma_to_t(__lowerCamelCase ,		__lowerCamelCase	) for sigma in sigmas]	)

    lowercase__						:	Union[str, Any]         = np.concatenate([sigmas, [0.0]]	).astype(np.floataa	)

    lowercase__						:	List[Any]         = torch.from_numpy(__lowerCamelCase	).to(device=__lowerCamelCase	)
    if str(__lowerCamelCase	).startswith("""mps"""	):
       # mps does not support float64
       lowercase__						:	Dict         = torch.from_numpy(__lowerCamelCase	).to(__lowerCamelCase ,		dtype=torch.floataa	)
    else:
       lowercase__						:	Any         = torch.from_numpy(__lowerCamelCase	).to(device=__lowerCamelCase	)

    lowercase__						:	str         = []



 def 						UpperCAmelCase    (	self   :   List[Any] ,		UpperCAmelCase__   :   Dict ,		UpperCAmelCase__   :   str	):

    """simple docstring"""






    lowercase__						:	Dict         = np.log(__lowerCamelCase	)

    # get distribution
    lowercase__						:	Any         = log_sigma - log_sigmas[:, np.newaxis]

    # get sigmas range
    lowercase__						:	List[str]         = np.cumsum((dists >= 0) ,		axis=0	).argmax(axis=0	).clip(max=log_sigmas.shape[0] - 2	)
    lowercase__						:	int         = low_idx + 1

    lowercase__						:	Optional[int]         = log_sigmas[low_idx]
    lowercase__						:	int         = log_sigmas[high_idx]

    # interpolate sigmas
    lowercase__						:	Optional[int]         = (low - log_sigma) / (low - high)
    lowercase__						:	Optional[int]         = np.clip(__lowerCamelCase ,		0 ,		1	)

    # transform interpolation to time range
    lowercase__						:	Any         = (1 - w) * low_idx + w * high_idx
    lowercase__						:	int         = t.reshape(sigma.shape	)
    return t



 def 						UpperCAmelCase    (	self   :   List[str] ,		UpperCAmelCase__   :   List[Any]	):

    """simple docstring"""






    lowercase__						:	float         = in_sigmas[-1].item()
    lowercase__						:	float         = in_sigmas[0].item()

    lowercase__						:	str         = 7.0  # 7.0 is the value used in the paper
    lowercase__						:	Dict         = np.linspace(0 ,		1 ,		self.num_inference_steps	)
    lowercase__						:	int         = sigma_min ** (1 / rho)
    lowercase__						:	Optional[int]         = sigma_max ** (1 / rho)
    lowercase__						:	Tuple         = (max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
    return sigmas



 def 						UpperCAmelCase    (	self   :   str ,		UpperCAmelCase__   :   str ,		UpperCAmelCase__   :   List[str] ,		UpperCAmelCase__   :   Dict ,		UpperCAmelCase__   :   Dict = 4 ,		UpperCAmelCase__   :   Tuple = True ,		):

    """simple docstring"""






    if not self.is_scale_input_called:
       warnings.warn(
           """The `scale_model_input` function should be called before `step` to ensure correct denoising. """
           """See `StableDiffusionPipeline` for a usage example."""	)

    if isinstance(__lowerCamelCase ,		torch.Tensor	):
       lowercase__						:	Optional[int]         = timestep.to(self.timesteps.device	)
    lowercase__						:	str         = (self.timesteps == timestep).nonzero().item()
    lowercase__						:	Union[str, Any]         = self.sigmas[step_index]

    # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
    if self.config.prediction_type == "epsilon":
       lowercase__						:	Optional[int]         = sample - sigma * model_output
    elif self.config.prediction_type == "v_prediction":
       # * c_out + input * c_skip
       lowercase__						:	List[str]         = model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
    elif self.config.prediction_type == "sample":
       lowercase__						:	int         = model_output
    else:
       raise ValueError(
           f'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`'''	)

    # 2. Convert to an ODE derivative
    lowercase__						:	List[Any]         = (sample - pred_original_sample) / sigma
    self.derivatives.append(__lowerCamelCase	)
    if len(self.derivatives	) > order:
       self.derivatives.pop(0	)

    # 3. Compute linear multistep coefficients
    lowercase__						:	List[str]         = min(step_index + 1 ,		__lowerCamelCase	)
    lowercase__						:	Optional[Any]         = [self.get_lms_coefficient(__lowerCamelCase ,		__lowerCamelCase ,		__lowerCamelCase	) for curr_order in range(__lowerCamelCase	)]

    # 4. Compute previous sample based on the derivatives path
    lowercase__						:	Union[str, Any]         = sample + sum(
        coeff * derivative for coeff, derivative in zip(__lowerCamelCase ,		reversed(self.derivatives	)	)	)

    if not return_dict:
       return (prev_sample,)

    return LMSDiscreteSchedulerOutput(prev_sample=__lowerCamelCase ,		pred_original_sample=__lowerCamelCase	)



 def 						UpperCAmelCase    (	self   :   int ,		UpperCAmelCase__   :   Tuple ,		UpperCAmelCase__   :   Dict ,		UpperCAmelCase__   :   int ,		):

    """simple docstring"""






    lowercase__						:	Dict         = self.sigmas.to(device=original_samples.device ,		dtype=original_samples.dtype	)
    if original_samples.device.type == "mps" and torch.is_floating_point(__lowerCamelCase	):
       # mps does not support float64
       lowercase__						:	Union[str, Any]         = self.timesteps.to(original_samples.device ,		dtype=torch.floataa	)
       lowercase__						:	Tuple         = timesteps.to(original_samples.device ,		dtype=torch.floataa	)
    else:
       lowercase__						:	Optional[int]         = self.timesteps.to(original_samples.device	)
       lowercase__						:	Union[str, Any]         = timesteps.to(original_samples.device	)

    lowercase__						:	str         = [(schedule_timesteps == t).nonzero().item() for t in timesteps]

    lowercase__						:	Optional[Any]         = sigmas[step_indices].flatten()
    while len(sigma.shape	) < len(original_samples.shape	):
       lowercase__						:	str         = sigma.unsqueeze(-1	)

    lowercase__						:	List[Any]         = original_samples + noise * sigma
    return noisy_samples







 def __len__(	self   :   str	):

    """simple docstring"""






    return self.config.num_train_timesteps










from typing import SupportsIndex

import numpy as np
from scipy.ndimage import convolve
def       a__						(  A		,				A		,				A					):
   lowercase__						:	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
   lowercase__						:	Tuple         = np.stack(
       np.meshgrid(np.arange(0		,				image.shape[1]					)		,				np.arange(0		,				image.shape[0]					)					)		,				2					)
   lowercase__						:	Union[str, Any]         = np.round(grid - flow					).astype(np.intaa					)

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

   lowercase__						:	Tuple         = image[grid[:, :, 1], grid[:, :, 0]]

   # Set pixels at invalid locations to 0
   lowercase__						:	str         = 0

   return warped
def       a__						(  A		,				A		,				A		,				A = None		,				):
   if alpha is None:
      lowercase__						:	Tuple         = 0.1

   # Initialize flow
   lowercase__						:	Optional[int]         = np.zeros_like(A					)
   lowercase__						:	Tuple         = np.zeros_like(A					)

   # Prepare kernels for the calculation of the derivatives and the average velocity
   lowercase__						:	Tuple         = np.array([[-1, 1], [-1, 1]]					) * 0.2_5
   lowercase__						:	int         = np.array([[-1, -1], [1, 1]]					) * 0.2_5
   lowercase__						:	Optional[int]         = np.array([[1, 1], [1, 1]]					) * 0.2_5
   lowercase__						:	Optional[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(A					):
      lowercase__						:	Optional[int]         = warp(A		,				A		,				A					)
      lowercase__						:	Union[str, Any]         = convolve(A		,				A					) + convolve(A		,				A					)
      lowercase__						:	Union[str, Any]         = convolve(A		,				A					) + convolve(A		,				A					)
      lowercase__						:	str         = convolve(A		,				A					) + convolve(A		,				-kernel_t					)

      lowercase__						:	str         = convolve(A		,				A					)
      lowercase__						:	Tuple         = convolve(A		,				A					)

      # This updates the flow as proposed in the paper (Step 12)
      lowercase__						:	List[Any]         = (
          derivative_x * avg_horizontal_velocity
          + derivative_y * avg_vertical_velocity
          + derivative_t
      )
      lowercase__						:	Union[str, Any]         = update / (alpha**2 + derivative_x**2 + derivative_y**2)

      lowercase__						:	Optional[int]         = avg_horizontal_velocity - derivative_x * update
      lowercase__						:	int         = avg_vertical_velocity - derivative_y * update

   return horizontal_flow, vertical_flow


if __name__ == "__main__":
 import doctest

 doctest.testmod()


