'''simple docstring'''




# XXX: we want transformers master here - in the absense of conftest manipulating sys.path:
# hack it in for now:
import sys
from pathlib import Path


__lowercase        =    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(4_2)

__lowercase        =    {'''base''': '''patrickvonplaten/wav2vec2_tiny_random''', '''robust''': '''patrickvonplaten/wav2vec2_tiny_random_robust'''}

__lowercase        =    '''zero2'''
__lowercase        =    '''zero3'''
__lowercase        =    [ZEROa, ZEROa]






def 							snake_case__				(     _A:   List[Any] ,  _A:   int ,  _A:   Optional[int]					)       ->      Tuple:

		'''simple docstring'''


		lowerCAmelCase     =    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
__lowercase        =    list(itertools.product(stages, models.keys()))






@slow
@require_deepspeed
@require_torch_gpu
class 							a__(   lowerCAmelCase__      ):


			'''simple docstring'''






			@parameterized.expand(__lowerCAmelCase			,       name_func=__lowerCAmelCase)
			def 				a_			(  self			,       __lowerCAmelCase			,       __lowerCAmelCase):




					"""simple docstring"""


					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			,       __lowerCAmelCase			,       __lowerCAmelCase):




					"""simple docstring"""


					self.run_and_check(
					    stage=__lowerCAmelCase			,       model=__lowerCAmelCase			,       distributed=__lowerCAmelCase			,       fpaa=__lowerCAmelCase			,       )



			@parameterized.expand(__lowerCAmelCase			,       name_func=__lowerCAmelCase)
			def 				a_			(  self			,       __lowerCAmelCase			,       __lowerCAmelCase):




					"""simple docstring"""


					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			,       __lowerCAmelCase			,       __lowerCAmelCase):




					"""simple docstring"""


					self.run_and_check(
					    stage=__lowerCAmelCase			,       model=__lowerCAmelCase			,       distributed=__lowerCAmelCase			,       fpaa=__lowerCAmelCase			,       )



			def 				a_			(  self			,       __lowerCAmelCase):




					"""simple docstring"""


					pass



			def 				a_			(  self			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase = 10			,       __lowerCAmelCase = True			,       __lowerCAmelCase = True			,       __lowerCAmelCase = True			,       ):




					"""simple docstring"""


					lowerCAmelCase     =    models[model]

					lowerCAmelCase     =    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			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase = 10			,       __lowerCAmelCase = 1			,       __lowerCAmelCase = True			,       __lowerCAmelCase = True			,       ):




					"""simple docstring"""


					lowerCAmelCase     =    self.get_auto_remove_tmp_dir("""./xxx"""			,       after=__lowerCAmelCase)
					lowerCAmelCase     =    f"\n            --model_name_or_path {model_name}\n            --dataset_name hf-internal-testing/librispeech_asr_dummy\n            --dataset_config_name clean\n            --train_split_name validation\n            --validation_split_name validation\n            --output_dir {output_dir}\n            --num_train_epochs {str(__lowerCAmelCase)}\n            --per_device_train_batch_size 2\n            --per_device_eval_batch_size 2\n            --evaluation_strategy steps\n            --learning_rate 5e-4\n            --warmup_steps 8\n            --orthography timit\n            --preprocessing_num_workers 1\n            --group_by_length\n            --freeze_feature_extractor\n            --report_to none\n            --save_steps 0\n            --eval_steps {eval_steps}\n            --report_to none\n        ".split()

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

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

					lowerCAmelCase     =    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			,       __lowerCAmelCase=False):




					"""simple docstring"""


					lowerCAmelCase     =    min(2			,       get_gpu_count()) if distributed else 1
					return f"deepspeed --num_nodes 1 --num_gpus {num_gpus}".split()





'''simple docstring'''




import os
from distutils.util import strtobool






def 							snake_case__				(     _A:   int ,  _A:   List[Any]					)       ->      Union[str, Any]:

		'''simple docstring'''


		for e in env_keys:
				lowerCAmelCase     =    int(os.environ.get(_A ,  -1					)					)
				if val >= 0:
						return val
		return default






def 							snake_case__				(     _A:   str ,  _A:   List[Any]=False					)       ->      int:

		'''simple docstring'''


		lowerCAmelCase     =    os.environ.get(_A ,  str(_A					)					)
		return strtobool(_A					) == 1  # As its name indicates `strtobool` actually returns an int...






def 							snake_case__				(     _A:   Dict ,  _A:   List[Any]="no"					)       ->      str:

		'''simple docstring'''


		lowerCAmelCase     =    os.environ.get(_A ,  str(_A					)					)
		return value




