






'''simple docstring'''
from __future__ import annotations

import os
import tempfile
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import is_tensorflow_text_available, is_tf_available
from transformers.testing_utils import require_tensorflow_text, require_tf, slow

from ..test_modeling_tf_common import floats_tensor
from .test_framework_agnostic import GenerationIntegrationTestsMixin


if is_tf_available():
 import tensorflow as tf

 from transformers import (
     AutoTokenizer,
     TFAutoModelForCausalLM,
     TFAutoModelForSeqaSeqLM,
     TFAutoModelForSpeechSeqaSeq,
     TFAutoModelForVisionaSeq,
     TFBartForConditionalGeneration,
     TFLogitsProcessorList,
     TFMinLengthLogitsProcessor,
     tf_top_k_top_p_filtering,
 )

if is_tensorflow_text_available():
 import tensorflow_text as text


@require_tf
class 							_UpperCAmelCase    (		unittest.TestCase      ):




    def 					lowerCAmelCase__					(      self	:							Optional[int]      ):






       '''simple docstring'''






       lowercase_						:     Union[str, Any]							     =  tf.convert_to_tensor(
           [
               [
                   8.222_0991,  # 3rd highest value; idx. 0
                   -0.562_0044,
                   5.2322_9752,
                   4.038_6393,
                   -6.879_8378,
                   -0.5478_5802,
                   -3.201_2153,
                   2.9277_7176,
                   1.8817_1953,
                   7.3534_1276,  # 5th highest value; idx. 9
                   8.4320_7833,  # 2nd highest value; idx. 10
                   -9.8571_1836,
                   -5.9620_9236,
                   -1.1303_9161,
                   -7.111_5294,
                   -0.836_9633,
                   -5.318_6408,
                   7.0642_7407,
                   0.8136_9344,
                   -0.8202_3817,
                   -5.917_9796,
                   0.5881_3443,
                   -6.9977_8438,
                   4.7155_1189,
                   -0.1877_1637,
                   7.4402_0759,  # 4th highest value; idx. 25
                   9.3845_0987,  # 1st highest value; idx. 26
                   2.1266_2941,
                   -9.3256_2038,
                   2.3565_2522,
               ],  # cummulative prob of 5 highest values <= 0.6
               [
                   0.5842_5518,
                   4.5313_9238,
                   -5.5751_0464,
                   -6.2803_0699,
                   -7.1952_9503,
                   -4.0212_2551,
                   1.3933_7037,
                   -6.0670_7057,
                   1.5948_0517,
                   -9.64_3119,
                   0.0390_7799,
                   0.6723_1762,
                   -8.8820_6726,
                   6.2711_5922,  # 4th highest value; idx. 13
                   2.2852_0723,
                   4.8276_7506,
                   4.3042_1368,
                   8.827_5313,  # 2nd highest value; idx. 17
                   5.4402_9958,  # 5th highest value; idx. 18
                   -4.473_5794,
                   7.3857_9536,  # 3rd highest value; idx. 20
                   -2.9105_1663,
                   2.6194_6077,
                   -2.567_4762,
                   -9.4895_9302,
                   -4.0292_2645,
                   -1.3541_6918,
                   9.6770_2323,  # 1st highest value; idx. 27
                   -5.8947_8553,
                   1.8537_0467,
               ],  # cummulative prob of 5 highest values <= 0.6
           ]	,  dtype=tf.floataa	,  )

       lowercase_						:     List[Any]							     =  tf.convert_to_tensor(
           [[0, 0], [0, 9], [0, 1_0], [0, 2_5], [0, 2_6], [1, 1_3], [1, 1_7], [1, 1_8], [1, 2_0], [1, 2_7]]	,  dtype=tf.intaa	,  )  # expected non filtered idx as noted above

       lowercase_						:     Tuple							     =  tf.convert_to_tensor(
           [8.22_2099, 7.353_4126, 8.43_2078, 7.440_2075, 9.3_8451, 6.27_1159, 8.82_7531, 5.440_2995, 7.385_7956, 9.67_7023]	,  dtype=tf.floataa	,  )  # expected non filtered values as noted above

       lowercase_						:     Dict							     =  tf_top_k_top_p_filtering(a	,  top_k=1_0	,  top_p=0.6	,  min_tokens_to_keep=4      )

       lowercase_						:     Dict							     =  output[output != -float("inf"      )]
       lowercase_						:     Optional[int]							     =  tf.cast(
           tf.where(tf.not_equal(a	,  tf.constant(-float("inf"      )	,  dtype=tf.floataa      )      )      )	,  dtype=tf.intaa	,  )

       tf.debugging.assert_near(a	,  a	,  rtol=1e-12      )
       tf.debugging.assert_equal(a	,  a      )


