code: 
  
  
  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    A					(     A_				):
  							UpperCamelCase_     :							torch.FloatTensor
  							UpperCamelCase_     :							Optional[torch.FloatTensor] =None
  
  
  
  
  
  
  
  def  _lowerCamelCase(			lowercase__     ,				lowercase__=0.999     ,				lowercase__="cosine"     ,				)		->			List[Any]:
  
  		'''simple docstring'''
  
  
  
  
  
  
  
  		if alpha_transform_type == "cosine":
  
  				def alpha_bar_fn(lowercase__			):
  						return math.cos((t + 0.008) / 1.008 * math.pi / 2			) ** 2
  
  		elif alpha_transform_type == "exp":
  
  				def alpha_bar_fn(lowercase__			):
  						return math.exp(t * -12.0			)
  
  		else:
  				raise ValueError(F'Unsupported alpha_tranform_type: {alpha_transform_type}'			)
  
  		__lowercase=    []
  		for i in range(lowercase__			):
  				__lowercase=    i / num_diffusion_timesteps
  				__lowercase=    (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    A					(     A_      ,       A_				):
  							UpperCamelCase_     :							Any =[e.name for e in KarrasDiffusionSchedulers]
  							UpperCamelCase_     :							Union[str, Any] =1
  
  
  							@register_to_config
  							def __init__(self   ,       lowerCAmelCase = 1_0_0_0   ,       lowerCAmelCase = 0.00_01   ,       lowerCAmelCase = 0.02   ,       lowerCAmelCase = "linear"   ,       lowerCAmelCase = None   ,       lowerCAmelCase = False   ,       lowerCAmelCase = "epsilon"   ,       lowerCAmelCase = "linspace"   ,       lowerCAmelCase = 0   ,       ):
  									if trained_betas is not None:
  											__lowercase=    torch.tensor(lowerCAmelCase   ,       dtype=torch.floataa      )
  									elif beta_schedule == "linear":
  											__lowercase=    torch.linspace(lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       dtype=torch.floataa      )
  									elif beta_schedule == "scaled_linear":
  											# this schedule is very specific to the latent diffusion model.
  											__lowercase=    (
  											    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=    betas_for_alpha_bar(lowerCAmelCase      )
  									else:
  											raise NotImplementedError(f'{beta_schedule} does is not implemented for {self.__class__}'      )
  
  									__lowercase=    1.0 - self.betas
  									__lowercase=    torch.cumprod(self.alphas   ,       dim=0      )
  
  									__lowercase=    np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5      )
  									__lowercase=    np.concatenate([sigmas[::-1], [0.0]]      ).astype(np.floataa      )
  									__lowercase=    torch.from_numpy(lowerCAmelCase      )
  
  									# setable values
  									__lowercase=    None
  									__lowercase=    use_karras_sigmas
  									self.set_timesteps(lowerCAmelCase   ,       lowerCAmelCase      )
  									__lowercase=    []
  									__lowercase=    False
  
  
  							@property
  							def     _A (self      ):
  									# 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     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase      ):
  									if isinstance(lowerCAmelCase   ,       torch.Tensor      ):
  											__lowercase=    timestep.to(self.timesteps.device      )
  									__lowercase=    (self.timesteps == timestep).nonzero().item()
  									__lowercase=    self.sigmas[step_index]
  									__lowercase=    sample / ((sigma**2 + 1) ** 0.5)
  									__lowercase=    True
  									return sample
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase      ):
  
  									def lms_derivative(lowerCAmelCase      ):
  											__lowercase=    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=    integrate.quad(lowerCAmelCase   ,       self.sigmas[t]   ,       self.sigmas[t + 1]   ,       epsrel=1E-4      )[0]
  
  									return integrated_coeff
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase = None      ):
  									__lowercase=    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=    np.linspace(0   ,       self.config.num_train_timesteps - 1   ,       lowerCAmelCase   ,       dtype=lowerCAmelCase      )[
  											    ::-1
  											].copy()
  									elif self.config.timestep_spacing == "leading":
  											__lowercase=    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=    (np.arange(0   ,       lowerCAmelCase      ) * step_ratio).round()[::-1].copy().astype(lowerCAmelCase      )
  											timesteps += self.config.steps_offset
  									elif self.config.timestep_spacing == "trailing":
  											__lowercase=    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=    (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=    np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5      )
  									__lowercase=    np.log(lowerCAmelCase      )
  									__lowercase=    np.interp(lowerCAmelCase   ,       np.arange(0   ,       len(lowerCAmelCase      )      )   ,       lowerCAmelCase      )
  
  									if self.use_karras_sigmas:
  											__lowercase=    self._convert_to_karras(in_sigmas=lowerCAmelCase      )
  											__lowercase=    np.array([self._sigma_to_t(lowerCAmelCase   ,       lowerCAmelCase      ) for sigma in sigmas]      )
  
  									__lowercase=    np.concatenate([sigmas, [0.0]]      ).astype(np.floataa      )
  
  									__lowercase=    torch.from_numpy(lowerCAmelCase      ).to(device=lowerCAmelCase      )
  									if str(lowerCAmelCase      ).startswith('mps'      ):
  											# mps does not support float64
  											__lowercase=    torch.from_numpy(lowerCAmelCase      ).to(lowerCAmelCase   ,       dtype=torch.floataa      )
  									else:
  											__lowercase=    torch.from_numpy(lowerCAmelCase      ).to(device=lowerCAmelCase      )
  
  									__lowercase=    []
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase      ):
  									# get log sigma
  									__lowercase=    np.log(lowerCAmelCase      )
  
  									# get distribution
  									__lowercase=    log_sigma - log_sigmas[:, np.newaxis]
  
  									# get sigmas range
  									__lowercase=    np.cumsum((dists >= 0)   ,       axis=0      ).argmax(axis=0      ).clip(max=log_sigmas.shape[0] - 2      )
  									__lowercase=    low_idx + 1
  
  									__lowercase=    log_sigmas[low_idx]
  									__lowercase=    log_sigmas[high_idx]
  
  									# interpolate sigmas
  									__lowercase=    (low - log_sigma) / (low - high)
  									__lowercase=    np.clip(lowerCAmelCase   ,       0   ,       1      )
  
  									# transform interpolation to time range
  									__lowercase=    (1 - w) * low_idx + w * high_idx
  									__lowercase=    t.reshape(sigma.shape      )
  									return t
  
  
  							def     _A (self   ,       lowerCAmelCase      ):
  									__lowercase=    in_sigmas[-1].item()
  									__lowercase=    in_sigmas[0].item()
  
  									__lowercase=    7.0  # 7.0 is the value used in the paper
  									__lowercase=    np.linspace(0   ,       1   ,       self.num_inference_steps      )
  									__lowercase=    sigma_min ** (1 / rho)
  									__lowercase=    sigma_max ** (1 / rho)
  									__lowercase=    (max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
  									return sigmas
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase = 4   ,       lowerCAmelCase = True   ,       ):
  									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=    timestep.to(self.timesteps.device      )
  									__lowercase=    (self.timesteps == timestep).nonzero().item()
  									__lowercase=    self.sigmas[step_index]
  
  									# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
  									if self.config.prediction_type == "epsilon":
  											__lowercase=    sample - sigma * model_output
  									elif self.config.prediction_type == "v_prediction":
  											# * c_out + input * c_skip
  											__lowercase=    model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
  									elif self.config.prediction_type == "sample":
  											__lowercase=    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=    (sample - pred_original_sample) / sigma
  									self.derivatives.append(lowerCAmelCase      )
  									if len(self.derivatives      ) > order:
  											self.derivatives.pop(0      )
  
  									# 3. Compute linear multistep coefficients
  									__lowercase=    min(step_index + 1   ,       lowerCAmelCase      )
  									__lowercase=    [self.get_lms_coefficient(lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase      ) for curr_order in range(lowerCAmelCase      )]
  
  									# 4. Compute previous sample based on the derivatives path
  									__lowercase=    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     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       ):
  									# Make sure sigmas and timesteps have the same device and dtype as original_samples
  									__lowercase=    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=    self.timesteps.to(original_samples.device   ,       dtype=torch.floataa      )
  											__lowercase=    timesteps.to(original_samples.device   ,       dtype=torch.floataa      )
  									else:
  											__lowercase=    self.timesteps.to(original_samples.device      )
  											__lowercase=    timesteps.to(original_samples.device      )
  
  									__lowercase=    [(schedule_timesteps == t).nonzero().item() for t in timesteps]
  
  									__lowercase=    sigmas[step_indices].flatten()
  									while len(sigma.shape      ) < len(original_samples.shape      ):
  											__lowercase=    sigma.unsqueeze(-1      )
  
  									__lowercase=    original_samples + noise * sigma
  									return noisy_samples
  
  
  
  
  
  
  
  							def __len__(self      ):
  									return self.config.num_train_timesteps
  
  
  
  
  
  
  
