






'''simple docstring'''





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


a_      = logging.get_logger(__name__)

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


a_      = {
    'AutoTokenizer': 'PreTrainedTokenizerBase',
    'AutoFeatureExtractor': 'FeatureExtractionMixin',
    'AutoImageProcessor': 'ImageProcessingMixin',
}





class 		SCREAMING_SNAKE_CASE__   (     lowercase_    ):
		_UpperCAmelCase         =['''feature_extractor''', '''tokenizer''']
		# Names need to be attr_class for attr in attributes
		_UpperCAmelCase         =None
		_UpperCAmelCase         =None
		_UpperCAmelCase         =None



		def __init__(    self:    Optional[Any]				,   *a:    Optional[int]				,   **a:    Tuple)  ->List[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(a				,   self.attributes):
														if attribute_name in kwargs:
																				raise TypeError(f"""Got multiple values for argument {attribute_name}.""")
														else:
																				a_				     =   arg

								if len(a) != len(self.attributes):
														raise ValueError(
														    f"""This processor requires {len(self.attributes)} arguments: {', '.join(self.attributes)}. Got """
														    f"""{len(a)} 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():
														a_				     =   getattr(self				,   f"""{attribute_name}_class""")
														# Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
														a_				     =   AUTO_TO_BASE_CLASS_MAPPING.get(a				,   a)
														if isinstance(a				,   a):
																				a_				     =   tuple(getattr(a				,   a) for n in class_name if n is not None)
														else:
																				a_				     =   getattr(a				,   a)

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

														setattr(self				,   a				,   a)



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






								'''simple docstring'''


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



		def _lowerCAmelCase     (    self:    Union[str, Any]				,   a:    Union[str, Any]				,   a:    bool = False				,   **a:    Optional[Any])  ->Any:






								'''simple docstring'''


								os.makedirs(a				,   exist_ok=a)

								if push_to_hub:
														a_				     =   kwargs.pop("commit_message"				,   a)
														a_				     =   kwargs.pop("repo_id"				,   save_directory.split(os.path.sep)[-1])
														a_				     =   self._create_repo(a				,   **a)
														a_				     =   self._get_files_timestamps(a)
								# 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:
														a_				     =   [getattr(self				,   a) for attribute_name in self.attributes]
														a_				     =   [(a.init_kwargs if isinstance(a				,   a) else a) for a in attrs]
														custom_object_save(self				,   a				,   config=a)

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

								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:
																				a_				     =   getattr(self				,   a)
																				if isinstance(a				,   a):
																										del attribute.init_kwargs["auto_map"]

								if push_to_hub:
														self._upload_modified_files(
														    a				,   a				,   a				,   commit_message=a				,   token=kwargs.get("use_auth_token")				,   )



		@classmethod
		def _lowerCAmelCase     (    cls:    str				,   a:    Union[str, os.PathLike]				,   a:    Optional[Union[str, os.PathLike]] = None				,   a:    bool = False				,   a:    bool = False				,   a:    Optional[Union[str, bool]] = None				,   a:    str = "main"				,   **a:    Tuple				,   )  ->str:






								'''simple docstring'''


								a_				     =   cache_dir
								a_				     =   force_download
								a_				     =   local_files_only
								a_				     =   revision

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

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

								a_				     =   cls._get_arguments_from_pretrained(a				,   **a)
								return cls(*a)



		@classmethod
		def _lowerCAmelCase     (    cls:    Optional[Any]				,   a:    Optional[int]="AutoProcessor")  ->Dict:






								'''simple docstring'''


								if not isinstance(a				,   a):
														a_				     =   auto_class.__name__

								import transformers.models.auto as auto_module

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

								a_				     =   auto_class



		@classmethod
		def _lowerCAmelCase     (    cls:    List[Any]				,   a:    Optional[Any]				,   **a:    Optional[int])  ->int:






								'''simple docstring'''


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

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




		@property
		def _lowerCAmelCase     (    self:    Union[str, Any])  ->int:






								'''simple docstring'''


								a_				     =   getattr(self				,   self.attributes[0])
								return getattr(a				,   "model_input_names"				,   a)





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











'''simple docstring'''





from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
							import torch

a_      = logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
							from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):



							def 			__UpperCAmelCase				(cls				)		->				Optional[Any]:

													'''simple docstring'''


													return cls



def 			__UpperCAmelCase				(lowercase__	,lowercase__ = None	,lowercase__ = None	,lowercase__ = None	,lowercase__ = None	,)		->				Union[str, Any]:

						'''simple docstring'''


						a_				     =   device
						a_				     =   shape[0]

						a_				     =   layout or torch.strided
						a_				     =   device or torch.device("cpu"				)

						if generator is not None:
												a_				     =   generator.device.type if not isinstance(lowercase__	,lowercase__				) else generator[0].device.type
												if gen_device_type != device.type and gen_device_type == "cpu":
																		a_				     =   "cpu"
																		if device != "mps":
																								logger.info(
																								    F"""The passed generator was created on 'cpu' even though a tensor on {device} was expected."""
																								    F""" Tensors will be created on 'cpu' and then moved to {device}. Note that one can probably"""
																								    F""" slighly speed up this function by passing a generator that was created on the {device} device."""				)
												elif gen_device_type != device.type and gen_device_type == "cuda":
																		raise ValueError(F"""Cannot generate a {device} tensor from a generator of type {gen_device_type}."""				)

						if isinstance(lowercase__	,lowercase__				):
												a_				     =   (1,) + shape[1:]
												a_				     =   [
												    torch.randn(lowercase__	,generator=generator[i]	,device=lowercase__	,dtype=lowercase__	,layout=lowercase__				)
												    for i in range(lowercase__				)
												]
												a_				     =   torch.cat(lowercase__	,dim=0				).to(lowercase__				)
						else:
												a_				     =   torch.randn(lowercase__	,generator=lowercase__	,device=lowercase__	,dtype=lowercase__	,layout=lowercase__				).to(lowercase__				)

						return latents



def 			__UpperCAmelCase				(lowercase__				)		->				Tuple:

						'''simple docstring'''


						if is_torch_version("<"	,"2.0.0"				) or not hasattr(lowercase__	,"_dynamo"				):
												return False
						return isinstance(lowercase__	,torch._dynamo.eval_frame.OptimizedModule				)



