



"""simple docstring"""




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      UpperCAmelCase_  (    _UpperCamelCase	,      unittest.TestCase						):
      __SCREAMING_SNAKE_CASE						:       Union[str, Any]								=			VideoToVideoSDPipeline
      __SCREAMING_SNAKE_CASE						:       Dict								=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'video'}						) - {'image', 'width', 'height'}
      __SCREAMING_SNAKE_CASE						:       Any								=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'video'}						) - {'image'}
      __SCREAMING_SNAKE_CASE						:       Tuple								=			PipelineTesterMixin.required_optional_params - {'latents'}
      __SCREAMING_SNAKE_CASE						:       Tuple								=			False

      # No `output_type`.
      __SCREAMING_SNAKE_CASE						:       List[Any]								=			frozenset(
          [
              'num_inference_steps',
              'generator',
              'latents',
              'return_dict',
              'callback',
              'callback_steps',
          ]						)



      def 						snake_case_ (					self					:     Union[str, Any]						):
           torch.manual_seed(0						)
           _UpperCAmelCase      :     Tuple        =   UNetaDConditionModel(
               block_out_channels=(3_2, 6_4, 6_4, 6_4)					,     layers_per_block=2					,     sample_size=3_2					,     in_channels=4					,     out_channels=4					,     down_block_types=("CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "DownBlock3D")					,     up_block_types=("UpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D")					,     cross_attention_dim=3_2					,     attention_head_dim=4					,     )
           _UpperCAmelCase      :     Tuple        =   DDIMScheduler(
               beta_start=0.00_085					,     beta_end=0.012					,     beta_schedule="scaled_linear"					,     clip_sample=lowerCamelCase_					,     set_alpha_to_one=lowerCamelCase_					,     )
           torch.manual_seed(0						)
           _UpperCAmelCase      :     Any        =   AutoencoderKL(
               block_out_channels=[3_2, 6_4]					,     in_channels=3					,     out_channels=3					,     down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]					,     up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]					,     latent_channels=4					,     sample_size=1_2_8					,     )
           torch.manual_seed(0						)
           _UpperCAmelCase      :     Optional[int]        =   CLIPTextConfig(
               bos_token_id=0					,     eos_token_id=2					,     hidden_size=3_2					,     intermediate_size=3_7					,     layer_norm_eps=1e-05					,     num_attention_heads=4					,     num_hidden_layers=5					,     pad_token_id=1					,     vocab_size=1_0_0_0					,     hidden_act="gelu"					,     projection_dim=5_1_2					,     )
           _UpperCAmelCase      :     List[str]        =   CLIPTextModel(lowerCamelCase_						)
           _UpperCAmelCase      :     Union[str, Any]        =   CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)

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



      def 						snake_case_ (					self					:     Dict					,     A					:     Optional[Any]					,     A					:     Union[str, Any]=0						):
           # 3 frames
           _UpperCAmelCase      :     List[str]        =   floats_tensor((1, 3, 3, 3_2, 3_2)					,     rng=random.Random(lowerCamelCase_						)						).to(lowerCamelCase_						)

           if str(lowerCamelCase_						).startswith("mps"						):
                _UpperCAmelCase      :     int        =   torch.manual_seed(lowerCamelCase_						)
           else:
                _UpperCAmelCase      :     Optional[int]        =   torch.Generator(device=lowerCamelCase_						).manual_seed(lowerCamelCase_						)
           _UpperCAmelCase      :     List[Any]        =   {
               "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 						snake_case_ (					self					:     Union[str, Any]						):
           _UpperCAmelCase      :     Tuple        =   "cpu"  # ensure determinism for the device-dependent torch.Generator
           _UpperCAmelCase      :     List[str]        =   self.get_dummy_components()
           _UpperCAmelCase      :     str        =   VideoToVideoSDPipeline(**lowerCamelCase_						)
           _UpperCAmelCase      :     str        =   sd_pipe.to(lowerCamelCase_						)
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase_						)

           _UpperCAmelCase      :     List[str]        =   self.get_dummy_inputs(lowerCamelCase_						)
           _UpperCAmelCase      :     Dict        =   "np"
           _UpperCAmelCase      :     List[Any]        =   sd_pipe(**lowerCamelCase_						).frames
           _UpperCAmelCase      :     Optional[Any]        =   frames[0][-3:, -3:, -1]

           assert frames[0].shape == (3_2, 3_2, 3)
           _UpperCAmelCase      :     Tuple        =   np.array([1_0_6, 1_1_7, 1_1_3, 1_7_4, 1_3_7, 1_1_2, 1_4_8, 1_5_1, 1_3_1]						)

           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 						snake_case_ (					self					:     Any						):
           self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCamelCase_					,     expected_max_diff=5e-3						)



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



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



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



      def 						snake_case_ (					self					:     Optional[Any]						):
           return super().test_progress_bar()







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



      def 						snake_case_ (					self					:     Tuple						):
           _UpperCAmelCase      :     Union[str, Any]        =   VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"					,     torch_dtype=torch.floataa						)
           pipe.enable_model_cpu_offload()

           # 10 frames
           _UpperCAmelCase      :     Dict        =   torch.Generator(device="cpu"						).manual_seed(0						)
           _UpperCAmelCase      :     Dict        =   torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)					,     generator=lowerCamelCase_						)
           _UpperCAmelCase      :     str        =   video.to("cuda"						)

           _UpperCAmelCase      :     Any        =   "Spiderman is surfing"

           _UpperCAmelCase      :     Tuple        =   pipe(lowerCamelCase_					,     video=lowerCamelCase_					,     generator=lowerCamelCase_					,     num_inference_steps=3					,     output_type="pt"						).frames

           _UpperCAmelCase      :     Dict        =   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










