



'''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				:						Union[str, Any]		,    __snake_case				:						Dict		,    __snake_case				:						str=13		,    __snake_case				:						Dict=10		,    __snake_case				:						int=3		,    __snake_case				:						str=2		,    __snake_case				:						List[str]=2		,    __snake_case				:						Tuple=True		,    __snake_case				:						str=True		,    __snake_case				:						str=32		,    __snake_case				:						List[Any]=5		,    __snake_case				:						Union[str, Any]=4		,    __snake_case				:						List[str]=37		,    __snake_case				:						str="gelu"		,    __snake_case				:						Optional[Any]=0.1		,    __snake_case				:						Dict=0.1		,    __snake_case				:						Tuple=10		,    __snake_case				:						List[str]=0.02		,    __snake_case				:						Tuple="divided_space_time"		,    __snake_case				:						int=None		,    ):
      a		:      Union[str, Any]       =  parent
      a		:      Any       =  batch_size
      a		:      Dict       =  image_size
      a		:      Any       =  num_channels
      a		:      Optional[int]       =  patch_size
      a		:      List[Any]       =  num_frames
      a		:      Tuple       =  is_training
      a		:      str       =  use_labels
      a		:      int       =  hidden_size
      a		:      List[Any]       =  num_hidden_layers
      a		:      List[Any]       =  num_attention_heads
      a		:      int       =  intermediate_size
      a		:      Dict       =  hidden_act
      a		:      List[Any]       =  hidden_dropout_prob
      a		:      List[str]       =  attention_probs_dropout_prob
      a		:      Tuple       =  attention_type
      a		:      Union[str, Any]       =  initializer_range
      a		:      str       =  scope
      a		:      List[str]       =  num_labels

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







    def     lowercase_ (   self				:						List[str]			):
      a		:      Dict       =  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		:      str       =  ids_tensor([self.batch_size]		,    self.num_labels			)

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

      return config, pixel_values, labels







    def     lowercase_ (   self				:						int			):
      a		:      Dict       =  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[Any]       =  self.num_labels
      return config







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







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

      a		:      List[Any]       =  model(__UpperCamelCase			)

      # verify the logits shape
      a		:      Dict       =  torch.Size((self.batch_size, self.num_labels)			)
      self.parent.assertEqual(result.logits.shape		,    __UpperCamelCase			)







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

@require_torch
class 							a__(      _UpperCAmelCase	,				_UpperCAmelCase	,				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[str]			):
      a		:      Union[str, Any]       =  TimesformerModelTester(self			)
      a		:      Optional[Any]       =  ConfigTester(
          self		,    config_class=__UpperCamelCase		,    has_text_modality=__UpperCamelCase		,    hidden_size=37			)







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

      if return_labels:
        if model_class in get_values(__UpperCamelCase			):
          a		:      List[Any]       =  torch.zeros(
              self.model_tester.batch_size		,    dtype=torch.long		,    device=__UpperCamelCase			)

      return inputs_dict







    def     lowercase_ (   self				:						Union[str, Any]			):
      self.config_tester.run_common_tests()







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







    def     lowercase_ (   self				:						str			):
      a     ,			a		:      Dict       =  self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a		:      Dict       =  model_class(__UpperCamelCase			)
        self.assertIsInstance(model.get_input_embeddings()		,    (nn.Module)			)
        a		:      Optional[int]       =  model.get_output_embeddings()
        self.assertTrue(x is None or isinstance(__UpperCamelCase		,    nn.Linear			)			)







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

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

        a		:      Any       =  ['pixel_values']
        self.assertListEqual(arg_names[:1]		,    __UpperCamelCase			)







    def     lowercase_ (   self				:						Union[str, Any]			):
      a		:      Any       =  self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*__UpperCamelCase			)







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







    @slow
    def     lowercase_ (   self				:						Optional[int]			):






      for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        a		:      str       =  TimesformerModel.from_pretrained(__UpperCamelCase			)
        self.assertIsNotNone(__UpperCamelCase			)







    def     lowercase_ (   self				:						int			):






      if not self.has_attentions:
        pass

      else:
        a     ,			a		:      str       =  self.model_tester.prepare_config_and_inputs_for_common()
        a		:      Union[str, Any]       =  True

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

          a		:      str       =  True
          a		:      Dict       =  False
          a		:      str       =  True
          a		:      Optional[int]       =  model_class(__UpperCamelCase			)
          model.to(__UpperCamelCase			)
          model.eval()
          with torch.no_grad():
            a		:      List[str]       =  model(**self._prepare_for_class(__UpperCamelCase		,    __UpperCamelCase			)			)
          a		:      int       =  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"]
          a		:      Dict       =  True
          a		:      List[str]       =  model_class(__UpperCamelCase			)
          model.to(__UpperCamelCase			)
          model.eval()
          with torch.no_grad():
            a		:      Any       =  model(**self._prepare_for_class(__UpperCamelCase		,    __UpperCamelCase			)			)
          a		:      str       =  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]		,    )
          a		:      str       =  len(__UpperCamelCase			)

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

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

          a		:      Any       =  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				:						Union[str, Any]			):






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

        with torch.no_grad():
          a		:      Any       =  model(**self._prepare_for_class(__UpperCamelCase		,    __UpperCamelCase			)			)

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

        a		:      Union[str, Any]       =  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		:      Tuple       =  True
        check_hidden_states_output(__UpperCamelCase		,    __UpperCamelCase		,    __UpperCamelCase			)

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

        check_hidden_states_output(__UpperCamelCase		,    __UpperCamelCase		,    __UpperCamelCase			)
def    lowerCamelCase__ (		):
  a		:      Dict       =  hf_hub_download(
      repo_id='hf-internal-testing/spaghetti-video'				,    filename='eating_spaghetti.npy'				,    repo_type='dataset'       )
  a		:      str       =  np.load(_A       )
  return list(_A       )

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







    @cached_property
    def     lowercase_ (   self				:						Tuple			):
      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		:      Optional[Any]       =  TimesformerForVideoClassification.from_pretrained('facebook/timesformer-base-finetuned-k400'			).to(
          __UpperCamelCase			)

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

      # forward pass
      with torch.no_grad():
        a		:      Union[str, Any]       =  model(**__UpperCamelCase			)

      # verify the logits
      a		:      Tuple       =  torch.Size((1, 4_00)			)
      self.assertEqual(outputs.logits.shape		,    __UpperCamelCase			)

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

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




'''simple docstring'''






import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


lowerCAmelCase:      Any		 =		False

@skip_mps
class 							a__(      lowerCamelCase__	,				lowerCamelCase__	,				lowerCamelCase__	,				unittest.TestCase			):
    lowercase__				=					StableDiffusionAttendAndExcitePipeline
    lowercase__				=					False
    lowercase__				=					TEXT_TO_IMAGE_PARAMS
    lowercase__				=					TEXT_TO_IMAGE_BATCH_PARAMS.union({"""token_indices"""}			)
    lowercase__				=					TEXT_TO_IMAGE_IMAGE_PARAMS
    lowercase__				=					TEXT_TO_IMAGE_IMAGE_PARAMS







    @classmethod
    def     lowercase_ (   cls				:						Union[str, Any]			):
      super().setUpClass()
      torch.use_deterministic_algorithms(__snake_case			)







    @classmethod
    def     lowercase_ (   cls				:						Optional[Any]			):
      super().tearDownClass()
      torch.use_deterministic_algorithms(__snake_case			)







    def     lowercase_ (   self				:						List[Any]			):
      torch.manual_seed(0			)
      a		:      List[Any]       =  UNetaDConditionModel(
          block_out_channels=(32, 64)		,    layers_per_block=1		,    sample_size=32		,    in_channels=4		,    out_channels=4		,    down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')		,    up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')		,    cross_attention_dim=32		,    attention_head_dim=(2, 4)		,    use_linear_projection=__snake_case		,    )
      a		:      Optional[int]       =  DDIMScheduler(
          beta_start=0.00085		,    beta_end=0.012		,    beta_schedule='scaled_linear'		,    clip_sample=__snake_case		,    set_alpha_to_one=__snake_case		,    )
      torch.manual_seed(0			)
      a		:      str       =  AutoencoderKL(
          block_out_channels=[32, 64]		,    in_channels=3		,    out_channels=3		,    down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']		,    up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']		,    latent_channels=4		,    sample_size=1_28		,    )
      torch.manual_seed(0			)
      a		:      str       =  CLIPTextConfig(
          bos_token_id=0		,    eos_token_id=2		,    hidden_size=32		,    intermediate_size=37		,    layer_norm_eps=1e-0_5		,    num_attention_heads=4		,    num_hidden_layers=5		,    pad_token_id=1		,    vocab_size=10_00		,    hidden_act='gelu'		,    projection_dim=5_12		,    )
      a		:      Any       =  CLIPTextModel(__snake_case			)
      a		:      Tuple       =  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'			)

      a		:      List[str]       =  {
          'unet': unet,
          'scheduler': scheduler,
          'vae': vae,
          'text_encoder': text_encoder,
          'tokenizer': tokenizer,
          'safety_checker': None,
          'feature_extractor': None,
      }

      return components







    def     lowercase_ (   self				:						str		,    __snake_case				:						Tuple		,    __snake_case				:						Optional[int]=0			):
      if str(__snake_case			).startswith('mps'			):
        a		:      Any       =  torch.manual_seed(__snake_case			)
      else:
        a		:      Any       =  torch.Generator(device=__snake_case			).manual_seed(__snake_case			)
      a		:      int       =  {
          'prompt': 'a cat and a frog',
          'token_indices': [2, 5],
          'generator': generator,
          'num_inference_steps': 1,
          'guidance_scale': 6.0,
          'output_type': 'numpy',
          'max_iter_to_alter': 2,
          'thresholds': {0: 0.7},
      }
      return inputs







    def     lowercase_ (   self				:						List[Any]			):
      a		:      Union[str, Any]       =  'cpu'

      a		:      Any       =  self.get_dummy_components()
      a		:      List[str]       =  self.pipeline_class(**__snake_case			)
      pipe.to(__snake_case			)
      pipe.set_progress_bar_config(disable=__snake_case			)

      a		:      Any       =  self.get_dummy_inputs(__snake_case			)
      a		:      int       =  pipe(**__snake_case			).images
      a		:      int       =  image[0, -3:, -3:, -1]

      self.assertEqual(image.shape		,    (1, 64, 64, 3)			)
      a		:      str       =  np.array(
          [0.63905364, 0.62897307, 0.48599017, 0.5133624, 0.5550048, 0.45769516, 0.50326973, 0.5023139, 0.45384496]			)
      a		:      int       =  np.abs(image_slice.flatten() - expected_slice			).max()
      self.assertLessEqual(__snake_case		,    1e-3			)







    def     lowercase_ (   self				:						Dict			):
      super().test_cpu_offload_forward_pass(expected_max_diff=5e-4			)







    def     lowercase_ (   self				:						Tuple			):
      # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
      self._test_inference_batch_consistent(batch_sizes=[1, 2]			)







    def     lowercase_ (   self				:						Union[str, Any]			):
      self._test_inference_batch_single_identical(batch_size=2		,    expected_max_diff=7e-4			)







    def     lowercase_ (   self				:						Tuple			):
      super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3			)







    def     lowercase_ (   self				:						str			):
      super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4			)







    def     lowercase_ (   self				:						Any			):
      super().test_save_load_local(expected_max_difference=5e-4			)







    def     lowercase_ (   self				:						List[Any]			):
      super().test_save_load_optional_components(expected_max_difference=4e-4			)

@require_torch_gpu
@slow
class 							a__(      unittest.TestCase			):







    @classmethod
    def     lowercase_ (   cls				:						Union[str, Any]			):
      super().setUpClass()
      torch.use_deterministic_algorithms(__snake_case			)







    @classmethod
    def     lowercase_ (   cls				:						Union[str, Any]			):
      super().tearDownClass()
      torch.use_deterministic_algorithms(__snake_case			)







    def     lowercase_ (   self				:						Union[str, Any]			):
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()







    def     lowercase_ (   self				:						Optional[Any]			):
      a		:      List[Any]       =  torch.manual_seed(51			)

      a		:      Dict       =  StableDiffusionAttendAndExcitePipeline.from_pretrained(
          'CompVis/stable-diffusion-v1-4'		,    safety_checker=__snake_case		,    torch_dtype=torch.floataa			)
      pipe.to('cuda'			)

      a		:      Optional[Any]       =  'a painting of an elephant with glasses'
      a		:      Any       =  [5, 7]

      a		:      Tuple       =  pipe(
          prompt=__snake_case		,    token_indices=__snake_case		,    guidance_scale=7.5		,    generator=__snake_case		,    num_inference_steps=5		,    max_iter_to_alter=5		,    output_type='numpy'		,    ).images[0]

      a		:      str       =  load_numpy(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'			)
      assert np.abs((expected_image - image).max()			) < 5e-1