
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 	A__				(	unittest.TestCase      ):






				def 				_lowerCamelCase						(		self   : str  ):






							'''simple docstring'''







							lowerCAmelCase__ : Dict   =						torch.randn(5       ,	2  )
							lowerCAmelCase__ : Dict   =						torch.device('cuda'  ) if torch.cuda.is_available() else torch.device('cpu'  )

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

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

							lowerCAmelCase__ : Tuple   =						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  )

							lowerCAmelCase__ : List[Any]   =						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  )

							lowerCAmelCase__ : Union[str, Any]   =						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 				_lowerCamelCase						(		self   : int  ):






							'''simple docstring'''







							with self.assertRaises(UpperCAmelCase_  ) as cm:
										lowerCAmelCase__ : int   =						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 				_lowerCamelCase						(		self   : Union[str, Any]  ):






							'''simple docstring'''







							lowerCAmelCase__ : Union[str, Any]   =						torch.tensor([1, 2, 3, 4, 5]  )
							self.assertEqual(listify(UpperCAmelCase_  )       ,	[1, 2, 3, 4, 5]  )

							lowerCAmelCase__ : Tuple   =						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]]  )

							lowerCAmelCase__ : Optional[int]   =						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 				_lowerCamelCase						(		self   : List[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 				_lowerCamelCase						(		self   : Tuple  ):






							'''simple docstring'''







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






				@require_cuda
				def 				_lowerCamelCase						(		self   : int  ):






							'''simple docstring'''







							lowerCAmelCase__ : str   =						RegressionModel()
							lowerCAmelCase__ : Any   =						model.forward
							lowerCAmelCase__ : Any   =						torch.cuda.amp.autocast(dtype=torch.floataa  )(model.forward  )
							lowerCAmelCase__ : int   =						convert_outputs_to_fpaa(model.forward  )
							lowerCAmelCase__ : Tuple   =						extract_model_from_parallel(UpperCAmelCase_       ,	keep_fpaa_wrapper=UpperCAmelCase_  )
							lowerCAmelCase__ : Dict   =						pickle.dumps(UpperCAmelCase_  )






				@require_cuda
				@require_torch_min_version(version='2.0'  )
				def 				_lowerCamelCase						(		self   : Dict  ):






							'''simple docstring'''







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






				def 				_lowerCamelCase						(		self   : int  ):






							'''simple docstring'''







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

							self.assertEqual(UpperCAmelCase_       ,	UpperCAmelCase_  )






				@require_torch_min_version(version='2.0'  )
				def 				_lowerCamelCase						(		self   : Optional[int]  ):






							'''simple docstring'''







							lowerCAmelCase__ : int   =						RegressionModel()
							lowerCAmelCase__ : Optional[int]   =						torch.compile(UpperCAmelCase_  )

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

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






				def 				_lowerCamelCase						(		self   : Optional[int]  ):






							'''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']  )  )






from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


UpperCAmelCase__    :      Optional[Any]       					={
    '''configuration_rwkv''': ['''RWKV_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''RwkvConfig''', '''RwkvOnnxConfig'''],
}

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							UpperCAmelCase__    :      Optional[Any]       					=[
							    '''RWKV_PRETRAINED_MODEL_ARCHIVE_LIST''',
							    '''RwkvForCausalLM''',
							    '''RwkvModel''',
							    '''RwkvPreTrainedModel''',
							]


if TYPE_CHECKING:
							from .configuration_rwkv import RWKV_PRETRAINED_CONFIG_ARCHIVE_MAP, RwkvConfig, RwkvOnnxConfig

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_rwkv import (
														    RWKV_PRETRAINED_MODEL_ARCHIVE_LIST,
														    RwkvForCausalLM,
														    RwkvModel,
														    RwkvPreTrainedModel,
														)
else:
							import sys

							UpperCAmelCase__    :      int       					=_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)





