
import torch
from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.bert.modeling_bert import (
    BERT_INPUTS_DOCSTRING,
    BERT_START_DOCSTRING,
    BertEmbeddings,
    BertLayer,
    BertPooler,
    BertPreTrainedModel,
)


def 							a__					(							snake_case							):





  """simple docstring"""
  __SCREAMING_SNAKE_CASE      :	List[str]					=		torch.exp(_UpperCAmelCase							)
  __SCREAMING_SNAKE_CASE      :	str					=		torch.sum(_UpperCAmelCase       ,			dim=1							)  # sum of exp(x_i)
  __SCREAMING_SNAKE_CASE      :	int					=		torch.sum(x * exp_x       ,			dim=1							)  # sum of x_i * exp(x_i)
  return torch.log(_UpperCAmelCase							) - B / A





class     __UpperCamelCase (							nn.Module       ):






     """simple docstring"""







     def __init__(				self							:							int     ,						_A							:							Optional[int] ):







       """simple docstring"""




       super().__init__()
       __SCREAMING_SNAKE_CASE      :	str					=		config.output_attentions
       __SCREAMING_SNAKE_CASE      :	Any					=		config.output_hidden_states
       __SCREAMING_SNAKE_CASE      :	List[Any]					=		nn.ModuleList([BertLayer(_snake_case ) for _ in range(config.num_hidden_layers )] )
       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		nn.ModuleList([BertHighway(_snake_case ) for _ in range(config.num_hidden_layers )] )

       __SCREAMING_SNAKE_CASE      :	Any					=		[-1 for _ in range(config.num_hidden_layers )]



     def  UpperCAmelCase__							(				self							:							List[str]     ,						_A							:							Optional[int] ):







       """simple docstring"""




       if (type(_snake_case ) is float) or (type(_snake_case ) is int):
         for i in range(len(self.early_exit_entropy ) ):
           __SCREAMING_SNAKE_CASE      :	Any					=		x
       else:
         __SCREAMING_SNAKE_CASE      :	Tuple					=		x



     def  UpperCAmelCase__							(				self							:							str     ,						_A							:							List[str] ):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE      :	Dict					=		pooler.state_dict()
       for highway in self.highway:
         for name, param in highway.pooler.state_dict().items():
           param.copy_(loaded_model[name] )




     def  UpperCAmelCase__							(				self							:							List[Any]     ,						_A							:							Any     ,						_A							:							Union[str, Any]=None     ,						_A							:							Optional[int]=None     ,						_A							:							List[Any]=None     ,						_A							:							int=None     ,						):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE      :	Dict					=		()
       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		()
       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		()
       for i, layer_module in enumerate(self.layer ):
         if self.output_hidden_states:
           __SCREAMING_SNAKE_CASE      :	Dict					=		all_hidden_states + (hidden_states,)

         __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		layer_module(
             _snake_case     ,						_snake_case     ,						head_mask[i]     ,						_snake_case     ,						_snake_case )
         __SCREAMING_SNAKE_CASE      :	Any					=		layer_outputs[0]

         if self.output_attentions:
           __SCREAMING_SNAKE_CASE      :	Tuple					=		all_attentions + (layer_outputs[1],)

         __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		(hidden_states,)
         if self.output_hidden_states:
           __SCREAMING_SNAKE_CASE      :	Tuple					=		current_outputs + (all_hidden_states,)
         if self.output_attentions:
           __SCREAMING_SNAKE_CASE      :	str					=		current_outputs + (all_attentions,)

         __SCREAMING_SNAKE_CASE      :	str					=		self.highway[i](_snake_case )
         # logits, pooled_output

         if not self.training:
           __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		highway_exit[0]
           __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		entropy(_snake_case )
           __SCREAMING_SNAKE_CASE      :	str					=		highway_exit + (highway_entropy,)  # logits, hidden_states(?), entropy
           __SCREAMING_SNAKE_CASE      :	str					=		all_highway_exits + (highway_exit,)

           if highway_entropy < self.early_exit_entropy[i]:
             __SCREAMING_SNAKE_CASE      :	Any					=		(highway_logits,) + current_outputs[1:] + (all_highway_exits,)
             raise HighwayException(_snake_case     ,						i + 1 )
         else:
           __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		all_highway_exits + (highway_exit,)

               # Add last layer
       if self.output_hidden_states:
         __SCREAMING_SNAKE_CASE      :	List[str]					=		all_hidden_states + (hidden_states,)

       __SCREAMING_SNAKE_CASE      :	Any					=		(hidden_states,)
       if self.output_hidden_states:
         __SCREAMING_SNAKE_CASE      :	Any					=		outputs + (all_hidden_states,)
       if self.output_attentions:
         __SCREAMING_SNAKE_CASE      :	List[str]					=		outputs + (all_attentions,)

       __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		outputs + (all_highway_exits,)
       return outputs  # last-layer hidden state, (all hidden states), (all attentions), all highway exits






