


'''simple docstring'''





import os
import shutil
import sys
import tempfile
import unittest
from pathlib import Path

import pytest

import transformers
from transformers import (
    BERT_PRETRAINED_CONFIG_ARCHIVE_MAP,
    GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP,
    AutoTokenizer,
    BertConfig,
    BertTokenizer,
    BertTokenizerFast,
    CTRLTokenizer,
    GPTaTokenizer,
    GPTaTokenizerFast,
    PreTrainedTokenizerFast,
    RobertaTokenizer,
    RobertaTokenizerFast,
    is_tokenizers_available,
)
from transformers.models.auto.configuration_auto import CONFIG_MAPPING, AutoConfig
from transformers.models.auto.tokenization_auto import (
    TOKENIZER_MAPPING,
    get_tokenizer_config,
    tokenizer_class_from_name,
)
from transformers.models.roberta.configuration_roberta import RobertaConfig
from transformers.testing_utils import (
    DUMMY_DIFF_TOKENIZER_IDENTIFIER,
    DUMMY_UNKNOWN_IDENTIFIER,
    SMALL_MODEL_IDENTIFIER,
    RequestCounter,
    require_tokenizers,
    slow,
)


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
    from test_module.custom_tokenization_fast import CustomTokenizerFast




class 				UpperCAmelCase			(			unittest.TestCase						):


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[int] ):
    UpperCAmelCase__	:Optional[Any]      =      0


 @slow
 def        __SCREAMING_SNAKE_CASE					(      self							:	Any ):
    for model_name in (x for x in BERT_PRETRAINED_CONFIG_ARCHIVE_MAP.keys() if "japanese" not in x):
       UpperCAmelCase__	:List[Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
       self.assertIsNotNone(__lowerCamelCase )
       self.assertIsInstance(__lowerCamelCase			,       (BertTokenizer, BertTokenizerFast) )
       self.assertGreater(len(__lowerCamelCase )			,       0 )

    for model_name in GPT2_PRETRAINED_CONFIG_ARCHIVE_MAP.keys():
       UpperCAmelCase__	:Union[str, Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
       self.assertIsNotNone(__lowerCamelCase )
       self.assertIsInstance(__lowerCamelCase			,       (GPTaTokenizer, GPTaTokenizerFast) )
       self.assertGreater(len(__lowerCamelCase )			,       0 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[str] ):
    UpperCAmelCase__	:List[Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
    self.assertIsInstance(__lowerCamelCase			,       (BertTokenizer, BertTokenizerFast) )
    self.assertEqual(tokenizer.vocab_size			,       1_2 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    UpperCAmelCase__	:List[Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
    self.assertIsInstance(__lowerCamelCase			,       (RobertaTokenizer, RobertaTokenizerFast) )
    self.assertEqual(tokenizer.vocab_size			,       2_0 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int ):
    UpperCAmelCase__	:int      =      AutoConfig.from_pretrained(__lowerCamelCase )
    self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )
    # Check that tokenizer_type ≠ model_type
    UpperCAmelCase__	:str      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       config=__lowerCamelCase )
    self.assertIsInstance(__lowerCamelCase			,       (BertTokenizer, BertTokenizerFast) )
    self.assertEqual(tokenizer.vocab_size			,       1_2 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    with tempfile.TemporaryDirectory() as tmp_dir:
       shutil.copy('''./tests/fixtures/vocab.txt'''			,       os.path.join(__lowerCamelCase			,       '''vocab.txt''' ) )

       UpperCAmelCase__	:Optional[Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       tokenizer_type='''bert'''			,       use_fast=__lowerCamelCase )
       self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )

    with tempfile.TemporaryDirectory() as tmp_dir:
       shutil.copy('''./tests/fixtures/vocab.json'''			,       os.path.join(__lowerCamelCase			,       '''vocab.json''' ) )
       shutil.copy('''./tests/fixtures/merges.txt'''			,       os.path.join(__lowerCamelCase			,       '''merges.txt''' ) )

       UpperCAmelCase__	:List[str]      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       tokenizer_type='''gpt2'''			,       use_fast=__lowerCamelCase )
       self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict ):
    with tempfile.TemporaryDirectory() as tmp_dir:
       shutil.copy('''./tests/fixtures/vocab.txt'''			,       os.path.join(__lowerCamelCase			,       '''vocab.txt''' ) )

       UpperCAmelCase__	:Any      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       tokenizer_type='''bert''' )
       self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )

    with tempfile.TemporaryDirectory() as tmp_dir:
       shutil.copy('''./tests/fixtures/vocab.json'''			,       os.path.join(__lowerCamelCase			,       '''vocab.json''' ) )
       shutil.copy('''./tests/fixtures/merges.txt'''			,       os.path.join(__lowerCamelCase			,       '''merges.txt''' ) )

       UpperCAmelCase__	:Dict      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       tokenizer_type='''gpt2''' )
       self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    with pytest.raises(__lowerCamelCase ):
       AutoTokenizer.from_pretrained('''./'''			,       tokenizer_type='''xxx''' )


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	Tuple ):
    for tokenizer_class in [BertTokenizer, BertTokenizerFast, AutoTokenizer]:
       UpperCAmelCase__	:int      =      tokenizer_class.from_pretrained('''wietsedv/bert-base-dutch-cased''' )
       self.assertIsInstance(__lowerCamelCase			,       (BertTokenizer, BertTokenizerFast) )

       if isinstance(__lowerCamelCase			,       __lowerCamelCase ):
          self.assertEqual(tokenizer.basic_tokenizer.do_lower_case			,       __lowerCamelCase )
       else:
          self.assertEqual(tokenizer.do_lower_case			,       __lowerCamelCase )

       self.assertEqual(tokenizer.model_max_length			,       5_1_2 )


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    for tokenizer_class in [BertTokenizer, BertTokenizerFast, AutoTokenizer]:
       with self.assertRaisesRegex(
           __lowerCamelCase			,       '''julien-c/herlolip-not-exists is not a local folder and is not a valid model identifier'''			,       ):
          UpperCAmelCase__	:Optional[int]      =      tokenizer_class.from_pretrained('''julien-c/herlolip-not-exists''' )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Any ):
    # tests: https://github.com/huggingface/transformers/pull/13251
    # 1. models with `-`, e.g. xlm-roberta -> xlm_roberta
    # 2. models that don't remap 1-1 from model-name to model file, e.g., openai-gpt -> openai
    UpperCAmelCase__	:Optional[Any]      =      TOKENIZER_MAPPING.values()
    UpperCAmelCase__	:Optional[Any]      =      []

    for slow_tok, fast_tok in tokenizers:
       if slow_tok is not None:
          tokenizer_names.append(slow_tok.__name__ )

       if fast_tok is not None:
          tokenizer_names.append(fast_tok.__name__ )

    for tokenizer_name in tokenizer_names:
       # must find the right class
       tokenizer_class_from_name(__lowerCamelCase )


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	int ):
    self.assertIsInstance(AutoTokenizer.from_pretrained('''bert-base-cased'''			,       use_fast=__lowerCamelCase )			,       __lowerCamelCase )
    self.assertIsInstance(AutoTokenizer.from_pretrained('''bert-base-cased''' )			,       __lowerCamelCase )


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	Tuple ):
    UpperCAmelCase__	:Optional[Any]      =      AutoTokenizer.from_pretrained('''distilbert-base-uncased'''			,       do_lower_case=__lowerCamelCase )
    UpperCAmelCase__	:Optional[Any]      =      '''Hello, world. How are you?'''
    UpperCAmelCase__	:Tuple      =      tokenizer.tokenize(__lowerCamelCase )
    self.assertEqual('''[UNK]'''			,       tokens[0] )

    UpperCAmelCase__	:Optional[Any]      =      AutoTokenizer.from_pretrained('''microsoft/mpnet-base'''			,       do_lower_case=__lowerCamelCase )
    UpperCAmelCase__	:Union[str, Any]      =      tokenizer.tokenize(__lowerCamelCase )
    self.assertEqual('''[UNK]'''			,       tokens[0] )


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    UpperCAmelCase__	:Optional[int]      =      AutoTokenizer.from_pretrained('''robot-test/dummy-tokenizer-fast-with-model-config''' )
    self.assertEqual(type(__lowerCamelCase )			,       __lowerCamelCase )
    self.assertEqual(tokenizer.model_max_length			,       5_1_2 )
    self.assertEqual(tokenizer.vocab_size			,       3_0_0_0_0 )
    self.assertEqual(tokenizer.unk_token			,       '''[UNK]''' )
    self.assertEqual(tokenizer.padding_side			,       '''right''' )
    self.assertEqual(tokenizer.truncation_side			,       '''right''' )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    UpperCAmelCase__	:Dict      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
    self.assertIsInstance(__lowerCamelCase			,       (BertTokenizer, BertTokenizerFast) )
    with tempfile.TemporaryDirectory() as tmp_dir:
       tokenizer.save_pretrained(__lowerCamelCase )
       UpperCAmelCase__	:Optional[Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )

    self.assertIsInstance(__lowerCamelCase			,       tokenizer.__class__ )
    self.assertEqual(tokenizera.vocab_size			,       1_2 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int ):
    UpperCAmelCase__	:Tuple      =      AutoTokenizer.from_pretrained('''ctrl''' )
    # There is no fast CTRL so this always gives us a slow tokenizer.
    self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    # Check we can load the tokenizer config of an online model.
    UpperCAmelCase__	:Union[str, Any]      =      get_tokenizer_config('''bert-base-cased''' )
    UpperCAmelCase__	:Optional[int]      =      config.pop('''_commit_hash'''			,       __lowerCamelCase )
    # If we ever update bert-base-cased tokenizer config, this dict here will need to be updated.
    self.assertEqual(__lowerCamelCase			,       {'''do_lower_case''': False} )

    # This model does not have a tokenizer_config so we get back an empty dict.
    UpperCAmelCase__	:int      =      get_tokenizer_config(__lowerCamelCase )
    self.assertDictEqual(__lowerCamelCase			,       {} )

    # A tokenizer saved with `save_pretrained` always creates a tokenizer config.
    UpperCAmelCase__	:Optional[Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
    with tempfile.TemporaryDirectory() as tmp_dir:
       tokenizer.save_pretrained(__lowerCamelCase )
       UpperCAmelCase__	:List[Any]      =      get_tokenizer_config(__lowerCamelCase )

    # Check the class of the tokenizer was properly saved (note that it always saves the slow class).
    self.assertEqual(config['''tokenizer_class''']			,       '''BertTokenizer''' )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Any ):
    try:
       AutoConfig.register('''custom'''			,       __lowerCamelCase )

       AutoTokenizer.register(__lowerCamelCase			,       slow_tokenizer_class=__lowerCamelCase )
       # Trying to register something existing in the Transformers library will raise an error
       with self.assertRaises(__lowerCamelCase ):
          AutoTokenizer.register(__lowerCamelCase			,       slow_tokenizer_class=__lowerCamelCase )

       UpperCAmelCase__	:Optional[Any]      =      CustomTokenizer.from_pretrained(__lowerCamelCase )
       with tempfile.TemporaryDirectory() as tmp_dir:
          tokenizer.save_pretrained(__lowerCamelCase )

          UpperCAmelCase__	:List[str]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
          self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )

    finally:
       if "custom" in CONFIG_MAPPING._extra_content:
          del CONFIG_MAPPING._extra_content["custom"]
       if CustomConfig in TOKENIZER_MAPPING._extra_content:
          del TOKENIZER_MAPPING._extra_content[CustomConfig]


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    try:
       AutoConfig.register('''custom'''			,       __lowerCamelCase )

       # Can register in two steps
       AutoTokenizer.register(__lowerCamelCase			,       slow_tokenizer_class=__lowerCamelCase )
       self.assertEqual(TOKENIZER_MAPPING[CustomConfig]			,       (CustomTokenizer, None) )
       AutoTokenizer.register(__lowerCamelCase			,       fast_tokenizer_class=__lowerCamelCase )
       self.assertEqual(TOKENIZER_MAPPING[CustomConfig]			,       (CustomTokenizer, CustomTokenizerFast) )

       del TOKENIZER_MAPPING._extra_content[CustomConfig]
       # Can register in one step
       AutoTokenizer.register(
           __lowerCamelCase			,       slow_tokenizer_class=__lowerCamelCase			,       fast_tokenizer_class=__lowerCamelCase )
       self.assertEqual(TOKENIZER_MAPPING[CustomConfig]			,       (CustomTokenizer, CustomTokenizerFast) )

       # Trying to register something existing in the Transformers library will raise an error
       with self.assertRaises(__lowerCamelCase ):
          AutoTokenizer.register(__lowerCamelCase			,       fast_tokenizer_class=__lowerCamelCase )

       # We pass through a bert tokenizer fast cause there is no converter slow to fast for our new toknizer
       # and that model does not have a tokenizer.json
       with tempfile.TemporaryDirectory() as tmp_dir:
          UpperCAmelCase__	:Union[str, Any]      =      BertTokenizerFast.from_pretrained(__lowerCamelCase )
          bert_tokenizer.save_pretrained(__lowerCamelCase )
          UpperCAmelCase__	:Union[str, Any]      =      CustomTokenizerFast.from_pretrained(__lowerCamelCase )

       with tempfile.TemporaryDirectory() as tmp_dir:
          tokenizer.save_pretrained(__lowerCamelCase )

          UpperCAmelCase__	:Union[str, Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase )
          self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )

          UpperCAmelCase__	:Optional[int]      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       use_fast=__lowerCamelCase )
          self.assertIsInstance(__lowerCamelCase			,       __lowerCamelCase )

    finally:
       if "custom" in CONFIG_MAPPING._extra_content:
          del CONFIG_MAPPING._extra_content["custom"]
       if CustomConfig in TOKENIZER_MAPPING._extra_content:
          del TOKENIZER_MAPPING._extra_content[CustomConfig]


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    # If remote code is not set, we will time out when asking whether to load the model.
    with self.assertRaises(__lowerCamelCase ):
       UpperCAmelCase__	:Any      =      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer''' )
    # If remote code is disabled, we can't load this config.
    with self.assertRaises(__lowerCamelCase ):
       UpperCAmelCase__	:int      =      AutoTokenizer.from_pretrained(
           '''hf-internal-testing/test_dynamic_tokenizer'''			,       trust_remote_code=__lowerCamelCase )

    UpperCAmelCase__	:Optional[Any]      =      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer'''			,       trust_remote_code=__lowerCamelCase )
    self.assertTrue(tokenizer.special_attribute_present )
    # Test tokenizer can be reloaded.
    with tempfile.TemporaryDirectory() as tmp_dir:
       tokenizer.save_pretrained(__lowerCamelCase )
       UpperCAmelCase__	:str      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       trust_remote_code=__lowerCamelCase )
    self.assertTrue(reloaded_tokenizer.special_attribute_present )

    if is_tokenizers_available():
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizerFast''' )
       self.assertEqual(reloaded_tokenizer.__class__.__name__			,       '''NewTokenizerFast''' )

       # Test we can also load the slow version
       UpperCAmelCase__	:Tuple      =      AutoTokenizer.from_pretrained(
           '''hf-internal-testing/test_dynamic_tokenizer'''			,       trust_remote_code=__lowerCamelCase			,       use_fast=__lowerCamelCase )
       self.assertTrue(tokenizer.special_attribute_present )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizer''' )
       # Test tokenizer can be reloaded.
       with tempfile.TemporaryDirectory() as tmp_dir:
          tokenizer.save_pretrained(__lowerCamelCase )
          UpperCAmelCase__	:List[Any]      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       trust_remote_code=__lowerCamelCase			,       use_fast=__lowerCamelCase )
       self.assertEqual(reloaded_tokenizer.__class__.__name__			,       '''NewTokenizer''' )
       self.assertTrue(reloaded_tokenizer.special_attribute_present )
    else:
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizer''' )
       self.assertEqual(reloaded_tokenizer.__class__.__name__			,       '''NewTokenizer''' )


 @require_tokenizers
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):




    class 				UpperCAmelCase			(			_snake_case						):
     UpperCAmelCase				=					False




    class 				UpperCAmelCase			(			_snake_case						):
     UpperCAmelCase				=					NewTokenizer
     UpperCAmelCase				=					False

    try:
       AutoConfig.register('''custom'''			,       __lowerCamelCase )
       AutoTokenizer.register(__lowerCamelCase			,       slow_tokenizer_class=__lowerCamelCase )
       AutoTokenizer.register(__lowerCamelCase			,       fast_tokenizer_class=__lowerCamelCase )
       # If remote code is not set, the default is to use local
       UpperCAmelCase__	:str      =      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer''' )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizerFast''' )
       self.assertFalse(tokenizer.special_attribute_present )
       UpperCAmelCase__	:str      =      AutoTokenizer.from_pretrained('''hf-internal-testing/test_dynamic_tokenizer'''			,       use_fast=__lowerCamelCase )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizer''' )
       self.assertFalse(tokenizer.special_attribute_present )

       # If remote code is disabled, we load the local one.
       UpperCAmelCase__	:Union[str, Any]      =      AutoTokenizer.from_pretrained(
           '''hf-internal-testing/test_dynamic_tokenizer'''			,       trust_remote_code=__lowerCamelCase )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizerFast''' )
       self.assertFalse(tokenizer.special_attribute_present )
       UpperCAmelCase__	:str      =      AutoTokenizer.from_pretrained(
           '''hf-internal-testing/test_dynamic_tokenizer'''			,       trust_remote_code=__lowerCamelCase			,       use_fast=__lowerCamelCase )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizer''' )
       self.assertFalse(tokenizer.special_attribute_present )

       # If remote is enabled, we load from the Hub
       UpperCAmelCase__	:int      =      AutoTokenizer.from_pretrained(
           '''hf-internal-testing/test_dynamic_tokenizer'''			,       trust_remote_code=__lowerCamelCase )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizerFast''' )
       self.assertTrue(tokenizer.special_attribute_present )
       UpperCAmelCase__	:List[str]      =      AutoTokenizer.from_pretrained(
           '''hf-internal-testing/test_dynamic_tokenizer'''			,       trust_remote_code=__lowerCamelCase			,       use_fast=__lowerCamelCase )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizer''' )
       self.assertTrue(tokenizer.special_attribute_present )

    finally:
       if "custom" in CONFIG_MAPPING._extra_content:
          del CONFIG_MAPPING._extra_content["custom"]
       if CustomConfig in TOKENIZER_MAPPING._extra_content:
          del TOKENIZER_MAPPING._extra_content[CustomConfig]


 def        __SCREAMING_SNAKE_CASE					(      self							:	str ):
    UpperCAmelCase__	:str      =      AutoTokenizer.from_pretrained(
        '''hf-internal-testing/test_dynamic_tokenizer_legacy'''			,       trust_remote_code=__lowerCamelCase )
    self.assertTrue(tokenizer.special_attribute_present )
    if is_tokenizers_available():
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizerFast''' )

       # Test we can also load the slow version
       UpperCAmelCase__	:Dict      =      AutoTokenizer.from_pretrained(
           '''hf-internal-testing/test_dynamic_tokenizer_legacy'''			,       trust_remote_code=__lowerCamelCase			,       use_fast=__lowerCamelCase )
       self.assertTrue(tokenizer.special_attribute_present )
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizer''' )
    else:
       self.assertEqual(tokenizer.__class__.__name__			,       '''NewTokenizer''' )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    with self.assertRaisesRegex(
        __lowerCamelCase			,       '''bert-base is not a local folder and is not a valid model identifier''' ):
       UpperCAmelCase__	:Union[str, Any]      =      AutoTokenizer.from_pretrained('''bert-base''' )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    with self.assertRaisesRegex(
        __lowerCamelCase			,       r'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)''' ):
       UpperCAmelCase__	:Any      =      AutoTokenizer.from_pretrained(__lowerCamelCase			,       revision='''aaaaaa''' )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    # Make sure we have cached the tokenizer.
    UpperCAmelCase__	:Union[str, Any]      =      AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert''' )
    with RequestCounter() as counter:
       UpperCAmelCase__	:List[str]      =      AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert''' )
       self.assertEqual(counter.get_request_count			,       0 )
       self.assertEqual(counter.head_request_count			,       1 )
       self.assertEqual(counter.other_request_count			,       0 )







