



"""simple docstring"""






import ast
import os
import re
import shutil
import tempfile
import unittest
from unittest import mock

import torch

from accelerate.test_utils.examples import compare_against_test
from accelerate.test_utils.testing import TempDirTestCase, require_trackers, run_command, slow
from accelerate.utils import write_basic_config


# DataLoaders built from `test_samples/MRPC` for quick testing
# Should mock `{script_name}.get_dataloaders` via:
# @mock.patch("{script_name}.get_dataloaders", mocked_dataloaders)

lowerCAmelCase__        =				[
    '''cross_validation.py''',
    '''gradient_accumulation.py''',
    '''local_sgd.py''',
    '''multi_process_metrics.py''',
    '''memory.py''',
    '''automatic_gradient_accumulation.py''',
    '''fsdp_with_peak_mem_tracking.py''',
    '''deepspeed_with_config_support.py''',
    '''megatron_lm_gpt_pretraining.py''',
]






class        __snake_case       (	unittest.TestCase):




  def 						SCREAMING_SNAKE_CASE				( self     :						List[str]						,					__lowerCAmelCase     :						str						,					__lowerCAmelCase     :						bool						,					__lowerCAmelCase     :						str = None						,					__lowerCAmelCase     :						list = None		):






         """simple docstring"""






         _lowerCamelCase						:				str    			= None
         _lowerCamelCase						:				Tuple    			= os.path.abspath(os.path.join('''examples'''						,					'''by_feature'''		)		)
         _lowerCamelCase						:				List[Any]    			= os.path.abspath('''examples'''		)
         for item in os.listdir(__lowerCAmelCase		):
                if item not in EXCLUDE_EXAMPLES:
                       _lowerCamelCase						:				Optional[int]    			= os.path.join(__lowerCAmelCase						,					__lowerCAmelCase		)
                       if os.path.isfile(__lowerCAmelCase		) and ".py" in item_path:
                              with self.subTest(
                                  tested_script=__lowerCAmelCase						,					feature_script=__lowerCAmelCase						,					tested_section='''main()''' if parser_only else '''training_function()'''						,					):
                                     _lowerCamelCase						:				Tuple    			= compare_against_test(
                                         os.path.join(__lowerCAmelCase						,					__lowerCAmelCase		)						,					__lowerCAmelCase						,					__lowerCAmelCase						,					__lowerCAmelCase		)
                                     _lowerCamelCase						:				Union[str, Any]    			= '''\n'''.join(__lowerCAmelCase		)
                                     if special_strings is not None:
                                            for string in special_strings:
                                                   _lowerCamelCase						:				Dict    			= diff.replace(__lowerCAmelCase						,					''''''		)
                                     self.assertEqual(__lowerCAmelCase						,					''''''		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[int]		):






         """simple docstring"""






         self.one_complete_example('''complete_nlp_example.py'''						,					__lowerCAmelCase		)
         self.one_complete_example('''complete_nlp_example.py'''						,					__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Any		):






         """simple docstring"""






         _lowerCamelCase						:				Optional[Any]    			= os.path.abspath(os.path.join('''examples'''						,					'''cv_example.py'''		)		)
         _lowerCamelCase						:				List[Any]    			= [
             ''' ''' * 1_6 + '''{\n\n''',
             ''' ''' * 2_0 + '''"accuracy": eval_metric["accuracy"],\n\n''',
             ''' ''' * 2_0 + '''"f1": eval_metric["f1"],\n\n''',
             ''' ''' * 2_0 + '''"train_loss": total_loss.item() / len(train_dataloader),\n\n''',
             ''' ''' * 2_0 + '''"epoch": epoch,\n\n''',
             ''' ''' * 1_6 + '''},\n\n''',
             ''' ''' * 1_6 + '''step=epoch,\n''',
             ''' ''' * 1_2,
             ''' ''' * 8 + '''for step, batch in enumerate(active_dataloader):\n''',
         ]
         self.one_complete_example('''complete_cv_example.py'''						,					__lowerCAmelCase						,					__lowerCAmelCase						,					__lowerCAmelCase		)
         self.one_complete_example('''complete_cv_example.py'''						,					__lowerCAmelCase						,					__lowerCAmelCase						,					__lowerCAmelCase		)






