
# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion
import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.schedulers.scheduling_utils import SchedulerMixin
from diffusers.utils import BaseOutput, deprecate

@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->DDIM
class __snake_case		(      lowerCAmelCase     ):
    _a		:						torch.FloatTensor
    _a		:						Optional[torch.FloatTensor]=				None







def 						_snake_case(  SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__=0.999					,	SCREAMING_SNAKE_CASE__="cosine"					,	) ->    Optional[Any]:
       if alpha_transform_type == "cosine":

              def alpha_bar_fn(SCREAMING_SNAKE_CASE__				):
                     return math.cos((t + 0.008) / 1.008 * math.pi / 2				) ** 2

       elif alpha_transform_type == "exp":

              def alpha_bar_fn(SCREAMING_SNAKE_CASE__				):
                     return math.exp(t * -12.0				)

       else:
              raise ValueError(f"Unsupported alpha_tranform_type: {alpha_transform_type}"				)

       lowercase							:     Tuple       	=      []
       for i in range(SCREAMING_SNAKE_CASE__				):
              lowercase							:     List[str]       	=      i / num_diffusion_timesteps
              lowercase							:     List[str]       	=      (i + 1) / num_diffusion_timesteps
              betas.append(min(1 - alpha_bar_fn(SCREAMING_SNAKE_CASE__				) / alpha_bar_fn(SCREAMING_SNAKE_CASE__				)					,	SCREAMING_SNAKE_CASE__				)				)
       return torch.tensor(SCREAMING_SNAKE_CASE__					,	dtype=torch.floataa				)




