





'''simple docstring'''


from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


lowerCAmelCase :					Union[str, Any] 							=logging.get_logger(__name__)
@add_end_docstrings(_lowerCAmelCase							)
class 				a_       (    _lowerCAmelCase							):





     def __init__(			self		: str     ,				**lowercase		: str						):





        """simple docstring"""




        super().__init__(**lowercase						)

        if self.framework != "pt":
           raise ValueError(F'The {self.__class__} is only available in PyTorch.'						)
        # No specific FOR_XXX available yet





     def __call__(			self		: Any     ,				lowercase		: Union[np.ndarray, bytes, str]     ,				**lowercase		: Dict						):





        """simple docstring"""




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





     def 							lowercase__				(			self		: Dict     ,				**lowercase		: int						):





        """simple docstring"""




        lowercase_      :str         =			{}
        if "candidate_labels" in kwargs:
           lowercase_      :Tuple         =			kwargs["candidate_labels"]
        if "hypothesis_template" in kwargs:
           lowercase_      :Optional[Any]         =			kwargs["hypothesis_template"]

        return preprocess_params, {}, {}





     def 							lowercase__				(			self		: Dict     ,				lowercase		: Any     ,				lowercase		: Dict=None     ,				lowercase		: Optional[Any]="This is a sound of {}."						):





        """simple docstring"""




        if isinstance(lowercase     ,				lowercase						):
           if audio.startswith("http://"						) or audio.startswith("https://"						):
              # We need to actually check for a real protocol, otherwise it's impossible to use a local file
              # like http_huggingface_co.png
              lowercase_      :Union[str, Any]         =			requests.get(lowercase						).content
           else:
              with open(lowercase     ,				"rb"						) as f:
                 lowercase_      :Dict         =			f.read()

        if isinstance(lowercase     ,				lowercase						):
           lowercase_      :str         =			ffmpeg_read(lowercase     ,				self.feature_extractor.sampling_rate						)

        if not isinstance(lowercase     ,				np.ndarray						):
           raise ValueError("We expect a numpy ndarray as input"						)
        if len(audio.shape						) != 1:
           raise ValueError("We expect a single channel audio input for ZeroShotAudioClassificationPipeline"						)

        lowercase_      :Dict         =			self.feature_extractor(
            [audio]     ,				sampling_rate=self.feature_extractor.sampling_rate     ,				return_tensors="pt"						)
        lowercase_      :Optional[Any]         =			candidate_labels
        lowercase_      :Optional[int]         =			[hypothesis_template.format(lowercase						) for x in candidate_labels]
        lowercase_      :List[str]         =			self.tokenizer(lowercase     ,				return_tensors=self.framework     ,				padding=lowercase						)
        lowercase_      :Any         =			[text_inputs]
        return inputs





     def 							lowercase__				(			self		: Any     ,				lowercase		: str						):





        """simple docstring"""




        lowercase_      :int         =			model_inputs.pop("candidate_labels"						)
        lowercase_      :Optional[int]         =			model_inputs.pop("text_inputs"						)
        if isinstance(text_inputs[0]     ,				lowercase						):
           lowercase_      :int         =			text_inputs[0]
        else:
           # Batching case.
           lowercase_      :int         =			text_inputs[0][0]

        lowercase_      :List[Any]         =			self.model(**lowercase     ,				**lowercase						)

        lowercase_      :Optional[Any]         =			{
            "candidate_labels": candidate_labels,
            "logits": outputs.logits_per_audio,
        }
        return model_outputs







     def 							lowercase__				(			self		: Any     ,				lowercase		: List[str]						):





        """simple docstring"""




        lowercase_      :List[Any]         =			model_outputs.pop("candidate_labels"						)
        lowercase_      :List[str]         =			model_outputs["logits"][0]

        if self.framework == "pt":
           lowercase_      :Dict         =			logits.softmax(dim=0						)
           lowercase_      :Dict         =			probs.tolist()
        else:
           raise ValueError("`tf` framework not supported."						)

        lowercase_      :str         =			[
            {"score": score, "label": candidate_label}
            for score, candidate_label in sorted(zip(lowercase     ,				lowercase						)     ,				key=lambda lowercase						: -x[0]						)
        ]
        return result









