



'''simple docstring'''
def A__					(  UpperCAmelCase_						,						UpperCAmelCase_ = 1_0_0_0							):



      if (not isinstance(__A						,						__A							)) or (digit_position <= 0):
            raise ValueError('Digit position must be a positive integer'							)
      elif (not isinstance(__A						,						__A							)) or (precision < 0):
            raise ValueError('Precision must be a nonnegative integer'							)

      # compute an approximation of (16 ** (n - 1)) * pi whose fractional part is mostly
      # accurate
      _UpperCamelCase      :	Tuple				       =     (
          4 * _subsum(__A						,						1						,						__A							)
          - 2 * _subsum(__A						,						4						,						__A							)
          - _subsum(__A						,						5						,						__A							)
          - _subsum(__A						,						6						,						__A							)
      )

      # return the first hex digit of the fractional part of the result
      return hex(int((sum_result % 1) * 1_6							)							)[2:]






def A__					(  UpperCAmelCase_						,						UpperCAmelCase_						,						UpperCAmelCase_							):
      _UpperCamelCase      :	Any				       =     0.0
      for sum_index in range(digit_pos_to_extract + precision							):
            _UpperCamelCase      :	Any				       =     8 * sum_index + denominator_addend
            if sum_index < digit_pos_to_extract:
                  # if the exponential term is an integer and we mod it by the denominator
                  # before dividing, only the integer part of the sum will change;
                  # the fractional part will not
                  _UpperCamelCase      :	Dict				       =     pow(
                      1_6						,						digit_pos_to_extract - 1 - sum_index						,						__A							)
            else:
                  _UpperCamelCase      :	str				       =     pow(1_6						,						digit_pos_to_extract - 1 - sum_index							)
            total += exponential_term / denominator
      return total


if __name__ == "__main__":
     import doctest

     doctest.testmod()







'''simple docstring'''
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 							lowercase__			(							lowercase     ):
	lowercase__						       =    42
	lowercase__						       =    None






