



'''simple docstring'''






import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig
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
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch
     from torch import nn

     from transformers import (
         MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING,
         TimesformerForVideoClassification,
         TimesformerModel,
     )
     from transformers.models.timesformer.modeling_timesformer import TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from transformers import VideoMAEImageProcessor

class 				a__:







    def __init__( self					:    Optional[Any]		,     __snake_case					:    Tuple		,     __snake_case					:    Dict=13		,     __snake_case					:    int=10		,     __snake_case					:    Any=3		,     __snake_case					:    Optional[int]=2		,     __snake_case					:    int=2		,     __snake_case					:    List[Any]=True		,     __snake_case					:    Any=True		,     __snake_case					:    List[Any]=32		,     __snake_case					:    Dict=5		,     __snake_case					:    Optional[int]=4		,     __snake_case					:    Optional[Any]=37		,     __snake_case					:    Any="gelu"		,     __snake_case					:    Any=0.1		,     __snake_case					:    Any=0.1		,     __snake_case					:    Optional[int]=10		,     __snake_case					:    Optional[Any]=0.02		,     __snake_case					:    Dict="divided_space_time"		,     __snake_case					:    int=None		,     ):
      a		:   Dict	  =	parent
      a		:   List[str]	  =	batch_size
      a		:   List[Any]	  =	image_size
      a		:   str	  =	num_channels
      a		:   str	  =	patch_size
      a		:   Union[str, Any]	  =	num_frames
      a		:   str	  =	is_training
      a		:   str	  =	use_labels
      a		:   Optional[int]	  =	hidden_size
      a		:   Dict	  =	num_hidden_layers
      a		:   str	  =	num_attention_heads
      a		:   Union[str, Any]	  =	intermediate_size
      a		:   Optional[Any]	  =	hidden_act
      a		:   Optional[Any]	  =	hidden_dropout_prob
      a		:   Optional[Any]	  =	attention_probs_dropout_prob
      a		:   Tuple	  =	attention_type
      a		:   Optional[int]	  =	initializer_range
      a		:   Any	  =	scope
      a		:   int	  =	num_labels

      # in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
      a		:   str	  =	(image_size // patch_size) ** 2
      a		:   Optional[Any]	  =	(num_frames) * self.num_patches_per_frame + 1







    def      lowercase_			( self					:    Tuple      ):
      a		:   List[str]	  =	floats_tensor(
          [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]      )

      a		:   List[str]	  =	None
      if self.use_labels:
        a		:   Any	  =	ids_tensor([self.batch_size]		,     self.num_labels      )

      a		:   Optional[int]	  =	self.get_config()

      return config, pixel_values, labels







    def      lowercase_			( self					:    List[str]      ):
      a		:   Optional[Any]	  =	TimesformerConfig(
          image_size=self.image_size		,     patch_size=self.patch_size		,     num_channels=self.num_channels		,     num_frames=self.num_frames		,     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		,     attention_type=self.attention_type		,     )
      a		:   List[str]	  =	self.num_labels
      return config







    def      lowercase_			( self					:    Union[str, Any]		,     __snake_case					:    Optional[int]		,     __snake_case					:    str		,     __snake_case					:    Dict      ):
      a		:   List[str]	  =	TimesformerModel(config=__snake_case      )
      model.to(__snake_case      )
      model.eval()
      a		:   Union[str, Any]	  =	model(__snake_case      )
      self.parent.assertEqual(result.last_hidden_state.shape		,     (self.batch_size, self.seq_length, self.hidden_size)      )







    def      lowercase_			( self					:    List[Any]		,     __snake_case					:    Optional[Any]		,     __snake_case					:    Tuple		,     __snake_case					:    List[Any]      ):
      a		:   Union[str, Any]	  =	TimesformerForVideoClassification(__snake_case      )
      model.to(__snake_case      )
      model.eval()

      a		:   List[Any]	  =	model(__snake_case      )

      # verify the logits shape
      a		:   Union[str, Any]	  =	torch.Size((self.batch_size, self.num_labels)      )
      self.parent.assertEqual(result.logits.shape		,     __snake_case      )







    def      lowercase_			( self					:    Any      ):
      a		:   Tuple	  =	self.prepare_config_and_inputs()
      a  ,      a  ,      a		:   Union[str, Any]	  =	config_and_inputs
      a		:   List[Any]	  =	{'pixel_values': pixel_values}
      return config, inputs_dict

@require_torch
class 				a__(   lowerCamelCase__						,					lowerCamelCase__						,					unittest.TestCase      ):
    lowercase__					=       (TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
    lowercase__					=       (
        {"""feature-extraction""": TimesformerModel, """video-classification""": TimesformerForVideoClassification}
        if is_torch_available()
        else {}
    )

    lowercase__					=       False
    lowercase__					=       False
    lowercase__					=       False
    lowercase__					=       False







    def      lowercase_			( self					:    List[Any]      ):
      a		:   Optional[int]	  =	TimesformerModelTester(self      )
      a		:   Optional[Any]	  =	ConfigTester(
          self		,     config_class=__snake_case		,     has_text_modality=__snake_case		,     hidden_size=37      )







    def      lowercase_			( self					:    str		,     __snake_case					:    Dict		,     __snake_case					:    Union[str, Any]		,     __snake_case					:    Optional[int]=False      ):
      a		:   List[Any]	  =	copy.deepcopy(__snake_case      )

      if return_labels:
        if model_class in get_values(__snake_case      ):
          a		:   Dict	  =	torch.zeros(
              self.model_tester.batch_size		,     dtype=torch.long		,     device=__snake_case      )

      return inputs_dict







    def      lowercase_			( self					:    int      ):
      self.config_tester.run_common_tests()







    @unittest.skip(reason='TimeSformer does not use inputs_embeds'      )
    def      lowercase_			( self					:    Tuple      ):
      pass







    def      lowercase_			( self					:    Optional[Any]      ):
      a  ,      a		:   Dict	  =	self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a		:   str	  =	model_class(__snake_case      )
        self.assertIsInstance(model.get_input_embeddings()		,     (nn.Module)      )
        a		:   str	  =	model.get_output_embeddings()
        self.assertTrue(x is None or isinstance(__snake_case		,     nn.Linear      )      )







    def      lowercase_			( self					:    Dict      ):
      a  ,      a		:   Optional[int]	  =	self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a		:   Union[str, Any]	  =	model_class(__snake_case      )
        a		:   Optional[Any]	  =	inspect.signature(model.forward      )
        # signature.parameters is an OrderedDict => so arg_names order is deterministic
        a		:   str	  =	[*signature.parameters.keys()]

        a		:   int	  =	['pixel_values']
        self.assertListEqual(arg_names[:1]		,     __snake_case      )







    def      lowercase_			( self					:    Dict      ):
      a		:   List[Any]	  =	self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*__snake_case      )







    def      lowercase_			( self					:    Optional[Any]      ):
      a		:   List[Any]	  =	self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_video_classification(*__snake_case      )







    @slow
    def      lowercase_			( self					:    Union[str, Any]      ):
      for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        a		:   Optional[int]	  =	TimesformerModel.from_pretrained(__snake_case      )
        self.assertIsNotNone(__snake_case      )







    def      lowercase_			( self					:    Tuple      ):
      if not self.has_attentions:
        pass

      else:
        a  ,      a		:   Dict	  =	self.model_tester.prepare_config_and_inputs_for_common()
        a		:   str	  =	True

        for model_class in self.all_model_classes:
          a		:   Optional[Any]	  =	self.model_tester.seq_length
          a		:   str	  =	self.model_tester.num_frames

          a		:   Any	  =	True
          a		:   Union[str, Any]	  =	False
          a		:   Dict	  =	True
          a		:   Optional[int]	  =	model_class(__snake_case      )
          model.to(__snake_case      )
          model.eval()
          with torch.no_grad():
            a		:   List[Any]	  =	model(**self._prepare_for_class(__snake_case		,     __snake_case      )      )
          a		:   Tuple	  =	outputs.attentions
          self.assertEqual(len(__snake_case      )		,     self.model_tester.num_hidden_layers      )

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

          # attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
          self.assertListEqual(
              list(attentions[0].shape[-3:]      )		,     [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]		,     )
          a		:   Union[str, Any]	  =	len(__snake_case      )

          # Check attention is always last and order is fine
          a		:   List[Any]	  =	True
          a		:   Tuple	  =	True
          a		:   Optional[int]	  =	model_class(__snake_case      )
          model.to(__snake_case      )
          model.eval()
          with torch.no_grad():
            a		:   Optional[int]	  =	model(**self._prepare_for_class(__snake_case		,     __snake_case      )      )

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

          a		:   List[str]	  =	outputs.attentions

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

          # attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
          self.assertListEqual(
              list(self_attentions[0].shape[-3:]      )		,     [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]		,     )







    def      lowercase_			( self					:    int      ):
      def check_hidden_states_output(__snake_case					:    Optional[int]		,     __snake_case					:    Optional[Any]		,     __snake_case					:    Union[str, Any]      ):
        a		:   Dict	  =	model_class(__snake_case      )
        model.to(__snake_case      )
        model.eval()

        with torch.no_grad():
          a		:   Tuple	  =	model(**self._prepare_for_class(__snake_case		,     __snake_case      )      )

        a		:   Tuple	  =	outputs.hidden_states
        a		:   int	  =	self.model_tester.num_hidden_layers + 1
        self.assertEqual(len(__snake_case      )		,     __snake_case      )

        a		:   Optional[int]	  =	self.model_tester.seq_length

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

      a  ,      a		:   str	  =	self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a		:   Optional[int]	  =	True
        check_hidden_states_output(__snake_case		,     __snake_case		,     __snake_case      )

        # check that output_hidden_states also work using config
        del inputs_dict["output_hidden_states"]
        a		:   Optional[Any]	  =	True

        check_hidden_states_output(__snake_case		,     __snake_case		,     __snake_case      )
def  lowerCamelCase__			(		):
  a		:   Optional[int]	  =	hf_hub_download(
      repo_id='hf-internal-testing/spaghetti-video'					,     filename='eating_spaghetti.npy'					,     repo_type='dataset'							)
  a		:   Optional[int]	  =	np.load(_A							)
  return list(_A							)

@require_torch
@require_vision
class 				a__(   unittest.TestCase      ):







    @cached_property
    def      lowercase_			( self					:    Any      ):
      # logits were tested with a different mean and std, so we use the same here
      return (
          VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]		,     image_std=[0.5, 0.5, 0.5]      )
          if is_vision_available()
          else None
      )







    @slow
    def      lowercase_			( self					:    str      ):
      a		:   Dict	  =	TimesformerForVideoClassification.from_pretrained('facebook/timesformer-base-finetuned-k400'      ).to(
          __snake_case      )

      a		:   str	  =	self.default_image_processor
      a		:   Optional[int]	  =	prepare_video()
      a		:   str	  =	image_processor(video[:8]		,     return_tensors='pt'      ).to(__snake_case      )

      # forward pass
      with torch.no_grad():
        a		:   str	  =	model(**__snake_case      )

      # verify the logits
      a		:   List[str]	  =	torch.Size((1, 4_00)      )
      self.assertEqual(outputs.logits.shape		,     __snake_case      )

      a		:   int	  =	torch.tensor([-0.3016, -0.7713, -0.4205]      ).to(__snake_case      )

      self.assertTrue(torch.allclose(outputs.logits[0, :3]		,     __snake_case		,     atol=1e-4      )      )




