


"""simple docstring"""






import unittest

from transformers import SpeechaTextaConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
 import torch

 from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
     SpeechaTextaDecoder,
     SpeechaTextaForCausalLM,
 )


@require_torch
class 	A_       :





  """simple docstring"""

  def __init__(						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase=99  ,					__UpperCAmelCase=13  ,					__UpperCAmelCase=16  ,					__UpperCAmelCase=7  ,					__UpperCAmelCase=True  ,					__UpperCAmelCase=True  ,					__UpperCAmelCase=True  ,					__UpperCAmelCase=False  ,					__UpperCAmelCase=True  ,					__UpperCAmelCase=2  ,					__UpperCAmelCase=32  ,					__UpperCAmelCase=4  ,					__UpperCAmelCase=4  ,					__UpperCAmelCase=30  ,					__UpperCAmelCase=0  ,					__UpperCAmelCase=1  ,					__UpperCAmelCase=2  ,					__UpperCAmelCase=None  ,					)				->					Optional[int]:
     a   :	int						=    parent
     a   :	Dict						=    batch_size
     a   :	List[Any]						=    decoder_seq_length
     # For common tests
     a   :	Any						=    self.decoder_seq_length
     a   :	Optional[Any]						=    is_training
     a   :	Union[str, Any]						=    use_attention_mask
     a   :	int						=    use_labels

     a   :	Optional[int]						=    vocab_size
     a   :	List[str]						=    d_model
     a   :	Tuple						=    d_model
     a   :	List[Any]						=    decoder_layers
     a   :	str						=    decoder_layers
     a   :	Union[str, Any]						=    decoder_ffn_dim
     a   :	List[str]						=    decoder_attention_heads
     a   :	Dict						=    decoder_attention_heads
     a   :	Tuple						=    eos_token_id
     a   :	str						=    bos_token_id
     a   :	List[Any]						=    pad_token_id
     a   :	List[str]						=    decoder_start_token_id
     a   :	Union[str, Any]						=    use_cache
     a   :	str						=    max_position_embeddings

     a   :	Optional[Any]						=    None
     a   :	Tuple						=    decoder_seq_length
     a   :	List[str]						=    2
     a   :	List[str]						=    1
  def 							lowercase_     (						self						)				->					Optional[int]:
     a   :	List[Any]						=    ids_tensor([self.batch_size, self.decoder_seq_length]  ,					self.vocab_size						)

     a   :	Optional[Any]						=    None
     if self.use_attention_mask:
        a   :	Any						=    ids_tensor([self.batch_size, self.decoder_seq_length]  ,					vocab_size=2						)

     a   :	str						=    None
     if self.use_labels:
        a   :	List[str]						=    ids_tensor([self.batch_size, self.decoder_seq_length]  ,					self.vocab_size						)

     a   :	Any						=    SpeechaTextaConfig(
         vocab_size=self.vocab_size  ,					d_model=self.d_model  ,					decoder_layers=self.decoder_layers  ,					decoder_ffn_dim=self.decoder_ffn_dim  ,					decoder_attention_heads=self.decoder_attention_heads  ,					eos_token_id=self.eos_token_id  ,					bos_token_id=self.bos_token_id  ,					use_cache=self.use_cache  ,					pad_token_id=self.pad_token_id  ,					decoder_start_token_id=self.decoder_start_token_id  ,					max_position_embeddings=self.max_position_embeddings  ,					)

     return (
         config,
         input_ids,
         attention_mask,
         lm_labels,
     )
  def 							lowercase_     (						self  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					__UpperCAmelCase  ,					)				->					int:
     a   :	Dict						=    True
     a   :	Any						=    SpeechaTextaDecoder(config=__a						).to(__a						).eval()
     a   :	Dict						=    input_ids[:2]

     input_ids[input_ids == 0] += 1
     # first forward pass
     a   :	Optional[Any]						=    model(__a  ,					use_cache=__a						)
     a   :	int						=    model(__a						)
     a   :	Union[str, Any]						=    model(__a  ,					use_cache=__a						)

     self.parent.assertTrue(len(__a						) == len(__a						)						)
     self.parent.assertTrue(len(__a						) == len(__a						) + 1						)

     a   :	Tuple						=    outputs['past_key_values']

     # create hypothetical next token and extent to next_input_ids
     a   :	Tuple						=    ids_tensor((2, 1)  ,					config.vocab_size - 1						) + 1

     # append to next input_ids and
     a   :	List[Any]						=    torch.cat([input_ids, next_tokens]  ,					dim=-1						)

     print(__a						)

     a   :	Tuple						=    model(__a						)['last_hidden_state']
     a   :	int						=    model(__a  ,					past_key_values=__a						)['last_hidden_state']

     # select random slice
     a   :	Any						=    ids_tensor((1,)  ,					output_from_past.shape[-1]						).item()
     a   :	str						=    output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
     a   :	Dict						=    output_from_past[:, 0, random_slice_idx].detach()

     # test that outputs are equal for slice
     assert torch.allclose(__a  ,					__a  ,					atol=1E-3						)






  def 							lowercase_     (						self						)				->					Tuple:
     a   :	Tuple						=    self.prepare_config_and_inputs()
     (
         (
         a					
     )	, (
         a					
     )	, (
         a					
     )	, (
         a					
     )	, 					
     )   :	Optional[Any]						=    config_and_inputs

     a   :	Optional[Any]						=    {
         'input_ids': input_ids,
         'attention_mask': attention_mask,
     }
     return config, inputs_dict



