

"""simple docstring"""





from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__UpperCamelCase    : List[str]				 =						logging.get_logger(__name__)







class 							SCREAMING_SNAKE_CASE   (						a_			):






					"""simple docstring"""



					lowercase__												=			["input_features", "attention_mask"]







					def __init__(			self     :							Union[str, Any]  ,lowercase_     :							Tuple=8_0  ,lowercase_     :							Optional[int]=1_6_0_0_0  ,lowercase_     :							Optional[Any]=8_0  ,lowercase_     :							List[str]=0.0  ,lowercase_     :							Dict=True  ,lowercase_     :							int=True  ,lowercase_     :							Union[str, Any]=True  ,**lowercase_     :							List[Any]  ,):
						super().__init__(feature_size=lowercase_  ,sampling_rate=lowercase_  ,padding_value=lowercase_  ,**lowercase_						)
						lowerCAmelCase__      :		Optional[Any]    							=      num_mel_bins
						lowerCAmelCase__      :		str    							=      do_ceptral_normalize
						lowerCAmelCase__      :		List[Any]    							=      normalize_means
						lowerCAmelCase__      :		Optional[Any]    							=      normalize_vars
						lowerCAmelCase__      :		int    							=      True







					def 			__lowerCAmelCase					(			self     :							Union[str, Any]  ,lowercase_     :							np.ndarray  ,):
						lowerCAmelCase__      :		Union[str, Any]    							=      waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
						lowerCAmelCase__      :		Union[str, Any]    							=      torch.from_numpy(lowercase_						).unsqueeze(0						)
						lowerCAmelCase__      :		List[str]    							=      ta_kaldi.fbank(lowercase_  ,num_mel_bins=self.num_mel_bins  ,sample_frequency=self.sampling_rate						)
						return features.numpy()







					@staticmethod
					def 			__lowerCAmelCase					(			lowercase_     :							np.ndarray  ,lowercase_     :							int  ,lowercase_     :							Optional[bool] = True  ,lowercase_     :							Optional[bool] = True  ,lowercase_     :							float = 0.0  ,):
						# make sure we normalize float32 arrays
						if normalize_means:
							lowerCAmelCase__      :		Union[str, Any]    							=      x[:input_length].mean(axis=0						)
							lowerCAmelCase__      :		int    							=      np.subtract(lowercase_  ,lowercase_						)
						if normalize_vars:
							lowerCAmelCase__      :		str    							=      x[:input_length].std(axis=0						)
							lowerCAmelCase__      :		Optional[int]    							=      np.divide(lowercase_  ,lowercase_						)

						if input_length < x.shape[0]:
							lowerCAmelCase__      :		Optional[int]    							=      padding_value

						# make sure array is in float32
						lowerCAmelCase__      :		int    							=      x.astype(np.floataa						)

						return x







					def 			__lowerCAmelCase					(			self     :							Union[str, Any]  ,lowercase_     :							List[np.ndarray]  ,lowercase_     :							Optional[np.ndarray] = None						):
						lowerCAmelCase__      :		List[Any]    							=      attention_mask.sum(-1						) if attention_mask is not None else [x.shape[0] for x in input_features]
						return [
						    self.utterance_cmvn(lowercase_  ,lowercase_  ,self.normalize_means  ,self.normalize_vars  ,self.padding_value						)
						    for x, n in zip(lowercase_  ,lowercase_						)
						]







					def __call__(			self     :							Any  ,lowercase_     :							Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]  ,lowercase_     :							Union[bool, str, PaddingStrategy] = False  ,lowercase_     :							Optional[int] = None  ,lowercase_     :							bool = False  ,lowercase_     :							Optional[int] = None  ,lowercase_     :							Optional[Union[str, TensorType]] = None  ,lowercase_     :							Optional[int] = None  ,lowercase_     :							Optional[bool] = None  ,**lowercase_     :							Optional[Any]  ,):

						if sampling_rate is not None:
							if sampling_rate != self.sampling_rate:
								raise ValueError(
								    F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
								    F' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'
								    F' {self.sampling_rate} and not {sampling_rate}.'						)
						else:
							logger.warning(
							    '''It is strongly recommended to pass the `sampling_rate` argument to this function. '''
							    '''Failing to do so can result in silent errors that might be hard to debug.'''						)

						lowerCAmelCase__      :		Union[str, Any]    							=      isinstance(lowercase_  ,np.ndarray						) and len(raw_speech.shape						) > 1
						if is_batched_numpy and len(raw_speech.shape						) > 2:
							raise ValueError(F'Only mono-channel audio is supported for input to {self}'						)
						lowerCAmelCase__      :		Optional[int]    							=      is_batched_numpy or (
						    isinstance(lowercase_  ,(list, tuple)						) and (isinstance(raw_speech[0]  ,(np.ndarray, tuple, list)						))
						)

						if is_batched:
							lowerCAmelCase__      :		Dict    							=      [np.asarray(lowercase_  ,dtype=np.floataa						) for speech in raw_speech]
						elif not is_batched and not isinstance(lowercase_  ,np.ndarray						):
							lowerCAmelCase__      :		Optional[int]    							=      np.asarray(lowercase_  ,dtype=np.floataa						)
						elif isinstance(lowercase_  ,np.ndarray						) and raw_speech.dtype is np.dtype(np.floataa						):
							lowerCAmelCase__      :		Any    							=      raw_speech.astype(np.floataa						)

						# always return batch
						if not is_batched:
							lowerCAmelCase__      :		str    							=      [raw_speech]

						# extract fbank features
						lowerCAmelCase__      :		List[str]    							=      [self._extract_fbank_features(lowercase_						) for waveform in raw_speech]

						# convert into correct format for padding
						lowerCAmelCase__      :		List[str]    							=      BatchFeature({'''input_features''': features}						)

						lowerCAmelCase__      :		List[Any]    							=      self.pad(
						    lowercase_  ,padding=lowercase_  ,max_length=lowercase_  ,truncation=lowercase_  ,pad_to_multiple_of=lowercase_  ,return_attention_mask=lowercase_  ,**lowercase_  ,)

						# make sure list is in array format
						lowerCAmelCase__      :		Optional[int]    							=      padded_inputs.get('''input_features'''						)
						if isinstance(input_features[0]  ,lowercase_						):
							lowerCAmelCase__      :		List[str]    							=      [np.asarray(lowercase_  ,dtype=np.floataa						) for feature in input_features]

						lowerCAmelCase__      :		Dict    							=      padded_inputs.get('''attention_mask'''						)
						if attention_mask is not None:
							lowerCAmelCase__      :		Optional[int]    							=      [np.asarray(lowercase_  ,dtype=np.intaa						) for array in attention_mask]

						# Utterance-level cepstral mean and variance normalization
						if self.do_ceptral_normalize:
							lowerCAmelCase__      :		Optional[Any]    							=      (
							    np.array(lowercase_  ,dtype=np.intaa						)
							    if self._get_padding_strategies(lowercase_  ,max_length=lowercase_						) is not PaddingStrategy.DO_NOT_PAD
							    else None
							)
							lowerCAmelCase__      :		List[str]    							=      self.normalize(
							    padded_inputs['''input_features''']  ,attention_mask=lowercase_						)

						if return_tensors is not None:
							lowerCAmelCase__      :		Dict    							=      padded_inputs.convert_to_tensors(lowercase_						)

						return padded_inputs







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


