code: 
  
  
  # DISCLAIMER: This file is strongly influenced by https://github.com/yang-song/score_sde_pytorch
  
  from dataclasses import dataclass
  from typing import Optional, Tuple, Union
  
  import flax
  import jax.numpy as jnp
  from jax import random
  
  from ..configuration_utils import ConfigMixin, register_to_config
  from .scheduling_utils_flax import FlaxSchedulerMixin, FlaxSchedulerOutput, broadcast_to_shape_from_left
  
  @flax.struct.dataclass
  class 						snake_case			:
  			'''simple docstring'''
  
  			UpperCAmelCase       :       Optional[jnp.ndarray]       = None
  			UpperCAmelCase       :       Optional[jnp.ndarray]       = None
  			UpperCAmelCase       :       Optional[jnp.ndarray]       = None
  
  			@classmethod
  			def _lowercase (	cls     :     Dict					)					->					Optional[Any]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return cls()
  
  @dataclass
  class 						snake_case			(       lowerCAmelCase__	):
  			'''simple docstring'''
  
  			UpperCAmelCase       :       ScoreSdeVeSchedulerState
  			UpperCAmelCase       :       jnp.ndarray
  			UpperCAmelCase       :       Optional[jnp.ndarray]       = None
  
  class 						snake_case			(       lowerCAmelCase__	, lowerCAmelCase__	):
  			'''simple docstring'''
  
  			@property
  			def _lowercase (	self     :     List[str]					)					->					Dict:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return True
  
  			@register_to_config
  			def __init__(	self     :     str							,							lowerCAmelCase_     :     int = 2_000							,							lowerCAmelCase_     :     float = 0.15							,							lowerCAmelCase_     :     float = 0.01							,							lowerCAmelCase_     :     float = 1_348.0							,							lowerCAmelCase_     :     float = 1e-5							,							lowerCAmelCase_     :     int = 1							,							)					->					Optional[Any]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								pass
  
  			def _lowercase (	self     :     Dict					)					->					Optional[int]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       ScoreSdeVeSchedulerState.create()
  								return self.set_sigmas(
  								    lowerCAmelCase_							,							self.config.num_train_timesteps							,							self.config.sigma_min							,							self.config.sigma_max							,							self.config.sampling_eps							,							)
  
  			def _lowercase (	self     :     str							,							lowerCAmelCase_     :     ScoreSdeVeSchedulerState							,							lowerCAmelCase_     :     int							,							lowerCAmelCase_     :     Tuple = ()							,							lowerCAmelCase_     :     float = None					)					->					ScoreSdeVeSchedulerState:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       sampling_eps if sampling_eps is not None else self.config.sampling_eps
  
  								SCREAMING_SNAKE_CASE_	      =       jnp.linspace(1							,							lowerCAmelCase_							,							lowerCAmelCase_					)
  								return state.replace(timesteps=lowerCAmelCase_					)
  
  			def _lowercase (	self     :     Any							,							lowerCAmelCase_     :     ScoreSdeVeSchedulerState							,							lowerCAmelCase_     :     int							,							lowerCAmelCase_     :     float = None							,							lowerCAmelCase_     :     float = None							,							lowerCAmelCase_     :     float = None							,							)					->					ScoreSdeVeSchedulerState:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       sigma_min if sigma_min is not None else self.config.sigma_min
  								SCREAMING_SNAKE_CASE_	      =       sigma_max if sigma_max is not None else self.config.sigma_max
  								SCREAMING_SNAKE_CASE_	      =       sampling_eps if sampling_eps is not None else self.config.sampling_eps
  								if state.timesteps is None:
  													SCREAMING_SNAKE_CASE_	      =       self.set_timesteps(lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_					)
  
  								SCREAMING_SNAKE_CASE_	      =       jnp.exp(jnp.linspace(jnp.log(lowerCAmelCase_					)							,							jnp.log(lowerCAmelCase_					)							,							lowerCAmelCase_					)					)
  								SCREAMING_SNAKE_CASE_	      =       jnp.array([sigma_min * (sigma_max / sigma_min) ** t for t in state.timesteps]					)
  
  								return state.replace(discrete_sigmas=lowerCAmelCase_							,							sigmas=lowerCAmelCase_					)
  
  			def _lowercase (	self     :     List[str]							,							lowerCAmelCase_     :     Optional[int]							,							lowerCAmelCase_     :     int							,							lowerCAmelCase_     :     Dict					)					->					List[Any]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return jnp.where(timesteps == 0							,							jnp.zeros_like(lowerCAmelCase_					)							,							state.discrete_sigmas[timesteps - 1]					)
  
  			def _lowercase (	self     :     Tuple							,							lowerCAmelCase_     :     ScoreSdeVeSchedulerState							,							lowerCAmelCase_     :     jnp.ndarray							,							lowerCAmelCase_     :     int							,							lowerCAmelCase_     :     jnp.ndarray							,							lowerCAmelCase_     :     random.KeyArray							,							lowerCAmelCase_     :     bool = True							,							)					->					Union[FlaxSdeVeOutput, Tuple]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								if state.timesteps is None:
  													raise ValueError(
  													    '''`state.timesteps` is not set, you need to run \'set_timesteps\' after creating the scheduler'''					)
  
  								SCREAMING_SNAKE_CASE_	      =       timestep * jnp.ones(
  								    sample.shape[0]							,							)
  								SCREAMING_SNAKE_CASE_	      =       (timestep * (len(state.timesteps					) - 1)).long()
  
  								SCREAMING_SNAKE_CASE_	      =       state.discrete_sigmas[timesteps]
  								SCREAMING_SNAKE_CASE_	      =       self.get_adjacent_sigma(lowerCAmelCase_							,							lowerCAmelCase_							,							lowerCAmelCase_					)
  								SCREAMING_SNAKE_CASE_	      =       jnp.zeros_like(lowerCAmelCase_					)
  								SCREAMING_SNAKE_CASE_	      =       (sigma**2 - adjacent_sigma**2) ** 0.5
  
  								# equation 6 in the paper: the model_output modeled by the network is grad_x log pt(x)
  								# also equation 47 shows the analog from SDE models to ancestral sampling methods
  								SCREAMING_SNAKE_CASE_	      =       diffusion.flatten()
  								SCREAMING_SNAKE_CASE_	      =       broadcast_to_shape_from_left(lowerCAmelCase_							,							sample.shape					)
  								SCREAMING_SNAKE_CASE_	      =       drift - diffusion**2 * model_output
  
  								#  equation 6: sample noise for the diffusion term of
  								SCREAMING_SNAKE_CASE_	      =       random.split(lowerCAmelCase_							,							num=1					)
  								SCREAMING_SNAKE_CASE_	      =       random.normal(key=lowerCAmelCase_							,							shape=sample.shape					)
  								SCREAMING_SNAKE_CASE_	      =       sample - drift  # subtract because `dt` is a small negative timestep
  								# TODO is the variable diffusion the correct scaling term for the noise?
  								SCREAMING_SNAKE_CASE_	      =       prev_sample_mean + diffusion * noise  # add impact of diffusion field g
  
  								if not return_dict:
  													return (prev_sample, prev_sample_mean, state)
  
  								return FlaxSdeVeOutput(prev_sample=lowerCAmelCase_							,							prev_sample_mean=lowerCAmelCase_							,							state=lowerCAmelCase_					)
  
  			def _lowercase (	self     :     List[Any]							,							lowerCAmelCase_     :     ScoreSdeVeSchedulerState							,							lowerCAmelCase_     :     jnp.ndarray							,							lowerCAmelCase_     :     jnp.ndarray							,							lowerCAmelCase_     :     random.KeyArray							,							lowerCAmelCase_     :     bool = True							,							)					->					Union[FlaxSdeVeOutput, Tuple]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								if state.timesteps is None:
  													raise ValueError(
  													    '''`state.timesteps` is not set, you need to run \'set_timesteps\' after creating the scheduler'''					)
  
  								# For small batch sizes, the paper "suggest replacing norm(z) with sqrt(d), where d is the dim. of z"
  								# sample noise for correction
  								SCREAMING_SNAKE_CASE_	      =       random.split(lowerCAmelCase_							,							num=1					)
  								SCREAMING_SNAKE_CASE_	      =       random.normal(key=lowerCAmelCase_							,							shape=sample.shape					)
  
  								# compute step size from the model_output, the noise, and the snr
  								SCREAMING_SNAKE_CASE_	      =       jnp.linalg.norm(lowerCAmelCase_					)
  								SCREAMING_SNAKE_CASE_	      =       jnp.linalg.norm(lowerCAmelCase_					)
  								SCREAMING_SNAKE_CASE_	      =       (self.config.snr * noise_norm / grad_norm) ** 2 * 2
  								SCREAMING_SNAKE_CASE_	      =       step_size * jnp.ones(sample.shape[0]					)
  
  								# compute corrected sample: model_output term and noise term
  								SCREAMING_SNAKE_CASE_	      =       step_size.flatten()
  								SCREAMING_SNAKE_CASE_	      =       broadcast_to_shape_from_left(lowerCAmelCase_							,							sample.shape					)
  								SCREAMING_SNAKE_CASE_	      =       sample + step_size * model_output
  								SCREAMING_SNAKE_CASE_	      =       prev_sample_mean + ((step_size * 2) ** 0.5) * noise
  
  								if not return_dict:
  													return (prev_sample, state)
  
  								return FlaxSdeVeOutput(prev_sample=lowerCAmelCase_							,							state=lowerCAmelCase_					)
  
  
  
  
  			def __len__(	self     :     Union[str, Any]					)					->					Dict:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return self.config.num_train_timesteps
  
  
  
