






'''simple docstring'''







from ...configuration_utils import PretrainedConfig
from ...utils import logging


_A	:			Dict         =logging.get_logger(__name__)

_A	:			Any         ={
    '''google/pegasus-large''': '''https://huggingface.co/google/pegasus-large/resolve/main/config.json''',
    # See all PEGASUS models at https://huggingface.co/models?filter=pegasus
}

class _lowercase	(				_lowercase		):
							a				      =       """pegasus"""
							a				      =       ["""past_key_values"""]
							a				      =       {"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}




							def __init__(							self:				Optional[Any]			,							UpperCamelCase__:				Dict=50_265			,							UpperCamelCase__:				Tuple=1_024			,							UpperCamelCase__:				Dict=12			,							UpperCamelCase__:				List[Any]=4_096			,							UpperCamelCase__:				Any=16			,							UpperCamelCase__:				Optional[int]=12			,							UpperCamelCase__:				Optional[Any]=4_096			,							UpperCamelCase__:				List[str]=16			,							UpperCamelCase__:				List[str]=0.0			,							UpperCamelCase__:				List[Any]=0.0			,							UpperCamelCase__:				Tuple=True			,							UpperCamelCase__:				Optional[Any]=True			,							UpperCamelCase__:				Dict="gelu"			,							UpperCamelCase__:				Optional[Any]=1_024			,							UpperCamelCase__:				int=0.1			,							UpperCamelCase__:				List[Any]=0.0			,							UpperCamelCase__:				List[Any]=0.0			,							UpperCamelCase__:				Union[str, Any]=0.02			,							UpperCamelCase__:				List[Any]=0			,							UpperCamelCase__:				Any=False			,							UpperCamelCase__:				Any=0			,							UpperCamelCase__:				str=1			,							UpperCamelCase__:				Union[str, Any]=1			,							**UpperCamelCase__:				Any			,							):
														lowerCamelCase__					:	int         =       vocab_size
														lowerCamelCase__					:	Union[str, Any]         =       max_position_embeddings
														lowerCamelCase__					:	List[Any]         =       d_model
														lowerCamelCase__					:	Union[str, Any]         =       encoder_ffn_dim
														lowerCamelCase__					:	Dict         =       encoder_layers
														lowerCamelCase__					:	Optional[int]         =       encoder_attention_heads
														lowerCamelCase__					:	Union[str, Any]         =       decoder_ffn_dim
														lowerCamelCase__					:	Tuple         =       decoder_layers
														lowerCamelCase__					:	Optional[int]         =       decoder_attention_heads
														lowerCamelCase__					:	Any         =       dropout
														lowerCamelCase__					:	str         =       attention_dropout
														lowerCamelCase__					:	str         =       activation_dropout
														lowerCamelCase__					:	Dict         =       activation_function
														lowerCamelCase__					:	int         =       init_std
														lowerCamelCase__					:	Union[str, Any]         =       encoder_layerdrop
														lowerCamelCase__					:	List[Any]         =       decoder_layerdrop
														lowerCamelCase__					:	List[str]         =       use_cache
														lowerCamelCase__					:	Optional[int]         =       encoder_layers
														lowerCamelCase__					:	List[Any]         =       scale_embedding  # scale factor will be sqrt(d_model) if True
														super().__init__(
														    pad_token_id=UpperCamelCase__			,							eos_token_id=UpperCamelCase__			,							is_encoder_decoder=UpperCamelCase__			,							decoder_start_token_id=UpperCamelCase__			,							forced_eos_token_id=UpperCamelCase__			,							**UpperCamelCase__			,							)




							@property
							def 		lowerCamelCase_				(							self:				str     ):
														return self.encoder_attention_heads




							@property
							def 		lowerCamelCase_				(							self:				str     ):
														return self.d_model






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_:Dict             =						logging.get_logger(__name__)






class SCREAMING_SNAKE_CASE__     (		SCREAMING_SNAKE_CASE__    ):




    '''simple docstring'''






    __lowerCamelCase			: Union[str, Any]					=      ["input_features", "is_longer"]






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






    def 			_lowerCAmelCase					(       self		):
        A :       Any          =   copy.deepcopy(self.__dict__		)
        A :       Tuple          =   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 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__ = None		):
        A :       Optional[Any]          =   spectrogram(
            lowerCamelCase__,		window_function(self.fft_window_size,		"""hann"""		),		frame_length=self.fft_window_size,		hop_length=self.hop_length,		power=2.0,		mel_filters=lowerCamelCase__,		log_mel="""dB""",		)
        return log_mel_spectrogram.T






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

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

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






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

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

            if truncation == "fusion":
                A :       Union[str, Any]          =   self._np_extract_fbank_features(lowerCamelCase__,		self.mel_filters		)
                A :       Tuple          =   np.stack([input_mel, input_mel, input_mel, input_mel],		axis=0		)
            else:
                A :       List[str]          =   self._np_extract_fbank_features(lowerCamelCase__,		self.mel_filters_slaney		)[None, :]

        return input_mel, longer






    def __call__(       self,		lowerCamelCase__,		lowerCamelCase__ = None,		lowerCamelCase__ = None,		lowerCamelCase__ = None,		lowerCamelCase__ = None,		lowerCamelCase__ = None,		**lowerCamelCase__,		):
        A :       Any          =   truncation if truncation is not None else self.truncation
        A :       Union[str, Any]          =   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."""		)

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

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

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

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

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

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

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

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

        A :       Union[str, Any]          =   {"""input_features""": input_mel, """is_longer""": is_longer}
        A :       Tuple          =   BatchFeature(lowerCamelCase__		)

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

        return input_features


