





import os
import warnings
from pathlib import Path
from typing import Optional, Union

from .dynamic_module_utils import custom_object_save
from .tokenization_utils_base import PreTrainedTokenizerBase
from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging


__lowercase													=			logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
__lowercase													=			direct_transformers_import(Path(__file__).parent)


__lowercase													=			{
    """AutoTokenizer""": """PreTrainedTokenizerBase""",
    """AutoFeatureExtractor""": """FeatureExtractionMixin""",
    """AutoImageProcessor""": """ImageProcessingMixin""",
}

class   __snake_case				(			_a   ):






				'''simple docstring'''
				_lowerCAmelCase     =	["""feature_extractor""", """tokenizer"""]
				# Names need to be attr_class for attr in attributes
				_lowerCAmelCase     =	None
				_lowerCAmelCase     =	None
				_lowerCAmelCase     =	None





				def __init__(   self						:		List[str]			,						*snake_case						:		Dict			,						**snake_case						:		Any							):

						"""simple docstring"""





						for key in kwargs:
								if key not in self.attributes:
										raise TypeError(f'''Unexpected keyword argument {key}.'''							)
						for arg, attribute_name in zip(snake_case			,						self.attributes							):
								if attribute_name in kwargs:
										raise TypeError(f'''Got multiple values for argument {attribute_name}.'''							)
								else:
										__lowerCAmelCase			=				arg

						if len(snake_case							) != len(self.attributes							):
								raise ValueError(
								    f'''This processor requires {len(self.attributes							)} arguments: {", ".join(self.attributes							)}. Got '''
								    f'''{len(snake_case							)} arguments instead.'''							)

						# Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
						for attribute_name, arg in kwargs.items():
								__lowerCAmelCase			=				getattr(self			,						f'''{attribute_name}_class'''							)
								# Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
								__lowerCAmelCase			=				AUTO_TO_BASE_CLASS_MAPPING.get(snake_case			,						snake_case							)
								if isinstance(snake_case			,						snake_case							):
										__lowerCAmelCase			=				tuple(getattr(snake_case			,						snake_case							) for n in class_name if n is not None							)
								else:
										__lowerCAmelCase			=				getattr(snake_case			,						snake_case							)

								if not isinstance(snake_case			,						snake_case							):
										raise ValueError(
										    f'''Received a {type(snake_case							).__name__} for argument {attribute_name}, but a {class_name} was expected.'''							)

								setattr(self			,						snake_case			,						snake_case							)





				def __repr__(   self						:		List[str]							):

						"""simple docstring"""





						__lowerCAmelCase			=				[f'''- {name}: {repr(getattr(self			,						snake_case							)							)}''' for name in self.attributes]
						__lowerCAmelCase			=				'''\n'''.join(snake_case							)
						return f'''{self.__class__.__name__}:\n{attributes_repr}'''





				def 	A		(   self						:		List[str]			,						snake_case						:		Dict			,						snake_case						:		bool = False			,						**snake_case						:		Tuple							):

						"""simple docstring"""





						os.makedirs(snake_case			,						exist_ok=snake_case							)

						if push_to_hub:
								__lowerCAmelCase			=				kwargs.pop('''commit_message'''			,						snake_case							)
								__lowerCAmelCase			=				kwargs.pop('''repo_id'''			,						save_directory.split(os.path.sep							)[-1]							)
								__lowerCAmelCase			=				self._create_repo(snake_case			,						**snake_case							)
								__lowerCAmelCase			=				self._get_files_timestamps(snake_case							)
						# If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
						# loaded from the Hub.
						if self._auto_class is not None:
								__lowerCAmelCase			=				[getattr(self			,						snake_case							) for attribute_name in self.attributes]
								__lowerCAmelCase			=				[(a.init_kwargs if isinstance(snake_case			,						snake_case							) else a) for a in attrs]
								custom_object_save(self			,						snake_case			,						config=snake_case							)

						for attribute_name in self.attributes:
								__lowerCAmelCase			=				getattr(self			,						snake_case							)
								# Include the processor class in the attribute config so this processor can then be reloaded with the
								# `AutoProcessor` API.
								if hasattr(snake_case			,						'''_set_processor_class'''							):
										attribute._set_processor_class(self.__class__.__name__							)
								attribute.save_pretrained(snake_case							)

						if self._auto_class is not None:
								# We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
								for attribute_name in self.attributes:
										__lowerCAmelCase			=				getattr(self			,						snake_case							)
										if isinstance(snake_case			,						snake_case							):
												del attribute.init_kwargs["auto_map"]

						if push_to_hub:
								self._upload_modified_files(
								    snake_case			,						snake_case			,						snake_case			,						commit_message=snake_case			,						token=kwargs.get('''use_auth_token'''							)			,						)





				@classmethod
				def 	A		(   cls						:		Any			,						snake_case						:		Union[str, os.PathLike]			,						snake_case						:		Optional[Union[str, os.PathLike]] = None			,						snake_case						:		bool = False			,						snake_case						:		bool = False			,						snake_case						:		Optional[Union[str, bool]] = None			,						snake_case						:		str = "main"			,						**snake_case						:		Optional[int]			,						):

						"""simple docstring"""





						__lowerCAmelCase			=				cache_dir
						__lowerCAmelCase			=				force_download
						__lowerCAmelCase			=				local_files_only
						__lowerCAmelCase			=				revision

						__lowerCAmelCase			=				kwargs.pop('''use_auth_token'''			,						snake_case							)
						if use_auth_token is not None:
								warnings.warn(
								    '''The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers.'''			,						snake_case							)
								if token is not None:
										raise ValueError(
										    '''`token` and `use_auth_token` are both specified. Please set only the argument `token`.'''							)
								__lowerCAmelCase			=				use_auth_token

						if token is not None:
								# change to `token` in a follow-up PR
								__lowerCAmelCase			=				token

						__lowerCAmelCase			=				cls._get_arguments_from_pretrained(snake_case			,						**snake_case							)
						return cls(*snake_case							)





				@classmethod
				def 	A		(   cls						:		Union[str, Any]			,						snake_case						:		Dict="AutoProcessor"							):

						"""simple docstring"""





						if not isinstance(snake_case			,						snake_case							):
								__lowerCAmelCase			=				auto_class.__name__

						import transformers.models.auto as auto_module

						if not hasattr(snake_case			,						snake_case							):
								raise ValueError(f'''{auto_class} is not a valid auto class.'''							)

						__lowerCAmelCase			=				auto_class





				@classmethod
				def 	A		(   cls						:		List[Any]			,						snake_case						:		int			,						**snake_case						:		Optional[int]							):

						"""simple docstring"""





						__lowerCAmelCase			=				[]
						for attribute_name in cls.attributes:
								__lowerCAmelCase			=				getattr(cls			,						f'''{attribute_name}_class'''							)
								if isinstance(snake_case			,						snake_case							):
										__lowerCAmelCase			=				tuple(getattr(snake_case			,						snake_case							) if n is not None else None for n in class_name							)
										__lowerCAmelCase			=				kwargs.get('''use_fast'''			,						snake_case							)
										if use_fast and classes[1] is not None:
												__lowerCAmelCase			=				classes[1]
										else:
												__lowerCAmelCase			=				classes[0]
								else:
										__lowerCAmelCase			=				getattr(snake_case			,						snake_case							)

								args.append(attribute_class.from_pretrained(snake_case			,						**snake_case							)							)
						return args





				@property
				def 	A		(   self						:		Tuple							):

						"""simple docstring"""





						__lowerCAmelCase			=				getattr(self			,						self.attributes[0]							)
						return getattr(snake_case			,						'''model_input_names'''			,						snake_case							)