def A__					(  UpperCAmelCase_						,						UpperCAmelCase_=0.999						,						UpperCAmelCase_="cosine"						,						):
						if alpha_transform_type == "cosine":

												def alpha_bar_fn(UpperCAmelCase_							):
																		return math.cos((t + 0.008) / 1.008 * math.pi / 2							) ** 2

						elif alpha_transform_type == "exp":

												def alpha_bar_fn(UpperCAmelCase_							):
																		return math.exp(t * -12.0							)

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

						_UpperCamelCase      :	Tuple				       =     []
						for i in range(UpperCAmelCase_							):
												_UpperCamelCase      :	List[str]				       =     i / num_diffusion_timesteps
												_UpperCamelCase      :	Optional[int]				       =     (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 							lowercase__			(							lowercase		,      lowercase     ):
	lowercase__						       =    [e.name for e in KarrasDiffusionSchedulers]
	lowercase__						       =    1





	@register_to_config
	def __init__(				self      :       int				,lowerCamelCase__      :       int = 1000				,lowerCamelCase__      :       float = 0.0_0_0_1				,lowerCamelCase__      :       float = 0.0_2				,lowerCamelCase__      :       str = "linear"				,lowerCamelCase__      :       Optional[Union[np.ndarray, List[float]]] = None				,lowerCamelCase__      :       Optional[bool] = False				,lowerCamelCase__      :       str = "epsilon"				,lowerCamelCase__      :       str = "linspace"				,lowerCamelCase__      :       int = 0				,):







							'''simple docstring'''
							if trained_betas is not None:
													_UpperCamelCase      :	Union[str, Any]				       =     torch.tensor(lowerCamelCase__				,dtype=torch.floataa				)
							elif beta_schedule == "linear":
													_UpperCamelCase      :	Union[str, Any]				       =     torch.linspace(lowerCamelCase__				,lowerCamelCase__				,lowerCamelCase__				,dtype=torch.floataa				)
							elif beta_schedule == "scaled_linear":
													# this schedule is very specific to the latent diffusion model.
													_UpperCamelCase      :	Optional[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
													_UpperCamelCase      :	Dict				       =     betas_for_alpha_bar(lowerCamelCase__				)
							else:
													raise NotImplementedError(F'{beta_schedule} does is not implemented for {self.__class__}'				)

							_UpperCamelCase      :	List[Any]				       =     1.0 - self.betas
							_UpperCamelCase      :	int				       =     torch.cumprod(self.alphas				,dim=0				)

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

							# setable values
							_UpperCamelCase      :	List[str]				       =     None
							_UpperCamelCase      :	List[Any]				       =     use_karras_sigmas
							self.set_timesteps(lowerCamelCase__				,lowerCamelCase__				)
							_UpperCamelCase      :	Any				       =     []
							_UpperCamelCase      :	Tuple				       =     False





	@property
	def 			UpperCamelCase_						(				self      :       Union[str, Any]				):







							'''simple docstring'''
							# standard deviation of the initial noise distribution
							if self.config.timestep_spacing in ["linspace", "trailing"]:
													return self.sigmas.max()

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





	def 			UpperCamelCase_						(				self      :       Optional[int]				,lowerCamelCase__      :       torch.FloatTensor				,lowerCamelCase__      :       Union[float, torch.FloatTensor]				):







							'''simple docstring'''
							if isinstance(lowerCamelCase__				,torch.Tensor				):
													_UpperCamelCase      :	List[str]				       =     timestep.to(self.timesteps.device				)
							_UpperCamelCase      :	Optional[Any]				       =     (self.timesteps == timestep).nonzero().item()
							_UpperCamelCase      :	Any				       =     self.sigmas[step_index]
							_UpperCamelCase      :	Dict				       =     sample / ((sigma**2 + 1) ** 0.5)
							_UpperCamelCase      :	Optional[Any]				       =     True
							return sample





	def 			UpperCamelCase_						(				self      :       List[Any]				,lowerCamelCase__      :       Optional[int]				,lowerCamelCase__      :       int				,lowerCamelCase__      :       Optional[Any]				):







							'''simple docstring'''

							def lms_derivative(lowerCamelCase__      :       str				):
													_UpperCamelCase      :	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

							_UpperCamelCase      :	Union[str, Any]				       =     integrate.quad(lowerCamelCase__				,self.sigmas[t]				,self.sigmas[t + 1]				,epsrel=1E-4				)[0]

							return integrated_coeff





	def 			UpperCamelCase_						(				self      :       Dict				,lowerCamelCase__      :       int				,lowerCamelCase__      :       Union[str, torch.device] = None				):







							'''simple docstring'''
							_UpperCamelCase      :	Optional[int]				       =     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":
													_UpperCamelCase      :	int				       =     np.linspace(0				,self.config.num_train_timesteps - 1				,lowerCamelCase__				,dtype=lowerCamelCase__				)[
													    ::-1
													].copy()
							elif self.config.timestep_spacing == "leading":
													_UpperCamelCase      :	Optional[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
													_UpperCamelCase      :	str				       =     (np.arange(0				,lowerCamelCase__				) * step_ratio).round()[::-1].copy().astype(lowerCamelCase__				)
													timesteps += self.config.steps_offset
							elif self.config.timestep_spacing == "trailing":
													_UpperCamelCase      :	Optional[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
													_UpperCamelCase      :	Optional[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\'.'				)

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

							if self.use_karras_sigmas:
													_UpperCamelCase      :	Optional[int]				       =     self._convert_to_karras(in_sigmas=lowerCamelCase__				)
													_UpperCamelCase      :	str				       =     np.array([self._sigma_to_t(lowerCamelCase__				,lowerCamelCase__				) for sigma in sigmas]				)

							_UpperCamelCase      :	List[str]				       =     np.concatenate([sigmas, [0.0]]				).astype(np.floataa				)

							_UpperCamelCase      :	Any				       =     torch.from_numpy(lowerCamelCase__				).to(device=lowerCamelCase__				)
							if str(lowerCamelCase__				).startswith('mps'				):
													# mps does not support float64
													_UpperCamelCase      :	Optional[Any]				       =     torch.from_numpy(lowerCamelCase__				).to(lowerCamelCase__				,dtype=torch.floataa				)
							else:
													_UpperCamelCase      :	Any				       =     torch.from_numpy(lowerCamelCase__				).to(device=lowerCamelCase__				)

							_UpperCamelCase      :	Union[str, Any]				       =     []





	def 			UpperCamelCase_						(				self      :       Any				,lowerCamelCase__      :       Any				,lowerCamelCase__      :       List[str]				):







							'''simple docstring'''
							# get log sigma
							_UpperCamelCase      :	Tuple				       =     np.log(lowerCamelCase__				)

							# get distribution
							_UpperCamelCase      :	Dict				       =     log_sigma - log_sigmas[:, np.newaxis]

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

							_UpperCamelCase      :	Optional[int]				       =     log_sigmas[low_idx]
							_UpperCamelCase      :	Union[str, Any]				       =     log_sigmas[high_idx]

							# interpolate sigmas
							_UpperCamelCase      :	List[Any]				       =     (low - log_sigma) / (low - high)
							_UpperCamelCase      :	int				       =     np.clip(lowerCamelCase__				,0				,1				)

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





	def 			UpperCamelCase_						(				self      :       Optional[Any]				,lowerCamelCase__      :       torch.FloatTensor				):







							'''simple docstring'''
							_UpperCamelCase      :	float				       =     in_sigmas[-1].item()
							_UpperCamelCase      :	float				       =     in_sigmas[0].item()

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





	def 			UpperCamelCase_						(				self      :       Optional[Any]				,lowerCamelCase__      :       torch.FloatTensor				,lowerCamelCase__      :       Union[float, torch.FloatTensor]				,lowerCamelCase__      :       torch.FloatTensor				,lowerCamelCase__      :       int = 4				,lowerCamelCase__      :       bool = 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				):
													_UpperCamelCase      :	Union[str, Any]				       =     timestep.to(self.timesteps.device				)
							_UpperCamelCase      :	Any				       =     (self.timesteps == timestep).nonzero().item()
							_UpperCamelCase      :	Optional[int]				       =     self.sigmas[step_index]

							# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
							if self.config.prediction_type == "epsilon":
													_UpperCamelCase      :	Dict				       =     sample - sigma * model_output
							elif self.config.prediction_type == "v_prediction":
													# * c_out + input * c_skip
													_UpperCamelCase      :	str				       =     model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
							elif self.config.prediction_type == "sample":
													_UpperCamelCase      :	Any				       =     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
							_UpperCamelCase      :	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
							_UpperCamelCase      :	List[Any]				       =     min(step_index + 1				,lowerCamelCase__				)
							_UpperCamelCase      :	Optional[int]				       =     [self.get_lms_coefficient(lowerCamelCase__				,lowerCamelCase__				,lowerCamelCase__				) for curr_order in range(lowerCamelCase__				)]

							# 4. Compute previous sample based on the derivatives path
							_UpperCamelCase      :	int				       =     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      :       Optional[Any]				,lowerCamelCase__      :       torch.FloatTensor				,lowerCamelCase__      :       torch.FloatTensor				,lowerCamelCase__      :       torch.FloatTensor				,):







							'''simple docstring'''
							# Make sure sigmas and timesteps have the same device and dtype as original_samples
							_UpperCamelCase      :	Tuple				       =     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
													_UpperCamelCase      :	str				       =     self.timesteps.to(original_samples.device				,dtype=torch.floataa				)
													_UpperCamelCase      :	Optional[Any]				       =     timesteps.to(original_samples.device				,dtype=torch.floataa				)
							else:
													_UpperCamelCase      :	Union[str, Any]				       =     self.timesteps.to(original_samples.device				)
													_UpperCamelCase      :	Union[str, Any]				       =     timesteps.to(original_samples.device				)

							_UpperCamelCase      :	Union[str, Any]				       =     [(schedule_timesteps == t).nonzero().item() for t in timesteps]

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

							_UpperCamelCase      :	str				       =     original_samples + noise * sigma
							return noisy_samples





	def __len__(				self      :       List[Any]				):







							'''simple docstring'''
							return self.config.num_train_timesteps


