





'''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


A__				:   Dict      =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__				:   Optional[int]      ={'''base''': '''patrickvonplaten/wav2vec2_tiny_random''', '''robust''': '''patrickvonplaten/wav2vec2_tiny_random_robust'''}

A__				:   List[str]      ='''zero2'''
A__				:   List[Any]      ='''zero3'''
A__				:   str      =[ZEROa, ZEROa]
def    UpperCamelCase__      (		lowerCAmelCase							,	lowerCAmelCase							,	lowerCAmelCase  ):





      """simple docstring"""


      _lowerCAmelCase							     =							parameterized.to_safe_name("""_""".join(str(lowerCAmelCase  ) for x in param.args  )  )
      return f"{func.__name__}_{param_based_name}"


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





@slow
@require_deepspeed
@require_torch_gpu
class  UpperCAmelCase     (			snake_case_   ):




      @parameterized.expand(__snake_case			,			name_func=__snake_case  )
      def 				lowercase__    (			self			: List[str]			,			__snake_case			: Tuple			,			__snake_case			: Optional[int]  )				->      List[str]:
            self.run_and_check(
                stage=__snake_case			,			model=__snake_case			,			distributed=__snake_case			,			fpaa=__snake_case			,			)




      @require_torch_multi_gpu
      @parameterized.expand(__snake_case			,			name_func=__snake_case  )
      def 				lowercase__    (			self			: List[Any]			,			__snake_case			: Union[str, Any]			,			__snake_case			: int  )				->      int:
            self.run_and_check(
                stage=__snake_case			,			model=__snake_case			,			distributed=__snake_case			,			fpaa=__snake_case			,			)




      @parameterized.expand(__snake_case			,			name_func=__snake_case  )
      def 				lowercase__    (			self			: Any			,			__snake_case			: Optional[int]			,			__snake_case			: Dict  )				->      Tuple:
            self.run_and_check(
                stage=__snake_case			,			model=__snake_case			,			distributed=__snake_case			,			fpaa=__snake_case			,			)




      @require_torch_multi_gpu
      @parameterized.expand(__snake_case			,			name_func=__snake_case  )
      def 				lowercase__    (			self			: Tuple			,			__snake_case			: Optional[int]			,			__snake_case			: Dict  )				->      Union[str, Any]:
            self.run_and_check(
                stage=__snake_case			,			model=__snake_case			,			distributed=__snake_case			,			fpaa=__snake_case			,			)




      def 				lowercase__    (			self			: Optional[Any]			,			__snake_case			: Union[str, 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 				lowercase__    (			self			: Dict			,			__snake_case			: str			,			__snake_case			: str			,			__snake_case			: int = 10			,			__snake_case			: bool = True			,			__snake_case			: bool = True			,			__snake_case			: bool = True			,			)				->      Optional[int]:
            _lowerCAmelCase							     =							models[model]

            _lowerCAmelCase							     =							self.run_trainer(
                stage=__snake_case			,			model_name=__snake_case			,			eval_steps=__snake_case			,			num_train_epochs=1			,			distributed=__snake_case			,			fpaa=__snake_case			,			)

            self.do_checks(__snake_case  )

            return output_dir




      def 				lowercase__    (			self			: Optional[int]			,			__snake_case			: str			,			__snake_case			: str			,			__snake_case			: int = 10			,			__snake_case			: int = 1			,			__snake_case			: bool = True			,			__snake_case			: bool = True			,			)				->      Optional[Any]:
            _lowerCAmelCase							     =							self.get_auto_remove_tmp_dir("""./xxx"""			,			after=__snake_case  )
            _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(__snake_case  )}\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(__snake_case  )

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

            return output_dir




      def 				lowercase__    (			self			: Tuple			,			__snake_case			: List[Any]=False  )				->      List[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)
            _lowerCAmelCase							     =							min(2			,			get_gpu_count()  ) if distributed else 1
            return f"deepspeed --num_nodes 1 --num_gpus {num_gpus}".split()








from abc import ABC, abstractmethod
from argparse import ArgumentParser

class    lowerCAmelCase__     (	__lowercase			):






		@staticmethod
		@abstractmethod
		def  __A       (		SCREAMING_SNAKE_CASE__		:							ArgumentParser  )     ->							str:
						raise NotImplementedError()







		@abstractmethod
		def  __A       (		self		:							Optional[int]  )     ->							Union[str, Any]:
						raise NotImplementedError()



