




'''simple docstring'''





import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModelWithProjection, PreTrainedModel

from ...utils import logging


_SCREAMING_SNAKE_CASE							:   Tuple      		= logging.get_logger(__name__)




class  _snake_case		(	lowercase_							):
			lowerCAmelCase_     :      Dict						       =    CLIPConfig

			lowerCAmelCase_     :      Dict						       =    ["CLIPEncoderLayer"]




			def __init__(					self				,							a__							)				->				Dict:

										'''simple docstring'''




										super().__init__(a__							)

										snake_case_						=      CLIPVisionModelWithProjection(config.vision_config							)

										snake_case_						=      nn.Linear(config.vision_config.projection_dim				,							1							)
										snake_case_						=      nn.Linear(config.vision_config.projection_dim				,							1							)




			@torch.no_grad()
			def    lowerCAmelCase__				(					self				,							a__				,							a__				,							a__=0.5				,							a__=0.5							)				->				Any:

										'''simple docstring'''




										snake_case_						=      self.vision_model(a__							)[0]

										snake_case_						=      self.p_head(a__							)
										snake_case_						=      nsfw_detected.flatten()
										snake_case_						=      nsfw_detected > p_threshold
										snake_case_						=      nsfw_detected.tolist()

										if any(a__							):
																	logger.warning(
																	    "Potential NSFW content was detected in one or more images. A black image will be returned instead."
																	    " Try again with a different prompt and/or seed."							)

										for idx, nsfw_detected_ in enumerate(a__							):
																	if nsfw_detected_:
																								snake_case_						=      np.zeros(images[idx].shape							)

										snake_case_						=      self.w_head(a__							)
										snake_case_						=      watermark_detected.flatten()
										snake_case_						=      watermark_detected > w_threshold
										snake_case_						=      watermark_detected.tolist()

										if any(a__							):
																	logger.warning(
																	    "Potential watermarked content was detected in one or more images. A black image will be returned instead."
																	    " Try again with a different prompt and/or seed."							)

										for idx, watermark_detected_ in enumerate(a__							):
																	if watermark_detected_:
																								snake_case_						=      np.zeros(images[idx].shape							)

										return images, nsfw_detected, watermark_detected





'''simple docstring'''




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


__SCREAMING_SNAKE_CASE       :     List[str]     =  logging.get_logger(__name__)

