






import sys
from pathlib import Path


__A					=	Path(__file__).resolve().parents[3] / "src"
sys.path.insert(1, str(git_repo_path))

import dataclasses  # noqa
import io  # noqa
import itertools  # noqa
import json  # noqa
import os  # noqa
import unittest  # noqa
from copy import deepcopy  # noqa

from parameterized import parameterized  # noqa
from transformers import TrainingArguments, is_torch_available  # noqa
from transformers.deepspeed import is_deepspeed_available  # noqa
from transformers.file_utils import WEIGHTS_NAME  # noqa
from transformers.testing_utils import (  # noqa
    CaptureLogger,
    ExtendSysPath,
    TestCasePlus,
    execute_subprocess_async,
    get_gpu_count,
    mockenv_context,
    require_deepspeed,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)
from transformers.trainer_utils import set_seed  # noqa


set_seed(42)

__A					=	{"base": "patrickvonplaten/wav2vec2_tiny_random", "robust": "patrickvonplaten/wav2vec2_tiny_random_robust"}

__A					=	"zero2"
__A					=	"zero3"
__A					=	[ZEROa, ZEROa]







def       lowercase__	(					A_:						Optional[int]  ,      A_:						Union[str, Any]  ,      A_:						Optional[Any]		)	->				List[str]:







       """simple docstring"""






       __UpperCAmelCase     				=parameterized.to_safe_name("""_""".join(str(A_		) for x in param.args		)		)
       return F'''{func.__name__}_{param_based_name}'''


# Cartesian-product of zero stages with models to test
__A					=	list(itertools.product(stages, models.keys()))





