





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


__A							:			Union[str, Any]         =			logging.get_logger(__name__)







class      lowerCamelCase(						__snake_case	):
     '''simple docstring'''


     __magic_name__   =			['input_features', 'is_longer']




     def __init__(    self				,     snake_case_=64				,     snake_case_=4_8000				,     snake_case_=480				,     snake_case_=10				,     snake_case_=1024				,     snake_case_=0.0				,     snake_case_=False				,     snake_case_ = 0				,     snake_case_ = 1_4000				,     snake_case_ = None				,     snake_case_ = "fusion"				,     snake_case_ = "repeatpad"				,     **snake_case_				,     ):
         super().__init__(
             feature_size=snake_case_				,     sampling_rate=snake_case_				,     padding_value=snake_case_				,     return_attention_mask=snake_case_				,     **snake_case_				,     )
         _A     =							top_db
         _A     =							truncation
         _A     =							padding
         _A     =							fft_window_size
         _A     =							(fft_window_size >> 1) + 1
         _A     =							hop_length
         _A     =							max_length_s
         _A     =							max_length_s * sampling_rate
         _A     =							sampling_rate
         _A     =							frequency_min
         _A     =							frequency_max
         _A     =							mel_filter_bank(
             num_frequency_bins=self.nb_frequency_bins				,     num_mel_filters=snake_case_				,     min_frequency=snake_case_				,     max_frequency=snake_case_				,     sampling_rate=snake_case_				,     norm=snake_case_				,     mel_scale='htk'				,     )
         _A     =							mel_filter_bank(
             num_frequency_bins=self.nb_frequency_bins				,     num_mel_filters=snake_case_				,     min_frequency=snake_case_				,     max_frequency=snake_case_				,     sampling_rate=snake_case_				,     norm='slaney'				,     mel_scale='slaney'				,     )




     def        lowerCAmelCase__		(    self							):
         _A     =							copy.deepcopy(self.__dict__							)
         _A     =							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				,     snake_case_				,     snake_case_ = None							):
         _A     =							spectrogram(
             snake_case_				,     window_function(self.fft_window_size				,     'hann'							)				,     frame_length=self.fft_window_size				,     hop_length=self.hop_length				,     power=2.0				,     mel_filters=snake_case_				,     log_mel='dB'				,     )
         return log_mel_spectrogram.T




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

         _A     =							mel[idx_front : idx_front + chunk_frames, :]
         _A     =							mel[idx_middle : idx_middle + chunk_frames, :]
         _A     =							mel[idx_back : idx_back + chunk_frames, :]

         _A     =							torch.tensor(mel[None, None, :]							)
         _A     =							torch.nn.functional.interpolate(
             snake_case_				,     size=[chunk_frames, 64]				,     mode='bilinear'				,     align_corners=snake_case_							)
         _A     =							mel_shrink[0][0].numpy()
         _A     =							np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]				,     axis=0							)
         return mel_fusion




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

         else:
             _A     =							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     =							int(max_length / len(snake_case_							)							)
                     _A     =							np.stack(np.tile(snake_case_				,     n_repeat + 1							)							)[:max_length]
                 if padding == "repeatpad":
                     _A     =							int(max_length / len(snake_case_							)							)
                     _A     =							np.stack(np.tile(snake_case_				,     snake_case_							)							)
                 _A     =							np.pad(snake_case_				,     (0, max_length - waveform.shape[0])				,     mode='constant'				,     constant_values=0							)

             if truncation == "fusion":
                 _A     =							self._np_extract_fbank_features(snake_case_				,     self.mel_filters							)
                 _A     =							np.stack([input_mel, input_mel, input_mel, input_mel]				,     axis=0							)
             else:
                 _A     =							self._np_extract_fbank_features(snake_case_				,     self.mel_filters_slaney							)[None, :]

         return input_mel, longer




     def __call__(    self				,     snake_case_				,     snake_case_ = None				,     snake_case_ = None				,     snake_case_ = None				,     snake_case_ = None				,     snake_case_ = None				,     **snake_case_				,     ):
         _A     =							truncation if truncation is not None else self.truncation
         _A     =							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     =							isinstance(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}"							)
         _A     =							is_batched_numpy or (
             isinstance(snake_case_				,     (list, tuple)							) and (isinstance(raw_speech[0]				,     (np.ndarray, tuple, list)							))
         )

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

         # always return batch
         if not is_batched:
             _A     =							[np.asarray(snake_case_							)]

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

         _A     =							[]
         _A     =							[]
         for mel, longer in padded_inputs:
             input_mel.append(snake_case_							)
             is_longer.append(snake_case_							)

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

         if isinstance(input_mel[0]				,     snake_case_							):
             _A     =							[np.asarray(snake_case_				,     dtype=np.floataa							) for feature in input_mel]

         # is_longer is a list of bool
         _A     =							[[longer] for longer in is_longer]

         _A     =							{'input_features': input_mel, 'is_longer': is_longer}
         _A     =							BatchFeature(snake_case_							)

         if return_tensors is not None:
             _A     =							input_features.convert_to_tensors(snake_case_							)

         return input_features












