
"""simple docstring"""





import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


@skip_mps
class A_	(UpperCAmelCase__				,unittest.TestCase      ):




  '''simple docstring'''
  SCREAMING_SNAKE_CASE__			:						int       =		TextToVideoSDPipeline
  SCREAMING_SNAKE_CASE__			:						List[str]       =		TEXT_TO_IMAGE_PARAMS
  SCREAMING_SNAKE_CASE__			:						str       =		TEXT_TO_IMAGE_BATCH_PARAMS
  # No `output_type`.
  SCREAMING_SNAKE_CASE__			:						List[Any]       =		frozenset(
      [
          """num_inference_steps""",
          """generator""",
          """latents""",
          """return_dict""",
          """callback""",
          """callback_steps""",
      ]      )







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     torch.manual_seed(0  )
     UpperCAmelCase_					:					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				,   )
     UpperCAmelCase_					:					Optional[Any]			     =    DDIMScheduler(
         beta_start=0.0_00_85				,   beta_end=0.0_12				,   beta_schedule="scaled_linear"				,   clip_sample=lowercase_				,   set_alpha_to_one=lowercase_				,   )
     torch.manual_seed(0  )
     UpperCAmelCase_					:					Union[str, Any]			     =    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  )
     UpperCAmelCase_					:					Optional[Any]			     =    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=1000				,   hidden_act="gelu"				,   projection_dim=512				,   )
     UpperCAmelCase_					:					List[Any]			     =    CLIPTextModel(lowercase_  )
     UpperCAmelCase_					:					Tuple			     =    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"  )

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







  def        UpperCamelCase__					(		self				,   lowercase_				,   lowercase_=0  ):

     """simple docstring"""


     if str(lowercase_  ).startswith("mps"  ):
        UpperCAmelCase_					:					Optional[Any]			     =    torch.manual_seed(lowercase_  )
     else:
        UpperCAmelCase_					:					int			     =    torch.Generator(device=lowercase_  ).manual_seed(lowercase_  )
     UpperCAmelCase_					:					Union[str, Any]			     =    {
         '''prompt''': '''A painting of a squirrel eating a burger''',
         '''generator''': generator,
         '''num_inference_steps''': 2,
         '''guidance_scale''': 6.0,
         '''output_type''': '''pt''',
     }
     return inputs







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					Optional[Any]			     =    '''cpu'''  # ensure determinism for the device-dependent torch.Generator
     UpperCAmelCase_					:					Any			     =    self.get_dummy_components()
     UpperCAmelCase_					:					List[Any]			     =    TextToVideoSDPipeline(**lowercase_  )
     UpperCAmelCase_					:					Optional[Any]			     =    sd_pipe.to(lowercase_  )
     sd_pipe.set_progress_bar_config(disable=lowercase_  )

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

     assert frames[0].shape == (64, 64, 3)
     UpperCAmelCase_					:					List[str]			     =    np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]  )

     assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1E-2







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowercase_				,   expected_max_diff=3E-3  )







  @unittest.skipIf(
      torch_device != "cuda" or not is_xformers_available()				,   reason="XFormers attention is only available with CUDA and `xformers` installed"				,   )
  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowercase_				,   expected_max_diff=1E-2  )







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

     """simple docstring"""
     pass







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

     """simple docstring"""
     pass







  @unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."  )
  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     pass







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     return super().test_progress_bar()




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




  '''simple docstring'''







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					Dict			     =    load_numpy(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"  )

     UpperCAmelCase_					:					Optional[int]			     =    TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"  )
     UpperCAmelCase_					:					str			     =    DPMSolverMultistepScheduler.from_config(pipe.scheduler.config  )
     UpperCAmelCase_					:					Tuple			     =    pipe.to("cuda"  )

     UpperCAmelCase_					:					List[str]			     =    '''Spiderman is surfing'''
     UpperCAmelCase_					:					Any			     =    torch.Generator(device="cpu"  ).manual_seed(0  )

     UpperCAmelCase_					:					str			     =    pipe(lowercase_				,   generator=lowercase_				,   num_inference_steps=25				,   output_type="pt"  ).frames
     UpperCAmelCase_					:					Optional[Any]			     =    video_frames.cpu().numpy()

     assert np.abs(expected_video - video  ).mean() < 5E-2







  def        UpperCamelCase__					(		self  ):

     """simple docstring"""
     UpperCAmelCase_					:					Optional[int]			     =    load_numpy(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"  )

     UpperCAmelCase_					:					List[str]			     =    TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"  )
     UpperCAmelCase_					:					Tuple			     =    pipe.to("cuda"  )

     UpperCAmelCase_					:					Tuple			     =    '''Spiderman is surfing'''
     UpperCAmelCase_					:					int			     =    torch.Generator(device="cpu"  ).manual_seed(0  )

     UpperCAmelCase_					:					Optional[Any]			     =    pipe(lowercase_				,   generator=lowercase_				,   num_inference_steps=2				,   output_type="pt"  ).frames
     UpperCAmelCase_					:					List[Any]			     =    video_frames.cpu().numpy()

     assert np.abs(expected_video - video  ).mean() < 5E-2










"""simple docstring"""






from __future__ import annotations




class 							lowerCAmelCase__				:






      '''simple docstring'''

      def __init__(						self     :  Any						,	lowercase_     :  int = 0):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  List[Any]    						=		key
      def _SCREAMING_SNAKE_CASE (						self     :  Any						,	lowercase_     :  str						,	lowercase_     :  int):




          '''simple docstring'''


          assert isinstance(lowercase_						,	lowercase_) and isinstance(lowercase_						,	lowercase_)

          SCREAMING_SNAKE_CASE_      :  Dict    						=		key or self.__key or 1

          # make sure key is an appropriate size
          key %= 255

          return [chr(ord(lowercase_) ^ key) for ch in content]
      def _SCREAMING_SNAKE_CASE (						self     :  str						,	lowercase_     :  str						,	lowercase_     :  int):




          '''simple docstring'''


          assert isinstance(lowercase_						,	lowercase_) and isinstance(lowercase_						,	lowercase_)

          SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		key or self.__key or 1

          # make sure key is an appropriate size
          key %= 255

          return [chr(ord(lowercase_) ^ key) for ch in content]
      def _SCREAMING_SNAKE_CASE (						self     :  Tuple						,	lowercase_     :  str						,	lowercase_     :  int = 0):




          '''simple docstring'''


          assert isinstance(lowercase_						,	lowercase_) and isinstance(lowercase_						,	lowercase_)

          SCREAMING_SNAKE_CASE_      :  int    						=		key or self.__key or 1

          # make sure key can be any size
          while key > 255:
              key -= 255

          # This will be returned
          SCREAMING_SNAKE_CASE_      :  List[str]    						=		''''''

          for ch in content:
              ans += chr(ord(lowercase_) ^ key)

          return ans
      def _SCREAMING_SNAKE_CASE (						self     :  int						,	lowercase_     :  str						,	lowercase_     :  int = 0):




          '''simple docstring'''


          assert isinstance(lowercase_						,	lowercase_) and isinstance(lowercase_						,	lowercase_)

          SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		key or self.__key or 1

          # make sure key can be any size
          while key > 255:
              key -= 255

          # This will be returned
          SCREAMING_SNAKE_CASE_      :  List[Any]    						=		''''''

          for ch in content:
              ans += chr(ord(lowercase_) ^ key)

          return ans
      def _SCREAMING_SNAKE_CASE (						self     :  str						,	lowercase_     :  str						,	lowercase_     :  int = 0):




          '''simple docstring'''


          assert isinstance(lowercase_						,	lowercase_) and isinstance(lowercase_						,	lowercase_)

          try:
              with open(lowercase_) as fin, open('''encrypt.out'''						,	'''w+''') as fout:
                  # actual encrypt-process
                  for line in fin:
                      fout.write(self.encrypt_string(lowercase_						,	lowercase_))

          except OSError:
              return False

          return True



      def _SCREAMING_SNAKE_CASE (						self     :  Optional[int]						,	lowercase_     :  str						,	lowercase_     :  int):




          '''simple docstring'''


          assert isinstance(lowercase_						,	lowercase_) and isinstance(lowercase_						,	lowercase_)

          try:
              with open(lowercase_) as fin, open('''decrypt.out'''						,	'''w+''') as fout:
                  # actual encrypt-process
                  for line in fin:
                      fout.write(self.decrypt_string(lowercase_						,	lowercase_))

          except OSError:
              return False

          return True


# Tests
# crypt = XORCipher()
# key = 67

# # test encrypt
# print(crypt.encrypt("hallo welt",key))
# # test decrypt
# print(crypt.decrypt(crypt.encrypt("hallo welt",key), key))

# # test encrypt_string
# print(crypt.encrypt_string("hallo welt",key))

# # test decrypt_string
# print(crypt.decrypt_string(crypt.encrypt_string("hallo welt",key),key))

# if (crypt.encrypt_file("test.txt",key)):
#       print("encrypt successful")
# else:
#       print("encrypt unsuccessful")

# if (crypt.decrypt_file("encrypt.out",key)):
#       print("decrypt successful")
# else:
#       print("decrypt unsuccessful")


