



import argparse
import logging
import os
import sys
import tempfile
from pathlib import Path

import lightning_base
import pytest
import pytorch_lightning as pl
import torch
from convert_pl_checkpoint_to_hf import convert_pl_to_hf
from distillation import distill_main
from finetune import SummarizationModule, main
from huggingface_hub import list_models
from parameterized import parameterized
from run_eval import generate_summaries_or_translations
from torch import nn

from transformers import AutoConfig, AutoModelForSeqaSeqLM
from transformers.testing_utils import CaptureStderr, CaptureStdout, TestCasePlus, require_torch_gpu, slow
from utils import label_smoothed_nll_loss, lmap, load_json


logging.basicConfig(level=logging.DEBUG)

UpperCamelCase_									=   logging.getLogger()
UpperCamelCase_									=   torch.cuda.is_available()
UpperCamelCase_									=   {
    '''max_tokens_per_batch''': None,
    '''supervise_forward''': True,
    '''normalize_hidden''': True,
    '''label_smoothing''': 0.2,
    '''eval_max_gen_length''': None,
    '''eval_beams''': 1,
    '''val_metric''': '''loss''',
    '''save_top_k''': 1,
    '''adafactor''': True,
    '''early_stopping_patience''': 2,
    '''logger_name''': '''default''',
    '''length_penalty''': 0.5,
    '''cache_dir''': '''''',
    '''task''': '''summarization''',
    '''num_workers''': 2,
    '''alpha_hid''': 0,
    '''freeze_embeds''': True,
    '''enc_only''': False,
    '''tgt_suffix''': '''''',
    '''resume_from_checkpoint''': None,
    '''sortish_sampler''': True,
    '''student_decoder_layers''': 1,
    '''val_check_interval''': 1.0,
    '''output_dir''': '''''',
    '''fp16''': False,  # TODO(SS): set this to CUDA_AVAILABLE if ci installs apex or start using native amp
    '''no_teacher''': False,
    '''fp16_opt_level''': '''O1''',
    '''gpus''': 1 if CUDA_AVAILABLE else 0,
    '''n_tpu_cores''': 0,
    '''max_grad_norm''': 1.0,
    '''do_train''': True,
    '''do_predict''': True,
    '''accumulate_grad_batches''': 1,
    '''server_ip''': '''''',
    '''server_port''': '''''',
    '''seed''': 42,
    '''model_name_or_path''': '''sshleifer/bart-tiny-random''',
    '''config_name''': '''''',
    '''tokenizer_name''': '''facebook/bart-large''',
    '''do_lower_case''': False,
    '''learning_rate''': 0.3,
    '''lr_scheduler''': '''linear''',
    '''weight_decay''': 0.0,
    '''adam_epsilon''': 1E-0_8,
    '''warmup_steps''': 0,
    '''max_epochs''': 1,
    '''train_batch_size''': 2,
    '''eval_batch_size''': 2,
    '''max_source_length''': 12,
    '''max_target_length''': 12,
    '''val_max_target_length''': 12,
    '''test_max_target_length''': 12,
    '''fast_dev_run''': False,
    '''no_cache''': False,
    '''n_train''': -1,
    '''n_val''': -1,
    '''n_test''': -1,
    '''student_encoder_layers''': 1,
    '''freeze_encoder''': False,
    '''auto_scale_batch_size''': False,
    '''overwrite_output_dir''': False,
    '''student''': None,
}







def       lowerCamelCase_  (    _a       :						Path   ,		_a       :						list    ):



    '''simple docstring'''




    UpperCAmelCase_     :       Dict  		=					"""\n""".join(_a    )
    Path(_a    ).open("""w"""    ).writelines(_a    )


