




from copy import deepcopy
class 						_A :






  def __init__(  self		:     Optional[int]				,						_A		:     list[int] | None = None				,						_A		:     int | None = None				)		->   None:






       """simple docstring"""







       if arr is None and size is not None:
            lowercase :       Any       =			size
            lowercase :       List[Any]       =			[0] * size
       elif arr is not None:
            self.init(_A				)
       else:
            raise ValueError('''Either arr or size must be specified'''				)






  def  __a      (  self		:     List[Any]				,						_A		:     list[int]				)		->   None:






       """simple docstring"""







       lowercase :       int       =			len(_A				)
       lowercase :       Union[str, Any]       =			deepcopy(_A				)
       for i in range(1				,						self.size				):
            lowercase :       List[str]       =			self.next_(_A				)
            if j < self.size:
                 self.tree[j] += self.tree[i]






  def  __a      (  self		:     str				)		->   list[int]:






       """simple docstring"""







       lowercase :       int       =			self.tree[:]
       for i in range(self.size - 1				,						0				,						-1				):
            lowercase :       int       =			self.next_(_A				)
            if j < self.size:
                 arr[j] -= arr[i]
       return arr






  @staticmethod
  def  __a      (  _A		:     int				)		->   int:






       """simple docstring"""







       return index + (index & (-index))






  @staticmethod
  def  __a      (  _A		:     int				)		->   int:






       """simple docstring"""







       return index - (index & (-index))






  def  __a      (  self		:     Union[str, Any]				,						_A		:     int				,						_A		:     int				)		->   None:






       """simple docstring"""







       if index == 0:
            self.tree[0] += value
            return
       while index < self.size:
            self.tree[index] += value
            lowercase :       List[Any]       =			self.next_(_A				)






  def  __a      (  self		:     Any				,						_A		:     int				,						_A		:     int				)		->   None:






       """simple docstring"""







       self.add(_A				,						value - self.get(_A				)				)






  def  __a      (  self		:     Any				,						_A		:     int				)		->   int:






       """simple docstring"""







       if right == 0:
            return 0
       lowercase :       Union[str, Any]       =			self.tree[0]
       right -= 1  # make right inclusive
       while right > 0:
            result += self.tree[right]
            lowercase :       Union[str, Any]       =			self.prev(_A				)
       return result






  def  __a      (  self		:     List[Any]				,						_A		:     int				,						_A		:     int				)		->   int:






       """simple docstring"""







       return self.prefix(_A				) - self.prefix(_A				)






  def  __a      (  self		:     Dict				,						_A		:     int				)		->   int:






       """simple docstring"""







       return self.query(_A				,						index + 1				)






  def  __a      (  self		:     List[Any]				,						_A		:     int				)		->   int:






       """simple docstring"""







       value -= self.tree[0]
       if value < 0:
            return -1

       lowercase :       Optional[int]       =			1  # Largest power of 2 <= size
       while j * 2 < self.size:
            j *= 2

       lowercase :       Optional[Any]       =			0

       while j > 0:
            if i + j < self.size and self.tree[i + j] <= value:
                 value -= self.tree[i + j]
                 i += j
            j //= 2
       return i


if __name__ == "__main__":
 import doctest

 doctest.testmod()





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

from huggingface_hub import HfFolder, Repository, create_repo, delete_repo
from requests.exceptions import HTTPError

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


