




'''simple docstring'''







import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VivitConfig
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor


if is_torch_available():
   import torch
   from torch import nn

   from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VivitForVideoClassification, VivitModel
   from transformers.models.vivit.modeling_vivit import VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
   from transformers import VivitImageProcessor


class       _a  :
       def __init__(     self						:     str       ,    lowercase						:     Dict       ,    lowercase						:     str=2       ,    lowercase						:     int=True       ,    lowercase						:     List[str]=True       ,    lowercase						:     int=10       ,    lowercase						:     Dict=10       ,    lowercase						:     Any=8       ,    lowercase						:     Optional[int]=[2, 4, 4]       ,    lowercase						:     str=3       ,    lowercase						:     Optional[Any]=768       ,    lowercase						:     int=5       ,    lowercase						:     int=4       ,    lowercase						:     int=37       ,    lowercase						:     Union[str, Any]="gelu_fast"       ,    lowercase						:     Union[str, Any]=0.0       ,    lowercase						:     List[str]=0.0       ,    lowercase						:     int=0.02       ,    lowercase						:     str=1E-06       ,    lowercase						:     str=True       ,    lowercase						:     int=None       ,    ):



           '''simple docstring'''

           UpperCAmelCase      	=		parent
           UpperCAmelCase      	=		batch_size
           UpperCAmelCase      	=		is_training
           UpperCAmelCase      	=		use_labels
           UpperCAmelCase      	=		num_labels
           UpperCAmelCase      	=		image_size
           UpperCAmelCase      	=		num_frames
           UpperCAmelCase      	=		tubelet_size
           UpperCAmelCase      	=		num_channels
           UpperCAmelCase      	=		hidden_size
           UpperCAmelCase      	=		num_hidden_layers
           UpperCAmelCase      	=		num_attention_heads
           UpperCAmelCase      	=		intermediate_size
           UpperCAmelCase      	=		hidden_act
           UpperCAmelCase      	=		hidden_dropout_prob
           UpperCAmelCase      	=		attention_probs_dropout_prob
           UpperCAmelCase      	=		initializer_range
           UpperCAmelCase      	=		layer_norm_eps
           UpperCAmelCase      	=		qkv_bias
           UpperCAmelCase      	=		scope

           UpperCAmelCase      	=		(
               (self.image_size // self.tubelet_size[2])
               * (self.image_size // self.tubelet_size[1])
               * (self.num_frames // self.tubelet_size[0])
           ) + 1  # CLS token
       def    A				(     self						:     int    ):



           '''simple docstring'''

           UpperCAmelCase      	=		floats_tensor(
               [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]    )

           UpperCAmelCase      	=		None
           if self.use_labels:
               UpperCAmelCase      	=		ids_tensor([self.batch_size]       ,    self.num_labels    )

           UpperCAmelCase      	=		self.get_config()

           return config, pixel_values, labels
       def    A				(     self						:     Any    ):



           '''simple docstring'''

           UpperCAmelCase      	=		VivitConfig(
               num_frames=self.num_frames       ,    image_size=self.image_size       ,    tubelet_size=self.tubelet_size       ,    num_channels=self.num_channels       ,    hidden_size=self.hidden_size       ,    num_hidden_layers=self.num_hidden_layers       ,    num_attention_heads=self.num_attention_heads       ,    intermediate_size=self.intermediate_size       ,    hidden_act=self.hidden_act       ,    hidden_dropout_prob=self.hidden_dropout_prob       ,    attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,    initializer_range=self.initializer_range       ,    layer_norm_eps=self.layer_norm_eps       ,    qkv_bias=self.qkv_bias       ,    )
           UpperCAmelCase      	=		self.num_labels
           return config
       def    A				(     self						:     List[str]       ,    lowercase						:     Optional[Any]       ,    lowercase						:     Optional[int]       ,    lowercase						:     Optional[int]    ):



           '''simple docstring'''

           UpperCAmelCase      	=		VivitModel(config=lowercase    )
           model.to(lowercase    )
           model.eval()
           UpperCAmelCase      	=		model(lowercase    )
           self.parent.assertEqual(result.last_hidden_state.shape       ,    (self.batch_size, self.seq_length, self.hidden_size)    )
       def    A				(     self						:     Optional[Any]       ,    lowercase						:     int       ,    lowercase						:     Optional[Any]       ,    lowercase						:     int    ):



           '''simple docstring'''

           UpperCAmelCase      	=		VivitForVideoClassification(lowercase    )
           model.to(lowercase    )
           model.eval()

           UpperCAmelCase      	=		model(lowercase    )

           # verify the logits shape
           UpperCAmelCase      	=		torch.Size((self.batch_size, self.num_labels)    )
           self.parent.assertEqual(result.logits.shape       ,    lowercase    )







       def    A				(     self						:     Any    ):



           '''simple docstring'''

           UpperCAmelCase      	=		self.prepare_config_and_inputs()
           UpperCAmelCase	,					UpperCAmelCase	,					UpperCAmelCase      	=		config_and_inputs
           UpperCAmelCase      	=		{'''pixel_values''': pixel_values}
           return config, inputs_dict






@require_torch
class       _a  (     __a							,							unittest.TestCase			):
       __a    :					str           = (VivitModel, VivitForVideoClassification) if is_torch_available() else ()

       __a    :					Union[str, Any]           = False
       __a    :					int           = False
       __a    :					List[Any]           = False
       __a    :					Tuple           = False
       def    A				(     self						:     Optional[Any]    ):



           '''simple docstring'''

           UpperCAmelCase      	=		VivitModelTester(self    )
           UpperCAmelCase      	=		ConfigTester(self       ,    config_class=lowercase       ,    has_text_modality=lowercase       ,    hidden_size=37    )
       def    A				(     self						:     List[Any]       ,    lowercase						:     str       ,    lowercase						:     Dict       ,    lowercase						:     List[str]=False    ):



           '''simple docstring'''

           UpperCAmelCase      	=		copy.deepcopy(lowercase    )

           if return_labels:
               if model_class in get_values(lowercase    ):
                   UpperCAmelCase      	=		torch.zeros(
                       self.model_tester.batch_size       ,    dtype=torch.long       ,    device=lowercase    )

           return inputs_dict
       def    A				(     self						:     Tuple    ):



           '''simple docstring'''

           self.config_tester.run_common_tests()
       @unittest.skip(reason='''Vivit does not use inputs_embeds'''    )
       def    A				(     self						:     Any    ):



           '''simple docstring'''

           pass
       def    A				(     self						:     Optional[Any]    ):



           '''simple docstring'''

           UpperCAmelCase	,					UpperCAmelCase      	=		self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
               UpperCAmelCase      	=		model_class(lowercase    )
               self.assertIsInstance(model.get_input_embeddings()       ,    (nn.Module)    )
               UpperCAmelCase      	=		model.get_output_embeddings()
               self.assertTrue(x is None or isinstance(lowercase       ,    nn.Linear    )    )
       def    A				(     self						:     Tuple    ):



           '''simple docstring'''

           UpperCAmelCase	,					UpperCAmelCase      	=		self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
               UpperCAmelCase      	=		model_class(lowercase    )
               UpperCAmelCase      	=		inspect.signature(model.forward    )
               # signature.parameters is an OrderedDict => so arg_names order is deterministic
               UpperCAmelCase      	=		[*signature.parameters.keys()]

               UpperCAmelCase      	=		['''pixel_values''', '''head_mask''']
               self.assertListEqual(arg_names[:2]       ,    lowercase    )
       def    A				(     self						:     Dict    ):



           '''simple docstring'''

           UpperCAmelCase      	=		self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_model(*lowercase    )
       def    A				(     self						:     Union[str, Any]    ):



           '''simple docstring'''

           UpperCAmelCase      	=		self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_for_video_classification(*lowercase    )
       @slow
       def    A				(     self						:     Union[str, Any]    ):



           '''simple docstring'''

           for model_name in VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               UpperCAmelCase      	=		VivitModel.from_pretrained(lowercase    )
               self.assertIsNotNone(lowercase    )
       def    A				(     self						:     List[Any]    ):



           '''simple docstring'''

           UpperCAmelCase	,					UpperCAmelCase      	=		self.model_tester.prepare_config_and_inputs_for_common()
           UpperCAmelCase      	=		True

           for model_class in self.all_model_classes:
               UpperCAmelCase      	=		self.model_tester.seq_length

               UpperCAmelCase      	=		True
               UpperCAmelCase      	=		False
               UpperCAmelCase      	=		True
               UpperCAmelCase      	=		model_class(lowercase    )
               model.to(lowercase    )
               model.eval()
               with torch.no_grad():
                   UpperCAmelCase      	=		model(**self._prepare_for_class(lowercase       ,    lowercase    )    )
               UpperCAmelCase      	=		outputs.attentions
               self.assertEqual(len(lowercase    )       ,    self.model_tester.num_hidden_layers    )

               # check that output_attentions also work using config
               del inputs_dict["output_attentions"]
               UpperCAmelCase      	=		True
               UpperCAmelCase      	=		model_class(lowercase    )
               model.to(lowercase    )
               model.eval()
               with torch.no_grad():
                   UpperCAmelCase      	=		model(**self._prepare_for_class(lowercase       ,    lowercase    )    )
               UpperCAmelCase      	=		outputs.attentions
               self.assertEqual(len(lowercase    )       ,    self.model_tester.num_hidden_layers    )

               self.assertListEqual(
                   list(attentions[0].shape[-3:]    )       ,    [self.model_tester.num_attention_heads, seq_len, seq_len]       ,    )
               UpperCAmelCase      	=		len(lowercase    )

               # Check attention is always last and order is fine
               UpperCAmelCase      	=		True
               UpperCAmelCase      	=		True
               UpperCAmelCase      	=		model_class(lowercase    )
               model.to(lowercase    )
               model.eval()
               with torch.no_grad():
                   UpperCAmelCase      	=		model(**self._prepare_for_class(lowercase       ,    lowercase    )    )

               self.assertEqual(out_len + 1       ,    len(lowercase    )    )

               UpperCAmelCase      	=		outputs.attentions

               self.assertEqual(len(lowercase    )       ,    self.model_tester.num_hidden_layers    )

               self.assertListEqual(
                   list(self_attentions[0].shape[-3:]    )       ,    [self.model_tester.num_attention_heads, seq_len, seq_len]       ,    )
       def    A				(     self						:     List[Any]    ):



           '''simple docstring'''

           def check_hidden_states_output(lowercase						:     int       ,    lowercase						:     Union[str, Any]       ,    lowercase						:     str    ):
               UpperCAmelCase      	=		model_class(lowercase    )
               model.to(lowercase    )
               model.eval()

               with torch.no_grad():
                   UpperCAmelCase      	=		model(**self._prepare_for_class(lowercase       ,    lowercase    )    )

               UpperCAmelCase      	=		outputs.hidden_states
               UpperCAmelCase      	=		self.model_tester.num_hidden_layers + 1
               self.assertEqual(len(lowercase    )       ,    lowercase    )

               UpperCAmelCase      	=		self.model_tester.seq_length

               self.assertListEqual(
                   list(hidden_states[0].shape[-2:]    )       ,    [seq_length, self.model_tester.hidden_size]       ,    )

           UpperCAmelCase	,					UpperCAmelCase      	=		self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
               UpperCAmelCase      	=		True
               check_hidden_states_output(lowercase       ,    lowercase       ,    lowercase    )

               # check that output_hidden_states also work using config
               del inputs_dict["output_hidden_states"]
               UpperCAmelCase      	=		True

               check_hidden_states_output(lowercase       ,    lowercase       ,    lowercase    )







       @unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''    )
       def    A				(     self						:     List[Any]    ):



           '''simple docstring'''

           pass






def 						snake_case_							():
    UpperCAmelCase      	=		hf_hub_download(
        repo_id='''hf-internal-testing/spaghetti-video'''   ,       filename='''eating_spaghetti_32_frames.npy'''   ,       repo_type='''dataset'''					)
    UpperCAmelCase      	=		np.load(_a					)
    return list(_a					)







@require_torch
@require_vision
class       _a  (     unittest.TestCase			):
       @cached_property
       def    A				(     self						:     Optional[int]    ):



           '''simple docstring'''

           return VivitImageProcessor() if is_vision_available() else None







       @slow
       def    A				(     self						:     Tuple    ):



           '''simple docstring'''

           UpperCAmelCase      	=		VivitForVideoClassification.from_pretrained('''google/vivit-b-16x2-kinetics400'''    ).to(lowercase    )

           UpperCAmelCase      	=		self.default_image_processor
           UpperCAmelCase      	=		prepare_video()
           UpperCAmelCase      	=		image_processor(lowercase       ,    return_tensors='''pt'''    ).to(lowercase    )

           # forward pass
           with torch.no_grad():
               UpperCAmelCase      	=		model(**lowercase    )

           # verify the logits
           UpperCAmelCase      	=		torch.Size((1, 400)    )
           self.assertEqual(outputs.logits.shape       ,    lowercase    )

           # taken from original model
           UpperCAmelCase      	=		torch.tensor([-1.0543, 2.0764, -0.2104, 0.4439, -0.9658]    ).to(lowercase    )

           self.assertTrue(torch.allclose(outputs.logits[0, :5]       ,    lowercase       ,    atol=1E-4    )    )










'''simple docstring'''







import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
   from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

A			   =logging.get_logger(__name__)





def 						snake_case_							(_a     :							bytes   ,       _a     :							int					):
    UpperCAmelCase      	=		F"{sampling_rate}"
    UpperCAmelCase      	=		'''1'''
    UpperCAmelCase      	=		'''f32le'''
    UpperCAmelCase      	=		[
        '''ffmpeg''',
        '''-i''',
        '''pipe:0''',
        '''-ac''',
        ac,
        '''-ar''',
        ar,
        '''-f''',
        format_for_conversion,
        '''-hide_banner''',
        '''-loglevel''',
        '''quiet''',
        '''pipe:1''',
    ]

    try:
        UpperCAmelCase      	=		subprocess.Popen(_a   ,       stdin=subprocess.PIPE   ,       stdout=subprocess.PIPE					)
    except FileNotFoundError:
        raise ValueError('''ffmpeg was not found but is required to load audio files from filename'''					)
    UpperCAmelCase      	=		ffmpeg_process.communicate(_a					)
    UpperCAmelCase      	=		output_stream[0]

    UpperCAmelCase      	=		np.frombuffer(_a   ,       np.floataa					)
    if audio.shape[0] == 0:
        raise ValueError('''Malformed soundfile'''					)
    return audio







@add_end_docstrings(__a			)
class       _a  (     __a			):
       def __init__(     self						:     Dict       ,    *lowercase						:     Dict       ,    **lowercase						:     Any    ):



           '''simple docstring'''

           UpperCAmelCase      	=		5
           super().__init__(*lowercase       ,    **lowercase    )

           if self.framework != "pt":
               raise ValueError(f"The {self.__class__} is only available in PyTorch."    )

           self.check_model_type(lowercase    )
       def __call__(     self						:     List[Any]       ,    lowercase						:     Union[np.ndarray, bytes, str]       ,    **lowercase						:     Dict       ,    ):



           '''simple docstring'''

           return super().__call__(lowercase       ,    **lowercase    )
       def    A				(     self						:     Tuple       ,    lowercase						:     Dict=None       ,    **lowercase						:     List[str]    ):



           '''simple docstring'''

           UpperCAmelCase      	=		{}
           if top_k is not None:
               if top_k > self.model.config.num_labels:
                   UpperCAmelCase      	=		self.model.config.num_labels
               UpperCAmelCase      	=		top_k
           return {}, {}, postprocess_params
       def    A				(     self						:     List[Any]       ,    lowercase						:     Union[str, Any]    ):



           '''simple docstring'''

           if isinstance(lowercase       ,    lowercase    ):
               if inputs.startswith('''http://'''    ) or inputs.startswith('''https://'''    ):
                   # We need to actually check for a real protocol, otherwise it's impossible to use a local file
                   # like http_huggingface_co.png
                   UpperCAmelCase      	=		requests.get(lowercase    ).content
               else:
                   with open(lowercase       ,    '''rb'''    ) as f:
                       UpperCAmelCase      	=		f.read()

           if isinstance(lowercase       ,    lowercase    ):
               UpperCAmelCase      	=		ffmpeg_read(lowercase       ,    self.feature_extractor.sampling_rate    )

           if isinstance(lowercase       ,    lowercase    ):
               # Accepting `"array"` which is the key defined in `datasets` for
               # better integration
               if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
                   raise ValueError(
                       '''When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a '''
                       '''"raw" key containing the numpy array representing the audio and a "sampling_rate" key, '''
                       '''containing the sampling_rate associated with that array'''    )

               UpperCAmelCase      	=		inputs.pop('''raw'''       ,    lowercase    )
               if _inputs is None:
                   # Remove path which will not be used from `datasets`.
                   inputs.pop('''path'''       ,    lowercase    )
                   UpperCAmelCase      	=		inputs.pop('''array'''       ,    lowercase    )
               UpperCAmelCase      	=		inputs.pop('''sampling_rate'''    )
               UpperCAmelCase      	=		_inputs
               if in_sampling_rate != self.feature_extractor.sampling_rate:
                   import torch

                   if is_torchaudio_available():
                       from torchaudio import functional as F
                   else:
                       raise ImportError(
                           '''torchaudio is required to resample audio samples in AudioClassificationPipeline. '''
                           '''The torchaudio package can be installed through: `pip install torchaudio`.'''    )

                   UpperCAmelCase      	=		F.resample(
                       torch.from_numpy(lowercase    )       ,    lowercase       ,    self.feature_extractor.sampling_rate    ).numpy()

           if not isinstance(lowercase       ,    np.ndarray    ):
               raise ValueError('''We expect a numpy ndarray as input'''    )
           if len(inputs.shape    ) != 1:
               raise ValueError('''We expect a single channel audio input for AudioClassificationPipeline'''    )

           UpperCAmelCase      	=		self.feature_extractor(
               lowercase       ,    sampling_rate=self.feature_extractor.sampling_rate       ,    return_tensors='''pt'''    )
           return processed
       def    A				(     self						:     Optional[Any]       ,    lowercase						:     List[str]    ):



           '''simple docstring'''

           UpperCAmelCase      	=		self.model(**lowercase    )
           return model_outputs







       def    A				(     self						:     Union[str, Any]       ,    lowercase						:     Optional[Any]       ,    lowercase						:     List[str]=5    ):



           '''simple docstring'''

           UpperCAmelCase      	=		model_outputs.logits[0].softmax(-1    )
           UpperCAmelCase	,					UpperCAmelCase      	=		probs.topk(lowercase    )

           UpperCAmelCase      	=		scores.tolist()
           UpperCAmelCase      	=		ids.tolist()

           UpperCAmelCase      	=		[{'''score''': score, '''label''': self.model.config.idalabel[_id]} for score, _id in zip(lowercase       ,    lowercase    )]

           return labels




