import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


__lowerCamelCase      :			Union[str, Any]							       =		random.Random()

if is_torch_available():
   import torch



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	int			,					__UpperCamelCase   :	Dict=1.0			,					__UpperCamelCase   :	Optional[int]=None			,					__UpperCamelCase   :	List[Any]=None     )							-> List[str]:







  """simple docstring"""





  if rng is None:
    SCREAMING_SNAKE_CASE__      							=				global_rng

  SCREAMING_SNAKE_CASE__      							=				[]
  for batch_idx in range(shape[0]     ):
    values.append([]     )
    for _ in range(shape[1]     ):
      values[-1].append(rng.random() * scale     )

  return values







class __snake_case    (						unittest.TestCase		):




   def __init__(   self      :    Any	,     _lowercase      :    List[str]	,     _lowercase      :    Optional[Any]=7	,     _lowercase      :    List[str]=4_00	,     _lowercase      :    int=20_00	,     _lowercase      :    Optional[int]=1	,     _lowercase      :    Optional[int]=0.0	,     _lowercase      :    Optional[Any]=1_60_00	,     _lowercase      :    Any=True	,     _lowercase      :    int=True	,     ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				parent
     SCREAMING_SNAKE_CASE__      							=				batch_size
     SCREAMING_SNAKE_CASE__      							=				min_seq_length
     SCREAMING_SNAKE_CASE__      							=				max_seq_length
     SCREAMING_SNAKE_CASE__      							=				(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
     SCREAMING_SNAKE_CASE__      							=				feature_size
     SCREAMING_SNAKE_CASE__      							=				padding_value
     SCREAMING_SNAKE_CASE__      							=				sampling_rate
     SCREAMING_SNAKE_CASE__      							=				return_attention_mask
     SCREAMING_SNAKE_CASE__      							=				do_normalize




   def      __a						(   self      :    Optional[int]				):



     """simple docstring"""






     return {
         "feature_size": self.feature_size,
         "padding_value": self.padding_value,
         "sampling_rate": self.sampling_rate,
         "return_attention_mask": self.return_attention_mask,
         "do_normalize": self.do_normalize,
     }




   def      __a						(   self      :    str	,     _lowercase      :    str=False	,     _lowercase      :    int=False				):



     """simple docstring"""






     def _flatten(_lowercase      :    Dict				):
       return list(itertools.chain(*_lowercase				)				)

     if equal_length:
       SCREAMING_SNAKE_CASE__      							=				floats_list((self.batch_size, self.max_seq_length)				)
     else:
       # make sure that inputs increase in size
       SCREAMING_SNAKE_CASE__      							=				[
           _flatten(floats_list((x, self.feature_size)				)				)
           for x in range(self.min_seq_length	,     self.max_seq_length	,     self.seq_length_diff				)
       ]

     if numpify:
       SCREAMING_SNAKE_CASE__      							=				[np.asarray(_lowercase				) for x in speech_inputs]

     return speech_inputs







@require_torch
@require_torchaudio
class __snake_case    (						lowerCamelCase_       ,		unittest.TestCase		):
   lowerCAmelCase_														=							ASTFeatureExtractor




   def      __a						(   self      :    str				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				ASTFeatureExtractionTester(self				)




   def      __a						(   self      :    str				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()				)
     # create three inputs of length 800, 1000, and 1200
     SCREAMING_SNAKE_CASE__      							=				[floats_list((1, x)				)[0] for x in range(8_00	,     14_00	,     2_00				)]
     SCREAMING_SNAKE_CASE__      							=				[np.asarray(_lowercase				) for speech_input in speech_inputs]

     # Test not batched input
     SCREAMING_SNAKE_CASE__      							=				feat_extract(speech_inputs[0]	,     return_tensors="""np"""				).input_values
     SCREAMING_SNAKE_CASE__      							=				feat_extract(np_speech_inputs[0]	,     return_tensors="""np"""				).input_values
     self.assertTrue(np.allclose(_lowercase	,     _lowercase	,     atol=1E-3				)				)

     # Test batched
     SCREAMING_SNAKE_CASE__      							=				feat_extract(_lowercase	,     padding=_lowercase	,     return_tensors="""np"""				).input_values
     SCREAMING_SNAKE_CASE__      							=				feat_extract(_lowercase	,     padding=_lowercase	,     return_tensors="""np"""				).input_values
     for enc_seq_a, enc_seq_a in zip(_lowercase	,     _lowercase				):
       self.assertTrue(np.allclose(_lowercase	,     _lowercase	,     atol=1E-3				)				)

     # Test 2-D numpy arrays are batched.
     SCREAMING_SNAKE_CASE__      							=				[floats_list((1, x)				)[0] for x in (8_00, 8_00, 8_00)]
     SCREAMING_SNAKE_CASE__      							=				np.asarray(_lowercase				)
     SCREAMING_SNAKE_CASE__      							=				feat_extract(_lowercase	,     return_tensors="""np"""				).input_values
     SCREAMING_SNAKE_CASE__      							=				feat_extract(_lowercase	,     return_tensors="""np"""				).input_values
     for enc_seq_a, enc_seq_a in zip(_lowercase	,     _lowercase				):
       self.assertTrue(np.allclose(_lowercase	,     _lowercase	,     atol=1E-3				)				)




   @require_torch
   def      __a						(   self      :    Optional[int]				):



     """simple docstring"""






     import torch

     SCREAMING_SNAKE_CASE__      							=				self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()				)
     SCREAMING_SNAKE_CASE__      							=				np.random.rand(1_00				).astype(np.floataa				)
     SCREAMING_SNAKE_CASE__      							=				np_speech_inputs.tolist()

     for inputs in [py_speech_inputs, np_speech_inputs]:
       SCREAMING_SNAKE_CASE__      							=				feature_extractor.pad([{"""input_values""": inputs}]	,     return_tensors="""np"""				)
       self.assertTrue(np_processed.input_values.dtype == np.floataa				)
       SCREAMING_SNAKE_CASE__      							=				feature_extractor.pad([{"""input_values""": inputs}]	,     return_tensors="""pt"""				)
       self.assertTrue(pt_processed.input_values.dtype == torch.floataa				)




   def      __a						(   self      :    str	,     _lowercase      :    Optional[Any]				):



     """simple docstring"""






     from datasets import load_dataset

     SCREAMING_SNAKE_CASE__      							=				load_dataset("""hf-internal-testing/librispeech_asr_dummy"""	,     """clean"""	,     split="""validation"""				)
     # automatic decoding with librispeech
     SCREAMING_SNAKE_CASE__      							=				ds.sort("""id"""				).select(range(_lowercase				)				)[:num_samples]["""audio"""]

     return [x["array"] for x in speech_samples]




   @require_torch
   def      __a						(   self      :    List[Any]				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				torch.tensor(
         [-0.98_94, -1.27_76, -0.90_66, -1.27_76, -0.93_49, -1.26_09, -1.03_86, -1.27_76,
          -1.15_61, -1.27_76, -1.20_52, -1.27_23, -1.21_90, -1.21_32, -1.27_76, -1.11_33,
          -1.19_53, -1.13_43, -1.15_84, -1.22_03, -1.17_70, -1.24_74, -1.23_81, -1.19_36,
          -0.92_70, -0.83_17, -0.80_49, -0.77_06, -0.75_65, -0.78_69]				)
     # fmt: on

     SCREAMING_SNAKE_CASE__      							=				self._load_datasamples(1				)
     SCREAMING_SNAKE_CASE__      							=				ASTFeatureExtractor()
     SCREAMING_SNAKE_CASE__      							=				feature_extractor(_lowercase	,     return_tensors="""pt"""				).input_values
     self.assertEquals(input_values.shape	,     (1, 10_24, 1_28)				)
     self.assertTrue(torch.allclose(input_values[0, 0, :30]	,     _lowercase	,     atol=1E-4				)				)

import argparse
import os

import transformers

from .convert_slow_tokenizer import SLOW_TO_FAST_CONVERTERS
from .utils import logging


logging.set_verbosity_info()

__lowerCamelCase      :			int							       =		logging.get_logger(__name__)


__lowerCamelCase      :			Tuple							       =		{name: getattr(transformers, name + '''Fast''') for name in SLOW_TO_FAST_CONVERTERS}



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	int			,					__UpperCamelCase   :	Optional[int]			,					__UpperCamelCase   :	str			,					__UpperCamelCase   :	Optional[Any]     )							-> Union[str, Any]:







  """simple docstring"""





  if tokenizer_name is not None and tokenizer_name not in TOKENIZER_CLASSES:
    raise ValueError(f"""Unrecognized tokenizer name, should be one of {list(TOKENIZER_CLASSES.keys()     )}."""     )

  if tokenizer_name is None:
    SCREAMING_SNAKE_CASE__      							=				TOKENIZER_CLASSES
  else:
    SCREAMING_SNAKE_CASE__      							=				{tokenizer_name: getattr(__UpperCamelCase			,					tokenizer_name + """Fast"""     )}

  logger.info(f"""Loading tokenizer classes: {tokenizer_names}"""     )

  for tokenizer_name in tokenizer_names:
    SCREAMING_SNAKE_CASE__      							=				TOKENIZER_CLASSES[tokenizer_name]

    SCREAMING_SNAKE_CASE__      							=				True
    if checkpoint_name is None:
      SCREAMING_SNAKE_CASE__      							=				list(tokenizer_class.max_model_input_sizes.keys()     )
    else:
      SCREAMING_SNAKE_CASE__      							=				[checkpoint_name]

    logger.info(f"""For tokenizer {tokenizer_class.__class__.__name__} loading checkpoints: {checkpoint_names}"""     )

    for checkpoint in checkpoint_names:
      logger.info(f"""Loading {tokenizer_class.__class__.__name__} {checkpoint}"""     )

      # Load tokenizer
      SCREAMING_SNAKE_CASE__      							=				tokenizer_class.from_pretrained(__UpperCamelCase			,					force_download=__UpperCamelCase     )

      # Save fast tokenizer
      logger.info(f"""Save fast tokenizer to {dump_path} with prefix {checkpoint} add_prefix {add_prefix}"""     )

      # For organization names we create sub-directories
      if "/" in checkpoint:
        SCREAMING_SNAKE_CASE__							,					SCREAMING_SNAKE_CASE__      							=				checkpoint.split("""/"""     )
        SCREAMING_SNAKE_CASE__      							=				os.path.join(__UpperCamelCase			,					__UpperCamelCase     )
      elif add_prefix:
        SCREAMING_SNAKE_CASE__      							=				checkpoint
        SCREAMING_SNAKE_CASE__      							=				dump_path
      else:
        SCREAMING_SNAKE_CASE__      							=				None
        SCREAMING_SNAKE_CASE__      							=				dump_path

      logger.info(f"""=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}"""     )

      if checkpoint in list(tokenizer.pretrained_vocab_files_map.values()     )[0]:
        SCREAMING_SNAKE_CASE__      							=				list(tokenizer.pretrained_vocab_files_map.values()     )[0][checkpoint]
        SCREAMING_SNAKE_CASE__      							=				file_path.split(__UpperCamelCase     )[-1][0]
        if next_char == "/":
          SCREAMING_SNAKE_CASE__      							=				os.path.join(__UpperCamelCase			,					__UpperCamelCase     )
          SCREAMING_SNAKE_CASE__      							=				None

        logger.info(f"""=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}"""     )

      SCREAMING_SNAKE_CASE__      							=				tokenizer.save_pretrained(
          __UpperCamelCase			,					legacy_format=__UpperCamelCase			,					filename_prefix=__UpperCamelCase     )
      logger.info(f"""=> File names {file_names}"""     )

      for file_name in file_names:
        if not file_name.endswith("""tokenizer.json"""     ):
          os.remove(__UpperCamelCase     )
          logger.info(f"""=> removing {file_name}"""     )


if __name__ == "__main__":
   __lowerCamelCase      :			str							       =		argparse.ArgumentParser()
   # Required parameters
   parser.add_argument(
       '''--dump_path''', default=None, type=str, required=True, help='''Path to output generated fast tokenizer files.'''
   )
   parser.add_argument(
       '''--tokenizer_name''',
       default=None,
       type=str,
       help=(
           F"""Optional tokenizer type selected in the list of {list(TOKENIZER_CLASSES.keys())}. If not given, will """
           '''download and convert all the checkpoints from AWS.'''
       ),
   )
   parser.add_argument(
       '''--checkpoint_name''',
       default=None,
       type=str,
       help='''Optional checkpoint name. If not given, will download and convert the canonical checkpoints from AWS.''',
   )
   parser.add_argument(
       '''--force_download''',
       action='''store_true''',
       help='''Re-download checkpoints.''',
   )
   __lowerCamelCase      :			Any							       =		parser.parse_args()

   convert_slow_checkpoint_to_fast(args.tokenizer_name, args.checkpoint_name, args.dump_path, args.force_download)
