


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


snake_case       :					Dict        =  namedtuple('''ExampleNamedTuple''', '''a b c''')







class 	_snake_case						(   unittest.TestCase							):




 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Any											= torch.randn(5      ,						2					)
       a		:Any											= torch.device('''cuda'''					) if torch.cuda.is_available() else torch.device('''cpu'''					)

       a		:int											= send_to_device(_lowerCamelCase      ,						_lowerCamelCase					)
       self.assertTrue(torch.equal(resulta.cpu()      ,						_lowerCamelCase					)					)

       a		:List[Any]											= send_to_device((tensor, [tensor, tensor], 1)      ,						_lowerCamelCase					)
       self.assertIsInstance(_lowerCamelCase      ,						_lowerCamelCase					)
       self.assertTrue(torch.equal(resulta[0].cpu()      ,						_lowerCamelCase					)					)
       self.assertIsInstance(resulta[1]      ,						_lowerCamelCase					)
       self.assertTrue(torch.equal(resulta[1][0].cpu()      ,						_lowerCamelCase					)					)
       self.assertTrue(torch.equal(resulta[1][1].cpu()      ,						_lowerCamelCase					)					)
       self.assertEqual(resulta[2]      ,						1					)

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

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

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




 def  SCREAMING_SNAKE_CASE__       (	self					):
       with self.assertRaises(_lowerCamelCase					) as cm:
             a		:List[Any]											= recursively_apply(torch.tensor      ,						(torch.tensor(1					), 1)      ,						error_on_other_type=_lowerCamelCase					)
       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  SCREAMING_SNAKE_CASE__       (	self					):
       a		:List[str]											= torch.tensor([1, 2, 3, 4, 5]					)
       self.assertEqual(listify(_lowerCamelCase					)      ,						[1, 2, 3, 4, 5]					)

       a		:int											= torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)
       self.assertEqual(listify(_lowerCamelCase					)      ,						[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)

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




 def  SCREAMING_SNAKE_CASE__       (	self					):
       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  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Any											= RegressionModel()
       a		:List[str]											= model.forward
       a		:Optional[int]											= convert_outputs_to_fpaa(model.forward					)
       a		:List[str]											= extract_model_from_parallel(_lowerCamelCase      ,						keep_fpaa_wrapper=_lowerCamelCase					)
       a		:Optional[Any]											= pickle.dumps(_lowerCamelCase					)




 @require_cuda
 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Union[str, Any]											= RegressionModel()
       a		:str											= model.forward
       a		:Dict											= torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
       a		:int											= convert_outputs_to_fpaa(model.forward					)
       a		:Optional[Any]											= extract_model_from_parallel(_lowerCamelCase      ,						keep_fpaa_wrapper=_lowerCamelCase					)
       a		:Optional[Any]											= pickle.dumps(_lowerCamelCase					)




 @require_cuda
 @require_torch_min_version(version='''2.0'''					)
 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Optional[Any]											= RegressionModel()
       a		:List[str]											= model.forward
       a		:Any											= torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
       a		:Tuple											= convert_outputs_to_fpaa(model.forward					)
       a		:List[str]											= torch.compile(model.forward      ,						backend='''inductor'''					)
       a		:Optional[int]											= torch.randn(4      ,						10					).cuda()
       a		:Optional[Any]											= model(_lowerCamelCase					)




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

       self.assertEqual(_lowerCamelCase      ,						_lowerCamelCase					)




 @require_torch_min_version(version='''2.0'''					)
 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Union[str, Any]											= RegressionModel()
       a		:int											= torch.compile(_lowerCamelCase					)

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

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




 def  SCREAMING_SNAKE_CASE__       (	self					):
       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_convnext import ConvNextImageProcessor


snake_case       :					Tuple        =  logging.get_logger(__name__)







class 	_snake_case						(   _snake_case							):




 def __init__(	self      ,						*_lowerCamelCase      ,						**_lowerCamelCase					):
       warnings.warn(
           '''The class ConvNextFeatureExtractor is deprecated and will be removed in version 5 of Transformers.'''
           ''' Please use ConvNextImageProcessor instead.'''      ,						_lowerCamelCase      ,						)
       super().__init__(*_lowerCamelCase      ,						**_lowerCamelCase					)





