
'''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


__lowerCAmelCase												=	get_tests_dir('fixtures/dummy_feature_extractor_config.json')
__lowerCAmelCase												=	get_tests_dir('fixtures/vocab.json')
__lowerCAmelCase												=	get_tests_dir('fixtures')


class       _lowerCAmelCase   (   unittest.TestCase				):







							'''simple docstring'''

							lowerCAmelCase_ 						=						["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]

							def 	lowercase							(self  )					->       Union[str, Any]:
														_snake_case       =  0

							def 	lowercase							(self  )					->       Optional[int]:
														_snake_case       =  AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h"""  )
														self.assertIsInstance(UpperCAmelCase  ,  UpperCAmelCase  )

							def 	lowercase							(self  )					->       List[Any]:
														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(UpperCAmelCase  )
																					processor.save_pretrained(UpperCAmelCase  )

																					_snake_case       =  AutoProcessor.from_pretrained(UpperCAmelCase  )

														self.assertIsInstance(UpperCAmelCase  ,  UpperCAmelCase  )

							def 	lowercase							(self  )					->       Union[str, Any]:
														with tempfile.TemporaryDirectory() as tmpdirname:
																					# copy relevant files
																					copyfile(UpperCAmelCase  ,  os.path.join(UpperCAmelCase  ,  UpperCAmelCase  )  )
																					copyfile(UpperCAmelCase  ,  os.path.join(UpperCAmelCase  ,  """vocab.json"""  )  )

																					_snake_case       =  AutoProcessor.from_pretrained(UpperCAmelCase  )

														self.assertIsInstance(UpperCAmelCase  ,  UpperCAmelCase  )

							def 	lowercase							(self  )					->       List[Any]:
														with tempfile.TemporaryDirectory() as tmpdirname:
																					_snake_case       =  WavaVecaFeatureExtractor()
																					_snake_case       =  AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""  )

																					_snake_case       =  WavaVecaProcessor(UpperCAmelCase  ,  UpperCAmelCase  )

																					# save in new folder
																					processor.save_pretrained(UpperCAmelCase  )

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

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

																					_snake_case       =  AutoProcessor.from_pretrained(UpperCAmelCase  )

														self.assertIsInstance(UpperCAmelCase  ,  UpperCAmelCase  )

							def 	lowercase							(self  )					->       Optional[Any]:
														with tempfile.TemporaryDirectory() as tmpdirname:
																					_snake_case       =  WavaVecaFeatureExtractor()
																					_snake_case       =  AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""  )

																					_snake_case       =  WavaVecaProcessor(UpperCAmelCase  ,  UpperCAmelCase  )

																					# save in new folder
																					processor.save_pretrained(UpperCAmelCase  )

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

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

																					_snake_case       =  AutoProcessor.from_pretrained(UpperCAmelCase  )

														self.assertIsInstance(UpperCAmelCase  ,  UpperCAmelCase  )

							def 	lowercase							(self  )					->       str:
														with tempfile.TemporaryDirectory() as tmpdirname:
																					_snake_case       =  WavaVecaConfig(processor_class="""Wav2Vec2Processor"""  )
																					model_config.save_pretrained(UpperCAmelCase  )
																					# copy relevant files
																					copyfile(UpperCAmelCase  ,  os.path.join(UpperCAmelCase  ,  """vocab.json"""  )  )
																					# create emtpy sample processor
																					with open(os.path.join(UpperCAmelCase  ,  UpperCAmelCase  )  ,  """w"""  ) as f:
																												f.write("""{}"""  )

																					_snake_case       =  AutoProcessor.from_pretrained(UpperCAmelCase  )

														self.assertIsInstance(UpperCAmelCase  ,  UpperCAmelCase  )

							def 	lowercase							(self  )					->       Any:
														# If remote code is not set, we will time out when asking whether to load the model.
														with self.assertRaises(UpperCAmelCase  ):
																					_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(UpperCAmelCase  ):
																					_snake_case       =  AutoProcessor.from_pretrained(
																					    """hf-internal-testing/test_dynamic_processor"""  ,  trust_remote_code=UpperCAmelCase  )

														_snake_case       =  AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor"""  ,  trust_remote_code=UpperCAmelCase  )
														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=UpperCAmelCase  ,  use_fast=UpperCAmelCase  )
																					_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 	lowercase							(self  )					->       Tuple:
														try:
																					AutoConfig.register("""custom"""  ,  UpperCAmelCase  )
																					AutoFeatureExtractor.register(UpperCAmelCase  ,  UpperCAmelCase  )
																					AutoTokenizer.register(UpperCAmelCase  ,  slow_tokenizer_class=UpperCAmelCase  )
																					AutoProcessor.register(UpperCAmelCase  ,  UpperCAmelCase  )
																					# Trying to register something existing in the Transformers library will raise an error
																					with self.assertRaises(UpperCAmelCase  ):
																												AutoProcessor.register(UpperCAmelCase  ,  UpperCAmelCase  )

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

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

																					_snake_case       =  CustomProcessor(UpperCAmelCase  ,  UpperCAmelCase  )

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

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


														class       _lowerCAmelCase   (   __snake_case				):







																					'''simple docstring'''

																					lowerCAmelCase_ 						=						False


														class       _lowerCAmelCase   (   __snake_case				):







																					'''simple docstring'''

																					lowerCAmelCase_ 						=						False


														class       _lowerCAmelCase   (   __snake_case				):







																					'''simple docstring'''

																					lowerCAmelCase_ 						=						"AutoFeatureExtractor"
																					lowerCAmelCase_ 						=						"AutoTokenizer"
																					lowerCAmelCase_ 						=						False

														try:
																					AutoConfig.register("""custom"""  ,  UpperCAmelCase  )
																					AutoFeatureExtractor.register(UpperCAmelCase  ,  UpperCAmelCase  )
																					AutoTokenizer.register(UpperCAmelCase  ,  slow_tokenizer_class=UpperCAmelCase  )
																					AutoProcessor.register(UpperCAmelCase  ,  UpperCAmelCase  )
																					# 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=UpperCAmelCase  )
																					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=UpperCAmelCase  )
																					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 	lowercase							(self  )					->       List[Any]:
														_snake_case       =  AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-bert"""  )
														self.assertEqual(processor.__class__.__name__  ,  """BertTokenizerFast"""  )





							def 	lowercase							(self  )					->       str:
														_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_ 						=						["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]

							@classmethod
							def 	lowercase							(cls  )					->       Tuple:
														_snake_case       =  TOKEN
														HfFolder.save_token(UpperCAmelCase  )

							@classmethod
							def 	lowercase							(cls  )					->       Tuple:
														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 	lowercase							(self  )					->       List[Any]:
														_snake_case       =  WavaVecaProcessor.from_pretrained(UpperCAmelCase  )
														with tempfile.TemporaryDirectory() as tmp_dir:
																					processor.save_pretrained(
																					    os.path.join(UpperCAmelCase  ,  """test-processor"""  )  ,  push_to_hub=UpperCAmelCase  ,  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(UpperCAmelCase  ,  getattr(new_processor.feature_extractor  ,  UpperCAmelCase  )  )
																					self.assertDictEqual(new_processor.tokenizer.get_vocab()  ,  processor.tokenizer.get_vocab()  )

							def 	lowercase							(self  )					->       Any:
														_snake_case       =  WavaVecaProcessor.from_pretrained(UpperCAmelCase  )

														with tempfile.TemporaryDirectory() as tmp_dir:
																					processor.save_pretrained(
																					    os.path.join(UpperCAmelCase  ,  """test-processor-org"""  )  ,  push_to_hub=UpperCAmelCase  ,  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(UpperCAmelCase  ,  getattr(new_processor.feature_extractor  ,  UpperCAmelCase  )  )
																					self.assertDictEqual(new_processor.tokenizer.get_vocab()  ,  processor.tokenizer.get_vocab()  )





							def 	lowercase							(self  )					->       List[str]:
														CustomFeatureExtractor.register_for_auto_class()
														CustomTokenizer.register_for_auto_class()
														CustomProcessor.register_for_auto_class()

														_snake_case       =  CustomFeatureExtractor.from_pretrained(UpperCAmelCase  )

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

														_snake_case       =  CustomProcessor(UpperCAmelCase  ,  UpperCAmelCase  )

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

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

																					repo.push_to_hub()

														_snake_case       =  AutoProcessor.from_pretrained(f"""{USER}/test-dynamic-processor"""  ,  trust_remote_code=UpperCAmelCase  )
														# 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 random import randint, random






def  __SCREAMING_SNAKE_CASE				(  _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE = False							, _SCREAMING_SNAKE_CASE = False							, _SCREAMING_SNAKE_CASE = 5							, ):
							_snake_case       =  [[-1] * number_of_cells]  # Create a highway without any car
							_snake_case       =  0
							_snake_case       =  max(_SCREAMING_SNAKE_CASE							, 0				)
							while i < number_of_cells:
														_snake_case       =  (
														    randint(0							, _SCREAMING_SNAKE_CASE				) if random_speed else initial_speed
														)  # Place the cars
														i += (
														    randint(1							, max_speed * 2				) if random_frequency else frequency
														)  # Arbitrary number, may need tuning
							return highway






def  __SCREAMING_SNAKE_CASE				(  _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE				):
							_snake_case       =  0
							_snake_case       =  highway_now[car_index + 1 :]
							for cell in range(len(_SCREAMING_SNAKE_CASE				)				):  # May need a better name for this
														if cells[cell] != -1:  # If the cell is not empty then
																					return distance  # we have the distance we wanted
														distance += 1
							# Here if the car is near the end of the highway
							return distance + get_distance(_SCREAMING_SNAKE_CASE							, -1				)






def  __SCREAMING_SNAKE_CASE				(  _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE				):
							_snake_case       =  len(_SCREAMING_SNAKE_CASE				)
							# Beforce calculations, the highway is empty
							_snake_case       =  [-1] * number_of_cells

							for car_index in range(_SCREAMING_SNAKE_CASE				):
														if highway_now[car_index] != -1:
																					# Add 1 to the current speed of the car and cap the speed
																					_snake_case       =  min(highway_now[car_index] + 1							, _SCREAMING_SNAKE_CASE				)
																					# Number of empty cell before the next car
																					_snake_case       =  get_distance(_SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE				) - 1
																					# We can't have the car causing an accident
																					_snake_case       =  min(next_highway[car_index]							, _SCREAMING_SNAKE_CASE				)
																					if random() < probability:
																												# Randomly, a driver will slow down
																												_snake_case       =  max(next_highway[car_index] - 1							, 0				)
							return next_highway






def  __SCREAMING_SNAKE_CASE				(  _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE				):
							_snake_case       =  len(highway[0]				)

							for i in range(_SCREAMING_SNAKE_CASE				):
														_snake_case       =  update(highway[i]							, _SCREAMING_SNAKE_CASE							, _SCREAMING_SNAKE_CASE				)
														_snake_case       =  [-1] * number_of_cells

														for car_index in range(_SCREAMING_SNAKE_CASE				):
																					_snake_case       =  next_speeds_calculated[car_index]
																					if speed != -1:
																												# Change the position based on the speed (with % to create the loop)
																												_snake_case       =  (car_index + speed) % number_of_cells
																												# Commit the change of position
																												_snake_case       =  speed
														highway.append(_SCREAMING_SNAKE_CASE				)

							return highway


if __name__ == "__main__":
				import doctest

				doctest.testmod()