@require_torch
class 	A_       (     _lowercase      ,  _lowercase      ,  _lowercase      ,  unittest.TestCase   ):





  """simple docstring"""

  lowercase		:	Dict   =						(SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
  lowercase		:	Optional[Any]   =						(SpeechaTextaForCausalLM,) if is_torch_available() else ()
  lowercase		:	List[Any]   =						{"text-generation": SpeechaTextaForCausalLM} if is_torch_available() else {}
  lowercase		:	str   =						True
  lowercase		:	int   =						False
  def 							lowercase_     (						self  ,					)				->					str:
     a   :	Tuple						=    SpeechaTextaStandaloneDecoderModelTester(self  ,					is_training=__a						)
     a   :	Tuple						=    ConfigTester(self  ,					config_class=__a						)
  def 							lowercase_     (						self						)				->					Dict:
     pass
  def 							lowercase_     (						self						)				->					List[str]:
     pass
  def 							lowercase_     (						self						)				->					Any:
     pass
  def 							lowercase_     (						self						)				->					Any:
     self.config_tester.run_common_tests()
  def 							lowercase_     (						self						)				->					Optional[Any]:
     a   :	Optional[Any]						=    self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_decoder_model_past(*__a						)






  def 							lowercase_     (						self						)				->					Union[str, Any]:
     return












"""simple docstring"""





import argparse
import logging

import datasets
from tokenizers import Tokenizer, decoders, normalizers, pre_tokenizers, processors
from tokenizers.models import Unigram
from tokenizers.trainers import UnigramTrainer

from transformers import AlbertTokenizerFast


lowerCAmelCase__        =      logging.getLogger(__name__)


def    a__       (       ):




		"""simple docstring"""


		UpperCamelCase =						argparse.ArgumentParser(description="Train a unigram tokenizer on the wikitext dataset."     )
		parser.add_argument(
		    "--dataset_name"	,						type=_SCREAMING_SNAKE_CASE	,						default="wikitext"	,						help="Name of the training. Explore datasets at: hf.co/datasets."	,						)
		parser.add_argument(
		    "--dataset_config"	,						type=_SCREAMING_SNAKE_CASE	,						default="wikitext-103-raw-v1"	,						help="Configuration name of the dataset."     )
		parser.add_argument(
		    "--batch_size"	,						type=_SCREAMING_SNAKE_CASE	,						default=1_000	,						help="Batch size during training."	,						)
		parser.add_argument(
		    "--vocab_size"	,						type=_SCREAMING_SNAKE_CASE	,						default=10_048	,						help="Size of the desired vocabulary."	,						)
		parser.add_argument(
		    "--limit"	,						default=_SCREAMING_SNAKE_CASE	,						type=_SCREAMING_SNAKE_CASE	,						help="Limit the number of shards (used for debugging)."	,						)
		parser.add_argument(
		    "--export_to_hub"	,						action="store_true"	,						)

		UpperCamelCase =						parser.parse_args()
		return args


def    a__       (       _SCREAMING_SNAKE_CASE     ):




		"""simple docstring"""


		UpperCamelCase =						datasets.load_dataset(args.dataset_name	,						args.dataset_config	,						split="train"     )

		if args.limit is not None:
				UpperCamelCase =						min(len(_SCREAMING_SNAKE_CASE     )	,						args.limit     )
				UpperCamelCase =						dataset.select(range(_SCREAMING_SNAKE_CASE     )     )
				logger.info(F"Limiting the dataset to {args.limit} entries."     )

		def batch_iterator():
				for i in range(0	,						len(_SCREAMING_SNAKE_CASE     )	,						args.batch_size     ):
						yield dataset[i : i + args.batch_size]["text"]

    # Prepare the tokenizer.
		UpperCamelCase =						Tokenizer(Unigram()     )
		UpperCamelCase =						normalizers.Sequence([normalizers.Replace("``"	,						"\""     ), normalizers.Replace("''"	,						"\""     )]     )
		UpperCamelCase =						pre_tokenizers.Metaspace()

		# Prepare the trainer.
		UpperCamelCase =						UnigramTrainer(
		    unk_token="<unk>"	,						special_tokens=["[CLS]", "[SEP]", "<unk>", "<pad>", "[MASK]"]	,						vocab_size=args.vocab_size	,						)

		logger.info("Training the tokenizer."     )
		tokenizer.train_from_iterator(batch_iterator()	,						trainer=_SCREAMING_SNAKE_CASE     )
		logger.info("Tokenizer training complete!"     )

		UpperCamelCase =						tokenizer.token_to_id("[CLS]"     )
		UpperCamelCase =						tokenizer.token_to_id("[SEP]"     )
		UpperCamelCase =						processors.TemplateProcessing(
		    single="[CLS]:0 $A:0 [SEP]:0"	,						pair="[CLS]:0 $A:0 [SEP]:0 $B:1 [SEP]:1"	,						special_tokens=[
		        ("[CLS]", cls_token_id),
		        ("[SEP]", sep_token_id),
		    ]	,						)
		UpperCamelCase =						decoders.Metaspace()

		if args.export_to_hub:
				logger.info("Exporting the trained tokenzier to Hub."     )
				UpperCamelCase =						AlbertTokenizerFast(tokenizer_object=_SCREAMING_SNAKE_CASE     )
				new_tokenizer.push_to_hub("unigram-tokenizer-dataset"     )


if __name__ == "__main__":
						lowerCAmelCase__        =      parse_args()
						main(args)






