


'''simple docstring'''

import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VideoMAEConfig
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,
         VideoMAEForPreTraining,
         VideoMAEForVideoClassification,
         VideoMAEModel,
     )
     from transformers.models.videomae.modeling_videomae import VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from transformers import VideoMAEImageProcessor





class      snake_case    :





  """simple docstring"""




  def __init__(    self			,  UpperCamelCase			,  UpperCamelCase=13			,  UpperCamelCase=10			,  UpperCamelCase=3			,  UpperCamelCase=2			,  UpperCamelCase=2			,  UpperCamelCase=2			,  UpperCamelCase=True			,  UpperCamelCase=True			,  UpperCamelCase=32			,  UpperCamelCase=5			,  UpperCamelCase=4			,  UpperCamelCase=37			,  UpperCamelCase="gelu"			,  UpperCamelCase=0.1			,  UpperCamelCase=0.1			,  UpperCamelCase=10			,  UpperCamelCase=0.02			,  UpperCamelCase=0.9			,  UpperCamelCase=None			,  ):







   """simple docstring"""
   lowerCamelCase_      =  parent
   lowerCamelCase_      =  batch_size
   lowerCamelCase_      =  image_size
   lowerCamelCase_      =  num_channels
   lowerCamelCase_      =  patch_size
   lowerCamelCase_      =  tubelet_size
   lowerCamelCase_      =  num_frames
   lowerCamelCase_      =  is_training
   lowerCamelCase_      =  use_labels
   lowerCamelCase_      =  hidden_size
   lowerCamelCase_      =  num_hidden_layers
   lowerCamelCase_      =  num_attention_heads
   lowerCamelCase_      =  intermediate_size
   lowerCamelCase_      =  hidden_act
   lowerCamelCase_      =  hidden_dropout_prob
   lowerCamelCase_      =  attention_probs_dropout_prob
   lowerCamelCase_      =  type_sequence_label_size
   lowerCamelCase_      =  initializer_range
   lowerCamelCase_      =  mask_ratio
   lowerCamelCase_      =  scope

   # in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
   lowerCamelCase_      =  (image_size // patch_size) ** 2
   lowerCamelCase_      =  (num_frames // tubelet_size) * self.num_patches_per_frame

   # use this variable to define bool_masked_pos
   lowerCamelCase_      =  int(mask_ratio * self.seq_length     )






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  floats_tensor(
       [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]     )

   lowerCamelCase_      =  None
   if self.use_labels:
    lowerCamelCase_      =  ids_tensor([self.batch_size]			,  self.type_sequence_label_size     )

   lowerCamelCase_      =  self.get_config()

   return config, pixel_values, labels






  def   snake_case		(    self     ):







   """simple docstring"""
   return VideoMAEConfig(
       image_size=self.image_size			,  patch_size=self.patch_size			,  num_channels=self.num_channels			,  num_frames=self.num_frames			,  tubelet_size=self.tubelet_size			,  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			,  is_decoder=UpperCamelCase			,  initializer_range=self.initializer_range			,  )






  def   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase     ):







   """simple docstring"""
   lowerCamelCase_      =  VideoMAEModel(config=UpperCamelCase     )
   model.to(UpperCamelCase     )
   model.eval()
   lowerCamelCase_      =  model(UpperCamelCase     )
   self.parent.assertEqual(result.last_hidden_state.shape			,  (self.batch_size, self.seq_length, self.hidden_size)     )






  def   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase     ):







   """simple docstring"""
   lowerCamelCase_      =  VideoMAEForPreTraining(UpperCamelCase     )
   model.to(UpperCamelCase     )
   model.eval()
   # important: each video needs to have the same number of masked patches
   # hence we define a single mask, which we then repeat for each example in the batch
   lowerCamelCase_      =  torch.ones((self.num_masks,)     )
   lowerCamelCase_      =  torch.cat([mask, torch.zeros(self.seq_length - mask.size(0     )     )]     )
   lowerCamelCase_      =  mask.expand(self.batch_size			,  -1     ).bool()

   lowerCamelCase_      =  model(UpperCamelCase			,  UpperCamelCase     )
   # model only returns predictions for masked patches
   lowerCamelCase_      =  mask.sum().item()
   lowerCamelCase_      =  3 * self.tubelet_size * self.patch_size**2
   self.parent.assertEqual(result.logits.shape			,  (self.batch_size, num_masked_patches, decoder_num_labels)     )







  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  self.prepare_config_and_inputs()
   lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_      =  config_and_inputs
   lowerCamelCase_      =  {"pixel_values": pixel_values}
   return config, inputs_dict





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





  """simple docstring"""




  _lowerCamelCase		    =  (
      (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
  )
  _lowerCamelCase		    =  (
      {"feature-extraction": VideoMAEModel, "video-classification": VideoMAEForVideoClassification}
      if is_torch_available()
      else {}
  )

  _lowerCamelCase		    =  False
  _lowerCamelCase		    =  False
  _lowerCamelCase		    =  False
  _lowerCamelCase		    =  False






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  VideoMAEModelTester(self     )
   lowerCamelCase_      =  ConfigTester(self			,  config_class=UpperCamelCase			,  has_text_modality=UpperCamelCase			,  hidden_size=37     )






  def   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase=False     ):







   """simple docstring"""
   lowerCamelCase_      =  copy.deepcopy(UpperCamelCase     )

   if model_class == VideoMAEForPreTraining:
    # important: each video needs to have the same number of masked patches
    # hence we define a single mask, which we then repeat for each example in the batch
    lowerCamelCase_      =  torch.ones((self.model_tester.num_masks,)     )
    lowerCamelCase_      =  torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0     )     )]     )
    lowerCamelCase_      =  mask.expand(self.model_tester.batch_size			,  -1     ).bool()
    lowerCamelCase_      =  bool_masked_pos.to(UpperCamelCase     )

   if return_labels:
    if model_class in [
        *get_values(UpperCamelCase     ),
    ]:
     lowerCamelCase_      =  torch.zeros(
         self.model_tester.batch_size			,  dtype=torch.long			,  device=UpperCamelCase     )

   return inputs_dict






  def   snake_case		(    self     ):







   """simple docstring"""
   self.config_tester.run_common_tests()






  @unittest.skip(reason="VideoMAE does not use inputs_embeds"     )
  def   snake_case		(    self     ):







   """simple docstring"""
   pass






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_ ,lowerCamelCase_      =  self.model_tester.prepare_config_and_inputs_for_common()

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






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_ ,lowerCamelCase_      =  self.model_tester.prepare_config_and_inputs_for_common()

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

    lowerCamelCase_      =  ["pixel_values"]
    self.assertListEqual(arg_names[:1]			,  UpperCamelCase     )






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_model(*UpperCamelCase     )






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_for_pretraining(*UpperCamelCase     )






  @slow
  def   snake_case		(    self     ):







   """simple docstring"""
   for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
    lowerCamelCase_      =  VideoMAEModel.from_pretrained(UpperCamelCase     )
    self.assertIsNotNone(UpperCamelCase     )






  def   snake_case		(    self     ):







   """simple docstring"""
   if not self.has_attentions:
    pass

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

    for model_class in self.all_model_classes:
     lowerCamelCase_      =  self.model_tester.seq_length - self.model_tester.num_masks
     lowerCamelCase_      =  (
         num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
     )

     lowerCamelCase_      =  True
     lowerCamelCase_      =  False
     lowerCamelCase_      =  True
     lowerCamelCase_      =  model_class(UpperCamelCase     )
     model.to(UpperCamelCase     )
     model.eval()
     with torch.no_grad():
      lowerCamelCase_      =  model(**self._prepare_for_class(UpperCamelCase			,  UpperCamelCase     )     )
     lowerCamelCase_      =  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"]
     lowerCamelCase_      =  True
     lowerCamelCase_      =  model_class(UpperCamelCase     )
     model.to(UpperCamelCase     )
     model.eval()
     with torch.no_grad():
      lowerCamelCase_      =  model(**self._prepare_for_class(UpperCamelCase			,  UpperCamelCase     )     )
     lowerCamelCase_      =  outputs.attentions
     self.assertEqual(len(UpperCamelCase     )			,  self.model_tester.num_hidden_layers     )

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

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

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

     lowerCamelCase_      =  outputs.attentions

     self.assertEqual(len(UpperCamelCase     )			,  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   snake_case		(    self     ):







   """simple docstring"""
   def check_hidden_states_output(UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase     ):
    lowerCamelCase_      =  model_class(UpperCamelCase     )
    model.to(UpperCamelCase     )
    model.eval()

    with torch.no_grad():
     lowerCamelCase_      =  model(**self._prepare_for_class(UpperCamelCase			,  UpperCamelCase     )     )

    lowerCamelCase_      =  outputs.hidden_states
    lowerCamelCase_      =  self.model_tester.num_hidden_layers + 1
    self.assertEqual(len(UpperCamelCase     )			,  UpperCamelCase     )

    lowerCamelCase_      =  self.model_tester.seq_length - self.model_tester.num_masks
    lowerCamelCase_      =  num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length

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

   lowerCamelCase_ ,lowerCamelCase_      =  self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
    lowerCamelCase_      =  True
    check_hidden_states_output(UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase     )

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

    check_hidden_states_output(UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase     )







  @unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."     )
  def   snake_case		(    self     ):







   """simple docstring"""
   pass





def __snake_case					(      ):
 lowerCamelCase_      =  hf_hub_download(
     repo_id="hf-internal-testing/spaghetti-video"					,				filename="eating_spaghetti.npy"					,				repo_type="dataset"		)
 lowerCamelCase_      =  np.load(UpperCAmelCase_		)
 return list(UpperCAmelCase_		)







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





  """simple docstring"""




  @cached_property
  def   snake_case		(    self     ):







   """simple docstring"""
   # 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   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  VideoMAEForVideoClassification.from_pretrained("MCG-NJU/videomae-base-finetuned-kinetics"     ).to(
       UpperCamelCase     )

   lowerCamelCase_      =  self.default_image_processor
   lowerCamelCase_      =  prepare_video()
   lowerCamelCase_      =  image_processor(UpperCamelCase			,  return_tensors="pt"     ).to(UpperCamelCase     )

   # forward pass
   with torch.no_grad():
    lowerCamelCase_      =  model(**UpperCamelCase     )

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

   lowerCamelCase_      =  torch.tensor([0.3_669, -0.0_688, -0.2_421]     ).to(UpperCamelCase     )

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







  @slow
  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  VideoMAEForPreTraining.from_pretrained("MCG-NJU/videomae-base-short"     ).to(UpperCamelCase     )

   lowerCamelCase_      =  self.default_image_processor
   lowerCamelCase_      =  prepare_video()
   lowerCamelCase_      =  image_processor(UpperCamelCase			,  return_tensors="pt"     ).to(UpperCamelCase     )

   # add boolean mask, indicating which patches to mask
   lowerCamelCase_      =  hf_hub_download(repo_id="hf-internal-testing/bool-masked-pos"			,  filename="bool_masked_pos.pt"     )
   lowerCamelCase_      =  torch.load(UpperCamelCase     )

   # forward pass
   with torch.no_grad():
    lowerCamelCase_      =  model(**UpperCamelCase     )

   # verify the logits
   lowerCamelCase_      =  torch.Size([1, 1408, 1536]     )
   lowerCamelCase_      =  torch.tensor(
       [[0.7_994, 0.9_612, 0.8_508], [0.7_401, 0.8_958, 0.8_302], [0.5_862, 0.7_468, 0.7_325]]			,  device=UpperCamelCase     )
   self.assertEqual(outputs.logits.shape			,  UpperCamelCase     )
   self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]			,  UpperCamelCase			,  atol=1e-4     )     )

   # verify the loss (`config.norm_pix_loss` = `True`)
   lowerCamelCase_      =  torch.tensor([0.5_142]			,  device=UpperCamelCase     )
   self.assertTrue(torch.allclose(outputs.loss			,  UpperCamelCase			,  atol=1e-4     )     )

   # verify the loss (`config.norm_pix_loss` = `False`)
   lowerCamelCase_      =  VideoMAEForPreTraining.from_pretrained("MCG-NJU/videomae-base-short"			,  norm_pix_loss=UpperCamelCase     ).to(
       UpperCamelCase     )

   with torch.no_grad():
    lowerCamelCase_      =  model(**UpperCamelCase     )

   lowerCamelCase_      =  torch.tensor(torch.tensor([0.6_469]     )			,  device=UpperCamelCase     )
   self.assertTrue(torch.allclose(outputs.loss			,  UpperCamelCase			,  atol=1e-4     )     )