@mock.patch.dict(os.environ  ,     {"TESTING_MOCKED_DATALOADERS": "1"})
class        __snake_case       (	_lowercase):
  snake_case__				:   Tuple  			=						False




  @classmethod
  def 						SCREAMING_SNAKE_CASE				( cls     :						Optional[Any]		):






         """simple docstring"""






         super().setUpClass()
         _lowerCamelCase						:				Tuple    			= tempfile.mkdtemp()
         _lowerCamelCase						:				List[Any]    			= os.path.join(cls._tmpdir						,					'''default_config.yml'''		)

         write_basic_config(save_location=cls.configPath		)
         _lowerCamelCase						:				int    			= ['''accelerate''', '''launch''', '''--config_file''', cls.configPath]




  @classmethod
  def 						SCREAMING_SNAKE_CASE				( cls     :						List[str]		):






         """simple docstring"""






         super().tearDownClass()
         shutil.rmtree(cls._tmpdir		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[int]		):






         """simple docstring"""






         _lowerCamelCase						:				Dict    			= f'''
        examples/by_feature/checkpointing.py
        --checkpointing_steps epoch
        --output_dir {self.tmpdir}
        '''.split()
         run_command(self._launch_args + testargs		)
         self.assertTrue(os.path.exists(os.path.join(self.tmpdir						,					'''epoch_0'''		)		)		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Dict		):






         """simple docstring"""






         _lowerCamelCase						:				Optional[int]    			= f'''
        examples/by_feature/checkpointing.py
        --checkpointing_steps 1
        --output_dir {self.tmpdir}
        '''.split()
         _lowerCamelCase						:				Any    			= run_command(self._launch_args + testargs		)
         self.assertTrue(os.path.exists(os.path.join(self.tmpdir						,					'''step_2'''		)		)		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Dict		):






         """simple docstring"""






         _lowerCamelCase						:				Union[str, Any]    			= f'''
        examples/by_feature/checkpointing.py
        --resume_from_checkpoint {os.path.join(self.tmpdir						,					"epoch_0"		)}
        '''.split()
         _lowerCamelCase						:				List[str]    			= run_command(self._launch_args + testargs						,					return_stdout=__lowerCAmelCase		)
         self.assertNotIn('''epoch 0:'''						,					__lowerCAmelCase		)
         self.assertIn('''epoch 1:'''						,					__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[Any]		):






         """simple docstring"""






         _lowerCamelCase						:				Any    			= f'''
        examples/by_feature/checkpointing.py
        --resume_from_checkpoint {os.path.join(self.tmpdir						,					"step_2"		)}
        '''.split()
         _lowerCamelCase						:				int    			= run_command(self._launch_args + testargs						,					return_stdout=__lowerCAmelCase		)
         if torch.cuda.is_available():
                _lowerCamelCase						:				str    			= torch.cuda.device_count()
         else:
                _lowerCamelCase						:				List[str]    			= 1
         if num_processes > 1:
                self.assertNotIn('''epoch 0:'''						,					__lowerCAmelCase		)
                self.assertIn('''epoch 1:'''						,					__lowerCAmelCase		)
         else:
                self.assertIn('''epoch 0:'''						,					__lowerCAmelCase		)
                self.assertIn('''epoch 1:'''						,					__lowerCAmelCase		)




  @slow
  def 						SCREAMING_SNAKE_CASE				( self     :						Dict		):






         """simple docstring"""






         _lowerCamelCase						:				List[Any]    			= '''
        examples/by_feature/cross_validation.py
        --num_folds 2
        '''.split()
         with mock.patch.dict(os.environ						,					{'''TESTING_MOCKED_DATALOADERS''': '''0'''}		):
                _lowerCamelCase						:				List[str]    			= run_command(self._launch_args + testargs						,					return_stdout=__lowerCAmelCase		)
                _lowerCamelCase						:				int    			= re.findall('''({.+})'''						,					__lowerCAmelCase		)
                _lowerCamelCase						:				List[str]    			= [r for r in results if '''accuracy''' in r][-1]
                _lowerCamelCase						:				Dict    			= ast.literal_eval(__lowerCAmelCase		)
                self.assertGreaterEqual(results['''accuracy''']						,					0.75		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[int]		):






         """simple docstring"""






         _lowerCamelCase						:				Optional[int]    			= ['''examples/by_feature/multi_process_metrics.py''']
         run_command(self._launch_args + testargs		)




  @require_trackers
  @mock.patch.dict(os.environ						,					{'''WANDB_MODE''': '''offline'''}		)
  def 						SCREAMING_SNAKE_CASE				( self     :						Dict		):






         """simple docstring"""






         with tempfile.TemporaryDirectory() as tmpdir:
                _lowerCamelCase						:				Tuple    			= f'''
            examples/by_feature/tracking.py
            --with_tracking
            --project_dir {tmpdir}
            '''.split()
                run_command(self._launch_args + testargs		)
                self.assertTrue(os.path.exists(os.path.join(__lowerCAmelCase						,					'''tracking'''		)		)		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase						:				Union[str, Any]    			= ['''examples/by_feature/gradient_accumulation.py''']
         run_command(self._launch_args + testargs		)




  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[int]		):






         """simple docstring"""






         _lowerCamelCase						:				int    			= ['''examples/by_feature/local_sgd.py''']
         run_command(self._launch_args + testargs		)










"""simple docstring"""
from torch import nn

def 				snake_case      (   A__						):
			if act_fn in ["swish", "silu"]:
						return nn.SiLU()
			elif act_fn == "mish":
						return nn.Mish()
			elif act_fn == "gelu":
						return nn.GELU()
			else:
						raise ValueError(F"""Unsupported activation function: {act_fn}"""						)



