



'''simple docstring'''





import asyncio
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from distutils.util import strtobool
from functools import partial
from pathlib import Path
from typing import List, Union
from unittest import mock

import torch

from ..state import AcceleratorState, PartialState
from ..utils import (
    gather,
    is_bnb_available,
    is_comet_ml_available,
    is_datasets_available,
    is_deepspeed_available,
    is_mps_available,
    is_safetensors_available,
    is_tensorboard_available,
    is_torch_version,
    is_tpu_available,
    is_transformers_available,
    is_wandb_available,
    is_xpu_available,
)





def   __snake_case			(    _UpperCAmelCase     ,				_UpperCAmelCase=False       ):
    try:
        __a		 =       os.environ[key]
    except KeyError:
        # KEY isn't set, default to `default`.
        __a		 =       default
    else:
        # KEY is set, convert it to True or False.
        try:
            __a		 =       strtobool(__A       )
        except ValueError:
            # More values are supported, but let's keep the message simple.
            raise ValueError(f'If set, {key} must be yes or no.'       )
    return _value


__snake_case		:str   	=  parse_flag_from_env('''RUN_SLOW''', default=False)





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skip('''Test was skipped'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(_run_slow_tests     ,				'''test is slow'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(not torch.cuda.is_available()     ,				'''test requires only a CPU'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(torch.cuda.is_available()     ,				'''test requires a GPU'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_xpu_available()     ,				'''test requires a XPU'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_mps_available()     ,				'''test requires a `mps` backend support in `torch`'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(
        is_transformers_available() and is_datasets_available()     ,				'''test requires the Hugging Face suite'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_bnb_available()     ,				'''test requires the bitsandbytes library'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_tpu_available()     ,				'''test requires TPU'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(torch.cuda.device_count() == 1     ,				'''test requires a GPU'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(torch.xpu.device_count() == 1     ,				'''test requires a XPU'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(torch.cuda.device_count() > 1     ,				'''test requires multiple GPUs'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(torch.xpu.device_count() > 1     ,				'''test requires multiple XPUs'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_safetensors_available()     ,				'''test requires safetensors'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_deepspeed_available()     ,				'''test requires DeepSpeed'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_torch_version('''>='''     ,				'''1.12.0'''       )     ,				'''test requires torch version >= 1.12.0'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase=None     ,				_UpperCAmelCase=None       ):
    if test_case is None:
        return partial(__A     ,				version=__A       )
    return unittest.skipUnless(is_torch_version('''>='''     ,				__A       )     ,				f'test requires torch version >= {version}'       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_tensorboard_available()     ,				'''test requires Tensorboard'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_wandb_available()     ,				'''test requires wandb'''       )(__A       )





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(is_comet_ml_available()     ,				'''test requires comet_ml'''       )(__A       )


__snake_case		:Union[str, Any]   	=  (
    any([is_wandb_available(), is_tensorboard_available()]) and not is_comet_ml_available()
)





def   __snake_case			(    _UpperCAmelCase       ):
    return unittest.skipUnless(
        _atleast_one_tracker_available     ,				'''test requires at least one tracker to be available and for `comet_ml` to not be installed'''     ,				)(__A       )

class 					_A (			unittest.TestCase					):
       UpperCamelCase__						:					List[Any]		      =		True



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


           '''simple docstring'''

           __a		 =       tempfile.mkdtemp()



       @classmethod
       def 	_lowerCamelCase    (	cls							:  int):


           '''simple docstring'''

           if os.path.exists(cls.tmpdir):
               shutil.rmtree(cls.tmpdir)





       def 	_lowerCamelCase    (	self							:  Any):


           '''simple docstring'''

           if self.clear_on_setup:
               for path in Path(self.tmpdir).glob('''**/*'''):
                   if path.is_file():
                       path.unlink()
                   elif path.is_dir():
                       shutil.rmtree(_UpperCamelCase)


class 					_A (			unittest.TestCase					):



       def 	_lowerCamelCase    (	self							:  int):


           '''simple docstring'''

           super().tearDown()
           # Reset the state of the AcceleratorState singleton.
           AcceleratorState._reset_state()
           PartialState._reset_state()


class 					_A (			unittest.TestCase					):



       def 	_lowerCamelCase    (	self							:  List[Any]  ,	__SCREAMING_SNAKE_CASE							:  Union[mock.Mock, List[mock.Mock]]):


           '''simple docstring'''

           __a		 =       mocks if isinstance(_UpperCamelCase  ,	(tuple, list)) else [mocks]
           for m in self.mocks:
               m.start()
               self.addCleanup(m.stop)





def   __snake_case			(    _UpperCAmelCase       ):
    __a		 =       AcceleratorState()
    __a		 =       tensor[None].clone().to(state.device       )
    __a		 =       gather(__A       ).cpu()
    __a		 =       tensor[0].cpu()
    for i in range(tensors.shape[0]       ):
        if not torch.equal(tensors[i]     ,				__A       ):
            return False
    return True

class 					_A :



       def __init__(	self							:  List[str]  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  Tuple):


           '''simple docstring'''

           __a		 =       returncode
           __a		 =       stdout
           __a		 =       stderr





async def   __snake_case			(    _UpperCAmelCase     ,				_UpperCAmelCase       ):
    while True:
        __a		 =       await stream.readline()
        if line:
            callback(__A       )
        else:
            break





async def   __snake_case			(    _UpperCAmelCase     ,				_UpperCAmelCase=None     ,				_UpperCAmelCase=None     ,				_UpperCAmelCase=None     ,				_UpperCAmelCase=False     ,				_UpperCAmelCase=False       ):
    if echo:
        print('''\nRunning: '''     ,				''' '''.join(__A       )       )

    __a		 =       await asyncio.create_subprocess_exec(
        cmd[0]     ,				*cmd[1:]     ,				stdin=__A     ,				stdout=asyncio.subprocess.PIPE     ,				stderr=asyncio.subprocess.PIPE     ,				env=__A     ,				)

    # note: there is a warning for a possible deadlock when using `wait` with huge amounts of data in the pipe
    # https://docs.python.org/3/library/asyncio-subprocess.html#asyncio.asyncio.subprocess.Process.wait
    #
    # If it starts hanging, will need to switch to the following code. The problem is that no data
    # will be seen until it's done and if it hangs for example there will be no debug info.
    # out, err = await p.communicate()
    # return _RunOutput(p.returncode, out, err)

    __a		 =       []
    __a		 =       []

    def tee(_UpperCAmelCase     ,				_UpperCAmelCase     ,				_UpperCAmelCase     ,				_UpperCAmelCase=""       ):
        __a		 =       line.decode('''utf-8'''       ).rstrip()
        sink.append(__A       )
        if not quiet:
            print(__A     ,				__A     ,				file=__A       )

    # XXX: the timeout doesn't seem to make any difference here
    await asyncio.wait(
        [
            asyncio.create_task(_read_stream(p.stdout     ,				lambda _UpperCAmelCase       : tee(__A     ,				__A     ,				sys.stdout     ,				label='''stdout:'''       )       )       ),
            asyncio.create_task(_read_stream(p.stderr     ,				lambda _UpperCAmelCase       : tee(__A     ,				__A     ,				sys.stderr     ,				label='''stderr:'''       )       )       ),
        ]     ,				timeout=__A     ,				)
    return _RunOutput(await p.wait()     ,				__A     ,				__A       )





def   __snake_case			(    _UpperCAmelCase     ,				_UpperCAmelCase=None     ,				_UpperCAmelCase=None     ,				_UpperCAmelCase=180     ,				_UpperCAmelCase=False     ,				_UpperCAmelCase=True       ):
    __a		 =       asyncio.get_event_loop()
    __a		 =       loop.run_until_complete(
        _stream_subprocess(__A     ,				env=__A     ,				stdin=__A     ,				timeout=__A     ,				quiet=__A     ,				echo=__A       )       )

    __a		 =       """ """.join(__A       )
    if result.returncode > 0:
        __a		 =       """\n""".join(result.stderr       )
        raise RuntimeError(
            f'\'{cmd_str}\' failed with returncode {result.returncode}\n\n'
            f'The combined stderr from workers follows:\n{stderr}'       )

    return result

class 					_A (			__lowerCAmelCase					):
       pass





def   __snake_case			(    _UpperCAmelCase     ,				_UpperCAmelCase=False       ):
    try:
        __a		 =       subprocess.check_output(__A     ,				stderr=subprocess.STDOUT       )
        if return_stdout:
            if hasattr(__A     ,				'''decode'''       ):
                __a		 =       output.decode('''utf-8'''       )
            return output
    except subprocess.CalledProcessError as e:
        raise SubprocessCallException(
            f'Command `{" ".join(__A       )}` failed with the following error:\n\n{e.output.decode()}'       ) from e










# 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


__snake_case		:Union[str, Any]   	=  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)

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

__snake_case		:List[Any]   	=  '''zero2'''
__snake_case		:Optional[Any]   	=  '''zero3'''
__snake_case		:str   	=  [ZEROa, ZEROa]





def   __snake_case			(    _UpperCAmelCase     ,				_UpperCAmelCase     ,				_UpperCAmelCase       ):
    # 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
    __a		 =       parameterized.to_safe_name('''_'''.join(str(_UpperCAmelCase       ) for x in param.args       )       )
    return f'{func.__name__}_{param_based_name}'


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

@slow
@require_deepspeed
@require_torch_gpu
class 					_A (			__UpperCAmelCase					):



       @parameterized.expand(__SCREAMING_SNAKE_CASE  ,	name_func=__SCREAMING_SNAKE_CASE)
       def 	_lowerCamelCase    (	self							:  Dict  ,	__SCREAMING_SNAKE_CASE							:  Optional[Any]  ,	__SCREAMING_SNAKE_CASE							:  Any):


           '''simple docstring'''

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



       @require_torch_multi_gpu
       @parameterized.expand(__SCREAMING_SNAKE_CASE  ,	name_func=__SCREAMING_SNAKE_CASE)
       def 	_lowerCamelCase    (	self							:  str  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  List[Any]):


           '''simple docstring'''

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



       @parameterized.expand(__SCREAMING_SNAKE_CASE  ,	name_func=__SCREAMING_SNAKE_CASE)
       def 	_lowerCamelCase    (	self							:  Tuple  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  Any):


           '''simple docstring'''

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



       @require_torch_multi_gpu
       @parameterized.expand(__SCREAMING_SNAKE_CASE  ,	name_func=__SCREAMING_SNAKE_CASE)
       def 	_lowerCamelCase    (	self							:  List[Any]  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  List[Any]):


           '''simple docstring'''

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



       def 	_lowerCamelCase    (	self							:  List[Any]  ,	__SCREAMING_SNAKE_CASE							:  List[str]):


           '''simple docstring'''

           pass



       def 	_lowerCamelCase    (	self							:  Any  ,	__SCREAMING_SNAKE_CASE							:  str  ,	__SCREAMING_SNAKE_CASE							:  str  ,	__SCREAMING_SNAKE_CASE							:  int = 10  ,	__SCREAMING_SNAKE_CASE							:  bool = True  ,	__SCREAMING_SNAKE_CASE							:  bool = True  ,	__SCREAMING_SNAKE_CASE							:  bool = True  ,	):


           '''simple docstring'''

           __a		 =       models[model]

           __a		 =       self.run_trainer(
               stage=__SCREAMING_SNAKE_CASE  ,	model_name=__SCREAMING_SNAKE_CASE  ,	eval_steps=__SCREAMING_SNAKE_CASE  ,	num_train_epochs=1  ,	distributed=__SCREAMING_SNAKE_CASE  ,	fpaa=__SCREAMING_SNAKE_CASE  ,	)

           self.do_checks(__SCREAMING_SNAKE_CASE)

           return output_dir



       def 	_lowerCamelCase    (	self							:  List[str]  ,	__SCREAMING_SNAKE_CASE							:  str  ,	__SCREAMING_SNAKE_CASE							:  str  ,	__SCREAMING_SNAKE_CASE							:  int = 10  ,	__SCREAMING_SNAKE_CASE							:  int = 1  ,	__SCREAMING_SNAKE_CASE							:  bool = True  ,	__SCREAMING_SNAKE_CASE							:  bool = True  ,	):


           '''simple docstring'''

           __a		 =       self.get_auto_remove_tmp_dir('''./xxx'''  ,	after=__SCREAMING_SNAKE_CASE)
           __a		 =       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(__SCREAMING_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
           __a		 =       F'--deepspeed {self.test_file_dir_str}/ds_config_wav2vec2_{stage}.json'.split()
           __a		 =       [F'{self.examples_dir_str}/research_projects/wav2vec2/run_asr.py']
           __a		 =       self.get_launcher(__SCREAMING_SNAKE_CASE)

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

           return output_dir





       def 	_lowerCamelCase    (	self							:  int  ,	__SCREAMING_SNAKE_CASE							:  List[Any]=False):


           '''simple docstring'''

           __a		 =       min(2  ,	get_gpu_count()) if distributed else 1
           return F'deepspeed --num_nodes 1 --num_gpus {num_gpus}'.split()