class 	lowerCamelCase_	(snake_case__    ):
      '''simple docstring'''

      __UpperCamelCase:  Optional[int]						=						["input_features", "is_longer"]




      def __init__(       self						:				str						,					A						:				int=64						,					A						:				Dict=48000						,					A						:				str=480						,					A						:				List[Any]=10						,					A						:				Optional[Any]=1024						,					A						:				Tuple=0.0						,					A						:				List[Any]=False						,					A						:				float = 0						,					A						:				float = 14000						,					A						:				int = None						,					A						:				str = "fusion"						,					A						:				str = "repeatpad"						,					**A						:				Dict						,					):
             super().__init__(
                 feature_size=A						,					sampling_rate=A						,					padding_value=A						,					return_attention_mask=A						,					**A						,					)
             _UpperCAmelCase     :  Optional[Any]							      =    top_db
             _UpperCAmelCase     :  Dict							      =    truncation
             _UpperCAmelCase     :  List[Any]							      =    padding
             _UpperCAmelCase     :  Optional[Any]							      =    fft_window_size
             _UpperCAmelCase     :  Dict							      =    (fft_window_size >> 1) + 1
             _UpperCAmelCase     :  Any							      =    hop_length
             _UpperCAmelCase     :  Tuple							      =    max_length_s
             _UpperCAmelCase     :  str							      =    max_length_s * sampling_rate
             _UpperCAmelCase     :  Any							      =    sampling_rate
             _UpperCAmelCase     :  Optional[int]							      =    frequency_min
             _UpperCAmelCase     :  str							      =    frequency_max
             _UpperCAmelCase     :  Union[str, Any]							      =    mel_filter_bank(
                 num_frequency_bins=self.nb_frequency_bins						,					num_mel_filters=A						,					min_frequency=A						,					max_frequency=A						,					sampling_rate=A						,					norm=A						,					mel_scale="htk"						,					)
             _UpperCAmelCase     :  Tuple							      =    mel_filter_bank(
                 num_frequency_bins=self.nb_frequency_bins						,					num_mel_filters=A						,					min_frequency=A						,					max_frequency=A						,					sampling_rate=A						,					norm="slaney"						,					mel_scale="slaney"						,					)




      def      _A	(       self						:				List[str]       ):
             _UpperCAmelCase     :  Union[str, Any]							      =    copy.deepcopy(self.__dict__       )
             _UpperCAmelCase     :  Dict							      =    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      _A	(       self						:				Optional[Any]						,					A						:				np.array						,					A						:				Optional[np.array] = None       ):
             _UpperCAmelCase     :  Dict							      =    spectrogram(
                 A						,					window_function(self.fft_window_size						,					"hann"       )						,					frame_length=self.fft_window_size						,					hop_length=self.hop_length						,					power=2.0						,					mel_filters=A						,					log_mel="dB"						,					)
             return log_mel_spectrogram.T




      def      _A	(       self						:				str						,					A						:				str						,					A						:				List[str]						,					A						:				List[Any]       ):
             _UpperCAmelCase     :  List[str]							      =    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     :  Optional[Any]							      =    [0]
             if len(ranges[2]       ) == 0:
                    # if the audio is too short, we just use the first chunk
                    _UpperCAmelCase     :  Tuple							      =    [0]
             # randomly choose index for each part
             _UpperCAmelCase     :  Dict							      =    np.random.choice(ranges[0]       )
             _UpperCAmelCase     :  str							      =    np.random.choice(ranges[1]       )
             _UpperCAmelCase     :  Tuple							      =    np.random.choice(ranges[2]       )

             _UpperCAmelCase     :  str							      =    mel[idx_front : idx_front + chunk_frames, :]
             _UpperCAmelCase     :  str							      =    mel[idx_middle : idx_middle + chunk_frames, :]
             _UpperCAmelCase     :  List[Any]							      =    mel[idx_back : idx_back + chunk_frames, :]

             _UpperCAmelCase     :  Dict							      =    torch.tensor(mel[None, None, :]       )
             _UpperCAmelCase     :  Optional[Any]							      =    torch.nn.functional.interpolate(
                 A						,					size=[chunk_frames, 64]						,					mode="bilinear"						,					align_corners=A       )
             _UpperCAmelCase     :  List[str]							      =    mel_shrink[0][0].numpy()
             _UpperCAmelCase     :  str							      =    np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]						,					axis=0       )
             return mel_fusion




      def      _A	(       self						:				List[Any]						,					A						:				np.array						,					A						:				List[str]						,					A						:				Any						,					A						:				Optional[int]       ):
             if waveform.shape[0] > max_length:
                    if truncation == "rand_trunc":
                           _UpperCAmelCase     :  int							      =    True
                           # random crop to max_length (for compatibility) -> this should be handled by self.pad
                           _UpperCAmelCase     :  str							      =    len(A       ) - max_length
                           _UpperCAmelCase     :  str							      =    np.random.randint(0						,					overflow + 1       )
                           _UpperCAmelCase     :  int							      =    waveform[idx : idx + max_length]
                           _UpperCAmelCase     :  Any							      =    self._np_extract_fbank_features(A						,					self.mel_filters_slaney       )[None, :]
                    elif truncation == "fusion":
                           _UpperCAmelCase     :  Tuple							      =    self._np_extract_fbank_features(A						,					self.mel_filters       )
                           _UpperCAmelCase     :  List[str]							      =    max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                           _UpperCAmelCase     :  Optional[Any]							      =    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     :  Any							      =    np.stack([mel, mel, mel, mel]						,					axis=0       )
                                  _UpperCAmelCase     :  int							      =    False
                           else:
                                  _UpperCAmelCase     :  Tuple							      =    self._random_mel_fusion(A						,					A						,					A       )
                                  _UpperCAmelCase     :  Any							      =    True
                    else:
                           raise NotImplementedError(F"""data_truncating {truncation} not implemented"""       )

             else:
                    _UpperCAmelCase     :  Optional[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     :  str							      =    int(max_length / len(A       )       )
                                  _UpperCAmelCase     :  Dict							      =    np.stack(np.tile(A						,					n_repeat + 1       )       )[:max_length]
                           if padding == "repeatpad":
                                  _UpperCAmelCase     :  Dict							      =    int(max_length / len(A       )       )
                                  _UpperCAmelCase     :  List[str]							      =    np.stack(np.tile(A						,					A       )       )
                           _UpperCAmelCase     :  Optional[Any]							      =    np.pad(A						,					(0, max_length - waveform.shape[0])						,					mode="constant"						,					constant_values=0       )

                    if truncation == "fusion":
                           _UpperCAmelCase     :  str							      =    self._np_extract_fbank_features(A						,					self.mel_filters       )
                           _UpperCAmelCase     :  Optional[int]							      =    np.stack([input_mel, input_mel, input_mel, input_mel]						,					axis=0       )
                    else:
                           _UpperCAmelCase     :  List[str]							      =    self._np_extract_fbank_features(A						,					self.mel_filters_slaney       )[None, :]

             return input_mel, longer





      def __call__(       self						:				Union[str, Any]						,					A						:				Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]						,					A						:				str = None						,					A						:				Optional[str] = None						,					A						:				Optional[int] = None						,					A						:				Optional[int] = None						,					A						:				Optional[Union[str, TensorType]] = None						,					**A						:				List[str]						,					):
             _UpperCAmelCase     :  int							      =    truncation if truncation is not None else self.truncation
             _UpperCAmelCase     :  Optional[int]							      =    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     :  Any							      =    isinstance(A						,					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(A						,					(list, tuple)       ) and (isinstance(raw_speech[0]						,					(np.ndarray, tuple, list)       ))
             )

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

             # always return batch
             if not is_batched:
                    _UpperCAmelCase     :  List[str]							      =    [np.asarray(A       )]

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

             _UpperCAmelCase     :  int							      =    []
             _UpperCAmelCase     :  Optional[Any]							      =    []
             for mel, longer in padded_inputs:
                    input_mel.append(A       )
                    is_longer.append(A       )

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

             if isinstance(input_mel[0]						,					A       ):
                    _UpperCAmelCase     :  List[str]							      =    [np.asarray(A						,					dtype=np.floataa       ) for feature in input_mel]

             # is_longer is a list of bool
             _UpperCAmelCase     :  Tuple							      =    [[longer] for longer in is_longer]

             _UpperCAmelCase     :  Optional[Any]							      =    {"input_features": input_mel, "is_longer": is_longer}
             _UpperCAmelCase     :  Tuple							      =    BatchFeature(A       )

             if return_tensors is not None:
                    _UpperCAmelCase     :  List[Any]							      =    input_features.convert_to_tensors(A       )

             return input_features

