"""simple docstring"""
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


SCREAMING_SNAKE_CASE__   =					get_tests_dir("fixtures/dummy_feature_extractor_config.json")
SCREAMING_SNAKE_CASE__   =					get_tests_dir("fixtures/vocab.json")
SCREAMING_SNAKE_CASE__   =					get_tests_dir("fixtures")



class 					lowerCAmelCase_      (     unittest.TestCase   ):



				"""simple docstring"""
				_lowerCAmelCase       :      Optional[int]     					=  ["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]

				def 						snake_case	(			self      ):






										"""simple docstring"""

										snake_case        =     0

				def 						snake_case	(			self      ):






										"""simple docstring"""

										snake_case        =     AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'      )
										self.assertIsInstance(lowerCAmelCase						,      lowerCAmelCase      )

				def 						snake_case	(			self      ):






										"""simple docstring"""

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

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

																snake_case        =     AutoProcessor.from_pretrained(lowerCAmelCase      )

										self.assertIsInstance(lowerCAmelCase						,      lowerCAmelCase      )

				def 						snake_case	(			self      ):






										"""simple docstring"""

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

																snake_case        =     AutoProcessor.from_pretrained(lowerCAmelCase      )

										self.assertIsInstance(lowerCAmelCase						,      lowerCAmelCase      )

				def 						snake_case	(			self      ):






										"""simple docstring"""

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

																snake_case        =     WavaVecaProcessor(lowerCAmelCase						,      lowerCAmelCase      )

																# save in new folder
																processor.save_pretrained(lowerCAmelCase      )

																# drop `processor_class` in tokenizer
																with open(os.path.join(lowerCAmelCase						,      lowerCAmelCase      )						,      'r'      ) as f:
																						snake_case        =     json.load(lowerCAmelCase      )
																						config_dict.pop('processor_class'      )

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

																snake_case        =     AutoProcessor.from_pretrained(lowerCAmelCase      )

										self.assertIsInstance(lowerCAmelCase						,      lowerCAmelCase      )

				def 						snake_case	(			self      ):






										"""simple docstring"""

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

																snake_case        =     WavaVecaProcessor(lowerCAmelCase						,      lowerCAmelCase      )

																# save in new folder
																processor.save_pretrained(lowerCAmelCase      )

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

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

																snake_case        =     AutoProcessor.from_pretrained(lowerCAmelCase      )

										self.assertIsInstance(lowerCAmelCase						,      lowerCAmelCase      )

				def 						snake_case	(			self      ):






										"""simple docstring"""

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

																snake_case        =     AutoProcessor.from_pretrained(lowerCAmelCase      )

										self.assertIsInstance(lowerCAmelCase						,      lowerCAmelCase      )

				def 						snake_case	(			self      ):






										"""simple docstring"""

										with self.assertRaises(lowerCAmelCase      ):
																snake_case        =     AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'      )
										# If remote code is disabled, we can't load this config.
										with self.assertRaises(lowerCAmelCase      ):
																snake_case        =     AutoProcessor.from_pretrained(
																    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowerCAmelCase      )

										snake_case        =     AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowerCAmelCase      )
										self.assertTrue(processor.special_attribute_present      )
										self.assertEqual(processor.__class__.__name__						,      'NewProcessor'      )

										snake_case        =     processor.feature_extractor
										self.assertTrue(feature_extractor.special_attribute_present      )
										self.assertEqual(feature_extractor.__class__.__name__						,      'NewFeatureExtractor'      )

										snake_case        =     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
																snake_case        =     AutoProcessor.from_pretrained(
																    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowerCAmelCase						,      use_fast=lowerCAmelCase      )
																snake_case        =     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 						snake_case	(			self      ):






										"""simple docstring"""

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

																# Now that the config is registered, it can be used as any other config with the auto-API
																snake_case        =     CustomFeatureExtractor.from_pretrained(lowerCAmelCase      )

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

																snake_case        =     CustomProcessor(lowerCAmelCase						,      lowerCAmelCase      )

																with tempfile.TemporaryDirectory() as tmp_dir:
																						processor.save_pretrained(lowerCAmelCase      )
																						snake_case        =     AutoProcessor.from_pretrained(lowerCAmelCase      )
																						self.assertIsInstance(lowerCAmelCase						,      lowerCAmelCase      )

										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 						snake_case	(			self      ):






										"""simple docstring"""

										class 					lowerCAmelCase_      (     lowerCAmelCase   ):



														"""simple docstring"""
														_lowerCAmelCase       :      List[Any]     					=  False





										class 					lowerCAmelCase_      (     lowerCAmelCase   ):



														"""simple docstring"""
														_lowerCAmelCase       :      str     					=  False





										class 					lowerCAmelCase_      (     lowerCAmelCase   ):



														"""simple docstring"""
														_lowerCAmelCase       :      Optional[Any]     					=  """AutoFeatureExtractor"""
														_lowerCAmelCase       :      Dict     					=  """AutoTokenizer"""
														_lowerCAmelCase       :      Optional[Any]     					=  False

										try:
																AutoConfig.register('custom'						,      lowerCAmelCase      )
																AutoFeatureExtractor.register(lowerCAmelCase						,      lowerCAmelCase      )
																AutoTokenizer.register(lowerCAmelCase						,      slow_tokenizer_class=lowerCAmelCase      )
																AutoProcessor.register(lowerCAmelCase						,      lowerCAmelCase      )
																# If remote code is not set, the default is to use local classes.
																snake_case        =     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.
																snake_case        =     AutoProcessor.from_pretrained(
																    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowerCAmelCase      )
																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.
																snake_case        =     AutoProcessor.from_pretrained(
																    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowerCAmelCase      )
																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 						snake_case	(			self      ):






										"""simple docstring"""

										snake_case        =     AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-bert'      )
										self.assertEqual(processor.__class__.__name__						,      'BertTokenizerFast'      )

				def 						snake_case	(			self      ):






										"""simple docstring"""

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