@slow
@require_deepspeed
@require_torch_gpu
class 				_A			(     a__				):
    """simple docstring"""

    @parameterized.expand(lowerCamelCase_ ,     name_func=lowerCamelCase_      )
    def _a							(				self      :     Any ,     __SCREAMING_SNAKE_CASE      :     List[str] ,     __SCREAMING_SNAKE_CASE      :     Any      )      ->							Any:
           self.run_and_check(
               stage=lowerCamelCase_ ,     model=lowerCamelCase_ ,     distributed=lowerCamelCase_ ,     fpaa=lowerCamelCase_ ,     )



    @require_torch_multi_gpu
    @parameterized.expand(lowerCamelCase_ ,     name_func=lowerCamelCase_      )
    def _a							(				self      :     int ,     __SCREAMING_SNAKE_CASE      :     Tuple ,     __SCREAMING_SNAKE_CASE      :     Any      )      ->							int:
           self.run_and_check(
               stage=lowerCamelCase_ ,     model=lowerCamelCase_ ,     distributed=lowerCamelCase_ ,     fpaa=lowerCamelCase_ ,     )



    @parameterized.expand(lowerCamelCase_ ,     name_func=lowerCamelCase_      )
    def _a							(				self      :     List[Any] ,     __SCREAMING_SNAKE_CASE      :     Any ,     __SCREAMING_SNAKE_CASE      :     List[Any]      )      ->							Tuple:
           self.run_and_check(
               stage=lowerCamelCase_ ,     model=lowerCamelCase_ ,     distributed=lowerCamelCase_ ,     fpaa=lowerCamelCase_ ,     )



    @require_torch_multi_gpu
    @parameterized.expand(lowerCamelCase_ ,     name_func=lowerCamelCase_      )
    def _a							(				self      :     Dict ,     __SCREAMING_SNAKE_CASE      :     str ,     __SCREAMING_SNAKE_CASE      :     Optional[int]      )      ->							Optional[int]:
           self.run_and_check(
               stage=lowerCamelCase_ ,     model=lowerCamelCase_ ,     distributed=lowerCamelCase_ ,     fpaa=lowerCamelCase_ ,     )



    def _a							(				self      :     Optional[int] ,     __SCREAMING_SNAKE_CASE      :     Any      )      ->							Union[str, Any]:
           # XXX: run_asr is premature and doesn't save any results
           # so all we check for now is that the process didn't fail
           pass



    def _a							(				self      :     Tuple ,     __SCREAMING_SNAKE_CASE      :     Dict ,     __SCREAMING_SNAKE_CASE      :     Union[str, Any] ,     __SCREAMING_SNAKE_CASE      :     Tuple = 10 ,     __SCREAMING_SNAKE_CASE      :     Tuple = True ,     __SCREAMING_SNAKE_CASE      :     Union[str, Any] = True ,     __SCREAMING_SNAKE_CASE      :     int = True ,     )      ->							Optional[int]:
           __UpperCAmelCase     				=models[model]

           __UpperCAmelCase     				=self.run_trainer(
               stage=lowerCamelCase_ ,     model_name=lowerCamelCase_ ,     eval_steps=lowerCamelCase_ ,     num_train_epochs=1 ,     distributed=lowerCamelCase_ ,     fpaa=lowerCamelCase_ ,     )

           self.do_checks(lowerCamelCase_      )

           return output_dir



    def _a							(				self      :     List[str] ,     __SCREAMING_SNAKE_CASE      :     Union[str, Any] ,     __SCREAMING_SNAKE_CASE      :     Any ,     __SCREAMING_SNAKE_CASE      :     Union[str, Any] = 10 ,     __SCREAMING_SNAKE_CASE      :     Any = 1 ,     __SCREAMING_SNAKE_CASE      :     Any = True ,     __SCREAMING_SNAKE_CASE      :     Optional[int] = True ,     )      ->							Optional[int]:
           __UpperCAmelCase     				=self.get_auto_remove_tmp_dir("""./xxx""" ,     after=lowerCamelCase_      )
           __UpperCAmelCase     				=f'''
            --model_name_or_path {model_name}
            --dataset_name hf-internal-testing/librispeech_asr_dummy
            --dataset_config_name clean
            --train_split_name validation
            --validation_split_name validation
            --output_dir {output_dir}
            --num_train_epochs {str(lowerCamelCase_      )}
            --per_device_train_batch_size 2
            --per_device_eval_batch_size 2
            --evaluation_strategy steps
            --learning_rate 5e-4
            --warmup_steps 8
            --orthography timit
            --preprocessing_num_workers 1
            --group_by_length
            --freeze_feature_extractor
            --report_to none
            --save_steps 0
            --eval_steps {eval_steps}
            --report_to none
        '''.split()

           if fpaa:
                  args.extend(["""--fp16"""]      )

           # currently ds_config_wav2vec2_zero.json requires "zero_optimization.find_unused_parameters": true,
           # hence the separate config files
           __UpperCAmelCase     				=f'''--deepspeed {self.test_file_dir_str}/ds_config_wav2vec2_{stage}.json'''.split()
           __UpperCAmelCase     				=[f'''{self.examples_dir_str}/research_projects/wav2vec2/run_asr.py''']
           __UpperCAmelCase     				=self.get_launcher(lowerCamelCase_      )

           __UpperCAmelCase     				=launcher + script + args + ds_args
           # keep for quick debug
           # print(" ".join([f"\nPYTHONPATH={self.src_dir_str}"] +cmd)); die
           execute_subprocess_async(lowerCamelCase_ ,     env=self.get_env()      )

           return output_dir



    def _a							(				self      :     Optional[Any] ,     __SCREAMING_SNAKE_CASE      :     Optional[Any]=False      )      ->							Any:
           # 1. explicitly set --num_nodes=1 just in case these tests end up run on a multi-node setup
           # - it won't be able to handle that
           # 2. for now testing with just 2 gpus max (since some quality tests may give different
           # results with mode gpus because we use very little data)
           __UpperCAmelCase     				=min(2 ,     get_gpu_count()      ) if distributed else 1
           return f'''deepspeed --num_nodes 1 --num_gpus {num_gpus}'''.split()








'''simple docstring'''






import unittest

import torch
from torch import nn

from accelerate.test_utils import require_cuda
from accelerate.utils.memory import find_executable_batch_size, release_memory


def    _snake_case  (	)		->       Union[str, Any]:
							raise RuntimeError('''CUDA out of memory.'''			)




