





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							:      List[str]						=	namedtuple('''ExampleNamedTuple''', '''a b c''')


class 					__magic_name__  (							unittest.TestCase							):





     """simple docstring"""



     def 						SCREAMING_SNAKE_CASE		(							self       :Optional[int]  ):





        '''simple docstring'''




        A_					:				Tuple          =			torch.randn(5			,      2  )
        A_					:				List[str]          =			torch.device("cuda"  ) if torch.cuda.is_available() else torch.device("cpu"  )

        A_					:				Any          =			send_to_device(snake_case			,      snake_case  )
        self.assertTrue(torch.equal(resulta.cpu()			,      snake_case  )  )

        A_					:				int          =			send_to_device((tensor, [tensor, tensor], 1)			,      snake_case  )
        self.assertIsInstance(snake_case			,      snake_case  )
        self.assertTrue(torch.equal(resulta[0].cpu()			,      snake_case  )  )
        self.assertIsInstance(resulta[1]			,      snake_case  )
        self.assertTrue(torch.equal(resulta[1][0].cpu()			,      snake_case  )  )
        self.assertTrue(torch.equal(resulta[1][1].cpu()			,      snake_case  )  )
        self.assertEqual(resulta[2]			,      1  )

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

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

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



     def 						SCREAMING_SNAKE_CASE		(							self       :str  ):





        '''simple docstring'''




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





        '''simple docstring'''




        A_					:				Dict          =			torch.tensor([1, 2, 3, 4, 5]  )
        self.assertEqual(listify(snake_case  )			,      [1, 2, 3, 4, 5]  )

        A_					:				Tuple          =			torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]  )
        self.assertEqual(listify(snake_case  )			,      [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]  )

        A_					:				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(snake_case  )			,      [[[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  ):





        '''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       :List[str]  ):





        '''simple docstring'''




        A_					:				List[str]          =			RegressionModel()
        A_					:				Tuple          =			model.forward
        A_					:				Dict          =			convert_outputs_to_fpaa(model.forward  )
        A_					:				int          =			extract_model_from_parallel(snake_case			,      keep_fpaa_wrapper=snake_case  )
        A_					:				Dict          =			pickle.dumps(snake_case  )



     @require_cuda
     def 						SCREAMING_SNAKE_CASE		(							self       :List[str]  ):





        '''simple docstring'''




        A_					:				Union[str, Any]          =			RegressionModel()
        A_					:				Tuple          =			model.forward
        A_					:				List[str]          =			torch.cuda.amp.autocast(dtype=torch.floataa  )(model.forward  )
        A_					:				List[Any]          =			convert_outputs_to_fpaa(model.forward  )
        A_					:				Any          =			extract_model_from_parallel(snake_case			,      keep_fpaa_wrapper=snake_case  )
        A_					:				Optional[Any]          =			pickle.dumps(snake_case  )



     @require_cuda
     @require_torch_min_version(version="2.0"  )
     def 						SCREAMING_SNAKE_CASE		(							self       :Any  ):





        '''simple docstring'''




        A_					:				str          =			RegressionModel()
        A_					:				List[str]          =			model.forward
        A_					:				int          =			torch.cuda.amp.autocast(dtype=torch.floataa  )(model.forward  )
        A_					:				Optional[Any]          =			convert_outputs_to_fpaa(model.forward  )
        A_					:				List[Any]          =			torch.compile(model.forward			,      backend="inductor"  )
        A_					:				Tuple          =			torch.randn(4			,      10  ).cuda()
        A_					:				List[str]          =			model(snake_case  )



     def 						SCREAMING_SNAKE_CASE		(							self       :Optional[Any]  ):





        '''simple docstring'''




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

        self.assertEqual(snake_case			,      snake_case  )



     @require_torch_min_version(version="2.0"  )
     def 						SCREAMING_SNAKE_CASE		(							self       :Tuple  ):





        '''simple docstring'''




        A_					:				Union[str, Any]          =			RegressionModel()
        A_					:				Dict          =			torch.compile(snake_case  )

        # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
        A_					:				int          =			torch.nn.parallel.DataParallel(snake_case  )
        A_					:				str          =			torch.compile(snake_case  )
        A_					:				Any          =			extract_model_from_parallel(snake_case  )

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





     def 						SCREAMING_SNAKE_CASE		(							self       :List[str]  ):





        '''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 __future__ import annotations

def      __snake_case						(					_lowerCAmelCase : int	) ->						list[int]:
   assert (
       isinstance(_lowerCAmelCase				,				_lowerCAmelCase	) and mask > 0
   ), f"mask needs to be positive integer, your input {mask}"
   A_					:				Optional[Any]          =			[]
   A_					:				Optional[int]          =			mask

   while submask:
      all_submasks.append(_lowerCAmelCase	)
      A_					:				List[str]          =			(submask - 1) & mask

   return all_submasks


if __name__ == "__main__":
 import doctest

 doctest.testmod()