@add_start_docstrings(
    '''The Bert Model transformer with early exiting (DeeBERT). ''' ,						a__ ,						)
class     __UpperCamelCase (							a__       ):






     """simple docstring"""







     def __init__(				self							:							str     ,						_A							:							List[Any] ):







       """simple docstring"""




       super().__init__(_snake_case )
       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		config

       __SCREAMING_SNAKE_CASE      :	int					=		BertEmbeddings(_snake_case )
       __SCREAMING_SNAKE_CASE      :	Dict					=		DeeBertEncoder(_snake_case )
       __SCREAMING_SNAKE_CASE      :	Tuple					=		BertPooler(_snake_case )

       self.init_weights()



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







       """simple docstring"""




       self.encoder.init_highway_pooler(self.pooler )



     def  UpperCAmelCase__							(				self							:							List[Any] ):







       """simple docstring"""




       return self.embeddings.word_embeddings



     def  UpperCAmelCase__							(				self							:							int     ,						_A							:							Tuple ):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE      :	Any					=		value



     def  UpperCAmelCase__							(				self							:							Tuple     ,						_A							:							List[str] ):







       """simple docstring"""




       for layer, heads in heads_to_prune.items():
         self.encoder.layer[layer].attention.prune_heads(_snake_case )




     @add_start_docstrings_to_model_forward(_snake_case )
     def  UpperCAmelCase__							(				self							:							Union[str, Any]     ,						_A							:							Optional[int]=None     ,						_A							:							List[str]=None     ,						_A							:							Any=None     ,						_A							:							Any=None     ,						_A							:							Optional[Any]=None     ,						_A							:							Optional[Any]=None     ,						_A							:							List[str]=None     ,						_A							:							int=None     ,						):







       """simple docstring"""




       if input_ids is not None and inputs_embeds is not None:
         raise ValueError('''You cannot specify both input_ids and inputs_embeds at the same time''' )
       elif input_ids is not None:
         __SCREAMING_SNAKE_CASE      :	Dict					=		input_ids.size()
       elif inputs_embeds is not None:
         __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		inputs_embeds.size()[:-1]
       else:
         raise ValueError('''You have to specify either input_ids or inputs_embeds''' )

       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		input_ids.device if input_ids is not None else inputs_embeds.device

       if attention_mask is None:
         __SCREAMING_SNAKE_CASE      :	Dict					=		torch.ones(_snake_case     ,						device=_snake_case )
       if encoder_attention_mask is None:
         __SCREAMING_SNAKE_CASE      :	str					=		torch.ones(_snake_case     ,						device=_snake_case )
       if token_type_ids is None:
         __SCREAMING_SNAKE_CASE      :	Optional[int]					=		torch.zeros(_snake_case     ,						dtype=torch.long     ,						device=_snake_case )

       # We can provide a self-attention mask of dimensions [batch_size, from_seq_length, to_seq_length]
       # ourselves in which case we just need to make it broadcastable to all heads.
       __SCREAMING_SNAKE_CASE      :	int					=		self.get_extended_attention_mask(_snake_case     ,						_snake_case     ,						_snake_case )

       # If a 2D ou 3D attention mask is provided for the cross-attention
       # we need to make broadcastable to [batch_size, num_heads, seq_length, seq_length]
       if encoder_attention_mask.dim() == 3:
         __SCREAMING_SNAKE_CASE      :	Dict					=		encoder_attention_mask[:, None, :, :]
       if encoder_attention_mask.dim() == 2:
         __SCREAMING_SNAKE_CASE      :	int					=		encoder_attention_mask[:, None, None, :]

       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		encoder_extended_attention_mask.to(
           dtype=next(self.parameters() ).dtype )  # fp16 compatibility
       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		(1.0 - encoder_extended_attention_mask) * -1_0000.0

       # Prepare head mask if needed
       # 1.0 in head_mask indicate we keep the head
       # attention_probs has shape bsz x n_heads x N x N
       # input head_mask has shape [num_heads] or [num_hidden_layers x num_heads]
       # and head_mask is converted to shape [num_hidden_layers x batch x num_heads x seq_length x seq_length]
       __SCREAMING_SNAKE_CASE      :	Any					=		self.get_head_mask(_snake_case     ,						self.config.num_hidden_layers )

       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		self.embeddings(
           input_ids=_snake_case     ,						position_ids=_snake_case     ,						token_type_ids=_snake_case     ,						inputs_embeds=_snake_case )
       __SCREAMING_SNAKE_CASE      :	List[str]					=		self.encoder(
           _snake_case     ,						attention_mask=_snake_case     ,						head_mask=_snake_case     ,						encoder_hidden_states=_snake_case     ,						encoder_attention_mask=_snake_case     ,						)
       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		encoder_outputs[0]
       __SCREAMING_SNAKE_CASE      :	List[Any]					=		self.pooler(_snake_case )

       __SCREAMING_SNAKE_CASE      :	Dict					=		(
           sequence_output,
           pooled_output,
       ) + encoder_outputs[
           1:
       ]  # add hidden_states and attentions if they are here
       return outputs  # sequence_output, pooled_output, (hidden_states), (attentions), highway exits






