



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 					_lowercase      (a_					,      unittest.TestCase					):
				'''simple docstring'''







				lowercase__        =  TextToVideoSDPipeline
				lowercase__        =  TEXT_TO_IMAGE_PARAMS
				lowercase__        =  TEXT_TO_IMAGE_BATCH_PARAMS
				# No `output_type`.
				lowercase__        =  frozenset(
				    [
				        """num_inference_steps""",
				        """generator""",
				        """latents""",
				        """return_dict""",
				        """callback""",
				        """callback_steps""",
				    ]					)

				def 					_lowerCamelCase    (       self			):

									'''simple docstring'''


									torch.manual_seed(0			)
									UpperCamelCase_						     =      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   ,       )
									UpperCamelCase_						     =      DDIMScheduler(
									    beta_start=0.00_085   ,       beta_end=0.012   ,       beta_schedule="scaled_linear"   ,       clip_sample=__a   ,       set_alpha_to_one=__a   ,       )
									torch.manual_seed(0			)
									UpperCamelCase_						     =      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			)
									UpperCamelCase_						     =      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=1000   ,       hidden_act="gelu"   ,       projection_dim=512   ,       )
									UpperCamelCase_						     =      CLIPTextModel(__a			)
									UpperCamelCase_						     =      CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"			)

									UpperCamelCase_						     =      {
									    "unet": unet,
									    "scheduler": scheduler,
									    "vae": vae,
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									}
									return components

				def 					_lowerCamelCase    (       self   ,       snake_case__   ,       snake_case__=0			):

									'''simple docstring'''


									if str(__a			).startswith("mps"			):
														UpperCamelCase_						     =      torch.manual_seed(__a			)
									else:
														UpperCamelCase_						     =      torch.Generator(device=__a			).manual_seed(__a			)
									UpperCamelCase_						     =      {
									    "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 					_lowerCamelCase    (       self			):

									'''simple docstring'''


									UpperCamelCase_						     =      "cpu"  # ensure determinism for the device-dependent torch.Generator
									UpperCamelCase_						     =      self.get_dummy_components()
									UpperCamelCase_						     =      TextToVideoSDPipeline(**__a			)
									UpperCamelCase_						     =      sd_pipe.to(__a			)
									sd_pipe.set_progress_bar_config(disable=__a			)

									UpperCamelCase_						     =      self.get_dummy_inputs(__a			)
									UpperCamelCase_						     =      "np"
									UpperCamelCase_						     =      sd_pipe(**__a			).frames
									UpperCamelCase_						     =      frames[0][-3:, -3:, -1]

									assert frames[0].shape == (64, 64, 3)
									UpperCamelCase_						     =      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 					_lowerCamelCase    (       self			):

									'''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 					_lowerCamelCase    (       self			):

									'''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 					_lowerCamelCase    (       self			):

									'''simple docstring'''


									pass

				@unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."			)
				def 					_lowerCamelCase    (       self			):

									'''simple docstring'''


									pass

				@unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."			)
				def 					_lowerCamelCase    (       self			):

									'''simple docstring'''


									pass

				def 					_lowerCamelCase    (       self			):

									'''simple docstring'''


									return super().test_progress_bar()




@slow
@skip_mps
class 					_lowercase      (unittest.TestCase					):
				'''simple docstring'''







				def 					_lowerCamelCase    (       self			):

									'''simple docstring'''


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

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

									UpperCamelCase_						     =      "Spiderman is surfing"
									UpperCamelCase_						     =      torch.Generator(device="cpu"			).manual_seed(0			)

									UpperCamelCase_						     =      pipe(__a   ,       generator=__a   ,       num_inference_steps=25   ,       output_type="pt"			).frames
									UpperCamelCase_						     =      video_frames.cpu().numpy()

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

				def 					_lowerCamelCase    (       self			):

									'''simple docstring'''


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

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

									UpperCamelCase_						     =      "Spiderman is surfing"
									UpperCamelCase_						     =      torch.Generator(device="cpu"			).manual_seed(0			)

									UpperCamelCase_						     =      pipe(__a   ,       generator=__a   ,       num_inference_steps=2   ,       output_type="pt"			).frames
									UpperCamelCase_						     =      video_frames.cpu().numpy()

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





import pytest

from datasets.utils.sharding import _distribute_shards, _number_of_shards_in_gen_kwargs, _split_gen_kwargs






@pytest.mark.parametrize(
    "kwargs, expected"				, [
        ({"num_shards": 0, "max_num_jobs": 1}, []),
        ({"num_shards": 10, "max_num_jobs": 1}, [range(10  )]),
        ({"num_shards": 10, "max_num_jobs": 10}, [range(_lowerCamelCase				, i + 1  ) for i in range(10  )]),
        ({"num_shards": 1, "max_num_jobs": 10}, [range(1  )]),
        ({"num_shards": 10, "max_num_jobs": 3}, [range(0				, 4  ), range(4				, 7  ), range(7				, 10  )]),
        ({"num_shards": 3, "max_num_jobs": 10}, [range(0				, 1  ), range(1				, 2  ), range(2				, 3  )]),
    ]				, )
def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							List[Any]    =    _distribute_shards(**_lowerCamelCase  )
							assert out == expected






@pytest.mark.parametrize(
    "gen_kwargs, max_num_jobs, expected"				, [
        ({"foo": 0}, 10, [{"foo": 0}]),
        ({"shards": [0, 1, 2, 3]}, 1, [{"shards": [0, 1, 2, 3]}]),
        ({"shards": [0, 1, 2, 3]}, 4, [{"shards": [0]}, {"shards": [1]}, {"shards": [2]}, {"shards": [3]}]),
        ({"shards": [0, 1]}, 4, [{"shards": [0]}, {"shards": [1]}]),
        ({"shards": [0, 1, 2, 3]}, 2, [{"shards": [0, 1]}, {"shards": [2, 3]}]),
    ]				, )
def 				A				(					_lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							Optional[int]    =    _split_gen_kwargs(_lowerCamelCase				, _lowerCamelCase  )
							assert out == expected






@pytest.mark.parametrize(
    "gen_kwargs, expected"				, [
        ({"foo": 0}, 1),
        ({"shards": [0]}, 1),
        ({"shards": [0, 1, 2, 3]}, 4),
        ({"shards": [0, 1, 2, 3], "foo": 0}, 4),
        ({"shards": [0, 1, 2, 3], "other": (0, 1)}, 4),
        ({"shards": [0, 1, 2, 3], "shards2": [0, 1]}, RuntimeError),
    ]				, )
def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							if expected is RuntimeError:
														with pytest.raises(_lowerCamelCase  ):
																					_number_of_shards_in_gen_kwargs(_lowerCamelCase  )
							else:
														_lowerCAmelCase     :							Optional[int]    =    _number_of_shards_in_gen_kwargs(_lowerCamelCase  )
														assert out == expected