@require_tf
class 							_UpperCAmelCase    (		unittest.TestCase					,       snake_case      ):
    # setting framework_dependent_parameters needs to be gated, just like its contents' imports
    if is_tf_available():
        __lowerCamelCase: Optional[int]									=     {
            'AutoModelForCausalLM': TFAutoModelForCausalLM,
            'AutoModelForSpeechSeq2Seq': TFAutoModelForSpeechSeqaSeq,
            'AutoModelForSeq2SeqLM': TFAutoModelForSeqaSeqLM,
            'AutoModelForVision2Seq': TFAutoModelForVisionaSeq,
            'LogitsProcessorList': TFLogitsProcessorList,
            'MinLengthLogitsProcessor': TFMinLengthLogitsProcessor,
            'create_tensor_fn': tf.convert_to_tensor,
            'floats_tensor': floats_tensor,
            'return_tensors': 'tf',
        }




    @slow
    def 					lowerCAmelCase__					(      self	:							Optional[int]      ):






       '''simple docstring'''






       lowercase_						:     Any							     =  TFAutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2"      )
       lowercase_						:     Dict							     =  2
       lowercase_						:     List[str]							     =  2


       class 							_UpperCAmelCase    (		tf.Module      ):




           def __init__(      self	:							Any	,  a	:							Optional[Any]      ):






              '''simple docstring'''






              super(a	,  self      ).__init__()
              lowercase_						:     Optional[int]							     =  model





           @tf.function(
               input_signature=(
                   tf.TensorSpec((None, input_length)	,  tf.intaa	,  name="input_ids"      ),
                   tf.TensorSpec((None, input_length)	,  tf.intaa	,  name="attention_mask"      ),
               )	,  jit_compile=a	,  )
           def 					lowerCAmelCase__					(      self	:							int	,  a	:							List[str]	,  a	:							List[str]      ):






              '''simple docstring'''






              lowercase_						:     Dict							     =  self.model.generate(
                  input_ids=a	,  attention_mask=a	,  max_new_tokens=a	,  return_dict_in_generate=a	,  )
              return {"sequences": outputs["sequences"]}


       lowercase_						:     List[str]							     =  [[2, 0], [1_0_2, 1_0_3]]
       lowercase_						:     Tuple							     =  [[1, 0], [1, 1]]
       lowercase_						:     List[Any]							     =  DummyModel(model=a      )
       with tempfile.TemporaryDirectory() as tmp_dir:
          tf.saved_model.save(a	,  a	,  signatures={"serving_default": dummy_model.serving}      )
          lowercase_						:     Optional[int]							     =  tf.saved_model.load(a      ).signatures["serving_default"]
          for batch_size in range(1	,  len(a      ) + 1      ):
             lowercase_						:     int							     =  {
                 "input_ids": tf.constant(dummy_input_ids[:batch_size]      ),
                 "attention_mask": tf.constant(dummy_attention_masks[:batch_size]      ),
             }
             lowercase_						:     List[Any]							     =  serving_func(**a      )["sequences"]
             lowercase_						:     Any							     =  test_model.generate(**a	,  max_new_tokens=a      )
             tf.debugging.assert_equal(a	,  a      )




    @slow
    def 					lowerCAmelCase__					(      self	:							str      ):






       '''simple docstring'''






       lowercase_						:     Union[str, Any]							     =  TFAutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2"      )
       lowercase_						:     int							     =  1
       lowercase_						:     Any							     =  2


       class 							_UpperCAmelCase    (		tf.Module      ):




           def __init__(      self	:							int	,  a	:							str      ):






              '''simple docstring'''






              super(a	,  self      ).__init__()
              lowercase_						:     str							     =  model





           @tf.function(
               input_signature=(
                   tf.TensorSpec((batch_size, None)	,  tf.intaa	,  name="input_ids"      ),
                   tf.TensorSpec((batch_size, None)	,  tf.intaa	,  name="attention_mask"      ),
               )	,  jit_compile=a	,  )
           def 					lowerCAmelCase__					(      self	:							List[str]	,  a	:							Union[str, Any]	,  a	:							Any      ):






              '''simple docstring'''






              lowercase_						:     Optional[Any]							     =  self.model.generate(
                  input_ids=a	,  attention_mask=a	,  max_new_tokens=a	,  return_dict_in_generate=a	,  )
              return {"sequences": outputs["sequences"]}


       lowercase_						:     str							     =  [[2], [1_0_2, 1_0_3]]
       lowercase_						:     List[str]							     =  [[1], [1, 1]]
       lowercase_						:     Optional[Any]							     =  DummyModel(model=a      )
       with tempfile.TemporaryDirectory() as tmp_dir:
          tf.saved_model.save(a	,  a	,  signatures={"serving_default": dummy_model.serving}      )
          lowercase_						:     str							     =  tf.saved_model.load(a      ).signatures["serving_default"]
          for input_row in range(len(a      )      ):
             lowercase_						:     Tuple							     =  {
                 "input_ids": tf.constant([dummy_input_ids[input_row]]      ),
                 "attention_mask": tf.constant([dummy_attention_masks[input_row]]      ),
             }
             lowercase_						:     List[str]							     =  serving_func(**a      )["sequences"]
             lowercase_						:     List[Any]							     =  test_model.generate(**a	,  max_new_tokens=a      )
             tf.debugging.assert_equal(a	,  a      )




    @slow
    @require_tensorflow_text
    def 					lowerCAmelCase__					(      self	:							Tuple      ):






       '''simple docstring'''






       with tempfile.TemporaryDirectory() as tmp_dir:
          # file needed to load the TF tokenizer
          hf_hub_download(repo_id="google/flan-t5-small"	,  filename="spiece.model"	,  local_dir=a      )


          class 							_UpperCAmelCase    (		tf.keras.layers.Layer      ):




              def __init__(      self	:							Dict      ):






                 '''simple docstring'''






                 super().__init__()
                 lowercase_						:     Tuple							     =  text.SentencepieceTokenizer(
                     model=tf.io.gfile.GFile(os.path.join(a	,  "spiece.model"      )	,  "rb"      ).read()      )
                 lowercase_						:     List[Any]							     =  TFAutoModelForSeqaSeqLM.from_pretrained("hf-internal-testing/tiny-random-t5"      )





              def 					lowerCAmelCase__					(      self	:							Union[str, Any]	,  a	:							Tuple	,  *a	:							List[Any]	,  **a	:							str      ):






                 '''simple docstring'''






                 lowercase_						:     List[Any]							     =  self.tokenizer.tokenize(a      )
                 lowercase_     ,		lowercase_						:     int							     =  text.pad_model_inputs(
                     a	,  max_seq_length=6_4	,  pad_value=self.model.config.pad_token_id      )
                 lowercase_						:     Tuple							     =  self.model.generate(input_ids=a	,  attention_mask=a      )
                 return self.tokenizer.detokenize(a      )


          lowercase_						:     Optional[Any]							     =  CompleteSentenceTransformer()
          lowercase_						:     Optional[Any]							     =  tf.keras.layers.Input(shape=(1,)	,  dtype=tf.string	,  name="inputs"      )
          lowercase_						:     str							     =  complete_model(a      )
          lowercase_						:     Any							     =  tf.keras.Model(a	,  a      )
          keras_model.save(a      )




    def 					lowerCAmelCase__					(      self	:							Tuple      ):






       '''simple docstring'''






       lowercase_						:     Tuple							     =  {
           "do_sample": True,
           "num_beams": 1,
           "top_p": 0.7,
           "top_k": 1_0,
           "temperature": 0.7,
       }
       lowercase_						:     List[str]							     =  1_4

       lowercase_						:     Optional[Any]							     =  AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"      )
       lowercase_						:     List[str]							     =  "Hello, my dog is cute and"
       lowercase_						:     List[str]							     =  tokenizer(a	,  return_tensors="tf"      )
       lowercase_						:     Any							     =  TFAutoModelForCausalLM.from_pretrained("hf-internal-testing/tiny-random-gpt2"      )

       lowercase_						:     Dict							     =  6_3_8
       # forces the generation to happen on CPU, to avoid GPU-related quirks
       with tf.device(":/CPU:0"      ):
          tf.random.set_seed(0      )
          lowercase_						:     Union[str, Any]							     =  model.generate(**a	,  eos_token_id=a	,  **a      )
       self.assertTrue(expectation == len(generated_tokens[0]      )      )

       lowercase_						:     Dict							     =  [6_3_8, 1_9_8]
       with tf.device(":/CPU:0"      ):
          tf.random.set_seed(0      )
          lowercase_						:     Dict							     =  model.generate(**a	,  eos_token_id=a	,  **a      )
       self.assertTrue(expectation == len(generated_tokens[0]      )      )





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






       '''simple docstring'''






       lowercase_						:     str							     =  AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-bart"      )
       lowercase_						:     Optional[Any]							     =  "Hugging Face is a technology company based in New York and Paris."
       lowercase_						:     List[Any]							     =  bart_tokenizer(a	,  return_tensors="tf"      ).input_ids
       lowercase_						:     Any							     =  TFBartForConditionalGeneration.from_pretrained("hf-internal-testing/tiny-random-bart"      )
       lowercase_						:     Optional[int]							     =  bart_model.generate(a      ).numpy()


       class 							_UpperCAmelCase    (		snake_case      ):




           def 					lowerCAmelCase__					(      self	:							Dict	,  a	:							List[Any]	,  a	:							str=None	,  **a	:							List[str]      ):






              '''simple docstring'''






              return super().call(a	,  **a      )


       lowercase_						:     Any							     =  FakeBart.from_pretrained("hf-internal-testing/tiny-random-bart"      )
       lowercase_						:     Tuple							     =  bart_model.generate(a	,  foo="bar"      ).numpy()
       self.assertTrue(np.array_equal(a	,  a      )      )


       class 							_UpperCAmelCase    (		bart_model.model.encoder.__class__      ):




           def 					lowerCAmelCase__					(      self	:							List[str]	,  a	:							Optional[int]	,  **a	:							Tuple      ):






              '''simple docstring'''






              return super().call(a	,  **a      )


       lowercase_						:     Dict							     =  FakeEncoder(bart_model.config	,  bart_model.model.shared      )
       lowercase_						:     Optional[Any]							     =  fake_encoder

       # Normal generation still works (the output will be different because the encoder weights are different)
       lowercase_						:     Any							     =  bart_model.generate(a      ).numpy()
       with self.assertRaises(a      ):
          # FakeEncoder.call() accepts **kwargs -> no filtering -> value error due to unexpected input "foo"
          bart_model.generate(a	,  foo="bar"      )












