
'''simple docstring'''





import math
from collections import defaultdict
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin, SchedulerOutput







def snake_case_     (					lowercase__   ,       lowercase__=0.999   ,       lowercase__="cosine"   ,       ):
  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}"""		)

  UpperCAmelCase__      :					str				       =							[]
  for i in range(lowercase__		):
    UpperCAmelCase__      :					Any				       =							i / num_diffusion_timesteps
    UpperCAmelCase__      :					Union[str, Any]				       =							(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 							UpperCAmelCase_							(    A					,				A				):

 '''simple docstring'''
 lowercase_					:    Dict				=							[e.name for e in KarrasDiffusionSchedulers]
 lowercase_					:    List[str]				=							2







 @register_to_config
 def __init__(					self		:  Optional[Any]			,				snake_case__		:  int = 10_00			,				snake_case__		:  float = 0.00085			,				snake_case__		:  float = 0.012			,				snake_case__		:  str = "linear"			,				snake_case__		:  Optional[Union[np.ndarray, List[float]]] = None			,				snake_case__		:  str = "epsilon"			,				snake_case__		:  str = "linspace"			,				snake_case__		:  int = 0			,				):



   '''simple docstring'''




   if trained_betas is not None:
     UpperCAmelCase__      :					Dict				       =							torch.tensor(snake_case__			,				dtype=torch.floataa )
   elif beta_schedule == "linear":
     UpperCAmelCase__      :					Dict				       =							torch.linspace(snake_case__			,				snake_case__			,				snake_case__			,				dtype=torch.floataa )
   elif beta_schedule == "scaled_linear":
     # this schedule is very specific to the latent diffusion model.
     UpperCAmelCase__      :					Union[str, Any]				       =							(
         torch.linspace(beta_start**0.5			,				beta_end**0.5			,				snake_case__			,				dtype=torch.floataa ) ** 2
     )
   elif beta_schedule == "squaredcos_cap_v2":
     # Glide cosine schedule
     UpperCAmelCase__      :					Optional[int]				       =							betas_for_alpha_bar(snake_case__ )
   else:
     raise NotImplementedError(F"""{beta_schedule} does is not implemented for {self.__class__}""" )

   UpperCAmelCase__      :					Optional[Any]				       =							1.0 - self.betas
   UpperCAmelCase__      :					Union[str, Any]				       =							torch.cumprod(self.alphas			,				dim=0 )

   #  set all values
   self.set_timesteps(snake_case__			,				snake_case__			,				snake_case__ )







 def 							UpperCamelCase  (					self		:  Dict			,				snake_case__		:  Optional[int]			,				snake_case__		:  Tuple=None ):



   '''simple docstring'''




   if schedule_timesteps is None:
     UpperCAmelCase__      :					List[str]				       =							self.timesteps

   UpperCAmelCase__      :					List[Any]				       =							(schedule_timesteps == timestep).nonzero()

   # The sigma index that is taken for the **very** first `step`
   # is always the second index (or the last index if there is only 1)
   # This way we can ensure we don't accidentally skip a sigma in
   # case we start in the middle of the denoising schedule (e.g. for image-to-image)
   if len(self._index_counter ) == 0:
     UpperCAmelCase__      :					str				       =							1 if len(snake_case__ ) > 1 else 0
   else:
     UpperCAmelCase__      :					List[str]				       =							timestep.cpu().item() if torch.is_tensor(snake_case__ ) else timestep
     UpperCAmelCase__      :					Tuple				       =							self._index_counter[timestep_int]

   return indices[pos].item()







 @property
 def 							UpperCamelCase  (					self		:  Tuple ):



   '''simple docstring'''




   if self.config.timestep_spacing in ["linspace", "trailing"]:
     return self.sigmas.max()

   return (self.sigmas.max() ** 2 + 1) ** 0.5







 def 							UpperCamelCase  (					self		:  Union[str, Any]			,				snake_case__		:  torch.FloatTensor			,				snake_case__		:  Union[float, torch.FloatTensor]			,				):



   '''simple docstring'''




   UpperCAmelCase__      :					Dict				       =							self.index_for_timestep(snake_case__ )

   if self.state_in_first_order:
     UpperCAmelCase__      :					Dict				       =							self.sigmas[step_index]
   else:
     UpperCAmelCase__      :					Optional[Any]				       =							self.sigmas_interpol[step_index]

   UpperCAmelCase__      :					Dict				       =							sample / ((sigma**2 + 1) ** 0.5)
   return sample







 def 							UpperCamelCase  (					self		:  Optional[int]			,				snake_case__		:  int			,				snake_case__		:  Union[str, torch.device] = None			,				snake_case__		:  Optional[int] = None			,				):



   '''simple docstring'''




   UpperCAmelCase__      :					str				       =							num_inference_steps

   UpperCAmelCase__      :					Optional[int]				       =							num_train_timesteps or self.config.num_train_timesteps

   # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
   if self.config.timestep_spacing == "linspace":
     UpperCAmelCase__      :					Tuple				       =							np.linspace(0			,				num_train_timesteps - 1			,				snake_case__			,				dtype=snake_case__ )[::-1].copy()
   elif self.config.timestep_spacing == "leading":
     UpperCAmelCase__      :					Dict				       =							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
     UpperCAmelCase__      :					Tuple				       =							(np.arange(0			,				snake_case__ ) * step_ratio).round()[::-1].copy().astype(snake_case__ )
     timesteps += self.config.steps_offset
   elif self.config.timestep_spacing == "trailing":
     UpperCAmelCase__      :					List[Any]				       =							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
     UpperCAmelCase__      :					int				       =							(np.arange(snake_case__			,				0			,				-step_ratio )).round().copy().astype(snake_case__ )
     timesteps -= 1
   else:
     raise ValueError(
         F"""{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'.""" )

   UpperCAmelCase__      :					Optional[Any]				       =							np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5 )
   UpperCAmelCase__      :					Optional[Any]				       =							torch.from_numpy(np.log(snake_case__ ) ).to(snake_case__ )

   UpperCAmelCase__      :					List[Any]				       =							np.interp(snake_case__			,				np.arange(0			,				len(snake_case__ ) )			,				snake_case__ )
   UpperCAmelCase__      :					Dict				       =							np.concatenate([sigmas, [0.0]] ).astype(np.floataa )
   UpperCAmelCase__      :					Optional[int]				       =							torch.from_numpy(snake_case__ ).to(device=snake_case__ )

   # interpolate sigmas
   UpperCAmelCase__      :					Optional[Any]				       =							sigmas.log().lerp(sigmas.roll(1 ).log()			,				0.5 ).exp()

   UpperCAmelCase__      :					str				       =							torch.cat([sigmas[:1], sigmas[1:].repeat_interleave(2 ), sigmas[-1:]] )
   UpperCAmelCase__      :					Tuple				       =							torch.cat(
       [sigmas_interpol[:1], sigmas_interpol[1:].repeat_interleave(2 ), sigmas_interpol[-1:]] )

   if str(snake_case__ ).startswith("mps" ):
     # mps does not support float64
     UpperCAmelCase__      :					Any				       =							torch.from_numpy(snake_case__ ).to(snake_case__			,				dtype=torch.floataa )
   else:
     UpperCAmelCase__      :					str				       =							torch.from_numpy(snake_case__ ).to(snake_case__ )

   # interpolate timesteps
   UpperCAmelCase__      :					Tuple				       =							self.sigma_to_t(snake_case__ ).to(snake_case__			,				dtype=timesteps.dtype )
   UpperCAmelCase__      :					Any				       =							torch.stack((timesteps_interpol[1:-1, None], timesteps[1:, None])			,				dim=-1 ).flatten()

   UpperCAmelCase__      :					List[Any]				       =							torch.cat([timesteps[:1], interleaved_timesteps] )

   UpperCAmelCase__      :					Optional[Any]				       =							None

   # for exp beta schedules, such as the one for `pipeline_shap_e.py`
   # we need an index counter
   UpperCAmelCase__      :					List[str]				       =							defaultdict(snake_case__ )







 def 							UpperCamelCase  (					self		:  Optional[int]			,				snake_case__		:  Dict ):



   '''simple docstring'''




   UpperCAmelCase__      :					List[str]				       =							sigma.log()

   # get distribution
   UpperCAmelCase__      :					List[Any]				       =							log_sigma - self.log_sigmas[:, None]

   # get sigmas range
   UpperCAmelCase__      :					str				       =							dists.ge(0 ).cumsum(dim=0 ).argmax(dim=0 ).clamp(max=self.log_sigmas.shape[0] - 2 )
   UpperCAmelCase__      :					List[Any]				       =							low_idx + 1

   UpperCAmelCase__      :					Optional[Any]				       =							self.log_sigmas[low_idx]
   UpperCAmelCase__      :					Optional[int]				       =							self.log_sigmas[high_idx]

   # interpolate sigmas
   UpperCAmelCase__      :					Tuple				       =							(low - log_sigma) / (low - high)
   UpperCAmelCase__      :					Tuple				       =							w.clamp(0			,				1 )

   # transform interpolation to time range
   UpperCAmelCase__      :					str				       =							(1 - w) * low_idx + w * high_idx
   UpperCAmelCase__      :					Optional[int]				       =							t.view(sigma.shape )
   return t







 @property
 def 							UpperCamelCase  (					self		:  int ):



   '''simple docstring'''




   return self.sample is None







 def 							UpperCamelCase  (					self		:  Any			,				snake_case__		:  Union[torch.FloatTensor, np.ndarray]			,				snake_case__		:  Union[float, torch.FloatTensor]			,				snake_case__		:  Union[torch.FloatTensor, np.ndarray]			,				snake_case__		:  bool = True			,				):



   '''simple docstring'''




   UpperCAmelCase__      :					int				       =							self.index_for_timestep(snake_case__ )

   # advance index counter by 1
   UpperCAmelCase__      :					int				       =							timestep.cpu().item() if torch.is_tensor(snake_case__ ) else timestep
   self._index_counter[timestep_int] += 1

   if self.state_in_first_order:
     UpperCAmelCase__      :					List[Any]				       =							self.sigmas[step_index]
     UpperCAmelCase__      :					Tuple				       =							self.sigmas_interpol[step_index + 1]
     UpperCAmelCase__      :					List[Any]				       =							self.sigmas[step_index + 1]
   else:
     # 2nd order / KDPM2's method
     UpperCAmelCase__      :					Optional[int]				       =							self.sigmas[step_index - 1]
     UpperCAmelCase__      :					List[str]				       =							self.sigmas_interpol[step_index]
     UpperCAmelCase__      :					str				       =							self.sigmas[step_index]

   # currently only gamma=0 is supported. This usually works best anyways.
   # We can support gamma in the future but then need to scale the timestep before
   # passing it to the model which requires a change in API
   UpperCAmelCase__      :					Any				       =							0
   UpperCAmelCase__      :					List[str]				       =							sigma * (gamma + 1)  # Note: sigma_hat == sigma for now

   # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
   if self.config.prediction_type == "epsilon":
     UpperCAmelCase__      :					Optional[int]				       =							sigma_hat if self.state_in_first_order else sigma_interpol
     UpperCAmelCase__      :					List[Any]				       =							sample - sigma_input * model_output
   elif self.config.prediction_type == "v_prediction":
     UpperCAmelCase__      :					Optional[Any]				       =							sigma_hat if self.state_in_first_order else sigma_interpol
     UpperCAmelCase__      :					List[str]				       =							model_output * (-sigma_input / (sigma_input**2 + 1) ** 0.5) + (
         sample / (sigma_input**2 + 1)
     )
   elif self.config.prediction_type == "sample":
     raise NotImplementedError("prediction_type not implemented yet: sample" )
   else:
     raise ValueError(
         F"""prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`""" )

   if self.state_in_first_order:
     # 2. Convert to an ODE derivative for 1st order
     UpperCAmelCase__      :					Tuple				       =							(sample - pred_original_sample) / sigma_hat
     # 3. delta timestep
     UpperCAmelCase__      :					Optional[int]				       =							sigma_interpol - sigma_hat

     # store for 2nd order step
     UpperCAmelCase__      :					Any				       =							sample
   else:
     # DPM-Solver-2
     # 2. Convert to an ODE derivative for 2nd order
     UpperCAmelCase__      :					List[str]				       =							(sample - pred_original_sample) / sigma_interpol

     # 3. delta timestep
     UpperCAmelCase__      :					Optional[int]				       =							sigma_next - sigma_hat

     UpperCAmelCase__      :					Dict				       =							self.sample
     UpperCAmelCase__      :					str				       =							None

   UpperCAmelCase__      :					List[Any]				       =							sample + derivative * dt

   if not return_dict:
     return (prev_sample,)

   return SchedulerOutput(prev_sample=snake_case__ )







 def 							UpperCamelCase  (					self		:  str			,				snake_case__		:  torch.FloatTensor			,				snake_case__		:  torch.FloatTensor			,				snake_case__		:  torch.FloatTensor			,				):



   '''simple docstring'''




   UpperCAmelCase__      :					List[str]				       =							self.sigmas.to(device=original_samples.device			,				dtype=original_samples.dtype )
   if original_samples.device.type == "mps" and torch.is_floating_point(snake_case__ ):
     # mps does not support float64
     UpperCAmelCase__      :					Optional[Any]				       =							self.timesteps.to(original_samples.device			,				dtype=torch.floataa )
     UpperCAmelCase__      :					Tuple				       =							timesteps.to(original_samples.device			,				dtype=torch.floataa )
   else:
     UpperCAmelCase__      :					str				       =							self.timesteps.to(original_samples.device )
     UpperCAmelCase__      :					Union[str, Any]				       =							timesteps.to(original_samples.device )

   UpperCAmelCase__      :					str				       =							[self.index_for_timestep(snake_case__			,				snake_case__ ) for t in timesteps]

   UpperCAmelCase__      :					str				       =							sigmas[step_indices].flatten()
   while len(sigma.shape ) < len(original_samples.shape ):
     UpperCAmelCase__      :					List[Any]				       =							sigma.unsqueeze(-1 )

   UpperCAmelCase__      :					Dict				       =							original_samples + noise * sigma
   return noisy_samples







 def __len__(					self		:  List[str] ):



   '''simple docstring'''




   return self.config.num_train_timesteps





