





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  ):






			def     lowerCamelCase						(self					)  ->     List[Any]:


							'''simple docstring'''



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

							snake_case_		:					Any			    =						send_to_device(__magic_name__      ,							__magic_name__					)
							self.assertTrue(torch.equal(resulta.cpu()      ,							__magic_name__					)					)

							snake_case_		:					Tuple			    =						send_to_device((tensor, [tensor, tensor], 1)      ,							__magic_name__					)
							self.assertIsInstance(__magic_name__      ,							__magic_name__					)
							self.assertTrue(torch.equal(resulta[0].cpu()      ,							__magic_name__					)					)
							self.assertIsInstance(resulta[1]      ,							__magic_name__					)
							self.assertTrue(torch.equal(resulta[1][0].cpu()      ,							__magic_name__					)					)
							self.assertTrue(torch.equal(resulta[1][1].cpu()      ,							__magic_name__					)					)
							self.assertEqual(resulta[2]      ,							1					)

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

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

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






			def     lowerCamelCase						(self					)  ->     Dict:


							'''simple docstring'''



							with self.assertRaises(__magic_name__					) as cm:
											snake_case_		:					List[Any]			    =						recursively_apply(torch.tensor      ,							(torch.tensor(1					), 1)      ,							error_on_other_type=__magic_name__					)
							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     lowerCamelCase						(self					)  ->     Any:


							'''simple docstring'''



							snake_case_		:					List[Any]			    =						torch.tensor([1, 2, 3, 4, 5]					)
							self.assertEqual(listify(__magic_name__					)      ,							[1, 2, 3, 4, 5]					)

							snake_case_		:					str			    =						torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)
							self.assertEqual(listify(__magic_name__					)      ,							[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)

							snake_case_		:					Dict			    =						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(__magic_name__					)      ,							[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]					)






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


							'''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     lowerCamelCase						(self					)  ->     Optional[int]:


							'''simple docstring'''



							snake_case_		:					Dict			    =						RegressionModel()
							snake_case_		:					Any			    =						model.forward
							snake_case_		:					List[Any]			    =						convert_outputs_to_fpaa(model.forward					)
							snake_case_		:					Optional[int]			    =						extract_model_from_parallel(__magic_name__      ,							keep_fpaa_wrapper=__magic_name__					)
							snake_case_		:					List[str]			    =						pickle.dumps(__magic_name__					)






			@require_cuda
			def     lowerCamelCase						(self					)  ->     Optional[Any]:


							'''simple docstring'''



							snake_case_		:					Dict			    =						RegressionModel()
							snake_case_		:					Optional[int]			    =						model.forward
							snake_case_		:					List[str]			    =						torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
							snake_case_		:					int			    =						convert_outputs_to_fpaa(model.forward					)
							snake_case_		:					Optional[int]			    =						extract_model_from_parallel(__magic_name__      ,							keep_fpaa_wrapper=__magic_name__					)
							snake_case_		:					str			    =						pickle.dumps(__magic_name__					)






			@require_cuda
			@require_torch_min_version(version='''2.0'''					)
			def     lowerCamelCase						(self					)  ->     Dict:


							'''simple docstring'''



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






			def     lowerCamelCase						(self					)  ->     int:


							'''simple docstring'''



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

							self.assertEqual(__magic_name__      ,							__magic_name__					)






			@require_torch_min_version(version='''2.0'''					)
			def     lowerCamelCase						(self					)  ->     Any:


							'''simple docstring'''



							snake_case_		:					Optional[Any]			    =						RegressionModel()
							snake_case_		:					Dict			    =						torch.compile(__magic_name__					)

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

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






			def     lowerCamelCase						(self					)  ->     Tuple:


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

from ...utils import logging
from .image_processing_levit import LevitImageProcessor


lowerCAmelCase_												=       logging.get_logger(__name__)



class       __lowerCAmelCase (    _a  ):






			def __init__(self      ,							*__magic_name__      ,							**__magic_name__					)  ->     None:


							'''simple docstring'''



							warnings.warn(
							    '''The class LevitFeatureExtractor is deprecated and will be removed in version 5 of Transformers. Please'''
							    ''' use LevitImageProcessor instead.'''      ,							__magic_name__      ,							)
							super().__init__(*__magic_name__      ,							**__magic_name__					)
