





import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


lowerCAmelCase_        =			namedtuple("""ExampleNamedTuple""", """a b c""")



class 				_lowerCAmelCase			(	unittest.TestCase							):


  '''simple docstring'''







  def UpperCamelCase_       (					self					:    str							):


    '''simple docstring'''





    _snake_case      : Union[str, Any]														=		torch.randn(5					, 2							)
    _snake_case      : Optional[Any]														=		torch.device('cuda'							) if torch.cuda.is_available() else torch.device('cpu'							)

    _snake_case      : List[Any]														=		send_to_device(UpperCamelCase					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta.cpu()					, UpperCamelCase							)							)

    _snake_case      : Optional[Any]														=		send_to_device((tensor, [tensor, tensor], 1)					, UpperCamelCase							)
    self.assertIsInstance(UpperCamelCase					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta[0].cpu()					, UpperCamelCase							)							)
    self.assertIsInstance(resulta[1]					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta[1][0].cpu()					, UpperCamelCase							)							)
    self.assertTrue(torch.equal(resulta[1][1].cpu()					, UpperCamelCase							)							)
    self.assertEqual(resulta[2]					, 1							)

    _snake_case      : List[Any]														=		send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}					, UpperCamelCase							)
    self.assertIsInstance(UpperCamelCase					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta['a'].cpu()					, UpperCamelCase							)							)
    self.assertIsInstance(resulta['b']					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta['b'][0].cpu()					, UpperCamelCase							)							)
    self.assertTrue(torch.equal(resulta['b'][1].cpu()					, UpperCamelCase							)							)
    self.assertEqual(resulta['c']					, 1							)

    _snake_case      : List[str]														=		send_to_device(ExampleNamedTuple(a=UpperCamelCase					, b=[tensor, tensor]					, c=1							)					, UpperCamelCase							)
    self.assertIsInstance(UpperCamelCase					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta.a.cpu()					, UpperCamelCase							)							)
    self.assertIsInstance(resulta.b					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta.b[0].cpu()					, UpperCamelCase							)							)
    self.assertTrue(torch.equal(resulta.b[1].cpu()					, UpperCamelCase							)							)
    self.assertEqual(resulta.c					, 1							)

    _snake_case      : List[str]														=		send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}							)					, UpperCamelCase							)
    self.assertIsInstance(UpperCamelCase					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta['a'].cpu()					, UpperCamelCase							)							)
    self.assertIsInstance(resulta['b']					, UpperCamelCase							)
    self.assertTrue(torch.equal(resulta['b'][0].cpu()					, UpperCamelCase							)							)
    self.assertTrue(torch.equal(resulta['b'][1].cpu()					, UpperCamelCase							)							)
    self.assertEqual(resulta['c']					, 1							)







  def UpperCamelCase_       (					self					:    Dict							):


    '''simple docstring'''





    with self.assertRaises(UpperCamelCase							) as cm:
      _snake_case      : str														=		recursively_apply(torch.tensor					, (torch.tensor(1							), 1)					, error_on_other_type=UpperCamelCase							)
    self.assertEqual(
        str(cm.exception							)					, 'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'					, )







  def UpperCamelCase_       (					self					:    str							):


    '''simple docstring'''





    _snake_case      : int														=		torch.tensor([1, 2, 3, 4, 5]							)
    self.assertEqual(listify(UpperCamelCase							)					, [1, 2, 3, 4, 5]							)

    _snake_case      : Dict														=		torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]							)
    self.assertEqual(listify(UpperCamelCase							)					, [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]							)

    _snake_case      : List[str]														=		torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]							)
    self.assertEqual(
        listify(UpperCamelCase							)					, [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]							)







  def UpperCamelCase_       (					self					:    str							):


    '''simple docstring'''





    with patch_environment(aa=1					, BB=2							):
      self.assertEqual(os.environ.get('AA'							)					, '1'							)
      self.assertEqual(os.environ.get('BB'							)					, '2'							)

    self.assertNotIn('AA'					, os.environ							)
    self.assertNotIn('BB'					, os.environ							)







  def UpperCamelCase_       (					self					:    str							):


    '''simple docstring'''





    _snake_case      : Any														=		RegressionModel()
    _snake_case      : Dict														=		model.forward
    _snake_case      : List[Any]														=		convert_outputs_to_fpaa(model.forward							)
    _snake_case      : Union[str, Any]														=		extract_model_from_parallel(UpperCamelCase					, keep_fpaa_wrapper=UpperCamelCase							)
    _snake_case      : Any														=		pickle.dumps(UpperCamelCase							)







  @require_cuda
  def UpperCamelCase_       (					self					:    List[Any]							):


    '''simple docstring'''





    _snake_case      : Dict														=		RegressionModel()
    _snake_case      : Any														=		model.forward
    _snake_case      : Union[str, Any]														=		torch.cuda.amp.autocast(dtype=torch.floataa							)(model.forward							)
    _snake_case      : Optional[int]														=		convert_outputs_to_fpaa(model.forward							)
    _snake_case      : Union[str, Any]														=		extract_model_from_parallel(UpperCamelCase					, keep_fpaa_wrapper=UpperCamelCase							)
    _snake_case      : Optional[int]														=		pickle.dumps(UpperCamelCase							)







  @require_cuda
  @require_torch_min_version(version='2.0'							)
  def UpperCamelCase_       (					self					:    List[str]							):


    '''simple docstring'''





    _snake_case      : Dict														=		RegressionModel()
    _snake_case      : Tuple														=		model.forward
    _snake_case      : Tuple														=		torch.cuda.amp.autocast(dtype=torch.floataa							)(model.forward							)
    _snake_case      : Dict														=		convert_outputs_to_fpaa(model.forward							)
    _snake_case      : List[Any]														=		torch.compile(model.forward					, backend='inductor'							)
    _snake_case      : str														=		torch.randn(4					, 10							).cuda()
    _snake_case      : List[str]														=		model(UpperCamelCase							)







  def UpperCamelCase_       (					self					:    Any							):


    '''simple docstring'''





    _snake_case      : Any														=		RegressionModel()
    # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
    _snake_case      : str														=		torch.nn.parallel.DataParallel(UpperCamelCase							)
    _snake_case      : Optional[Any]														=		extract_model_from_parallel(UpperCamelCase							)

    self.assertEqual(UpperCamelCase					, UpperCamelCase							)







  @require_torch_min_version(version='2.0'							)
  def UpperCamelCase_       (					self					:    str							):


    '''simple docstring'''





    _snake_case      : int														=		RegressionModel()
    _snake_case      : List[str]														=		torch.compile(UpperCamelCase							)

    # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
    _snake_case      : Any														=		torch.nn.parallel.DataParallel(UpperCamelCase							)
    _snake_case      : Union[str, Any]														=		torch.compile(UpperCamelCase							)
    _snake_case      : Optional[Any]														=		extract_model_from_parallel(UpperCamelCase							)

    self.assertEqual(compiled_model._orig_mod					, compiled_model_unwrapped._orig_mod							)







  def UpperCamelCase_       (					self					:    Optional[Any]							):


    '''simple docstring'''





    self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3]							)]							)					, torch.device('cpu'							)							)
    self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3]							)}							)					, torch.device('cpu'							)							)
    self.assertIsNone(find_device([1, 'a']							)							)





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