import json
import os
from datetime import date
from pathlib import Path

from tabulate import DataRow, TableFormat, tabulate


UpperCAmelCase_									=	TableFormat(
    lineabove=None,
    linebelowheader=None,
    linebetweenrows=None,
    linebelow=None,
    headerrow=DataRow('', '|', '|'),
    datarow=DataRow('', '|', '|'),
    padding=1,
    with_header_hide=None,
)


UpperCAmelCase_									=	[]
UpperCAmelCase_									=	[]

UpperCAmelCase_									=	{'type': 'section', 'text': {'type': 'plain_text', 'text': 'No failed tests! 🤗', 'emoji': True}}

UpperCAmelCase_									=	[
    {
        'type': 'header',
        'text': {
            'type': 'plain_text',
            'text': f"""🤗 Accelerate nightly {os.environ.get("TEST_TYPE", "")} test results""",
            'emoji': True,
        },
    }
]

UpperCAmelCase_									=	0
for log in Path().glob('*.log'):
       UpperCAmelCase_									=	0
       with open(log, 'r') as f:
              for line in f:
                     UpperCAmelCase_									=	json.loads(line)
                     if line.get('nodeid', '') != "":
                            UpperCAmelCase_									=	line['nodeid']
                            if line.get('duration', None) is not None:
                                   UpperCAmelCase_									=	f"""{line["duration"]:.4f}"""
                                   if line.get('outcome', '') == "failed":
                                          section_num_failed += 1
                                          failed.append([test, duration, log.name.split('_')[0]])
                                          total_num_failed += 1
       group_info.append([str(log), section_num_failed, failed])
       UpperCAmelCase_									=	[]
       log.unlink()

