

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


__A      		=     random.Random()

if is_torch_available():
       import torch



def 				snake_case_(_UpperCamelCase			,	_UpperCamelCase=1.0			,	_UpperCamelCase=None			,	_UpperCamelCase=None					)							->     Optional[int]:



  """simple docstring"""






  if rng is None:
    _snake_case						=							global_rng

  _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   lowercase_   (  unittest.TestCase					):



      def __init__(      self				:    Dict      , A__				:    List[Any]      , A__				:    Optional[int]=7      , A__				:    Optional[int]=400      , A__				:    Tuple=2000      , A__				:    str=1      , A__				:    str=0.0      , A__				:    Optional[Any]=16000      , A__				:    List[Any]=True      , A__				:    Optional[Any]=True      , )   ->					List[Any]:
        _snake_case						=							parent
        _snake_case						=							batch_size
        _snake_case						=							min_seq_length
        _snake_case						=							max_seq_length
        _snake_case						=							(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        _snake_case						=							feature_size
        _snake_case						=							padding_value
        _snake_case						=							sampling_rate
        _snake_case						=							return_attention_mask
        _snake_case						=							do_normalize



      def 						UpperCamelCase_		(      self				:    int		)   ->					Optional[int]:
        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 						UpperCamelCase_		(      self				:    List[Any]      , A__				:    Tuple=False      , A__				:    Tuple=False		)   ->					Union[str, Any]:
        def _flatten(A__				:    List[str]		):
          return list(itertools.chain(*A__		)		)

        if equal_length:
          _snake_case						=							floats_list((self.batch_size, self.max_seq_length)		)
        else:
          # make sure that inputs increase in size
          _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:
          _snake_case						=							[np.asarray(A__		) for x in speech_inputs]

        return speech_inputs



@require_torch
@require_torchaudio
class   lowercase_   (  __lowercase       ,							unittest.TestCase					):
      UpperCamelCase_      :		List[Any]										=    ASTFeatureExtractor



      def 						UpperCamelCase_		(      self				:    Optional[Any]		)   ->					List[Any]:
        _snake_case						=							ASTFeatureExtractionTester(self		)



      def 						UpperCamelCase_		(      self				:    Dict		)   ->					Optional[Any]:
        # Tests that all call wrap to encode_plus and batch_encode_plus
        _snake_case						=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
        # create three inputs of length 800, 1000, and 1200
        _snake_case						=							[floats_list((1, x)		)[0] for x in range(800      , 1400      , 200		)]
        _snake_case						=							[np.asarray(A__		) for speech_input in speech_inputs]

        # Test not batched input
        _snake_case						=							feat_extract(speech_inputs[0]      , return_tensors='''np'''		).input_values
        _snake_case						=							feat_extract(np_speech_inputs[0]      , return_tensors='''np'''		).input_values
        self.assertTrue(np.allclose(A__      , A__      , atol=1e-3		)		)

        # Test batched
        _snake_case						=							feat_extract(A__      , padding=A__      , return_tensors='''np'''		).input_values
        _snake_case						=							feat_extract(A__      , padding=A__      , return_tensors='''np'''		).input_values
        for enc_seq_a, enc_seq_a in zip(A__      , A__		):
          self.assertTrue(np.allclose(A__      , A__      , atol=1e-3		)		)

        # Test 2-D numpy arrays are batched.
        _snake_case						=							[floats_list((1, x)		)[0] for x in (800, 800, 800)]
        _snake_case						=							np.asarray(A__		)
        _snake_case						=							feat_extract(A__      , return_tensors='''np'''		).input_values
        _snake_case						=							feat_extract(A__      , return_tensors='''np'''		).input_values
        for enc_seq_a, enc_seq_a in zip(A__      , A__		):
          self.assertTrue(np.allclose(A__      , A__      , atol=1e-3		)		)



      @require_torch
      def 						UpperCamelCase_		(      self				:    Dict		)   ->					str:
        import torch

        _snake_case						=							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()		)
        _snake_case						=							np.random.rand(100		).astype(np.floataa		)
        _snake_case						=							np_speech_inputs.tolist()

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



      def 						UpperCamelCase_		(      self				:    List[Any]      , A__				:    Optional[Any]		)   ->					Tuple:
        from datasets import load_dataset

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

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







      @require_torch
      def 						UpperCamelCase_		(      self				:    List[str]		)   ->					Dict:
        # fmt: off
        _snake_case						=							torch.tensor(
            [-0.9894, -1.2776, -0.9066, -1.2776, -0.9349, -1.2609, -1.0386, -1.2776,
             -1.1561, -1.2776, -1.2052, -1.2723, -1.2190, -1.2132, -1.2776, -1.1133,
             -1.1953, -1.1343, -1.1584, -1.2203, -1.1770, -1.2474, -1.2381, -1.1936,
             -0.9270, -0.8317, -0.8049, -0.7706, -0.7565, -0.7869]		)
        # fmt: on

        _snake_case						=							self._load_datasamples(1		)
        _snake_case						=							ASTFeatureExtractor()
        _snake_case						=							feature_extractor(A__      , return_tensors='''pt'''		).input_values
        self.assertEquals(input_values.shape      , (1, 1024, 128)		)
        self.assertTrue(torch.allclose(input_values[0, 0, :30]      , A__      , atol=1e-4		)		)