class     __UpperCamelCase (							a__       ):






     """simple docstring"""







     def __init__(				self							:							List[Any]     ,						_A							:							Dict     ,						_A							:							str ):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		message
       __SCREAMING_SNAKE_CASE      :	Tuple					=		exit_layer  # start from 1!






class     __UpperCamelCase (							nn.Module       ):






     """simple docstring"""







     def __init__(				self							:							Any     ,						_A							:							List[str] ):







       """simple docstring"""




       super().__init__()
       __SCREAMING_SNAKE_CASE      :	Optional[int]					=		BertPooler(_snake_case )
       __SCREAMING_SNAKE_CASE      :	Dict					=		nn.Dropout(config.hidden_dropout_prob )
       __SCREAMING_SNAKE_CASE      :	Dict					=		nn.Linear(config.hidden_size     ,						config.num_labels )




     def  UpperCAmelCase__							(				self							:							Any     ,						_A							:							List[Any] ):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE      :	Tuple					=		encoder_outputs[0]
       __SCREAMING_SNAKE_CASE      :	int					=		self.pooler(_snake_case )
       # "return" pooler_output

       # BertModel
       __SCREAMING_SNAKE_CASE      :	Dict					=		(pooler_input, pooler_output) + encoder_outputs[1:]
       # "return" bmodel_output

       # Dropout and classification
       __SCREAMING_SNAKE_CASE      :	str					=		bmodel_output[1]

       __SCREAMING_SNAKE_CASE      :	List[str]					=		self.dropout(_snake_case )
       __SCREAMING_SNAKE_CASE      :	List[str]					=		self.classifier(_snake_case )

       return logits, pooled_output






@add_start_docstrings(
    '''Bert Model (with early exiting - DeeBERT) with a classifier on top,
    also takes care of multi-layer training. ''' ,						a__ ,						)