UpperCAmelCase_									=	''
UpperCAmelCase_									=	[]
if total_num_failed > 0:
       for name, num_failed, failed_tests in group_info:
              if num_failed > 0:
                     if num_failed == 1:
                            message += f"*{name[1:]}: {num_failed} failed test*\n"
                     else:
                            message += f"*{name[1:]}: {num_failed} failed tests*\n"
                     UpperCAmelCase_									=	[]
                     UpperCAmelCase_									=	{}
                     for test in failed_tests:
                            UpperCAmelCase_									=	test[0].split('::')
                            UpperCAmelCase_									=	data[0].split('/')[-1]
                            if data[0] not in filesafailed:
                                   UpperCAmelCase_									=	[data[1:]]
                            else:
                                   filesafailed[data[0]] += [data[1:]]
                            failed_table.append(data)

                     UpperCAmelCase_									=	[test[0] for test in failed_table]
                     UpperCAmelCase_									=	list(set(files))
                     # Count number of instances in failed_tests
                     UpperCAmelCase_									=	[]
                     for file in individual_files:
                            table.append([file, len(filesafailed[file])])

                     UpperCAmelCase_									=	tabulate(
                         table,
                         headers=['Test Location', 'Num Failed'],
                         tablefmt=hf_table_format,
                         stralign='right',
                     )
                     message += f"\n```\n{failed_table}\n```"
                     all_filesafailed.append(filesafailed)
       if len(message) > 3_000:
              UpperCAmelCase_									=	'Too many failed tests, please see the full report in the Action results.'
              UpperCAmelCase_									=	len(err) + 10
              UpperCAmelCase_									=	message[: 3_000 - offset] + f"""\n...\n```\n{err}"""
       print(f"""### {message}""")