'''simple docstring'''
import argparse

import torch

from transformers import (
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaForAudioFrameClassification,
    WavaVecaForSequenceClassification,
    WavaVecaForXVector,
    logging,
)


logging.set_verbosity_info()
UpperCamelCase__			=      logging.get_logger(__name__)




def 		__SCREAMING_SNAKE_CASE       (	_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase				):


   """simple docstring"""




   lowercase_						:     Optional[Any]							     =  WavaVecaForSequenceClassification.from_pretrained(_UpperCamelCase    ,		config=_UpperCamelCase				)
   lowercase_						:     Optional[int]							     =  downstream_dict["projector.weight"]
   lowercase_						:     str							     =  downstream_dict["projector.bias"]
   lowercase_						:     int							     =  downstream_dict["model.post_net.linear.weight"]
   lowercase_						:     Optional[Any]							     =  downstream_dict["model.post_net.linear.bias"]
   return model




def 		__SCREAMING_SNAKE_CASE       (	_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase				):


   """simple docstring"""




   lowercase_						:     Tuple							     =  WavaVecaForAudioFrameClassification.from_pretrained(_UpperCamelCase    ,		config=_UpperCamelCase				)
   lowercase_						:     Any							     =  downstream_dict["model.linear.weight"]
   lowercase_						:     List[str]							     =  downstream_dict["model.linear.bias"]
   return model