UpperCamelCase_									=   [''' Sam ate lunch today.''', '''Sams lunch ingredients.''']
UpperCamelCase_									=   ['''A very interesting story about what I ate for lunch.''', '''Avocado, celery, turkey, coffee''']
UpperCamelCase_									=   '''patrickvonplaten/t5-tiny-random'''
UpperCamelCase_									=   '''sshleifer/t5-tinier-random'''
UpperCamelCase_									=   '''sshleifer/bart-tiny-random'''
UpperCamelCase_									=   '''sshleifer/tiny-mbart'''
UpperCamelCase_									=   '''sshleifer/tiny-marian-en-de'''
UpperCamelCase_									=   '''stas/tiny-wmt19-en-de'''


UpperCamelCase_									=   logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)
logging.disable(logging.CRITICAL)  # remove noisy download output from tracebacks







def       lowerCamelCase_  (    _a       :						int    ):



    '''simple docstring'''




    for split in ["train", "val", "test"]:
        _dump_articles(os.path.join(_a   ,		F'''{split}.source'''    )   ,		_a    )
        _dump_articles(os.path.join(_a   ,		F'''{split}.target'''    )   ,		_a    )
    return tmp_dir







class       _snake_case				(     __snake_case ):






   '''simple docstring'''






   @classmethod
   def       A__    (    cls:     Any							)	->      Tuple:
       logging.disable(logging.CRITICAL							)  # remove noisy download output from tracebacks
       return cls


   @slow
   @require_torch_gpu
   def       A__    (    self:     Optional[int]							)	->      Tuple:
       UpperCAmelCase_     :       int  		=					list_models()
       UpperCAmelCase_     :       Optional[int]  		=					"""sshleifer"""
       UpperCAmelCase_     :       Optional[Any]  		=					[x.modelId for x in model_list if x.modelId.startswith(lowerCamelCase_							)]
       UpperCAmelCase_     :       Optional[int]  		=					["""sshleifer/blenderbot-3B""", """sshleifer/blenderbot-90M"""]
       UpperCAmelCase_     :       Tuple  		=					[]
       for m in model_ids:
           if m in allowed_to_be_broken:
               continue
           try:
               AutoConfig.from_pretrained(lowerCamelCase_							)
           except Exception:
               failures.append(lowerCamelCase_							)
       assert not failures, F'''The following models could not be loaded through AutoConfig: {failures}'''


   def       A__    (    self:     List[Any]							)	->      Dict:
       UpperCAmelCase_     :       List[str]  		=					{"""student_encoder_layers""": 2, """student_decoder_layers""": 1, """no_teacher""": True}
       self._test_distiller_cli(lowerCamelCase_							)


   def       A__    (    self:     Tuple							)	->      Any:
       UpperCAmelCase_     :       int  		=					{
           """student_encoder_layers""": 2,
           """student_decoder_layers""": 1,
           """max_epochs""": 4,
           """val_check_interval""": 0.2_5,
           """alpha_hid""": 2.0,
           """model_name_or_path""": """IGNORE_THIS_IT_DOESNT_GET_USED""",
       }
       UpperCAmelCase_     :       Optional[int]  		=					self._test_distiller_cli(lowerCamelCase_  ,check_contents=lowerCamelCase_							)

       UpperCAmelCase_     :       Optional[Any]  		=					list(Path(model.output_dir							).glob("""*.ckpt"""							)							)
       self.assertEqual(1  ,len(lowerCamelCase_							)							)
       UpperCAmelCase_     :       int  		=					list(Path(model.output_dir							).glob("""**/*.bin"""							)							)
       self.assertEqual(len(lowerCamelCase_							)  ,2							)
       UpperCAmelCase_     :       List[Any]  		=					lmap(str.strip  ,Path(model.hparams.data_dir							).joinpath("""test.source"""							).open().readlines()							)
       UpperCAmelCase_     :       Tuple  		=					tempfile.mktemp()  # XXX: not being cleaned up
       generate_summaries_or_translations(lowerCamelCase_  ,lowerCamelCase_  ,str(model.output_dir / """best_tfmr"""							)							)
       self.assertTrue(Path(lowerCamelCase_							).exists()							)

       UpperCAmelCase_     :       Optional[int]  		=					self.get_auto_remove_tmp_dir()
       convert_pl_to_hf(ckpts[0]  ,transformer_ckpts[0].parent  ,lowerCamelCase_							)
       assert os.path.exists(os.path.join(lowerCamelCase_  ,"""pytorch_model.bin"""							)							)


   def       A__    (    self:     Tuple							)	->      List[str]:
       UpperCAmelCase_     :       Optional[Any]  		=					AutoModelForSeqaSeqLM.from_pretrained(lowerCamelCase_							)
       UpperCAmelCase_		, UpperCAmelCase_     :       Any  		=					model.dummy_inputs["""input_ids"""], model.dummy_inputs["""attention_mask"""]
       UpperCAmelCase_     :       Union[str, Any]  		=					torch.tensor([[0, 4, 8, 2], [0, 8, 2, 1]]  ,dtype=torch.long  ,device=model.device							)
       UpperCAmelCase_     :       List[str]  		=					target_ids[:, :-1].contiguous()  # Why this line?
       UpperCAmelCase_     :       int  		=					target_ids[:, 1:].clone()  # why clone?
       UpperCAmelCase_     :       str  		=					model(
           lowerCamelCase_  ,attention_mask=lowerCamelCase_  ,decoder_input_ids=lowerCamelCase_  ,labels=lowerCamelCase_  ,use_cache=lowerCamelCase_							).loss

       UpperCAmelCase_     :       Tuple  		=					model(lowerCamelCase_  ,attention_mask=lowerCamelCase_  ,decoder_input_ids=lowerCamelCase_  ,use_cache=lowerCamelCase_							).logits

       UpperCAmelCase_     :       str  		=					nn.functional.log_softmax(lowerCamelCase_  ,dim=-1							)
       UpperCAmelCase_		, UpperCAmelCase_     :       Dict  		=					label_smoothed_nll_loss(
           lowerCamelCase_  ,lowerCamelCase_  ,0.1  ,ignore_index=model.config.pad_token_id							)
       with self.assertRaises(lowerCamelCase_							):
           # TODO: understand why this breaks
           self.assertEqual(lowerCamelCase_  ,lowerCamelCase_							)


   def       A__    (    self:     int							)	->      int:
       UpperCAmelCase_     :       List[str]  		=					{
           """student_encoder_layers""": 2,
           """student_decoder_layers""": 1,
           """num_train_epochs""": 4,
           """val_check_interval""": 0.2_5,
           """alpha_hid""": 2.0,
           """task""": """translation""",
           """model_name_or_path""": """IGNORE_THIS_IT_DOESNT_GET_USED""",
           """tokenizer_name""": MBART_TINY,
           """teacher""": MBART_TINY,
           """src_lang""": """en_XX""",
           """tgt_lang""": """ro_RO""",
       }
       UpperCAmelCase_     :       int  		=					self._test_distiller_cli(lowerCamelCase_  ,check_contents=lowerCamelCase_							)
       assert model.model.config.model_type == "mbart"

       UpperCAmelCase_     :       List[str]  		=					list(Path(model.output_dir							).glob("""*.ckpt"""							)							)
       self.assertEqual(1  ,len(lowerCamelCase_							)							)
       UpperCAmelCase_     :       Optional[int]  		=					list(Path(model.output_dir							).glob("""**/*.bin"""							)							)
       UpperCAmelCase_     :       Union[str, Any]  		=					list(Path(model.output_dir							).glob("""best_tfmr/*"""							)							)
       assert len(lowerCamelCase_							) > 2
       self.assertEqual(len(lowerCamelCase_							)  ,2							)


   def       A__    (    self:     Dict							)	->      Optional[int]:
       UpperCAmelCase_     :       Any  		=					{
           """student_encoder_layers""": 1,
           """student_decoder_layers""": 1,
           """alpha_hid""": 2.0,
           """teacher""": T5_TINY,
           """model_name_or_path""": T5_TINY,
           """tokenizer_name""": T5_TINY,
       }
       self._test_distiller_cli(lowerCamelCase_							)


   def       A__    (    self:     Any							)	->      List[Any]:
       UpperCAmelCase_     :       str  		=					{
           """teacher""": T5_TINY,
           """student""": T5_TINIER,
           """model_name_or_path""": T5_TINIER,
           """tokenizer_name""": T5_TINIER,
       }
       self._test_distiller_cli(lowerCamelCase_							)






   def       A__    (    self:     Optional[int]  ,lowerCamelCase_:     int  ,lowerCamelCase_:     Union[str, Any]=True							)	->      Union[str, Any]:
       UpperCAmelCase_     :       Tuple  		=					{
           """label_smoothing""": 0.0,
           """early_stopping_patience""": -1,
           """train_batch_size""": 1,
           """eval_batch_size""": 2,
           """max_epochs""": 2,
           """alpha_mlm""": 0.2,
           """alpha_ce""": 0.8,
           """do_predict""": True,
           """model_name_or_path""": """sshleifer/tinier_bart""",
           """teacher""": CHEAP_ARGS["""model_name_or_path"""],
           """val_check_interval""": 0.5,
       }
       default_updates.update(lowerCamelCase_							)
       UpperCAmelCase_     :       dict  		=					CHEAP_ARGS.copy()
       UpperCAmelCase_     :       int  		=					make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()							)
       UpperCAmelCase_     :       Optional[int]  		=					self.get_auto_remove_tmp_dir()

       args_d.update(data_dir=lowerCamelCase_  ,output_dir=lowerCamelCase_  ,**lowerCamelCase_							)
       UpperCAmelCase_     :       Optional[Any]  		=					distill_main(argparse.Namespace(**lowerCamelCase_							)							)
       if not check_contents:
           return model
       UpperCAmelCase_     :       List[Any]  		=					os.listdir(lowerCamelCase_							)
       UpperCAmelCase_     :       Tuple  		=					{os.path.basename(lowerCamelCase_							) for p in contents}
       UpperCAmelCase_     :       Tuple  		=					[p for p in contents if p.endswith("""ckpt"""							)]
       assert len(lowerCamelCase_							) > 0

       self.assertIn("""test_generations.txt"""  ,lowerCamelCase_							)
       self.assertIn("""test_results.txt"""  ,lowerCamelCase_							)

       UpperCAmelCase_     :       List[str]  		=					load_json(model.metrics_save_path							)
       UpperCAmelCase_     :       List[Any]  		=					metrics["""val"""][-1]
       self.assertGreaterEqual(last_step_stats["""val_avg_gen_time"""]  ,0.0_1							)
       self.assertGreaterEqual(1.0  ,last_step_stats["""val_avg_gen_time"""]							)
       self.assertIsInstance(last_step_stats[F'''val_avg_{model.val_metric}''']  ,lowerCamelCase_							)
       UpperCAmelCase_     :       Dict  		=					int(args_d["""max_epochs"""] * (1 / args_d["""val_check_interval"""]) + 1							)
       self.assertEqual(len(metrics["""val"""]							)  ,lowerCamelCase_							)
       self.assertEqual(len(metrics["""test"""]							)  ,1							)
       return model







