
"""simple docstring"""
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,
)


lowercase		:						Optional[Any] =					namedtuple('ExampleNamedTuple', 'a b c')




class    _snake_case  (						unittest.TestCase       ):






							def 							__lowercase  (					self						:			List[Any]    ):






												"""simple docstring"""



												__a:						Optional[int]		    =					torch.randn(5						, 2    )
												__a:						Optional[int]		    =					torch.device("""cuda"""    ) if torch.cuda.is_available() else torch.device("""cpu"""    )

												__a:						Dict		    =					send_to_device(snake_case__						, snake_case__    )
												self.assertTrue(torch.equal(resulta.cpu()						, snake_case__    )    )

												__a:						Tuple		    =					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:						int		    =					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:						List[str]		    =					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:						Tuple		    =					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 							__lowercase  (					self						:			Optional[Any]    ):






												"""simple docstring"""



												with self.assertRaises(snake_case__    ) as cm:
																	__a:						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 							__lowercase  (					self						:			str    ):






												"""simple docstring"""



												__a:						str		    =					torch.tensor([1, 2, 3, 4, 5]    )
												self.assertEqual(listify(snake_case__    )						, [1, 2, 3, 4, 5]    )

												__a:						Any		    =					torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]    )
												self.assertEqual(listify(snake_case__    )						, [[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]    )

												__a:						Optional[int]		    =					torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]    )
												self.assertEqual(
												    listify(snake_case__    )						, [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]    )






							def 							__lowercase  (					self						:			Optional[int]    ):






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






												"""simple docstring"""



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






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






												"""simple docstring"""



												__a:						Optional[int]		    =					RegressionModel()
												__a:						Dict		    =					model.forward
												__a:						str		    =					torch.cuda.amp.autocast(dtype=torch.floataa    )(model.forward    )
												__a:						List[str]		    =					convert_outputs_to_fpaa(model.forward    )
												__a:						List[str]		    =					extract_model_from_parallel(snake_case__						, keep_fpaa_wrapper=snake_case__    )
												__a:						List[Any]		    =					pickle.dumps(snake_case__    )






							@require_cuda
							@require_torch_min_version(version="""2.0"""    )
							def 							__lowercase  (					self						:			Union[str, Any]    ):






												"""simple docstring"""



												__a:						Tuple		    =					RegressionModel()
												__a:						int		    =					model.forward
												__a:						Tuple		    =					torch.cuda.amp.autocast(dtype=torch.floataa    )(model.forward    )
												__a:						Optional[Any]		    =					convert_outputs_to_fpaa(model.forward    )
												__a:						Optional[int]		    =					torch.compile(model.forward						, backend="""inductor"""    )
												__a:						List[str]		    =					torch.randn(4						, 1_0    ).cuda()
												__a:						Union[str, Any]		    =					model(snake_case__    )






							def 							__lowercase  (					self						:			Any    ):






												"""simple docstring"""



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

												self.assertEqual(snake_case__						, snake_case__    )






							@require_torch_min_version(version="""2.0"""    )
							def 							__lowercase  (					self						:			List[Any]    ):






												"""simple docstring"""



												__a:						str		    =					RegressionModel()
												__a:						Any		    =					torch.compile(snake_case__    )

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

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






							def 							__lowercase  (					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"""]    )    )


"""simple docstring"""
from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowercase		:						int =					logging.get_logger(__name__)

lowercase		:						Any =					{
    'facebook/regnet-y-040': 'https://huggingface.co/facebook/regnet-y-040/blob/main/config.json',
}




class    _snake_case  (						_lowercase       ):
							SCREAMING_SNAKE_CASE		     =						'''regnet'''
							SCREAMING_SNAKE_CASE		     =						['''x''', '''y''']






							def __init__(					self						:			Union[str, Any]						, snake_case__						:			Any=3						, snake_case__						:			str=3_2						, snake_case__						:			Tuple=[1_2_8, 1_9_2, 5_1_2, 1_0_8_8]						, snake_case__						:			Union[str, Any]=[2, 6, 1_2, 2]						, snake_case__						:			str=6_4						, snake_case__						:			Optional[int]="y"						, snake_case__						:			Tuple="relu"						, **snake_case__						:			str						, ):






												"""simple docstring"""



												super().__init__(**snake_case__    )
												if layer_type not in self.layer_types:
																	raise ValueError(f"""layer_type={layer_type} is not one of {','.join(self.layer_types    )}"""    )
												__a:						Any		    =					num_channels
												__a:						List[Any]		    =					embedding_size
												__a:						Optional[int]		    =					hidden_sizes
												__a:						Optional[int]		    =					depths
												__a:						Optional[Any]		    =					groups_width
												__a:						Tuple		    =					layer_type
												__a:						str		    =					hidden_act
												# always downsample in the first stage
												__a:						List[str]		    =					True
