code: 
  
  
  from dataclasses import dataclass
  from typing import Optional, Tuple, Union
  
  import flax
  import jax.numpy as jnp
  from scipy import integrate
  
  from ..configuration_utils import ConfigMixin, register_to_config
  from .scheduling_utils_flax import (
      CommonSchedulerState,
      FlaxKarrasDiffusionSchedulers,
      FlaxSchedulerMixin,
      FlaxSchedulerOutput,
      broadcast_to_shape_from_left,
  )
  
  
  
  
  
  @flax.struct.dataclass
  class    A					:
  							UpperCamelCase_     :							CommonSchedulerState
  
  							# setable values
  							UpperCamelCase_     :							jnp.ndarray
  							UpperCamelCase_     :							jnp.ndarray
  							UpperCamelCase_     :							jnp.ndarray
  							UpperCamelCase_     :							Optional[int] =None
  
  							# running values
  							UpperCamelCase_     :							Optional[jnp.ndarray] =None
  
  
  							@classmethod
  							def     _A (cls   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase      ):
  									return cls(common=lowerCAmelCase   ,       init_noise_sigma=lowerCAmelCase   ,       timesteps=lowerCAmelCase   ,       sigmas=lowerCAmelCase      )
  
  
  
  
  
  @dataclass
  class    A					(     A_				):
  							UpperCamelCase_     :							LMSDiscreteSchedulerState
  
  
  
  
  
  class    A					(     A_      ,       A_				):
  							UpperCamelCase_     :							List[str] =[e.name for e in FlaxKarrasDiffusionSchedulers]
  
  							UpperCamelCase_     :							jnp.dtype
  
  
  							@property
  							def     _A (self      ):
  									return True
  
  
  							@register_to_config
  							def __init__(self   ,       lowerCAmelCase = 1_0_0_0   ,       lowerCAmelCase = 0.00_01   ,       lowerCAmelCase = 0.02   ,       lowerCAmelCase = "linear"   ,       lowerCAmelCase = None   ,       lowerCAmelCase = "epsilon"   ,       lowerCAmelCase = jnp.floataa   ,       ):
  									__lowercase=    dtype
  
  
  							def     _A (self   ,       lowerCAmelCase = None      ):
  									if common is None:
  											__lowercase=    CommonSchedulerState.create(self      )
  
  									__lowercase=    jnp.arange(0   ,       self.config.num_train_timesteps      ).round()[::-1]
  									__lowercase=    ((1 - common.alphas_cumprod) / common.alphas_cumprod) ** 0.5
  
  									# standard deviation of the initial noise distribution
  									__lowercase=    sigmas.max()
  
  									return LMSDiscreteSchedulerState.create(
  									    common=lowerCAmelCase   ,       init_noise_sigma=lowerCAmelCase   ,       timesteps=lowerCAmelCase   ,       sigmas=lowerCAmelCase   ,       )
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase      ):
  									((__lowercase),							)=    jnp.where(state.timesteps == timestep   ,       size=1      )
  									__lowercase=    step_index[0]
  
  									__lowercase=    state.sigmas[step_index]
  									__lowercase=    sample / ((sigma**2 + 1) ** 0.5)
  									return sample
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase      ):
  
  									def lms_derivative(lowerCAmelCase      ):
  											__lowercase=    1.0
  											for k in range(lowerCAmelCase      ):
  													if current_order == k:
  															continue
  													prod *= (tau - state.sigmas[t - k]) / (state.sigmas[t - current_order] - state.sigmas[t - k])
  											return prod
  
  									__lowercase=    integrate.quad(lowerCAmelCase   ,       state.sigmas[t]   ,       state.sigmas[t + 1]   ,       epsrel=1E-4      )[0]
  
  									return integrated_coeff
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase = ()      ):
  									__lowercase=    jnp.linspace(self.config.num_train_timesteps - 1   ,       0   ,       lowerCAmelCase   ,       dtype=self.dtype      )
  
  									__lowercase=    jnp.floor(lowerCAmelCase      ).astype(jnp.intaa      )
  									__lowercase=    jnp.ceil(lowerCAmelCase      ).astype(jnp.intaa      )
  
  									__lowercase=    jnp.mod(lowerCAmelCase   ,       1.0      )
  
  									__lowercase=    ((1 - state.common.alphas_cumprod) / state.common.alphas_cumprod) ** 0.5
  									__lowercase=    (1 - frac) * sigmas[low_idx] + frac * sigmas[high_idx]
  									__lowercase=    jnp.concatenate([sigmas, jnp.array([0.0]   ,       dtype=self.dtype      )]      )
  
  									__lowercase=    timesteps.astype(jnp.intaa      )
  
  									# initial running values
  									__lowercase=    jnp.zeros((0,) + shape   ,       dtype=self.dtype      )
  
  									return state.replace(
  									    timesteps=lowerCAmelCase   ,       sigmas=lowerCAmelCase   ,       num_inference_steps=lowerCAmelCase   ,       derivatives=lowerCAmelCase   ,       )
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase = 4   ,       lowerCAmelCase = True   ,       ):
  									if state.num_inference_steps is None:
  											raise ValueError(
  											    'Number of inference steps is \'None\', you need to run \'set_timesteps\' after creating the scheduler'      )
  
  									__lowercase=    state.sigmas[timestep]
  
  									# 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))
  									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
  									__lowercase=    state.replace(derivatives=jnp.append(state.derivatives   ,       lowerCAmelCase      )      )
  									if len(state.derivatives      ) > order:
  											__lowercase=    state.replace(derivatives=jnp.delete(state.derivatives   ,       0      )      )
  
  									# 3. Compute linear multistep coefficients
  									__lowercase=    min(timestep + 1   ,       lowerCAmelCase      )
  									__lowercase=    [self.get_lms_coefficient(lowerCAmelCase   ,       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(state.derivatives      )      )      )
  
  									if not return_dict:
  											return (prev_sample, state)
  
  									return FlaxLMSSchedulerOutput(prev_sample=lowerCAmelCase   ,       state=lowerCAmelCase      )
  
  
  							def     _A (self   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       lowerCAmelCase   ,       ):
  									__lowercase=    state.sigmas[timesteps].flatten()
  									__lowercase=    broadcast_to_shape_from_left(lowerCAmelCase   ,       noise.shape      )
  
  									__lowercase=    original_samples + noise * sigma
  
  									return noisy_samples
  
  
  
  
  
  
  
  							def __len__(self      ):
  									return self.config.num_train_timesteps
  
  
  
  
  
  
  
