

'''simple docstring'''






import inspect
import unittest

from huggingface_hub import hf_hub_download

from transformers import ASTConfig
from transformers.testing_utils import require_torch, require_torchaudio, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_torchaudio_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
 import torch
 from torch import nn

 from transformers import ASTForAudioClassification, ASTModel
 from transformers.models.audio_spectrogram_transformer.modeling_audio_spectrogram_transformer import (
     AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
 )


if is_torchaudio_available():
 import torchaudio

 from transformers import ASTFeatureExtractor






class     lowercase_			:




  """simple docstring"""




  def __init__(   self     :  Optional[int]      ,lowercase__     :  str      ,lowercase__     :  Optional[int]=1_3      ,lowercase__     :  Union[str, Any]=2      ,lowercase__     :  List[Any]=2_4      ,lowercase__     :  List[Any]=1_6      ,lowercase__     :  Any=True      ,lowercase__     :  Dict=True      ,lowercase__     :  int=3_2      ,lowercase__     :  Tuple=5      ,lowercase__     :  List[str]=4      ,lowercase__     :  Any=3_7      ,lowercase__     :  int="gelu"      ,lowercase__     :  List[str]=0.1      ,lowercase__     :  Optional[Any]=0.1      ,lowercase__     :  Tuple=1_0      ,lowercase__     :  str=0.0_2      ,lowercase__     :  str=None      ,lowercase__     :  List[str]=2      ,lowercase__     :  int=2      ,):
   __lowercase													=				parent
   __lowercase													=				batch_size
   __lowercase													=				patch_size
   __lowercase													=				max_length
   __lowercase													=				num_mel_bins
   __lowercase													=				is_training
   __lowercase													=				use_labels
   __lowercase													=				hidden_size
   __lowercase													=				num_hidden_layers
   __lowercase													=				num_attention_heads
   __lowercase													=				intermediate_size
   __lowercase													=				hidden_act
   __lowercase													=				hidden_dropout_prob
   __lowercase													=				attention_probs_dropout_prob
   __lowercase													=				type_sequence_label_size
   __lowercase													=				initializer_range
   __lowercase													=				scope
   __lowercase													=				frequency_stride
   __lowercase													=				time_stride

   # in AST, the seq length equals the number of patches + 2 (we add 2 for the [CLS] and distillation tokens)
   __lowercase													=				(self.num_mel_bins - self.patch_size) // self.frequency_stride + 1
   __lowercase													=				(self.max_length - self.patch_size) // self.time_stride + 1
   __lowercase													=				frequency_out_dimension * time_out_dimension
   __lowercase													=				num_patches + 2






  def       SCREAMING_SNAKE_CASE   (   self     :  List[str]    ):
   __lowercase													=				floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]    )

   __lowercase													=				None
   if self.use_labels:
    __lowercase													=				ids_tensor([self.batch_size]      ,self.type_sequence_label_size    )

   __lowercase													=				self.get_config()

   return config, input_values, labels






  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[int]    ):
   return ASTConfig(
       patch_size=self.patch_size      ,max_length=self.max_length      ,num_mel_bins=self.num_mel_bins      ,hidden_size=self.hidden_size      ,num_hidden_layers=self.num_hidden_layers      ,num_attention_heads=self.num_attention_heads      ,intermediate_size=self.intermediate_size      ,hidden_act=self.hidden_act      ,hidden_dropout_prob=self.hidden_dropout_prob      ,attention_probs_dropout_prob=self.attention_probs_dropout_prob      ,is_decoder=lowerCAmelCase_      ,initializer_range=self.initializer_range      ,frequency_stride=self.frequency_stride      ,time_stride=self.time_stride      ,)






  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[int]      ,lowercase__     :  List[str]      ,lowercase__     :  Any      ,lowercase__     :  Tuple    ):
   __lowercase													=				ASTModel(config=lowerCAmelCase_    )
   model.to(lowerCAmelCase_    )
   model.eval()
   __lowercase													=				model(lowerCAmelCase_    )
   self.parent.assertEqual(result.last_hidden_state.shape      ,(self.batch_size, self.seq_length, self.hidden_size)    )






  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[int]    ):
   __lowercase													=				self.prepare_config_and_inputs()
   (
       (
       __lowercase						
   )						,			(
       __lowercase						
   )						,			(
       __lowercase						
   )						,									
   )													=				config_and_inputs
   __lowercase													=				{'''input_values''': input_values}
   return config, inputs_dict






