




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					):



   '''simple docstring'''






   snake_case_   :	torch.FloatTensor
   snake_case_   :	Optional[torch.FloatTensor]       =						None



def  lowercase    (				SCREAMING_SNAKE_CASE__    :		Tuple			,    SCREAMING_SNAKE_CASE__    :		str=0.9_9_9			,    SCREAMING_SNAKE_CASE__    :		List[Any]="cosine"			,    )					->       Tuple:
 if alpha_transform_type == "cosine":

  def alpha_bar_fn(SCREAMING_SNAKE_CASE__    :		Union[str, Any]			):
   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(SCREAMING_SNAKE_CASE__    :		str			):
   return math.exp(t * -12.0			)

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

 _snake_case      :					List[Any]					=   []
 for i in range(lowercase__			):
  _snake_case      :					Any					=   i / num_diffusion_timesteps
  _snake_case      :					Any					=   (i + 1) / num_diffusion_timesteps
  betas.append(min(1 - alpha_bar_fn(lowercase__			) / alpha_bar_fn(lowercase__			)			,    lowercase__			)			)
 return torch.tensor(lowercase__			,    dtype=torch.floataa			)







class 						snake_case  (		__UpperCAmelCase   ,__UpperCAmelCase					):



   '''simple docstring'''






   snake_case_   :	str       =						[e.name for e in KarrasDiffusionSchedulers]
   snake_case_   :	int       =						1

   @register_to_config
   def __init__(   self							:     Union[str, Any]			,						lowerCAmelCase							:     Any = 1000			,						lowerCAmelCase							:     Optional[Any] = 0.0_001			,						lowerCAmelCase							:     Optional[Any] = 0.02			,						lowerCAmelCase							:     Tuple = "linear"			,						lowerCAmelCase							:     Tuple = None			,						lowerCAmelCase							:     Union[str, Any] = False			,						lowerCAmelCase							:     Optional[int] = "epsilon"			,						lowerCAmelCase							:     Optional[Any] = "linspace"			,						lowerCAmelCase							:     Optional[int] = 0			,						)			->			Any:

    """simple docstring"""







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

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

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

    # setable values
    _snake_case      :					str					=   None
    _snake_case      :					Tuple					=   use_karras_sigmas
    self.set_timesteps(lowerCamelCase__			,						lowerCamelCase__)
    _snake_case      :					Tuple					=   []
    _snake_case      :					int					=   False

   @property
   def        UpperCamelCase_					(   self							:     Any)			->			Any:

    """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							:     Optional[Any]			,						lowerCAmelCase							:     Optional[int]			,						lowerCAmelCase							:     int)			->			torch.FloatTensor:

    """simple docstring"""







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

   def        UpperCamelCase_					(   self							:     List[str]			,						lowerCAmelCase							:     Optional[Any]			,						lowerCAmelCase							:     Tuple			,						lowerCAmelCase							:     str)			->			int:

    """simple docstring"""







    def lms_derivative(lowerCAmelCase							:     List[Any]):
     _snake_case      :					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

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

    return integrated_coeff

   def        UpperCamelCase_					(   self							:     Dict			,						lowerCAmelCase							:     Optional[Any]			,						lowerCAmelCase							:     Any = None)			->			Optional[Any]:

    """simple docstring"""







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

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

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

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

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

    _snake_case      :					List[str]					=   []

   def        UpperCamelCase_					(   self							:     Optional[Any]			,						lowerCAmelCase							:     Optional[Any]			,						lowerCAmelCase							:     str)			->			Any:

    """simple docstring"""







    _snake_case      :					int					=   np.log(lowerCamelCase__)

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

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

    _snake_case      :					int					=   log_sigmas[low_idx]
    _snake_case      :					Tuple					=   log_sigmas[high_idx]

    # interpolate sigmas
    _snake_case      :					Union[str, Any]					=   (low - log_sigma) / (low - high)
    _snake_case      :					Dict					=   np.clip(lowerCamelCase__			,						0			,						1)

    # transform interpolation to time range
    _snake_case      :					List[Any]					=   (1 - w) * low_idx + w * high_idx
    _snake_case      :					Optional[Any]					=   t.reshape(sigma.shape)
    return t

   def        UpperCamelCase_					(   self							:     Any			,						lowerCAmelCase							:     Optional[Any])			->			torch.FloatTensor:

    """simple docstring"""







    _snake_case      :					Optional[Any]					=   in_sigmas[-1].item()
    _snake_case      :					List[str]					=   in_sigmas[0].item()

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

   def        UpperCamelCase_					(   self							:     Tuple			,						lowerCAmelCase							:     str			,						lowerCAmelCase							:     int			,						lowerCAmelCase							:     Optional[Any]			,						lowerCAmelCase							:     Union[str, Any] = 4			,						lowerCAmelCase							:     Tuple = True			,						)			->			Union[LMSDiscreteSchedulerOutput, Tuple]:

    """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):
     _snake_case      :					List[Any]					=   timestep.to(self.timesteps.device)
    _snake_case      :					int					=   (self.timesteps == timestep).nonzero().item()
    _snake_case      :					Tuple					=   self.sigmas[step_index]

    # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
    if self.config.prediction_type == "epsilon":
     _snake_case      :					List[str]					=   sample - sigma * model_output
    elif self.config.prediction_type == "v_prediction":
     # * c_out + input * c_skip
     _snake_case      :					List[Any]					=   model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
    elif self.config.prediction_type == "sample":
     _snake_case      :					Tuple					=   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
    _snake_case      :					Optional[Any]					=   (sample - pred_original_sample) / sigma
    self.derivatives.append(lowerCamelCase__)
    if len(self.derivatives) > order:
     self.derivatives.pop(0)

    # 3. Compute linear multistep coefficients
    _snake_case      :					Any					=   min(step_index + 1			,						lowerCamelCase__)
    _snake_case      :					int					=   [self.get_lms_coefficient(lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__) for curr_order in range(lowerCamelCase__)]

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

    """simple docstring"""







    _snake_case      :					Union[str, Any]					=   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
     _snake_case      :					List[str]					=   self.timesteps.to(original_samples.device			,						dtype=torch.floataa)
     _snake_case      :					Optional[Any]					=   timesteps.to(original_samples.device			,						dtype=torch.floataa)
    else:
     _snake_case      :					Tuple					=   self.timesteps.to(original_samples.device)
     _snake_case      :					Tuple					=   timesteps.to(original_samples.device)

    _snake_case      :					Optional[Any]					=   [(schedule_timesteps == t).nonzero().item() for t in timesteps]

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

    _snake_case      :					Tuple					=   original_samples + noise * sigma
    return noisy_samples


   def __len__(   self							:     str)			->			Optional[Any]:

    """simple docstring"""







    return self.config.num_train_timesteps





