



"""simple docstring"""
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 		lowerCamelCase__    (  cls	:   List[Any]    ,						lowerCAmelCase	:   CommonSchedulerState    ,						lowerCAmelCase	:   jnp.ndarray    ,						lowerCAmelCase	:   jnp.ndarray    ,						lowerCAmelCase	:   jnp.ndarray						)       ->   Union[str, Any]:


     '''simple docstring'''




     return cls(common=lowerCAmelCase    ,						init_noise_sigma=lowerCAmelCase    ,						timesteps=lowerCAmelCase    ,						sigmas=lowerCAmelCase						)







@dataclass
class    a		(					UpperCAmelCase__    ):
  UpperCamelCase		:			LMSDiscreteSchedulerState







class    a		(					UpperCAmelCase__		, UpperCAmelCase__    ):
  UpperCamelCase		:			List[Any]         =   [e.name for e in FlaxKarrasDiffusionSchedulers]

  UpperCamelCase		:			jnp.dtype



  @property
  def 		lowerCamelCase__    (  self	:   List[Any]						)       ->   Any:


     '''simple docstring'''




     return True



  @register_to_config
  def __init__(  self	:   Dict    ,						lowerCAmelCase	:   int = 1000    ,						lowerCAmelCase	:   float = 0.0_0_0_1    ,						lowerCAmelCase	:   float = 0.0_2    ,						lowerCAmelCase	:   str = "linear"    ,						lowerCAmelCase	:   Optional[jnp.ndarray] = None    ,						lowerCAmelCase	:   str = "epsilon"    ,						lowerCAmelCase	:   jnp.dtype = jnp.floataa    ,						)       ->   List[Any]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	int            =dtype



  def 		lowerCamelCase__    (  self	:   Optional[Any]    ,						lowerCAmelCase	:   Optional[CommonSchedulerState] = None						)       ->   LMSDiscreteSchedulerState:


     '''simple docstring'''




     if common is None:
        SCREAMING_SNAKE_CASE_:	Optional[Any]            =CommonSchedulerState.create(self						)

     SCREAMING_SNAKE_CASE_:	Optional[Any]            =jnp.arange(0    ,						self.config.num_train_timesteps						).round()[::-1]
     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =((1 - common.alphas_cumprod) / common.alphas_cumprod) ** 0.5

     # standard deviation of the initial noise distribution
     SCREAMING_SNAKE_CASE_:	List[Any]            =sigmas.max()

     return LMSDiscreteSchedulerState.create(
         common=lowerCAmelCase    ,						init_noise_sigma=lowerCAmelCase    ,						timesteps=lowerCAmelCase    ,						sigmas=lowerCAmelCase    ,						)



  def 		lowerCamelCase__    (  self	:   Tuple    ,						lowerCAmelCase	:   LMSDiscreteSchedulerState    ,						lowerCAmelCase	:   jnp.ndarray    ,						lowerCAmelCase	:   int						)       ->   jnp.ndarray:


     '''simple docstring'''




     ((SCREAMING_SNAKE_CASE_)     ,						):	Optional[Any]            =jnp.where(state.timesteps == timestep    ,						size=1						)
     SCREAMING_SNAKE_CASE_:	List[Any]            =step_index[0]

     SCREAMING_SNAKE_CASE_:	Optional[Any]            =state.sigmas[step_index]
     SCREAMING_SNAKE_CASE_:	int            =sample / ((sigma**2 + 1) ** 0.5)
     return sample



  def 		lowerCamelCase__    (  self	:   Optional[int]    ,						lowerCAmelCase	:   LMSDiscreteSchedulerState    ,						lowerCAmelCase	:   str    ,						lowerCAmelCase	:   Optional[int]    ,						lowerCAmelCase	:   Dict						)       ->   int:


     '''simple docstring'''




     def lms_derivative(lowerCAmelCase	:   str						):
        SCREAMING_SNAKE_CASE_:	Union[str, Any]            =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

     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =integrate.quad(lowerCAmelCase    ,						state.sigmas[t]    ,						state.sigmas[t + 1]    ,						epsrel=1E-4						)[0]

     return integrated_coeff



  def 		lowerCamelCase__    (  self	:   Tuple    ,						lowerCAmelCase	:   LMSDiscreteSchedulerState    ,						lowerCAmelCase	:   int    ,						lowerCAmelCase	:   Tuple = ()						)       ->   LMSDiscreteSchedulerState:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Tuple            =jnp.linspace(self.config.num_train_timesteps - 1    ,						0    ,						lowerCAmelCase    ,						dtype=self.dtype						)

     SCREAMING_SNAKE_CASE_:	List[str]            =jnp.floor(lowerCAmelCase						).astype(jnp.intaa						)
     SCREAMING_SNAKE_CASE_:	Optional[Any]            =jnp.ceil(lowerCAmelCase						).astype(jnp.intaa						)

     SCREAMING_SNAKE_CASE_:	int            =jnp.mod(lowerCAmelCase    ,						1.0						)

     SCREAMING_SNAKE_CASE_:	str            =((1 - state.common.alphas_cumprod) / state.common.alphas_cumprod) ** 0.5
     SCREAMING_SNAKE_CASE_:	Optional[int]            =(1 - frac) * sigmas[low_idx] + frac * sigmas[high_idx]
     SCREAMING_SNAKE_CASE_:	List[str]            =jnp.concatenate([sigmas, jnp.array([0.0]    ,						dtype=self.dtype						)]						)

     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =timesteps.astype(jnp.intaa						)

     # initial running values
     SCREAMING_SNAKE_CASE_:	Tuple            =jnp.zeros((0,) + shape    ,						dtype=self.dtype						)

     return state.replace(
         timesteps=lowerCAmelCase    ,						sigmas=lowerCAmelCase    ,						num_inference_steps=lowerCAmelCase    ,						derivatives=lowerCAmelCase    ,						)



  def 		lowerCamelCase__    (  self	:   int    ,						lowerCAmelCase	:   LMSDiscreteSchedulerState    ,						lowerCAmelCase	:   jnp.ndarray    ,						lowerCAmelCase	:   int    ,						lowerCAmelCase	:   jnp.ndarray    ,						lowerCAmelCase	:   int = 4    ,						lowerCAmelCase	:   bool = True    ,						)       ->   Union[FlaxLMSSchedulerOutput, 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"""						)

     SCREAMING_SNAKE_CASE_:	Tuple            =state.sigmas[timestep]

     # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
     if self.config.prediction_type == "epsilon":
        SCREAMING_SNAKE_CASE_:	Optional[Any]            =sample - sigma * model_output
     elif self.config.prediction_type == "v_prediction":
        # * c_out + input * c_skip
        SCREAMING_SNAKE_CASE_:	Optional[Any]            =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
     SCREAMING_SNAKE_CASE_:	Optional[Any]            =(sample - pred_original_sample) / sigma
     SCREAMING_SNAKE_CASE_:	Tuple            =state.replace(derivatives=jnp.append(state.derivatives    ,						lowerCAmelCase						)						)
     if len(state.derivatives						) > order:
        SCREAMING_SNAKE_CASE_:	Optional[Any]            =state.replace(derivatives=jnp.delete(state.derivatives    ,						0						)						)

     # 3. Compute linear multistep coefficients
     SCREAMING_SNAKE_CASE_:	List[Any]            =min(timestep + 1    ,						lowerCAmelCase						)
     SCREAMING_SNAKE_CASE_:	Optional[Any]            =[self.get_lms_coefficient(lowerCAmelCase    ,						lowerCAmelCase    ,						lowerCAmelCase    ,						lowerCAmelCase						) for curr_order in range(lowerCAmelCase						)]

     # 4. Compute previous sample based on the derivatives path
     SCREAMING_SNAKE_CASE_:	int            =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 		lowerCamelCase__    (  self	:   List[Any]    ,						lowerCAmelCase	:   LMSDiscreteSchedulerState    ,						lowerCAmelCase	:   jnp.ndarray    ,						lowerCAmelCase	:   jnp.ndarray    ,						lowerCAmelCase	:   jnp.ndarray    ,						)       ->   jnp.ndarray:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	str            =state.sigmas[timesteps].flatten()
     SCREAMING_SNAKE_CASE_:	List[Any]            =broadcast_to_shape_from_left(lowerCAmelCase    ,						noise.shape						)

     SCREAMING_SNAKE_CASE_:	str            =original_samples + noise * sigma

     return noisy_samples





  def __len__(  self	:   str						)       ->   Optional[Any]:


     '''simple docstring'''




     return self.config.num_train_timesteps








"""simple docstring"""
import shutil
import tempfile
import unittest

import pytest

from transformers import WhisperTokenizer, is_speech_available
from transformers.testing_utils import require_sentencepiece, require_torch, require_torchaudio

from .test_feature_extraction_whisper import floats_list


if is_speech_available():
   from transformers import WhisperFeatureExtractor, WhisperProcessor


_UpperCAmelCase       =     5_0_3_5_8
_UpperCAmelCase       =     5_0_3_6_2
@require_torch
@require_torchaudio
@require_sentencepiece
class    a		(					unittest.TestCase    ):



  def 		lowerCamelCase__    (  self	:   Dict						)       ->   Tuple:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	List[Any]            ="""openai/whisper-small.en"""
     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =tempfile.mkdtemp()



  def 		lowerCamelCase__    (  self	:   Any    ,						**lowerCAmelCase	:   str						)       ->   str:


     '''simple docstring'''




     return WhisperTokenizer.from_pretrained(self.checkpoint    ,						**lowerCAmelCase						)



  def 		lowerCamelCase__    (  self	:   str    ,						**lowerCAmelCase	:   Optional[Any]						)       ->   Optional[Any]:


     '''simple docstring'''




     return WhisperFeatureExtractor.from_pretrained(self.checkpoint    ,						**lowerCAmelCase						)



  def 		lowerCamelCase__    (  self	:   Tuple						)       ->   Optional[Any]:


     '''simple docstring'''




     shutil.rmtree(self.tmpdirname						)



  def 		lowerCamelCase__    (  self	:   int						)       ->   str:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Optional[int]            =self.get_tokenizer()
     SCREAMING_SNAKE_CASE_:	Optional[int]            =self.get_feature_extractor()

     SCREAMING_SNAKE_CASE_:	Optional[Any]            =WhisperProcessor(tokenizer=lowerCAmelCase    ,						feature_extractor=lowerCAmelCase						)

     processor.save_pretrained(self.tmpdirname						)
     SCREAMING_SNAKE_CASE_:	Dict            =WhisperProcessor.from_pretrained(self.tmpdirname						)

     self.assertEqual(processor.tokenizer.get_vocab()    ,						tokenizer.get_vocab()						)
     self.assertIsInstance(processor.tokenizer    ,						lowerCAmelCase						)

     self.assertEqual(processor.feature_extractor.to_json_string()    ,						feature_extractor.to_json_string()						)
     self.assertIsInstance(processor.feature_extractor    ,						lowerCAmelCase						)



  def 		lowerCamelCase__    (  self	:   str						)       ->   Optional[Any]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =WhisperProcessor(tokenizer=self.get_tokenizer()    ,						feature_extractor=self.get_feature_extractor()						)
     processor.save_pretrained(self.tmpdirname						)

     SCREAMING_SNAKE_CASE_:	int            =self.get_tokenizer(bos_token="""(BOS)"""    ,						eos_token="""(EOS)"""						)
     SCREAMING_SNAKE_CASE_:	Dict            =self.get_feature_extractor(do_normalize=lowerCAmelCase    ,						padding_value=1.0						)

     SCREAMING_SNAKE_CASE_:	Optional[int]            =WhisperProcessor.from_pretrained(
         self.tmpdirname    ,						bos_token="""(BOS)"""    ,						eos_token="""(EOS)"""    ,						do_normalize=lowerCAmelCase    ,						padding_value=1.0						)

     self.assertEqual(processor.tokenizer.get_vocab()    ,						tokenizer_add_kwargs.get_vocab()						)
     self.assertIsInstance(processor.tokenizer    ,						lowerCAmelCase						)

     self.assertEqual(processor.feature_extractor.to_json_string()    ,						feature_extractor_add_kwargs.to_json_string()						)
     self.assertIsInstance(processor.feature_extractor    ,						lowerCAmelCase						)



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


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Tuple            =self.get_feature_extractor()
     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =self.get_tokenizer()

     SCREAMING_SNAKE_CASE_:	Dict            =WhisperProcessor(tokenizer=lowerCAmelCase    ,						feature_extractor=lowerCAmelCase						)

     SCREAMING_SNAKE_CASE_:	Dict            =floats_list((3, 1000)						)

     SCREAMING_SNAKE_CASE_:	Optional[Any]            =feature_extractor(lowerCAmelCase    ,						return_tensors="""np"""						)
     SCREAMING_SNAKE_CASE_:	str            =processor(lowerCAmelCase    ,						return_tensors="""np"""						)

     for key in input_feat_extract.keys():
        self.assertAlmostEqual(input_feat_extract[key].sum()    ,						input_processor[key].sum()    ,						delta=1E-2						)



  def 		lowerCamelCase__    (  self	:   Tuple						)       ->   Any:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	List[str]            =self.get_feature_extractor()
     SCREAMING_SNAKE_CASE_:	str            =self.get_tokenizer()

     SCREAMING_SNAKE_CASE_:	List[Any]            =WhisperProcessor(tokenizer=lowerCAmelCase    ,						feature_extractor=lowerCAmelCase						)

     SCREAMING_SNAKE_CASE_:	List[str]            ="""This is a test string"""

     SCREAMING_SNAKE_CASE_:	Tuple            =processor(text=lowerCAmelCase						)

     SCREAMING_SNAKE_CASE_:	List[Any]            =tokenizer(lowerCAmelCase						)

     for key in encoded_tok.keys():
        self.assertListEqual(encoded_tok[key]    ,						encoded_processor[key]						)



  def 		lowerCamelCase__    (  self	:   Optional[Any]						)       ->   Any:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Optional[int]            =self.get_feature_extractor()
     SCREAMING_SNAKE_CASE_:	str            =self.get_tokenizer()

     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =WhisperProcessor(tokenizer=lowerCAmelCase    ,						feature_extractor=lowerCAmelCase						)

     SCREAMING_SNAKE_CASE_:	List[str]            =[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

     SCREAMING_SNAKE_CASE_:	Optional[int]            =processor.batch_decode(lowerCAmelCase						)
     SCREAMING_SNAKE_CASE_:	Dict            =tokenizer.batch_decode(lowerCAmelCase						)

     self.assertListEqual(lowerCAmelCase    ,						lowerCAmelCase						)



  def 		lowerCamelCase__    (  self	:   Union[str, Any]						)       ->   Dict:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	str            =self.get_feature_extractor()
     SCREAMING_SNAKE_CASE_:	List[Any]            =self.get_tokenizer()

     SCREAMING_SNAKE_CASE_:	Optional[Any]            =WhisperProcessor(tokenizer=lowerCAmelCase    ,						feature_extractor=lowerCAmelCase						)

     self.assertListEqual(
         processor.model_input_names    ,						feature_extractor.model_input_names    ,						msg="""`processor` and `feature_extractor` model input names do not match"""    ,						)



  def 		lowerCamelCase__    (  self	:   Optional[int]						)       ->   Union[str, Any]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Optional[Any]            =self.get_feature_extractor()
     SCREAMING_SNAKE_CASE_:	Any            =self.get_tokenizer()

     SCREAMING_SNAKE_CASE_:	List[Any]            =WhisperProcessor(tokenizer=lowerCAmelCase    ,						feature_extractor=lowerCAmelCase						)
     SCREAMING_SNAKE_CASE_:	Any            =processor.get_decoder_prompt_ids(task="""transcribe"""    ,						no_timestamps=lowerCAmelCase						)

     self.assertIsInstance(lowerCAmelCase    ,						lowerCAmelCase						)
     for ids in forced_decoder_ids:
        self.assertIsInstance(lowerCAmelCase    ,						(list, tuple)						)

     SCREAMING_SNAKE_CASE_:	Tuple            =[TRANSCRIBE, NOTIMESTAMPS]
     self.assertListEqual([ids[-1] for ids in forced_decoder_ids]    ,						lowerCAmelCase						)



  def 		lowerCamelCase__    (  self	:   List[str]						)       ->   List[str]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =WhisperProcessor(tokenizer=self.get_tokenizer()    ,						feature_extractor=self.get_feature_extractor()						)
     SCREAMING_SNAKE_CASE_:	Optional[int]            =processor.get_prompt_ids("""Mr. Quilter"""						)
     SCREAMING_SNAKE_CASE_:	List[str]            =processor.tokenizer.decode(lowerCAmelCase						)

     self.assertListEqual(prompt_ids.tolist()    ,						[5_0360, 1770, 13, 2264, 346, 353]						)
     self.assertEqual(lowerCAmelCase    ,						"""<|startofprev|> Mr. Quilter"""						)



  def 		lowerCamelCase__    (  self	:   Dict						)       ->   Tuple:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Optional[Any]            =WhisperProcessor(tokenizer=self.get_tokenizer()    ,						feature_extractor=self.get_feature_extractor()						)
     SCREAMING_SNAKE_CASE_:	int            =processor.get_prompt_ids(""""""						)
     SCREAMING_SNAKE_CASE_:	Union[str, Any]            =processor.tokenizer.decode(lowerCAmelCase						)

     self.assertListEqual(prompt_ids.tolist()    ,						[5_0360, 220]						)
     self.assertEqual(lowerCAmelCase    ,						"""<|startofprev|> """						)





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


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:	Dict            =WhisperProcessor(tokenizer=self.get_tokenizer()    ,						feature_extractor=self.get_feature_extractor()						)

     def _test_prompt_error_raised_helper(lowerCAmelCase	:   List[str]    ,						lowerCAmelCase	:   int						):
        with pytest.raises(lowerCAmelCase						) as excinfo:
           processor.get_prompt_ids(lowerCAmelCase						)
        SCREAMING_SNAKE_CASE_:	List[Any]            =f'''Encountered text in the prompt corresponding to disallowed special token: {special_token}.'''
        self.assertEqual(lowerCAmelCase    ,						str(excinfo.value						)						)

     _test_prompt_error_raised_helper("""<|startofprev|> test"""    ,						"""<|startofprev|>"""						)
     _test_prompt_error_raised_helper("""test <|notimestamps|>"""    ,						"""<|notimestamps|>"""						)
     _test_prompt_error_raised_helper("""test <|zh|> test <|transcribe|>"""    ,						"""<|zh|>"""						)



