"""simple docstring"""



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


SCREAMING_SNAKE_CASE__:List[Any]			=	logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__:List[str]			=	{
    """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 				snake_case__     :


 def __init__(				self  ,						lowerCamelCase=None  ,						**lowerCamelCase		):
        logger.info("`diffusers.OnnxRuntimeModel` is experimental and might change in the future."		)
        __a     				=  model
        __a     				=  kwargs.get("model_save_dir"  ,						lowerCamelCase		)
        __a     				=  kwargs.get("latest_model_name"  ,						lowerCamelCase		)


 def __call__(				self  ,						**lowerCamelCase		):
        __a     				=  {k: np.array(lowerCamelCase		) for k, v in kwargs.items()}
        return self.model.run(lowerCamelCase  ,						lowerCamelCase		)


 @staticmethod
 def 		a__ (				lowerCamelCase  ,						lowerCamelCase=None  ,						lowerCamelCase=None		):
        if provider is None:
               logger.info("No onnxruntime provider specified, using CPUExecutionProvider"		)
               __a     				=  "CPUExecutionProvider"

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


 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase = None  ,						**lowerCamelCase		):
        __a     				=  file_name if file_name is not None else ONNX_WEIGHTS_NAME

        __a     				=  self.model_save_dir.joinpath(self.latest_model_name		)
        __a     				=  Path(lowerCamelCase		).joinpath(lowerCamelCase		)
        try:
               shutil.copyfile(lowerCamelCase  ,						lowerCamelCase		)
        except shutil.SameFileError:
               pass

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


 def 		a__ (				self  ,						lowerCamelCase  ,						**lowerCamelCase  ,						):
        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 		a__ (				cls  ,						lowerCamelCase  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = False  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						**lowerCamelCase  ,						):
        __a     				=  file_name if file_name is not None else ONNX_WEIGHTS_NAME
        # load model from local directory
        if os.path.isdir(lowerCamelCase		):
               __a     				=  OnnxRuntimeModel.load_model(
                   os.path.join(lowerCamelCase  ,						lowerCamelCase		)  ,						provider=lowerCamelCase  ,						sess_options=lowerCamelCase		)
               __a     				=  Path(lowerCamelCase		)
        # load model from hub
        else:
               # download model
               __a     				=  hf_hub_download(
                   repo_id=lowerCamelCase  ,						filename=lowerCamelCase  ,						use_auth_token=lowerCamelCase  ,						revision=lowerCamelCase  ,						cache_dir=lowerCamelCase  ,						force_download=lowerCamelCase  ,						)
               __a     				=  Path(lowerCamelCase		).parent
               __a     				=  Path(lowerCamelCase		).name
               __a     				=  OnnxRuntimeModel.load_model(lowerCamelCase  ,						provider=lowerCamelCase  ,						sess_options=lowerCamelCase		)
        return cls(model=lowerCamelCase  ,						**lowerCamelCase		)






 @classmethod
 def 		a__ (				cls  ,						lowerCamelCase  ,						lowerCamelCase = True  ,						lowerCamelCase = None  ,						lowerCamelCase = None  ,						**lowerCamelCase  ,						):
        __a     				=  None
        if len(str(lowerCamelCase		).split("@"		)		) == 2:
               __a				, __a     				=  model_id.split("@"		)

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

"""simple docstring"""



from ...processing_utils import ProcessorMixin






class 				snake_case__     (   snake_case_   ):
 _snake_case   :			List[str]           =		"""WhisperFeatureExtractor"""
 _snake_case   :			Any           =		"""WhisperTokenizer"""


 def __init__(				self  ,						lowerCamelCase  ,						lowerCamelCase		):
        super().__init__(lowerCamelCase  ,						lowerCamelCase		)
        __a     				=  self.feature_extractor
        __a     				=  False


 def 		a__ (				self  ,						lowerCamelCase=None  ,						lowerCamelCase=None  ,						lowerCamelCase=True		):
        return self.tokenizer.get_decoder_prompt_ids(task=lowerCamelCase  ,						language=lowerCamelCase  ,						no_timestamps=lowerCamelCase		)


 def __call__(				self  ,						*lowerCamelCase  ,						**lowerCamelCase		):
        # For backward compatibility
        if self._in_target_context_manager:
               return self.current_processor(*lowerCamelCase  ,						**lowerCamelCase		)

        __a     				=  kwargs.pop("audio"  ,						lowerCamelCase		)
        __a     				=  kwargs.pop("sampling_rate"  ,						lowerCamelCase		)
        __a     				=  kwargs.pop("text"  ,						lowerCamelCase		)
        if len(lowerCamelCase		) > 0:
               __a     				=  args[0]
               __a     				=  args[1:]

        if audio is None and text is None:
               raise ValueError("You need to specify either an `audio` or `text` input to process."		)

        if audio is not None:
               __a     				=  self.feature_extractor(lowerCamelCase  ,						*lowerCamelCase  ,						sampling_rate=lowerCamelCase  ,						**lowerCamelCase		)
        if text is not None:
               __a     				=  self.tokenizer(lowerCamelCase  ,						**lowerCamelCase		)

        if text is None:
               return inputs

        elif audio is None:
               return encodings
        else:
               __a     				=  encodings["input_ids"]
               return inputs


 def 		a__ (				self  ,						*lowerCamelCase  ,						**lowerCamelCase		):
        return self.tokenizer.batch_decode(*lowerCamelCase  ,						**lowerCamelCase		)


 def 		a__ (				self  ,						*lowerCamelCase  ,						**lowerCamelCase		):
        return self.tokenizer.decode(*lowerCamelCase  ,						**lowerCamelCase		)






 def 		a__ (				self  ,						lowerCamelCase  ,						lowerCamelCase="np"		):
        return self.tokenizer.get_prompt_ids(lowerCamelCase  ,						return_tensors=lowerCamelCase		)
