


"""simple docstring"""






# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion

from dataclasses import dataclass
from typing import Optional, Tuple, Union

import flax
import jax.numpy as jnp

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils_flax import (
    CommonSchedulerState,
    FlaxKarrasDiffusionSchedulers,
    FlaxSchedulerMixin,
    FlaxSchedulerOutput,
    add_noise_common,
    get_velocity_common,
)


@flax.struct.dataclass
class 			SCREAMING_SNAKE_CASE_  :




							'''simple docstring'''





							__magic_name__					:     CommonSchedulerState
							__magic_name__					:     jnp.ndarray

							# setable values
							__magic_name__					:     jnp.ndarray
							__magic_name__					:     jnp.ndarray
							__magic_name__					:     Optional[int]				  =  None




							@classmethod
							def        UpperCAmelCase (						cls	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					)   ->	Tuple:







								'''simple docstring'''




								return cls(
								    common=lowerCamelCase__	,					final_alpha_cumprod=lowerCamelCase__	,					init_noise_sigma=lowerCamelCase__	,					timesteps=lowerCamelCase__	,					)


@dataclass
class 			SCREAMING_SNAKE_CASE_  (   _lowercase):




							'''simple docstring'''





							__magic_name__					:     DDIMSchedulerState


class 			SCREAMING_SNAKE_CASE_  (   _lowercase	,      _lowercase):




							'''simple docstring'''





							__magic_name__					:     Optional[int]				  =  [e.name for e in FlaxKarrasDiffusionSchedulers]

							__magic_name__					:     jnp.dtype




							@property
							def        UpperCAmelCase (						self)   ->	Tuple:







								'''simple docstring'''




								return True




							@register_to_config
							def __init__(						self	,					lowerCamelCase__ = 1_000	,					lowerCamelCase__ = 0.00_01	,					lowerCamelCase__ = 0.02	,					lowerCamelCase__ = "linear"	,					lowerCamelCase__ = None	,					lowerCamelCase__ = True	,					lowerCamelCase__ = 0	,					lowerCamelCase__ = "epsilon"	,					lowerCamelCase__ = jnp.floataa	,					)   ->	int:







								'''simple docstring'''




								snake_case__  :						Optional[int]		=      dtype




							def        UpperCAmelCase (						self	,					lowerCamelCase__ = None)   ->	DDIMSchedulerState:







								'''simple docstring'''




								if common is None:
									snake_case__  :						int		=      CommonSchedulerState.create(self)

								# At every step in ddim, we are looking into the previous alphas_cumprod
								# For the final step, there is no previous alphas_cumprod because we are already at 0
								# `set_alpha_to_one` decides whether we set this parameter simply to one or
								# whether we use the final alpha of the "non-previous" one.
								snake_case__  :						Tuple		=      (
								    jnp.array(1.0	,					dtype=self.dtype) if self.config.set_alpha_to_one else common.alphas_cumprod[0]
								)

								# standard deviation of the initial noise distribution
								snake_case__  :						List[str]		=      jnp.array(1.0	,					dtype=self.dtype)

								snake_case__  :						Dict		=      jnp.arange(0	,					self.config.num_train_timesteps).round()[::-1]

								return DDIMSchedulerState.create(
								    common=lowerCamelCase__	,					final_alpha_cumprod=lowerCamelCase__	,					init_noise_sigma=lowerCamelCase__	,					timesteps=lowerCamelCase__	,					)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__ = None)   ->	jnp.ndarray:







								'''simple docstring'''




								return sample




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__ = ())   ->	DDIMSchedulerState:







								'''simple docstring'''




								snake_case__  :						List[str]		=      self.config.num_train_timesteps // num_inference_steps
								# creates integer timesteps by multiplying by ratio
								# rounding to avoid issues when num_inference_step is power of 3
								snake_case__  :						str		=      (jnp.arange(0	,					lowerCamelCase__) * step_ratio).round()[::-1] + self.config.steps_offset

								return state.replace(
								    num_inference_steps=lowerCamelCase__	,					timesteps=lowerCamelCase__	,					)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)   ->	Optional[int]:







								'''simple docstring'''




								snake_case__  :						Dict		=      state.common.alphas_cumprod[timestep]
								snake_case__  :						Any		=      jnp.where(
								    prev_timestep >= 0	,					state.common.alphas_cumprod[prev_timestep]	,					state.final_alpha_cumprod)
								snake_case__  :						Any		=      1 - alpha_prod_t
								snake_case__  :						str		=      1 - alpha_prod_t_prev

								snake_case__  :						Union[str, Any]		=      (beta_prod_t_prev / beta_prod_t) * (1 - alpha_prod_t / alpha_prod_t_prev)

								return variance




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__ = 0.0	,					lowerCamelCase__ = True	,					)   ->	Union[FlaxDDIMSchedulerOutput, Tuple]:







								'''simple docstring'''




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

								# See formulas (12) and (16) of DDIM paper https://arxiv.org/pdf/2010.02502.pdf
								# Ideally, read DDIM paper in-detail understanding

								# Notation (<variable name> -> <name in paper>
								# - pred_noise_t -> e_theta(x_t, t)
								# - pred_original_sample -> f_theta(x_t, t) or x_0
								# - std_dev_t -> sigma_t
								# - eta -> η
								# - pred_sample_direction -> "direction pointing to x_t"
								# - pred_prev_sample -> "x_t-1"

								# 1. get previous step value (=t-1)
								snake_case__  :						Optional[Any]		=      timestep - self.config.num_train_timesteps // state.num_inference_steps

								snake_case__  :						List[str]		=      state.common.alphas_cumprod
								snake_case__  :						Any		=      state.final_alpha_cumprod

								# 2. compute alphas, betas
								snake_case__  :						Optional[int]		=      alphas_cumprod[timestep]
								snake_case__  :						Dict		=      jnp.where(prev_timestep >= 0	,					alphas_cumprod[prev_timestep]	,					lowerCamelCase__)

								snake_case__  :						List[str]		=      1 - alpha_prod_t

								# 3. compute predicted original sample from predicted noise also called
								# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
								if self.config.prediction_type == "epsilon":
									snake_case__  :						List[Any]		=      (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
									snake_case__  :						Union[str, Any]		=      model_output
								elif self.config.prediction_type == "sample":
									snake_case__  :						Dict		=      model_output
									snake_case__  :						Tuple		=      (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
								elif self.config.prediction_type == "v_prediction":
									snake_case__  :						str		=      (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
									snake_case__  :						Optional[int]		=      (alpha_prod_t**0.5) * model_output + (beta_prod_t**0.5) * sample
								else:
									raise ValueError(
									    f"""prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample`, or"""
									    " `v_prediction`")

								# 4. compute variance: "sigma_t(η)" -> see formula (16)
								# σ_t = sqrt((1 − α_t−1)/(1 − α_t)) * sqrt(1 − α_t/α_t−1)
								snake_case__  :						List[str]		=      self._get_variance(lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)
								snake_case__  :						Optional[int]		=      eta * variance ** 0.5

								# 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
								snake_case__  :						List[Any]		=      (1 - alpha_prod_t_prev - std_dev_t**2) ** 0.5 * pred_epsilon

								# 6. compute x_t without "random noise" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
								snake_case__  :						str		=      alpha_prod_t_prev ** 0.5 * pred_original_sample + pred_sample_direction

								if not return_dict:
									return (prev_sample, state)

								return FlaxDDIMSchedulerOutput(prev_sample=lowerCamelCase__	,					state=lowerCamelCase__)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					)   ->	jnp.ndarray:







								'''simple docstring'''




								return add_noise_common(state.common	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)




							def        UpperCAmelCase (						self	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__	,					)   ->	jnp.ndarray:







								'''simple docstring'''




								return get_velocity_common(state.common	,					lowerCamelCase__	,					lowerCamelCase__	,					lowerCamelCase__)







							def __len__(						self)   ->	List[Any]:







								'''simple docstring'''




								return self.config.num_train_timesteps







import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
  import torch_xla.core.xla_model as xm  # noqa: F401

  _UpperCAmelCase    			=					True
except ImportError:
  _UpperCAmelCase    			=					False


# Cache this result has it's a C FFI call which can be pretty time-consuming
_UpperCAmelCase    			=					torch.distributed.is_available()


def 		_lowerCamelCase       (   _a  ):
  """simple docstring"""




  _lowerCamelCase      =						importlib.util.find_spec(_a  ) is not None
  if package_exists:
    try:
      _lowerCamelCase      =						importlib.metadata.metadata(_a  )
      return True
    except importlib.metadata.PackageNotFoundError:
      return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _torch_distributed_available


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  try:
    pass
  except ImportError:
    print(
        '''Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not'''
        ''' detected. If you see \"ValueError: Invalid backend: \'ccl\'\" error, please install Intel(R) oneCCL'''
        ''' Bindings for PyTorch*.'''  )
  return (
      importlib.util.find_spec('''torch_ccl'''  ) is not None
      or importlib.util.find_spec('''oneccl_bindings_for_pytorch'''  ) is not None
  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return importlib.metadata.version('''oneccl_bind_pt'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''transformer_engine'''  )


@lru_cache
def 		_lowerCamelCase       (   _a=True  ):
  """simple docstring"""




  if torch.cuda.is_available():
    return False
  if _tpu_available and check_device:
    try:
      # Will raise a RuntimeError if no XLA configuration is found
      _lowerCamelCase      =						xm.xla_device()
      return True
    except RuntimeError:
      return False
  return _tpu_available


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''deepspeed'''  )


def 		_lowerCamelCase       (   _a=False  ):
  """simple docstring"""




  if is_tpu_available():
    return not ignore_tpu
  if torch.cuda.is_available():
    return torch.cuda.is_bfaa_supported()
  if is_npu_available():
    return False
  return True


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  _lowerCamelCase      =						_is_package_available('''bitsandbytes'''  )
  if package_exists:
    _lowerCamelCase      =						version.parse(importlib.metadata.version('''bitsandbytes'''  )  )
    return compare_versions(_a			,   '''>='''			,   '''0.39.0'''  )
  return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  _lowerCamelCase      =						_is_package_available('''bitsandbytes'''  )
  if package_exists:
    _lowerCamelCase      =						version.parse(importlib.metadata.version('''bitsandbytes'''  )  )
    return compare_versions(_a			,   '''>='''			,   '''0.37.2'''  )
  return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''bitsandbytes'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  if strtobool(os.environ.get('''ACCELERATE_USE_MEGATRON_LM'''			,   '''False'''  )  ) == 1:
    _lowerCamelCase      =						importlib.util.find_spec('''megatron'''  ) is not None
    if package_exists:
      try:
        _lowerCamelCase      =						parse(importlib.metadata.version('''megatron-lm'''  )  )
        return compare_versions(_a			,   '''>='''			,   '''2.2.0'''  )
      except Exception as e:
        warnings.warn(F'''Parse Megatron version failed. Exception:{e}'''  )
        return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''safetensors'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''transformers'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''datasets'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''aim'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''tensorboard'''  ) or _is_package_available('''tensorboardX'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''wandb'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''comet_ml'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''boto3'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  if _is_package_available('''rich'''  ):
    if "ACCELERATE_DISABLE_RICH" in os.environ:
      warnings.warn(
          '''`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`.'''  )
      return not parse_flag_from_env('''ACCELERATE_DISABLE_RICH'''			,   _a  )
    return parse_flag_from_env('''ACCELERATE_ENABLE_RICH'''			,   _a  )
  return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''sagemaker'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''tqdm'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''mlflow'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return is_torch_version('''>='''			,   '''1.12'''  ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  def get_major_and_minor_from_version(_a  ):
    return str(version.parse(_a  ).major  ) + "." + str(version.parse(_a  ).minor  )

  _lowerCamelCase      =						importlib.metadata.version('''torch'''  )
  if importlib.util.find_spec('''intel_extension_for_pytorch'''  ) is None:
    return False
  _lowerCamelCase      =						'''N/A'''
  try:
    _lowerCamelCase      =						importlib.metadata.version('''intel_extension_for_pytorch'''  )
  except importlib.metadata.PackageNotFoundError:
    return False
  _lowerCamelCase      =						get_major_and_minor_from_version(_torch_version  )
  _lowerCamelCase      =						get_major_and_minor_from_version(_ipex_version  )
  if torch_major_and_minor != ipex_major_and_minor:
    warnings.warn(
        F'''Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'''
        F''' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'''  )
    return False
  return True


@lru_cache
def 		_lowerCamelCase       (   _a=False  ):
  """simple docstring"""




  if importlib.util.find_spec('''torch'''  ) is None or importlib.util.find_spec('''torch_npu'''  ) is None:
    return False

  import torch
  import torch_npu  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no NPU is found
      _lowerCamelCase      =						torch.npu.device_count()
      return torch.npu.is_available()
    except RuntimeError:
      return False
  return hasattr(_a			,   '''npu'''  ) and torch.npu.is_available()






@lru_cache
def 		_lowerCamelCase       (   _a=False  ):
  """simple docstring"""




  if not parse_flag_from_env('''ACCELERATE_USE_XPU'''			,   default=_a  ):
    return False
  if is_ipex_available():
    import torch

    if is_torch_version('''<='''			,   '''1.12'''  ):
      return False
  else:
    return False

  import intel_extension_for_pytorch  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no XPU  is found
      _lowerCamelCase      =						torch.xpu.device_count()
      return torch.xpu.is_available()
    except RuntimeError:
      return False
  return hasattr(_a			,   '''xpu'''  ) and torch.xpu.is_available()




