



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

import numpy as np
import torch

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


__UpperCAmelCase			  =			logging.get_logger(__name__)



class 							lowerCAmelCase_						(    a__				):
       UpperCAmelCase__							:   int  	=    ["input_features", "is_longer"]
       def __init__(     self,     SCREAMING_SNAKE_CASE_=64,     SCREAMING_SNAKE_CASE_=4_8000,     SCREAMING_SNAKE_CASE_=480,     SCREAMING_SNAKE_CASE_=10,     SCREAMING_SNAKE_CASE_=1024,     SCREAMING_SNAKE_CASE_=0.0,     SCREAMING_SNAKE_CASE_=False,     SCREAMING_SNAKE_CASE_ = 0,     SCREAMING_SNAKE_CASE_ = 1_4000,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = "fusion",     SCREAMING_SNAKE_CASE_ = "repeatpad",     **SCREAMING_SNAKE_CASE_,     )	->						Optional[Any]:
          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_,     )
          UpperCamelCase       :    Any		      =      top_db
          UpperCamelCase       :    Optional[int]		      =      truncation
          UpperCamelCase       :    Any		      =      padding
          UpperCamelCase       :    Any		      =      fft_window_size
          UpperCamelCase       :    Tuple		      =      (fft_window_size >> 1) + 1
          UpperCamelCase       :    List[str]		      =      hop_length
          UpperCamelCase       :    Dict		      =      max_length_s
          UpperCamelCase       :    List[str]		      =      max_length_s * sampling_rate
          UpperCamelCase       :    int		      =      sampling_rate
          UpperCamelCase       :    Any		      =      frequency_min
          UpperCamelCase       :    Optional[Any]		      =      frequency_max
          UpperCamelCase       :    Dict		      =      mel_filter_bank(
              num_frequency_bins=self.nb_frequency_bins,     num_mel_filters=SCREAMING_SNAKE_CASE_,     min_frequency=SCREAMING_SNAKE_CASE_,     max_frequency=SCREAMING_SNAKE_CASE_,     sampling_rate=SCREAMING_SNAKE_CASE_,     norm=SCREAMING_SNAKE_CASE_,     mel_scale='htk',     )
          UpperCamelCase       :    Dict		      =      mel_filter_bank(
              num_frequency_bins=self.nb_frequency_bins,     num_mel_filters=SCREAMING_SNAKE_CASE_,     min_frequency=SCREAMING_SNAKE_CASE_,     max_frequency=SCREAMING_SNAKE_CASE_,     sampling_rate=SCREAMING_SNAKE_CASE_,     norm='slaney',     mel_scale='slaney',     )
       def    snake_case_			(     self			)	->						Dict[str, Any]:
          UpperCamelCase       :    Dict		      =      copy.deepcopy(self.__dict__			)
          UpperCamelCase       :    Any		      =      self.__class__.__name__
          if "mel_filters" in output:
             del output["mel_filters"]
          if "mel_filters_slaney" in output:
             del output["mel_filters_slaney"]
          return output
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = None			)	->						np.ndarray:
          UpperCamelCase       :    Dict		      =      spectrogram(
              SCREAMING_SNAKE_CASE_,     window_function(self.fft_window_size,     'hann'			),     frame_length=self.fft_window_size,     hop_length=self.hop_length,     power=2.0,     mel_filters=SCREAMING_SNAKE_CASE_,     log_mel='dB',     )
          return log_mel_spectrogram.T
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)	->						str:
          UpperCamelCase       :    Optional[int]		      =      np.array_split(list(range(0,     total_frames - chunk_frames + 1			)			),     3			)
          if len(ranges[1]			) == 0:
             # if the audio is too short, we just use the first chunk
             UpperCamelCase       :    Union[str, Any]		      =      [0]
          if len(ranges[2]			) == 0:
             # if the audio is too short, we just use the first chunk
             UpperCamelCase       :    Dict		      =      [0]
          # randomly choose index for each part
          UpperCamelCase       :    int		      =      np.random.choice(ranges[0]			)
          UpperCamelCase       :    List[Any]		      =      np.random.choice(ranges[1]			)
          UpperCamelCase       :    Any		      =      np.random.choice(ranges[2]			)

          UpperCamelCase       :    Union[str, Any]		      =      mel[idx_front : idx_front + chunk_frames, :]
          UpperCamelCase       :    Union[str, Any]		      =      mel[idx_middle : idx_middle + chunk_frames, :]
          UpperCamelCase       :    Optional[int]		      =      mel[idx_back : idx_back + chunk_frames, :]

          UpperCamelCase       :    List[Any]		      =      torch.tensor(mel[None, None, :]			)
          UpperCamelCase       :    List[Any]		      =      torch.nn.functional.interpolate(
              SCREAMING_SNAKE_CASE_,     size=[chunk_frames, 64],     mode='bilinear',     align_corners=SCREAMING_SNAKE_CASE_			)
          UpperCamelCase       :    str		      =      mel_shrink[0][0].numpy()
          UpperCamelCase       :    Dict		      =      np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back],     axis=0			)
          return mel_fusion
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)	->						np.array:
          if waveform.shape[0] > max_length:
             if truncation == "rand_trunc":
                UpperCamelCase       :    Dict		      =      True
                # random crop to max_length (for compatibility) -> this should be handled by self.pad
                UpperCamelCase       :    str		      =      len(SCREAMING_SNAKE_CASE_			) - max_length
                UpperCamelCase       :    Optional[Any]		      =      np.random.randint(0,     overflow + 1			)
                UpperCamelCase       :    int		      =      waveform[idx : idx + max_length]
                UpperCamelCase       :    Tuple		      =      self._np_extract_fbank_features(SCREAMING_SNAKE_CASE_,     self.mel_filters_slaney			)[None, :]
             elif truncation == "fusion":
                UpperCamelCase       :    Tuple		      =      self._np_extract_fbank_features(SCREAMING_SNAKE_CASE_,     self.mel_filters			)
                UpperCamelCase       :    Any		      =      max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                UpperCamelCase       :    List[str]		      =      mel.shape[0]
                if chunk_frames == total_frames:
                   # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
                   # In this case, we just use the whole audio.
                   UpperCamelCase       :    Optional[Any]		      =      np.stack([mel, mel, mel, mel],     axis=0			)
                   UpperCamelCase       :    Optional[int]		      =      False
                else:
                   UpperCamelCase       :    Optional[int]		      =      self._random_mel_fusion(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)
                   UpperCamelCase       :    Any		      =      True
             else:
                raise NotImplementedError(F"""data_truncating {truncation} not implemented"""			)

          else:
             UpperCamelCase       :    Any		      =      False
             # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
             if waveform.shape[0] < max_length:
                if padding == "repeat":
                   UpperCamelCase       :    Any		      =      int(max_length / len(SCREAMING_SNAKE_CASE_			)			)
                   UpperCamelCase       :    Tuple		      =      np.stack(np.tile(SCREAMING_SNAKE_CASE_,     n_repeat + 1			)			)[:max_length]
                if padding == "repeatpad":
                   UpperCamelCase       :    str		      =      int(max_length / len(SCREAMING_SNAKE_CASE_			)			)
                   UpperCamelCase       :    str		      =      np.stack(np.tile(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)			)
                UpperCamelCase       :    List[str]		      =      np.pad(SCREAMING_SNAKE_CASE_,     (0, max_length - waveform.shape[0]),     mode='constant',     constant_values=0			)

             if truncation == "fusion":
                UpperCamelCase       :    Tuple		      =      self._np_extract_fbank_features(SCREAMING_SNAKE_CASE_,     self.mel_filters			)
                UpperCamelCase       :    Tuple		      =      np.stack([input_mel, input_mel, input_mel, input_mel],     axis=0			)
             else:
                UpperCamelCase       :    Optional[int]		      =      self._np_extract_fbank_features(SCREAMING_SNAKE_CASE_,     self.mel_filters_slaney			)[None, :]

          return input_mel, longer






       def __call__(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     SCREAMING_SNAKE_CASE_ = None,     **SCREAMING_SNAKE_CASE_,     )	->						BatchFeature:
          UpperCamelCase       :    Optional[int]		      =      truncation if truncation is not None else self.truncation
          UpperCamelCase       :    List[str]		      =      padding if padding else self.padding

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

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

          if is_batched:
             UpperCamelCase       :    Tuple		      =      [np.asarray(SCREAMING_SNAKE_CASE_,     dtype=np.floataa			) for speech in raw_speech]
          elif not is_batched and not isinstance(SCREAMING_SNAKE_CASE_,     np.ndarray			):
             UpperCamelCase       :    Tuple		      =      np.asarray(SCREAMING_SNAKE_CASE_,     dtype=np.floataa			)
          elif isinstance(SCREAMING_SNAKE_CASE_,     np.ndarray			) and raw_speech.dtype is np.dtype(np.floataa			):
             UpperCamelCase       :    str		      =      raw_speech.astype(np.floataa			)

          # always return batch
          if not is_batched:
             UpperCamelCase       :    Any		      =      [np.asarray(SCREAMING_SNAKE_CASE_			)]

          # convert to mel spectrogram, truncate and pad if needed.
          UpperCamelCase       :    Dict		      =      [
              self._get_input_mel(SCREAMING_SNAKE_CASE_,     max_length if max_length else self.nb_max_samples,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)
              for waveform in raw_speech
          ]

          UpperCamelCase       :    str		      =      []
          UpperCamelCase       :    str		      =      []
          for mel, longer in padded_inputs:
             input_mel.append(SCREAMING_SNAKE_CASE_			)
             is_longer.append(SCREAMING_SNAKE_CASE_			)

          if truncation == "fusion" and sum(SCREAMING_SNAKE_CASE_			) == 0:
             # if no audio is longer than 10s, then randomly select one audio to be longer
             UpperCamelCase       :    Optional[Any]		      =      np.random.randint(0,     len(SCREAMING_SNAKE_CASE_			)			)
             UpperCamelCase       :    List[Any]		      =      True

          if isinstance(input_mel[0],     SCREAMING_SNAKE_CASE_			):
             UpperCamelCase       :    Dict		      =      [np.asarray(SCREAMING_SNAKE_CASE_,     dtype=np.floataa			) for feature in input_mel]

          # is_longer is a list of bool
          UpperCamelCase       :    int		      =      [[longer] for longer in is_longer]

          UpperCamelCase       :    Any		      =      {'input_features': input_mel, 'is_longer': is_longer}
          UpperCamelCase       :    Tuple		      =      BatchFeature(SCREAMING_SNAKE_CASE_			)

          if return_tensors is not None:
             UpperCamelCase       :    List[str]		      =      input_features.convert_to_tensors(SCREAMING_SNAKE_CASE_			)

          return input_features