def 		__SCREAMING_SNAKE_CASE       (	_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase				):


   """simple docstring"""




   lowercase_						:     Any							     =  WavaVecaForXVector.from_pretrained(_UpperCamelCase    ,		config=_UpperCamelCase				)
   lowercase_						:     str							     =  downstream_dict["connector.weight"]
   lowercase_						:     List[str]							     =  downstream_dict["connector.bias"]
   for i, kernel_size in enumerate(hf_config.tdnn_kernel				):
      lowercase_						:     Union[str, Any]							     =  downstream_dict[
          F"""model.framelevel_feature_extractor.module.{i}.kernel.weight"""
      ]
      lowercase_						:     Dict							     =  downstream_dict[F"""model.framelevel_feature_extractor.module.{i}.kernel.bias"""]

   lowercase_						:     Dict							     =  downstream_dict["model.utterancelevel_feature_extractor.linear1.weight"]
   lowercase_						:     List[Any]							     =  downstream_dict["model.utterancelevel_feature_extractor.linear1.bias"]
   lowercase_						:     Dict							     =  downstream_dict["model.utterancelevel_feature_extractor.linear2.weight"]
   lowercase_						:     Dict							     =  downstream_dict["model.utterancelevel_feature_extractor.linear2.bias"]
   lowercase_						:     Optional[Any]							     =  downstream_dict["objective.W"]
   return model




