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

						def 						a__ (					self		:			str			):
													_UpperCAmelCase  :			Tuple	       =		torch.randn(5	,						2			)
													_UpperCAmelCase  :			List[Any]	       =		torch.device("cuda"			) if torch.cuda.is_available() else torch.device("cpu"			)

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

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

													_UpperCAmelCase  :			Any	       =		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			)

													_UpperCAmelCase  :			Union[str, 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			)

													_UpperCAmelCase  :			Dict	       =		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 						a__ (					self		:			List[str]			):
													with self.assertRaises(_lowerCamelCase			) as cm:
																				_UpperCAmelCase  :			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 						a__ (					self		:			Union[str, Any]			):
													_UpperCAmelCase  :			Any	       =		torch.tensor([1, 2, 3, 4, 5]			)
													self.assertEqual(listify(_lowerCamelCase			)	,						[1, 2, 3, 4, 5]			)

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

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

						@require_cuda
						def 						a__ (					self		:			List[Any]			):
													_UpperCAmelCase  :			List[Any]	       =		RegressionModel()
													_UpperCAmelCase  :			int	       =		model.forward
													_UpperCAmelCase  :			Tuple	       =		torch.cuda.amp.autocast(dtype=torch.floataa			)(model.forward			)
													_UpperCAmelCase  :			Tuple	       =		convert_outputs_to_fpaa(model.forward			)
													_UpperCAmelCase  :			Optional[Any]	       =		extract_model_from_parallel(_lowerCamelCase	,						keep_fpaa_wrapper=_lowerCamelCase			)
													_UpperCAmelCase  :			List[str]	       =		pickle.dumps(_lowerCamelCase			)

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

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

													self.assertEqual(_lowerCamelCase	,						_lowerCamelCase			)

						@require_torch_min_version(version="2.0"			)
						def 						a__ (					self		:			Optional[Any]			):
													_UpperCAmelCase  :			Optional[Any]	       =		RegressionModel()
													_UpperCAmelCase  :			Union[str, Any]	       =		torch.compile(_lowerCamelCase			)

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

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






						def 						a__ (					self		:			List[str]			):
													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"]			)			)






def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )   ->      Dict:

							"""simple docstring"""






							if power < 0:
														return -1
							base %= mod
							_UpperCAmelCase  :			Optional[Any]	       =		1

							while power > 0:
														if power & 1:
																					_UpperCAmelCase  :			Optional[Any]	       =		(result * base) % mod
														_UpperCAmelCase  :			List[str]	       =		power >> 1
														_UpperCAmelCase  :			Optional[int]	       =		(base * base) % mod

							return result




def 		lowerCAmelCase_     (     )   ->      List[str]:

							"""simple docstring"""






							print(modular_exponential(3      ,	2_0_0      ,	1_3  )  )


if __name__ == "__main__":
				import doctest

				doctest.testmod()

				main()