style_context: 
  
  
  from typing import Dict, List, Optional, Tuple
  
  from ...tokenization_utils import AddedToken, PreTrainedTokenizer
  from ...utils import logging
  
  
  lowerCAmelCase					 =	logging.get_logger(__name__)
  
  
  
  
  
  class    A					(     A_				):
  							UpperCamelCase_     :							Any =['''input_ids''', '''attention_mask''']
  
  
  							def __init__(self   ,       lowerCAmelCase="[PAD]"   ,       lowerCAmelCase="[BOS]"   ,       lowerCAmelCase="[EOS]"   ,       lowerCAmelCase="[MASK]"   ,       lowerCAmelCase="[CLS]"   ,       lowerCAmelCase="[SEP]"   ,       lowerCAmelCase=2_0_4_8   ,       **lowerCAmelCase   ,       ):
  									__lowercase=    AddedToken(lowerCAmelCase   ,       lstrip=lowerCAmelCase   ,       rstrip=lowerCAmelCase      ) if isinstance(lowerCAmelCase   ,       lowerCAmelCase      ) else pad_token
  									__lowercase=    AddedToken(lowerCAmelCase   ,       lstrip=lowerCAmelCase   ,       rstrip=lowerCAmelCase      ) if isinstance(lowerCAmelCase   ,       lowerCAmelCase      ) else bos_token
  									__lowercase=    AddedToken(lowerCAmelCase   ,       lstrip=lowerCAmelCase   ,       rstrip=lowerCAmelCase      ) if isinstance(lowerCAmelCase   ,       lowerCAmelCase      ) else eos_token
  									__lowercase=    AddedToken(lowerCAmelCase   ,       lstrip=lowerCAmelCase   ,       rstrip=lowerCAmelCase      ) if isinstance(lowerCAmelCase   ,       lowerCAmelCase      ) else mask_token
  									__lowercase=    AddedToken(lowerCAmelCase   ,       lstrip=lowerCAmelCase   ,       rstrip=lowerCAmelCase      ) if isinstance(lowerCAmelCase   ,       lowerCAmelCase      ) else cls_token
  									__lowercase=    AddedToken(lowerCAmelCase   ,       lstrip=lowerCAmelCase   ,       rstrip=lowerCAmelCase      ) if isinstance(lowerCAmelCase   ,       lowerCAmelCase      ) else sep_token
  
  									super().__init__(
  									    pad_token=lowerCAmelCase   ,       bos_token=lowerCAmelCase   ,       eos_token=lowerCAmelCase   ,       mask_token=lowerCAmelCase   ,       cls_token=lowerCAmelCase   ,       sep_token=lowerCAmelCase   ,       model_max_length=lowerCAmelCase   ,       **lowerCAmelCase   ,       )
  
  									__lowercase=    2**8  # utf is 8 bits
  
  									# define special tokens dict
  									__lowercase=    {
  									    self.pad_token: 0,
  									    self.bos_token: 1,
  									    self.eos_token: 2,
  									    self.mask_token: 3,
  									    self.cls_token: 4,
  									    self.sep_token: 5,
  									}
  									__lowercase=    len(self.special_tokens_encoder      )
  									__lowercase=    {v: k for k, v in self.special_tokens_encoder.items()}
  
  
  							def     _A (self      ):
  									__lowercase=    self.special_tokens_encoder.copy()
  									vocab.update(self.added_tokens_encoder      )
  									for i in range(self._utf_vocab_size      ):
  											__lowercase=    chr(lowerCAmelCase      )
  											__lowercase=    i + len(self.special_tokens_encoder      )
  									return vocab
  
  
  							@property
  							def     _A (self      ):
  									return self._utf_vocab_size + self._num_special_tokens
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase = None   ,       lowerCAmelCase = False      ):
  									if already_has_special_tokens:
  											return super().get_special_tokens_mask(
  											    token_ids_a=lowerCAmelCase   ,       token_ids_a=lowerCAmelCase   ,       already_has_special_tokens=lowerCAmelCase      )
  
  									# normal case: some special tokens
  									if token_ids_a is None:
  											return [1] + [0] * len(lowerCAmelCase      ) + [1]
  									return [1] + ([0] * len(lowerCAmelCase      )) + [1] + ([0] * len(lowerCAmelCase      )) + [1]
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase = None      ):
  									if token_ids_a is None:
  											return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
  									else:
  											return [self.cls_token_id] + token_ids_a + [self.sep_token_id] + token_ids_a + [self.sep_token_id]
  
  
  							def     _A (self   ,       lowerCAmelCase      ):
  									__lowercase=    [chr(lowerCAmelCase      ) for i in text.encode('utf-8'      )]
  									return tokens
  
  
  							def     _A (self   ,       lowerCAmelCase      ):
  									if token in self.special_tokens_encoder:
  											__lowercase=    self.special_tokens_encoder[token]
  									elif token in self.added_tokens_encoder:
  											__lowercase=    self.added_tokens_encoder[token]
  									elif len(lowerCAmelCase      ) != 1:
  											__lowercase=    self.unk_token_id
  									else:
  											__lowercase=    ord(lowerCAmelCase      ) + self._num_special_tokens
  									return token_id
  
  
  							def     _A (self   ,       lowerCAmelCase      ):
  									if index in self.special_tokens_decoder:
  											__lowercase=    self.special_tokens_decoder[index]
  									elif index in self.added_tokens_decoder:
  											__lowercase=    self.added_tokens_decoder[index]
  									else:
  											__lowercase=    chr(index - self._num_special_tokens      )
  									return token
  
  
  							def     _A (self   ,       lowerCAmelCase      ):
  									__lowercase=    B''
  									for token in tokens:
  											if token in self.special_tokens_decoder:
  													__lowercase=    self.special_tokens_decoder[token].encode('utf-8'      )
  											elif token in self.added_tokens_decoder:
  													__lowercase=    self.special_tokens_decoder[token].encode('utf-8'      )
  											elif token in self.special_tokens_encoder:
  													__lowercase=    token.encode('utf-8'      )
  											elif token in self.added_tokens_encoder:
  													__lowercase=    token.encode('utf-8'      )
  											else:
  													__lowercase=    bytes([ord(lowerCAmelCase      )]      )
  											bstring += tok_string
  									__lowercase=    bstring.decode('utf-8'   ,       errors='replace'      )
  									return string
  
  
  
  
  
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase = None      ):
  									return ()
  
  
  
  
  
  
  