'''simple docstring'''





import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
    from .features import FeatureType


__lowerCamelCase					,		__lowerCamelCase					,		__lowerCamelCase												=						False, False, False




@dataclass
class 				UpperCAmelCase			:
 UpperCAmelCase				=					None
 UpperCAmelCase				=					True
 UpperCAmelCase				=					True
 UpperCAmelCase				=					None
 # Automatically constructed
 UpperCAmelCase				=					"dict"
 UpperCAmelCase				=					pa.struct({"bytes": pa.binary(), "path": pa.string()}						)
 UpperCAmelCase				=					field(default="Audio"						, init=_snake_case						, repr=_snake_case						)


 def __call__(      self							:	Any ):
    return self.pa_type


 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict			,       __lowerCamelCase							:	Union[str, bytes, dict] ):
    try:
       import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
    except ImportError as err:
       raise ImportError('''To support encoding audio data, please install \'soundfile\'.''' ) from err
    if isinstance(__lowerCamelCase			,       __lowerCamelCase ):
       return {"bytes": None, "path": value}
    elif isinstance(__lowerCamelCase			,       __lowerCamelCase ):
       return {"bytes": value, "path": None}
    elif "array" in value:
       # convert the audio array to wav bytes
       UpperCAmelCase__	:int      =      BytesIO()
       sf.write(__lowerCamelCase			,       value['''array''']			,       value['''sampling_rate''']			,       format='''wav''' )
       return {"bytes": buffer.getvalue(), "path": None}
    elif value.get('''path''' ) is not None and os.path.isfile(value['''path'''] ):
       # we set "bytes": None to not duplicate the data if they're already available locally
       if value["path"].endswith('''pcm''' ):
          # "PCM" only has raw audio bytes
          if value.get('''sampling_rate''' ) is None:
             # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
             raise KeyError('''To use PCM files, please specify a \'sampling_rate\' in Audio object''' )
          if value.get('''bytes''' ):
             # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
             UpperCAmelCase__	:List[Any]      =      np.frombuffer(value['''bytes''']			,       dtype=np.intaa ).astype(np.floataa ) / 3_2_7_6_7
          else:
             UpperCAmelCase__	:Optional[Any]      =      np.memmap(value['''path''']			,       dtype='''h'''			,       mode='''r''' ).astype(np.floataa ) / 3_2_7_6_7

          UpperCAmelCase__	:Optional[Any]      =      BytesIO(bytes() )
          sf.write(__lowerCamelCase			,       __lowerCamelCase			,       value['''sampling_rate''']			,       format='''wav''' )
          return {"bytes": buffer.getvalue(), "path": None}
       else:
          return {"bytes": None, "path": value.get('''path''' )}
    elif value.get('''bytes''' ) is not None or value.get('''path''' ) is not None:
       # store the audio bytes, and path is used to infer the audio format using the file extension
       return {"bytes": value.get('''bytes''' ), "path": value.get('''path''' )}
    else:
       raise ValueError(
           f'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.''' )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any]			,       __lowerCamelCase							:	dict			,       __lowerCamelCase							:	Optional[Dict[str, Union[str, bool, None]]] = None ):
    if not self.decode:
       raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.''' )

    UpperCAmelCase__   ,						UpperCAmelCase__	:str      =      (value['''path'''], BytesIO(value['''bytes'''] )) if value['''bytes'''] is not None else (value['''path'''], None)
    if path is None and file is None:
       raise ValueError(f'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.''' )

    try:
       import librosa
       import soundfile as sf
    except ImportError as err:
       raise ImportError('''To support decoding audio files, please install \'librosa\' and \'soundfile\'.''' ) from err

    UpperCAmelCase__	:List[str]      =      xsplitext(__lowerCamelCase )[1][1:].lower() if path is not None else None
    if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
       raise RuntimeError(
           '''Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '''
           '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. ''' )
    elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
       raise RuntimeError(
           '''Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '''
           '''You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. ''' )

    if file is None:
       UpperCAmelCase__	:Optional[Any]      =      token_per_repo_id or {}
       UpperCAmelCase__	:str      =      path.split('''::''' )[-1]
       try:
          UpperCAmelCase__	:Tuple      =      string_to_dict(__lowerCamelCase			,       config.HUB_DATASETS_URL )['''repo_id''']
          UpperCAmelCase__	:str      =      token_per_repo_id[repo_id]
       except (ValueError, KeyError):
          UpperCAmelCase__	:Tuple      =      None

       with xopen(__lowerCamelCase			,       '''rb'''			,       use_auth_token=__lowerCamelCase ) as f:
          UpperCAmelCase__   ,						UpperCAmelCase__	:Union[str, Any]      =      sf.read(__lowerCamelCase )

    else:
       UpperCAmelCase__   ,						UpperCAmelCase__	:List[Any]      =      sf.read(__lowerCamelCase )

    UpperCAmelCase__	:Optional[int]      =      array.T
    if self.mono:
       UpperCAmelCase__	:Any      =      librosa.to_mono(__lowerCamelCase )
    if self.sampling_rate and self.sampling_rate != sampling_rate:
       UpperCAmelCase__	:Union[str, Any]      =      librosa.resample(__lowerCamelCase			,       orig_sr=__lowerCamelCase			,       target_sr=self.sampling_rate )
       UpperCAmelCase__	:List[str]      =      self.sampling_rate

    return {"path": path, "array": array, "sampling_rate": sampling_rate}


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any] ):
    from .features import Value

    if self.decode:
       raise ValueError('''Cannot flatten a decoded Audio feature.''' )
    return {
        "bytes": Value('''binary''' ),
        "path": Value('''string''' ),
    }


 def        __SCREAMING_SNAKE_CASE					(      self							:	Any			,       __lowerCamelCase							:	Union[pa.StringArray, pa.StructArray] ):
    if pa.types.is_string(storage.type ):
       UpperCAmelCase__	:List[str]      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.binary() )
       UpperCAmelCase__	:Tuple      =      pa.StructArray.from_arrays([bytes_array, storage]			,       ['''bytes''', '''path''']			,       mask=storage.is_null() )
    elif pa.types.is_binary(storage.type ):
       UpperCAmelCase__	:str      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.string() )
       UpperCAmelCase__	:int      =      pa.StructArray.from_arrays([storage, path_array]			,       ['''bytes''', '''path''']			,       mask=storage.is_null() )
    elif pa.types.is_struct(storage.type ) and storage.type.get_all_field_indices('''array''' ):
       UpperCAmelCase__	:Any      =      pa.array([Audio().encode_example(__lowerCamelCase ) if x is not None else None for x in storage.to_pylist()] )
    elif pa.types.is_struct(storage.type ):
       if storage.type.get_field_index('''bytes''' ) >= 0:
          UpperCAmelCase__	:str      =      storage.field('''bytes''' )
       else:
          UpperCAmelCase__	:List[str]      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.binary() )
       if storage.type.get_field_index('''path''' ) >= 0:
          UpperCAmelCase__	:Optional[int]      =      storage.field('''path''' )
       else:
          UpperCAmelCase__	:Optional[int]      =      pa.array([None] * len(__lowerCamelCase )			,       type=pa.string() )
       UpperCAmelCase__	:List[str]      =      pa.StructArray.from_arrays([bytes_array, path_array]			,       ['''bytes''', '''path''']			,       mask=storage.is_null() )
    return array_cast(__lowerCamelCase			,       self.pa_type )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any]			,       __lowerCamelCase							:	pa.StructArray ):

    @no_op_if_value_is_null
    def path_to_bytes(__lowerCamelCase							:	Dict ):
       with xopen(__lowerCamelCase			,       '''rb''' ) as f:
          UpperCAmelCase__	:Any      =      f.read()
       return bytes_

    UpperCAmelCase__	:Union[str, Any]      =      pa.array(
        [
            (path_to_bytes(x['''path'''] ) if x['''bytes'''] is None else x['''bytes''']) if x is not None else None
            for x in storage.to_pylist()
        ]			,       type=pa.binary()			,       )
    UpperCAmelCase__	:Optional[int]      =      pa.array(
        [os.path.basename(__lowerCamelCase ) if path is not None else None for path in storage.field('''path''' ).to_pylist()]			,       type=pa.string()			,       )
    UpperCAmelCase__	:Optional[int]      =      pa.StructArray.from_arrays([bytes_array, path_array]			,       ['''bytes''', '''path''']			,       mask=bytes_array.is_null() )
    return array_cast(__lowerCamelCase			,       self.pa_type )



