




import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()





class 			__lowerCamelCase	(   UpperCamelCase__     ,    unittest.TestCase   ):




    """simple docstring"""

    snake_case__ =				KandinskyImgaImgPipeline
    snake_case__ =				["prompt", "image_embeds", "negative_image_embeds", "image"]
    snake_case__ =				[
        "prompt",
        "negative_prompt",
        "image_embeds",
        "negative_image_embeds",
        "image",
    ]
    snake_case__ =				[
        "generator",
        "height",
        "width",
        "strength",
        "guidance_scale",
        "negative_prompt",
        "num_inference_steps",
        "return_dict",
        "guidance_scale",
        "num_images_per_prompt",
        "output_type",
        "return_dict",
    ]
    snake_case__ =				False


    @property
    def        a				(	self					:					Union[str, Any]       )				->	Dict:
         return 32


    @property
    def        a				(	self					:					Optional[Any]       )				->	Any:
         return 32


    @property
    def        a				(	self					:					Tuple       )				->	Any:
         return self.time_input_dim


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


    @property
    def        a				(	self					:					str       )				->	List[str]:
         return 100


    @property
    def        a				(	self					:					Optional[int]       )				->	List[Any]:
         lowerCAmelCase__          =			XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"       )
         return tokenizer


    @property
    def        a				(	self					:					Optional[Any]       )				->	int:
         torch.manual_seed(0       )
         lowerCAmelCase__          =			MCLIPConfig(
             numDims=self.cross_attention_dim       ,     transformerDimensions=self.text_embedder_hidden_size       ,     hidden_size=self.text_embedder_hidden_size       ,     intermediate_size=37       ,     num_attention_heads=4       ,     num_hidden_layers=5       ,     vocab_size=1_005       ,     )

         lowerCAmelCase__          =			MultilingualCLIP(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			text_encoder.eval()

         return text_encoder


    @property
    def        a				(	self					:					Dict       )				->	Dict:
         torch.manual_seed(0       )

         lowerCAmelCase__          =			{
             "in_channels": 4,
             # Out channels is double in channels because predicts mean and variance
             "out_channels": 8,
             "addition_embed_type": "text_image",
             "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
             "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
             "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
             "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
             "layers_per_block": 1,
             "encoder_hid_dim": self.text_embedder_hidden_size,
             "encoder_hid_dim_type": "text_image_proj",
             "cross_attention_dim": self.cross_attention_dim,
             "attention_head_dim": 4,
             "resnet_time_scale_shift": "scale_shift",
             "class_embed_type": None,
         }

         lowerCAmelCase__          =			UNetaDConditionModel(**SCREAMING_SNAKE_CASE__       )
         return model


    @property
    def        a				(	self					:					int       )				->	Dict:
         return {
             "block_out_channels": [32, 64],
             "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
             "in_channels": 3,
             "latent_channels": 4,
             "layers_per_block": 1,
             "norm_num_groups": 8,
             "norm_type": "spatial",
             "num_vq_embeddings": 12,
             "out_channels": 3,
             "up_block_types": [
                 "AttnUpDecoderBlock2D",
                 "UpDecoderBlock2D",
             ],
             "vq_embed_dim": 4,
         }


    @property
    def        a				(	self					:					Dict       )				->	str:
         torch.manual_seed(0       )
         lowerCAmelCase__          =			VQModel(**self.dummy_movq_kwargs       )
         return model


    def        a				(	self					:					Tuple       )				->	Any:
         lowerCAmelCase__          =			self.dummy_text_encoder
         lowerCAmelCase__          =			self.dummy_tokenizer
         lowerCAmelCase__          =			self.dummy_unet
         lowerCAmelCase__          =			self.dummy_movq

         lowerCAmelCase__          =			{
             "num_train_timesteps": 1_000,
             "beta_schedule": "linear",
             "beta_start": 0.00_085,
             "beta_end": 0.012,
             "clip_sample": False,
             "set_alpha_to_one": False,
             "steps_offset": 0,
             "prediction_type": "epsilon",
             "thresholding": False,
         }

         lowerCAmelCase__          =			DDIMScheduler(**SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			{
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
             "unet": unet,
             "scheduler": scheduler,
             "movq": movq,
         }

         return components


    def        a				(	self					:					Tuple       ,     SCREAMING_SNAKE_CASE__					:					Dict       ,     SCREAMING_SNAKE_CASE__					:					str=0       )				->	Dict:
         lowerCAmelCase__          =			floats_tensor((1, self.cross_attention_dim)       ,     rng=random.Random(SCREAMING_SNAKE_CASE__       )       ).to(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			floats_tensor((1, self.cross_attention_dim)       ,     rng=random.Random(seed + 1       )       ).to(SCREAMING_SNAKE_CASE__       )
         # create init_image
         lowerCAmelCase__          =			floats_tensor((1, 3, 64, 64)       ,     rng=random.Random(SCREAMING_SNAKE_CASE__       )       ).to(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			image.cpu().permute(0       ,     2       ,     3       ,     1       )[0]
         lowerCAmelCase__          =			Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE__       )       ).convert("RGB"       ).resize((256, 256)       )

         if str(SCREAMING_SNAKE_CASE__       ).startswith("mps"       ):
              lowerCAmelCase__          =			torch.manual_seed(SCREAMING_SNAKE_CASE__       )
         else:
              lowerCAmelCase__          =			torch.Generator(device=SCREAMING_SNAKE_CASE__       ).manual_seed(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			{
             "prompt": "horse",
             "image": init_image,
             "image_embeds": image_embeds,
             "negative_image_embeds": negative_image_embeds,
             "generator": generator,
             "height": 64,
             "width": 64,
             "num_inference_steps": 10,
             "guidance_scale": 7.0,
             "strength": 0.2,
             "output_type": "np",
         }
         return inputs


    def        a				(	self					:					Union[str, Any]       )				->	Optional[Any]:
         lowerCAmelCase__          =			"cpu"

         lowerCAmelCase__          =			self.get_dummy_components()

         lowerCAmelCase__          =			self.pipeline_class(**SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			pipe.to(SCREAMING_SNAKE_CASE__       )

         pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			pipe(**self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )       )
         lowerCAmelCase__          =			output.images

         lowerCAmelCase__          =			pipe(
             **self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )       ,     return_dict=SCREAMING_SNAKE_CASE__       ,     )[0]

         lowerCAmelCase__          =			image[0, -3:, -3:, -1]
         lowerCAmelCase__          =			image_from_tuple[0, -3:, -3:, -1]

         assert image.shape == (1, 64, 64, 3)

         lowerCAmelCase__          =			np.array(
             [0.61_474_943, 0.6_073_539, 0.43_308_544, 0.5_928_269, 0.47_493_595, 0.46_755_973, 0.4_613_838, 0.45_368_797, 0.50_119_233]       )
         assert (
             np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2
         ), f' expected_slice {expected_slice}, but got {image_slice.flatten()}'
         assert (
             np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1e-2
         ), f' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'






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




    """simple docstring"""

    def        a				(	self					:					Optional[Any]       )				->	List[str]:
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()


    def        a				(	self					:					str       )				->	List[Any]:
         lowerCAmelCase__          =			load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinsky/kandinsky_img2img_frog.npy"       )

         lowerCAmelCase__          =			load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"       )
         lowerCAmelCase__          =			"A red cartoon frog, 4k"

         lowerCAmelCase__          =			KandinskyPriorPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-1-prior"       ,     torch_dtype=torch.floataa       )
         pipe_prior.to(SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			KandinskyImgaImgPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-1"       ,     torch_dtype=torch.floataa       )
         lowerCAmelCase__          =			pipeline.to(SCREAMING_SNAKE_CASE__       )

         pipeline.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			torch.Generator(device="cpu"       ).manual_seed(0       )
         lowerCAmelCase__     ,		lowerCAmelCase__          =			pipe_prior(
             SCREAMING_SNAKE_CASE__       ,     generator=SCREAMING_SNAKE_CASE__       ,     num_inference_steps=5       ,     negative_prompt=""       ,     ).to_tuple()

         lowerCAmelCase__          =			pipeline(
             SCREAMING_SNAKE_CASE__       ,     image=SCREAMING_SNAKE_CASE__       ,     image_embeds=SCREAMING_SNAKE_CASE__       ,     negative_image_embeds=SCREAMING_SNAKE_CASE__       ,     generator=SCREAMING_SNAKE_CASE__       ,     num_inference_steps=100       ,     height=768       ,     width=768       ,     strength=0.2       ,     output_type="np"       ,     )

         lowerCAmelCase__          =			output.images[0]

         assert image.shape == (768, 768, 3)

         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )






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 			__lowerCamelCase	(   unittest.TestCase   ):




    """simple docstring"""

    snake_case__ =				MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING


    def        a				(	self					:					str       ,     SCREAMING_SNAKE_CASE__					:					Tuple       ,     SCREAMING_SNAKE_CASE__					:					List[str]       ,     SCREAMING_SNAKE_CASE__					:					Optional[int]       )				->	Any:
         lowerCAmelCase__          =			hf_hub_download(
             repo_id="nateraw/video-demo"       ,     filename="archery.mp4"       ,     repo_type="dataset"       )
         lowerCAmelCase__          =			VideoClassificationPipeline(model=SCREAMING_SNAKE_CASE__       ,     image_processor=SCREAMING_SNAKE_CASE__       ,     top_k=2       )
         lowerCAmelCase__          =			[
             example_video_filepath,
             "https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4",
         ]
         return video_classifier, examples


    def        a				(	self					:					Any       ,     SCREAMING_SNAKE_CASE__					:					Dict       ,     SCREAMING_SNAKE_CASE__					:					Union[str, Any]       )				->	Tuple:
         for example in examples:
              lowerCAmelCase__          =			video_classifier(SCREAMING_SNAKE_CASE__       )

              self.assertEqual(
                  SCREAMING_SNAKE_CASE__       ,     [
                      {"score": ANY(SCREAMING_SNAKE_CASE__       ), "label": ANY(SCREAMING_SNAKE_CASE__       )},
                      {"score": ANY(SCREAMING_SNAKE_CASE__       ), "label": ANY(SCREAMING_SNAKE_CASE__       )},
                  ]       ,     )


    @require_torch
    def        a				(	self					:					Dict       )				->	Optional[Any]:
         lowerCAmelCase__          =			"hf-internal-testing/tiny-random-VideoMAEForVideoClassification"
         lowerCAmelCase__          =			VideoMAEFeatureExtractor(
             size={"shortest_edge": 10}       ,     crop_size={"height": 10, "width": 10}       )
         lowerCAmelCase__          =			pipeline(
             "video-classification"       ,     model=SCREAMING_SNAKE_CASE__       ,     feature_extractor=SCREAMING_SNAKE_CASE__       ,     frame_sampling_rate=4       )

         lowerCAmelCase__          =			hf_hub_download(repo_id="nateraw/video-demo"       ,     filename="archery.mp4"       ,     repo_type="dataset"       )
         lowerCAmelCase__          =			video_classifier(SCREAMING_SNAKE_CASE__       ,     top_k=2       )
         self.assertEqual(
             nested_simplify(SCREAMING_SNAKE_CASE__       ,     decimals=4       )       ,     [{"score": 0.5_199, "label": "LABEL_0"}, {"score": 0.4_801, "label": "LABEL_1"}]       ,     )

         lowerCAmelCase__          =			video_classifier(
             [
                 video_file_path,
                 video_file_path,
             ]       ,     top_k=2       ,     )
         self.assertEqual(
             nested_simplify(SCREAMING_SNAKE_CASE__       ,     decimals=4       )       ,     [
                 [{"score": 0.5_199, "label": "LABEL_0"}, {"score": 0.4_801, "label": "LABEL_1"}],
                 [{"score": 0.5_199, "label": "LABEL_0"}, {"score": 0.4_801, "label": "LABEL_1"}],
             ]       ,     )


    @require_tf
    def        a				(	self					:					Optional[Any]       )				->	Optional[int]:
         pass
