





from __future__ import annotations

import unittest

from transformers import BlenderbotSmallConfig, BlenderbotSmallTokenizer, is_tf_available
from transformers.testing_utils import require_tf, require_tokenizers, slow
from transformers.utils import cached_property

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


if is_tf_available():
     import tensorflow as tf

     from transformers import TFAutoModelForSeqaSeqLM, TFBlenderbotSmallForConditionalGeneration, TFBlenderbotSmallModel

@require_tf
class      A       :
      _snake_case							     =BlenderbotSmallConfig
      _snake_case							     ={}
      _snake_case							     ='''gelu'''



      def __init__(	self:						Dict		,  _lowerCAmelCase:						Optional[int]		,  _lowerCAmelCase:						str=13		,  _lowerCAmelCase:						List[str]=7		,  _lowerCAmelCase:						Dict=True		,  _lowerCAmelCase:						Optional[int]=False		,  _lowerCAmelCase:						Dict=99		,  _lowerCAmelCase:						List[str]=32		,  _lowerCAmelCase:						Dict=2		,  _lowerCAmelCase:						Tuple=4		,  _lowerCAmelCase:						Union[str, Any]=37		,  _lowerCAmelCase:						Optional[Any]=0.1		,  _lowerCAmelCase:						List[str]=0.1		,  _lowerCAmelCase:						List[str]=20		,  _lowerCAmelCase:						List[Any]=2		,  _lowerCAmelCase:						List[Any]=1		,  _lowerCAmelCase:						List[Any]=0		,  )						->    Dict:
         '''simple docstring'''

         UpperCAmelCase_											=parent
         UpperCAmelCase_											=batch_size
         UpperCAmelCase_											=seq_length
         UpperCAmelCase_											=is_training
         UpperCAmelCase_											=use_labels
         UpperCAmelCase_											=vocab_size
         UpperCAmelCase_											=hidden_size
         UpperCAmelCase_											=num_hidden_layers
         UpperCAmelCase_											=num_attention_heads
         UpperCAmelCase_											=intermediate_size

         UpperCAmelCase_											=hidden_dropout_prob
         UpperCAmelCase_											=attention_probs_dropout_prob
         UpperCAmelCase_											=max_position_embeddings
         UpperCAmelCase_											=eos_token_id
         UpperCAmelCase_											=pad_token_id
         UpperCAmelCase_											=bos_token_id



      def 	lowerCAmelCase__    (	self:						str       )						->    Optional[Any]:
         '''simple docstring'''

         UpperCAmelCase_											=ids_tensor([self.batch_size, self.seq_length - 1]		,  self.vocab_size       )
         UpperCAmelCase_											=tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size       )		,  1       )
         UpperCAmelCase_											=tf.concat([input_ids, eos_tensor]		,  axis=1       )

         UpperCAmelCase_											=ids_tensor([self.batch_size, self.seq_length]		,  self.vocab_size       )

         UpperCAmelCase_											=self.config_cls(
             vocab_size=self.vocab_size		,  d_model=self.hidden_size		,  encoder_layers=self.num_hidden_layers		,  decoder_layers=self.num_hidden_layers		,  encoder_attention_heads=self.num_attention_heads		,  decoder_attention_heads=self.num_attention_heads		,  encoder_ffn_dim=self.intermediate_size		,  decoder_ffn_dim=self.intermediate_size		,  dropout=self.hidden_dropout_prob		,  attention_dropout=self.attention_probs_dropout_prob		,  max_position_embeddings=self.max_position_embeddings		,  eos_token_ids=[2]		,  bos_token_id=self.bos_token_id		,  pad_token_id=self.pad_token_id		,  decoder_start_token_id=self.pad_token_id		,  **self.config_updates		,  )
         UpperCAmelCase_											=prepare_blenderbot_small_inputs_dict(_lowerCAmelCase		,  _lowerCAmelCase		,  _lowerCAmelCase       )
         return config, inputs_dict






      def 	lowerCAmelCase__    (	self:						Optional[Any]		,  _lowerCAmelCase:						Optional[Any]		,  _lowerCAmelCase:						Tuple       )						->    Tuple:
         '''simple docstring'''

         UpperCAmelCase_											=TFBlenderbotSmallModel(config=_lowerCAmelCase       ).get_decoder()
         UpperCAmelCase_											=inputs_dict["input_ids"]

         UpperCAmelCase_											=input_ids[:1, :]
         UpperCAmelCase_											=inputs_dict["attention_mask"][:1, :]
         UpperCAmelCase_											=inputs_dict["head_mask"]
         UpperCAmelCase_											=1

         # first forward pass
         UpperCAmelCase_											=model(_lowerCAmelCase		,  attention_mask=_lowerCAmelCase		,  head_mask=_lowerCAmelCase		,  use_cache=_lowerCAmelCase       )

         UpperCAmelCase_						,       UpperCAmelCase_											=outputs.to_tuple()

         # create hypothetical next token and extent to next_input_ids
         UpperCAmelCase_											=ids_tensor((self.batch_size, 3)		,  config.vocab_size       )
         UpperCAmelCase_											=tf.cast(ids_tensor((self.batch_size, 3)		,  2       )		,  tf.inta       )

         # append to next input_ids and
         UpperCAmelCase_											=tf.concat([input_ids, next_tokens]		,  axis=-1       )
         UpperCAmelCase_											=tf.concat([attention_mask, next_attn_mask]		,  axis=-1       )

         UpperCAmelCase_											=model(_lowerCAmelCase		,  attention_mask=_lowerCAmelCase       )[0]
         UpperCAmelCase_											=model(_lowerCAmelCase		,  attention_mask=_lowerCAmelCase		,  past_key_values=_lowerCAmelCase       )[0]

         self.parent.assertEqual(next_tokens.shape[1]		,  output_from_past.shape[1]       )

         # select random slice
         UpperCAmelCase_											=int(ids_tensor((1,)		,  output_from_past.shape[-1]       )       )
         UpperCAmelCase_											=output_from_no_past[:, -3:, random_slice_idx]
         UpperCAmelCase_											=output_from_past[:, :, random_slice_idx]

         # test that outputs are equal for slice
         tf.debugging.assert_near(_lowerCAmelCase		,  _lowerCAmelCase		,  rtol=1e-3       )