__lowercase													=			copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
						__lowercase													=			ProcessorMixin.push_to_hub.__doc__.format(
						    object="""processor""", object_class="""AutoProcessor""", object_files="""processor files"""
						)







import inspect
import os
import unittest

import torch

import accelerate
from accelerate import debug_launcher
from accelerate.test_utils import (
    execute_subprocess_async,
    require_cpu,
    require_multi_gpu,
    require_single_gpu,
    test_sync,
)
from accelerate.utils import patch_environment

class   __snake_case				(			unittest.TestCase   ):






    '''simple docstring'''





    def 	A		(   self						:		int							):

      """simple docstring"""





      __lowerCAmelCase			=				inspect.getfile(accelerate.test_utils							)
      __lowerCAmelCase			=				os.path.sep.join(mod_file.split(os.path.sep							)[:-1] + ['''scripts''', '''test_sync.py''']							)





    @require_cpu
    def 	A		(   self						:		str							):

      """simple docstring"""





      debug_launcher(test_sync.main			,						num_processes=1							)





    @require_cpu
    def 	A		(   self						:		List[str]							):

      """simple docstring"""





      debug_launcher(test_sync.main							)





    @require_single_gpu
    def 	A		(   self						:		Optional[Any]							):

      """simple docstring"""





      test_sync.main()





    @require_multi_gpu
    def 	A		(   self						:		Union[str, Any]							):

      """simple docstring"""





      print(f'''Found {torch.cuda.device_count()} devices.'''							)
      __lowerCAmelCase			=				['''torchrun''', f'''--nproc_per_node={torch.cuda.device_count()}''', self.test_file_path]
      with patch_environment(omp_num_threads=1							):
        execute_subprocess_async(snake_case			,						env=os.environ.copy()							)