UpperCAmelCase							= logging.get_logger(__name__)

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


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




class       __lowerCAmelCase     (				a_  ):



 '''simple docstring'''
 _lowerCamelCase  :		Optional[int] 			=       ['feature_extractor', 'tokenizer']
 # Names need to be attr_class for attr in attributes
 _lowerCamelCase  :		List[Any] 			=       None
 _lowerCamelCase  :		Optional[Any] 			=       None
 _lowerCamelCase  :		Tuple 			=       None

 def __init__(	self				:						Tuple				,	*lowerCamelCase				:						Optional[Any]				,	**lowerCamelCase				:						List[Any]     )       ->							str:
        # Sanitize args and kwargs
        for key in kwargs:
               if key not in self.attributes:
                      raise TypeError(f"Unexpected keyword argument {key}."     )
        for arg, attribute_name in zip(lowerCamelCase				,	self.attributes     ):
               if attribute_name in kwargs:
                      raise TypeError(f"Got multiple values for argument {attribute_name}."     )
               else:
                      _UpperCamelCase   :				Any       =     arg

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

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

               setattr(self				,	lowerCamelCase				,	lowerCamelCase     )

 def __repr__(	self				:						Dict     )       ->							str:
        _UpperCamelCase   :				List[str]       =     [f"- {name}: {repr(getattr(self				,	lowerCamelCase     )     )}" for name in self.attributes]
        _UpperCamelCase   :				int       =     '\n'.join(lowerCamelCase     )
        return f"{self.__class__.__name__}:\n{attributes_repr}"

 def     __snake_case    (	self				:						Tuple				,	lowerCamelCase				:						int				,	lowerCamelCase				:						bool = False				,	**lowerCamelCase				:						Optional[Any]     )       ->							Any:
        os.makedirs(lowerCamelCase				,	exist_ok=lowerCamelCase     )

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

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

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

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

 @classmethod
 def     __snake_case    (	cls				:						Tuple				,	lowerCamelCase				:						Union[str, os.PathLike]				,	lowerCamelCase				:						Optional[Union[str, os.PathLike]] = None				,	lowerCamelCase				:						bool = False				,	lowerCamelCase				:						bool = False				,	lowerCamelCase				:						Optional[Union[str, bool]] = None				,	lowerCamelCase				:						str = "main"				,	**lowerCamelCase				:						Tuple				,	)       ->							Optional[Any]:
        _UpperCamelCase   :				Any       =     cache_dir
        _UpperCamelCase   :				Optional[int]       =     force_download
        _UpperCamelCase   :				int       =     local_files_only
        _UpperCamelCase   :				Dict       =     revision

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

        if token is not None:
               # change to `token` in a follow-up PR
               _UpperCamelCase   :				List[Any]       =     token

        _UpperCamelCase   :				List[Any]       =     cls._get_arguments_from_pretrained(lowerCamelCase				,	**lowerCamelCase     )
        return cls(*lowerCamelCase     )

 @classmethod
 def     __snake_case    (	cls				:						Dict				,	lowerCamelCase				:						Optional[Any]="AutoProcessor"     )       ->							Optional[int]:
        if not isinstance(lowerCamelCase				,	lowerCamelCase     ):
               _UpperCamelCase   :				Tuple       =     auto_class.__name__

        import transformers.models.auto as auto_module

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

        _UpperCamelCase   :				Union[str, Any]       =     auto_class

 @classmethod
 def     __snake_case    (	cls				:						int				,	lowerCamelCase				:						Any				,	**lowerCamelCase				:						List[Any]     )       ->							Optional[Any]:
        _UpperCamelCase   :				Optional[int]       =     []
        for attribute_name in cls.attributes:
               _UpperCamelCase   :				Dict       =     getattr(cls				,	f"{attribute_name}_class"     )
               if isinstance(lowerCamelCase				,	lowerCamelCase     ):
                      _UpperCamelCase   :				Any       =     tuple(getattr(lowerCamelCase				,	lowerCamelCase     ) if n is not None else None for n in class_name     )
                      _UpperCamelCase   :				List[Any]       =     kwargs.get('use_fast'				,	lowerCamelCase     )
                      if use_fast and classes[1] is not None:
                             _UpperCamelCase   :				Optional[int]       =     classes[1]
                      else:
                             _UpperCamelCase   :				Any       =     classes[0]
               else:
                      _UpperCamelCase   :				Any       =     getattr(lowerCamelCase				,	lowerCamelCase     )

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

 @property
 def     __snake_case    (	self				:						List[Any]     )       ->							Optional[Any]:
        _UpperCamelCase   :				List[str]       =     getattr(self				,	self.attributes[0]     )
        return getattr(lowerCamelCase				,	'model_input_names'				,	lowerCamelCase     )







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