def     a__ (					lowercase__  ,    lowercase__  ,    lowercase__  ,    lowercase__=None  ,    lowercase__=None  ,    lowercase__=None  ,    lowercase__=None  ,    lowercase__=None  ,    ):






   '''simple docstring'''






   if attention_mask is None:
      UpperCAmelCase_											=tf.cast(tf.math.not_equal(lowercase__  ,    config.pad_token_id  )  ,    tf.inta  )
   if decoder_attention_mask is None:
      UpperCAmelCase_											=tf.concat(
          [
              tf.ones(decoder_input_ids[:, :1].shape  ,    dtype=tf.inta  ),
              tf.cast(tf.math.not_equal(decoder_input_ids[:, 1:]  ,    config.pad_token_id  )  ,    tf.inta  ),
          ]  ,    axis=-1  ,    )
   if head_mask is None:
      UpperCAmelCase_											=tf.ones((config.encoder_layers, config.encoder_attention_heads)  )
   if decoder_head_mask is None:
      UpperCAmelCase_											=tf.ones((config.decoder_layers, config.decoder_attention_heads)  )
   if cross_attn_head_mask is None:
      UpperCAmelCase_											=tf.ones((config.decoder_layers, config.decoder_attention_heads)  )
   return {
       "input_ids": input_ids,
       "decoder_input_ids": decoder_input_ids,
       "attention_mask": attention_mask,
       "decoder_attention_mask": decoder_attention_mask,
       "head_mask": head_mask,
       "decoder_head_mask": decoder_head_mask,
       "cross_attn_head_mask": cross_attn_head_mask,
   }

