






"""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      _lowercase (     lowerCAmelCase		,  unittest.TestCase						):
      _a   :     Optional[Any]								=   VideoToVideoSDPipeline
      _a   :     Optional[int]								=   TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''}						) - {'''image''', '''width''', '''height'''}
      _a   :     int								=   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''}						) - {'''image'''}
      _a   :     Any								=   PipelineTesterMixin.required_optional_params - {'''latents'''}
      _a   :     List[str]								=   False

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


      def _UpperCamelCase						(    self				:	Optional[Any]						):





            """simple docstring"""







            torch.manual_seed(0						)
            __snake_case      :    int  			=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							,   )
            __snake_case      :    Any  			=DDIMScheduler(
                beta_start=0.0_0_0_8_5							,   beta_end=0.0_1_2							,   beta_schedule='''scaled_linear'''							,   clip_sample=a							,   set_alpha_to_one=a							,   )
            torch.manual_seed(0						)
            __snake_case      :    int  			=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						)
            __snake_case      :    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							,   )
            __snake_case      :    Union[str, Any]  			=CLIPTextModel(a						)
            __snake_case      :    List[Any]  			=CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

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


      def _UpperCamelCase						(    self				:	int							,   a				:	Dict							,   a				:	Dict=0						):





            """simple docstring"""







            __snake_case      :    int  			=floats_tensor((1, 3, 3, 3_2, 3_2)							,   rng=random.Random(a						)						).to(a						)

            if str(a						).startswith('''mps'''						):
                  __snake_case      :    Any  			=torch.manual_seed(a						)
            else:
                  __snake_case      :    List[Any]  			=torch.Generator(device=a						).manual_seed(a						)
            __snake_case      :    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 _UpperCamelCase						(    self				:	Optional[Any]						):





            """simple docstring"""







            __snake_case      :    List[Any]  			='''cpu'''  # ensure determinism for the device-dependent torch.Generator
            __snake_case      :    Optional[int]  			=self.get_dummy_components()
            __snake_case      :    Tuple  			=VideoToVideoSDPipeline(**a						)
            __snake_case      :    Optional[Any]  			=sd_pipe.to(a						)
            sd_pipe.set_progress_bar_config(disable=a						)

            __snake_case      :    Any  			=self.get_dummy_inputs(a						)
            __snake_case      :    Tuple  			='''np'''
            __snake_case      :    Optional[int]  			=sd_pipe(**a						).frames
            __snake_case      :    Any  			=frames[0][-3:, -3:, -1]

            assert frames[0].shape == (3_2, 3_2, 3)
            __snake_case      :    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 _UpperCamelCase						(    self				:	Any						):





            """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				:	Tuple						):





            """simple docstring"""







            pass


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





            """simple docstring"""







            pass


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





            """simple docstring"""







            pass







      def _UpperCamelCase						(    self				:	Tuple						):





            """simple docstring"""







            return super().test_progress_bar()



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


      def _UpperCamelCase						(    self				:	List[str]						):





            """simple docstring"""







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

            # 10 frames
            __snake_case      :    Optional[int]  			=torch.Generator(device='''cpu'''						).manual_seed(0						)
            __snake_case      :    int  			=torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)							,   generator=a						)
            __snake_case      :    Any  			=video.to('''cuda'''						)

            __snake_case      :    Optional[Any]  			='''Spiderman is surfing'''

            __snake_case      :    List[Any]  			=pipe(a							,   video=a							,   generator=a							,   num_inference_steps=3							,   output_type='''pt'''						).frames

            __snake_case      :    Any  			=np.array([-1.0_4_5_8_9_8_4, -1.1_2_7_9_2_9_7, -0.9_6_6_3_0_8_6, -0.9_1_5_0_3_9_0_6, -0.7_5_0_9_7_6_5_6]						)
            assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array						).sum() < 1e-2










"""simple docstring"""





from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCamelCase_	:				Optional[int]										= logging.get_logger(__name__)

UpperCamelCase_	:				Union[str, Any]										= {"""ctrl""": """https://huggingface.co/ctrl/resolve/main/config.json"""}


class      _lowercase (     lowerCAmelCase						):
      _a   :     str								=   '''ctrl'''
      _a   :     Optional[int]								=   ['''past_key_values''']
      _a   :     List[Any]								=   {
          '''max_position_embeddings''': '''n_positions''',
          '''hidden_size''': '''n_embd''',
          '''num_attention_heads''': '''n_head''',
          '''num_hidden_layers''': '''n_layer''',
      }


      def __init__(    self				:	str							,   a				:	Optional[int]=2_4_6_5_3_4							,   a				:	Optional[int]=2_5_6							,   a				:	Optional[int]=1_2_8_0							,   a				:	Any=8_1_9_2							,   a				:	int=4_8							,   a				:	Optional[Any]=1_6							,   a				:	str=0.1							,   a				:	str=0.1							,   a				:	Any=1e-6							,   a				:	Optional[int]=0.0_2							,   a				:	int=True							,   **a				:	str							,   ):





            """simple docstring"""







            __snake_case      :    Tuple  			=vocab_size
            __snake_case      :    Optional[Any]  			=n_positions
            __snake_case      :    List[Any]  			=n_embd
            __snake_case      :    Any  			=n_layer
            __snake_case      :    Any  			=n_head
            __snake_case      :    str  			=dff
            __snake_case      :    List[str]  			=resid_pdrop
            __snake_case      :    str  			=embd_pdrop
            __snake_case      :    Union[str, Any]  			=layer_norm_epsilon
            __snake_case      :    List[Any]  			=initializer_range

            __snake_case      :    Any  			=use_cache

            super().__init__(**a						)


