
import enum
import warnings

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
  import tensorflow as tf

  from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

if is_torch_available():
  from ..models.auto.modeling_auto import MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

lowercase  : Union[str, Any]						   =							logging.get_logger(__name__)

class __snake_case		(				enum.Enum					):
    _a		:						Any=    0
    _a		:						Tuple=    1






@add_end_docstrings(lowerCAmelCase					)
class __snake_case		(				lowerCAmelCase					):
    _a		:						int=    "generated"
    def __init__(  self		,*snake_case		,**snake_case				):



           '''simple docstring'''







           super().__init__(*snake_case		,**snake_case				)

           self.check_model_type(
               TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
               if self.framework == """tf"""
               else MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case=None		,snake_case=None		,snake_case=None		,snake_case=None		,snake_case=None		,snake_case=None		,**snake_case		,):



           '''simple docstring'''







           lowercase							:					str	     =				{}
           if truncation is not None:
                  lowercase							:					Union[str, Any]	     =				truncation

           lowercase							:					int	     =				generate_kwargs

           lowercase							:					Union[str, Any]	     =				{}
           if return_tensors is not None and return_type is None:
                  lowercase							:					Optional[Any]	     =				ReturnType.TENSORS if return_tensors else ReturnType.TEXT
           if return_type is not None:
                  lowercase							:					List[str]	     =				return_type

           if clean_up_tokenization_spaces is not None:
                  lowercase							:					Dict	     =				clean_up_tokenization_spaces

           if stop_sequence is not None:
                  lowercase							:					str	     =				self.tokenizer.encode(snake_case		,add_special_tokens=snake_case				)
                  if len(snake_case				) > 1:
                         warnings.warn(
                             """Stopping on a multiple token sequence is not yet supported on transformers. The first token of"""
                             """ the stop sequence will be used as the stop sequence string in the interim."""				)
                  lowercase							:					Optional[Any]	     =				stop_sequence_ids[0]

           return preprocess_params, forward_params, postprocess_params
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case		,snake_case				):



           '''simple docstring'''







           return True
    def 				_SCREAMING_SNAKE_CASE	(  self		,*snake_case		,snake_case				):



           '''simple docstring'''







           lowercase							:					Optional[int]	     =				self.model.config.prefix if self.model.config.prefix is not None else """"""
           if isinstance(args[0]		,snake_case				):
                  if self.tokenizer.pad_token_id is None:
                         raise ValueError("""Please make sure that the tokenizer has a pad_token_id when using a batch input"""				)
                  lowercase							:					List[Any]	     =				([prefix + arg for arg in args[0]],)
                  lowercase							:					int	     =				True

           elif isinstance(args[0]		,snake_case				):
                  lowercase							:					List[str]	     =				(prefix + args[0],)
                  lowercase							:					Dict	     =				False
           else:
                  raise ValueError(
                      f" `args[0]`: {args[0]} have the wrong format. The should be either of type `str` or type `list`"				)
           lowercase							:					Any	     =				self.tokenizer(*snake_case		,padding=snake_case		,truncation=snake_case		,return_tensors=self.framework				)
           # This is produced by tokenizers but is an invalid generate kwargs
           if "token_type_ids" in inputs:
                  del inputs["token_type_ids"]
           return inputs
    def __call__(  self		,*snake_case		,**snake_case				):



           '''simple docstring'''







           lowercase							:					str	     =				super().__call__(*snake_case		,**snake_case				)
           if (
               isinstance(args[0]		,snake_case				)
               and all(isinstance(snake_case		,snake_case				) for el in args[0]				)
               and all(len(snake_case				) == 1 for res in result				)
           ):
                  return [res[0] for res in result]
           return result
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case=TruncationStrategy.DO_NOT_TRUNCATE		,**snake_case				):



           '''simple docstring'''







           lowercase							:					Optional[int]	     =				self._parse_and_tokenize(snake_case		,truncation=snake_case		,**snake_case				)
           return inputs
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,**snake_case				):



           '''simple docstring'''







           if self.framework == "pt":
                  lowercase     ,      lowercase							:					str	     =				model_inputs["""input_ids"""].shape
           elif self.framework == "tf":
                  lowercase     ,      lowercase							:					List[Any]	     =				tf.shape(model_inputs["""input_ids"""]				).numpy()

           lowercase							:					Optional[Any]	     =				generate_kwargs.get("""min_length"""		,self.model.config.min_length				)
           lowercase							:					List[str]	     =				generate_kwargs.get("""max_length"""		,self.model.config.max_length				)
           self.check_inputs(snake_case		,generate_kwargs["""min_length"""]		,generate_kwargs["""max_length"""]				)
           lowercase							:					Optional[int]	     =				self.model.generate(**snake_case		,**snake_case				)
           lowercase							:					Optional[int]	     =				output_ids.shape[0]
           if self.framework == "pt":
                  lowercase							:					Optional[Any]	     =				output_ids.reshape(snake_case		,out_b // in_b		,*output_ids.shape[1:]				)
           elif self.framework == "tf":
                  lowercase							:					List[str]	     =				tf.reshape(snake_case		,(in_b, out_b // in_b, *output_ids.shape[1:])				)
           return {"output_ids": output_ids}
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case=ReturnType.TEXT		,snake_case=False				):



           '''simple docstring'''







           lowercase							:					List[str]	     =				[]
           for output_ids in model_outputs["output_ids"][0]:
                  if return_type == ReturnType.TENSORS:
                         lowercase							:					Tuple	     =				{f"{self.return_name}_token_ids": output_ids}
                  elif return_type == ReturnType.TEXT:
                         lowercase							:					List[str]	     =				{
                             f"{self.return_name}_text": self.tokenizer.decode(
                                 snake_case		,skip_special_tokens=snake_case		,clean_up_tokenization_spaces=snake_case		,)
                         }
                  records.append(snake_case				)
           return records






@add_end_docstrings(lowerCAmelCase					)
class __snake_case		(				lowerCAmelCase					):
    _a		:						Dict=    "summary"
    def __call__(  self		,*snake_case		,**snake_case				):



           '''simple docstring'''







           return super().__call__(*snake_case		,**snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case		,snake_case				):



           '''simple docstring'''







           if max_length < min_length:
                  logger.warning(f"Your min_length={min_length} must be inferior than your max_length={max_length}."				)

           if input_length < max_length:
                  logger.warning(
                      f"Your max_length is set to {max_length}, but your input_length is only {input_length}. Since this is "
                      """a summarization task, where outputs shorter than the input are typically wanted, you might """
                      f"consider decreasing max_length manually, e.g. summarizer('...', max_length={input_length//2})"				)






@add_end_docstrings(lowerCAmelCase					)
class __snake_case		(				lowerCAmelCase					):
    _a		:						Union[str, Any]=    "translation"
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case		,snake_case				):



           '''simple docstring'''







           if input_length > 0.9 * max_length:
                  logger.warning(
                      f"Your input_length: {input_length} is bigger than 0.9 * max_length: {max_length}. You might consider "
                      """increasing your max_length manually, e.g. translator('...', max_length=400)"""				)
           return True
    def 				_SCREAMING_SNAKE_CASE	(  self		,*snake_case		,snake_case=TruncationStrategy.DO_NOT_TRUNCATE		,snake_case=None		,snake_case=None				):



           '''simple docstring'''







           if getattr(self.tokenizer		,"""_build_translation_inputs"""		,snake_case				):
                  return self.tokenizer._build_translation_inputs(
                      *snake_case		,return_tensors=self.framework		,truncation=snake_case		,src_lang=snake_case		,tgt_lang=snake_case				)
           else:
                  return super()._parse_and_tokenize(*snake_case		,truncation=snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case=None		,snake_case=None		,**snake_case				):



           '''simple docstring'''







           lowercase     ,      lowercase     ,      lowercase							:					Optional[Any]	     =				super()._sanitize_parameters(**snake_case				)
           if src_lang is not None:
                  lowercase							:					List[Any]	     =				src_lang
           if tgt_lang is not None:
                  lowercase							:					List[str]	     =				tgt_lang
           if src_lang is None and tgt_lang is None:
                  # Backward compatibility, direct arguments use is preferred.
                  lowercase							:					str	     =				kwargs.get("""task"""		,self.task				)
                  lowercase							:					List[Any]	     =				task.split("""_"""				)
                  if task and len(snake_case				) == 4:
                         # translation, XX, to YY
                         lowercase							:					List[Any]	     =				items[1]
                         lowercase							:					List[str]	     =				items[3]
           return preprocess_params, forward_params, postprocess_params
    def __call__(  self		,*snake_case		,**snake_case				):



           '''simple docstring'''







           return super().__call__(*snake_case		,**snake_case				)


from typing import List, Optional, Union

import numpy as np

from ....audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ....feature_extraction_sequence_utils import SequenceFeatureExtractor
from ....feature_extraction_utils import BatchFeature
from ....file_utils import PaddingStrategy, TensorType
from ....utils import logging


lowercase  : Tuple						   =							logging.get_logger(__name__)

class __snake_case		(				lowerCAmelCase					):
    _a		:						Optional[Any]=    ["input_features", "attention_mask"]
    def __init__(  self		,snake_case=80		,snake_case=16000		,snake_case=0.0		,snake_case=10		,snake_case=25		,snake_case="hamming_window"		,snake_case=32_768.0		,snake_case=0.97		,snake_case=1.0		,snake_case=True		,snake_case=True		,snake_case=False		,**snake_case		,):



           '''simple docstring'''







           super().__init__(feature_size=snake_case		,sampling_rate=snake_case		,padding_value=snake_case		,**snake_case				)

           lowercase							:					Optional[Any]	     =				feature_size
           lowercase							:					List[Any]	     =				sampling_rate
           lowercase							:					int	     =				padding_value
           lowercase							:					Dict	     =				hop_length
           lowercase							:					List[str]	     =				win_length
           lowercase							:					List[Any]	     =				frame_signal_scale
           lowercase							:					List[Any]	     =				preemphasis_coeff
           lowercase							:					str	     =				mel_floor
           lowercase							:					int	     =				normalize_means
           lowercase							:					List[Any]	     =				normalize_vars
           lowercase							:					List[Any]	     =				win_function
           lowercase							:					int	     =				return_attention_mask

           lowercase							:					Any	     =				win_length * sampling_rate // 1000
           lowercase							:					Tuple	     =				hop_length * sampling_rate // 1000

           lowercase							:					Tuple	     =				optimal_fft_length(self.sample_size				)
           lowercase							:					Dict	     =				(self.n_fft // 2) + 1
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           if self.win_function == "hamming_window":
                  lowercase							:					Optional[Any]	     =				window_function(window_length=self.sample_size		,name=self.win_function		,periodic=snake_case				)
           else:
                  lowercase							:					Optional[Any]	     =				window_function(window_length=self.sample_size		,name=self.win_function				)

           lowercase							:					int	     =				mel_filter_bank(
               num_frequency_bins=self.n_freqs		,num_mel_filters=self.feature_size		,min_frequency=0.0		,max_frequency=self.sampling_rate / 2.0		,sampling_rate=self.sampling_rate		,)

           lowercase							:					Dict	     =				spectrogram(
               one_waveform * self.frame_signal_scale		,window=snake_case		,frame_length=self.sample_size		,hop_length=self.sample_stride		,fft_length=self.n_fft		,center=snake_case		,preemphasis=self.preemphasis_coeff		,mel_filters=snake_case		,mel_floor=self.mel_floor		,log_mel="""log"""		,)
           return msfc_features.T
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case		,snake_case				):



           '''simple docstring'''







           if self.normalize_means:
                  lowercase							:					List[Any]	     =				x[:input_length].mean(axis=0				)
                  lowercase							:					Dict	     =				np.subtract(snake_case		,snake_case				)
           if self.normalize_vars:
                  lowercase							:					List[Any]	     =				x[:input_length].std(axis=0				)
                  lowercase							:					List[Any]	     =				np.divide(snake_case		,snake_case				)

           if input_length < x.shape[0]:
                  lowercase							:					Any	     =				padding_value

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

           return x
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case = None				):



           '''simple docstring'''







           lowercase							:					Optional[Any]	     =				attention_mask.sum(-1				) if attention_mask is not None else [x.shape[0] for x in input_features]
           return [self._normalize_one(snake_case		,snake_case		,self.padding_value				) for x, n in zip(snake_case		,snake_case				)]
    def __call__(  self		,snake_case		,snake_case = False		,snake_case = None		,snake_case = False		,snake_case = None		,snake_case = None		,snake_case = None		,snake_case = None		,**snake_case		,):



           '''simple docstring'''







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

           lowercase							:					List[Any]	     =				isinstance(snake_case		,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}"				)
           lowercase							:					str	     =				is_batched_numpy or (
               isinstance(snake_case		,(list, tuple)				) and (isinstance(raw_speech[0]		,(np.ndarray, tuple, list)				))
           )

           if is_batched:
                  lowercase							:					List[str]	     =				[np.asarray(snake_case		,dtype=np.floataa				) for speech in raw_speech]
           elif not is_batched and not isinstance(snake_case		,np.ndarray				):
                  lowercase							:					int	     =				np.asarray(snake_case		,dtype=np.floataa				)
           elif isinstance(snake_case		,np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
                  lowercase							:					Any	     =				raw_speech.astype(np.floataa				)

           # always return batch
           if not is_batched:
                  lowercase							:					Dict	     =				[raw_speech]

           # extract fbank features
           lowercase							:					Tuple	     =				[self._extract_mfsc_features(snake_case				) for one_waveform in raw_speech]

           # convert into correct format for padding
           lowercase							:					Union[str, Any]	     =				BatchFeature({"""input_features""": features}				)

           lowercase							:					Optional[int]	     =				self.pad(
               snake_case		,padding=snake_case		,max_length=snake_case		,truncation=snake_case		,pad_to_multiple_of=snake_case		,return_attention_mask=snake_case		,**snake_case		,)
           # make sure list is in array format
           lowercase							:					Tuple	     =				padded_inputs.get("""input_features"""				)
           if isinstance(input_features[0]		,snake_case				):
                  lowercase							:					List[Any]	     =				[np.asarray(snake_case		,dtype=np.floataa				) for feature in input_features]

           lowercase							:					int	     =				padded_inputs.get("""attention_mask"""				)
           if attention_mask is not None:
                  lowercase							:					Any	     =				[np.asarray(snake_case		,dtype=np.intaa				) for array in attention_mask]

           if self.normalize_means or self.normalize_vars:
                  lowercase							:					List[str]	     =				(
                      np.array(snake_case		,dtype=np.intaa				)
                      if self._get_padding_strategies(snake_case		,max_length=snake_case				) is not PaddingStrategy.DO_NOT_PAD
                      and padding
                      else None
                  )
                  lowercase							:					List[str]	     =				self.normalize(
                      padded_inputs["""input_features"""]		,attention_mask=snake_case				)

           if return_tensors is not None:
                  lowercase							:					str	     =				padded_inputs.convert_to_tensors(snake_case				)

           return padded_inputs
