

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 (     _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 UpperCamelCase			(			self :      str    ):

     """simple docstring"""






     torch.manual_seed(0    )
     SCREAMING_SNAKE_CASE								=			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							,    )
     SCREAMING_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    )
     SCREAMING_SNAKE_CASE								=			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    )
     SCREAMING_SNAKE_CASE								=			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							,    )
     SCREAMING_SNAKE_CASE								=			CLIPTextModel(A_    )
     SCREAMING_SNAKE_CASE								=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

     SCREAMING_SNAKE_CASE								=			{
         'unet': unet,
         'scheduler': scheduler,
         'vae': vae,
         'text_encoder': text_encoder,
         'tokenizer': tokenizer,
     }
     return components

    def UpperCamelCase			(			self :      Optional[Any]							,    snake_case__ :      Union[str, Any]							,    snake_case__ :      Any=0    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			floats_tensor((1, 3, 3, 3_2, 3_2)							,    rng=random.Random(A_    )    ).to(A_    )

     if str(A_    ).startswith('mps'    ):
      SCREAMING_SNAKE_CASE								=			torch.manual_seed(A_    )
     else:
      SCREAMING_SNAKE_CASE								=			torch.Generator(device=A_    ).manual_seed(A_    )
     SCREAMING_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 UpperCamelCase			(			self :      str    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'cpu'  # ensure determinism for the device-dependent torch.Generator
     SCREAMING_SNAKE_CASE								=			self.get_dummy_components()
     SCREAMING_SNAKE_CASE								=			VideoToVideoSDPipeline(**A_    )
     SCREAMING_SNAKE_CASE								=			sd_pipe.to(A_    )
     sd_pipe.set_progress_bar_config(disable=A_    )

     SCREAMING_SNAKE_CASE								=			self.get_dummy_inputs(A_    )
     SCREAMING_SNAKE_CASE								=			'np'
     SCREAMING_SNAKE_CASE								=			sd_pipe(**A_    ).frames
     SCREAMING_SNAKE_CASE								=			frames[0][-3:, -3:, -1]

     assert frames[0].shape == (3_2, 3_2, 3)
     SCREAMING_SNAKE_CASE								=			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 UpperCamelCase			(			self :      int    ):

     """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 UpperCamelCase			(			self :      List[str]    ):

     """simple docstring"""






     pass

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

     """simple docstring"""






     pass

    @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'    )
    def UpperCamelCase			(			self :      List[str]    ):

     """simple docstring"""






     pass




    def UpperCamelCase			(			self :      Union[str, Any]    ):

     """simple docstring"""






     return super().test_progress_bar()




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

    def UpperCamelCase			(			self :      Any    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL'							,    torch_dtype=torch.floataa    )
     pipe.enable_model_cpu_offload()

     # 10 frames
     SCREAMING_SNAKE_CASE								=			torch.Generator(device='cpu'    ).manual_seed(0    )
     SCREAMING_SNAKE_CASE								=			torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)							,    generator=A_    )
     SCREAMING_SNAKE_CASE								=			video.to('cuda'    )

     SCREAMING_SNAKE_CASE								=			'Spiderman is surfing'

     SCREAMING_SNAKE_CASE								=			pipe(A_							,    video=A_							,    generator=A_							,    num_inference_steps=3							,    output_type='pt'    ).frames

     SCREAMING_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


def        __lowerCAmelCase    (       _UpperCamelCase		:  int	)		->							int:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			abs(_UpperCamelCase	)
 SCREAMING_SNAKE_CASE								=			0
 while n > 0:
  res += n % 10
  n //= 10
 return res


def        __lowerCAmelCase    (       _UpperCamelCase		:  int	)		->							int:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			abs(_UpperCamelCase	)
 return n if n < 10 else n % 10 + sum_of_digits(n // 10	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  int	)		->							int:
 '''simple docstring'''


 return sum(int(_UpperCamelCase	) for c in str(abs(_UpperCamelCase	)	)	)


def        __lowerCAmelCase    (       )		->							None:
 '''simple docstring'''


 from collections.abc import Callable
 from timeit import timeit

 def benchmark_a_function(_UpperCamelCase		:  Callable		,							_UpperCamelCase		:  int	) -> None:
  SCREAMING_SNAKE_CASE								=			f"""{func.__name__}({value})"""
  SCREAMING_SNAKE_CASE								=			timeit(f"""__main__.{call}"""		,							setup='import __main__'	)
  print(f"""{call:56} = {func(_UpperCamelCase	)} -- {timing:.4f} seconds"""	)

 for value in (26_21_44, 11_25_89_99_06_84_26_24, 1_26_76_50_60_02_28_22_94_01_49_67_03_20_53_76):
  for func in (sum_of_digits, sum_of_digits_recursion, sum_of_digits_compact):
   benchmark_a_function(_UpperCamelCase		,							_UpperCamelCase	)
  print()


if __name__ == "__main__":
      import doctest

      doctest.testmod()
      benchmark()

