


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

      # No `output_type`.
      snake_case__       :  List[Any]       =		frozenset(
          [
              '''num_inference_steps''',
              '''generator''',
              '''latents''',
              '''return_dict''',
              '''callback''',
              '''callback_steps''',
          ]    )
      def  SCREAMING_SNAKE_CASE							( self       : List[Any]  )	->    List[str]:
             torch.manual_seed(0  )
             a_      :	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  ,							)
             a_      :	List[str]							=						DDIMScheduler(
                 beta_start=0.00085  ,							beta_end=0.012  ,							beta_schedule='scaled_linear'  ,							clip_sample=SCREAMING_SNAKE_CASE__  ,							set_alpha_to_one=SCREAMING_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_      :	Optional[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[Any]							=						CLIPTextModel(SCREAMING_SNAKE_CASE__  )
             a_      :	Optional[Any]							=						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  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							SCREAMING_SNAKE_CASE__       : List[Any]=0  )	->    Dict:
             # 3 frames
             a_      :	Dict							=						floats_tensor((1, 3, 3, 3_2, 3_2)  ,							rng=random.Random(SCREAMING_SNAKE_CASE__  )  ).to(SCREAMING_SNAKE_CASE__  )

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

             a_      :	List[Any]							=						self.get_dummy_inputs(SCREAMING_SNAKE_CASE__  )
             a_      :	Dict							=						'np'
             a_      :	Any							=						sd_pipe(**SCREAMING_SNAKE_CASE__  ).frames
             a_      :	List[str]							=						frames[0][-3:, -3:, -1]

             assert frames[0].shape == (3_2, 3_2, 3)
             a_      :	List[str]							=						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  SCREAMING_SNAKE_CASE							( self       : str  )	->    Any:
             self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__  ,							expected_max_diff=5E-3  )
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[Any]  )	->    Optional[Any]:
             pass
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'  )
      def  SCREAMING_SNAKE_CASE							( self       : List[str]  )	->    Any:
             pass
      @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[Any]  )	->    Optional[int]:
             pass

      def  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  )	->    List[str]:
             return super().test_progress_bar()
@slow
@skip_mps
class 				SCREAMING_SNAKE_CASE__						(       unittest.TestCase    ):
      def  SCREAMING_SNAKE_CASE							( self       : Optional[Any]  )	->    Optional[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_      :	List[str]							=						torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)  ,							generator=SCREAMING_SNAKE_CASE__  )
             a_      :	Optional[int]							=						video.to('cuda'  )

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

             a_      :	str							=						pipe(SCREAMING_SNAKE_CASE__  ,							video=SCREAMING_SNAKE_CASE__  ,							generator=SCREAMING_SNAKE_CASE__  ,							num_inference_steps=3  ,							output_type='pt'  ).frames

             a_      :	List[Any]							=						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






from argparse import ArgumentParser

from .env import EnvironmentCommand







def  SCREAMING_SNAKE_CASE_		(    )			->	str:

       """simple docstring"""

       a_      :	str							=						ArgumentParser('Diffusers CLI tool'						,					usage='diffusers-cli <command> [<args>]'       )
       a_      :	Any							=						parser.add_subparsers(help='diffusers-cli command helpers'       )

       # Register commands
       EnvironmentCommand.register_subcommand(__A       )

       # Let's go
       a_      :	int							=						parser.parse_args()

       if not hasattr(__A						,					'func'       ):
              parser.print_help()
              exit(1       )

       # Run
       a_      :	Union[str, Any]							=						args.func(__A       )
       service.run()


if __name__ == "__main__":
   main()


