



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

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







    def      lowercase_			( self					:    Optional[int]      ):
      torch.manual_seed(0      )
      a		:   List[Any]	  =	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=__snake_case		,     set_alpha_to_one=__snake_case		,     )
      torch.manual_seed(0      )
      a		:   Dict	  =	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      )
      a		:   str	  =	CLIPTextConfig(
          bos_token_id=0		,     eos_token_id=2		,     hidden_size=32		,     intermediate_size=37		,     layer_norm_eps=1e-0_5		,     num_attention_heads=4		,     num_hidden_layers=5		,     pad_token_id=1		,     vocab_size=10_00		,     hidden_act='gelu'		,     projection_dim=5_12		,     )
      a		:   Union[str, Any]	  =	CLIPTextModel(__snake_case      )
      a		:   List[Any]	  =	CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )

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







    def      lowercase_			( self					:    List[Any]		,     __snake_case					:    Union[str, Any]		,     __snake_case					:    Optional[int]=0      ):
      # 3 frames
      a		:   List[str]	  =	floats_tensor((1, 3, 3, 32, 32)		,     rng=random.Random(__snake_case      )      ).to(__snake_case      )

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

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

      assert frames[0].shape == (32, 32, 3)
      a		:   Tuple	  =	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      lowercase_			( self					:    Any      ):
      self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__snake_case		,     expected_max_diff=5e-3      )







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







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







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







    def      lowercase_			( self					:    List[str]      ):
      return super().test_progress_bar()

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







    def      lowercase_			( self					:    Optional[Any]      ):
      a		:   Optional[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		:   Optional[Any]	  =	torch.randn((1, 10, 3, 10_24, 5_76)		,     generator=__snake_case      )
      a		:   int	  =	video.to('cuda'      )

      a		:   List[Any]	  =	'Spiderman is surfing'

      a		:   int	  =	pipe(__snake_case		,     video=__snake_case		,     generator=__snake_case		,     num_inference_steps=3		,     output_type='pt'      ).frames

      a		:   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



class     A__   :







    """simple docstring"""

    def __init__(			self     ,   __snake_case       ):
         snake_case											=							n
         snake_case											=							[None] * self.n
         snake_case											=							0  # index of the first element
         snake_case											=							0
         snake_case											=							0
    def __len__(			self       ):
         return self.size
    def 						a_     (			self       ):
         return self.size == 0
    def 						a_     (			self       ):
         return False if self.is_empty() else self.array[self.front]
    def 						a_     (			self     ,   __snake_case       ):
         if self.size >= self.n:
              raise Exception('''QUEUE IS FULL'''       )

         snake_case											=							data
         snake_case											=							(self.rear + 1) % self.n
         self.size += 1
         return self




    def 						a_     (			self       ):
         if self.size == 0:
              raise Exception('''UNDERFLOW'''       )

         snake_case											=							self.array[self.front]
         snake_case											=							None
         snake_case											=							(self.front + 1) % self.n
         self.size -= 1
         return temp