class     __UpperCamelCase (							a__       ):






     """simple docstring"""







     def __init__(				self							:							Optional[int]     ,						_A							:							Any ):







       """simple docstring"""




       super().__init__(_snake_case )
       __SCREAMING_SNAKE_CASE      :	Dict					=		config.num_labels
       __SCREAMING_SNAKE_CASE      :	int					=		config.num_hidden_layers

       __SCREAMING_SNAKE_CASE      :	List[Any]					=		DeeBertModel(_snake_case )
       __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		nn.Dropout(config.hidden_dropout_prob )
       __SCREAMING_SNAKE_CASE      :	Any					=		nn.Linear(config.hidden_size     ,						self.config.num_labels )

       self.init_weights()




     @add_start_docstrings_to_model_forward(_snake_case )
     def  UpperCAmelCase__							(				self							:							List[str]     ,						_A							:							List[Any]=None     ,						_A							:							Union[str, Any]=None     ,						_A							:							Dict=None     ,						_A							:							Optional[Any]=None     ,						_A							:							Optional[Any]=None     ,						_A							:							List[Any]=None     ,						_A							:							Union[str, Any]=None     ,						_A							:							Optional[int]=-1     ,						_A							:							int=False     ,						):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE      :	Any					=		self.num_layers
       try:
         __SCREAMING_SNAKE_CASE      :	List[Any]					=		self.bert(
             _snake_case     ,						attention_mask=_snake_case     ,						token_type_ids=_snake_case     ,						position_ids=_snake_case     ,						head_mask=_snake_case     ,						inputs_embeds=_snake_case     ,						)
         # sequence_output, pooled_output, (hidden_states), (attentions), highway exits

         __SCREAMING_SNAKE_CASE      :	str					=		outputs[1]

         __SCREAMING_SNAKE_CASE      :	Tuple					=		self.dropout(_snake_case )
         __SCREAMING_SNAKE_CASE      :	Optional[int]					=		self.classifier(_snake_case )
         __SCREAMING_SNAKE_CASE      :	str					=		(logits,) + outputs[2:]  # add hidden states and attention if they are here
       except HighwayException as e:
         __SCREAMING_SNAKE_CASE      :	Tuple					=		e.message
         __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		e.exit_layer
         __SCREAMING_SNAKE_CASE      :	List[str]					=		outputs[0]

       if not self.training:
         __SCREAMING_SNAKE_CASE      :	Optional[int]					=		entropy(_snake_case )
         __SCREAMING_SNAKE_CASE      :	List[str]					=		[]
         __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		[]
       if labels is not None:
         if self.num_labels == 1:
           #  We are doing regression
           __SCREAMING_SNAKE_CASE      :	List[str]					=		MSELoss()
           __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		loss_fct(logits.view(-1 )     ,						labels.view(-1 ) )
         else:
           __SCREAMING_SNAKE_CASE      :	int					=		CrossEntropyLoss()
           __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		loss_fct(logits.view(-1     ,						self.num_labels )     ,						labels.view(-1 ) )

         # work with highway exits
         __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		[]
         for highway_exit in outputs[-1]:
           __SCREAMING_SNAKE_CASE      :	Tuple					=		highway_exit[0]
           if not self.training:
             highway_logits_all.append(_snake_case )
             highway_entropy.append(highway_exit[2] )
           if self.num_labels == 1:
             #  We are doing regression
             __SCREAMING_SNAKE_CASE      :	Union[str, Any]					=		MSELoss()
             __SCREAMING_SNAKE_CASE      :	Optional[int]					=		loss_fct(highway_logits.view(-1 )     ,						labels.view(-1 ) )
           else:
             __SCREAMING_SNAKE_CASE      :	str					=		CrossEntropyLoss()
             __SCREAMING_SNAKE_CASE      :	Dict					=		loss_fct(highway_logits.view(-1     ,						self.num_labels )     ,						labels.view(-1 ) )
           highway_losses.append(_snake_case )

         if train_highway:
           __SCREAMING_SNAKE_CASE      :	str					=		(sum(highway_losses[:-1] ),) + outputs
           # exclude the final highway, of course
         else:
           __SCREAMING_SNAKE_CASE      :	str					=		(loss,) + outputs
       if not self.training:
         __SCREAMING_SNAKE_CASE      :	Dict					=		outputs + ((original_entropy, highway_entropy), exit_layer)
         if output_layer >= 0:
           __SCREAMING_SNAKE_CASE      :	Optional[Any]					=		(
               (outputs[0],) + (highway_logits_all[output_layer],) + outputs[2:]
           )  # use the highway of the last layer

       return outputs  # (loss), logits, (hidden_states), (attentions), (highway_exits)







