



'''simple docstring'''





import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()






@skip_mps
class __A    (						A     ,     unittest.TestCase		):



 '''simple docstring'''







 __lowerCamelCase					: str    = TextToVideoSDPipeline
 __lowerCamelCase					: Optional[Any]    = TEXT_TO_IMAGE_PARAMS
 __lowerCamelCase					: str    = TEXT_TO_IMAGE_BATCH_PARAMS
 # No `output_type`.
 __lowerCamelCase					: Tuple    = frozenset(
     [
         'num_inference_steps',
         'generator',
         'latents',
         'return_dict',
         'callback',
         'callback_steps',
     ]		)



 def       a__						(self )					->  Optional[int]:


      """simple docstring"""



      torch.manual_seed(0 )
      _a											=						UNetaDConditionModel(
          block_out_channels=(32, 64, 64, 64)				,     layers_per_block=2				,     sample_size=32				,     in_channels=4				,     out_channels=4				,     down_block_types=('''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''DownBlock3D''')				,     up_block_types=('''UpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''')				,     cross_attention_dim=32				,     attention_head_dim=4				,     )
      _a											=						DDIMScheduler(
          beta_start=0.00085				,     beta_end=0.012				,     beta_schedule='''scaled_linear'''				,     clip_sample=A				,     set_alpha_to_one=A				,     )
      torch.manual_seed(0 )
      _a											=						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=128				,     )
      torch.manual_seed(0 )
      _a											=						CLIPTextConfig(
          bos_token_id=0				,     eos_token_id=2				,     hidden_size=32				,     intermediate_size=37				,     layer_norm_eps=1E-05				,     num_attention_heads=4				,     num_hidden_layers=5				,     pad_token_id=1				,     vocab_size=1_000				,     hidden_act='''gelu'''				,     projection_dim=512				,     )
      _a											=						CLIPTextModel(A )
      _a											=						CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''' )

      _a											=						{
          '''unet''': unet,
          '''scheduler''': scheduler,
          '''vae''': vae,
          '''text_encoder''': text_encoder,
          '''tokenizer''': tokenizer,
      }
      return components



 def       a__						(self				,     A				,     A=0 )					->  Dict:


      """simple docstring"""



      if str(A ).startswith('''mps''' ):
           _a											=						torch.manual_seed(A )
      else:
           _a											=						torch.Generator(device=A ).manual_seed(A )
      _a											=						{
          '''prompt''': '''A painting of a squirrel eating a burger''',
          '''generator''': generator,
          '''num_inference_steps''': 2,
          '''guidance_scale''': 6.0,
          '''output_type''': '''pt''',
      }
      return inputs



 def       a__						(self )					->  Union[str, Any]:


      """simple docstring"""



      _a											=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
      _a											=						self.get_dummy_components()
      _a											=						TextToVideoSDPipeline(**A )
      _a											=						sd_pipe.to(A )
      sd_pipe.set_progress_bar_config(disable=A )

      _a											=						self.get_dummy_inputs(A )
      _a											=						'''np'''
      _a											=						sd_pipe(**A ).frames
      _a											=						frames[0][-3:, -3:, -1]

      assert frames[0].shape == (64, 64, 3)
      _a											=						np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0] )

      assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-2



 def       a__						(self )					->  Optional[int]:


      """simple docstring"""



      self._test_attention_slicing_forward_pass(test_mean_pixel_difference=A				,     expected_max_diff=3E-3 )



 @unittest.skipIf(
     torch_device != '''cuda''' or not is_xformers_available()				,     reason='''XFormers attention is only available with CUDA and `xformers` installed'''				,     )
 def       a__						(self )					->  int:


      """simple docstring"""



      self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A				,     expected_max_diff=1E-2 )



 @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.''' )
 def       a__						(self )					->  Dict:


      """simple docstring"""



      pass



 @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.''' )
 def       a__						(self )					->  Any:


      """simple docstring"""



      pass



 @unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.''' )
 def       a__						(self )					->  Optional[Any]:


      """simple docstring"""



      pass



 def       a__						(self )					->  Union[str, Any]:


      """simple docstring"""



      return super().test_progress_bar()






@slow
@skip_mps
class __A    (						unittest.TestCase		):



 '''simple docstring'''







 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						load_numpy(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy''' )

      _a											=						TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b''' )
      _a											=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
      _a											=						pipe.to('''cuda''' )

      _a											=						'''Spiderman is surfing'''
      _a											=						torch.Generator(device='''cpu''' ).manual_seed(0 )

      _a											=						pipe(A				,     generator=A				,     num_inference_steps=25				,     output_type='''pt''' ).frames
      _a											=						video_frames.cpu().numpy()

      assert np.abs(expected_video - video ).mean() < 5E-2



 def       a__						(self )					->  Union[str, Any]:


      """simple docstring"""



      _a											=						load_numpy(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy''' )

      _a											=						TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b''' )
      _a											=						pipe.to('''cuda''' )

      _a											=						'''Spiderman is surfing'''
      _a											=						torch.Generator(device='''cpu''' ).manual_seed(0 )

      _a											=						pipe(A				,     generator=A				,     num_inference_steps=2				,     output_type='''pt''' ).frames
      _a											=						video_frames.cpu().numpy()

      assert np.abs(expected_video - video ).mean() < 5E-2








