
'''simple docstring'''
import torch.nn as nn

from .imports import is_fpa_available


if is_fpa_available():
     import transformer_engine.pytorch as te







def 							A_  (    __SCREAMING_SNAKE_CASE    :	List[str]      ,  __SCREAMING_SNAKE_CASE    :	Any=True      ,  __SCREAMING_SNAKE_CASE    :	Optional[int]=True      ,  __SCREAMING_SNAKE_CASE    :	List[str]=True			)					->							List[str]:





  """simple docstring"""

  if not is_fpa_available():
    raise ImportError("""Using `convert_model` requires transformer_engine to be installed."""			)
  for name, module in model.named_children():
    if isinstance(_A      ,  nn.Linear			) and to_transformer_engine and _convert_linear:
      # Return early if the linear layer weights are not multiples of 16
      if any(p % 16 != 0 for p in module.weight.shape			):
        return
      __A		:  Dict       				=		module.bias is not None
      __A		:  Tuple       				=		te.Linear(
          module.in_features      ,  module.out_features      ,  bias=_A      ,  params_dtype=module.weight.dtype			)
      __A		:  int       				=		module.weight.data.clone()
      if has_bias:
        __A		:  int       				=		module.bias.data.clone()

      setattr(_A      ,  _A      ,  _A			)
    elif isinstance(_A      ,  nn.LayerNorm			) and to_transformer_engine and _convert_ln:
      __A		:  List[str]       				=		te.LayerNorm(module.normalized_shape[0]      ,  eps=module.eps      ,  params_dtype=module.weight.dtype			)
      __A		:  Any       				=		module.weight.data.clone()
      __A		:  Dict       				=		module.bias.data.clone()

      setattr(_A      ,  _A      ,  _A			)
    elif isinstance(_A      ,  te.Linear			) and not to_transformer_engine and _convert_linear:
      __A		:  List[str]       				=		module.bias is not None
      __A		:  Tuple       				=		nn.Linear(
          module.in_features      ,  module.out_features      ,  bias=_A      ,  params_dtype=module.weight.dtype			)
      __A		:  Optional[Any]       				=		module.weight.data.clone()
      if has_bias:
        __A		:  str       				=		module.bias.data.clone()

      setattr(_A      ,  _A      ,  _A			)
    elif isinstance(_A      ,  te.LayerNorm			) and not to_transformer_engine and _convert_ln:
      __A		:  str       				=		nn.LayerNorm(module.normalized_shape[0]      ,  eps=module.eps      ,  params_dtype=module.weight.dtype			)
      __A		:  Tuple       				=		module.weight.data.clone()
      __A		:  str       				=		module.bias.data.clone()

      setattr(_A      ,  _A      ,  _A			)
    else:
      convert_model(
          _A      ,  to_transformer_engine=_A      ,  _convert_linear=_convert_linear      ,  _convert_ln=_convert_ln      ,  )







def 							A_  (    __SCREAMING_SNAKE_CASE    :	Union[str, Any]			)					->							List[Any]:





  """simple docstring"""

  if not is_fpa_available():
    raise ImportError("""Using `has_transformer_engine_layers` requires transformer_engine to be installed."""			)
  for m in model.modules():
    if isinstance(_A      ,  (te.LayerNorm, te.Linear)			):
      return True
  return False











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


lowerCamelCase_					:			List[Any]     						=				namedtuple('''ExampleNamedTuple''', '''a b c''')
class  _SCREAMING_SNAKE_CASE				(		unittest.TestCase						):


			'''simple docstring'''





			def 					A   ( self				:    List[Any]      )				->     str:

									'''simple docstring'''







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

									UpperCamelCase__       =			send_to_device(lowercase			,   lowercase      )
									self.assertTrue(torch.equal(resulta.cpu()			,   lowercase      )      )

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

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

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

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






			def 					A   ( self				:    List[Any]      )				->     List[str]:

									'''simple docstring'''







									with self.assertRaises(lowercase      ) as cm:
															UpperCamelCase__       =			recursively_apply(torch.tensor			,   (torch.tensor(1      ), 1)			,   error_on_other_type=lowercase      )
									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				:    Tuple      )				->     Tuple:

									'''simple docstring'''







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

									UpperCamelCase__       =			torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )
									self.assertEqual(listify(lowercase      )			,   [[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]      )

									UpperCamelCase__       =			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(lowercase      )			,   [[[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 					A   ( self				:    Union[str, Any]      )				->     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 					A   ( self				:    Optional[int]      )				->     int:

									'''simple docstring'''







									UpperCamelCase__       =			RegressionModel()
									UpperCamelCase__       =			model.forward
									UpperCamelCase__       =			convert_outputs_to_fpaa(model.forward      )
									UpperCamelCase__       =			extract_model_from_parallel(lowercase			,   keep_fpaa_wrapper=lowercase      )
									UpperCamelCase__       =			pickle.dumps(lowercase      )






			@require_cuda
			def 					A   ( self				:    int      )				->     Optional[int]:

									'''simple docstring'''







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






			@require_cuda
			@require_torch_min_version(version="""2.0"""      )
			def 					A   ( self				:    Any      )				->     int:

									'''simple docstring'''







									UpperCamelCase__       =			RegressionModel()
									UpperCamelCase__       =			model.forward
									UpperCamelCase__       =			torch.cuda.amp.autocast(dtype=torch.floataa      )(model.forward      )
									UpperCamelCase__       =			convert_outputs_to_fpaa(model.forward      )
									UpperCamelCase__       =			torch.compile(model.forward			,   backend="""inductor"""      )
									UpperCamelCase__       =			torch.randn(4			,   1_0      ).cuda()
									UpperCamelCase__       =			model(lowercase      )






			def 					A   ( self				:    Union[str, Any]      )				->     Union[str, Any]:

									'''simple docstring'''







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

									self.assertEqual(lowercase			,   lowercase      )






			@require_torch_min_version(version="""2.0"""      )
			def 					A   ( self				:    Tuple      )				->     Tuple:

									'''simple docstring'''







									UpperCamelCase__       =			RegressionModel()
									UpperCamelCase__       =			torch.compile(lowercase      )

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

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







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