else:
       UpperCAmelCase_									=	'No failed tests! 🤗'
       print(f"""## {message}""")
       payload.append(no_error_payload)

if os.environ.get('TEST_TYPE', '') != "":
       from slack_sdk import WebClient

       UpperCAmelCase_									=	WebClient(token=os.environ['SLACK_API_TOKEN'])
       if message != "No failed tests! 🤗":
              UpperCAmelCase_									=	{
                  'type': 'section',
                  'text': {
                      'type': 'mrkdwn',
                      'text': message,
                  },
              }
              payload.append(md_report)
              UpperCAmelCase_									=	{
                  'type': 'section',
                  'text': {
                      'type': 'mrkdwn',
                      'text': '*For more details:*',
                  },
                  'accessory': {
                      'type': 'button',
                      'text': {
                          'type': 'plain_text',
                          'text': 'Check Action results',
                          'emoji': True,
                      },
                      'url': f"""https://github.com/{os.environ["GITHUB_REPOSITORY"]}/actions/runs/{os.environ["GITHUB_RUN_ID"]}""",
                  },
              }
              payload.append(action_button)
              UpperCAmelCase_									=	{
                  'type': 'context',
                  'elements': [
                      {
                          'type': 'plain_text',
                          'text': f"""Nightly {os.environ.get("TEST_TYPE")} test results for {date.today()}""",
                      }
                  ],
              }
              payload.append(date_report)
       UpperCAmelCase_									=	client.chat_postMessage(channel='#accelerate-ci-daily', text=message, blocks=payload)
       UpperCAmelCase_									=	response.data['ts']
       for failed_file in all_filesafailed:
              for test_location, test_failures in failed_file.items():
                     # Keep only the first instance of the test name
                     UpperCAmelCase_									=	''
                     for i, row in enumerate(test_failures):
                            if row[0] != test_class:
                                   UpperCAmelCase_									=	row[0]
                            else:
                                   UpperCAmelCase_									=	''

                     UpperCAmelCase_									=	{
                         'type': 'section',
                         'text': {
                             'type': 'mrkdwn',
                             'text': f"""Test location: {test_location}\n```\n{tabulate(test_failures, headers=["Class", "Test"], tablefmt=hf_table_format, stralign="right")}\n```""",
                         },
                     }

                     client.chat_postMessage(
                         channel='#accelerate-ci-daily',
                         thread_ts=ts,
                         blocks=[payload],
                     )