import os
import warnings
from pathlib import Path
from typing import Optional, Union

from .dynamic_module_utils import custom_object_save
from .tokenization_utils_base import PreTrainedTokenizerBase
from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging


a__      =		logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
a__      =		direct_transformers_import(Path(__file__).parent)


a__      =		{
    """AutoTokenizer""": """PreTrainedTokenizerBase""",
    """AutoFeatureExtractor""": """FeatureExtractionMixin""",
    """AutoImageProcessor""": """ImageProcessingMixin""",
}







class 						snake_case  (		SCREAMING_SNAKE_CASE_					):



   '''simple docstring'''






   snake_case_   :	List[str]       =						["""feature_extractor""", """tokenizer"""]
   # Names need to be attr_class for attr in attributes
   snake_case_   :	Optional[Any]       =						None
   snake_case_   :	List[str]       =						None
   snake_case_   :	List[str]       =						None

   def __init__(   self							:     int			,						*lowerCAmelCase							:     Any			,						**lowerCAmelCase							:     int)			->			Optional[Any]:

    """simple docstring"""







    for key in kwargs:
     if key not in self.attributes:
      raise TypeError(F'''Unexpected keyword argument {key}.''')
    for arg, attribute_name in zip(lowerCAmelCase			,						self.attributes):
     if attribute_name in kwargs:
      raise TypeError(F'''Got multiple values for argument {attribute_name}.''')
     else:
      _snake_case      :					List[str]					=   arg

    if len(lowerCAmelCase) != len(self.attributes):
     raise ValueError(
         F'''This processor requires {len(self.attributes)} arguments: {', '.join(self.attributes)}. Got '''
         F'''{len(lowerCAmelCase)} arguments instead.''')

    # Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
    for attribute_name, arg in kwargs.items():
     _snake_case      :					Optional[Any]					=   getattr(self			,						F'''{attribute_name}_class''')
     # Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
     _snake_case      :					List[str]					=   AUTO_TO_BASE_CLASS_MAPPING.get(lowerCAmelCase			,						lowerCAmelCase)
     if isinstance(lowerCAmelCase			,						lowerCAmelCase):
      _snake_case      :					Optional[Any]					=   tuple(getattr(lowerCAmelCase			,						lowerCAmelCase) for n in class_name if n is not None)
     else:
      _snake_case      :					List[Any]					=   getattr(lowerCAmelCase			,						lowerCAmelCase)

     if not isinstance(lowerCAmelCase			,						lowerCAmelCase):
      raise ValueError(
          F'''Received a {type(lowerCAmelCase).__name__} for argument {attribute_name}, but a {class_name} was expected.''')

     setattr(self			,						lowerCAmelCase			,						lowerCAmelCase)

   def __repr__(   self							:     int)			->			Dict:

    """simple docstring"""







    _snake_case      :					Union[str, Any]					=   [F'''- {name}: {repr(getattr(self			,						lowerCAmelCase))}''' for name in self.attributes]
    _snake_case      :					str					=   """\n""".join(lowerCAmelCase)
    return F'''{self.__class__.__name__}:\n{attributes_repr}'''

   def        UpperCamelCase_					(   self							:     Optional[int]			,						lowerCAmelCase							:     Optional[int]			,						lowerCAmelCase							:     bool = False			,						**lowerCAmelCase							:     Union[str, Any])			->			Dict:

    """simple docstring"""







    os.makedirs(lowerCAmelCase			,						exist_ok=lowerCAmelCase)

    if push_to_hub:
     _snake_case      :					str					=   kwargs.pop("""commit_message"""			,						lowerCAmelCase)
     _snake_case      :					Union[str, Any]					=   kwargs.pop("""repo_id"""			,						save_directory.split(os.path.sep)[-1])
     _snake_case      :					Tuple					=   self._create_repo(lowerCAmelCase			,						**lowerCAmelCase)
     _snake_case      :					Any					=   self._get_files_timestamps(lowerCAmelCase)
    # If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
    # loaded from the Hub.
    if self._auto_class is not None:
     _snake_case      :					Any					=   [getattr(self			,						lowerCAmelCase) for attribute_name in self.attributes]
     _snake_case      :					Any					=   [(a.init_kwargs if isinstance(lowerCAmelCase			,						lowerCAmelCase) else a) for a in attrs]
     custom_object_save(self			,						lowerCAmelCase			,						config=lowerCAmelCase)

    for attribute_name in self.attributes:
     _snake_case      :					Optional[int]					=   getattr(self			,						lowerCAmelCase)
     # Include the processor class in the attribute config so this processor can then be reloaded with the
     # `AutoProcessor` API.
     if hasattr(lowerCAmelCase			,						"""_set_processor_class"""):
      attribute._set_processor_class(self.__class__.__name__)
     attribute.save_pretrained(lowerCAmelCase)

    if self._auto_class is not None:
     # We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
     for attribute_name in self.attributes:
      _snake_case      :					List[str]					=   getattr(self			,						lowerCAmelCase)
      if isinstance(lowerCAmelCase			,						lowerCAmelCase):
       del attribute.init_kwargs["auto_map"]

    if push_to_hub:
     self._upload_modified_files(
         lowerCAmelCase			,						lowerCAmelCase			,						lowerCAmelCase			,						commit_message=lowerCAmelCase			,						token=kwargs.get("""use_auth_token""")			,						)

   @classmethod
   def        UpperCamelCase_					(   cls							:     List[Any]			,						lowerCAmelCase							:     Union[str, os.PathLike]			,						lowerCAmelCase							:     Optional[Union[str, os.PathLike]] = None			,						lowerCAmelCase							:     bool = False			,						lowerCAmelCase							:     bool = False			,						lowerCAmelCase							:     Optional[Union[str, bool]] = None			,						lowerCAmelCase							:     str = "main"			,						**lowerCAmelCase							:     Dict			,						)			->			Tuple:

    """simple docstring"""







    _snake_case      :					Dict					=   cache_dir
    _snake_case      :					Optional[Any]					=   force_download
    _snake_case      :					Dict					=   local_files_only
    _snake_case      :					Any					=   revision

    _snake_case      :					Dict					=   kwargs.pop("""use_auth_token"""			,						lowerCAmelCase)
    if use_auth_token is not None:
     warnings.warn(
         """The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers."""			,						lowerCAmelCase)
     if token is not None:
      raise ValueError(
          """`token` and `use_auth_token` are both specified. Please set only the argument `token`.""")
     _snake_case      :					str					=   use_auth_token

    if token is not None:
     # change to `token` in a follow-up PR
     _snake_case      :					Tuple					=   token

    _snake_case      :					Tuple					=   cls._get_arguments_from_pretrained(lowerCAmelCase			,						**lowerCAmelCase)
    return cls(*lowerCAmelCase)

   @classmethod
   def        UpperCamelCase_					(   cls							:     Optional[Any]			,						lowerCAmelCase							:     str="AutoProcessor")			->			int:

    """simple docstring"""







    if not isinstance(lowerCAmelCase			,						lowerCAmelCase):
     _snake_case      :					List[Any]					=   auto_class.__name__

    import transformers.models.auto as auto_module

    if not hasattr(lowerCAmelCase			,						lowerCAmelCase):
     raise ValueError(F'''{auto_class} is not a valid auto class.''')

    _snake_case      :					List[str]					=   auto_class

   @classmethod
   def        UpperCamelCase_					(   cls							:     List[Any]			,						lowerCAmelCase							:     Any			,						**lowerCAmelCase							:     Union[str, Any])			->			Optional[Any]:

    """simple docstring"""







    _snake_case      :					Optional[Any]					=   []
    for attribute_name in cls.attributes:
     _snake_case      :					Optional[Any]					=   getattr(cls			,						F'''{attribute_name}_class''')
     if isinstance(lowerCAmelCase			,						lowerCAmelCase):
      _snake_case      :					Tuple					=   tuple(getattr(lowerCAmelCase			,						lowerCAmelCase) if n is not None else None for n in class_name)
      _snake_case      :					str					=   kwargs.get("""use_fast"""			,						lowerCAmelCase)
      if use_fast and classes[1] is not None:
       _snake_case      :					Any					=   classes[1]
      else:
       _snake_case      :					Union[str, Any]					=   classes[0]
     else:
      _snake_case      :					Optional[Any]					=   getattr(lowerCAmelCase			,						lowerCAmelCase)

     args.append(attribute_class.from_pretrained(lowerCAmelCase			,						**lowerCAmelCase))
    return args


   @property
   def        UpperCamelCase_					(   self							:     Optional[int])			->			str:

    """simple docstring"""







    _snake_case      :					Tuple					=   getattr(self			,						self.attributes[0])
    return getattr(lowerCAmelCase			,						"""model_input_names"""			,						lowerCAmelCase)
a__      =		copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
  a__      =		ProcessorMixin.push_to_hub.__doc__.format(
      object="""processor""", object_class="""AutoProcessor""", object_files="""processor files"""
  )