import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
 import torch

if is_datasets_available():
 from datasets import load_dataset

__A							:			str         =			random.Random()





def 				__lowerCAmelCase(       _SCREAMING_SNAKE_CASE							,     _SCREAMING_SNAKE_CASE=1.0							,     _SCREAMING_SNAKE_CASE=None							,     _SCREAMING_SNAKE_CASE=None				)							-> Union[str, Any]:




    """simple docstring"""




    if rng is None:
        _A     =							global_rng

    _A     =							[]
    for batch_idx in range(shape[0]				):
        values.append([]				)
        for _ in range(shape[1]				):
            values[-1].append(rng.random() * scale				)

    return values







class      lowerCamelCase(						unittest.TestCase	):
     '''simple docstring'''


     def __init__(    self				,     snake_case_				,     snake_case_=7				,     snake_case_=400				,     snake_case_=2000				,     snake_case_=2048				,     snake_case_=128				,     snake_case_=1				,     snake_case_=512				,     snake_case_=30				,     snake_case_=4_4100				,     ):
         _A     =							parent
         _A     =							batch_size
         _A     =							min_seq_length
         _A     =							max_seq_length
         _A     =							(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
         _A     =							spectrogram_length
         _A     =							feature_size
         _A     =							num_audio_channels
         _A     =							hop_length
         _A     =							chunk_length
         _A     =							sampling_rate




     def        lowerCAmelCase__		(    self							):
         return {
             "spectrogram_length": self.spectrogram_length,
             "feature_size": self.feature_size,
             "num_audio_channels": self.num_audio_channels,
             "hop_length": self.hop_length,
             "chunk_length": self.chunk_length,
             "sampling_rate": self.sampling_rate,
         }




     def        lowerCAmelCase__		(    self				,     snake_case_=False				,     snake_case_=False							):
         def _flatten(snake_case_							):
             return list(itertools.chain(*snake_case_							)							)

         if equal_length:
             _A     =							[floats_list((self.max_seq_length, self.feature_size)							) for _ in range(self.batch_size							)]
         else:
             # make sure that inputs increase in size
             _A     =							[
                 floats_list((x, self.feature_size)							)
                 for x in range(self.min_seq_length				,     self.max_seq_length				,     self.seq_length_diff							)
             ]
         if numpify:
             _A     =							[np.asarray(snake_case_							) for x in speech_inputs]
         return speech_inputs







@require_torch
@require_torchaudio
class      lowerCamelCase(						__snake_case			, unittest.TestCase	):
     '''simple docstring'''


     __magic_name__   =			TvltFeatureExtractor




     def        lowerCAmelCase__		(    self							):
         _A     =							TvltFeatureExtractionTester(self							)




     def        lowerCAmelCase__		(    self							):
         _A     =							self.feature_extraction_class(**self.feat_extract_dict							)
         self.assertTrue(hasattr(snake_case_				,     'spectrogram_length'							)							)
         self.assertTrue(hasattr(snake_case_				,     'feature_size'							)							)
         self.assertTrue(hasattr(snake_case_				,     'num_audio_channels'							)							)
         self.assertTrue(hasattr(snake_case_				,     'hop_length'							)							)
         self.assertTrue(hasattr(snake_case_				,     'chunk_length'							)							)
         self.assertTrue(hasattr(snake_case_				,     'sampling_rate'							)							)




     def        lowerCAmelCase__		(    self							):
         _A     =							self.feature_extraction_class(**self.feat_extract_dict							)

         with tempfile.TemporaryDirectory() as tmpdirname:
             _A     =							feat_extract_first.save_pretrained(snake_case_							)[0]
             check_json_file_has_correct_format(snake_case_							)
             _A     =							self.feature_extraction_class.from_pretrained(snake_case_							)

         _A     =							feat_extract_first.to_dict()
         _A     =							feat_extract_second.to_dict()
         _A     =							dict_first.pop('mel_filters'							)
         _A     =							dict_second.pop('mel_filters'							)
         self.assertTrue(np.allclose(snake_case_				,     snake_case_							)							)
         self.assertEqual(snake_case_				,     snake_case_							)




     def        lowerCAmelCase__		(    self							):
         _A     =							self.feature_extraction_class(**self.feat_extract_dict							)

         with tempfile.TemporaryDirectory() as tmpdirname:
             _A     =							os.path.join(snake_case_				,     'feat_extract.json'							)
             feat_extract_first.to_json_file(snake_case_							)
             _A     =							self.feature_extraction_class.from_json_file(snake_case_							)

         _A     =							feat_extract_first.to_dict()
         _A     =							feat_extract_second.to_dict()
         _A     =							dict_first.pop('mel_filters'							)
         _A     =							dict_second.pop('mel_filters'							)
         self.assertTrue(np.allclose(snake_case_				,     snake_case_							)							)
         self.assertEqual(snake_case_				,     snake_case_							)




     def        lowerCAmelCase__		(    self							):
         # Initialize feature_extractor
         _A     =							self.feature_extraction_class(**self.feat_extract_dict							)

         # create three inputs of length 800, 1000, and 1200
         _A     =							[floats_list((1, x)							)[0] for x in range(800				,     1400				,     200							)]
         _A     =							[np.asarray(snake_case_							) for speech_input in speech_inputs]

         # Test not batched input
         _A     =							feature_extractor(np_speech_inputs[0]				,     return_tensors='np'				,     sampling_rate=4_4100							).audio_values

         self.assertTrue(encoded_audios.ndim == 4							)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size							)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length							)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels							)

         # Test batched
         _A     =							feature_extractor(snake_case_				,     return_tensors='np'				,     sampling_rate=4_4100							).audio_values

         self.assertTrue(encoded_audios.ndim == 4							)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size							)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length							)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels							)

         # Test audio masking
         _A     =							feature_extractor(
             snake_case_				,     return_tensors='np'				,     sampling_rate=4_4100				,     mask_audio=snake_case_							).audio_values

         self.assertTrue(encoded_audios.ndim == 4							)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size							)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length							)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels							)

         # Test 2-D numpy arrays are batched.
         _A     =							[floats_list((1, x)							)[0] for x in (800, 800, 800)]
         _A     =							np.asarray(snake_case_							)
         _A     =							feature_extractor(snake_case_				,     return_tensors='np'				,     sampling_rate=4_4100							).audio_values
         self.assertTrue(encoded_audios.ndim == 4							)
         self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size							)
         self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length							)
         self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels							)




     def        lowerCAmelCase__		(    self				,     snake_case_							):
         _A     =							load_dataset('hf-internal-testing/librispeech_asr_dummy'				,     'clean'				,     split='validation'							)
         # automatic decoding with librispeech
         _A     =							ds.sort('id'							).select(range(snake_case_							)							)[:num_samples]['audio']

         return [x["array"] for x in speech_samples]




     def        lowerCAmelCase__		(    self							):
         _A     =							self._load_datasamples(1							)
         _A     =							TvltFeatureExtractor()
         _A     =							feature_extractor(snake_case_				,     return_tensors='pt'							).audio_values

         self.assertEquals(audio_values.shape				,     (1, 1, 192, 128)							)

         _A     =							torch.tensor([[-0.3032, -0.2708], [-0.4434, -0.4007]]							)
         self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]				,     snake_case_				,     atol=1E-4							)							)