__UpperCamelCase    : int				 =						logging.get_logger(__name__)

__UpperCamelCase    : Optional[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 							SCREAMING_SNAKE_CASE   :






					"""simple docstring"""



					def __init__(			self     :							Optional[Any]  ,lowercase_     :							Union[str, Any]=None  ,**lowercase_     :							List[str]						):
						logger.info('''`diffusers.OnnxRuntimeModel` is experimental and might change in the future.'''						)
						lowerCAmelCase__      :		List[str]    							=      model
						lowerCAmelCase__      :		Optional[Any]    							=      kwargs.get('''model_save_dir'''  ,lowercase_						)
						lowerCAmelCase__      :		str    							=      kwargs.get('''latest_model_name'''  ,lowercase_						)







					def __call__(			self     :							Any  ,**lowercase_     :							Dict						):
						lowerCAmelCase__      :		List[Any]    							=      {k: np.array(lowercase_						) for k, v in kwargs.items()}
						return self.model.run(lowercase_  ,lowercase_						)







					@staticmethod
					def 			__lowerCAmelCase					(			lowercase_     :							Union[str, Path]  ,lowercase_     :							str=None  ,lowercase_     :							str=None						):
						if provider is None:
							logger.info('''No onnxruntime provider specified, using CPUExecutionProvider'''						)
							lowerCAmelCase__      :		Optional[Any]    							=      '''CPUExecutionProvider'''

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







					def 			__lowerCAmelCase					(			self     :							Dict  ,lowercase_     :							Union[str, Path]  ,lowercase_     :							Optional[str] = None  ,**lowercase_     :							str						):
						lowerCAmelCase__      :		Any    							=      file_name if file_name is not None else ONNX_WEIGHTS_NAME

						lowerCAmelCase__      :		Tuple    							=      self.model_save_dir.joinpath(self.latest_model_name						)
						lowerCAmelCase__      :		List[Any]    							=      Path(lowercase_						).joinpath(lowercase_						)
						try:
							shutil.copyfile(lowercase_  ,lowercase_						)
						except shutil.SameFileError:
							pass

						# copy external weights (for models >2GB)
						lowerCAmelCase__      :		Dict    							=      self.model_save_dir.joinpath(lowercase_						)
						if src_path.exists():
							lowerCAmelCase__      :		Union[str, Any]    							=      Path(lowercase_						).joinpath(lowercase_						)
							try:
								shutil.copyfile(lowercase_  ,lowercase_						)
							except shutil.SameFileError:
								pass







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

						os.makedirs(lowercase_  ,exist_ok=lowercase_						)

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







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







					@classmethod
					def 			__lowerCAmelCase					(			cls     :							Any  ,lowercase_     :							Union[str, Path]  ,lowercase_     :							bool = True  ,lowercase_     :							Optional[str] = None  ,lowercase_     :							Optional[str] = None  ,**lowercase_     :							Any  ,):
						lowerCAmelCase__      :		Union[str, Any]    							=      None
						if len(str(lowercase_						).split('''@'''						)						) == 2:
							lowerCAmelCase__							,lowerCAmelCase__      :		str    							=      model_id.split('''@'''						)

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




