
"""simple docstring"""



from math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


a					:	Tuple							=		logging.get_logger(__name__)




class 		a_						( _UpperCAmelCase	):
    a      :				Tuple     =				['audio_values', 'audio_mask']







    def __init__(			self			:       Optional[Any]     ,					__UpperCamelCase			:       Optional[Any]=20_48     ,					__UpperCamelCase			:       Any=1     ,					__UpperCamelCase			:       Any=[16, 16]     ,					__UpperCamelCase			:       Optional[int]=1_28     ,					__UpperCamelCase			:       Union[str, Any]=4_41_00     ,					__UpperCamelCase			:       List[str]=86     ,					__UpperCamelCase			:       int=20_48     ,					__UpperCamelCase			:       Optional[Any]=0.0     ,					**__UpperCamelCase			:       Dict     ,					)		->Dict:






        '''simple docstring'''






        super().__init__(
            feature_size=__UpperCamelCase     ,					sampling_rate=__UpperCamelCase     ,					padding_value=__UpperCamelCase     ,					**__UpperCamelCase     ,					)

        _UpperCAmelCase          =   spectrogram_length
        _UpperCAmelCase          =   num_channels
        _UpperCAmelCase          =   patch_size
        _UpperCAmelCase          =   feature_size // self.patch_size[1]
        _UpperCAmelCase          =   n_fft
        _UpperCAmelCase          =   sampling_rate // hop_length_to_sampling_rate
        _UpperCAmelCase          =   sampling_rate
        _UpperCAmelCase          =   padding_value
        _UpperCAmelCase          =   mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2     ,					num_mel_filters=__UpperCamelCase     ,					min_frequency=0.0     ,					max_frequency=2_20_50.0     ,					sampling_rate=__UpperCamelCase     ,					norm="""slaney"""     ,					mel_scale="""slaney"""     ,					).T







    def      _snake_case					(			self			:       Dict     ,					__UpperCamelCase			:       np.array     )		->np.ndarray:






        '''simple docstring'''






        _UpperCAmelCase          =   spectrogram(
            __UpperCamelCase     ,					window_function(self.n_fft     ,					"""hann"""     )     ,					frame_length=self.n_fft     ,					hop_length=self.hop_length     ,					power=2.0     ,					mel_filters=self.mel_filters.T     ,					log_mel="""dB"""     ,					db_range=80.0     ,					)
        _UpperCAmelCase          =   log_spec[:, :-1]
        _UpperCAmelCase          =   log_spec - 20.0
        _UpperCAmelCase          =   np.clip(log_spec / 40.0     ,					-2.0     ,					0.0     ) + 1.0
        return log_spec







    def __call__(			self			:       Optional[Any]     ,					__UpperCamelCase			:       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]     ,					__UpperCamelCase			:       Optional[Union[str, TensorType]] = None     ,					__UpperCamelCase			:       Optional[bool] = True     ,					__UpperCamelCase			:       Optional[int] = None     ,					__UpperCamelCase			:       bool = False     ,					__UpperCamelCase			:       bool = False     ,					**__UpperCamelCase			:       str     ,					)		->BatchFeature:






        '''simple docstring'''






        if sampling_rate is not None:
            if sampling_rate != self.sampling_rate:
                raise ValueError(
                    """This feature extractor is set to support sampling rate"""
                    f""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
                    f""" 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          =   isinstance(__UpperCamelCase     ,					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          =   is_batched_numpy or (
            isinstance(__UpperCamelCase     ,					(list, tuple)     ) and (isinstance(raw_speech[0]     ,					(np.ndarray, tuple, list)     ))
        )
        if is_batched:
            _UpperCAmelCase          =   [np.asarray([speech]     ,					dtype=np.floataa     ).T for speech in raw_speech]
        elif not is_batched and not isinstance(__UpperCamelCase     ,					np.ndarray     ):
            _UpperCAmelCase          =   np.asarray(__UpperCamelCase     ,					dtype=np.floataa     )
        elif isinstance(__UpperCamelCase     ,					np.ndarray     ) and raw_speech.dtype is np.dtype(np.floataa     ):
            _UpperCAmelCase          =   raw_speech.astype(np.floataa     )
        # always return batch
        if not is_batched:
            _UpperCAmelCase          =   [np.asarray([raw_speech]     ).T]

        # Convert audio signals to log mel spectrograms, truncate by time axis
        _UpperCAmelCase          =   [
            self._np_extract_fbank_features(waveform.squeeze()     ).T[: self.spectrogram_length] for waveform in raw_speech
        ]
        if isinstance(audio_features[0]     ,					__UpperCamelCase     ):
            _UpperCAmelCase          =   [np.asarray(__UpperCamelCase     ,					dtype=np.floataa     ) for feature in audio_features]

        # Create audio attention mask
        _UpperCAmelCase          =   max(
            [ceil(feature.shape[0] / self.patch_size[0]     ) * self.freq_len for feature in audio_features]     )  # The maximum number of audio patches in a batch
        if return_attention_mask:
            _UpperCAmelCase          =   [
                (ceil(feature.shape[0] / self.patch_size[0]     ) * self.freq_len) * [1]
                + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]     ) * self.freq_len) * [0]
                for feature in audio_features
            ]
            _UpperCAmelCase          =   np.array(__UpperCamelCase     ).astype(np.floataa     )

        # convert into correct format for padding
        _UpperCAmelCase          =   max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
        _UpperCAmelCase          =   np.ones([len(__UpperCamelCase     ), 1, max_time_len, self.feature_size]     ).astype(np.floataa     )
        _UpperCAmelCase          =   padded_audio_features * self.padding_value
        for i in range(len(__UpperCamelCase     )     ):
            _UpperCAmelCase          =   audio_features[i]
            _UpperCAmelCase          =   feature

        # return as BatchFeature
        if return_attention_mask:
            _UpperCAmelCase          =   {"""audio_values""": padded_audio_features, """audio_mask""": audio_mask}
        else:
            _UpperCAmelCase          =   {"""audio_values""": padded_audio_features}

        _UpperCAmelCase          =   BatchFeature(data=__UpperCamelCase     ,					tensor_type=__UpperCamelCase     )
        return encoded_inputs

"""simple docstring"""



import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin




class 		a_						( _UpperCAmelCase	):
    a      :				Any     =				['image_processor', 'tokenizer']
    a      :				Optional[int]     =				'AutoImageProcessor'
    a      :				Any     =				'AutoTokenizer'







    def __init__(			self			:       List[str]     ,					__UpperCamelCase			:       List[Any]=None     ,					__UpperCamelCase			:       List[str]=None     ,					**__UpperCamelCase			:       List[Any]     )		->Dict:






        '''simple docstring'''






        _UpperCAmelCase          =   None
        if "feature_extractor" in kwargs:
            warnings.warn(
                """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
                """ instead."""     ,					__UpperCamelCase     ,					)
            _UpperCAmelCase          =   kwargs.pop("""feature_extractor"""     )

        _UpperCAmelCase          =   image_processor if image_processor is not None else feature_extractor
        if image_processor is None:
            raise ValueError("""You need to specify an `image_processor`."""     )
        if tokenizer is None:
            raise ValueError("""You need to specify a `tokenizer`."""     )

        super().__init__(__UpperCamelCase     ,					__UpperCamelCase     )
        _UpperCAmelCase          =   self.image_processor
        _UpperCAmelCase          =   False







    def __call__(			self			:       Union[str, Any]     ,					*__UpperCamelCase			:       str     ,					**__UpperCamelCase			:       Optional[Any]     )		->List[str]:






        '''simple docstring'''






        if self._in_target_context_manager:
            return self.current_processor(*__UpperCamelCase     ,					**__UpperCamelCase     )

        _UpperCAmelCase          =   kwargs.pop("""images"""     ,					__UpperCamelCase     )
        _UpperCAmelCase          =   kwargs.pop("""text"""     ,					__UpperCamelCase     )
        if len(__UpperCamelCase     ) > 0:
            _UpperCAmelCase          =   args[0]
            _UpperCAmelCase          =   args[1:]

        if images is None and text is None:
            raise ValueError("""You need to specify either an `images` or `text` input to process."""     )

        if images is not None:
            _UpperCAmelCase          =   self.image_processor(__UpperCamelCase     ,					*__UpperCamelCase     ,					**__UpperCamelCase     )
        if text is not None:
            _UpperCAmelCase          =   self.tokenizer(__UpperCamelCase     ,					**__UpperCamelCase     )

        if text is None:
            return inputs
        elif images is None:
            return encodings
        else:
            _UpperCAmelCase          =   encodings["""input_ids"""]
            return inputs







    def      _snake_case					(			self			:       Union[str, Any]     ,					*__UpperCamelCase			:       int     ,					**__UpperCamelCase			:       Tuple     )		->Tuple:






        '''simple docstring'''






        return self.tokenizer.batch_decode(*__UpperCamelCase     ,					**__UpperCamelCase     )







    def      _snake_case					(			self			:       Tuple     ,					*__UpperCamelCase			:       int     ,					**__UpperCamelCase			:       Union[str, Any]     )		->int:






        '''simple docstring'''






        return self.tokenizer.decode(*__UpperCamelCase     ,					**__UpperCamelCase     )







    @contextmanager
    def      _snake_case					(			self			:       Tuple     )		->Union[str, Any]:






        '''simple docstring'''






        warnings.warn(
            """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
            """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
            """your images inputs, or in a separate call."""     )
        _UpperCAmelCase          =   True
        _UpperCAmelCase          =   self.tokenizer
        yield
        _UpperCAmelCase          =   self.image_processor
        _UpperCAmelCase          =   False







    def      _snake_case					(			self			:       str     ,					__UpperCamelCase			:       Dict     ,					__UpperCamelCase			:       Optional[Any]=False     ,					__UpperCamelCase			:       Union[str, Any]=None     )		->List[str]:






        '''simple docstring'''






        if added_vocab is None:
            _UpperCAmelCase          =   self.tokenizer.get_added_vocab()

        _UpperCAmelCase          =   {}

        while tokens:
            _UpperCAmelCase          =   re.search(r"""<s_(.*?)>"""     ,					__UpperCamelCase     ,					re.IGNORECASE     )
            if start_token is None:
                break
            _UpperCAmelCase          =   start_token.group(1     )
            _UpperCAmelCase          =   re.search(rf"""</s_{key}>"""     ,					__UpperCamelCase     ,					re.IGNORECASE     )
            _UpperCAmelCase          =   start_token.group()
            if end_token is None:
                _UpperCAmelCase          =   tokens.replace(__UpperCamelCase     ,					""""""     )
            else:
                _UpperCAmelCase          =   end_token.group()
                _UpperCAmelCase          =   re.escape(__UpperCamelCase     )
                _UpperCAmelCase          =   re.escape(__UpperCamelCase     )
                _UpperCAmelCase          =   re.search(f"""{start_token_escaped}(.*?){end_token_escaped}"""     ,					__UpperCamelCase     ,					re.IGNORECASE     )
                if content is not None:
                    _UpperCAmelCase          =   content.group(1     ).strip()
                    if r"<s_" in content and r"</s_" in content:  # non-leaf node
                        _UpperCAmelCase          =   self.tokenajson(__UpperCamelCase     ,					is_inner_value=__UpperCamelCase     ,					added_vocab=__UpperCamelCase     )
                        if value:
                            if len(__UpperCamelCase     ) == 1:
                                _UpperCAmelCase          =   value[0]
                            _UpperCAmelCase          =   value
                    else:  # leaf nodes
                        _UpperCAmelCase          =   []
                        for leaf in content.split(r"""<sep/>"""     ):
                            _UpperCAmelCase          =   leaf.strip()
                            if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
                                _UpperCAmelCase          =   leaf[1:-2]  # for categorical special tokens
                            output[key].append(__UpperCamelCase     )
                        if len(output[key]     ) == 1:
                            _UpperCAmelCase          =   output[key][0]

                _UpperCAmelCase          =   tokens[tokens.find(__UpperCamelCase     ) + len(__UpperCamelCase     ) :].strip()
                if tokens[:6] == r"<sep/>":  # non-leaf nodes
                    return [output] + self.tokenajson(tokens[6:]     ,					is_inner_value=__UpperCamelCase     ,					added_vocab=__UpperCamelCase     )

        if len(__UpperCamelCase     ):
            return [output] if is_inner_value else output
        else:
            return [] if is_inner_value else {"text_sequence": tokens}







    @property
    def      _snake_case					(			self			:       Tuple     )		->Optional[int]:






        '''simple docstring'''






        warnings.warn(
            """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""     ,					__UpperCamelCase     ,					)
        return self.image_processor_class







    @property
    def      _snake_case					(			self			:       List[str]     )		->Dict:






        '''simple docstring'''






        warnings.warn(
            """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""     ,					__UpperCamelCase     ,					)
        return self.image_processor