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			):
  _SCREAMING_SNAKE_CASE     					=      TextToVideoSDPipeline
  _SCREAMING_SNAKE_CASE     					=      TEXT_TO_IMAGE_PARAMS
  _SCREAMING_SNAKE_CASE     					=      TEXT_TO_IMAGE_BATCH_PARAMS
  # No `output_type`.
  _SCREAMING_SNAKE_CASE     					=      frozenset(
      [
          """num_inference_steps""",
          """generator""",
          """latents""",
          """return_dict""",
          """callback""",
          """callback_steps""",
      ]			)

  def 		_UpperCAmelCase (		self			:   Any							):
   torch.manual_seed(0							)
   __a							=	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							=	DDIMScheduler(
       beta_start=0.0_00_85  ,				beta_end=0.0_12  ,				beta_schedule="scaled_linear"  ,				clip_sample=_UpperCAmelCase  ,				set_alpha_to_one=_UpperCAmelCase  ,				)
   torch.manual_seed(0							)
   __a							=	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							=	CLIPTextConfig(
       bos_token_id=0  ,				eos_token_id=2  ,				hidden_size=32  ,				intermediate_size=37  ,				layer_norm_eps=1E-05  ,				num_attention_heads=4  ,				num_hidden_layers=5  ,				pad_token_id=1  ,				vocab_size=10_00  ,				hidden_act="gelu"  ,				projection_dim=5_12  ,				)
   __a							=	CLIPTextModel(_UpperCAmelCase							)
   __a							=	CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							)

   __a							=	{
       "unet": unet,
       "scheduler": scheduler,
       "vae": vae,
       "text_encoder": text_encoder,
       "tokenizer": tokenizer,
   }
   return components

  def 		_UpperCAmelCase (		self			:   Optional[Any]  ,				__SCREAMING_SNAKE_CASE			:   Tuple  ,				__SCREAMING_SNAKE_CASE			:   Tuple=0							):
   if str(_UpperCAmelCase							).startswith("mps"							):
    __a							=	torch.manual_seed(_UpperCAmelCase							)
   else:
    __a							=	torch.Generator(device=_UpperCAmelCase							).manual_seed(_UpperCAmelCase							)
   __a							=	{
       "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			:   Optional[Any]							):
   __a							=	"cpu"  # ensure determinism for the device-dependent torch.Generator
   __a							=	self.get_dummy_components()
   __a							=	TextToVideoSDPipeline(**_UpperCAmelCase							)
   __a							=	sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __a							=	self.get_dummy_inputs(_UpperCAmelCase							)
   __a							=	"np"
   __a							=	sd_pipe(**_UpperCAmelCase							).frames
   __a							=	frames[0][-3:, -3:, -1]

   assert frames[0].shape == (64, 64, 3)
   __a							=	np.array([1_58.0, 1_60.0, 1_53.0, 1_25.0, 1_00.0, 1_21.0, 1_11.0, 93.0, 1_13.0]							)

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

  def 		_UpperCAmelCase (		self			:   List[Any]							):
   self._test_attention_slicing_forward_pass(test_mean_pixel_difference=_UpperCAmelCase  ,				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			:   Optional[Any]							):
   self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_UpperCAmelCase  ,				expected_max_diff=1E-2							)

  @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."							)
  def 		_UpperCAmelCase (		self			:   Optional[int]							):
   pass

  @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."							)
  def 		_UpperCAmelCase (		self			:   List[str]							):
   pass

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

  def 		_UpperCAmelCase (		self			:   Tuple							):
   return super().test_progress_bar()
@slow
@skip_mps
class  A_			(							unittest.TestCase			):

  def 		_UpperCAmelCase (		self			:   str							):
   __a							=	load_numpy(
       "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"							)

   __a							=	TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"							)
   __a							=	DPMSolverMultistepScheduler.from_config(pipe.scheduler.config							)
   __a							=	pipe.to("cuda"							)

   __a							=	"Spiderman is surfing"
   __a							=	torch.Generator(device="cpu"							).manual_seed(0							)

   __a							=	pipe(_UpperCAmelCase  ,				generator=_UpperCAmelCase  ,				num_inference_steps=25  ,				output_type="pt"							).frames
   __a							=	video_frames.cpu().numpy()

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

  def 		_UpperCAmelCase (		self			:   int							):
   __a							=	load_numpy(
       "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"							)

   __a							=	TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"							)
   __a							=	pipe.to("cuda"							)

   __a							=	"Spiderman is surfing"
   __a							=	torch.Generator(device="cpu"							).manual_seed(0							)

   __a							=	pipe(_UpperCAmelCase  ,				generator=_UpperCAmelCase  ,				num_inference_steps=2  ,				output_type="pt"							).frames
   __a							=	video_frames.cpu().numpy()

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










import re
from pathlib import Path
from unittest import TestCase

import pytest


@pytest.mark.integration
class     _a							(	UpperCAmelCase__     ):






  """simple docstring"""





  def  _UpperCAmelCase  (  self		,		_UpperCAmelCase			)      ->			Dict:

     with open(_UpperCAmelCase		,		encoding='utf-8'			) as input_file:
        UpperCamelCase_						=  re.compile(R'(?!.*\b(?:encoding|rb|w|wb|w+|wb+|ab|ab+)\b)(?<=\s)(open)\((.*)\)'			)
        UpperCamelCase_						=  input_file.read()
        UpperCamelCase_						=  regexp.search(_UpperCAmelCase			)

     return match




  def  _UpperCAmelCase  (  self		,		_UpperCAmelCase			)      ->			Dict:

     with open(_UpperCAmelCase		,		encoding='utf-8'			) as input_file:
        UpperCamelCase_						=  re.compile(R'#[^\r\n]*print\(|\"[^\r\n]*print\(|\"\"\".*?print\(.*?\"\"\"|(print\()'		,		re.DOTALL			)
        UpperCamelCase_						=  input_file.read()
        # use `re.finditer` to handle the case where the ignored groups would be matched first by `re.search`
        UpperCamelCase_						=  regexp.finditer(_UpperCAmelCase			)

     UpperCamelCase_						=  [match for match in matches if match is not None and match.group(1			) is not None]
     return matches[0] if matches else None




  def  _UpperCAmelCase  (  self			)      ->			List[str]:
     UpperCamelCase_						=  Path('./datasets'			)
     UpperCamelCase_						=  list(dataset_paths.absolute().glob('**/*.py'			)			)

     for dataset in dataset_files:
        if self._no_encoding_on_file_open(str(_UpperCAmelCase			)			):
           raise AssertionError(f"""open(...) must use utf-8 encoding in {dataset}"""			)




  def  _UpperCAmelCase  (  self			)      ->			Optional[int]:
     UpperCamelCase_						=  Path('./datasets'			)
     UpperCamelCase_						=  list(dataset_paths.absolute().glob('**/*.py'			)			)

     for dataset in dataset_files:
        if self._no_print_statements(str(_UpperCAmelCase			)			):
           raise AssertionError(f"""print statement found in {dataset}. Use datasets.logger/logging instead."""			)





