




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 						lowercase							(   UpperCamelCase__,unittest.TestCase		):
     _a							 =						VideoToVideoSDPipeline
     _a							 =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}		) - {"image", "width", "height"}
     _a							 =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}		) - {"image"}
     _a							 =						PipelineTesterMixin.required_optional_params - {"latents"}
     _a							 =						False

     # No `output_type`.
     _a							 =						frozenset(
         [
             "num_inference_steps",
             "generator",
             "latents",
             "return_dict",
             "callback",
             "callback_steps",
         ]		)

     def 					a__	(  self				)						->							Optional[Any]:
           torch.manual_seed(0				)
           _A							:     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		,     )
           _A							:     int			=						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				)
           _A							:     str			=						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				)
           _A							:     Tuple			=						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		,     )
           _A							:     Optional[Any]			=						CLIPTextModel(_a				)
           _A							:     Optional[int]			=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""				)

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

     def 					a__	(  self		,     _a		,     _a=0				)						->							List[Any]:
           # 3 frames
           _A							:     List[str]			=						floats_tensor((1, 3, 3, 32, 32)		,     rng=random.Random(_a				)				).to(_a				)

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

           _A							:     int			=						self.get_dummy_inputs(_a				)
           _A							:     Optional[int]			=						"""np"""
           _A							:     int			=						sd_pipe(**_a				).frames
           _A							:     Union[str, Any]			=						frames[0][-3:, -3:, -1]

           assert frames[0].shape == (32, 32, 3)
           _A							:     Dict			=						np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]				)

           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				)						->							Optional[int]:
           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				)						->							Tuple:
           pass

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

     @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""				)
     def 					a__	(  self				)						->							List[Any]:
           pass



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







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

     def 					a__	(  self				)						->							Any:
           _A							:     Union[str, Any]			=						VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""		,     torch_dtype=torch.floataa				)
           pipe.enable_model_cpu_offload()

           # 10 frames
           _A							:     List[Any]			=						torch.Generator(device="""cpu"""				).manual_seed(0				)
           _A							:     int			=						torch.randn((1, 10, 3, 1024, 576)		,     generator=_a				)
           _A							:     int			=						video.to("""cuda"""				)

           _A							:     Optional[Any]			=						"""Spiderman is surfing"""

           _A							:     List[Any]			=						pipe(_a		,     video=_a		,     generator=_a		,     num_inference_steps=3		,     output_type="""pt"""				).frames

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







import torch

from diffusers import DPMSolverSDEScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import require_torchsde

from .test_schedulers import SchedulerCommonTest