class       _snake_case				(     __snake_case ):






   '''simple docstring'''






   @parameterized.expand(
       [T5_TINY, BART_TINY, MBART_TINY, MARIAN_TINY, FSMT_TINY]  ,)
   def       A__    (    self:     List[str]  ,lowerCamelCase_:     Dict							)	->      Optional[int]:
       UpperCAmelCase_     :       dict  		=					CHEAP_ARGS.copy()
       UpperCAmelCase_     :       Optional[Any]  		=					"""translation""" if model in [MBART_TINY, MARIAN_TINY, FSMT_TINY] else """summarization"""
       UpperCAmelCase_     :       List[Any]  		=					0.1 if task == """translation""" else 0

       UpperCAmelCase_     :       int  		=					make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()							)
       UpperCAmelCase_     :       Any  		=					self.get_auto_remove_tmp_dir()
       args_d.update(
           data_dir=lowerCamelCase_  ,model_name_or_path=lowerCamelCase_  ,tokenizer_name=lowerCamelCase_  ,train_batch_size=2  ,eval_batch_size=2  ,output_dir=lowerCamelCase_  ,do_predict=lowerCamelCase_  ,task=lowerCamelCase_  ,src_lang="""en_XX"""  ,tgt_lang="""ro_RO"""  ,freeze_encoder=lowerCamelCase_  ,freeze_embeds=lowerCamelCase_  ,)
       assert "n_train" in args_d
       UpperCAmelCase_     :       Optional[int]  		=					argparse.Namespace(**lowerCamelCase_							)
       UpperCAmelCase_     :       List[str]  		=					main(lowerCamelCase_							)

       UpperCAmelCase_     :       List[str]  		=					module.model.get_input_embeddings()
       assert not input_embeds.weight.requires_grad
       if model == T5_TINY:
           UpperCAmelCase_     :       Dict  		=					module.model.lm_head
           assert not lm_head.weight.requires_grad
           assert (lm_head.weight == input_embeds.weight).all().item()
       elif model == FSMT_TINY:
           UpperCAmelCase_     :       List[str]  		=					module.model.model
           UpperCAmelCase_     :       Union[str, Any]  		=					fsmt.decoder.embed_positions
           assert not embed_pos.weight.requires_grad
           assert not fsmt.decoder.embed_tokens.weight.requires_grad
           # check that embeds are not the same
           assert fsmt.decoder.embed_tokens != fsmt.encoder.embed_tokens
       else:
           UpperCAmelCase_     :       str  		=					module.model.model
           UpperCAmelCase_     :       List[str]  		=					bart.decoder.embed_positions
           assert not embed_pos.weight.requires_grad
           assert not bart.shared.weight.requires_grad
           # check that embeds are the same
           assert bart.decoder.embed_tokens == bart.encoder.embed_tokens
           assert bart.decoder.embed_tokens == bart.shared

       UpperCAmelCase_     :       Optional[Any]  		=					load_json(module.output_dir / """text_batch.json"""							)
       assert isinstance(lowerCamelCase_  ,lowerCamelCase_							)
       assert len(lowerCamelCase_							) >= 4


   def       A__    (    self:     int							)	->      Optional[int]:
       UpperCAmelCase_     :       dict  		=					CHEAP_ARGS.copy()

       UpperCAmelCase_     :       List[Any]  		=					"""summarization"""
       UpperCAmelCase_     :       Union[str, Any]  		=					make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()							)

       args_d.update(
           data_dir=lowerCamelCase_  ,tokenizer_name=lowerCamelCase_  ,train_batch_size=2  ,eval_batch_size=2  ,do_predict=lowerCamelCase_  ,task=lowerCamelCase_  ,src_lang="""en_XX"""  ,tgt_lang="""ro_RO"""  ,freeze_encoder=lowerCamelCase_  ,freeze_embeds=lowerCamelCase_  ,)

       # test models whose config includes the extra_model_args
       UpperCAmelCase_     :       Union[str, Any]  		=					BART_TINY
       UpperCAmelCase_     :       Dict  		=					self.get_auto_remove_tmp_dir()
       UpperCAmelCase_     :       Union[str, Any]  		=					args_d.copy()
       args_da.update(
           model_name_or_path=lowerCamelCase_  ,output_dir=lowerCamelCase_  ,)
       UpperCAmelCase_     :       Optional[Any]  		=					("""encoder_layerdrop""", """decoder_layerdrop""", """dropout""", """attention_dropout""")
       for p in extra_model_params:
           UpperCAmelCase_     :       List[str]  		=					0.5
       UpperCAmelCase_     :       Optional[Any]  		=					argparse.Namespace(**lowerCamelCase_							)
       UpperCAmelCase_     :       int  		=					main(lowerCamelCase_							)
       for p in extra_model_params:
           assert getattr(model.config  ,lowerCamelCase_							) == 0.5, F'''failed to override the model config for param {p}'''

       # test models whose config doesn't include the extra_model_args
       UpperCAmelCase_     :       List[Any]  		=					T5_TINY
       UpperCAmelCase_     :       str  		=					self.get_auto_remove_tmp_dir()
       UpperCAmelCase_     :       Tuple  		=					args_d.copy()
       args_da.update(
           model_name_or_path=lowerCamelCase_  ,output_dir=lowerCamelCase_  ,)
       UpperCAmelCase_     :       List[str]  		=					"""encoder_layerdrop"""
       UpperCAmelCase_     :       Optional[Any]  		=					0.5
       UpperCAmelCase_     :       Optional[Any]  		=					argparse.Namespace(**lowerCamelCase_							)
       with pytest.raises(lowerCamelCase_							) as excinfo:
           UpperCAmelCase_     :       Optional[Any]  		=					main(lowerCamelCase_							)
       assert str(excinfo.value							) == F'''model config doesn\'t have a `{unsupported_param}` attribute'''






   def       A__    (    self:     Optional[Any]							)	->      Any:
       UpperCAmelCase_     :       dict  		=					CHEAP_ARGS.copy()

       UpperCAmelCase_     :       Tuple  		=					"""summarization"""
       UpperCAmelCase_     :       Any  		=					make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()							)

       UpperCAmelCase_     :       Tuple  		=					BART_TINY
       UpperCAmelCase_     :       int  		=					self.get_auto_remove_tmp_dir()

       args_d.update(
           data_dir=lowerCamelCase_  ,model_name_or_path=lowerCamelCase_  ,output_dir=lowerCamelCase_  ,tokenizer_name=lowerCamelCase_  ,train_batch_size=2  ,eval_batch_size=2  ,do_predict=lowerCamelCase_  ,task=lowerCamelCase_  ,src_lang="""en_XX"""  ,tgt_lang="""ro_RO"""  ,freeze_encoder=lowerCamelCase_  ,freeze_embeds=lowerCamelCase_  ,)

       # emulate finetune.py
       UpperCAmelCase_     :       List[str]  		=					argparse.ArgumentParser()
       UpperCAmelCase_     :       Dict  		=					pl.Trainer.add_argparse_args(lowerCamelCase_							)
       UpperCAmelCase_     :       Optional[Any]  		=					SummarizationModule.add_model_specific_args(lowerCamelCase_  ,os.getcwd()							)
       UpperCAmelCase_     :       int  		=					{"""--help""": True}

       # --help test
       with pytest.raises(lowerCamelCase_							) as excinfo:
           with CaptureStdout() as cs:
               UpperCAmelCase_     :       Union[str, Any]  		=					parser.parse_args(lowerCamelCase_							)
           assert False, "--help is expected to sys.exit"
       assert excinfo.type == SystemExit
       UpperCAmelCase_     :       Union[str, Any]  		=					lightning_base.arg_to_scheduler_metavar
       assert expected in cs.out, "--help is expected to list the supported schedulers"

       # --lr_scheduler=non_existing_scheduler test
       UpperCAmelCase_     :       Any  		=					"""non_existing_scheduler"""
       UpperCAmelCase_     :       Optional[Any]  		=					{F'''--lr_scheduler={unsupported_param}'''}
       with pytest.raises(lowerCamelCase_							) as excinfo:
           with CaptureStderr() as cs:
               UpperCAmelCase_     :       List[Any]  		=					parser.parse_args(lowerCamelCase_							)
           assert False, "invalid argument is expected to sys.exit"
       assert excinfo.type == SystemExit
       UpperCAmelCase_     :       Optional[Any]  		=					F'''invalid choice: \'{unsupported_param}\''''
       assert expected in cs.err, F'''should have bailed on invalid choice of scheduler {unsupported_param}'''

       # --lr_scheduler=existing_scheduler test
       UpperCAmelCase_     :       Optional[Any]  		=					"""cosine"""
       UpperCAmelCase_     :       Optional[int]  		=					args_d.copy()
       UpperCAmelCase_     :       Union[str, Any]  		=					supported_param
       UpperCAmelCase_     :       Any  		=					argparse.Namespace(**lowerCamelCase_							)
       UpperCAmelCase_     :       Optional[int]  		=					main(lowerCamelCase_							)
       assert (
           getattr(model.hparams  ,"""lr_scheduler"""							) == supported_param
       ), F'''lr_scheduler={supported_param} shouldn\'t fail'''