@torch.no_grad()
def 		__SCREAMING_SNAKE_CASE       (	_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase				):


   """simple docstring"""




   lowercase_						:     Tuple							     =  torch.load(_UpperCamelCase    ,		map_location="cpu"				)

   lowercase_						:     Dict							     =  checkpoint["Downstream"]

   lowercase_						:     Optional[Any]							     =  WavaVecaConfig.from_pretrained(_UpperCamelCase				)
   lowercase_						:     Dict							     =  WavaVecaFeatureExtractor.from_pretrained(
       _UpperCamelCase    ,		return_attention_mask=_UpperCamelCase    ,		do_normalize=_UpperCamelCase				)

   lowercase_						:     Dict							     =  hf_config.architectures[0]
   if arch.endswith("ForSequenceClassification"				):
      lowercase_						:     Any							     =  convert_classification(_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase				)
   elif arch.endswith("ForAudioFrameClassification"				):
      lowercase_						:     Optional[int]							     =  convert_diarization(_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase				)
   elif arch.endswith("ForXVector"				):
      lowercase_						:     List[Any]							     =  convert_xvector(_UpperCamelCase    ,		_UpperCamelCase    ,		_UpperCamelCase				)
   else:
      raise NotImplementedError(F"""S3PRL weights conversion is not supported for {arch}"""				)

   if hf_config.use_weighted_layer_sum:
      lowercase_						:     List[str]							     =  checkpoint["Featurizer"]["weights"]

   hf_feature_extractor.save_pretrained(_UpperCamelCase				)
   hf_model.save_pretrained(_UpperCamelCase				)


if __name__ == "__main__":
 UpperCamelCase__			=      argparse.ArgumentParser()
 parser.add_argument(
     '--base_model_name', default=None, type=str, help='Name of the huggingface pretrained base model.'
 )
 parser.add_argument('--config_path', default=None, type=str, help='Path to the huggingface classifier config.')
 parser.add_argument('--checkpoint_path', default=None, type=str, help='Path to the s3prl checkpoint.')
 parser.add_argument('--model_dump_path', default=None, type=str, help='Path to the final converted model.')
 UpperCamelCase__			=      parser.parse_args()
 convert_saprl_checkpoint(args.base_model_name, args.config_path, args.checkpoint_path, args.model_dump_path)