style_context: 
  
  
  import json
  import os
  from collections.abc import Mapping
  from typing import Dict, List, Optional, Union
  
  import numpy as np
  import torch
  
  from .imports import is_safetensors_available
  
  
  
  
  
  
  
  def 							UpperCAmelCase				(     UpperCAmelCase				,UpperCAmelCase				,UpperCAmelCase				,UpperCAmelCase=None      )->		Dict:
  
  
  
  
  
  
  
  					'''simple docstring'''
  
  
  
  					SCREAMING_SNAKE_CASE_	      =       None
  					# Check the string instead of the dtype to be compatible with versions of PyTorch that don't have bfloat16.
  					if str(weight.dtype      ) == "torch.bfloat16":
  										# Need to reinterpret the underlined data as int16 since NumPy does not handle bfloat16s.
  										SCREAMING_SNAKE_CASE_	      =       weight.view(torch.intaa      )
  										SCREAMING_SNAKE_CASE_	      =       '''bfloat16'''
  					SCREAMING_SNAKE_CASE_	      =       weight.cpu().numpy()
  					SCREAMING_SNAKE_CASE_	      =       os.path.join(UpperCAmelCase				,f'''{weight_name}.dat'''      )
  					if index is not None:
  										if dtype is None:
  															SCREAMING_SNAKE_CASE_	      =       str(array.dtype      )
  										SCREAMING_SNAKE_CASE_	      =       {'''dtype''': dtype, '''shape''': list(array.shape      )}
  					if array.ndim == 0:
  										SCREAMING_SNAKE_CASE_	      =       array[None]
  					SCREAMING_SNAKE_CASE_	      =       np.memmap(UpperCAmelCase				,dtype=array.dtype				,mode='''w+'''				,shape=array.shape      )
  					SCREAMING_SNAKE_CASE_	      =       array[:]
  					file_array.flush()
  					return index
  
  
  
  
  
  
  
  def 							UpperCAmelCase				(     UpperCAmelCase				,UpperCAmelCase      )->		List[str]:
  
  
  
  
  
  
  
  					'''simple docstring'''
  
  
  
  					SCREAMING_SNAKE_CASE_	      =       tuple(weight_info['''shape''']      )
  					if shape == ():
  										# NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
  										SCREAMING_SNAKE_CASE_	      =       (1,)
  
  					SCREAMING_SNAKE_CASE_	      =       weight_info['''dtype''']
  					if dtype == "bfloat16":
  										# NumPy does not support bfloat16 so this was saved as a int16
  										SCREAMING_SNAKE_CASE_	      =       '''int16'''
  
  					SCREAMING_SNAKE_CASE_	      =       np.memmap(UpperCAmelCase				,dtype=UpperCAmelCase				,shape=UpperCAmelCase				,mode='''r'''      )
  
  					if len(weight_info['''shape''']      ) == 0:
  										SCREAMING_SNAKE_CASE_	      =       weight[0]
  					SCREAMING_SNAKE_CASE_	      =       torch.tensor(UpperCAmelCase      )
  					if weight_info["dtype"] == "bfloat16":
  										SCREAMING_SNAKE_CASE_	      =       weight.view(torch.bfloataa      )
  
  					return weight
  
  
  
  
  
  
  
  def 							UpperCAmelCase				(     UpperCAmelCase				,UpperCAmelCase      )->		str:
  
  
  
  
  
  
  
  					'''simple docstring'''
  
  
  
  					if index is None or len(UpperCAmelCase      ) == 0:
  										# Nothing to save
  										return
  
  					SCREAMING_SNAKE_CASE_	      =       os.path.join(UpperCAmelCase				,'''index.json'''      )
  					if os.path.isfile(UpperCAmelCase      ):
  										with open(UpperCAmelCase				,'''r'''				,encoding='''utf-8'''      ) as f:
  															SCREAMING_SNAKE_CASE_	      =       json.load(UpperCAmelCase      )
  					else:
  										SCREAMING_SNAKE_CASE_	      =       {}
  					current_index.update(UpperCAmelCase      )
  
  					with open(UpperCAmelCase				,'''w'''				,encoding='''utf-8'''      ) as f:
  										json.dump(UpperCAmelCase				,UpperCAmelCase				,indent=2      )
  
  
  
  
  
  
  
  def 							UpperCAmelCase				(     UpperCAmelCase				,UpperCAmelCase      )->		Optional[Any]:
  
  
  
  
  
  
  
  					'''simple docstring'''
  
  
  
  					os.makedirs(UpperCAmelCase				,exist_ok=UpperCAmelCase      )
  					SCREAMING_SNAKE_CASE_	      =       {}
  					for name, parameter in state_dict.items():
  										SCREAMING_SNAKE_CASE_	      =       offload_weight(UpperCAmelCase				,UpperCAmelCase				,UpperCAmelCase				,index=UpperCAmelCase      )
  
  					# Update index
  					save_offload_index(UpperCAmelCase				,UpperCAmelCase      )
  
  class 						snake_case			(       lowerCAmelCase__	):
  			'''simple docstring'''
  
  			def __init__(	self     :     int							,							lowerCAmelCase_     :     Mapping							,							lowerCAmelCase_     :     str					)					->					Dict:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								SCREAMING_SNAKE_CASE_	      =       dataset
  								SCREAMING_SNAKE_CASE_	      =       prefix
  
  			def __getitem__(	self     :     Optional[int]							,							lowerCAmelCase_     :     Optional[int]					)					->					Union[str, Any]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return self.dataset[F'''{self.prefix}{key}''']
  
  			def __iter__(	self     :     Tuple					)					->					str:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return iter([key for key in self.dataset if key.startswith(self.prefix					)]					)
  
  
  
  
  			def __len__(	self     :     Dict					)					->					Optional[int]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return len(self.dataset					)
  
  class 						snake_case			(       lowerCAmelCase__	):
  			'''simple docstring'''
  
  			def __init__(	self     :     int							,							lowerCAmelCase_     :     Dict[str, torch.Tensor] = None							,							lowerCAmelCase_     :     Optional[Union[str, os.PathLike]] = None							,							lowerCAmelCase_     :     Mapping = None							,							lowerCAmelCase_     :     int=None							,							)					->					Tuple:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								if state_dict is None and save_folder is None:
  													raise ValueError('''Need either a `state_dict` or a `save_folder` containing offloaded weights.'''					)
  
  								SCREAMING_SNAKE_CASE_	      =       {} if state_dict is None else state_dict
  								SCREAMING_SNAKE_CASE_	      =       save_folder
  								if index is None and save_folder is not None:
  													with open(os.path.join(lowerCAmelCase_							,							'''index.json'''					)					) as f:
  																		SCREAMING_SNAKE_CASE_	      =       json.load(lowerCAmelCase_					)
  								SCREAMING_SNAKE_CASE_	      =       {} if index is None else index
  								SCREAMING_SNAKE_CASE_	      =       list(self.state_dict.keys()					)
  								self.all_keys.extend([key for key in self.index if key not in self.all_keys]					)
  								SCREAMING_SNAKE_CASE_	      =       device
  
  			def __getitem__(	self     :     str							,							lowerCAmelCase_     :     str					)					->					Union[str, Any]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								if key in self.state_dict:
  													return self.state_dict[key]
  								SCREAMING_SNAKE_CASE_	      =       self.index[key]
  								if weight_info.get('''safetensors_file'''					) is not None:
  													if not is_safetensors_available():
  																		raise ImportError('''These offloaded weights require the use of safetensors: `pip install safetensors`.'''					)
  
  													from safetensors import safe_open
  
  													SCREAMING_SNAKE_CASE_	      =       '''cpu''' if self.device is None else self.device
  													with safe_open(weight_info['''safetensors_file''']							,							framework='''pt'''							,							device=lowerCAmelCase_					) as f:
  																		SCREAMING_SNAKE_CASE_	      =       f.get_tensor(weight_info.get('''weight_name'''							,							lowerCAmelCase_					)					)
  
  													if "dtype" in weight_info:
  																		return tensor.to(getattr(lowerCAmelCase_							,							weight_info['''dtype''']					)					)
  													else:
  																		return tensor
  
  								SCREAMING_SNAKE_CASE_	      =       os.path.join(self.save_folder							,							F'''{key}.dat'''					)
  								return load_offloaded_weight(lowerCAmelCase_							,							lowerCAmelCase_					)
  
  			def __iter__(	self     :     str					)					->					Optional[int]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return iter(self.all_keys					)
  
  
  
  
  			def __len__(	self     :     Optional[Any]					)					->					Optional[Any]:
  
  
  
  
  								"""simple docstring"""
  
  
  
  
  
  
  								return len(self.all_keys					)
  
  
  
  
  
  
  
  def 							UpperCAmelCase				(     UpperCAmelCase				,UpperCAmelCase      )->		List[str]:
  
  
  
  
  
  
  
  					'''simple docstring'''
  
  
  
  					SCREAMING_SNAKE_CASE_	      =       {}
  					for module_name in submodule_names:
  										# We want to catch module_name parameter (module_name.xxx) or potentially module_name, but not any of the
  										# submodules that could being like module_name (transformers.h.1 and transformers.h.10 for instance)
  										result.update(
  										    {
  										        key: param
  										        for key, param in state_dict.items()
  										        if key == module_name or key.startswith(module_name + '''.'''      )
  										    }      )
  					return result
  
  
  