'''simple docstring'''






# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

from ..models.auto import AutoModelForVisionaSeq
from ..utils import requires_backends
from .base import PipelineTool


if TYPE_CHECKING:
     from PIL import Image

class 				a__(   lowerCamelCase__      ):
    lowercase__					=       """Salesforce/blip-image-captioning-base"""
    lowercase__					=       (
        """This is a tool that generates a description of an image. It takes an input named `image` which should be the """
        """image to caption, and returns a text that contains the description in English."""
    )
    lowercase__					=       """image_captioner"""
    lowercase__					=       AutoModelForVisionaSeq

    lowercase__					=       ["""image"""]
    lowercase__					=       ["""text"""]







    def __init__( self					:    Tuple		,     *__snake_case					:    int		,     **__snake_case					:    List[Any]      ):
      requires_backends(self		,     ['vision']      )
      super().__init__(*__snake_case		,     **__snake_case      )







    def      lowercase_			( self					:    Optional[int]		,     __snake_case					:    "Image"      ):
      return self.pre_processor(images=__snake_case		,     return_tensors='pt'      )







    def      lowercase_			( self					:    Optional[Any]		,     __snake_case					:    str      ):
      return self.model.generate(**__snake_case      )







    def      lowercase_			( self					:    Optional[Any]		,     __snake_case					:    Tuple      ):
      return self.pre_processor.batch_decode(__snake_case		,     skip_special_tokens=__snake_case      )[0].strip()