@require_torchsde
class 						lowercase							(   UpperCamelCase__		):
     _a							 =						(DPMSolverSDEScheduler,)
     _a							 =						1_0

     def 					a__	(  self		,     **_a				)						->							Optional[Any]:
           _A							:     str			=						{
               """num_train_timesteps""": 1100,
               """beta_start""": 0.0001,
               """beta_end""": 0.02,
               """beta_schedule""": """linear""",
               """noise_sampler_seed""": 0,
           }

           config.update(**_a				)
           return config

     def 					a__	(  self				)						->							Tuple:
           for timesteps in [10, 50, 100, 1000]:
                 self.check_over_configs(num_train_timesteps=_a				)

     def 					a__	(  self				)						->							Optional[int]:
           for beta_start, beta_end in zip([0.00001, 0.0001, 0.001]		,     [0.0002, 0.002, 0.02]				):
                 self.check_over_configs(beta_start=_a		,     beta_end=_a				)

     def 					a__	(  self				)						->							Any:
           for schedule in ["linear", "scaled_linear"]:
                 self.check_over_configs(beta_schedule=_a				)

     def 					a__	(  self				)						->							Optional[int]:
           for prediction_type in ["epsilon", "v_prediction"]:
                 self.check_over_configs(prediction_type=_a				)

     def 					a__	(  self				)						->							Optional[int]:
           _A							:     Any			=						self.scheduler_classes[0]
           _A							:     List[str]			=						self.get_scheduler_config()
           _A							:     Optional[Any]			=						scheduler_class(**_a				)

           scheduler.set_timesteps(self.num_inference_steps				)

           _A							:     Dict			=						self.dummy_model()
           _A							:     Any			=						self.dummy_sample_deter * scheduler.init_noise_sigma
           _A							:     Dict			=						sample.to(_a				)

           for i, t in enumerate(scheduler.timesteps				):
                 _A							:     Optional[int]			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     str			=						model(_a		,     _a				)

                 _A							:     List[Any]			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     Optional[int]			=						output.prev_sample

           _A							:     Dict			=						torch.sum(torch.abs(_a				)				)
           _A							:     Dict			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 167.47821044921875				) < 1e-2
                 assert abs(result_mean.item() - 0.2178705964565277				) < 1e-3
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 171.59352111816406				) < 1e-2
                 assert abs(result_mean.item() - 0.22342906892299652				) < 1e-3
           else:
                 assert abs(result_sum.item() - 162.52383422851562				) < 1e-2
                 assert abs(result_mean.item() - 0.211619570851326				) < 1e-3

     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Dict			=						self.scheduler_classes[0]
           _A							:     Optional[int]			=						self.get_scheduler_config(prediction_type="""v_prediction"""				)
           _A							:     Optional[Any]			=						scheduler_class(**_a				)

           scheduler.set_timesteps(self.num_inference_steps				)

           _A							:     Tuple			=						self.dummy_model()
           _A							:     int			=						self.dummy_sample_deter * scheduler.init_noise_sigma
           _A							:     Tuple			=						sample.to(_a				)

           for i, t in enumerate(scheduler.timesteps				):
                 _A							:     int			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     Tuple			=						model(_a		,     _a				)

                 _A							:     Dict			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     Optional[int]			=						output.prev_sample

           _A							:     Optional[Any]			=						torch.sum(torch.abs(_a				)				)
           _A							:     List[Any]			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 124.77149200439453				) < 1e-2
                 assert abs(result_mean.item() - 0.16226289014816284				) < 1e-3
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 128.1663360595703				) < 1e-2
                 assert abs(result_mean.item() - 0.16688326001167297				) < 1e-3
           else:
                 assert abs(result_sum.item() - 119.8487548828125				) < 1e-2
                 assert abs(result_mean.item() - 0.1560530662536621				) < 1e-3

     def 					a__	(  self				)						->							List[str]:
           _A							:     Union[str, Any]			=						self.scheduler_classes[0]
           _A							:     List[Any]			=						self.get_scheduler_config()
           _A							:     List[str]			=						scheduler_class(**_a				)

           scheduler.set_timesteps(self.num_inference_steps		,     device=_a				)

           _A							:     Union[str, Any]			=						self.dummy_model()
           _A							:     Optional[Any]			=						self.dummy_sample_deter.to(_a				) * scheduler.init_noise_sigma

           for t in scheduler.timesteps:
                 _A							:     int			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     List[Any]			=						model(_a		,     _a				)

                 _A							:     Dict			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     Dict			=						output.prev_sample

           _A							:     str			=						torch.sum(torch.abs(_a				)				)
           _A							:     str			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 167.46957397460938				) < 1e-2
                 assert abs(result_mean.item() - 0.21805934607982635				) < 1e-3
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 171.59353637695312				) < 1e-2
                 assert abs(result_mean.item() - 0.22342908382415771				) < 1e-3
           else:
                 assert abs(result_sum.item() - 162.52383422851562				) < 1e-2
                 assert abs(result_mean.item() - 0.211619570851326				) < 1e-3



     def 					a__	(  self				)						->							Union[str, Any]:
           _A							:     List[Any]			=						self.scheduler_classes[0]
           _A							:     Optional[Any]			=						self.get_scheduler_config()
           _A							:     int			=						scheduler_class(**_a		,     use_karras_sigmas=_a				)

           scheduler.set_timesteps(self.num_inference_steps		,     device=_a				)

           _A							:     Optional[Any]			=						self.dummy_model()
           _A							:     Dict			=						self.dummy_sample_deter.to(_a				) * scheduler.init_noise_sigma
           _A							:     str			=						sample.to(_a				)

           for t in scheduler.timesteps:
                 _A							:     Optional[int]			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     List[Any]			=						model(_a		,     _a				)

                 _A							:     Dict			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     List[str]			=						output.prev_sample

           _A							:     str			=						torch.sum(torch.abs(_a				)				)
           _A							:     List[str]			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 176.66974135742188				) < 1e-2
                 assert abs(result_mean.item() - 0.23003872730981811				) < 1e-2
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 177.63653564453125				) < 1e-2
                 assert abs(result_mean.item() - 0.23003872730981811				) < 1e-2
           else:
                 assert abs(result_sum.item() - 170.3135223388672				) < 1e-2
                 assert abs(result_mean.item() - 0.23003872730981811				) < 1e-2