class 						a__							(				nn.Module    ):





							'''simple docstring'''






							def __init__(						self  )			->				int:
														super().__init__()
														lowerCAmelCase__			  =      nn.Linear(3   ,       4  )
														lowerCAmelCase__			  =      nn.BatchNormad(4  )
														lowerCAmelCase__			  =      nn.Linear(4   ,       5  )







							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_  )			->				Optional[Any]:
														return self.lineara(self.batchnorm(self.lineara(lowerCamelCase_  )  )  )




class 						a__							(				unittest.TestCase    ):





							'''simple docstring'''






							def     __SCREAMING_SNAKE_CASE				(						self  )			->				str:
														lowerCAmelCase__			  =      []

														@find_executable_batch_size(starting_batch_size=1_28  )
														def mock_training_loop_function(lowerCamelCase_  ):
																					nonlocal batch_sizes
																					batch_sizes.append(lowerCamelCase_  )
																					if batch_size != 8:
																												raise_fake_out_of_memory()

														mock_training_loop_function()
														self.assertListEqual(lowerCamelCase_   ,       [1_28, 64, 32, 16, 8]  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[int]:
														lowerCAmelCase__			  =      []

														@find_executable_batch_size(starting_batch_size=1_28  )
														def mock_training_loop_function(lowerCamelCase_   ,       lowerCamelCase_  ):
																					nonlocal batch_sizes
																					batch_sizes.append(lowerCamelCase_  )
																					if batch_size != 8:
																												raise_fake_out_of_memory()
																					return batch_size, arga

														lowerCAmelCase__  ,   lowerCAmelCase__			  =      mock_training_loop_function('''hello'''  )
														self.assertListEqual(lowerCamelCase_   ,       [1_28, 64, 32, 16, 8]  )
														self.assertListEqual([bs, arga]   ,       [8, '''hello''']  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Tuple:
														@find_executable_batch_size(starting_batch_size=0  )
														def mock_training_loop_function(lowerCamelCase_  ):
																					pass

														with self.assertRaises(lowerCamelCase_  ) as cm:
																					mock_training_loop_function()
																					self.assertIn('''No executable batch size found, reached zero.'''   ,       cm.exception.args[0]  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Dict:
														@find_executable_batch_size(starting_batch_size=16  )
														def mock_training_loop_function(lowerCamelCase_  ):
																					if batch_size > 0:
																												raise_fake_out_of_memory()
																					pass

														with self.assertRaises(lowerCamelCase_  ) as cm:
																					mock_training_loop_function()
																					self.assertIn('''No executable batch size found, reached zero.'''   ,       cm.exception.args[0]  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[int]:
														@find_executable_batch_size(starting_batch_size=1_28  )
														def mock_training_loop_function(lowerCamelCase_   ,       lowerCamelCase_   ,       lowerCamelCase_  ):
																					if batch_size != 8:
																												raise raise_fake_out_of_memory()

														with self.assertRaises(lowerCamelCase_  ) as cm:
																					mock_training_loop_function(1_28   ,       '''hello'''   ,       '''world'''  )
																					self.assertIn('''Batch size was passed into `f`'''   ,       cm.exception.args[0]  )
																					self.assertIn('''`f(arg1=\'hello\', arg2=\'world\')'''   ,       cm.exception.args[0]  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Union[str, Any]:
														@find_executable_batch_size(starting_batch_size=16  )
														def mock_training_loop_function(lowerCamelCase_  ):
																					raise ValueError('''Oops, we had an error!'''  )

														with self.assertRaises(lowerCamelCase_  ) as cm:
																					mock_training_loop_function()
																					self.assertIn('''Oops, we had an error!'''   ,       cm.exception.args[0]  )







							@require_cuda
							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[Any]:
														lowerCAmelCase__			  =      torch.cuda.memory_allocated()
														lowerCAmelCase__			  =      ModelForTest()
														model.cuda()
														self.assertGreater(torch.cuda.memory_allocated()   ,       lowerCamelCase_  )
														lowerCAmelCase__			  =      release_memory(lowerCamelCase_  )
														self.assertEqual(torch.cuda.memory_allocated()   ,       lowerCamelCase_  )