@require_torch
class     lowercase_			(__lowerCAmelCase       , __lowerCAmelCase       , unittest.TestCase   ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE					:						Any           =    (
      (
          ASTModel,
          ASTForAudioClassification,
      )
      if is_torch_available()
      else ()
  )
  SCREAMING_SNAKE_CASE					:						Dict           =    (
      {'''audio-classification''': ASTForAudioClassification, '''feature-extraction''': ASTModel}
      if is_torch_available()
      else {}
  )
  SCREAMING_SNAKE_CASE					:						Dict           =    False
  SCREAMING_SNAKE_CASE					:						str           =    False
  SCREAMING_SNAKE_CASE					:						Union[str, Any]           =    False
  SCREAMING_SNAKE_CASE					:						List[str]           =    False






  def       SCREAMING_SNAKE_CASE   (   self     :  Dict      ,lowercase__     :  str      ,lowercase__     :  Union[str, Any]      ,lowercase__     :  Dict      ,lowercase__     :  Dict      ,lowercase__     :  List[Any]    ):
   if pipeline_test_casse_name == "AudioClassificationPipelineTests":
    return True

   return False






  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[Any]    ):
   __lowercase													=				ASTModelTester(self    )
   __lowercase													=				ConfigTester(self      ,config_class=lowerCAmelCase_      ,has_text_modality=lowerCAmelCase_      ,hidden_size=3_7    )






  def       SCREAMING_SNAKE_CASE   (   self     :  Any    ):
   self.config_tester.run_common_tests()






  @unittest.skip(reason='''AST does not use inputs_embeds'''    )
  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[int]    ):
   pass






  def       SCREAMING_SNAKE_CASE   (   self     :  List[str]    ):
   __lowercase						,			__lowercase													=				self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
    __lowercase													=				model_class(lowerCAmelCase_    )
    self.assertIsInstance(model.get_input_embeddings()      ,(nn.Module)    )
    __lowercase													=				model.get_output_embeddings()
    self.assertTrue(x is None or isinstance(lowerCAmelCase_      ,nn.Linear    )    )






  def       SCREAMING_SNAKE_CASE   (   self     :  Union[str, Any]    ):
   __lowercase						,			__lowercase													=				self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
    __lowercase													=				model_class(lowerCAmelCase_    )
    __lowercase													=				inspect.signature(model.forward    )
    # signature.parameters is an OrderedDict => so arg_names order is deterministic
    __lowercase													=				[*signature.parameters.keys()]

    __lowercase													=				['''input_values''']
    self.assertListEqual(arg_names[:1]      ,lowerCAmelCase_    )






  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[int]    ):
   __lowercase													=				self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_model(*lowerCAmelCase_    )






  @slow
  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[Any]    ):
   for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
    __lowercase													=				ASTModel.from_pretrained(lowerCAmelCase_    )
    self.assertIsNotNone(lowerCAmelCase_    )


def 	_A  (					):






 """simple docstring"""



 __lowercase													=				hf_hub_download(
     repo_id='''nielsr/audio-spectogram-transformer-checkpoint'''  ,       filename='''sample_audio.flac'''  ,       repo_type='''dataset'''   )

 __lowercase						,			__lowercase													=				torchaudio.load(snake_case__   )

 return audio, sampling_rate






@require_torch
@require_torchaudio
class     lowercase_			(unittest.TestCase   ):




  """simple docstring"""




  @cached_property
  def       SCREAMING_SNAKE_CASE   (   self     :  List[Any]    ):
   return (
       ASTFeatureExtractor.from_pretrained('''MIT/ast-finetuned-audioset-10-10-0.4593'''    )
       if is_torchaudio_available()
       else None
   )






  @slow
  def       SCREAMING_SNAKE_CASE   (   self     :  int    ):
   __lowercase													=				self.default_feature_extractor
   __lowercase													=				ASTForAudioClassification.from_pretrained('''MIT/ast-finetuned-audioset-10-10-0.4593'''    ).to(lowerCAmelCase_    )

   __lowercase													=				self.default_feature_extractor
   __lowercase						,			__lowercase													=				prepare_audio()
   __lowercase													=				audio.squeeze().numpy()
   __lowercase													=				feature_extractor(lowerCAmelCase_      ,sampling_rate=lowerCAmelCase_      ,return_tensors='''pt'''    ).to(lowerCAmelCase_    )

   # forward pass
   with torch.no_grad():
    __lowercase													=				model(**lowerCAmelCase_    )

   # verify the logits
   __lowercase													=				torch.Size((1, 5_2_7)    )
   self.assertEqual(outputs.logits.shape      ,lowerCAmelCase_    )

   __lowercase													=				torch.tensor([-0.8_7_6_0, -7.0_0_4_2, -8.6_6_0_2]    ).to(lowerCAmelCase_    )

   self.assertTrue(torch.allclose(outputs.logits[0, :3]      ,lowerCAmelCase_      ,atol=1e-4    )    )




'''simple docstring'''






import argparse
import logging
from collections import namedtuple

import torch
from model_bertabs import BertAbsSummarizer
from models.model_builder import AbsSummarizer  # The authors' implementation

from transformers import BertTokenizer


logging.basicConfig(level=logging.INFO)
lowerCAmelCase__      		=			logging.getLogger(__name__)


lowerCAmelCase__      		=			'''Hello world! cécé herlolip'''


lowerCAmelCase__      		=			namedtuple(
    '''BertAbsConfig''',
    [
        '''temp_dir''',
        '''large''',
        '''use_bert_emb''',
        '''finetune_bert''',
        '''encoder''',
        '''share_emb''',
        '''max_pos''',
        '''enc_layers''',
        '''enc_hidden_size''',
        '''enc_heads''',
        '''enc_ff_size''',
        '''enc_dropout''',
        '''dec_layers''',
        '''dec_hidden_size''',
        '''dec_heads''',
        '''dec_ff_size''',
        '''dec_dropout''',
    ],
)