@require_tf
class      A       (	__lowercase		,      __lowercase		,      unittest.TestCase						):
      _snake_case							     =(
          (TFBlenderbotSmallForConditionalGeneration, TFBlenderbotSmallModel) if is_tf_available() else ()
      )
      _snake_case							     =(TFBlenderbotSmallForConditionalGeneration,) if is_tf_available() else ()
      _snake_case							     =(
          {
              '''conversational''': TFBlenderbotSmallForConditionalGeneration,
              '''feature-extraction''': TFBlenderbotSmallModel,
              '''summarization''': TFBlenderbotSmallForConditionalGeneration,
              '''text2text-generation''': TFBlenderbotSmallForConditionalGeneration,
              '''translation''': TFBlenderbotSmallForConditionalGeneration,
          }
          if is_tf_available()
          else {}
      )
      _snake_case							     =True
      _snake_case							     =False
      _snake_case							     =False



      def 	lowerCAmelCase__    (	self:						Optional[int]       )						->    List[Any]:
         '''simple docstring'''

         UpperCAmelCase_											=TFBlenderbotSmallModelTester(self       )
         UpperCAmelCase_											=ConfigTester(self		,  config_class=_lowerCAmelCase       )



      def 	lowerCAmelCase__    (	self:						Any       )						->    int:
         '''simple docstring'''

         self.config_tester.run_common_tests()






      def 	lowerCAmelCase__    (	self:						Union[str, Any]       )						->    Any:
         '''simple docstring'''

         UpperCAmelCase_											=self.model_tester.prepare_config_and_inputs_for_common()
         self.model_tester.check_decoder_model_past_large_inputs(*_lowerCAmelCase       )

@require_tokenizers
@require_tf
class      A       (	unittest.TestCase						):
      _snake_case							     =[
          '''Social anxiety\nWow, I am never shy. Do you have anxiety?\nYes. I end up sweating and blushing and feel like '''
          '''  i\'m going to throw up.\nand why is that?'''
      ]
      _snake_case							     ='''facebook/blenderbot_small-90M'''



      @cached_property
      def 	lowerCAmelCase__    (	self:						Optional[Any]       )						->    Dict:
         '''simple docstring'''

         return BlenderbotSmallTokenizer.from_pretrained("facebook/blenderbot-90M"       )



      @cached_property
      def 	lowerCAmelCase__    (	self:						Union[str, Any]       )						->    str:
         '''simple docstring'''

         UpperCAmelCase_											=TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name       )
         return model






      @slow
      def 	lowerCAmelCase__    (	self:						Optional[int]       )						->    Tuple:
         '''simple docstring'''

         UpperCAmelCase_											=self.tokenizer(self.src_text		,  return_tensors="tf"       )
         UpperCAmelCase_											=self.model.generate(
             model_inputs.input_ids		,  attention_mask=model_inputs.attention_mask		,  num_beams=2		,  use_cache=_lowerCAmelCase		,  )
         UpperCAmelCase_											=self.tokenizer.batch_decode(generated_ids.numpy()		,  skip_special_tokens=_lowerCAmelCase       )[0]
         assert generated_words in (
             "i don't know. i just feel like i'm going to throw up. it's not fun.",
             "i'm not sure. i just feel like i've been feeling like i have to be in a certain place",
             "i'm not sure. i just feel like i've been in a bad situation.",
         )











import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()