import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import BatchEncoding, PreTrainedTokenizer
from ...utils import logging


__A      		=     logging.get_logger(__name__)

__A      		=     '''▁'''

__A      		=     {
    '''vocab_file''': '''vocab.json''',
    '''spm_file''': '''sentencepiece.bpe.model''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

__A      		=     {
    '''vocab_file''': {
        '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json''',
        '''facebook/m2m100_1.2B''': '''https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json''',
    },
    '''spm_file''': {
        '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model''',
        '''facebook/m2m100_1.2B''': '''https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model''',
    },
    '''tokenizer_config_file''': {
        '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json''',
        '''facebook/m2m100_1.2B''': '''https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json''',
    },
}

__A      		=     {
    '''facebook/m2m100_418M''': 10_24,
}

# fmt: off
__A      		=     {
    '''m2m100''': ['''af''', '''am''', '''ar''', '''ast''', '''az''', '''ba''', '''be''', '''bg''', '''bn''', '''br''', '''bs''', '''ca''', '''ceb''', '''cs''', '''cy''', '''da''', '''de''', '''el''', '''en''', '''es''', '''et''', '''fa''', '''ff''', '''fi''', '''fr''', '''fy''', '''ga''', '''gd''', '''gl''', '''gu''', '''ha''', '''he''', '''hi''', '''hr''', '''ht''', '''hu''', '''hy''', '''id''', '''ig''', '''ilo''', '''is''', '''it''', '''ja''', '''jv''', '''ka''', '''kk''', '''km''', '''kn''', '''ko''', '''lb''', '''lg''', '''ln''', '''lo''', '''lt''', '''lv''', '''mg''', '''mk''', '''ml''', '''mn''', '''mr''', '''ms''', '''my''', '''ne''', '''nl''', '''no''', '''ns''', '''oc''', '''or''', '''pa''', '''pl''', '''ps''', '''pt''', '''ro''', '''ru''', '''sd''', '''si''', '''sk''', '''sl''', '''so''', '''sq''', '''sr''', '''ss''', '''su''', '''sv''', '''sw''', '''ta''', '''th''', '''tl''', '''tn''', '''tr''', '''uk''', '''ur''', '''uz''', '''vi''', '''wo''', '''xh''', '''yi''', '''yo''', '''zh''', '''zu'''],
    '''wmt21''': ['''en''', '''ha''', '''is''', '''ja''', '''cs''', '''ru''', '''zh''', '''de''']
}