'''simple docstring'''


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 :					Optional[int] 							=logging.get_logger(__name__)
class 				a_       (    _lowerCAmelCase							):
     __A								=      ["input_features"]





     def __init__(			self		: Any     ,				lowercase		: Tuple=80     ,				lowercase		: Optional[int]=16_000     ,				lowercase		: Optional[Any]=160     ,				lowercase		: Optional[int]=30     ,				lowercase		: List[Any]=400     ,				lowercase		: Dict=0.0     ,				lowercase		: Tuple=False     ,				**lowercase		: Optional[int]     ,				):





        """simple docstring"""




        super().__init__(
            feature_size=lowercase     ,				sampling_rate=lowercase     ,				padding_value=lowercase     ,				return_attention_mask=lowercase     ,				**lowercase     ,				)
        lowercase_      :Optional[int]         =			n_fft
        lowercase_      :List[Any]         =			hop_length
        lowercase_      :Tuple         =			chunk_length
        lowercase_      :List[str]         =			chunk_length * sampling_rate
        lowercase_      :Optional[Any]         =			self.n_samples // hop_length
        lowercase_      :Any         =			sampling_rate
        lowercase_      :List[Any]         =			mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2     ,				num_mel_filters=lowercase     ,				min_frequency=0.0     ,				max_frequency=80_00.0     ,				sampling_rate=lowercase     ,				norm="slaney"     ,				mel_scale="slaney"     ,				)





     def 							lowercase__				(			self		: str     ,				lowercase		: np.array						):





        """simple docstring"""




        lowercase_      :Any         =			spectrogram(
            lowercase     ,				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"     ,				)
        lowercase_      :Any         =			log_spec[:, :-1]
        lowercase_      :List[Any]         =			np.maximum(lowercase     ,				log_spec.max() - 8.0						)
        lowercase_      :Dict         =			(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 							lowercase__				(			lowercase		: List[np.ndarray]     ,				lowercase		: List[np.ndarray]     ,				lowercase		: float = 0.0						):





        """simple docstring"""




        if attention_mask is not None:
           lowercase_      :Optional[int]         =			np.array(lowercase     ,				np.intaa						)
           lowercase_      :Any         =			[]

           for vector, length in zip(lowercase     ,				attention_mask.sum(-1						)						):
              lowercase_      :Dict         =			(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7						)
              if length < normed_slice.shape[0]:
                 lowercase_      :List[Any]         =			padding_value

              normed_input_values.append(lowercase						)
        else:
           lowercase_      :List[Any]         =			[(x - x.mean()) / np.sqrt(x.var() + 1e-7						) for x in input_values]

        return normed_input_values





     def __call__(			self		: Tuple     ,				lowercase		: Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]     ,				lowercase		: bool = True     ,				lowercase		: Optional[int] = None     ,				lowercase		: Optional[Union[str, TensorType]] = None     ,				lowercase		: Optional[bool] = None     ,				lowercase		: Optional[str] = "max_length"     ,				lowercase		: Optional[int] = None     ,				lowercase		: Optional[int] = None     ,				lowercase		: Optional[bool] = None     ,				**lowercase		: Union[str, Any]     ,				):





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

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

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

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

        lowercase_      :int         =			BatchFeature({"input_features": raw_speech}						)

        # convert into correct format for padding

        lowercase_      :Tuple         =			self.pad(
            lowercase     ,				padding=lowercase     ,				max_length=max_length if max_length else self.n_samples     ,				truncation=lowercase     ,				pad_to_multiple_of=lowercase     ,				return_attention_mask=return_attention_mask or do_normalize     ,				)

        # zero-mean and unit-variance normalization
        if do_normalize:
           lowercase_      :Union[str, Any]         =			self.zero_mean_unit_var_norm(
               padded_inputs["input_features"]     ,				attention_mask=padded_inputs["attention_mask"]     ,				padding_value=self.padding_value     ,				)
           lowercase_      :List[Any]         =			np.stack(padded_inputs["input_features"]     ,				axis=0						)

        # make sure list is in array format
        lowercase_      :Union[str, Any]         =			padded_inputs.get("input_features"						).transpose(2     ,				0     ,				1						)

        lowercase_      :List[str]         =			[self._np_extract_fbank_features(lowercase						) for waveform in input_features[0]]

        if isinstance(input_features[0]     ,				lowercase						):
           lowercase_      :Tuple         =			[np.asarray(lowercase     ,				dtype=np.floataa						) for feature in input_features]
        else:
           lowercase_      :Union[str, Any]         =			input_features

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

        if return_tensors is not None:
           lowercase_      :Tuple         =			padded_inputs.convert_to_tensors(lowercase						)

        return padded_inputs







     def 							lowercase__				(			self		: List[str]						):





        """simple docstring"""




        lowercase_      :Union[str, Any]         =			copy.deepcopy(self.__dict__						)
        lowercase_      :List[str]         =			self.__class__.__name__
        if "mel_filters" in output:
           del output["mel_filters"]
        return output