'''simple docstring'''
from collections import UserDict
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_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
							from ..tf_utils import stable_softmax

_SCREAMING_SNAKE_CASE			=						logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase				)
class   __UpperCAmelCase				(     lowerCAmelCase				):



					'''simple docstring'''





					def __init__(	self     :						int      ,       **_lowercase     :						List[Any])    ->							Union[str, Any]:
								super().__init__(**_lowercase)

								requires_backends(self      ,       'vision')
								self.check_model_type(
								    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
								    if self.framework == 'tf'
								    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING)



					def __call__(	self     :						Optional[Any]      ,       _lowercase     :						Union[str, List[str], "Image", List["Image"]]      ,       **_lowercase     :						List[Any])    ->							Any:
								return super().__call__(_lowercase      ,       **_lowercase)



					def     __snake_case						(	self     :						int      ,       **_lowercase     :						Union[str, Any])    ->							Any:
								A_									= {}
								if "candidate_labels" in kwargs:
											A_									= kwargs['candidate_labels']
								if "hypothesis_template" in kwargs:
											A_									= kwargs['hypothesis_template']

								return preprocess_params, {}, {}



					def     __snake_case						(	self     :						List[str]      ,       _lowercase     :						Tuple      ,       _lowercase     :						Any=None      ,       _lowercase     :						Optional[int]="This is a photo of {}.")    ->							Union[str, Any]:
								A_									= load_image(_lowercase)
								A_									= self.image_processor(images=[image]      ,       return_tensors=self.framework)
								A_									= candidate_labels
								A_									= [hypothesis_template.format(_lowercase) for x in candidate_labels]
								A_									= self.tokenizer(_lowercase      ,       return_tensors=self.framework      ,       padding=_lowercase)
								A_									= [text_inputs]
								return inputs



					def     __snake_case						(	self     :						Optional[int]      ,       _lowercase     :						Tuple)    ->							Optional[int]:
								A_									= model_inputs.pop('candidate_labels')
								A_									= model_inputs.pop('text_inputs')
								if isinstance(text_inputs[0]      ,       _lowercase):
											A_									= text_inputs[0]
								else:
											# Batching case.
											A_									= text_inputs[0][0]

								A_									= self.model(**_lowercase      ,       **_lowercase)

								A_									= {
								    'candidate_labels': candidate_labels,
								    'logits': outputs.logits_per_image,
								}
								return model_outputs



					def     __snake_case						(	self     :						List[str]      ,       _lowercase     :						int)    ->							Optional[int]:
								A_									= model_outputs.pop('candidate_labels')
								A_									= model_outputs['logits'][0]
								if self.framework == "pt":
											A_									= logits.softmax(dim=-1).squeeze(-1)
											A_									= probs.tolist()
											if not isinstance(_lowercase      ,       _lowercase):
														A_									= [scores]
								elif self.framework == "tf":
											A_									= stable_softmax(_lowercase      ,       axis=-1)
											A_									= probs.numpy().tolist()
								else:
											raise ValueError(F'Unsupported framework: {self.framework}')

								A_									= [
								    {'score': score, 'label': candidate_label}
								    for score, candidate_label in sorted(zip(_lowercase      ,       _lowercase)      ,       key=lambda _lowercase: -x[0])
								]
								return result