'''simple docstring'''





import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoImageProcessor, ViTImageProcessor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_image_processing import CustomImageProcessor  # noqa E402


lowercase_	 =						get_tests_dir("fixtures")






class __A    (						unittest.TestCase		):



 '''simple docstring'''







 def       a__						(self )					->  List[Any]:


      """simple docstring"""



      _a											=						mock.Mock()
      _a											=						500
      _a											=						{}
      _a											=						HTTPError
      _a											=						{}

      # Download this model to make sure it's in the cache.
      _a											=						ViTImageProcessor.from_pretrained('''hf-internal-testing/tiny-random-vit''' )
      # Under the mock environment we get a 500 error when trying to reach the model.
      with mock.patch('''requests.Session.request'''				,     return_value=A ) as mock_head:
           _a											=						ViTImageProcessor.from_pretrained('''hf-internal-testing/tiny-random-vit''' )
           # This check we did call the fake head request
           mock_head.assert_called()



 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						ViTImageProcessor.from_pretrained(
          '''https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json''' )



 def       a__						(self )					->  Dict:


      """simple docstring"""



      with self.assertRaises(A ):
           # config is in subfolder, the following should not work without specifying the subfolder
           _a											=						AutoImageProcessor.from_pretrained('''hf-internal-testing/stable-diffusion-all-variants''' )

      _a											=						AutoImageProcessor.from_pretrained(
          '''hf-internal-testing/stable-diffusion-all-variants'''				,     subfolder='''feature_extractor''' )

      self.assertIsNotNone(A )






@is_staging_test
class __A    (						unittest.TestCase		):



 '''simple docstring'''







 @classmethod
 def       a__						(cls )					->  Any:


      """simple docstring"""



      _a											=						TOKEN
      HfFolder.save_token(A )



 @classmethod
 def       a__						(cls )					->  Union[str, Any]:


      """simple docstring"""



      try:
           delete_repo(token=cls._token				,     repo_id='''test-image-processor''' )
      except HTTPError:
           pass

      try:
           delete_repo(token=cls._token				,     repo_id='''valid_org/test-image-processor-org''' )
      except HTTPError:
           pass

      try:
           delete_repo(token=cls._token				,     repo_id='''test-dynamic-image-processor''' )
      except HTTPError:
           pass



 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						ViTImageProcessor.from_pretrained(A )
      image_processor.push_to_hub('''test-image-processor'''				,     use_auth_token=self._token )

      _a											=						ViTImageProcessor.from_pretrained(f'''{USER}/test-image-processor''' )
      for k, v in image_processor.__dict__.items():
           self.assertEqual(A				,     getattr(A				,     A ) )

      # Reset repo
      delete_repo(token=self._token				,     repo_id='''test-image-processor''' )

      # Push to hub via save_pretrained
      with tempfile.TemporaryDirectory() as tmp_dir:
           image_processor.save_pretrained(
               A				,     repo_id='''test-image-processor'''				,     push_to_hub=A				,     use_auth_token=self._token )

      _a											=						ViTImageProcessor.from_pretrained(f'''{USER}/test-image-processor''' )
      for k, v in image_processor.__dict__.items():
           self.assertEqual(A				,     getattr(A				,     A ) )



 def       a__						(self )					->  List[Any]:


      """simple docstring"""



      _a											=						ViTImageProcessor.from_pretrained(A )
      image_processor.push_to_hub('''valid_org/test-image-processor'''				,     use_auth_token=self._token )

      _a											=						ViTImageProcessor.from_pretrained('''valid_org/test-image-processor''' )
      for k, v in image_processor.__dict__.items():
           self.assertEqual(A				,     getattr(A				,     A ) )

      # Reset repo
      delete_repo(token=self._token				,     repo_id='''valid_org/test-image-processor''' )

      # Push to hub via save_pretrained
      with tempfile.TemporaryDirectory() as tmp_dir:
           image_processor.save_pretrained(
               A				,     repo_id='''valid_org/test-image-processor-org'''				,     push_to_hub=A				,     use_auth_token=self._token )

      _a											=						ViTImageProcessor.from_pretrained('''valid_org/test-image-processor-org''' )
      for k, v in image_processor.__dict__.items():
           self.assertEqual(A				,     getattr(A				,     A ) )



 def       a__						(self )					->  Union[str, Any]:


      """simple docstring"""



      CustomImageProcessor.register_for_auto_class()
      _a											=						CustomImageProcessor.from_pretrained(A )

      image_processor.push_to_hub('''test-dynamic-image-processor'''				,     use_auth_token=self._token )

      # This has added the proper auto_map field to the config
      self.assertDictEqual(
          image_processor.auto_map				,     {'''AutoImageProcessor''': '''custom_image_processing.CustomImageProcessor'''}				,     )

      _a											=						AutoImageProcessor.from_pretrained(
          f'''{USER}/test-dynamic-image-processor'''				,     trust_remote_code=A )
      # Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module
      self.assertEqual(new_image_processor.__class__.__name__				,     '''CustomImageProcessor''' )