lowerCAmelCase_						=			get_tests_dir('fixtures/dummy_feature_extractor_config.json')
lowerCAmelCase_						=			get_tests_dir('fixtures/vocab.json')
lowerCAmelCase_						=			get_tests_dir('fixtures')
class 						_A (  unittest.TestCase				):
  _UpperCamelCase       :      int			=      ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






  def  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       lowercase :       Union[str, Any]       =			0






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






       """simple docstring"""







       lowercase :       Union[str, Any]       =			AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''				)
       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Any				)		->   int:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       str       =			WavaVecaConfig()
            lowercase :       Tuple       =			AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''				)

            # save in new folder
            model_config.save_pretrained(_A				)
            processor.save_pretrained(_A				)

            lowercase :       Optional[Any]       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Tuple				)		->   Union[str, Any]:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            # copy relevant files
            copyfile(_A				,						os.path.join(_A				,						_A				)				)
            copyfile(_A				,						os.path.join(_A				,						'''vocab.json'''				)				)

            lowercase :       Any       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Tuple				)		->   Any:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       List[Any]       =			WavaVecaFeatureExtractor()
            lowercase :       Optional[int]       =			AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''				)

            lowercase :       Union[str, Any]       =			WavaVecaProcessor(_A				,						_A				)

            # save in new folder
            processor.save_pretrained(_A				)

            # drop `processor_class` in tokenizer
            with open(os.path.join(_A				,						_A				)				,						'''r'''				) as f:
                 lowercase :       Union[str, Any]       =			json.load(_A				)
                 config_dict.pop('''processor_class'''				)

            with open(os.path.join(_A				,						_A				)				,						'''w'''				) as f:
                 f.write(json.dumps(_A				)				)

            lowercase :       int       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     str				)		->   List[str]:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       int       =			WavaVecaFeatureExtractor()
            lowercase :       Dict       =			AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''				)

            lowercase :       Optional[int]       =			WavaVecaProcessor(_A				,						_A				)

            # save in new folder
            processor.save_pretrained(_A				)

            # drop `processor_class` in feature extractor
            with open(os.path.join(_A				,						_A				)				,						'''r'''				) as f:
                 lowercase :       int       =			json.load(_A				)
                 config_dict.pop('''processor_class'''				)

            with open(os.path.join(_A				,						_A				)				,						'''w'''				) as f:
                 f.write(json.dumps(_A				)				)

            lowercase :       Any       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     List[Any]				)		->   Tuple:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       Dict       =			WavaVecaConfig(processor_class='''Wav2Vec2Processor'''				)
            model_config.save_pretrained(_A				)
            # copy relevant files
            copyfile(_A				,						os.path.join(_A				,						'''vocab.json'''				)				)
            # create emtpy sample processor
            with open(os.path.join(_A				,						_A				)				,						'''w'''				) as f:
                 f.write('''{}'''				)

            lowercase :       Tuple       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       with self.assertRaises(_A				):
            lowercase :       Optional[Any]       =			AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''				)
       # If remote code is disabled, we can't load this config.
       with self.assertRaises(_A				):
            lowercase :       List[str]       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)

       lowercase :       str       =			AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)
       self.assertTrue(processor.special_attribute_present				)
       self.assertEqual(processor.__class__.__name__				,						'''NewProcessor'''				)

       lowercase :       Union[str, Any]       =			processor.feature_extractor
       self.assertTrue(feature_extractor.special_attribute_present				)
       self.assertEqual(feature_extractor.__class__.__name__				,						'''NewFeatureExtractor'''				)

       lowercase :       Optional[int]       =			processor.tokenizer
       self.assertTrue(tokenizer.special_attribute_present				)
       if is_tokenizers_available():
            self.assertEqual(tokenizer.__class__.__name__				,						'''NewTokenizerFast'''				)

            # Test we can also load the slow version
            lowercase :       Any       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				,						use_fast=_A				)
            lowercase :       Any       =			new_processor.tokenizer
            self.assertTrue(new_tokenizer.special_attribute_present				)
            self.assertEqual(new_tokenizer.__class__.__name__				,						'''NewTokenizer'''				)
       else:
            self.assertEqual(tokenizer.__class__.__name__				,						'''NewTokenizer'''				)






  def  __a      (  self		:     Dict				)		->   Optional[Any]:






       """simple docstring"""







       try:
            AutoConfig.register('''custom'''				,						_A				)
            AutoFeatureExtractor.register(_A				,						_A				)
            AutoTokenizer.register(_A				,						slow_tokenizer_class=_A				)
            AutoProcessor.register(_A				,						_A				)
            # Trying to register something existing in the Transformers library will raise an error
            with self.assertRaises(_A				):
                 AutoProcessor.register(_A				,						_A				)

            # Now that the config is registered, it can be used as any other config with the auto-API
            lowercase :       Optional[Any]       =			CustomFeatureExtractor.from_pretrained(_A				)

            with tempfile.TemporaryDirectory() as tmp_dir:
                 lowercase :       List[Any]       =			os.path.join(_A				,						'''vocab.txt'''				)
                 with open(_A				,						'''w'''				,						encoding='''utf-8'''				) as vocab_writer:
                      vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]				)				)
                 lowercase :       Optional[int]       =			CustomTokenizer(_A				)

            lowercase :       Union[str, Any]       =			CustomProcessor(_A				,						_A				)

            with tempfile.TemporaryDirectory() as tmp_dir:
                 processor.save_pretrained(_A				)
                 lowercase :       Dict       =			AutoProcessor.from_pretrained(_A				)
                 self.assertIsInstance(_A				,						_A				)

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






  def  __a      (  self		:     str				)		->   Any:






       """simple docstring"""







       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      int			=      False


       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      Union[str, Any]			=      False


       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      Tuple			=      '''AutoFeatureExtractor'''
         _UpperCamelCase       :      Any			=      '''AutoTokenizer'''
         _UpperCamelCase       :      Any			=      False

       try:
            AutoConfig.register('''custom'''				,						_A				)
            AutoFeatureExtractor.register(_A				,						_A				)
            AutoTokenizer.register(_A				,						slow_tokenizer_class=_A				)
            AutoProcessor.register(_A				,						_A				)
            # If remote code is not set, the default is to use local classes.
            lowercase :       List[Any]       =			AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''				)
            self.assertEqual(processor.__class__.__name__				,						'''NewProcessor'''				)
            self.assertFalse(processor.special_attribute_present				)
            self.assertFalse(processor.feature_extractor.special_attribute_present				)
            self.assertFalse(processor.tokenizer.special_attribute_present				)

            # If remote code is disabled, we load the local ones.
            lowercase :       Any       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)
            self.assertEqual(processor.__class__.__name__				,						'''NewProcessor'''				)
            self.assertFalse(processor.special_attribute_present				)
            self.assertFalse(processor.feature_extractor.special_attribute_present				)
            self.assertFalse(processor.tokenizer.special_attribute_present				)

            # If remote is enabled, we load from the Hub.
            lowercase :       Union[str, Any]       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)
            self.assertEqual(processor.__class__.__name__				,						'''NewProcessor'''				)
            self.assertTrue(processor.special_attribute_present				)
            self.assertTrue(processor.feature_extractor.special_attribute_present				)
            self.assertTrue(processor.tokenizer.special_attribute_present				)

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






  def  __a      (  self		:     Tuple				)		->   Dict:






       """simple docstring"""







       lowercase :       Optional[int]       =			AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''				)
       self.assertEqual(processor.__class__.__name__				,						'''BertTokenizerFast'''				)






  def  __a      (  self		:     Any				)		->   int:






       """simple docstring"""







       lowercase :       Any       =			AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext'''				)
       self.assertEqual(processor.__class__.__name__				,						'''ConvNextImageProcessor'''				)


@is_staging_test
class 						_A (  unittest.TestCase				):
  _UpperCamelCase       :      int			=      ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






  @classmethod
  def  __a      (  cls		:     Tuple				)		->   List[Any]:






       """simple docstring"""







       lowercase :       str       =			TOKEN
       HfFolder.save_token(_A				)






  @classmethod
  def  __a      (  cls		:     List[Any]				)		->   Dict:






       """simple docstring"""







       try:
            delete_repo(token=cls._token				,						repo_id='''test-processor'''				)
       except HTTPError:
            pass

       try:
            delete_repo(token=cls._token				,						repo_id='''valid_org/test-processor-org'''				)
       except HTTPError:
            pass

       try:
            delete_repo(token=cls._token				,						repo_id='''test-dynamic-processor'''				)
       except HTTPError:
            pass






  def  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       lowercase :       Any       =			WavaVecaProcessor.from_pretrained(_A				)
       with tempfile.TemporaryDirectory() as tmp_dir:
            processor.save_pretrained(
                os.path.join(_A				,						'''test-processor'''				)				,						push_to_hub=_A				,						use_auth_token=self._token				)

            lowercase :       Any       =			WavaVecaProcessor.from_pretrained(f"""{USER}/test-processor"""				)
            for k, v in processor.feature_extractor.__dict__.items():
                 self.assertEqual(_A				,						getattr(new_processor.feature_extractor				,						_A				)				)
            self.assertDictEqual(new_processor.tokenizer.get_vocab()				,						processor.tokenizer.get_vocab()				)






  def  __a      (  self		:     Optional[Any]				)		->   str:






       """simple docstring"""







       lowercase :       int       =			WavaVecaProcessor.from_pretrained(_A				)

       with tempfile.TemporaryDirectory() as tmp_dir:
            processor.save_pretrained(
                os.path.join(_A				,						'''test-processor-org'''				)				,						push_to_hub=_A				,						use_auth_token=self._token				,						organization='''valid_org'''				,						)

            lowercase :       Dict       =			WavaVecaProcessor.from_pretrained('''valid_org/test-processor-org'''				)
            for k, v in processor.feature_extractor.__dict__.items():
                 self.assertEqual(_A				,						getattr(new_processor.feature_extractor				,						_A				)				)
            self.assertDictEqual(new_processor.tokenizer.get_vocab()				,						processor.tokenizer.get_vocab()				)






  def  __a      (  self		:     str				)		->   Dict:






       """simple docstring"""







       CustomFeatureExtractor.register_for_auto_class()
       CustomTokenizer.register_for_auto_class()
       CustomProcessor.register_for_auto_class()

       lowercase :       List[str]       =			CustomFeatureExtractor.from_pretrained(_A				)

       with tempfile.TemporaryDirectory() as tmp_dir:
            lowercase :       Optional[int]       =			os.path.join(_A				,						'''vocab.txt'''				)
            with open(_A				,						'''w'''				,						encoding='''utf-8'''				) as vocab_writer:
                 vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]				)				)
            lowercase :       Optional[Any]       =			CustomTokenizer(_A				)

       lowercase :       Optional[Any]       =			CustomProcessor(_A				,						_A				)

       with tempfile.TemporaryDirectory() as tmp_dir:
            create_repo(f"""{USER}/test-dynamic-processor"""				,						token=self._token				)
            lowercase :       List[str]       =			Repository(_A				,						clone_from=f"""{USER}/test-dynamic-processor"""				,						token=self._token				)
            processor.save_pretrained(_A				)

            # This has added the proper auto_map field to the feature extractor config
            self.assertDictEqual(
                processor.feature_extractor.auto_map				,						{
                    '''AutoFeatureExtractor''': '''custom_feature_extraction.CustomFeatureExtractor''',
                    '''AutoProcessor''': '''custom_processing.CustomProcessor''',
                }				,						)

            # This has added the proper auto_map field to the tokenizer config
            with open(os.path.join(_A				,						'''tokenizer_config.json'''				)				) as f:
                 lowercase :       Optional[Any]       =			json.load(_A				)
            self.assertDictEqual(
                tokenizer_config['''auto_map''']				,						{
                    '''AutoTokenizer''': ['''custom_tokenization.CustomTokenizer''', None],
                    '''AutoProcessor''': '''custom_processing.CustomProcessor''',
                }				,						)

            # The code has been copied from fixtures
            self.assertTrue(os.path.isfile(os.path.join(_A				,						'''custom_feature_extraction.py'''				)				)				)
            self.assertTrue(os.path.isfile(os.path.join(_A				,						'''custom_tokenization.py'''				)				)				)
            self.assertTrue(os.path.isfile(os.path.join(_A				,						'''custom_processing.py'''				)				)				)

            repo.push_to_hub()

       lowercase :       Tuple       =			AutoProcessor.from_pretrained(f"""{USER}/test-dynamic-processor"""				,						trust_remote_code=_A				)
       # Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
       self.assertEqual(new_processor.__class__.__name__				,						'''CustomProcessor'''				)