



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


UpperCAmelCase:   Union[str, Any]       			=							namedtuple("""ExampleNamedTuple""", """a b c""")





class    UpperCamelCase     (			unittest.TestCase    ):




	"""simple docstring"""






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

			_lowercase							:      int			=					send_to_device(UpperCAmelCase_							,UpperCAmelCase_					)
			self.assertTrue(torch.equal(resulta.cpu()							,UpperCAmelCase_					)					)

			_lowercase							:      str			=					send_to_device((tensor, [tensor, tensor], 1)							,UpperCAmelCase_					)
			self.assertIsInstance(UpperCAmelCase_							,UpperCAmelCase_					)
			self.assertTrue(torch.equal(resulta[0].cpu()							,UpperCAmelCase_					)					)
			self.assertIsInstance(resulta[1]							,UpperCAmelCase_					)
			self.assertTrue(torch.equal(resulta[1][0].cpu()							,UpperCAmelCase_					)					)
			self.assertTrue(torch.equal(resulta[1][1].cpu()							,UpperCAmelCase_					)					)
			self.assertEqual(resulta[2]							,1					)

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

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

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






	def 					lowerCamelCase__ (			self					):
			with self.assertRaises(UpperCAmelCase_					) as cm:
					_lowercase							:      List[str]			=					recursively_apply(torch.tensor							,(torch.tensor(1					), 1)							,error_on_other_type=UpperCAmelCase_					)
			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 					lowerCamelCase__ (			self					):
			_lowercase							:      Optional[Any]			=					torch.tensor([1, 2, 3, 4, 5]					)
			self.assertEqual(listify(UpperCAmelCase_					)							,[1, 2, 3, 4, 5]					)

			_lowercase							:      Any			=					torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)
			self.assertEqual(listify(UpperCAmelCase_					)							,[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]					)

			_lowercase							:      Any			=					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(UpperCAmelCase_					)							,[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]					)






	def 					lowerCamelCase__ (			self					):
			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 					lowerCamelCase__ (			self					):
			_lowercase							:      Union[str, Any]			=					RegressionModel()
			_lowercase							:      Optional[Any]			=					model.forward
			_lowercase							:      Tuple			=					convert_outputs_to_fpaa(model.forward					)
			_lowercase							:      str			=					extract_model_from_parallel(UpperCAmelCase_							,keep_fpaa_wrapper=UpperCAmelCase_					)
			_lowercase							:      List[str]			=					pickle.dumps(UpperCAmelCase_					)






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






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






	def 					lowerCamelCase__ (			self					):
			_lowercase							:      Any			=					RegressionModel()
			# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
			_lowercase							:      int			=					torch.nn.parallel.DataParallel(UpperCAmelCase_					)
			_lowercase							:      List[str]			=					extract_model_from_parallel(UpperCAmelCase_					)

			self.assertEqual(UpperCAmelCase_							,UpperCAmelCase_					)






	@require_torch_min_version(version="""2.0"""					)
	def 					lowerCamelCase__ (			self					):
			_lowercase							:      Any			=					RegressionModel()
			_lowercase							:      str			=					torch.compile(UpperCAmelCase_					)

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

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






	def 					lowerCamelCase__ (			self					):
			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 typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCAmelCase__							= {
    'configuration_deberta_v2': ['DEBERTA_V2_PRETRAINED_CONFIG_ARCHIVE_MAP', 'DebertaV2Config', 'DebertaV2OnnxConfig'],
    'tokenization_deberta_v2': ['DebertaV2Tokenizer'],
}

try:
    if not is_tokenizers_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCAmelCase__							= ['DebertaV2TokenizerFast']

try:
    if not is_tf_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCAmelCase__							= [
        'TF_DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST',
        'TFDebertaV2ForMaskedLM',
        'TFDebertaV2ForQuestionAnswering',
        'TFDebertaV2ForSequenceClassification',
        'TFDebertaV2ForTokenClassification',
        'TFDebertaV2Model',
        'TFDebertaV2PreTrainedModel',
    ]

try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCAmelCase__							= [
        'DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST',
        'DebertaV2ForMaskedLM',
        'DebertaV2ForMultipleChoice',
        'DebertaV2ForQuestionAnswering',
        'DebertaV2ForSequenceClassification',
        'DebertaV2ForTokenClassification',
        'DebertaV2Model',
        'DebertaV2PreTrainedModel',
    ]


if TYPE_CHECKING:
    from .configuration_deberta_va import (
        DEBERTA_V2_PRETRAINED_CONFIG_ARCHIVE_MAP,
        DebertaVaConfig,
        DebertaVaOnnxConfig,
    )
    from .tokenization_deberta_va import DebertaVaTokenizer

    try:
        if not is_tokenizers_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .tokenization_deberta_va_fast import DebertaVaTokenizerFast

    try:
        if not is_tf_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_tf_deberta_va import (
            TF_DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST,
            TFDebertaVaForMaskedLM,
            TFDebertaVaForQuestionAnswering,
            TFDebertaVaForSequenceClassification,
            TFDebertaVaForTokenClassification,
            TFDebertaVaModel,
            TFDebertaVaPreTrainedModel,
        )

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_deberta_va import (
            DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST,
            DebertaVaForMaskedLM,
            DebertaVaForMultipleChoice,
            DebertaVaForQuestionAnswering,
            DebertaVaForSequenceClassification,
            DebertaVaForTokenClassification,
            DebertaVaModel,
            DebertaVaPreTrainedModel,
        )

else:
    import sys

    lowerCAmelCase__							= _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