class      A       (	__lowercase		,      unittest.TestCase						):
      _snake_case							     =DanceDiffusionPipeline
      _snake_case							     =UNCONDITIONAL_AUDIO_GENERATION_PARAMS
      _snake_case							     =PipelineTesterMixin.required_optional_params - {
          '''callback''',
          '''latents''',
          '''callback_steps''',
          '''output_type''',
          '''num_images_per_prompt''',
      }
      _snake_case							     =UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
      _snake_case							     =False
      _snake_case							     =False



      def 	lowerCAmelCase__    (	self:						Tuple       )						->    Tuple:
         '''simple docstring'''

         torch.manual_seed(0       )
         UpperCAmelCase_											=UNetaDModel(
             block_out_channels=(32, 32, 64)		,  extra_in_channels=16		,  sample_size=512		,  sample_rate=1_6000		,  in_channels=2		,  out_channels=2		,  flip_sin_to_cos=_lowerCAmelCase		,  use_timestep_embedding=_lowerCAmelCase		,  time_embedding_type="fourier"		,  mid_block_type="UNetMidBlock1D"		,  down_block_types=("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")		,  up_block_types=("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")		,  )
         UpperCAmelCase_											=IPNDMScheduler()

         UpperCAmelCase_											={
             "unet": unet,
             "scheduler": scheduler,
         }
         return components



      def 	lowerCAmelCase__    (	self:						str		,  _lowerCAmelCase:						Tuple		,  _lowerCAmelCase:						List[str]=0       )						->    str:
         '''simple docstring'''

         if str(_lowerCAmelCase       ).startswith("mps"       ):
            UpperCAmelCase_											=torch.manual_seed(_lowerCAmelCase       )
         else:
            UpperCAmelCase_											=torch.Generator(device=_lowerCAmelCase       ).manual_seed(_lowerCAmelCase       )
         UpperCAmelCase_											={
             "batch_size": 1,
             "generator": generator,
             "num_inference_steps": 4,
         }
         return inputs



      def 	lowerCAmelCase__    (	self:						Dict       )						->    Tuple:
         '''simple docstring'''

         UpperCAmelCase_											="cpu"  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase_											=self.get_dummy_components()
         UpperCAmelCase_											=DanceDiffusionPipeline(**_lowerCAmelCase       )
         UpperCAmelCase_											=pipe.to(_lowerCAmelCase       )
         pipe.set_progress_bar_config(disable=_lowerCAmelCase       )

         UpperCAmelCase_											=self.get_dummy_inputs(_lowerCAmelCase       )
         UpperCAmelCase_											=pipe(**_lowerCAmelCase       )
         UpperCAmelCase_											=output.audios

         UpperCAmelCase_											=audio[0, -3:, -3:]

         assert audio.shape == (1, 2, components["unet"].sample_size)
         UpperCAmelCase_											=np.array([-0.72_65, 1.00_00, -0.83_88, 0.11_75, 0.94_98, -1.00_00]       )
         assert np.abs(audio_slice.flatten() - expected_slice       ).max() < 1e-2



      @skip_mps
      def 	lowerCAmelCase__    (	self:						Optional[int]       )						->    Any:
         '''simple docstring'''

         return super().test_save_load_local()



      @skip_mps
      def 	lowerCAmelCase__    (	self:						Union[str, Any]       )						->    int:
         '''simple docstring'''

         return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3       )



      @skip_mps
      def 	lowerCAmelCase__    (	self:						List[str]       )						->    Optional[int]:
         '''simple docstring'''

         return super().test_save_load_optional_components()



      @skip_mps
      def 	lowerCAmelCase__    (	self:						Optional[Any]       )						->    Union[str, Any]:
         '''simple docstring'''

         return super().test_attention_slicing_forward_pass()






      def 	lowerCAmelCase__    (	self:						Optional[int]       )						->    Optional[Any]:
         '''simple docstring'''

         super().test_inference_batch_single_identical(expected_max_diff=3e-3       )

@slow
@require_torch_gpu
class      A       (	unittest.TestCase						):



      def 	lowerCAmelCase__    (	self:						Any       )						->    Optional[Any]:
         '''simple docstring'''

         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()



      def 	lowerCAmelCase__    (	self:						Union[str, Any]       )						->    Tuple:
         '''simple docstring'''

         UpperCAmelCase_											=torch_device

         UpperCAmelCase_											=DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"       )
         UpperCAmelCase_											=pipe.to(_lowerCAmelCase       )
         pipe.set_progress_bar_config(disable=_lowerCAmelCase       )

         UpperCAmelCase_											=torch.manual_seed(0       )
         UpperCAmelCase_											=pipe(generator=_lowerCAmelCase		,  num_inference_steps=100		,  audio_length_in_s=4.0_96       )
         UpperCAmelCase_											=output.audios

         UpperCAmelCase_											=audio[0, -3:, -3:]

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         UpperCAmelCase_											=np.array([-0.01_92, -0.02_31, -0.03_18, -0.00_59, 0.00_02, -0.00_20]       )

         assert np.abs(audio_slice.flatten() - expected_slice       ).max() < 1e-2






      def 	lowerCAmelCase__    (	self:						Tuple       )						->    Any:
         '''simple docstring'''

         UpperCAmelCase_											=torch_device

         UpperCAmelCase_											=DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"		,  torch_dtype=torch.floataa       )
         UpperCAmelCase_											=pipe.to(_lowerCAmelCase       )
         pipe.set_progress_bar_config(disable=_lowerCAmelCase       )

         UpperCAmelCase_											=torch.manual_seed(0       )
         UpperCAmelCase_											=pipe(generator=_lowerCAmelCase		,  num_inference_steps=100		,  audio_length_in_s=4.0_96       )
         UpperCAmelCase_											=output.audios

         UpperCAmelCase_											=audio[0, -3:, -3:]

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         UpperCAmelCase_											=np.array([-0.03_67, -0.04_88, -0.07_71, -0.05_25, -0.04_44, -0.03_41]       )

         assert np.abs(audio_slice.flatten() - expected_slice       ).max() < 1e-2




