




# 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


_UpperCAmelCase	:						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)

_UpperCAmelCase	:						Optional[Any]			=     {"base": "patrickvonplaten/wav2vec2_tiny_random", "robust": "patrickvonplaten/wav2vec2_tiny_random_robust"}

_UpperCAmelCase	:						str			=     "zero2"
_UpperCAmelCase	:						Any			=     "zero3"
_UpperCAmelCase	:						List[str]			=     [ZEROa, ZEROa]



def       UpperCAmelCase__  (     lowerCamelCase,			lowerCamelCase,			lowerCamelCase   ):
  # 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
  lowercase    :List[Any]      	=						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
_UpperCAmelCase	:						int			=     list(itertools.product(stages, models.keys()))







@slow
@require_deepspeed
@require_torch_gpu
class    __lowerCAmelCase       (							lowerCAmelCase):






       @parameterized.expand(_lowerCAmelCase       ,					name_func=_lowerCAmelCase				)
       def 					SCREAMING_SNAKE_CASE    (   self:					List[Any]       ,					_lowerCAmelCase:					List[str]       ,					_lowerCAmelCase:					Optional[int]				):
         self.run_and_check(
             stage=_lowerCAmelCase       ,					model=_lowerCAmelCase       ,					distributed=_lowerCAmelCase       ,					fpaa=_lowerCAmelCase       ,					)






       @require_torch_multi_gpu
       @parameterized.expand(_lowerCAmelCase       ,					name_func=_lowerCAmelCase				)
       def 					SCREAMING_SNAKE_CASE    (   self:					Dict       ,					_lowerCAmelCase:					Any       ,					_lowerCAmelCase:					List[Any]				):
         self.run_and_check(
             stage=_lowerCAmelCase       ,					model=_lowerCAmelCase       ,					distributed=_lowerCAmelCase       ,					fpaa=_lowerCAmelCase       ,					)






       @parameterized.expand(_lowerCAmelCase       ,					name_func=_lowerCAmelCase				)
       def 					SCREAMING_SNAKE_CASE    (   self:					int       ,					_lowerCAmelCase:					List[Any]       ,					_lowerCAmelCase:					Optional[Any]				):
         self.run_and_check(
             stage=_lowerCAmelCase       ,					model=_lowerCAmelCase       ,					distributed=_lowerCAmelCase       ,					fpaa=_lowerCAmelCase       ,					)






       @require_torch_multi_gpu
       @parameterized.expand(_lowerCAmelCase       ,					name_func=_lowerCAmelCase				)
       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[int]       ,					_lowerCAmelCase:					int       ,					_lowerCAmelCase:					Optional[Any]				):
         self.run_and_check(
             stage=_lowerCAmelCase       ,					model=_lowerCAmelCase       ,					distributed=_lowerCAmelCase       ,					fpaa=_lowerCAmelCase       ,					)






       def 					SCREAMING_SNAKE_CASE    (   self:					Any       ,					_lowerCAmelCase:					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 					SCREAMING_SNAKE_CASE    (   self:					Optional[int]       ,					_lowerCAmelCase:					str       ,					_lowerCAmelCase:					str       ,					_lowerCAmelCase:					int = 10       ,					_lowerCAmelCase:					bool = True       ,					_lowerCAmelCase:					bool = True       ,					_lowerCAmelCase:					bool = True       ,					):
         lowercase    :Tuple      	=						models[model]

         lowercase    :Optional[int]      	=						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 					SCREAMING_SNAKE_CASE    (   self:					int       ,					_lowerCAmelCase:					str       ,					_lowerCAmelCase:					str       ,					_lowerCAmelCase:					int = 10       ,					_lowerCAmelCase:					int = 1       ,					_lowerCAmelCase:					bool = True       ,					_lowerCAmelCase:					bool = True       ,					):
         lowercase    :Optional[int]      	=						self.get_auto_remove_tmp_dir("./xxx"       ,					after=_lowerCAmelCase				)
         lowercase    :Dict      	=						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
         lowercase    :Union[str, Any]      	=						F"--deepspeed {self.test_file_dir_str}/ds_config_wav2vec2_{stage}.json".split()
         lowercase    :Optional[int]      	=						[F"{self.examples_dir_str}/research_projects/wav2vec2/run_asr.py"]
         lowercase    :Dict      	=						self.get_launcher(_lowerCAmelCase				)

         lowercase    :int      	=						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 					SCREAMING_SNAKE_CASE    (   self:					Dict       ,					_lowerCAmelCase:					Optional[int]=False				):
         # 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)
         lowercase    :int      	=						min(2       ,					get_gpu_count()				) if distributed else 1
         return F"deepspeed --num_nodes 1 --num_gpus {num_gpus}".split()









import os
import shutil
from pathlib import Path
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ..utils import ONNX_EXTERNAL_WEIGHTS_NAME, ONNX_WEIGHTS_NAME, is_onnx_available, logging


if is_onnx_available():
     import onnxruntime as ort


_UpperCAmelCase	:						Dict			=     logging.get_logger(__name__)

_UpperCAmelCase	:						List[Any]			=     {
    "tensor(bool)": np.bool_,
    "tensor(int8)": np.inta,
    "tensor(uint8)": np.uinta,
    "tensor(int16)": np.intaa,
    "tensor(uint16)": np.uintaa,
    "tensor(int32)": np.intaa,
    "tensor(uint32)": np.uintaa,
    "tensor(int64)": np.intaa,
    "tensor(uint64)": np.uintaa,
    "tensor(float16)": np.floataa,
    "tensor(float)": np.floataa,
    "tensor(double)": np.floataa,
}




class    __lowerCAmelCase       :






       def __init__(   self:					Dict       ,					_lowerCAmelCase:					Union[str, Any]=None       ,					**_lowerCAmelCase:					Any				):
         logger.info("`diffusers.OnnxRuntimeModel` is experimental and might change in the future."				)
         lowercase    :str      	=						model
         lowercase    :Any      	=						kwargs.get("model_save_dir"       ,					_lowerCAmelCase				)
         lowercase    :Dict      	=						kwargs.get("latest_model_name"       ,					_lowerCAmelCase				)






       def __call__(   self:					Optional[Any]       ,					**_lowerCAmelCase:					List[Any]				):
         lowercase    :List[str]      	=						{k: np.array(_lowerCAmelCase				) for k, v in kwargs.items()}
         return self.model.run(_lowerCAmelCase       ,					_lowerCAmelCase				)






       @staticmethod
       def 					SCREAMING_SNAKE_CASE    (   _lowerCAmelCase:					Union[str, Path]       ,					_lowerCAmelCase:					List[Any]=None       ,					_lowerCAmelCase:					List[Any]=None				):
         if provider is None:
           logger.info("No onnxruntime provider specified, using CPUExecutionProvider"				)
           lowercase    :int      	=						"CPUExecutionProvider"

         return ort.InferenceSession(_lowerCAmelCase       ,					providers=[provider]       ,					sess_options=_lowerCAmelCase				)






       def 					SCREAMING_SNAKE_CASE    (   self:					List[Any]       ,					_lowerCAmelCase:					Union[str, Path]       ,					_lowerCAmelCase:					Optional[str] = None       ,					**_lowerCAmelCase:					Any				):
         lowercase    :Optional[int]      	=						file_name if file_name is not None else ONNX_WEIGHTS_NAME

         lowercase    :Tuple      	=						self.model_save_dir.joinpath(self.latest_model_name				)
         lowercase    :Optional[int]      	=						Path(_lowerCAmelCase				).joinpath(_lowerCAmelCase				)
         try:
           shutil.copyfile(_lowerCAmelCase       ,					_lowerCAmelCase				)
         except shutil.SameFileError:
           pass

         # copy external weights (for models >2GB)
         lowercase    :Tuple      	=						self.model_save_dir.joinpath(_lowerCAmelCase				)
         if src_path.exists():
           lowercase    :Dict      	=						Path(_lowerCAmelCase				).joinpath(_lowerCAmelCase				)
           try:
             shutil.copyfile(_lowerCAmelCase       ,					_lowerCAmelCase				)
           except shutil.SameFileError:
             pass






       def 					SCREAMING_SNAKE_CASE    (   self:					Optional[Any]       ,					_lowerCAmelCase:					Union[str, os.PathLike]       ,					**_lowerCAmelCase:					int       ,					):
         if os.path.isfile(_lowerCAmelCase				):
           logger.error(F"Provided path ({save_directory}) should be a directory, not a file"				)
           return

         os.makedirs(_lowerCAmelCase       ,					exist_ok=_lowerCAmelCase				)

         # saving model weights/files
         self._save_pretrained(_lowerCAmelCase       ,					**_lowerCAmelCase				)






       @classmethod
       def 					SCREAMING_SNAKE_CASE    (   cls:					Dict       ,					_lowerCAmelCase:					Union[str, Path]       ,					_lowerCAmelCase:					Optional[Union[bool, str, None]] = None       ,					_lowerCAmelCase:					Optional[Union[str, None]] = None       ,					_lowerCAmelCase:					bool = False       ,					_lowerCAmelCase:					Optional[str] = None       ,					_lowerCAmelCase:					Optional[str] = None       ,					_lowerCAmelCase:					Optional[str] = None       ,					_lowerCAmelCase:					Optional["ort.SessionOptions"] = None       ,					**_lowerCAmelCase:					Optional[int]       ,					):
         lowercase    :List[str]      	=						file_name if file_name is not None else ONNX_WEIGHTS_NAME
         # load model from local directory
         if os.path.isdir(_lowerCAmelCase				):
           lowercase    :Union[str, Any]      	=						OnnxRuntimeModel.load_model(
               os.path.join(_lowerCAmelCase       ,					_lowerCAmelCase				)       ,					provider=_lowerCAmelCase       ,					sess_options=_lowerCAmelCase				)
           lowercase    :Optional[Any]      	=						Path(_lowerCAmelCase				)
         # load model from hub
         else:
           # download model
           lowercase    :str      	=						hf_hub_download(
               repo_id=_lowerCAmelCase       ,					filename=_lowerCAmelCase       ,					use_auth_token=_lowerCAmelCase       ,					revision=_lowerCAmelCase       ,					cache_dir=_lowerCAmelCase       ,					force_download=_lowerCAmelCase       ,					)
           lowercase    :Optional[int]      	=						Path(_lowerCAmelCase				).parent
           lowercase    :Tuple      	=						Path(_lowerCAmelCase				).name
           lowercase    :Tuple      	=						OnnxRuntimeModel.load_model(_lowerCAmelCase       ,					provider=_lowerCAmelCase       ,					sess_options=_lowerCAmelCase				)
         return cls(model=_lowerCAmelCase       ,					**_lowerCAmelCase				)






       @classmethod
       def 					SCREAMING_SNAKE_CASE    (   cls:					str       ,					_lowerCAmelCase:					Union[str, Path]       ,					_lowerCAmelCase:					bool = True       ,					_lowerCAmelCase:					Optional[str] = None       ,					_lowerCAmelCase:					Optional[str] = None       ,					**_lowerCAmelCase:					Any       ,					):
         lowercase    :List[str]      	=						None
         if len(str(_lowerCAmelCase				).split("@"				)				) == 2:
           lowercase	,    lowercase    :Tuple      	=						model_id.split("@"				)

         return cls._from_pretrained(
             model_id=_lowerCAmelCase       ,					revision=_lowerCAmelCase       ,					cache_dir=_lowerCAmelCase       ,					force_download=_lowerCAmelCase       ,					use_auth_token=_lowerCAmelCase       ,					**_lowerCAmelCase       ,					)