@is_staging_test
class 					lowerCAmelCase_      (     unittest.TestCase   ):



				"""simple docstring"""
				_lowerCAmelCase       :      Optional[Any]     					=  ["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]

				@classmethod
				def 						snake_case	(			cls      ):






										"""simple docstring"""

										snake_case        =     TOKEN
										HfFolder.save_token(lowerCAmelCase      )

				@classmethod
				def 						snake_case	(			cls      ):






										"""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 						snake_case	(			self      ):






										"""simple docstring"""

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

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

				def 						snake_case	(			self      ):






										"""simple docstring"""

										snake_case        =     WavaVecaProcessor.from_pretrained(lowerCAmelCase      )

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

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

				def 						snake_case	(			self      ):






										"""simple docstring"""

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

										snake_case        =     CustomFeatureExtractor.from_pretrained(lowerCAmelCase      )

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

										snake_case        =     CustomProcessor(lowerCAmelCase						,      lowerCAmelCase      )

										with tempfile.TemporaryDirectory() as tmp_dir:
																create_repo(F"""{USER}/test-dynamic-processor"""						,      token=self._token      )
																snake_case        =     Repository(lowerCAmelCase						,      clone_from=F"""{USER}/test-dynamic-processor"""						,      token=self._token      )
																processor.save_pretrained(lowerCAmelCase      )

																# 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(lowerCAmelCase						,      'tokenizer_config.json'      )      ) as f:
																						snake_case        =     json.load(lowerCAmelCase      )
																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(lowerCAmelCase						,      'custom_feature_extraction.py'      )      )      )
																self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase						,      'custom_tokenization.py'      )      )      )
																self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase						,      'custom_processing.py'      )      )      )

																repo.push_to_hub()

										snake_case        =     AutoProcessor.from_pretrained(F"""{USER}/test-dynamic-processor"""						,      trust_remote_code=lowerCAmelCase      )
										# 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'      )



"""simple docstring"""
from __future__ import annotations

from typing import Any



class 					lowerCAmelCase_      (     lowerCAmelCase   ):



				"""simple docstring"""
				pass





class 					lowerCAmelCase_      :



				"""simple docstring"""

				def __init__(			self						,      lowerCAmelCase      ):






										"""simple docstring"""

										snake_case        =     data
										snake_case        =     None

				def __iter__(			self      ):






										"""simple docstring"""

										snake_case        =     self
										snake_case        =     []
										while node:
																if node in visited:
																						raise ContainsLoopError
																visited.append(lowerCAmelCase      )
																yield node.data
																snake_case        =     node.next_node

				@property
				def 						snake_case	(			self      ):






										"""simple docstring"""

										try:
																list(self      )
																return False
										except ContainsLoopError:
																return True


if __name__ == "__main__":
					SCREAMING_SNAKE_CASE__   =					Node(1)
					SCREAMING_SNAKE_CASE__   =					Node(2)
					SCREAMING_SNAKE_CASE__   =					Node(3)
					SCREAMING_SNAKE_CASE__   =					Node(4)
					print(root_node.has_loop)  # False
					SCREAMING_SNAKE_CASE__   =					root_node.next_node
					print(root_node.has_loop)  # True

					SCREAMING_SNAKE_CASE__   =					Node(5)
					SCREAMING_SNAKE_CASE__   =					Node(6)
					SCREAMING_SNAKE_CASE__   =					Node(5)
					SCREAMING_SNAKE_CASE__   =					Node(6)
					print(root_node.has_loop)  # False

					SCREAMING_SNAKE_CASE__   =					Node(1)
					print(root_node.has_loop)  # False


