






import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

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


lowerCAmelCase_           = logging.get_logger(__name__)
class      UpperCamelCase (				snake_case__ ):







		"""simple docstring"""



		snake_case    =   ["input_features"]
		def __init__(  self		:       Tuple						,_SCREAMING_SNAKE_CASE		:       Union[str, Any]=8_0						,_SCREAMING_SNAKE_CASE		:       List[Any]=1_6_0_0_0						,_SCREAMING_SNAKE_CASE		:       int=1_6_0						,_SCREAMING_SNAKE_CASE		:       List[Any]=3_0						,_SCREAMING_SNAKE_CASE		:       Any=4_0_0						,_SCREAMING_SNAKE_CASE		:       Optional[int]=0.0						,_SCREAMING_SNAKE_CASE		:       Dict=False						,**_SCREAMING_SNAKE_CASE		:       Dict						,)    ->   List[str]:


						'''simple docstring'''
						super().__init__(
						    feature_size=_SCREAMING_SNAKE_CASE						,sampling_rate=_SCREAMING_SNAKE_CASE						,padding_value=_SCREAMING_SNAKE_CASE						,return_attention_mask=_SCREAMING_SNAKE_CASE						,**_SCREAMING_SNAKE_CASE						,)
						A	 =      n_fft
						A	 =      hop_length
						A	 =      chunk_length
						A	 =      chunk_length * sampling_rate
						A	 =      self.n_samples // hop_length
						A	 =      sampling_rate
						A	 =      mel_filter_bank(
						    num_frequency_bins=1 + n_fft // 2						,num_mel_filters=_SCREAMING_SNAKE_CASE						,min_frequency=0.0						,max_frequency=8000.0						,sampling_rate=_SCREAMING_SNAKE_CASE						,norm='slaney'						,mel_scale='slaney'						,)
		def       A(  self		:       Optional[int]						,_SCREAMING_SNAKE_CASE		:       np.array		)    ->   np.ndarray:


						'''simple docstring'''
						A	 =      spectrogram(
						    _SCREAMING_SNAKE_CASE						,window_function(self.n_fft						,'hann'		)						,frame_length=self.n_fft						,hop_length=self.hop_length						,power=2.0						,mel_filters=self.mel_filters						,log_mel='log10'						,)
						A	 =      log_spec[:, :-1]
						A	 =      np.maximum(_SCREAMING_SNAKE_CASE						,log_spec.max() - 8.0		)
						A	 =      (log_spec + 4.0) / 4.0
						return log_spec
		@staticmethod
		# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
		def       A(  _SCREAMING_SNAKE_CASE		:       List[np.ndarray]						,_SCREAMING_SNAKE_CASE		:       List[np.ndarray]						,_SCREAMING_SNAKE_CASE		:       float = 0.0		)    ->   List[np.ndarray]:


						'''simple docstring'''
						if attention_mask is not None:
										A	 =      np.array(_SCREAMING_SNAKE_CASE						,np.intaa		)
										A	 =      []

										for vector, length in zip(_SCREAMING_SNAKE_CASE						,attention_mask.sum(-1		)		):
														A	 =      (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7		)
														if length < normed_slice.shape[0]:
																		A	 =      padding_value

														normed_input_values.append(_SCREAMING_SNAKE_CASE		)
						else:
										A	 =      [(x - x.mean()) / np.sqrt(x.var() + 1E-7		) for x in input_values]

						return normed_input_values
		def __call__(  self		:       int						,_SCREAMING_SNAKE_CASE		:       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]						,_SCREAMING_SNAKE_CASE		:       bool = True						,_SCREAMING_SNAKE_CASE		:       Optional[int] = None						,_SCREAMING_SNAKE_CASE		:       Optional[Union[str, TensorType]] = None						,_SCREAMING_SNAKE_CASE		:       Optional[bool] = None						,_SCREAMING_SNAKE_CASE		:       Optional[str] = "max_length"						,_SCREAMING_SNAKE_CASE		:       Optional[int] = None						,_SCREAMING_SNAKE_CASE		:       Optional[int] = None						,_SCREAMING_SNAKE_CASE		:       Optional[bool] = None						,**_SCREAMING_SNAKE_CASE		:       Any						,)    ->   BatchFeature:


						'''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.__class__.__name__} was trained using a'
														    f' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'
														    f' was sampled with {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.'		)

						A	 =      isinstance(_SCREAMING_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}'		)
						A	 =      is_batched_numpy or (
						    isinstance(_SCREAMING_SNAKE_CASE						,(list, tuple)		) and (isinstance(raw_speech[0]						,(np.ndarray, tuple, list)		))
						)

						if is_batched:
										A	 =      [np.asarray([speech]						,dtype=np.floataa		).T for speech in raw_speech]
						elif not is_batched and not isinstance(_SCREAMING_SNAKE_CASE						,np.ndarray		):
										A	 =      np.asarray(_SCREAMING_SNAKE_CASE						,dtype=np.floataa		)
						elif isinstance(_SCREAMING_SNAKE_CASE						,np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
										A	 =      raw_speech.astype(np.floataa		)

						# always return batch
						if not is_batched:
										A	 =      [np.asarray([raw_speech]		).T]

						A	 =      BatchFeature({'input_features': raw_speech}		)

						# convert into correct format for padding

						A	 =      self.pad(
						    _SCREAMING_SNAKE_CASE						,padding=_SCREAMING_SNAKE_CASE						,max_length=max_length if max_length else self.n_samples						,truncation=_SCREAMING_SNAKE_CASE						,pad_to_multiple_of=_SCREAMING_SNAKE_CASE						,return_attention_mask=return_attention_mask or do_normalize						,)

						# zero-mean and unit-variance normalization
						if do_normalize:
										A	 =      self.zero_mean_unit_var_norm(
										    padded_inputs['input_features']						,attention_mask=padded_inputs['attention_mask']						,padding_value=self.padding_value						,)
										A	 =      np.stack(padded_inputs['input_features']						,axis=0		)

						# make sure list is in array format
						A	 =      padded_inputs.get('input_features'		).transpose(2						,0						,1		)

						A	 =      [self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE		) for waveform in input_features[0]]

						if isinstance(input_features[0]						,_SCREAMING_SNAKE_CASE		):
										A	 =      [np.asarray(_SCREAMING_SNAKE_CASE						,dtype=np.floataa		) for feature in input_features]
						else:
										A	 =      input_features

						if return_attention_mask:
										# rescale from sample (48000) to feature (3000)
										A	 =      padded_inputs['attention_mask'][:, :: self.hop_length]

						if return_tensors is not None:
										A	 =      padded_inputs.convert_to_tensors(_SCREAMING_SNAKE_CASE		)

						return padded_inputs







		def       A(  self		:       Optional[Any]		)    ->   Dict[str, Any]:


						'''simple docstring'''
						A	 =      copy.deepcopy(self.__dict__		)
						A	 =      self.__class__.__name__
						if "mel_filters" in output:
										del output["mel_filters"]
						return output













from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
					from PIL import Image

					from ..image_utils import load_image

if is_tf_available():
					from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_VISION_2_SEQ_MAPPING

if is_torch_available():
					import torch

					from ..models.auto.modeling_auto import MODEL_FOR_VISION_2_SEQ_MAPPING

lowerCAmelCase_           = logging.get_logger(__name__)
@add_end_docstrings(snake_case__ )
class      UpperCamelCase (				snake_case__ ):







		"""simple docstring"""



		def __init__(  self		:       str						,*_SCREAMING_SNAKE_CASE		:       List[Any]						,**_SCREAMING_SNAKE_CASE		:       List[Any]		)    ->   Dict:


						'''simple docstring'''
						super().__init__(*_SCREAMING_SNAKE_CASE						,**_SCREAMING_SNAKE_CASE		)
						requires_backends(self						,'vision'		)
						self.check_model_type(
						    TF_MODEL_FOR_VISION_2_SEQ_MAPPING if self.framework == 'tf' else MODEL_FOR_VISION_2_SEQ_MAPPING		)
		def       A(  self		:       Union[str, Any]						,_SCREAMING_SNAKE_CASE		:       List[Any]=None						,_SCREAMING_SNAKE_CASE		:       List[str]=None						,_SCREAMING_SNAKE_CASE		:       int=None		)    ->   Any:


						'''simple docstring'''
						A	 =      {}
						A	 =      {}

						if prompt is not None:
										A	 =      prompt

						if generate_kwargs is not None:
										A	 =      generate_kwargs
						if max_new_tokens is not None:
										if "generate_kwargs" not in forward_kwargs:
														A	 =      {}
										if "max_new_tokens" in forward_kwargs["generate_kwargs"]:
														raise ValueError(
														    '\'max_new_tokens\' is defined twice, once in \'generate_kwargs\' and once as a direct parameter,'
														    ' please use only one'		)
										A	 =      max_new_tokens
						return preprocess_params, forward_kwargs, {}
		def __call__(  self		:       int						,_SCREAMING_SNAKE_CASE		:       Union[str, List[str], "Image.Image", List["Image.Image"]]						,**_SCREAMING_SNAKE_CASE		:       List[str]		)    ->   Union[str, Any]:


						'''simple docstring'''
						return super().__call__(_SCREAMING_SNAKE_CASE						,**_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       Dict						,_SCREAMING_SNAKE_CASE		:       List[Any]						,_SCREAMING_SNAKE_CASE		:       Optional[Any]=None		)    ->   int:


						'''simple docstring'''
						A	 =      load_image(_SCREAMING_SNAKE_CASE		)

						if prompt is not None:
										if not isinstance(_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		):
														raise ValueError(
														    f'Received an invalid text input, got - {type(_SCREAMING_SNAKE_CASE		)} - but expected a single string. '
														    'Note also that one single text can be provided for conditional image to text generation.'		)

										A	 =      self.model.config.model_type

										if model_type == "git":
														A	 =      self.image_processor(images=_SCREAMING_SNAKE_CASE						,return_tensors=self.framework		)
														A	 =      self.tokenizer(text=_SCREAMING_SNAKE_CASE						,add_special_tokens=_SCREAMING_SNAKE_CASE		).input_ids
														A	 =      [self.tokenizer.cls_token_id] + input_ids
														A	 =      torch.tensor(_SCREAMING_SNAKE_CASE		).unsqueeze(0		)
														model_inputs.update({'input_ids': input_ids}		)

										elif model_type == "pix2struct":
														A	 =      self.image_processor(images=_SCREAMING_SNAKE_CASE						,header_text=_SCREAMING_SNAKE_CASE						,return_tensors=self.framework		)

										elif model_type != "vision-encoder-decoder":
														# vision-encoder-decoder does not support conditional generation
														A	 =      self.image_processor(images=_SCREAMING_SNAKE_CASE						,return_tensors=self.framework		)
														A	 =      self.tokenizer(_SCREAMING_SNAKE_CASE						,return_tensors=self.framework		)
														model_inputs.update(_SCREAMING_SNAKE_CASE		)

										else:
														raise ValueError(f'Model type {model_type} does not support conditional text generation'		)

						else:
										A	 =      self.image_processor(images=_SCREAMING_SNAKE_CASE						,return_tensors=self.framework		)

						if self.model.config.model_type == "git" and prompt is None:
										A	 =      None

						return model_inputs
		def       A(  self		:       str						,_SCREAMING_SNAKE_CASE		:       str						,_SCREAMING_SNAKE_CASE		:       Dict=None		)    ->   List[str]:


						'''simple docstring'''
						# Git model sets `model_inputs["input_ids"] = None` in `preprocess` (when `prompt=None`). In batch model, the
						# pipeline will group them into a list of `None`, which fail `_forward`. Avoid this by checking it first.
						if (
						    "input_ids" in model_inputs
						    and isinstance(model_inputs['input_ids']						,_SCREAMING_SNAKE_CASE		)
						    and all(x is None for x in model_inputs['input_ids']		)
						):
										A	 =      None

						if generate_kwargs is None:
										A	 =      {}
						# FIXME: We need to pop here due to a difference in how `generation.py` and `generation.tf_utils.py`
						#  parse inputs. In the Tensorflow version, `generate` raises an error if we don't use `input_ids` whereas
						#  the PyTorch version matches it with `self.model.main_input_name` or `self.model.encoder.main_input_name`
						#  in the `_prepare_model_inputs` method.
						A	 =      model_inputs.pop(self.model.main_input_name		)
						A	 =      self.model.generate(_SCREAMING_SNAKE_CASE						,**_SCREAMING_SNAKE_CASE						,**_SCREAMING_SNAKE_CASE		)
						return model_outputs







		def       A(  self		:       Dict						,_SCREAMING_SNAKE_CASE		:       Dict		)    ->   Optional[int]:


						'''simple docstring'''
						A	 =      []
						for output_ids in model_outputs:
										A	 =      {
										    'generated_text': self.tokenizer.decode(
										        _SCREAMING_SNAKE_CASE						,skip_special_tokens=_SCREAMING_SNAKE_CASE						,)
										}
										records.append(_SCREAMING_SNAKE_CASE		)
						return records





