

"""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  _lowerCAmelCase  :







 """simple docstring"""







 def __init__(      self							:      List[Any], UpperCAmelCase__							:      Tuple, UpperCAmelCase__							:      List[Any]=1_3, UpperCAmelCase__							:      Union[str, Any]=1_0, UpperCAmelCase__							:      Dict=3, UpperCAmelCase__							:      List[Any]=2, UpperCAmelCase__							:      List[Any]=2, UpperCAmelCase__							:      str=True, UpperCAmelCase__							:      Any=True, UpperCAmelCase__							:      Optional[int]=3_2, UpperCAmelCase__							:      str=5, UpperCAmelCase__							:      Dict=4, UpperCAmelCase__							:      str=3_7, UpperCAmelCase__							:      str="gelu", UpperCAmelCase__							:      Tuple=0.1, UpperCAmelCase__							:      Tuple=0.1, UpperCAmelCase__							:      Any=1_0, UpperCAmelCase__							:      Optional[Any]=0.02, UpperCAmelCase__							:      List[Any]="divided_space_time", UpperCAmelCase__							:      int=None, ):
    __lowercase								=						parent
    __lowercase								=						batch_size
    __lowercase								=						image_size
    __lowercase								=						num_channels
    __lowercase								=						patch_size
    __lowercase								=						num_frames
    __lowercase								=						is_training
    __lowercase								=						use_labels
    __lowercase								=						hidden_size
    __lowercase								=						num_hidden_layers
    __lowercase								=						num_attention_heads
    __lowercase								=						intermediate_size
    __lowercase								=						hidden_act
    __lowercase								=						hidden_dropout_prob
    __lowercase								=						attention_probs_dropout_prob
    __lowercase								=						attention_type
    __lowercase								=						initializer_range
    __lowercase								=						scope
    __lowercase								=						num_labels

    # in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
    __lowercase								=						(image_size // patch_size) ** 2
    __lowercase								=						(num_frames) * self.num_patches_per_frame + 1


 def  _lowercase			(      self							:      Tuple     ):
    __lowercase								=						floats_tensor(
        [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]     )

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

    __lowercase								=						self.get_config()

    return config, pixel_values, labels


 def  _lowercase			(      self							:      List[str]     ):
    __lowercase								=						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, )
    __lowercase								=						self.num_labels
    return config


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


 def  _lowercase			(      self							:      Optional[int], UpperCAmelCase__							:      Union[str, Any], UpperCAmelCase__							:      Dict, UpperCAmelCase__							:      int     ):
    __lowercase								=						TimesformerForVideoClassification(UpperCAmelCase__     )
    model.to(UpperCAmelCase__     )
    model.eval()

    __lowercase								=						model(UpperCAmelCase__     )

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



 def  _lowercase			(      self							:      Optional[int]     ):
    __lowercase								=						self.prepare_config_and_inputs()
    __lowercase		,__lowercase		,__lowercase								=						config_and_inputs
    __lowercase								=						{"pixel_values": pixel_values}
    return config, inputs_dict







@require_torch
class  _lowerCAmelCase  (     lowercase					,lowercase					,unittest.TestCase       ):







 """simple docstring"""







 __UpperCAmelCase     : Any       		=     (TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
 __UpperCAmelCase     : Optional[int]       		=     (
     {"feature-extraction": TimesformerModel, "video-classification": TimesformerForVideoClassification}
     if is_torch_available()
     else {}
 )

 __UpperCAmelCase     : List[Any]       		=     False
 __UpperCAmelCase     : str       		=     False
 __UpperCAmelCase     : int       		=     False
 __UpperCAmelCase     : Dict       		=     False


 def  _lowercase			(      self							:      Union[str, Any]     ):
    __lowercase								=						TimesformerModelTester(self     )
    __lowercase								=						ConfigTester(
        self, config_class=UpperCAmelCase__, has_text_modality=UpperCAmelCase__, hidden_size=3_7     )


 def  _lowercase			(      self							:      Union[str, Any], UpperCAmelCase__							:      Tuple, UpperCAmelCase__							:      Any, UpperCAmelCase__							:      Optional[int]=False     ):
    __lowercase								=						copy.deepcopy(UpperCAmelCase__     )

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

    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							:      str     ):
    pass


 def  _lowercase			(      self							:      Optional[int]     ):
    __lowercase		,__lowercase								=						self.model_tester.prepare_config_and_inputs_for_common()

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


 def  _lowercase			(      self							:      str     ):
    __lowercase		,__lowercase								=						self.model_tester.prepare_config_and_inputs_for_common()

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

       __lowercase								=						["pixel_values"]
       self.assertListEqual(arg_names[:1], UpperCAmelCase__     )


 def  _lowercase			(      self							:      Tuple     ):
    __lowercase								=						self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_model(*UpperCAmelCase__     )


 def  _lowercase			(      self							:      Any     ):
    __lowercase								=						self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_for_video_classification(*UpperCAmelCase__     )


 @slow
 def  _lowercase			(      self							:      int     ):
    for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
       __lowercase								=						TimesformerModel.from_pretrained(UpperCAmelCase__     )
       self.assertIsNotNone(UpperCAmelCase__     )


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

    else:
       __lowercase		,__lowercase								=						self.model_tester.prepare_config_and_inputs_for_common()
       __lowercase								=						True

       for model_class in self.all_model_classes:
          __lowercase								=						self.model_tester.seq_length
          __lowercase								=						self.model_tester.num_frames

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

          # check that output_attentions also work using config
          del inputs_dict["output_attentions"]
          __lowercase								=						True
          __lowercase								=						model_class(UpperCAmelCase__     )
          model.to(UpperCAmelCase__     )
          model.eval()
          with torch.no_grad():
             __lowercase								=						model(**self._prepare_for_class(UpperCAmelCase__, UpperCAmelCase__     )     )
          __lowercase								=						outputs.attentions
          self.assertEqual(len(UpperCAmelCase__     ), 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], )
          __lowercase								=						len(UpperCAmelCase__     )

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

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

          __lowercase								=						outputs.attentions

          self.assertEqual(len(UpperCAmelCase__     ), 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							:      Dict     ):
    def check_hidden_states_output(UpperCAmelCase__							:      int, UpperCAmelCase__							:      Union[str, Any], UpperCAmelCase__							:      Union[str, Any]     ):
       __lowercase								=						model_class(UpperCAmelCase__     )
       model.to(UpperCAmelCase__     )
       model.eval()

       with torch.no_grad():
          __lowercase								=						model(**self._prepare_for_class(UpperCAmelCase__, UpperCAmelCase__     )     )

       __lowercase								=						outputs.hidden_states
       __lowercase								=						self.model_tester.num_hidden_layers + 1
       self.assertEqual(len(UpperCAmelCase__     ), UpperCAmelCase__     )

       __lowercase								=						self.model_tester.seq_length

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

    __lowercase		,__lowercase								=						self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
       __lowercase								=						True
       check_hidden_states_output(UpperCAmelCase__, UpperCAmelCase__, UpperCAmelCase__     )

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

       check_hidden_states_output(UpperCAmelCase__, UpperCAmelCase__, UpperCAmelCase__     )







def   _A  (  )     ->						Optional[int]:
   '''simple docstring'''


   __lowercase								=						hf_hub_download(
       repo_id="hf-internal-testing/spaghetti-video",      filename="eating_spaghetti.npy",      repo_type="dataset")
   __lowercase								=						np.load(UpperCamelCase_)
   return list(UpperCamelCase_)




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







 """simple docstring"""







 @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							:      Optional[int]     ):
    __lowercase								=						TimesformerForVideoClassification.from_pretrained("facebook/timesformer-base-finetuned-k400"     ).to(
        UpperCAmelCase__     )

    __lowercase								=						self.default_image_processor
    __lowercase								=						prepare_video()
    __lowercase								=						image_processor(video[:8], return_tensors="pt"     ).to(UpperCAmelCase__     )

    # forward pass
    with torch.no_grad():
       __lowercase								=						model(**UpperCAmelCase__     )

    # verify the logits
    __lowercase								=						torch.Size((1, 4_0_0)     )
    self.assertEqual(outputs.logits.shape, UpperCAmelCase__     )

    __lowercase								=						torch.tensor([-0.3_016, -0.7_713, -0.4_205]     ).to(UpperCAmelCase__     )

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







"""simple docstring"""





from unittest import TestCase

from datasets import Dataset
from minhash_deduplication import deduplicate_dataset, make_duplicate_clusters




def   _A  (  )     ->						Dict:
   '''simple docstring'''


   __lowercase								=						{
       "repo_name": ["test_repo1", "test_repo2", "test_repo3"],
       "path": ["test_1.py", "test_2.py", "unit_test.py"],
       "content": ["a " * 20, "a " * 30, "b " * 7],
   }
   __lowercase								=						Dataset.from_dict(UpperCamelCase_)
   return dataset




class  _lowerCAmelCase  (     lowercase       ):







 """simple docstring"""







 def  _lowercase			(      self							:      int     ):
    __lowercase								=						get_dataset()
    __lowercase								=						make_duplicate_clusters(UpperCAmelCase__, 0.85     )
    self.assertEqual(len(duplicate_clusters[0]     ), 2     )



 def  _lowercase			(      self							:      Any     ):
    __lowercase								=						get_dataset()
    __lowercase		,__lowercase								=						deduplicate_dataset(UpperCAmelCase__     )
    self.assertEqual(len(UpperCAmelCase__     ), 2     )
    print(UpperCAmelCase__     )
    self.assertEqual(duplicate_clusters[0][0]["copies"], 2     )
    self.assertEqual(duplicate_clusters[0][0]["is_extreme"], UpperCAmelCase__     )




