2020 the huggingface team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache orglicenseslicense2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license xxx we want transformers master here in the absense of conftest manipulating sys path hack it in for now customize the test name generator function as we want both params to appear in the subtest name as by default it shows only the first param cartesianproduct of zero stages with models to test xxx runasr is premature and doesn t save any results so all we check for now is that the process didn t fail xxx need to do better validation beyond just that the run was successful split if fp16 args extendfp16 currently dsconfigwav2vec2zero json requires zerooptimization findunusedparameters true hence the separate config files dsargs fdeepspeed self testfiledirstrdsconfigwav2vec2stage json split script fself examplesdirstrresearchprojectswav2vec2runasr py launcher self getlauncherdistributed cmd launcher script args dsargs keep for quick debug print joinfnpythonpathself srcdirstr cmd die executesubprocessasynccmd envself getenv return outputdir def getlauncherself distributedfalse 1 explicitly set numnodes1 just in case these tests end up run on a multinode 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 numgpus min2 getgpucount if distributed else 1 return fdeepspeed numnodes 1 numgpus numgpus split 2020 the huggingface team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache org licenses license 2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license xxx we want transformers master here in the absense of conftest manipulating sys path hack it in for now noqa noqa noqa noqa noqa noqa noqa noqa noqa noqa noqa noqa noqa customize the test name generator function as we want both params to appear in the sub test name as by default it shows only the first param cartesian product of zero stages with models to test xxx run_asr is premature and doesn t save any results so all we check for now is that the process didn t fail xxx need to do better validation beyond just that the run was successful 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 num_train_epochs 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 currently ds_config_wav2vec2_zero json requires zero_optimization find_unused_parameters true hence the separate config files keep for quick debug print join f npythonpath self src_dir_str cmd die 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
import sys
from pathlib import Path


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

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

from parameterized import parameterized  
from transformers import TrainingArguments, is_torch_available  
from transformers.integrations.deepspeed import is_deepspeed_available  
from transformers.file_utils import WEIGHTS_NAME  
from transformers.testing_utils import (  
    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  


set_seed(42)

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

ZERO2 = "zero2"
ZERO3 = "zero3"
stages = [ZERO2, ZERO3]


def custom_name_func(func, param_num, param):
    
    
    param_based_name = parameterized.to_safe_name("_".join(str(x) for x in param.args))
    return f"{func.__name__}_{param_based_name}"



params = list(itertools.product(stages, models.keys()))


@slow
@require_deepspeed
@require_torch_gpu
class TestDeepSpeedWav2Vec2(TestCasePlus):
    @parameterized.expand(params, name_func=custom_name_func)
    def test_fp32_non_distributed(self, stage, model):
        self.run_and_check(
            stage=stage,
            model=model,
            distributed=False,
            fp16=False,
        )

    @require_torch_multi_gpu
    @parameterized.expand(params, name_func=custom_name_func)
    def test_fp32_distributed(self, stage, model):
        self.run_and_check(
            stage=stage,
            model=model,
            distributed=True,
            fp16=False,
        )

    @parameterized.expand(params, name_func=custom_name_func)
    def test_fp16_non_distributed(self, stage, model):
        self.run_and_check(
            stage=stage,
            model=model,
            distributed=False,
            fp16=True,
        )

    @require_torch_multi_gpu
    @parameterized.expand(params, name_func=custom_name_func)
    def test_fp16_distributed(self, stage, model):
        self.run_and_check(
            stage=stage,
            model=model,
            distributed=True,
            fp16=True,
        )

    def do_checks(self, output_dir):
        
        
        pass

    
    def run_and_check(
        self,
        stage: str,
        model: str,
        eval_steps: int = 10,
        distributed: bool = True,
        quality_checks: bool = True,
        fp16: bool = True,
    ):
        model_name = models[model]

        output_dir = self.run_trainer(
            stage=stage,
            model_name=model_name,
            eval_steps=eval_steps,
            num_train_epochs=1,
            distributed=distributed,
            fp16=fp16,
        )

        self.do_checks(output_dir)

        return output_dir

    def run_trainer(
        self,
        stage: str,
        model_name: str,
        eval_steps: int = 10,
        num_train_epochs: int = 1,
        distributed: bool = True,
        fp16: bool = True,
    ):
        output_dir = self.get_auto_remove_tmp_dir("./xxx", after=False)
        args = f.split()

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

        
        
        ds_args = f"--deepspeed {self.test_file_dir_str}/ds_config_wav2vec2_{stage}.json".split()
        script = [f"{self.examples_dir_str}/research_projects/wav2vec2/run_asr.py"]
        launcher = self.get_launcher(distributed)

        cmd = launcher + script + args + ds_args
        
        
        execute_subprocess_async(cmd, env=self.get_env())

        return output_dir

    def get_launcher(self, distributed=False):
        
        
        
        
        num_gpus = min(2, get_gpu_count()) if distributed else 1
        return f"deepspeed --num_nodes 1 --num_gpus {num_gpus}".split()