code: 
  
  
  
  
  
  '''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      UpperCAmelCase			( UpperCamelCase__       ):
   __lowercase     	=       42
   __lowercase     	=       None
  
  
  
  
  
  def 		UpperCamelCase       (							_lowerCamelCase      :      int		, _lowerCamelCase      :      Tuple=0.9_9_9		, _lowerCamelCase      :      Optional[Any]="cosine"		, ):
      if alpha_transform_type == "cosine":
  
          def alpha_bar_fn(_lowerCamelCase      :      List[str] ):
              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(_lowerCamelCase      :      Tuple ):
              return math.exp(t * -1_2.0 )
  
      else:
          raise ValueError(F"Unsupported alpha_tranform_type: {alpha_transform_type}" )
  
      A__       =    []
      for i in range(_lowerCamelCase ):
          A__       =    i / num_diffusion_timesteps
          A__       =    (i + 1) / num_diffusion_timesteps
          betas.append(min(1 - alpha_bar_fn(_lowerCamelCase ) / alpha_bar_fn(_lowerCamelCase )		, _lowerCamelCase ) )
      return torch.tensor(_lowerCamelCase		, dtype=torch.floataa )
  
  
  
  
  
  
  class      UpperCAmelCase			( UpperCamelCase__    , UpperCamelCase__       ):
   __lowercase     	=       [e.name for e in KarrasDiffusionSchedulers]
   __lowercase     	=       1
   @register_to_config
   def __init__(       self   :Union[str, Any]		, lowercase_   :int = 10_00		, lowercase_   :float = 0.0_0_0_1		, lowercase_   :float = 0.0_2		, lowercase_   :str = "linear"		, lowercase_   :Optional[Union[np.ndarray, List[float]]] = None		, lowercase_   :Optional[bool] = False		, lowercase_   :str = "epsilon"		, lowercase_   :str = "linspace"		, lowercase_   :int = 0		, )->							Tuple:
       if trained_betas is not None:
           A__       =    torch.tensor(lowercase_		, dtype=torch.floataa							)
       elif beta_schedule == "linear":
           A__       =    torch.linspace(lowercase_		, lowercase_		, lowercase_		, dtype=torch.floataa							)
       elif beta_schedule == "scaled_linear":
           # this schedule is very specific to the latent diffusion model.
           A__       =    (
               torch.linspace(beta_start**0.5		, beta_end**0.5		, lowercase_		, dtype=torch.floataa							) ** 2
           )
       elif beta_schedule == "squaredcos_cap_v2":
           # Glide cosine schedule
           A__       =    betas_for_alpha_bar(lowercase_							)
       else:
           raise NotImplementedError(F"{beta_schedule} does is not implemented for {self.__class__}"							)
  
       A__       =    1.0 - self.betas
       A__       =    torch.cumprod(self.alphas		, dim=0							)
  
       A__       =    np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5							)
       A__       =    np.concatenate([sigmas[::-1], [0.0]]							).astype(np.floataa							)
       A__       =    torch.from_numpy(lowercase_							)
  
       # setable values
       A__       =    None
       A__       =    use_karras_sigmas
       self.set_timesteps(lowercase_		, lowercase_							)
       A__       =    []
       A__       =    False
   @property
   def   UpperCAmelCase_							(       self   :List[Any]							)->							Optional[int]:
       # 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[Any]		, lowercase_   :torch.FloatTensor		, lowercase_   :Union[float, torch.FloatTensor]							)->							torch.FloatTensor:
       if isinstance(lowercase_		, torch.Tensor							):
           A__       =    timestep.to(self.timesteps.device							)
       A__       =    (self.timesteps == timestep).nonzero().item()
       A__       =    self.sigmas[step_index]
       A__       =    sample / ((sigma**2 + 1) ** 0.5)
       A__       =    True
       return sample
   def   UpperCAmelCase_							(       self   :List[Any]		, lowercase_   :int		, lowercase_   :Dict		, lowercase_   :Optional[int]							)->							Dict:
  
       def lms_derivative(lowercase_   :Optional[int]							):
           A__       =    1.0
           for k in range(lowercase_							):
               if current_order == k:
                   continue
               prod *= (tau - self.sigmas[t - k]) / (self.sigmas[t - current_order] - self.sigmas[t - k])
           return prod
  
       A__       =    integrate.quad(lowercase_		, self.sigmas[t]		, self.sigmas[t + 1]		, epsrel=1E-4							)[0]
  
       return integrated_coeff
   def   UpperCAmelCase_							(       self   :Optional[Any]		, lowercase_   :int		, lowercase_   :Union[str, torch.device] = None							)->							Dict:
       A__       =    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":
           A__       =    np.linspace(0		, self.config.num_train_timesteps - 1		, lowercase_		, dtype=lowercase_							)[
               ::-1
           ].copy()
       elif self.config.timestep_spacing == "leading":
           A__       =    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
           A__       =    (np.arange(0		, lowercase_							) * step_ratio).round()[::-1].copy().astype(lowercase_							)
           timesteps += self.config.steps_offset
       elif self.config.timestep_spacing == "trailing":
           A__       =    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
           A__       =    (np.arange(self.config.num_train_timesteps		, 0		, -step_ratio							)).round().copy().astype(lowercase_							)
           timesteps -= 1
       else:
           raise ValueError(
               F"{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'."							)
  
       A__       =    np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5							)
       A__       =    np.log(lowercase_							)
       A__       =    np.interp(lowercase_		, np.arange(0		, len(lowercase_							)							)		, lowercase_							)
  
       if self.use_karras_sigmas:
           A__       =    self._convert_to_karras(in_sigmas=lowercase_							)
           A__       =    np.array([self._sigma_to_t(lowercase_		, lowercase_							) for sigma in sigmas]							)
  
       A__       =    np.concatenate([sigmas, [0.0]]							).astype(np.floataa							)
  
       A__       =    torch.from_numpy(lowercase_							).to(device=lowercase_							)
       if str(lowercase_							).startswith("mps"							):
           # mps does not support float64
           A__       =    torch.from_numpy(lowercase_							).to(lowercase_		, dtype=torch.floataa							)
       else:
           A__       =    torch.from_numpy(lowercase_							).to(device=lowercase_							)
  
       A__       =    []
   def   UpperCAmelCase_							(       self   :List[Any]		, lowercase_   :int		, lowercase_   :List[Any]							)->							Dict:
       # get log sigma
       A__       =    np.log(lowercase_							)
  
       # get distribution
       A__       =    log_sigma - log_sigmas[:, np.newaxis]
  
       # get sigmas range
       A__       =    np.cumsum((dists >= 0)		, axis=0							).argmax(axis=0							).clip(max=log_sigmas.shape[0] - 2							)
       A__       =    low_idx + 1
  
       A__       =    log_sigmas[low_idx]
       A__       =    log_sigmas[high_idx]
  
       # interpolate sigmas
       A__       =    (low - log_sigma) / (low - high)
       A__       =    np.clip(lowercase_		, 0		, 1							)
  
       # transform interpolation to time range
       A__       =    (1 - w) * low_idx + w * high_idx
       A__       =    t.reshape(sigma.shape							)
       return t
   def   UpperCAmelCase_							(       self   :Optional[Any]		, lowercase_   :torch.FloatTensor							)->							torch.FloatTensor:
       A__       =    in_sigmas[-1].item()
       A__       =    in_sigmas[0].item()
  
       A__       =    7.0  # 7.0 is the value used in the paper
       A__       =    np.linspace(0		, 1		, self.num_inference_steps							)
       A__       =    sigma_min ** (1 / rho)
       A__       =    sigma_max ** (1 / rho)
       A__       =    (max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
       return sigmas
   def   UpperCAmelCase_							(       self   :List[Any]		, lowercase_   :torch.FloatTensor		, lowercase_   :Union[float, torch.FloatTensor]		, lowercase_   :torch.FloatTensor		, lowercase_   :int = 4		, lowercase_   :bool = True		, )->							Union[LMSDiscreteSchedulerOutput, Tuple]:
       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(lowercase_		, torch.Tensor							):
           A__       =    timestep.to(self.timesteps.device							)
       A__       =    (self.timesteps == timestep).nonzero().item()
       A__       =    self.sigmas[step_index]
  
       # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
       if self.config.prediction_type == "epsilon":
           A__       =    sample - sigma * model_output
       elif self.config.prediction_type == "v_prediction":
           # * c_out + input * c_skip
           A__       =    model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
       elif self.config.prediction_type == "sample":
           A__       =    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
       A__       =    (sample - pred_original_sample) / sigma
       self.derivatives.append(lowercase_							)
       if len(self.derivatives							) > order:
           self.derivatives.pop(0							)
  
       # 3. Compute linear multistep coefficients
       A__       =    min(step_index + 1		, lowercase_							)
       A__       =    [self.get_lms_coefficient(lowercase_		, lowercase_		, lowercase_							) for curr_order in range(lowercase_							)]
  
       # 4. Compute previous sample based on the derivatives path
       A__       =    sample + sum(
           coeff * derivative for coeff, derivative in zip(lowercase_		, reversed(self.derivatives							)							)							)
  
       if not return_dict:
           return (prev_sample,)
  
       return LMSDiscreteSchedulerOutput(prev_sample=lowercase_		, pred_original_sample=lowercase_							)
   def   UpperCAmelCase_							(       self   :Tuple		, lowercase_   :torch.FloatTensor		, lowercase_   :torch.FloatTensor		, lowercase_   :torch.FloatTensor		, )->							torch.FloatTensor:
       # Make sure sigmas and timesteps have the same device and dtype as original_samples
       A__       =    self.sigmas.to(device=original_samples.device		, dtype=original_samples.dtype							)
       if original_samples.device.type == "mps" and torch.is_floating_point(lowercase_							):
           # mps does not support float64
           A__       =    self.timesteps.to(original_samples.device		, dtype=torch.floataa							)
           A__       =    timesteps.to(original_samples.device		, dtype=torch.floataa							)
       else:
           A__       =    self.timesteps.to(original_samples.device							)
           A__       =    timesteps.to(original_samples.device							)
  
       A__       =    [(schedule_timesteps == t).nonzero().item() for t in timesteps]
  
       A__       =    sigmas[step_indices].flatten()
       while len(sigma.shape							) < len(original_samples.shape							):
           A__       =    sigma.unsqueeze(-1							)
  
       A__       =    original_samples + noise * sigma
       return noisy_samples
  
  
  
   def __len__(       self   :Optional[int]							)->							Union[str, Any]:
       return self.config.num_train_timesteps
  
  
  
  
style_context: 
  
  
  
  
  
  '''simple docstring'''
  
  import bisect
  import copy
  from collections import defaultdict
  
  import numpy as np
  from torch.utils.data import BatchSampler, Sampler
  
  from utils import logger
  
  
  
  
  
  def 		UpperCamelCase       (							_lowerCamelCase      :      str		, _lowerCamelCase      :      Optional[Any] ):
      A__       =    copy.deepcopy(_lowerCamelCase )
      A__       =    sorted(_lowerCamelCase )
      A__       =    [bisect.bisect_right(_lowerCamelCase		, _lowerCamelCase ) for y in x]
      return quantized
  
  
  
  
  
  def 		UpperCamelCase       (							_lowerCamelCase      :      Optional[int]		, _lowerCamelCase      :      Tuple=0 ):
      A__       =    np.arange(start=3		, stop=_lowerCamelCase		, step=4 ).tolist() if k > 0 else [10]
      A__       =    _quantize(_lowerCamelCase		, _lowerCamelCase )
      # count number of elements per group
      A__       =    np.unique(_lowerCamelCase		, return_counts=_lowerCamelCase )[1]
      A__       =    [0] + bins + [np.inf]
      logger.info("Using {} as bins for aspect lengths quantization".format(_lowerCamelCase ) )
      logger.info("Count of instances per bin: {}".format(_lowerCamelCase ) )
      return groups
  
  
  
  
  
  
  class      UpperCAmelCase			( UpperCamelCase__       ):
   def __init__(       self   :Optional[Any]		, lowercase_   :List[Any]		, lowercase_   :Optional[int]		, lowercase_   :str							)->							Dict:
       if not isinstance(lowercase_		, lowercase_							):
           raise ValueError(
               "sampler should be an instance of torch.utils.data.Sampler, but got sampler={}".format(lowercase_							)							)
       A__       =    sampler
       A__       =    group_ids
       A__       =    batch_size
   def __iter__(       self   :int							)->							Tuple:
       A__       =    defaultdict(lowercase_							)
       A__       =    defaultdict(lowercase_							)
  
       A__       =    0
       for idx in self.sampler:
           A__       =    self.group_ids[idx]
           buffer_per_group[group_id].append(lowercase_							)
           samples_per_group[group_id].append(lowercase_							)
           if len(buffer_per_group[group_id]							) == self.batch_size:
               yield buffer_per_group[group_id]  # TODO
               num_batches += 1
               del buffer_per_group[group_id]
           assert len(buffer_per_group[group_id]							) < self.batch_size
  
       # now we have run out of elements that satisfy
       # the group criteria, let's return the remaining
       # elements so that the size of the sampler is
       # deterministic
       A__       =    len(self							)
       A__       =    expected_num_batches - num_batches
       if num_remaining > 0:
           # for the remaining batches, group the batches by similar lengths
           A__       =    []
           for group_id, idxs in sorted(buffer_per_group.items()		, key=lambda lowercase_							: x[0]							):
               batch_idx.extend(lowercase_							)
               if len(lowercase_							) >= self.batch_size:
                   yield batch_idx[: self.batch_size]
                   A__       =    batch_idx[self.batch_size :]
                   num_remaining -= 1
           if len(lowercase_							) > 0:
               yield batch_idx
               num_remaining -= 1
       assert num_remaining == 0
  
  
  
   def __len__(       self   :int							)->							Tuple:
       return (len(self.sampler							) + self.batch_size - 1) // self.batch_size
  
  
  
  