"""simple docstring"""




from typing import Dict, Iterable, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


_lowerCAmelCase    : Tuple							=							logging.get_logger(__name__)






class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       str								=			['pixel_values']



      def __init__(					self					:     List[Any]					,     A					:     bool = True					,     A					:     Dict[str, int] = None					,     A					:     PILImageResampling = PILImageResampling.BICUBIC					,     A					:     bool = True					,     A					:     Dict[str, int] = None					,     A					:     bool = True					,     A					:     Union[int, float] = 1 / 2_5_5					,     A					:     bool = True					,     A					:     Optional[Union[float, Iterable[float]]] = IMAGENET_DEFAULT_MEAN					,     A					:     Optional[Union[float, Iterable[float]]] = IMAGENET_DEFAULT_STD					,     **A					:     List[Any]					,     ):
           super().__init__(**A						)
           _UpperCAmelCase      :     int        =   size if size is not None else {"shortest_edge": 2_2_4}
           _UpperCAmelCase      :     Tuple        =   get_size_dict(A					,     default_to_square=A						)
           _UpperCAmelCase      :     int        =   crop_size if crop_size is not None else {"height": 2_2_4, "width": 2_2_4}
           _UpperCAmelCase      :     Union[str, Any]        =   get_size_dict(A					,     param_name="crop_size"						)

           _UpperCAmelCase      :     str        =   do_resize
           _UpperCAmelCase      :     Union[str, Any]        =   size
           _UpperCAmelCase      :     Tuple        =   resample
           _UpperCAmelCase      :     List[Any]        =   do_center_crop
           _UpperCAmelCase      :     List[Any]        =   crop_size
           _UpperCAmelCase      :     Tuple        =   do_rescale
           _UpperCAmelCase      :     List[Any]        =   rescale_factor
           _UpperCAmelCase      :     Dict        =   do_normalize
           _UpperCAmelCase      :     Optional[Any]        =   image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
           _UpperCAmelCase      :     Any        =   image_std if image_std is not None else IMAGENET_DEFAULT_STD



      def 						snake_case_ (					self					:     Optional[int]					,     A					:     np.ndarray					,     A					:     Dict[str, int]					,     A					:     PILImageResampling = PILImageResampling.BICUBIC					,     A					:     Optional[Union[str, ChannelDimension]] = None					,     **A					:     Union[str, Any]					,     ):
           _UpperCAmelCase      :     Tuple        =   get_size_dict(A					,     default_to_square=A						)
           # size_dict is a dict with either keys "height" and "width" or "shortest_edge"
           if "shortest_edge" in size:
                _UpperCAmelCase      :     Any        =   int((2_5_6 / 2_2_4) * size["shortest_edge"]						)
                _UpperCAmelCase      :     Union[str, Any]        =   get_resize_output_image_size(A					,     size=A					,     default_to_square=A						)
                _UpperCAmelCase      :     str        =   {"height": output_size[0], "width": output_size[1]}
           if "height" not in size_dict or "width" not in size_dict:
                raise ValueError(
                    f'Size dict must have keys \'height\' and \'width\' or \'shortest_edge\'. Got {size_dict.keys()}'						)
           return resize(
               A					,     size=(size_dict["height"], size_dict["width"])					,     resample=A					,     data_format=A					,     **A						)



      def 						snake_case_ (					self					:     int					,     A					:     np.ndarray					,     A					:     Dict[str, int]					,     A					:     Optional[Union[str, ChannelDimension]] = None					,     **A					:     Optional[Any]					,     ):
           _UpperCAmelCase      :     Dict        =   get_size_dict(A						)
           if "height" not in size or "width" not in size:
                raise ValueError(f'Size dict must have keys \'height\' and \'width\'. Got {size.keys()}'						)
           return center_crop(A					,     size=(size["height"], size["width"])					,     data_format=A					,     **A						)



      def 						snake_case_ (					self					:     List[str]					,     A					:     np.ndarray					,     A					:     Union[int, float]					,     A					:     Optional[Union[str, ChannelDimension]] = None					,     **A					:     Dict					,     ):
           return rescale(A					,     scale=A					,     data_format=A					,     **A						)



      def 						snake_case_ (					self					:     List[str]					,     A					:     np.ndarray					,     A					:     Union[float, List[float]]					,     A					:     Union[float, List[float]]					,     A					:     Optional[Union[str, ChannelDimension]] = None					,     **A					:     str					,     ):
           return normalize(A					,     mean=A					,     std=A					,     data_format=A					,     **A						)



      def 						snake_case_ (					self					:     Tuple					,     A					:     ImageInput					,     A					:     Optional[bool] = None					,     A					:     Optional[Dict[str, int]] = None					,     A					:     PILImageResampling = None					,     A					:     Optional[bool] = None					,     A					:     Optional[Dict[str, int]] = None					,     A					:     Optional[bool] = None					,     A					:     Optional[float] = None					,     A					:     Optional[bool] = None					,     A					:     Optional[Union[float, Iterable[float]]] = None					,     A					:     Optional[Union[float, Iterable[float]]] = None					,     A					:     Optional[TensorType] = None					,     A					:     ChannelDimension = ChannelDimension.FIRST					,     **A					:     Dict					,     ):
           _UpperCAmelCase      :     Dict        =   do_resize if do_resize is not None else self.do_resize
           _UpperCAmelCase      :     List[str]        =   resample if resample is not None else self.resample
           _UpperCAmelCase      :     int        =   do_center_crop if do_center_crop is not None else self.do_center_crop
           _UpperCAmelCase      :     str        =   do_rescale if do_rescale is not None else self.do_rescale
           _UpperCAmelCase      :     Tuple        =   rescale_factor if rescale_factor is not None else self.rescale_factor
           _UpperCAmelCase      :     Optional[int]        =   do_normalize if do_normalize is not None else self.do_normalize
           _UpperCAmelCase      :     List[str]        =   image_mean if image_mean is not None else self.image_mean
           _UpperCAmelCase      :     Tuple        =   image_std if image_std is not None else self.image_std

           _UpperCAmelCase      :     Any        =   size if size is not None else self.size
           _UpperCAmelCase      :     Optional[int]        =   get_size_dict(A					,     default_to_square=A						)
           _UpperCAmelCase      :     List[str]        =   crop_size if crop_size is not None else self.crop_size
           _UpperCAmelCase      :     List[str]        =   get_size_dict(A					,     param_name="crop_size"						)

           _UpperCAmelCase      :     List[Any]        =   make_list_of_images(A						)

           if not valid_images(A						):
                raise ValueError(
                    "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
                    "torch.Tensor, tf.Tensor or jax.ndarray."						)

           if do_resize and size is None:
                raise ValueError("Size must be specified if do_resize is True."						)

           if do_center_crop and crop_size is None:
                raise ValueError("Crop size must be specified if do_center_crop is True."						)

           if do_rescale and rescale_factor is None:
                raise ValueError("Rescale factor must be specified if do_rescale is True."						)

           if do_normalize and (image_mean is None or image_std is None):
                raise ValueError("Image mean and std must be specified if do_normalize is True."						)

           # All transformations expect numpy arrays.
           _UpperCAmelCase      :     Any        =   [to_numpy_array(A						) for image in images]

           if do_resize:
                _UpperCAmelCase      :     str        =   [self.resize(A					,     A					,     A						) for image in images]

           if do_center_crop:
                _UpperCAmelCase      :     Dict        =   [self.center_crop(A					,     A						) for image in images]

           if do_rescale:
                _UpperCAmelCase      :     Dict        =   [self.rescale(A					,     A						) for image in images]

           if do_normalize:
                _UpperCAmelCase      :     Optional[int]        =   [self.normalize(A					,     A					,     A						) for image in images]

           _UpperCAmelCase      :     Union[str, Any]        =   [to_channel_dimension_format(A					,     A						) for image in images]

           _UpperCAmelCase      :     List[Any]        =   {"pixel_values": images}
           return BatchFeature(data=A					,     tensor_type=A						)