"""simple docstring"""
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							(   a__   , 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    UpperCamelCase__  (     self     ):







  """simple docstring"""






  torch.manual_seed(0     )
  lowerCAmelCase    				= UNetaDModel(
      block_out_channels=(32, 32, 64)				,       extra_in_channels=16				,       sample_size=5_12				,       sample_rate=1_60_00				,       in_channels=2				,       out_channels=2				,       flip_sin_to_cos=_snake_case				,       use_timestep_embedding=_snake_case				,       time_embedding_type='fourier'				,       mid_block_type='UNetMidBlock1D'				,       down_block_types=('DownBlock1DNoSkip', 'DownBlock1D', 'AttnDownBlock1D')				,       up_block_types=('AttnUpBlock1D', 'UpBlock1D', 'UpBlock1DNoSkip')				,       )
  lowerCAmelCase    				= IPNDMScheduler()

  lowerCAmelCase    				= {
      'unet': unet,
      'scheduler': scheduler,
  }
  return components



 def    UpperCamelCase__  (     self				,       _snake_case				,       _snake_case=0     ):







  """simple docstring"""






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



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  lowerCAmelCase    				= 'cpu'  # ensure determinism for the device-dependent torch.Generator
  lowerCAmelCase    				= self.get_dummy_components()
  lowerCAmelCase    				= DanceDiffusionPipeline(**_snake_case     )
  lowerCAmelCase    				= pipe.to(_snake_case     )
  pipe.set_progress_bar_config(disable=_snake_case     )

  lowerCAmelCase    				= self.get_dummy_inputs(_snake_case     )
  lowerCAmelCase    				= pipe(**_snake_case     )
  lowerCAmelCase    				= output.audios

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

  assert audio.shape == (1, 2, components["unet"].sample_size)
  lowerCAmelCase    				= np.array([-0.7_265, 1.0_000, -0.8_388, 0.1_175, 0.9_498, -1.0_000]     )
  assert np.abs(audio_slice.flatten() - expected_slice     ).max() < 1E-2



 @skip_mps
 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  return super().test_save_load_local()



 @skip_mps
 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






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



 @skip_mps
 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  return super().test_save_load_optional_components()



 @skip_mps
 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  return super().test_attention_slicing_forward_pass()



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






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







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



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






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



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  lowerCAmelCase    				= torch_device

  lowerCAmelCase    				= DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'     )
  lowerCAmelCase    				= pipe.to(_snake_case     )
  pipe.set_progress_bar_config(disable=_snake_case     )

  lowerCAmelCase    				= torch.manual_seed(0     )
  lowerCAmelCase    				= pipe(generator=_snake_case				,       num_inference_steps=1_00				,       audio_length_in_s=4.096     )
  lowerCAmelCase    				= output.audios

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

  assert audio.shape == (1, 2, pipe.unet.sample_size)
  lowerCAmelCase    				= np.array([-0.0_192, -0.0_231, -0.0_318, -0.0_059, 0.0_002, -0.0_020]     )

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



 def    UpperCamelCase__  (     self     ):







  """simple docstring"""






  lowerCAmelCase    				= torch_device

  lowerCAmelCase    				= DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'				,       torch_dtype=torch.floataa     )
  lowerCAmelCase    				= pipe.to(_snake_case     )
  pipe.set_progress_bar_config(disable=_snake_case     )

  lowerCAmelCase    				= torch.manual_seed(0     )
  lowerCAmelCase    				= pipe(generator=_snake_case				,       num_inference_steps=1_00				,       audio_length_in_s=4.096     )
  lowerCAmelCase    				= output.audios

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

  assert audio.shape == (1, 2, pipe.unet.sample_size)
  lowerCAmelCase    				= np.array([-0.0_367, -0.0_488, -0.0_771, -0.0_525, -0.0_444, -0.0_341]     )

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