class   lowercase_   (  __lowercase					):
      UpperCamelCase_      :		str										=    VOCAB_FILES_NAMES
      UpperCamelCase_      :		Optional[int]										=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      UpperCamelCase_      :		Dict										=    PRETRAINED_VOCAB_FILES_MAP
      UpperCamelCase_      :		Union[str, Any]										=    ["input_ids", "attention_mask"]

      UpperCamelCase_      :		List[int]										=    []
      UpperCamelCase_      :		List[int]										=    []



      def __init__(      self				:    str      , A__				:    str      , A__				:    Optional[Any]      , A__				:    Union[str, Any]=None      , A__				:    Dict=None      , A__				:    Any="<s>"      , A__				:    Union[str, Any]="</s>"      , A__				:    Tuple="</s>"      , A__				:    Dict="<pad>"      , A__				:    List[Any]="<unk>"      , A__				:    str="m2m100"      , A__				:    Optional[Dict[str, Any]] = None      , A__				:    List[Any]=8      , **A__				:    Union[str, Any]      , )   ->					None:
        _snake_case						=							{} if sp_model_kwargs is None else sp_model_kwargs

        _snake_case						=							language_codes
        _snake_case						=							FAIRSEQ_LANGUAGE_CODES[language_codes]
        _snake_case						=							{lang_code: f"""__{lang_code}__""" for lang_code in fairseq_language_code}

        _snake_case						=							kwargs.get('''additional_special_tokens'''      , []		)
        kwargs["additional_special_tokens"] += [
            self.get_lang_token(A__		)
            for lang_code in fairseq_language_code
            if self.get_lang_token(A__		) not in kwargs["additional_special_tokens"]
        ]

        super().__init__(
            src_lang=A__      , tgt_lang=A__      , bos_token=A__      , eos_token=A__      , sep_token=A__      , unk_token=A__      , pad_token=A__      , language_codes=A__      , sp_model_kwargs=self.sp_model_kwargs      , num_madeup_words=A__      , **A__      , )

        _snake_case						=							vocab_file
        _snake_case						=							load_json(A__		)
        _snake_case						=							{v: k for k, v in self.encoder.items()}
        _snake_case						=							spm_file
        _snake_case						=							load_spm(A__      , self.sp_model_kwargs		)

        _snake_case						=							len(self.encoder		)

        _snake_case						=							{
            self.get_lang_token(A__		): self.encoder_size + i for i, lang_code in enumerate(A__		)
        }
        _snake_case						=							{lang_code: self.encoder_size + i for i, lang_code in enumerate(A__		)}
        _snake_case						=							{v: k for k, v in self.lang_token_to_id.items()}

        _snake_case						=							src_lang if src_lang is not None else '''en'''
        _snake_case						=							tgt_lang
        _snake_case						=							self.get_lang_id(self._src_lang		)
        self.set_src_lang_special_tokens(self._src_lang		)

        _snake_case						=							num_madeup_words



      @property
      def 						UpperCamelCase_		(      self				:    int		)   ->					int:
        return len(self.encoder		) + len(self.lang_token_to_id		)



      @property
      def 						UpperCamelCase_		(      self				:    Dict		)   ->					str:
        return self._src_lang



      @src_lang.setter
      def 						UpperCamelCase_		(      self				:    List[str]      , A__				:    str		)   ->					None:
        _snake_case						=							new_src_lang
        self.set_src_lang_special_tokens(self._src_lang		)



      def 						UpperCamelCase_		(      self				:    Any      , A__				:    str		)   ->					List[str]:
        return self.sp_model.encode(A__      , out_type=A__		)



      def 						UpperCamelCase_		(      self				:    Optional[int]      , A__				:    Dict		)   ->					str:
        if token in self.lang_token_to_id:
          return self.lang_token_to_id[token]
        return self.encoder.get(A__      , self.encoder[self.unk_token]		)



      def 						UpperCamelCase_		(      self				:    Union[str, Any]      , A__				:    int		)   ->					str:
        if index in self.id_to_lang_token:
          return self.id_to_lang_token[index]
        return self.decoder.get(A__      , self.unk_token		)



      def 						UpperCamelCase_		(      self				:    Optional[int]      , A__				:    Optional[int]		)   ->					List[Any]:
        _snake_case						=							[]
        _snake_case						=							''''''
        for token in tokens:
          # make sure that special tokens are not decoded using sentencepiece model
          if token in self.all_special_tokens:
            out_string += self.sp_model.decode(A__		) + token
            _snake_case						=							[]
          else:
            current_sub_tokens.append(A__		)
        out_string += self.sp_model.decode(A__		)
        return out_string.strip()



      def 						UpperCamelCase_		(      self				:    str      , A__				:    List[int]      , A__				:    Optional[List[int]] = None      , A__				:    bool = False		)   ->					List[int]:

        if already_has_special_tokens:
          return super().get_special_tokens_mask(
              token_ids_a=A__      , token_ids_a=A__      , already_has_special_tokens=A__		)

        _snake_case						=							[1] * len(self.prefix_tokens		)
        _snake_case						=							[1] * len(self.suffix_tokens		)
        if token_ids_a is None:
          return prefix_ones + ([0] * len(A__		)) + suffix_ones
        return prefix_ones + ([0] * len(A__		)) + ([0] * len(A__		)) + suffix_ones



      def 						UpperCamelCase_		(      self				:    Tuple      , A__				:    List[int]      , A__				:    Optional[List[int]] = None		)   ->					List[int]:
        if token_ids_a is None:
          return self.prefix_tokens + token_ids_a + self.suffix_tokens
        # We don't expect to process pairs, but leave the pair logic for API consistency
        return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens



      def 						UpperCamelCase_		(      self				:    str		)   ->					Dict:
        _snake_case						=							{self.convert_ids_to_tokens(A__		): i for i in range(self.vocab_size		)}
        vocab.update(self.added_tokens_encoder		)
        return vocab



      def __getstate__(      self				:    str		)   ->					Dict:
        _snake_case						=							self.__dict__.copy()
        _snake_case						=							None
        return state



      def __setstate__(      self				:    Union[str, Any]      , A__				:    Dict		)   ->					None:
        _snake_case						=							d

        # for backward compatibility
        if not hasattr(self      , '''sp_model_kwargs'''		):
          _snake_case						=							{}

        _snake_case						=							load_spm(self.spm_file      , self.sp_model_kwargs		)



      def 						UpperCamelCase_		(      self				:    Any      , A__				:    str      , A__				:    Optional[str] = None		)   ->					Tuple[str]:
        _snake_case						=							Path(A__		)
        if not save_dir.is_dir():
          raise OSError(f"""{save_directory} should be a directory"""		)
        _snake_case						=							save_dir / (
            (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''vocab_file''']
        )
        _snake_case						=							save_dir / (
            (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''spm_file''']
        )

        save_json(self.encoder      , A__		)

        if os.path.abspath(self.spm_file		) != os.path.abspath(A__		) and os.path.isfile(self.spm_file		):
          copyfile(self.spm_file      , A__		)
        elif not os.path.isfile(self.spm_file		):
          with open(A__      , '''wb'''		) as fi:
            _snake_case						=							self.sp_model.serialized_model_proto()
            fi.write(A__		)

        return (str(A__		), str(A__		))



      def 						UpperCamelCase_		(      self				:    Optional[int]      , A__				:    List[str]      , A__				:    str = "en"      , A__				:    Optional[List[str]] = None      , A__				:    str = "ro"      , **A__				:    List[Any]      , )   ->					BatchEncoding:
        _snake_case						=							src_lang
        _snake_case						=							tgt_lang
        self.set_src_lang_special_tokens(self.src_lang		)
        return super().prepare_seqaseq_batch(A__      , A__      , **A__		)



      def 						UpperCamelCase_		(      self				:    List[str]      , A__				:    int      , A__				:    Optional[str]      , A__				:    Optional[str]      , **A__				:    Union[str, Any]		)   ->					Tuple:
        if src_lang is None or tgt_lang is None:
          raise ValueError('''Translation requires a `src_lang` and a `tgt_lang` for this model'''		)
        _snake_case						=							src_lang
        _snake_case						=							self(A__      , add_special_tokens=A__      , **A__		)
        _snake_case						=							self.get_lang_id(A__		)
        _snake_case						=							tgt_lang_id
        return inputs



      def 						UpperCamelCase_		(      self				:    Dict		)   ->					Optional[Any]:
        self.set_src_lang_special_tokens(self.src_lang		)



      def 						UpperCamelCase_		(      self				:    Optional[Any]		)   ->					Dict:
        self.set_tgt_lang_special_tokens(self.tgt_lang		)



      def 						UpperCamelCase_		(      self				:    List[Any]      , A__				:    str		)   ->					None:
        _snake_case						=							self.get_lang_token(A__		)
        _snake_case						=							self.lang_token_to_id[lang_token]
        _snake_case						=							[self.cur_lang_id]
        _snake_case						=							[self.eos_token_id]



      def 						UpperCamelCase_		(      self				:    List[str]      , A__				:    str		)   ->					None:
        _snake_case						=							self.get_lang_token(A__		)
        _snake_case						=							self.lang_token_to_id[lang_token]
        _snake_case						=							[self.cur_lang_id]
        _snake_case						=							[self.eos_token_id]



      def 						UpperCamelCase_		(      self				:    Dict      , A__				:    str		)   ->					str:
        return self.lang_code_to_token[lang]







      def 						UpperCamelCase_		(      self				:    Tuple      , A__				:    str		)   ->					int:
        _snake_case						=							self.get_lang_token(A__		)
        return self.lang_token_to_id[lang_token]



def 				snake_case_(_UpperCamelCase			,	_UpperCamelCase					)							->     sentencepiece.SentencePieceProcessor:



  """simple docstring"""






  _snake_case						=							sentencepiece.SentencePieceProcessor(**_UpperCamelCase					)
  spm.Load(str(_UpperCamelCase					)					)
  return spm



def 				snake_case_(_UpperCamelCase					)							->     Union[Dict, List]:



  """simple docstring"""






  with open(_UpperCamelCase			,	'''r'''					) as f:
    return json.load(_UpperCamelCase					)



def 				snake_case_(_UpperCamelCase			,	_UpperCamelCase					)							->     None:



  """simple docstring"""






  with open(_UpperCamelCase			,	'''w'''					) as f:
    json.dump(_UpperCamelCase			,	_UpperCamelCase			,	indent=2					)

