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

   # No `output_type`.
   _SCREAMING_SNAKE_CASE				:							Any     	=				frozenset(
       [
           "num_inference_steps",
           "generator",
           "latents",
           "return_dict",
           "callback",
           "callback_steps",
       ]   )
   def 				lowerCAmelCase      (self			: Any       ):
    torch.manual_seed(0       )
    __a			:  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 ,    )
    __a			:  str		    =  DDIMScheduler(
        beta_start=0.0_0085 ,    beta_end=0.012 ,    beta_schedule='''scaled_linear''' ,    clip_sample=snake_case_ ,    set_alpha_to_one=snake_case_ ,    )
    torch.manual_seed(0       )
    __a			:  Optional[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       )
    __a			:  List[Any]		    =  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 ,    )
    __a			:  List[str]		    =  CLIPTextModel(snake_case_       )
    __a			:  List[Any]		    =  CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )

    __a			:  Optional[int]		    =  {
        '''unet''': unet,
        '''scheduler''': scheduler,
        '''vae''': vae,
        '''text_encoder''': text_encoder,
        '''tokenizer''': tokenizer,
    }
    return components
   def 				lowerCAmelCase      (self			: Optional[Any] ,    snake_case_			: List[str] ,    snake_case_			: Union[str, Any]=0       ):
    # 3 frames
    __a			:  Union[str, Any]		    =  floats_tensor((1, 3, 3, 3_2, 3_2) ,    rng=random.Random(snake_case_       )       ).to(snake_case_       )

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

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

    assert frames[0].shape == (3_2, 3_2, 3)
    __a			:  Dict		    =  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 				lowerCAmelCase      (self			: Tuple       ):
    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 				lowerCAmelCase      (self			: Optional[int]       ):
    pass
   @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''       )
   def 				lowerCAmelCase      (self			: int       ):
    pass
   @unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''       )
   def 				lowerCAmelCase      (self			: List[str]       ):
    pass







   def 				lowerCAmelCase      (self			: Dict       ):
    return super().test_progress_bar()





@slow
@skip_mps
class 		UpperCamelCase__  (	unittest.TestCase   ):
   def 				lowerCAmelCase      (self			: List[Any]       ):
    __a			:  List[Any]		    =  VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL''' ,    torch_dtype=torch.floataa       )
    pipe.enable_model_cpu_offload()

    # 10 frames
    __a			:  Dict		    =  torch.Generator(device='''cpu'''       ).manual_seed(0       )
    __a			:  List[Any]		    =  torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6) ,    generator=snake_case_       )
    __a			:  str		    =  video.to('''cuda'''       )

    __a			:  Union[str, Any]		    =  '''Spiderman is surfing'''

    __a			:  Optional[int]		    =  pipe(snake_case_ ,    video=snake_case_ ,    generator=snake_case_ ,    num_inference_steps=3 ,    output_type='''pt'''       ).frames

    __a			:  Optional[int]		    =  np.array([-1.045_8984, -1.127_9297, -0.966_3086, -0.9150_3906, -0.7509_7656]       )
    assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array       ).sum() < 1E-2


def        __UpperCamelCase					(    lowerCAmelCase__							:    list  ):

 def merge(lowerCAmelCase__							:    list  ,       lowerCAmelCase__							:    list  ) -> list:

  def _merge():
   while left and right:
    yield (left if left[0] <= right[0] else right).pop(0  )
   yield from left
   yield from right

  return list(_merge()  )

 if len(lowerCAmelCase__  ) <= 1:
  return collection
 __a			:  str		    =  len(lowerCAmelCase__  ) // 2
 return merge(merge_sort(collection[:mid]  )  ,       merge_sort(collection[mid:]  )  )


if __name__ == "__main__":
      import doctest

      doctest.testmod()
      lowercase__          =input('Enter numbers separated by a comma:\n').strip()
      lowercase__          =[int(item) for item in user_input.split(',')]
      print(*merge_sort(unsorted), sep=',')