def 	_A  (					A__  ,       A__   ):






 """simple docstring"""



 __lowercase													=				BertAbsConfig(
     temp_dir='''.'''  ,       finetune_bert=A__  ,       large=A__  ,       share_emb=A__  ,       use_bert_emb=A__  ,       encoder='''bert'''  ,       max_pos=512  ,       enc_layers=6  ,       enc_hidden_size=512  ,       enc_heads=8  ,       enc_ff_size=512  ,       enc_dropout=0.2  ,       dec_layers=6  ,       dec_hidden_size=768  ,       dec_heads=8  ,       dec_ff_size=2048  ,       dec_dropout=0.2  ,       )
 __lowercase													=				torch.load(A__  ,       lambda A__  ,       A__   : storage   )
 __lowercase													=				AbsSummarizer(A__  ,       torch.device('''cpu'''   )  ,       A__   )
 original.eval()

 __lowercase													=				BertAbsSummarizer(A__  ,       torch.device('''cpu'''   )   )
 new_model.eval()

 # -------------------
 # Convert the weights
 # -------------------

 logging.info('''convert the model'''   )
 new_model.bert.load_state_dict(original.bert.state_dict()   )
 new_model.decoder.load_state_dict(original.decoder.state_dict()   )
 new_model.generator.load_state_dict(original.generator.state_dict()   )

 # ----------------------------------
 # Make sure the outpus are identical
 # ----------------------------------

 logging.info('''Make sure that the models\' outputs are identical'''   )
 __lowercase													=				BertTokenizer.from_pretrained('''bert-base-uncased'''   )

 # prepare the model inputs
 __lowercase													=				tokenizer.encode('''This is sample éàalj\'-.'''   )
 encoder_input_ids.extend([tokenizer.pad_token_id] * (512 - len(A__   ))   )
 __lowercase													=				torch.tensor(A__   ).unsqueeze(0   )
 __lowercase													=				tokenizer.encode('''This is sample 3 éàalj\'-.'''   )
 decoder_input_ids.extend([tokenizer.pad_token_id] * (512 - len(A__   ))   )
 __lowercase													=				torch.tensor(A__   ).unsqueeze(0   )

 # failsafe to make sure the weights reset does not affect the
 # loaded weights.
 assert torch.max(torch.abs(original.generator[0].weight - new_model.generator[0].weight   )   ) == 0

 # forward pass
 __lowercase													=				encoder_input_ids
 __lowercase													=				decoder_input_ids
 __lowercase													=				__lowercase													=				None
 __lowercase													=				None
 __lowercase													=				__lowercase													=				None
 __lowercase													=				__lowercase													=				None
 __lowercase													=				None

 # The original model does not apply the geneator layer immediatly but rather in
 # the beam search (where it combines softmax + linear layer). Since we already
 # apply the softmax in our generation process we only apply the linear layer here.
 # We make sure that the outputs of the full stack are identical
 __lowercase													=				original(A__  ,       A__  ,       A__  ,       A__  ,       A__  ,       A__  ,       A__   )[0]
 __lowercase													=				original.generator(A__   )

 __lowercase													=				new_model(
     A__  ,       A__  ,       A__  ,       A__  ,       A__   )[0]
 __lowercase													=				new_model.generator(A__   )

 __lowercase													=				torch.max(torch.abs(output_converted_model - output_original_model   )   ).item()
 print('''Maximum absolute difference beween weights: {:.2f}'''.format(A__   )   )
 __lowercase													=				torch.max(torch.abs(output_converted_generator - output_original_generator   )   ).item()
 print('''Maximum absolute difference beween weights: {:.2f}'''.format(A__   )   )

 __lowercase													=				torch.allclose(A__  ,       A__  ,       atol=1e-3   )
 if are_identical:
  logging.info('''all weights are equal up to 1e-3'''   )
 else:
  raise ValueError('''the weights are different. The new model is likely different from the original one.'''   )

 # The model has been saved with torch.save(model) and this is bound to the exact
 # directory structure. We save the state_dict instead.
 logging.info('''saving the model\'s state dictionary'''   )
 torch.save(
     new_model.state_dict()  ,       '''./bertabs-finetuned-cnndm-extractive-abstractive-summarization/pytorch_model.bin'''   )


if __name__ == "__main__":
 lowerCAmelCase__      		=			argparse.ArgumentParser()
 parser.add_argument(
     '''--bertabs_checkpoint_path''',
     default=None,
     type=str,
     required=True,
     help='''Path the official PyTorch dump.''',
 )
 parser.add_argument(
     '''--pytorch_dump_folder_path''',
     default=None,
     type=str,
     required=True,
     help='''Path to the output PyTorch model.''',
 )
 lowerCAmelCase__      		=			parser.parse_args()

 convert_bertabs_checkpoints(
     args.bertabs_checkpoint_path,
     args.pytorch_dump_folder_path,
 )

