





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


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

class 						UpperCamelCase__					( unittest.TestCase  ):



     '''simple docstring'''

     def 							SCREAMING_SNAKE_CASE__    (				self		:       int			)	->						int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		torch.randn(5		,2			)
           SCREAMING_SNAKE_CASE						=		torch.device("""cuda"""			) if torch.cuda.is_available() else torch.device("""cpu"""			)

           SCREAMING_SNAKE_CASE						=		send_to_device(__A		,__A			)
           self.assertTrue(torch.equal(resulta.cpu()		,__A			)			)

           SCREAMING_SNAKE_CASE						=		send_to_device((tensor, [tensor, tensor], 1)		,__A			)
           self.assertIsInstance(__A		,__A			)
           self.assertTrue(torch.equal(resulta[0].cpu()		,__A			)			)
           self.assertIsInstance(resulta[1]		,__A			)
           self.assertTrue(torch.equal(resulta[1][0].cpu()		,__A			)			)
           self.assertTrue(torch.equal(resulta[1][1].cpu()		,__A			)			)
           self.assertEqual(resulta[2]		,1			)

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

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

           SCREAMING_SNAKE_CASE						=		send_to_device(UserDict({"""a""": tensor, """b""": [tensor, tensor], """c""": 1}			)		,__A			)
           self.assertIsInstance(__A		,__A			)
           self.assertTrue(torch.equal(resulta["""a"""].cpu()		,__A			)			)
           self.assertIsInstance(resulta["""b"""]		,__A			)
           self.assertTrue(torch.equal(resulta["""b"""][0].cpu()		,__A			)			)
           self.assertTrue(torch.equal(resulta["""b"""][1].cpu()		,__A			)			)
           self.assertEqual(resulta["""c"""]		,1			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Optional[int]			)	->						Optional[int]:




           '''simple docstring'''


           with self.assertRaises(__A			) as cm:
                 SCREAMING_SNAKE_CASE						=		recursively_apply(torch.tensor		,(torch.tensor(1			), 1)		,error_on_other_type=__A			)
           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		:       Any			)	->						int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		torch.tensor([1, 2, 3, 4, 5]			)
           self.assertEqual(listify(__A			)		,[1, 2, 3, 4, 5]			)

           SCREAMING_SNAKE_CASE						=		torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]			)
           self.assertEqual(listify(__A			)		,[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]			)

           SCREAMING_SNAKE_CASE						=		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(__A			)		,[[[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		:       Tuple			)	->						List[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 							SCREAMING_SNAKE_CASE__    (				self		:       str			)	->						Union[str, Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		RegressionModel()
           SCREAMING_SNAKE_CASE						=		model.forward
           SCREAMING_SNAKE_CASE						=		convert_outputs_to_fpaa(model.forward			)
           SCREAMING_SNAKE_CASE						=		extract_model_from_parallel(__A		,keep_fpaa_wrapper=__A			)
           SCREAMING_SNAKE_CASE						=		pickle.dumps(__A			)






     @require_cuda
     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[Any]			)	->						Dict:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		RegressionModel()
           SCREAMING_SNAKE_CASE						=		model.forward
           SCREAMING_SNAKE_CASE						=		torch.cuda.amp.autocast(dtype=torch.floataa			)(model.forward			)
           SCREAMING_SNAKE_CASE						=		convert_outputs_to_fpaa(model.forward			)
           SCREAMING_SNAKE_CASE						=		extract_model_from_parallel(__A		,keep_fpaa_wrapper=__A			)
           SCREAMING_SNAKE_CASE						=		pickle.dumps(__A			)






     @require_cuda
     @require_torch_min_version(version="""2.0"""			)
     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[str]			)	->						int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		RegressionModel()
           SCREAMING_SNAKE_CASE						=		model.forward
           SCREAMING_SNAKE_CASE						=		torch.cuda.amp.autocast(dtype=torch.floataa			)(model.forward			)
           SCREAMING_SNAKE_CASE						=		convert_outputs_to_fpaa(model.forward			)
           SCREAMING_SNAKE_CASE						=		torch.compile(model.forward		,backend="""inductor"""			)
           SCREAMING_SNAKE_CASE						=		torch.randn(4		,10			).cuda()
           SCREAMING_SNAKE_CASE						=		model(__A			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       int			)	->						Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		RegressionModel()
           # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
           SCREAMING_SNAKE_CASE						=		torch.nn.parallel.DataParallel(__A			)
           SCREAMING_SNAKE_CASE						=		extract_model_from_parallel(__A			)

           self.assertEqual(__A		,__A			)






     @require_torch_min_version(version="""2.0"""			)
     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[str]			)	->						str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		RegressionModel()
           SCREAMING_SNAKE_CASE						=		torch.compile(__A			)

           # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
           SCREAMING_SNAKE_CASE						=		torch.nn.parallel.DataParallel(__A			)
           SCREAMING_SNAKE_CASE						=		torch.compile(__A			)
           SCREAMING_SNAKE_CASE						=		extract_model_from_parallel(__A			)

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






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Union[str, Any]			)	->						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"""]			)			)



'''simple docstring'''

from .prime_check import Test

Test()