import shutil
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from transformers import SpeechaTextTokenizer, is_speech_available
from transformers.models.speech_to_text.tokenization_speech_to_text import VOCAB_FILES_NAMES, save_json
from transformers.testing_utils import get_tests_dir, require_sentencepiece, require_torch, require_torchaudio
from transformers.utils import FEATURE_EXTRACTOR_NAME

from .test_feature_extraction_speech_to_text import floats_list


if is_speech_available():
       from transformers import SpeechaTextFeatureExtractor, SpeechaTextProcessor


_UpperCamelCase:					str     =get_tests_dir('fixtures/test_sentencepiece.model')





@require_torch
@require_torchaudio
@require_sentencepiece
class 						__lowercase(     unittest.TestCase   ):


       """simple docstring"""







       def SCREAMING_SNAKE_CASE_				(     self					:			List[str]       )					->   Any:
         _lowerCAmelCase						=		tempfile.mkdtemp()

         _lowerCAmelCase						=		['<s>', '<pad>', '</s>', '<unk>', '▁This', '▁is', '▁a', '▁t', 'est']
         _lowerCAmelCase						=		dict(zip(_lowerCAmelCase  ,      range(len(_lowerCAmelCase       )       )       )       )
         _lowerCAmelCase						=		Path(self.tmpdirname       )
         save_json(_lowerCAmelCase  ,      save_dir / VOCAB_FILES_NAMES['vocab_file']       )
         if not (save_dir / VOCAB_FILES_NAMES["spm_file"]).exists():
           copyfile(_lowerCAmelCase  ,      save_dir / VOCAB_FILES_NAMES['spm_file']       )

         _lowerCAmelCase						=		SpeechaTextTokenizer.from_pretrained(self.tmpdirname       )
         tokenizer.save_pretrained(self.tmpdirname       )

         _lowerCAmelCase						=		{
             'feature_size': 24,
             'num_mel_bins': 24,
             'padding_value': 0.0,
             'sampling_rate': 1_6000,
             'return_attention_mask': False,
             'do_normalize': True,
         }
         save_json(_lowerCAmelCase  ,      save_dir / FEATURE_EXTRACTOR_NAME       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Union[str, Any]  ,      **_lowerCAmelCase					:			Optional[Any]       )					->   List[str]:
         return SpeechaTextTokenizer.from_pretrained(self.tmpdirname  ,      **_lowerCAmelCase       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Optional[Any]  ,      **_lowerCAmelCase					:			Dict       )					->   Tuple:
         return SpeechaTextFeatureExtractor.from_pretrained(self.tmpdirname  ,      **_lowerCAmelCase       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Optional[Any]       )					->   int:
         shutil.rmtree(self.tmpdirname       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Optional[int]       )					->   Any:
         _lowerCAmelCase						=		self.get_tokenizer()
         _lowerCAmelCase						=		self.get_feature_extractor()

         _lowerCAmelCase						=		SpeechaTextProcessor(tokenizer=_lowerCAmelCase  ,      feature_extractor=_lowerCAmelCase       )

         processor.save_pretrained(self.tmpdirname       )
         _lowerCAmelCase						=		SpeechaTextProcessor.from_pretrained(self.tmpdirname       )

         self.assertEqual(processor.tokenizer.get_vocab()  ,      tokenizer.get_vocab()       )
         self.assertIsInstance(processor.tokenizer  ,      _lowerCAmelCase       )

         self.assertEqual(processor.feature_extractor.to_json_string()  ,      feature_extractor.to_json_string()       )
         self.assertIsInstance(processor.feature_extractor  ,      _lowerCAmelCase       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Optional[int]       )					->   int:
         _lowerCAmelCase						=		SpeechaTextProcessor(
             tokenizer=self.get_tokenizer()  ,      feature_extractor=self.get_feature_extractor()       )
         processor.save_pretrained(self.tmpdirname       )

         _lowerCAmelCase						=		self.get_tokenizer(bos_token='(BOS)'  ,      eos_token='(EOS)'       )
         _lowerCAmelCase						=		self.get_feature_extractor(do_normalize=_lowerCAmelCase  ,      padding_value=1.0       )

         _lowerCAmelCase						=		SpeechaTextProcessor.from_pretrained(
             self.tmpdirname  ,      bos_token='(BOS)'  ,      eos_token='(EOS)'  ,      do_normalize=_lowerCAmelCase  ,      padding_value=1.0       )

         self.assertEqual(processor.tokenizer.get_vocab()  ,      tokenizer_add_kwargs.get_vocab()       )
         self.assertIsInstance(processor.tokenizer  ,      _lowerCAmelCase       )

         self.assertEqual(processor.feature_extractor.to_json_string()  ,      feature_extractor_add_kwargs.to_json_string()       )
         self.assertIsInstance(processor.feature_extractor  ,      _lowerCAmelCase       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Dict       )					->   Union[str, Any]:
         _lowerCAmelCase						=		self.get_feature_extractor()
         _lowerCAmelCase						=		self.get_tokenizer()

         _lowerCAmelCase						=		SpeechaTextProcessor(tokenizer=_lowerCAmelCase  ,      feature_extractor=_lowerCAmelCase       )

         _lowerCAmelCase						=		floats_list((3, 1000)       )

         _lowerCAmelCase						=		feature_extractor(_lowerCAmelCase  ,      return_tensors='np'       )
         _lowerCAmelCase						=		processor(_lowerCAmelCase  ,      return_tensors='np'       )

         for key in input_feat_extract.keys():
           self.assertAlmostEqual(input_feat_extract[key].sum()  ,      input_processor[key].sum()  ,      delta=1e-2       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Dict       )					->   Any:
         _lowerCAmelCase						=		self.get_feature_extractor()
         _lowerCAmelCase						=		self.get_tokenizer()

         _lowerCAmelCase						=		SpeechaTextProcessor(tokenizer=_lowerCAmelCase  ,      feature_extractor=_lowerCAmelCase       )

         _lowerCAmelCase						=		'This is a test string'

         _lowerCAmelCase						=		processor(text=_lowerCAmelCase       )

         _lowerCAmelCase						=		tokenizer(_lowerCAmelCase       )

         for key in encoded_tok.keys():
           self.assertListEqual(encoded_tok[key]  ,      encoded_processor[key]       )



       def SCREAMING_SNAKE_CASE_				(     self					:			Tuple       )					->   Tuple:
         _lowerCAmelCase						=		self.get_feature_extractor()
         _lowerCAmelCase						=		self.get_tokenizer()

         _lowerCAmelCase						=		SpeechaTextProcessor(tokenizer=_lowerCAmelCase  ,      feature_extractor=_lowerCAmelCase       )

         _lowerCAmelCase						=		[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

         _lowerCAmelCase						=		processor.batch_decode(_lowerCAmelCase       )
         _lowerCAmelCase						=		tokenizer.batch_decode(_lowerCAmelCase       )

         self.assertListEqual(_lowerCAmelCase  ,      _lowerCAmelCase       )






       def SCREAMING_SNAKE_CASE_				(     self					:			Any       )					->   Union[str, Any]:
         _lowerCAmelCase						=		self.get_feature_extractor()
         _lowerCAmelCase						=		self.get_tokenizer()

         _lowerCAmelCase						=		SpeechaTextProcessor(tokenizer=_lowerCAmelCase  ,      feature_extractor=_lowerCAmelCase       )

         self.assertListEqual(
             processor.model_input_names  ,      feature_extractor.model_input_names  ,      msg='`processor` and `feature_extractor` model input names do not match'  ,      )
