




import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModel

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEImgaImgPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import floats_tensor, load_image, load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference







class 						lowercase							(   UpperCamelCase__,unittest.TestCase		):
     _a							 =						ShapEImgaImgPipeline
     _a							 =						["image"]
     _a							 =						["image"]
     _a							 =						[
         "num_images_per_prompt",
         "num_inference_steps",
         "generator",
         "latents",
         "guidance_scale",
         "frame_size",
         "output_type",
         "return_dict",
     ]
     _a							 =						False

     @property
     def 					a__	(  self				)						->							Optional[int]:
           return 32

     @property
     def 					a__	(  self				)						->							Tuple:
           return 32

     @property
     def 					a__	(  self				)						->							Optional[Any]:
           return self.time_input_dim * 4

     @property
     def 					a__	(  self				)						->							Optional[int]:
           return 8

     @property
     def 					a__	(  self				)						->							List[Any]:
           torch.manual_seed(0				)
           _A							:     Union[str, Any]			=						CLIPVisionConfig(
               hidden_size=self.text_embedder_hidden_size		,     image_size=64		,     projection_dim=self.text_embedder_hidden_size		,     intermediate_size=37		,     num_attention_heads=4		,     num_channels=3		,     num_hidden_layers=5		,     patch_size=1		,     )

           _A							:     List[Any]			=						CLIPVisionModel(_a				)
           return model

     @property
     def 					a__	(  self				)						->							Dict:
           _A							:     int			=						CLIPImageProcessor(
               crop_size=224		,     do_center_crop=_a		,     do_normalize=_a		,     do_resize=_a		,     image_mean=[0.48145466, 0.4578275, 0.40821073]		,     image_std=[0.26862954, 0.26130258, 0.27577711]		,     resample=3		,     size=224		,     )

           return image_processor

     @property
     def 					a__	(  self				)						->							Union[str, Any]:
           torch.manual_seed(0				)

           _A							:     str			=						{
               """num_attention_heads""": 2,
               """attention_head_dim""": 16,
               """embedding_dim""": self.time_input_dim,
               """num_embeddings""": 32,
               """embedding_proj_dim""": self.text_embedder_hidden_size,
               """time_embed_dim""": self.time_embed_dim,
               """num_layers""": 1,
               """clip_embed_dim""": self.time_input_dim * 2,
               """additional_embeddings""": 0,
               """time_embed_act_fn""": """gelu""",
               """norm_in_type""": """layer""",
               """embedding_proj_norm_type""": """layer""",
               """encoder_hid_proj_type""": None,
               """added_emb_type""": None,
           }

           _A							:     Optional[int]			=						PriorTransformer(**_a				)
           return model

     @property
     def 					a__	(  self				)						->							Union[str, Any]:
           torch.manual_seed(0				)

           _A							:     List[str]			=						{
               """param_shapes""": (
                   (self.renderer_dim, 93),
                   (self.renderer_dim, 8),
                   (self.renderer_dim, 8),
                   (self.renderer_dim, 8),
               ),
               """d_latent""": self.time_input_dim,
               """d_hidden""": self.renderer_dim,
               """n_output""": 12,
               """background""": (
                   0.1,
                   0.1,
                   0.1,
               ),
           }
           _A							:     int			=						ShapERenderer(**_a				)
           return model

     def 					a__	(  self				)						->							int:
           _A							:     Optional[Any]			=						self.dummy_prior
           _A							:     Tuple			=						self.dummy_image_encoder
           _A							:     str			=						self.dummy_image_processor
           _A							:     Any			=						self.dummy_renderer

           _A							:     str			=						HeunDiscreteScheduler(
               beta_schedule="""exp"""		,     num_train_timesteps=1024		,     prediction_type="""sample"""		,     use_karras_sigmas=_a		,     clip_sample=_a		,     clip_sample_range=1.0		,     )
           _A							:     Union[str, Any]			=						{
               """prior""": prior,
               """image_encoder""": image_encoder,
               """image_processor""": image_processor,
               """renderer""": renderer,
               """scheduler""": scheduler,
           }

           return components

     def 					a__	(  self		,     _a		,     _a=0				)						->							int:
           _A							:     str			=						floats_tensor((1, 3, 64, 64)		,     rng=random.Random(_a				)				).to(_a				)

           if str(_a				).startswith("""mps"""				):
                 _A							:     List[Any]			=						torch.manual_seed(_a				)
           else:
                 _A							:     Tuple			=						torch.Generator(device=_a				).manual_seed(_a				)
           _A							:     List[Any]			=						{
               """image""": input_image,
               """generator""": generator,
               """num_inference_steps""": 1,
               """frame_size""": 32,
               """output_type""": """np""",
           }
           return inputs

     def 					a__	(  self				)						->							Tuple:
           _A							:     List[Any]			=						"""cpu"""

           _A							:     List[str]			=						self.get_dummy_components()

           _A							:     str			=						self.pipeline_class(**_a				)
           _A							:     Optional[int]			=						pipe.to(_a				)

           pipe.set_progress_bar_config(disable=_a				)

           _A							:     Tuple			=						pipe(**self.get_dummy_inputs(_a				)				)
           _A							:     Tuple			=						output.images[0]
           _A							:     Dict			=						image[0, -3:, -3:, -1]

           assert image.shape == (20, 32, 32, 3)

           _A							:     Dict			=						np.array(
               [
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
               ]				)

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

     def 					a__	(  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 					a__	(  self				)						->							Tuple:
           _A							:     List[str]			=						torch_device == """cpu"""
           _A							:     Union[str, Any]			=						True
           self._test_inference_batch_single_identical(
               batch_size=2		,     test_max_difference=_a		,     relax_max_difference=_a		,     )



     def 					a__	(  self				)						->							Tuple:
           _A							:     int			=						self.get_dummy_components()
           _A							:     Optional[int]			=						self.pipeline_class(**_a				)
           _A							:     int			=						pipe.to(_a				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     int			=						1
           _A							:     List[str]			=						2

           _A							:     str			=						self.get_dummy_inputs(_a				)

           for key in inputs.keys():
                 if key in self.batch_params:
                       _A							:     str			=						batch_size * [inputs[key]]

           _A							:     Union[str, Any]			=						pipe(**_a		,     num_images_per_prompt=_a				)[0]

           assert images.shape[0] == batch_size * num_images_per_prompt







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

     def 					a__	(  self				)						->							int:
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()



     def 					a__	(  self				)						->							Dict:
           _A							:     List[str]			=						load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/shap_e/corgi.png"""				)
           _A							:     str			=						load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
               """/shap_e/test_shap_e_img2img_out.npy"""				)
           _A							:     List[Any]			=						ShapEImgaImgPipeline.from_pretrained("""openai/shap-e-img2img"""				)
           _A							:     Union[str, Any]			=						pipe.to(_a				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     Tuple			=						torch.Generator(device=_a				).manual_seed(0				)

           _A							:     Optional[Any]			=						pipe(
               _a		,     generator=_a		,     guidance_scale=3.0		,     num_inference_steps=64		,     frame_size=64		,     output_type="""np"""		,     ).images[0]

           assert images.shape == (20, 64, 64, 3)

           assert_mean_pixel_difference(_a		,     _a				)








import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY




@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class     SCREAMING_SNAKE_CASE__ (						unittest.TestCase      ):
 __SCREAMING_SNAKE_CASE							       =	MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING



 def 				UpperCamelCase			(							self,__lowerCamelCase,__lowerCamelCase,__lowerCamelCase ):
        A__											=						hf_hub_download(
            repo_id='''nateraw/video-demo''',filename='''archery.mp4''',repo_type='''dataset''' )
        A__											=						VideoClassificationPipeline(model=__lowerCamelCase,image_processor=__lowerCamelCase,top_k=2 )
        A__											=						[
            example_video_filepath,
            '''https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4''',
        ]
        return video_classifier, examples



 def 				UpperCamelCase			(							self,__lowerCamelCase,__lowerCamelCase ):
        for example in examples:
               A__											=						video_classifier(__lowerCamelCase )

               self.assertEqual(
                   __lowerCamelCase,[
                       {'''score''': ANY(__lowerCamelCase ), '''label''': ANY(__lowerCamelCase )},
                       {'''score''': ANY(__lowerCamelCase ), '''label''': ANY(__lowerCamelCase )},
                   ],)



 @require_torch
 def 				UpperCamelCase			(							self ):
        A__											=						'''hf-internal-testing/tiny-random-VideoMAEForVideoClassification'''
        A__											=						VideoMAEFeatureExtractor(
            size={'''shortest_edge''': 10},crop_size={'''height''': 10, '''width''': 10} )
        A__											=						pipeline(
            '''video-classification''',model=__lowerCamelCase,feature_extractor=__lowerCamelCase,frame_sampling_rate=4 )

        A__											=						hf_hub_download(repo_id='''nateraw/video-demo''',filename='''archery.mp4''',repo_type='''dataset''' )
        A__											=						video_classifier(__lowerCamelCase,top_k=2 )
        self.assertEqual(
            nested_simplify(__lowerCamelCase,decimals=4 ),[{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],)

        A__											=						video_classifier(
            [
                video_file_path,
                video_file_path,
            ],top_k=2,)
        self.assertEqual(
            nested_simplify(__lowerCamelCase,decimals=4 ),[
                [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
                [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
            ],)




 @require_tf
 def 				UpperCamelCase			(							self ):
        pass