style_context: 
  
  
  from dataclasses import dataclass
  from enum import Enum
  from typing import List, Optional, Union
  
  import numpy as np
  import PIL
  from PIL import Image
  
  from ...utils import BaseOutput, is_torch_available, is_transformers_available
  
  
  
  
  
  @dataclass
  class    A					(     A_				):
  							UpperCamelCase_     :							Optional[int] ={
  							    '''sld_warmup_steps''': 15,
  							    '''sld_guidance_scale''': 20,
  							    '''sld_threshold''': 0.0,
  							    '''sld_momentum_scale''': 0.0,
  							    '''sld_mom_beta''': 0.0,
  							}
  							UpperCamelCase_     :							Tuple ={
  							    '''sld_warmup_steps''': 10,
  							    '''sld_guidance_scale''': 1_000,
  							    '''sld_threshold''': 0.01,
  							    '''sld_momentum_scale''': 0.3,
  							    '''sld_mom_beta''': 0.4,
  							}
  							UpperCamelCase_     :							Union[str, Any] ={
  							    '''sld_warmup_steps''': 7,
  							    '''sld_guidance_scale''': 2_000,
  							    '''sld_threshold''': 0.025,
  							    '''sld_momentum_scale''': 0.5,
  							    '''sld_mom_beta''': 0.7,
  							}
  							UpperCamelCase_     :							List[Any] ={
  							    '''sld_warmup_steps''': 0,
  							    '''sld_guidance_scale''': 5_000,
  							    '''sld_threshold''': 1.0,
  							    '''sld_momentum_scale''': 0.5,
  							    '''sld_mom_beta''': 0.7,
  							}
  
  
  
  
  
  @dataclass
  class    A					(     A_				):
  							UpperCamelCase_     :							Union[List[PIL.Image.Image], np.ndarray]
  							UpperCamelCase_     :							Optional[List[bool]]
  							UpperCamelCase_     :							Optional[Union[List[PIL.Image.Image], np.ndarray]]
  							UpperCamelCase_     :							Optional[str]
  
  
  if is_transformers_available() and is_torch_available():
  	from .pipeline_stable_diffusion_safe import StableDiffusionPipelineSafe
  	from .safety_checker import SafeStableDiffusionSafetyChecker
  
  
  
  
  
  
  