






import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


@skip_mps
class   __SCREAMING_SNAKE_CASE(	_UpperCamelCase    ,  unittest.TestCase	):
							_UpperCAmelCase     =       VideoToVideoSDPipeline
							_UpperCAmelCase     =       TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}	) - {"image", "width", "height"}
							_UpperCAmelCase     =       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}	) - {"image"}
							_UpperCAmelCase     =       PipelineTesterMixin.required_optional_params - {"latents"}
							_UpperCAmelCase     =       False

							# No `output_type`.
							_UpperCAmelCase     =       frozenset(
							    [
							        "num_inference_steps",
							        "generator",
							        "latents",
							        "return_dict",
							        "callback",
							        "callback_steps",
							    ]	)
							def lowerCAmelCase_ (     self:      Optional[int]						)		->		Optional[int]:
											torch.manual_seed(0						)
											snake_case__									=     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       ,	)
											snake_case__									=     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						)
											snake_case__									=     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						)
											snake_case__									=     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=10_00       ,	hidden_act='gelu'       ,	projection_dim=5_12       ,	)
											snake_case__									=     CLIPTextModel(__a						)
											snake_case__									=     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'						)

											snake_case__									=     {
											    "unet": unet,
											    "scheduler": scheduler,
											    "vae": vae,
											    "text_encoder": text_encoder,
											    "tokenizer": tokenizer,
											}
											return components
							def lowerCAmelCase_ (     self:      Any       ,	UpperCamelCase:      Optional[int]       ,	UpperCamelCase:      List[Any]=0						)		->		int:
											# 3 frames
											snake_case__									=     floats_tensor((1, 3, 3, 32, 32)       ,	rng=random.Random(__a						)						).to(__a						)

											if str(__a						).startswith('mps'						):
															snake_case__									=     torch.manual_seed(__a						)
											else:
															snake_case__									=     torch.Generator(device=__a						).manual_seed(__a						)
											snake_case__									=     {
											    "prompt": "A painting of a squirrel eating a burger",
											    "video": video,
											    "generator": generator,
											    "num_inference_steps": 2,
											    "guidance_scale": 6.0,
											    "output_type": "pt",
											}
											return inputs
							def lowerCAmelCase_ (     self:      List[str]						)		->		List[Any]:
											snake_case__									=     "cpu"  # ensure determinism for the device-dependent torch.Generator
											snake_case__									=     self.get_dummy_components()
											snake_case__									=     VideoToVideoSDPipeline(**__a						)
											snake_case__									=     sd_pipe.to(__a						)
											sd_pipe.set_progress_bar_config(disable=__a						)

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

											assert frames[0].shape == (32, 32, 3)
											snake_case__									=     np.array([1_06, 1_17, 1_13, 1_74, 1_37, 1_12, 1_48, 1_51, 1_31]						)

											assert np.abs(image_slice.flatten() - expected_slice						).max() < 1e-2
							@unittest.skipIf(
							    torch_device != 'cuda' or not is_xformers_available()       ,	reason='XFormers attention is only available with CUDA and `xformers` installed'       ,	)
							def lowerCAmelCase_ (     self:      int						)		->		Optional[Any]:
											self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__a       ,	expected_max_diff=5e-3						)
							@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'						)
							def lowerCAmelCase_ (     self:      Tuple						)		->		Optional[int]:
											pass
							@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'						)
							def lowerCAmelCase_ (     self:      int						)		->		Any:
											pass
							@unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'						)
							def lowerCAmelCase_ (     self:      Any						)		->		Union[str, Any]:
											pass


							def lowerCAmelCase_ (     self:      str						)		->		List[Any]:
											return super().test_progress_bar()


@slow
@skip_mps
class   __SCREAMING_SNAKE_CASE(	unittest.TestCase	):
							def lowerCAmelCase_ (     self:      List[str]						)		->		List[str]:
											snake_case__									=     VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL'       ,	torch_dtype=torch.floataa						)
											pipe.enable_model_cpu_offload()

											# 10 frames
											snake_case__									=     torch.Generator(device='cpu'						).manual_seed(0						)
											snake_case__									=     torch.randn((1, 10, 3, 10_24, 5_76)       ,	generator=__a						)
											snake_case__									=     video.to('cuda'						)

											snake_case__									=     "Spiderman is surfing"

											snake_case__									=     pipe(__a       ,	video=__a       ,	generator=__a       ,	num_inference_steps=3       ,	output_type='pt'						).frames

											snake_case__									=     np.array([-1.0_458_984, -1.1_279_297, -0.9_663_086, -0.91_503_906, -0.75_097_656]						)
											assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array						).sum() < 1e-2












from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_tokenizers_available,
    is_torch_available,
    is_vision_available,
)


__UpperCamelCase					:		str	  =						{"""processing_layoutxlm""": ["""LayoutXLMProcessor"""]}

try:
							if not is_sentencepiece_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__UpperCamelCase					:		Dict	  =						["""LayoutXLMTokenizer"""]

try:
							if not is_tokenizers_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							__UpperCamelCase					:		str	  =						["""LayoutXLMTokenizerFast"""]

if TYPE_CHECKING:
							from .processing_layoutxlm import LayoutXLMProcessor

							try:
														if not is_sentencepiece_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .tokenization_layoutxlm import LayoutXLMTokenizer

							try:
														if not is_tokenizers_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .tokenization_layoutxlm_fast import LayoutXLMTokenizerFast

else:
							import sys

							__UpperCamelCase					:		Union[str, Any]	  =						_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)