class __snake_case		(      lowerCAmelCase					,	lowerCAmelCase     ):
    _a		:						Union[str, Any]=				1
    @register_to_config
    def __init__(  self		,snake_case = 1000		,snake_case = 0.0_001		,snake_case = 0.02		,snake_case = "linear"		,snake_case = None		,snake_case = True		,snake_case = True		,snake_case = 0		,snake_case = "epsilon"		,snake_case = 1.0		,**snake_case		,):



           '''simple docstring'''







           if kwargs.get("""set_alpha_to_one"""		,snake_case      ) is not None:
                  lowercase							:     Any       	=      (
                      """The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead."""
                  )
                  deprecate("""set_alpha_to_one"""		,"""1.0.0"""		,snake_case		,standard_warn=snake_case      )
                  lowercase							:     List[str]       	=      kwargs["""set_alpha_to_one"""]
           if trained_betas is not None:
                  lowercase							:     str       	=      torch.tensor(snake_case		,dtype=torch.floataa      )
           elif beta_schedule == "linear":
                  lowercase							:     Union[str, Any]       	=      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.
                  lowercase							:     int       	=      (
                      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
                  lowercase							:     List[Any]       	=      betas_for_alpha_bar(snake_case      )
           else:
                  raise NotImplementedError(f"{beta_schedule} does is not implemented for {self.__class__}"      )

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

           # At every step in inverted ddim, we are looking into the next alphas_cumprod
           # For the final step, there is no next alphas_cumprod, and the index is out of bounds
           # `set_alpha_to_zero` decides whether we set this parameter simply to zero
           # in this case, self.step() just output the predicted noise
           # or whether we use the final alpha of the "non-previous" one.
           lowercase							:     Optional[int]       	=      torch.tensor(0.0      ) if set_alpha_to_zero else self.alphas_cumprod[-1]

           # standard deviation of the initial noise distribution
           lowercase							:     Dict       	=      1.0

           # setable values
           lowercase							:     List[Any]       	=      None
           lowercase							:     Dict       	=      torch.from_numpy(np.arange(0		,snake_case      ).copy().astype(np.intaa      )      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = None      ):



           '''simple docstring'''







           return sample
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = None      ):



           '''simple docstring'''







           if num_inference_steps > self.config.num_train_timesteps:
                  raise ValueError(
                      f"`num_inference_steps`: {num_inference_steps} cannot be larger than `self.config.train_timesteps`:"
                      f" {self.config.num_train_timesteps} as the unet model trained with this scheduler can only handle"
                      f" maximal {self.config.num_train_timesteps} timesteps."      )

           lowercase							:     Any       	=      num_inference_steps
           lowercase							:     Any       	=      self.config.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
           lowercase							:     Union[str, Any]       	=      (np.arange(0		,snake_case      ) * step_ratio).round().copy().astype(np.intaa      )
           lowercase							:     Dict       	=      torch.from_numpy(snake_case      ).to(snake_case      )
           self.timesteps += self.config.steps_offset
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case = 0.0		,snake_case = False		,snake_case = None		,snake_case = True		,):



           '''simple docstring'''







           lowercase							:     List[str]       	=      timestep + self.config.num_train_timesteps // self.num_inference_steps

           # 2. compute alphas, betas
           # change original implementation to exactly match noise levels for analogous forward process
           lowercase							:     Optional[Any]       	=      self.alphas_cumprod[timestep]
           lowercase							:     str       	=      (
               self.alphas_cumprod[prev_timestep]
               if prev_timestep < self.config.num_train_timesteps
               else self.final_alpha_cumprod
           )

           lowercase							:     Any       	=      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":
                  lowercase							:     Union[str, Any]       	=      (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
                  lowercase							:     str       	=      model_output
           elif self.config.prediction_type == "sample":
                  lowercase							:     str       	=      model_output
                  lowercase							:     str       	=      (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
           elif self.config.prediction_type == "v_prediction":
                  lowercase							:     str       	=      (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
                  lowercase							:     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. Clip or threshold "predicted x_0"
           if self.config.clip_sample:
                  lowercase							:     int       	=      pred_original_sample.clamp(
                      -self.config.clip_sample_range		,self.config.clip_sample_range      )

           # 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
           lowercase							:     Dict       	=      (1 - alpha_prod_t_prev) ** 0.5 * pred_epsilon

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

           if not return_dict:
                  return (prev_sample, pred_original_sample)
           return DDIMSchedulerOutput(prev_sample=snake_case		,pred_original_sample=snake_case      )
    def __len__(  self      ):



           '''simple docstring'''







           return self.config.num_train_timesteps


from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
  import numpy as np
  import tensorflow as tf

  from transformers import (
      TF_FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
      FlaubertConfig,
      TFFlaubertForMultipleChoice,
      TFFlaubertForQuestionAnsweringSimple,
      TFFlaubertForSequenceClassification,
      TFFlaubertForTokenClassification,
      TFFlaubertModel,
      TFFlaubertWithLMHeadModel,
  )

class __snake_case		:
    def __init__(  self		,snake_case		,):



           '''simple docstring'''







           lowercase							:     Any       	=      parent
           lowercase							:     Tuple       	=      13
           lowercase							:     str       	=      7
           lowercase							:     Dict       	=      True
           lowercase							:     Dict       	=      True
           lowercase							:     str       	=      True
           lowercase							:     List[str]       	=      True
           lowercase							:     int       	=      True
           lowercase							:     Union[str, Any]       	=      False
           lowercase							:     Dict       	=      False
           lowercase							:     List[Any]       	=      False
           lowercase							:     List[Any]       	=      2
           lowercase							:     Optional[Any]       	=      99
           lowercase							:     int       	=      0
           lowercase							:     Tuple       	=      32
           lowercase							:     int       	=      2
           lowercase							:     Tuple       	=      4
           lowercase							:     List[Any]       	=      0.1
           lowercase							:     Tuple       	=      0.1
           lowercase							:     List[Any]       	=      512
           lowercase							:     int       	=      16
           lowercase							:     Dict       	=      2
           lowercase							:     int       	=      0.02
           lowercase							:     Union[str, Any]       	=      3
           lowercase							:     Any       	=      4
           lowercase							:     List[Any]       	=      """last"""
           lowercase							:     Tuple       	=      True
           lowercase							:     List[Any]       	=      None
           lowercase							:     Any       	=      0
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     List[Any]       	=      ids_tensor([self.batch_size, self.seq_length]		,self.vocab_size      )
           lowercase							:     List[str]       	=      random_attention_mask([self.batch_size, self.seq_length]		,dtype=tf.floataa      )

           lowercase							:     Tuple       	=      None
           if self.use_input_lengths:
                  lowercase							:     List[str]       	=      (
                      ids_tensor([self.batch_size]		,vocab_size=2      ) + self.seq_length - 2
                  )  # small variation of seq_length

           lowercase							:     Tuple       	=      None
           if self.use_token_type_ids:
                  lowercase							:     Optional[int]       	=      ids_tensor([self.batch_size, self.seq_length]		,self.n_langs      )

           lowercase							:     List[str]       	=      None
           lowercase							:     List[str]       	=      None
           lowercase							:     Optional[Any]       	=      None
           if self.use_labels:
                  lowercase							:     List[str]       	=      ids_tensor([self.batch_size]		,self.type_sequence_label_size      )
                  lowercase							:     List[str]       	=      ids_tensor([self.batch_size, self.seq_length]		,self.num_labels      )
                  lowercase							:     str       	=      ids_tensor([self.batch_size]		,2		,dtype=tf.floataa      )
                  lowercase							:     Optional[Any]       	=      ids_tensor([self.batch_size]		,self.num_choices      )

           lowercase							:     str       	=      FlaubertConfig(
               vocab_size=self.vocab_size		,n_special=self.n_special		,emb_dim=self.hidden_size		,n_layers=self.num_hidden_layers		,n_heads=self.num_attention_heads		,dropout=self.hidden_dropout_prob		,attention_dropout=self.attention_probs_dropout_prob		,gelu_activation=self.gelu_activation		,sinusoidal_embeddings=self.sinusoidal_embeddings		,asm=self.asm		,causal=self.causal		,n_langs=self.n_langs		,max_position_embeddings=self.max_position_embeddings		,initializer_range=self.initializer_range		,summary_type=self.summary_type		,use_proj=self.use_proj		,bos_token_id=self.bos_token_id		,)

           return (
               config,
               input_ids,
               token_type_ids,
               input_lengths,
               sequence_labels,
               token_labels,
               is_impossible_labels,
               choice_labels,
               input_mask,
           )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,):



           '''simple docstring'''







           lowercase							:     Tuple       	=      TFFlaubertModel(config=snake_case      )
           lowercase							:     str       	=      {"""input_ids""": input_ids, """lengths""": input_lengths, """langs""": token_type_ids}
           lowercase							:     Optional[Any]       	=      model(snake_case      )

           lowercase							:     List[Any]       	=      [input_ids, input_mask]
           lowercase							:     int       	=      model(snake_case      )
           self.parent.assertEqual(result.last_hidden_state.shape		,(self.batch_size, self.seq_length, self.hidden_size)      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,):



           '''simple docstring'''







           lowercase							:     List[Any]       	=      TFFlaubertWithLMHeadModel(snake_case      )

           lowercase							:     Optional[Any]       	=      {"""input_ids""": input_ids, """lengths""": input_lengths, """langs""": token_type_ids}
           lowercase							:     int       	=      model(snake_case      )

           self.parent.assertEqual(result.logits.shape		,(self.batch_size, self.seq_length, self.vocab_size)      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,):



           '''simple docstring'''







           lowercase							:     Tuple       	=      TFFlaubertForQuestionAnsweringSimple(snake_case      )

           lowercase							:     Union[str, Any]       	=      {"""input_ids""": input_ids, """lengths""": input_lengths}

           lowercase							:     Tuple       	=      model(snake_case      )

           self.parent.assertEqual(result.start_logits.shape		,(self.batch_size, self.seq_length)      )
           self.parent.assertEqual(result.end_logits.shape		,(self.batch_size, self.seq_length)      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,):



           '''simple docstring'''







           lowercase							:     Union[str, Any]       	=      TFFlaubertForSequenceClassification(snake_case      )

           lowercase							:     str       	=      {"""input_ids""": input_ids, """lengths""": input_lengths}

           lowercase							:     str       	=      model(snake_case      )

           self.parent.assertEqual(result.logits.shape		,(self.batch_size, self.type_sequence_label_size)      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,):



           '''simple docstring'''







           lowercase							:     Any       	=      self.num_labels
           lowercase							:     List[str]       	=      TFFlaubertForTokenClassification(config=snake_case      )
           lowercase							:     Dict       	=      {"""input_ids""": input_ids, """attention_mask""": input_mask, """token_type_ids""": token_type_ids}
           lowercase							:     int       	=      model(snake_case      )
           self.parent.assertEqual(result.logits.shape		,(self.batch_size, self.seq_length, self.num_labels)      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,):



           '''simple docstring'''







           lowercase							:     Any       	=      self.num_choices
           lowercase							:     Dict       	=      TFFlaubertForMultipleChoice(config=snake_case      )
           lowercase							:     Any       	=      tf.tile(tf.expand_dims(snake_case		,1      )		,(1, self.num_choices, 1)      )
           lowercase							:     Optional[Any]       	=      tf.tile(tf.expand_dims(snake_case		,1      )		,(1, self.num_choices, 1)      )
           lowercase							:     Dict       	=      tf.tile(tf.expand_dims(snake_case		,1      )		,(1, self.num_choices, 1)      )
           lowercase							:     Union[str, Any]       	=      {
               """input_ids""": multiple_choice_inputs_ids,
               """attention_mask""": multiple_choice_input_mask,
               """token_type_ids""": multiple_choice_token_type_ids,
           }
           lowercase							:     int       	=      model(snake_case      )
           self.parent.assertEqual(result.logits.shape		,(self.batch_size, self.num_choices)      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Any       	=      self.prepare_config_and_inputs()
           (
               (
               lowercase
           )	,			(
               lowercase
           )	,			(
               lowercase
           )	,			(
               lowercase
           )	,			(
               lowercase
           )	,			(
               lowercase
           )	,			(
               lowercase
           )	,			(
               lowercase
           )	,			(
               lowercase
           )	,			
           )							:     int       	=      config_and_inputs
           lowercase							:     List[str]       	=      {
               """input_ids""": input_ids,
               """token_type_ids""": token_type_ids,
               """langs""": token_type_ids,
               """lengths""": input_lengths,
           }
           return config, inputs_dict






@require_tf
class __snake_case		(      lowerCAmelCase					,	lowerCAmelCase					,	unittest.TestCase     ):
    _a		:						Dict=				(
        (
            TFFlaubertModel,
            TFFlaubertWithLMHeadModel,
            TFFlaubertForSequenceClassification,
            TFFlaubertForQuestionAnsweringSimple,
            TFFlaubertForTokenClassification,
            TFFlaubertForMultipleChoice,
        )
        if is_tf_available()
        else ()
    )
    _a		:						Optional[Any]=				(
        (TFFlaubertWithLMHeadModel,) if is_tf_available() else ()
    )  # TODO (PVP): Check other models whether language generation is also applicable
    _a		:						Any=				(
        {
            "feature-extraction": TFFlaubertModel,
            "fill-mask": TFFlaubertWithLMHeadModel,
            "question-answering": TFFlaubertForQuestionAnsweringSimple,
            "text-classification": TFFlaubertForSequenceClassification,
            "token-classification": TFFlaubertForTokenClassification,
            "zero-shot": TFFlaubertForSequenceClassification,
        }
        if is_tf_available()
        else {}
    )
    _a		:						Tuple=				False
    _a		:						int=				False
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case      ):



           '''simple docstring'''







           if (
               pipeline_test_casse_name == "QAPipelineTests"
               and tokenizer_name is not None
               and not tokenizer_name.endswith("""Fast"""      )
           ):
                  # `QAPipelineTests` fails for a few models when the slower tokenizer are used.
                  # (The slower tokenizers were never used for pipeline tests before the pipeline testing rework)
                  # TODO: check (and possibly fix) the `QAPipelineTests` with slower tokenizer
                  return True

           return False
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     str       	=      TFFlaubertModelTester(self      )
           lowercase							:     List[Any]       	=      ConfigTester(self		,config_class=snake_case		,emb_dim=37      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           self.config_tester.run_common_tests()
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Tuple       	=      self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_flaubert_model(*snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Optional[int]       	=      self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_flaubert_lm_head(*snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     int       	=      self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_flaubert_qa(*snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Optional[Any]       	=      self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_flaubert_sequence_classif(*snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     str       	=      self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_flaubert_for_token_classification(*snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     List[str]       	=      self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_flaubert_for_multiple_choice(*snake_case      )
    @slow
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           for model_name in TF_FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                  lowercase							:     Dict       	=      TFFlaubertModel.from_pretrained(snake_case      )
                  self.assertIsNotNone(snake_case      )






@require_tf
@require_sentencepiece
@require_tokenizers
class __snake_case		(      unittest.TestCase     ):
    @slow
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     List[Any]       	=      TFFlaubertModel.from_pretrained("""jplu/tf-flaubert-small-cased"""      )

           lowercase							:     int       	=      tf.convert_to_tensor(
               [[0, 158, 735, 2592, 1424, 6727, 82, 1]]		,dtype=tf.intaa		,)  # "J'aime flaubert !"

           lowercase							:     Dict       	=      model(snake_case      )[0]
           lowercase							:     Union[str, Any]       	=      tf.TensorShape((1, 8, 512)      )
           self.assertEqual(output.shape		,snake_case      )
           # compare the actual values for a slice.
           lowercase							:     Tuple       	=      tf.convert_to_tensor(
               [
                   [
                       [-1.8_768_773, -1.566_555, 0.27_072_418],
                       [-1.6_920_038, -0.5_873_505, 1.9_329_599],
                       [-2.9_563_985, -1.6_993_835, 1.7_972_052],
                   ]
               ]		,dtype=tf.floataa		,)

           self.assertTrue(np.allclose(output[:, :3, :3].numpy()		,expected_slice.numpy()		,atol=1e-4      )      )