'''simple docstring'''





import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / """utils"""))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


SCREAMING_SNAKE_CASE     =       get_tests_dir("""fixtures""")
SCREAMING_SNAKE_CASE     =       get_tests_dir("""fixtures/dummy_feature_extractor_config.json""")
SCREAMING_SNAKE_CASE     =       get_tests_dir("""fixtures/dummy-config.json""")






class 							UpperCAmelCase_							(    unittest.TestCase				):

 '''simple docstring'''







 def 							UpperCamelCase  (					self		:  int ):



   '''simple docstring'''




   UpperCAmelCase__      :					Any				       =							0







 def 							UpperCamelCase  (					self		:  Union[str, Any] ):



   '''simple docstring'''




   UpperCAmelCase__      :					Optional[int]				       =							AutoFeatureExtractor.from_pretrained("facebook/wav2vec2-base-960h" )
   self.assertIsInstance(snake_case__			,				snake_case__ )







 def 							UpperCamelCase  (					self		:  int ):



   '''simple docstring'''




   UpperCAmelCase__      :					Optional[int]				       =							AutoFeatureExtractor.from_pretrained(snake_case__ )
   self.assertIsInstance(snake_case__			,				snake_case__ )







 def 							UpperCamelCase  (					self		:  List[Any] ):



   '''simple docstring'''




   with tempfile.TemporaryDirectory() as tmpdirname:
     UpperCAmelCase__      :					Optional[Any]				       =							WavaVecaConfig()

     # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
     UpperCAmelCase__      :					Any				       =							AutoFeatureExtractor.from_pretrained(snake_case__ ).to_dict()

     config_dict.pop("feature_extractor_type" )
     UpperCAmelCase__      :					Union[str, Any]				       =							WavaVecaFeatureExtractor(**snake_case__ )

     # save in new folder
     model_config.save_pretrained(snake_case__ )
     config.save_pretrained(snake_case__ )

     UpperCAmelCase__      :					Tuple				       =							AutoFeatureExtractor.from_pretrained(snake_case__ )

     # make sure private variable is not incorrectly saved
     UpperCAmelCase__      :					List[Any]				       =							json.loads(config.to_json_string() )
     self.assertTrue("_processor_class" not in dict_as_saved )

   self.assertIsInstance(snake_case__			,				snake_case__ )







 def 							UpperCamelCase  (					self		:  int ):



   '''simple docstring'''




   UpperCAmelCase__      :					Any				       =							AutoFeatureExtractor.from_pretrained(snake_case__ )
   self.assertIsInstance(snake_case__			,				snake_case__ )







 def 							UpperCamelCase  (					self		:  Tuple ):



   '''simple docstring'''




   with self.assertRaisesRegex(
       snake_case__			,				"bert-base is not a local folder and is not a valid model identifier" ):
     UpperCAmelCase__      :					Optional[int]				       =							AutoFeatureExtractor.from_pretrained("bert-base" )







 def 							UpperCamelCase  (					self		:  Any ):



   '''simple docstring'''




   with self.assertRaisesRegex(
       snake_case__			,				R"aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)" ):
     UpperCAmelCase__      :					List[str]				       =							AutoFeatureExtractor.from_pretrained(snake_case__			,				revision="aaaaaa" )







 def 							UpperCamelCase  (					self		:  Union[str, Any] ):



   '''simple docstring'''




   with self.assertRaisesRegex(
       snake_case__			,				"hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."			,				):
     UpperCAmelCase__      :					Optional[Any]				       =							AutoFeatureExtractor.from_pretrained("hf-internal-testing/config-no-model" )







 def 							UpperCamelCase  (					self		:  Optional[Any] ):



   '''simple docstring'''




   with self.assertRaises(snake_case__ ):
     UpperCAmelCase__      :					List[str]				       =							AutoFeatureExtractor.from_pretrained(
         "hf-internal-testing/test_dynamic_feature_extractor" )
   # If remote code is disabled, we can't load this config.
   with self.assertRaises(snake_case__ ):
     UpperCAmelCase__      :					Any				       =							AutoFeatureExtractor.from_pretrained(
         "hf-internal-testing/test_dynamic_feature_extractor"			,				trust_remote_code=snake_case__ )

   UpperCAmelCase__      :					Optional[Any]				       =							AutoFeatureExtractor.from_pretrained(
       "hf-internal-testing/test_dynamic_feature_extractor"			,				trust_remote_code=snake_case__ )
   self.assertEqual(feature_extractor.__class__.__name__			,				"NewFeatureExtractor" )

   # Test feature extractor can be reloaded.
   with tempfile.TemporaryDirectory() as tmp_dir:
     feature_extractor.save_pretrained(snake_case__ )
     UpperCAmelCase__      :					Any				       =							AutoFeatureExtractor.from_pretrained(snake_case__			,				trust_remote_code=snake_case__ )
   self.assertEqual(reloaded_feature_extractor.__class__.__name__			,				"NewFeatureExtractor" )







 def 							UpperCamelCase  (					self		:  str ):



   '''simple docstring'''




   try:
     AutoConfig.register("custom"			,				snake_case__ )
     AutoFeatureExtractor.register(snake_case__			,				snake_case__ )
     # Trying to register something existing in the Transformers library will raise an error
     with self.assertRaises(snake_case__ ):
       AutoFeatureExtractor.register(snake_case__			,				snake_case__ )

     # Now that the config is registered, it can be used as any other config with the auto-API
     UpperCAmelCase__      :					str				       =							CustomFeatureExtractor.from_pretrained(snake_case__ )
     with tempfile.TemporaryDirectory() as tmp_dir:
       feature_extractor.save_pretrained(snake_case__ )
       UpperCAmelCase__      :					Optional[int]				       =							AutoFeatureExtractor.from_pretrained(snake_case__ )
       self.assertIsInstance(snake_case__			,				snake_case__ )

   finally:
     if "custom" in CONFIG_MAPPING._extra_content:
       del CONFIG_MAPPING._extra_content["custom"]
     if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
       del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]







 def 							UpperCamelCase  (					self		:  str ):



   '''simple docstring'''




   class 							UpperCAmelCase_							(    A				):

    '''simple docstring'''
    lowercase_					:    Optional[Any]				=							True

   try:
     AutoConfig.register("custom"			,				snake_case__ )
     AutoFeatureExtractor.register(snake_case__			,				snake_case__ )
     # If remote code is not set, the default is to use local
     UpperCAmelCase__      :					Any				       =							AutoFeatureExtractor.from_pretrained(
         "hf-internal-testing/test_dynamic_feature_extractor" )
     self.assertEqual(feature_extractor.__class__.__name__			,				"NewFeatureExtractor" )
     self.assertTrue(feature_extractor.is_local )

     # If remote code is disabled, we load the local one.
     UpperCAmelCase__      :					Optional[int]				       =							AutoFeatureExtractor.from_pretrained(
         "hf-internal-testing/test_dynamic_feature_extractor"			,				trust_remote_code=snake_case__ )
     self.assertEqual(feature_extractor.__class__.__name__			,				"NewFeatureExtractor" )
     self.assertTrue(feature_extractor.is_local )

     # If remote is enabled, we load from the Hub
     UpperCAmelCase__      :					int				       =							AutoFeatureExtractor.from_pretrained(
         "hf-internal-testing/test_dynamic_feature_extractor"			,				trust_remote_code=snake_case__ )
     self.assertEqual(feature_extractor.__class__.__name__			,				"NewFeatureExtractor" )
     self.assertTrue(not hasattr(snake_case__			,				"is_local" ) )

   finally:
     if "custom" in CONFIG_MAPPING._extra_content:
       del CONFIG_MAPPING._extra_content["custom"]
     if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
       del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]



