




"""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							=			namedtuple("""ExampleNamedTuple""", """a b c""")

class 					__lowercase				(	unittest.TestCase		):


				'''simple docstring'''


				def 					_lowerCamelCase    (      self      :   str					):
										snake_case__				:    Optional[Any]								=    torch.randn(5					,2					)
										snake_case__				:    Optional[Any]								=    torch.device('''cuda'''					) if torch.cuda.is_available() else torch.device('''cpu'''					)

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

										snake_case__				:    List[Any]								=    send_to_device((tensor, [tensor, tensor], 1)					,SCREAMING_SNAKE_CASE_					)
										self.assertIsInstance(SCREAMING_SNAKE_CASE_					,SCREAMING_SNAKE_CASE_					)
										self.assertTrue(torch.equal(resulta[0].cpu()					,SCREAMING_SNAKE_CASE_					)					)
										self.assertIsInstance(resulta[1]					,SCREAMING_SNAKE_CASE_					)
										self.assertTrue(torch.equal(resulta[1][0].cpu()					,SCREAMING_SNAKE_CASE_					)					)
										self.assertTrue(torch.equal(resulta[1][1].cpu()					,SCREAMING_SNAKE_CASE_					)					)
										self.assertEqual(resulta[2]					,1					)

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

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

										snake_case__				:    List[str]								=    send_to_device(UserDict({'''a''': tensor, '''b''': [tensor, tensor], '''c''': 1}					)					,SCREAMING_SNAKE_CASE_					)
										self.assertIsInstance(SCREAMING_SNAKE_CASE_					,SCREAMING_SNAKE_CASE_					)
										self.assertTrue(torch.equal(resulta['''a'''].cpu()					,SCREAMING_SNAKE_CASE_					)					)
										self.assertIsInstance(resulta['''b''']					,SCREAMING_SNAKE_CASE_					)
										self.assertTrue(torch.equal(resulta['''b'''][0].cpu()					,SCREAMING_SNAKE_CASE_					)					)
										self.assertTrue(torch.equal(resulta['''b'''][1].cpu()					,SCREAMING_SNAKE_CASE_					)					)
										self.assertEqual(resulta['''c''']					,1					)

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

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

										snake_case__				:    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(SCREAMING_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 					_lowerCamelCase    (      self      :   str					):
										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      :   Dict					):
										snake_case__				:    Optional[int]								=    RegressionModel()
										snake_case__				:    int								=    model.forward
										snake_case__				:    Union[str, Any]								=    convert_outputs_to_fpaa(model.forward					)
										snake_case__				:    int								=    extract_model_from_parallel(SCREAMING_SNAKE_CASE_					,keep_fpaa_wrapper=SCREAMING_SNAKE_CASE_					)
										snake_case__				:    List[str]								=    pickle.dumps(SCREAMING_SNAKE_CASE_					)

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

				@require_cuda
				@require_torch_min_version(version='''2.0'''					)
				def 					_lowerCamelCase    (      self      :   Tuple					):
										snake_case__				:    Tuple								=    RegressionModel()
										snake_case__				:    Tuple								=    model.forward
										snake_case__				:    List[Any]								=    torch.cuda.amp.autocast(dtype=torch.floataa					)(model.forward					)
										snake_case__				:    str								=    convert_outputs_to_fpaa(model.forward					)
										snake_case__				:    Optional[int]								=    torch.compile(model.forward					,backend='''inductor'''					)
										snake_case__				:    Optional[int]								=    torch.randn(4					,1_0					).cuda()
										snake_case__				:    int								=    model(SCREAMING_SNAKE_CASE_					)

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

										self.assertEqual(SCREAMING_SNAKE_CASE_					,SCREAMING_SNAKE_CASE_					)

				@require_torch_min_version(version='''2.0'''					)
				def 					_lowerCamelCase    (      self      :   Dict					):
										snake_case__				:    Optional[int]								=    RegressionModel()
										snake_case__				:    Optional[int]								=    torch.compile(SCREAMING_SNAKE_CASE_					)

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

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





				def 					_lowerCamelCase    (      self      :   List[str]					):
										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''']					)					)



# as due to their complexity multi-gpu tests could impact other tests, and to aid debug we have those in a separate module.

import os
import sys

from transformers.testing_utils import TestCasePlus, execute_subprocess_async, get_gpu_count, require_torch_gpu, slow

from .utils import load_json


class 						a__							(   __snake_case		):



			@classmethod
			def __SCREAMING_SNAKE_CASE							(						cls    ) ->    Optional[int]:
								return cls







			@slow
			@require_torch_gpu
			def __SCREAMING_SNAKE_CASE							(						self    ) ->    Dict:
								__a							=   self.get_auto_remove_tmp_dir()
								__a							=   f'''
            --model_name Helsinki-NLP/opus-mt-en-ro
            --save_dir {output_dir}
            --data_dir {self.test_file_dir_str}/test_data/wmt_en_ro
            --num_beams 2
            --task translation
        '''.split()

								# we want this test to run even if there is only one GPU, but if there are more we use them all
								__a							=   get_gpu_count()
								__a							=   f'''
            -m torch.distributed.launch
            --nproc_per_node={n_gpu}
            {self.test_file_dir}/run_distributed_eval.py
        '''.split()
								__a							=   [sys.executable] + distributed_args + args
								execute_subprocess_async(UpperCAmelCase      ,						env=self.get_env()    )

								__a							=   os.path.join(UpperCAmelCase      ,						'test_bleu.json'    )
								__a							=   load_json(UpperCAmelCase    )
								# print(metrics)
								self.assertGreaterEqual(metrics['bleu']      ,						2_5    )




