


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 		__A      (			_SCREAMING_SNAKE_CASE					,						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     A__						(		self   :Tuple ):
										'''simple docstring'''

										torch.manual_seed(0 )
										__magic_name__	:	Tuple								=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    , )
										__magic_name__	:	Any								=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 )
										__magic_name__	:	int								=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 )
										__magic_name__	:	List[Any]								=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    , )
										__magic_name__	:	Any								=CLIPTextModel(A_ )
										__magic_name__	:	Optional[int]								=CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

										__magic_name__	:	Optional[Any]								={
										    """unet""": unet,
										    """scheduler""": scheduler,
										    """vae""": vae,
										    """text_encoder""": text_encoder,
										    """tokenizer""": tokenizer,
										}
										return components

						def     A__						(		self   :Optional[Any]    , __snake_case   :Dict    , __snake_case   :Optional[int]=0 ):
										'''simple docstring'''

										__magic_name__	:	str								=floats_tensor((1, 3, 3, 32, 32)    , rng=random.Random(A_ ) ).to(A_ )

										if str(A_ ).startswith("""mps""" ):
														__magic_name__	:	Any								=torch.manual_seed(A_ )
										else:
														__magic_name__	:	Union[str, Any]								=torch.Generator(device=A_ ).manual_seed(A_ )
										__magic_name__	:	Dict								={
										    """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     A__						(		self   :Tuple ):
										'''simple docstring'''

										__magic_name__	:	Any								="""cpu"""  # ensure determinism for the device-dependent torch.Generator
										__magic_name__	:	str								=self.get_dummy_components()
										__magic_name__	:	Optional[Any]								=VideoToVideoSDPipeline(**A_ )
										__magic_name__	:	Any								=sd_pipe.to(A_ )
										sd_pipe.set_progress_bar_config(disable=A_ )

										__magic_name__	:	int								=self.get_dummy_inputs(A_ )
										__magic_name__	:	str								="""np"""
										__magic_name__	:	List[str]								=sd_pipe(**A_ ).frames
										__magic_name__	:	Any								=frames[0][-3:, -3:, -1]

										assert frames[0].shape == (32, 32, 3)
										__magic_name__	:	Dict								=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     A__						(		self   :Tuple ):
										'''simple docstring'''

										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     A__						(		self   :Any ):
										'''simple docstring'''

										pass

						@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline.""" )
						def     A__						(		self   :Union[str, Any] ):
										'''simple docstring'''

										pass

						@unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline.""" )
						def     A__						(		self   :Optional[int] ):
										'''simple docstring'''

										pass


						def     A__						(		self   :Union[str, Any] ):
										'''simple docstring'''

										return super().test_progress_bar()




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

						def     A__						(		self   :int ):
										'''simple docstring'''

										__magic_name__	:	Tuple								=VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""    , torch_dtype=torch.floataa )
										pipe.enable_model_cpu_offload()

										# 10 frames
										__magic_name__	:	Dict								=torch.Generator(device="""cpu""" ).manual_seed(0 )
										__magic_name__	:	List[str]								=torch.randn((1, 10, 3, 10_24, 5_76)    , generator=A_ )
										__magic_name__	:	Tuple								=video.to("""cuda""" )

										__magic_name__	:	str								="""Spiderman is surfing"""

										__magic_name__	:	int								=pipe(A_    , video=A_    , generator=A_    , num_inference_steps=3    , output_type="""pt""" ).frames

										__magic_name__	:	List[str]								=np.array([-1.0458984, -1.1279297, -0.9663086, -0.91503906, -0.75097656] )
										assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array ).sum() < 1E-2










from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
							from PIL import Image

							from ..image_utils import load_image

if is_tf_available():
							import tensorflow as tf

							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
							from ..tf_utils import stable_softmax

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING

UpperCAmelCase_		:				str									=     logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase__			)
class 		__A      (			UpperCamelCase__			):

						def __init__(		self   :Any    , *__snake_case   :List[Any]    , **__snake_case   :Union[str, Any] ):
										'''simple docstring'''

										super().__init__(*__snake_case    , **__snake_case )
										requires_backends(self    , """vision""" )
										self.check_model_type(
										    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
										    if self.framework == """tf"""
										    else MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING )

						def     A__						(		self   :int    , __snake_case   :Optional[Any]=None ):
										'''simple docstring'''

										__magic_name__	:	Any								={}
										if top_k is not None:
														__magic_name__	:	str								=top_k
										return {}, {}, postprocess_params

						def __call__(		self   :Tuple    , __snake_case   :Union[str, List[str], "Image.Image", List["Image.Image"]]    , **__snake_case   :Union[str, Any] ):
										'''simple docstring'''

										return super().__call__(__snake_case    , **__snake_case )

						def     A__						(		self   :str    , __snake_case   :Any ):
										'''simple docstring'''

										__magic_name__	:	Optional[Any]								=load_image(__snake_case )
										__magic_name__	:	Any								=self.image_processor(images=__snake_case    , return_tensors=self.framework )
										return model_inputs

						def     A__						(		self   :int    , __snake_case   :Optional[int] ):
										'''simple docstring'''

										__magic_name__	:	str								=self.model(**__snake_case )
										return model_outputs


						def     A__						(		self   :List[str]    , __snake_case   :int    , __snake_case   :Union[str, Any]=5 ):
										'''simple docstring'''

										if top_k > self.model.config.num_labels:
														__magic_name__	:	Dict								=self.model.config.num_labels

										if self.framework == "pt":
														__magic_name__	:	int								=model_outputs.logits.softmax(-1 )[0]
														__magic_name__				, __magic_name__	:	str								=probs.topk(__snake_case )
										elif self.framework == "tf":
														__magic_name__	:	Union[str, Any]								=stable_softmax(model_outputs.logits    , axis=-1 )[0]
														__magic_name__	:	List[str]								=tf.math.top_k(__snake_case    , k=__snake_case )
														__magic_name__				, __magic_name__	:	str								=topk.values.numpy(), topk.indices.numpy()
										else:
														raise ValueError(f"Unsupported framework: {self.framework}" )

										__magic_name__	:	str								=scores.tolist()
										__magic_name__	:	Union[str, Any]								=ids.tolist()
										return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__snake_case    , __snake_case )]




