



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


A:   int								=		logging.get_logger(__name__)



class       SCREAMING_SNAKE_CASE__  (					UpperCAmelCase__ ):
   __lowerCAmelCase  :   str						=						['input_features']

   def __init__(   self      ,     _SCREAMING_SNAKE_CASE=80      ,     _SCREAMING_SNAKE_CASE=16000      ,     _SCREAMING_SNAKE_CASE=160      ,     _SCREAMING_SNAKE_CASE=30      ,     _SCREAMING_SNAKE_CASE=400      ,     _SCREAMING_SNAKE_CASE=0.0      ,     _SCREAMING_SNAKE_CASE=False      ,     **_SCREAMING_SNAKE_CASE      ,     )  ->      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      ,     )
    UpperCAmelCase	:					int	=	n_fft
    UpperCAmelCase	:					str	=	hop_length
    UpperCAmelCase	:					Tuple	=	chunk_length
    UpperCAmelCase	:					Optional[Any]	=	chunk_length * sampling_rate
    UpperCAmelCase	:					str	=	self.n_samples // hop_length
    UpperCAmelCase	:					Any	=	sampling_rate
    UpperCAmelCase	:					List[Any]	=	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 	SCREAMING_SNAKE_CASE     (   self      ,     _SCREAMING_SNAKE_CASE						)  ->      np.ndarray:





    '''simple docstring'''




    UpperCAmelCase	:					Union[str, Any]	=	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"""      ,     )
    UpperCAmelCase	:					List[str]	=	log_spec[:, :-1]
    UpperCAmelCase	:					Any	=	np.maximum(_SCREAMING_SNAKE_CASE      ,     log_spec.max() - 8.0						)
    UpperCAmelCase	:					List[Any]	=	(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 	SCREAMING_SNAKE_CASE     (   _SCREAMING_SNAKE_CASE      ,     _SCREAMING_SNAKE_CASE      ,     _SCREAMING_SNAKE_CASE = 0.0						)  ->      List[np.ndarray]:





    '''simple docstring'''




    if attention_mask is not None:
     UpperCAmelCase	:					Optional[Any]	=	np.array(_SCREAMING_SNAKE_CASE      ,     np.intaa						)
     UpperCAmelCase	:					List[str]	=	[]

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

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

    return normed_input_values

   def __call__(   self      ,     _SCREAMING_SNAKE_CASE      ,     _SCREAMING_SNAKE_CASE = True      ,     _SCREAMING_SNAKE_CASE = None      ,     _SCREAMING_SNAKE_CASE = None      ,     _SCREAMING_SNAKE_CASE = None      ,     _SCREAMING_SNAKE_CASE = "max_length"      ,     _SCREAMING_SNAKE_CASE = None      ,     _SCREAMING_SNAKE_CASE = None      ,     _SCREAMING_SNAKE_CASE = None      ,     **_SCREAMING_SNAKE_CASE      ,     )  ->      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."""						)

    UpperCAmelCase	:					Tuple	=	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	:					Optional[Any]	=	is_batched_numpy or (
        isinstance(_SCREAMING_SNAKE_CASE      ,     (list, tuple)						) and (isinstance(raw_speech[0]      ,     (np.ndarray, tuple, list)						))
    )

    if is_batched:
     UpperCAmelCase	:					Any	=	[np.asarray([speech]      ,     dtype=np.floataa						).T for speech in raw_speech]
    elif not is_batched and not isinstance(_SCREAMING_SNAKE_CASE      ,     np.ndarray						):
     UpperCAmelCase	:					List[str]	=	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	:					List[Any]	=	raw_speech.astype(np.floataa						)

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

    UpperCAmelCase	:					Union[str, Any]	=	BatchFeature({"""input_features""": raw_speech}						)

    # convert into correct format for padding

    UpperCAmelCase	:					Optional[int]	=	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:
     UpperCAmelCase	:					Any	=	self.zero_mean_unit_var_norm(
         padded_inputs["""input_features"""]      ,     attention_mask=padded_inputs["""attention_mask"""]      ,     padding_value=self.padding_value      ,     )
     UpperCAmelCase	:					Tuple	=	np.stack(padded_inputs["""input_features"""]      ,     axis=0						)

    # make sure list is in array format
    UpperCAmelCase	:					Optional[int]	=	padded_inputs.get("""input_features"""						).transpose(2      ,     0      ,     1						)

    UpperCAmelCase	:					int	=	[self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE						) for waveform in input_features[0]]

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

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

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

    return padded_inputs




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





    '''simple docstring'''




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





"""simple docstring"""







# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import numpy as np
import torch

from ..models.clipseg import CLIPSegForImageSegmentation
from ..utils import is_vision_available, requires_backends
from .base import PipelineTool


if is_vision_available():
     from PIL import Image







class       UpperCamelCase__   (	lowercase_  ):




      """simple docstring"""




      _SCREAMING_SNAKE_CASE  				=     (
          """This is a tool that creates a segmentation mask of an image according to a label. It cannot create an image."""
          """It takes two arguments named `image` which should be the original image, and `label` which should be a text """
          """describing the elements what should be identified in the segmentation mask. The tool returns the mask."""
      )
      _SCREAMING_SNAKE_CASE  				=     """CIDAS/clipseg-rd64-refined"""
      _SCREAMING_SNAKE_CASE  				=     """image_segmenter"""
      _SCREAMING_SNAKE_CASE  				=     CLIPSegForImageSegmentation

      _SCREAMING_SNAKE_CASE  				=     ["""image""", """text"""]
      _SCREAMING_SNAKE_CASE  				=     ["""image"""]



      def __init__(      self				:      Dict    ,			*SCREAMING_SNAKE_CASE_				:      Tuple    ,			**SCREAMING_SNAKE_CASE_				:      Optional[int]       ):
        requires_backends(self    ,			['vision']       )
        super().__init__(*SCREAMING_SNAKE_CASE_    ,			**SCREAMING_SNAKE_CASE_       )



      def      SCREAMING_SNAKE_CASE__ (      self				:      Optional[int]    ,			SCREAMING_SNAKE_CASE_				:      "Image"    ,			SCREAMING_SNAKE_CASE_				:      str       ):
        return self.pre_processor(text=[label]    ,			images=[image]    ,			padding=SCREAMING_SNAKE_CASE_    ,			return_tensors='pt'       )



      def      SCREAMING_SNAKE_CASE__ (      self				:      List[str]    ,			SCREAMING_SNAKE_CASE_				:      int       ):
        with torch.no_grad():
          lowerCAmelCase_   :				List[str]								= self.model(**SCREAMING_SNAKE_CASE_       ).logits
        return logits





      def      SCREAMING_SNAKE_CASE__ (      self				:      Optional[int]    ,			SCREAMING_SNAKE_CASE_				:      Optional[Any]       ):
        lowerCAmelCase_   :				Dict								= outputs.cpu().detach().numpy()
        lowerCAmelCase_   :				Optional[Any]								= 0
        lowerCAmelCase_   :				Optional[Any]								= 1
        return Image.fromarray((array * 2_5_5).astype(np.uinta       )       )





