




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


UpperCAmelCase_						=logging.get_logger(__name__)


class 							__UpperCamelCase			( __UpperCAmelCase  ):





				'''simple docstring'''



				__a							: Tuple										=["""input_features""", """is_longer"""]



				def __init__( self       , UpperCAmelCase_=64       , UpperCAmelCase_=4_80_00       , UpperCAmelCase_=4_80       , UpperCAmelCase_=10       , UpperCAmelCase_=10_24       , UpperCAmelCase_=0.0       , UpperCAmelCase_=False       , UpperCAmelCase_ = 0       , UpperCAmelCase_ = 1_40_00       , UpperCAmelCase_ = None       , UpperCAmelCase_ = "fusion"       , UpperCAmelCase_ = "repeatpad"       , **UpperCAmelCase_       , ):
											super().__init__(
											    feature_size=UpperCAmelCase_       , sampling_rate=UpperCAmelCase_       , padding_value=UpperCAmelCase_       , return_attention_mask=UpperCAmelCase_       , **UpperCAmelCase_       , )
											lowerCAmelCase   =		top_db
											lowerCAmelCase   =		truncation
											lowerCAmelCase   =		padding
											lowerCAmelCase   =		fft_window_size
											lowerCAmelCase   =		(fft_window_size >> 1) + 1
											lowerCAmelCase   =		hop_length
											lowerCAmelCase   =		max_length_s
											lowerCAmelCase   =		max_length_s * sampling_rate
											lowerCAmelCase   =		sampling_rate
											lowerCAmelCase   =		frequency_min
											lowerCAmelCase   =		frequency_max
											lowerCAmelCase   =		mel_filter_bank(
											    num_frequency_bins=self.nb_frequency_bins       , num_mel_filters=UpperCAmelCase_       , min_frequency=UpperCAmelCase_       , max_frequency=UpperCAmelCase_       , sampling_rate=UpperCAmelCase_       , norm=UpperCAmelCase_       , mel_scale='''htk'''       , )
											lowerCAmelCase   =		mel_filter_bank(
											    num_frequency_bins=self.nb_frequency_bins       , num_mel_filters=UpperCAmelCase_       , min_frequency=UpperCAmelCase_       , max_frequency=UpperCAmelCase_       , sampling_rate=UpperCAmelCase_       , norm='''slaney'''       , mel_scale='''slaney'''       , )



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



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

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

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



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

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

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

											return input_mel, longer






				def __call__( self       , UpperCAmelCase_       , UpperCAmelCase_ = None       , UpperCAmelCase_ = None       , UpperCAmelCase_ = None       , UpperCAmelCase_ = None       , UpperCAmelCase_ = None       , **UpperCAmelCase_       , ):
											lowerCAmelCase   =		truncation if truncation is not None else self.truncation
											lowerCAmelCase   =		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.'''						)

											lowerCAmelCase   =		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}"""						)
											lowerCAmelCase   =		is_batched_numpy or (
											    isinstance(UpperCAmelCase_       , (list, tuple)						) and (isinstance(raw_speech[0]       , (np.ndarray, tuple, list)						))
											)

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

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

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

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

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

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

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

											lowerCAmelCase   =		{'''input_features''': input_mel, '''is_longer''': is_longer}
											lowerCAmelCase   =		BatchFeature(UpperCAmelCase_						)

											if return_tensors is not None:
																		lowerCAmelCase   =		input_features.convert_to_tensors(UpperCAmelCase_						)

											return input_features









import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY


@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class 							__UpperCamelCase			( unittest.TestCase  ):





				'''simple docstring'''



				__a							: str										=MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_       , UpperCAmelCase_						):
											lowerCAmelCase   =		hf_hub_download(
											    repo_id='''nateraw/video-demo'''       , filename='''archery.mp4'''       , repo_type='''dataset'''						)
											lowerCAmelCase   =		VideoClassificationPipeline(model=UpperCAmelCase_       , image_processor=UpperCAmelCase_       , top_k=2						)
											lowerCAmelCase   =		[
											    example_video_filepath,
											    '''https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4''',
											]
											return video_classifier, examples



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_						):
											for example in examples:
																		lowerCAmelCase   =		video_classifier(UpperCAmelCase_						)

																		self.assertEqual(
																		    UpperCAmelCase_       , [
																		        {'''score''': ANY(UpperCAmelCase_						), '''label''': ANY(UpperCAmelCase_						)},
																		        {'''score''': ANY(UpperCAmelCase_						), '''label''': ANY(UpperCAmelCase_						)},
																		    ]       , )



				@require_torch
				def       __snake_case		( self						):
											lowerCAmelCase   =		'''hf-internal-testing/tiny-random-VideoMAEForVideoClassification'''
											lowerCAmelCase   =		VideoMAEFeatureExtractor(
											    size={'''shortest_edge''': 10}       , crop_size={'''height''': 10, '''width''': 10}						)
											lowerCAmelCase   =		pipeline(
											    '''video-classification'''       , model=UpperCAmelCase_       , feature_extractor=UpperCAmelCase_       , frame_sampling_rate=4						)

											lowerCAmelCase   =		hf_hub_download(repo_id='''nateraw/video-demo'''       , filename='''archery.mp4'''       , repo_type='''dataset'''						)
											lowerCAmelCase   =		video_classifier(UpperCAmelCase_       , top_k=2						)
											self.assertEqual(
											    nested_simplify(UpperCAmelCase_       , decimals=4						)       , [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}]       , )

											lowerCAmelCase   =		video_classifier(
											    [
											        video_file_path,
											        video_file_path,
											    ]       , top_k=2       , )
											self.assertEqual(
											    nested_simplify(UpperCAmelCase_       , decimals=4						)       , [
											        [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
											        [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
											    ]       , )






				@require_tf
				def       __snake_case		( self						):
											pass



