

'''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 							_snake_case       (				a_		,     unittest.TestCase  ):
   _A				:   List[str]			      =    VideoToVideoSDPipeline
   _A				:   Dict			      =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''}  ) - {'''image''', '''width''', '''height'''}
   _A				:   List[Any]			      =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''}  ) - {'''image'''}
   _A				:   Tuple			      =    PipelineTesterMixin.required_optional_params - {'''latents'''}
   _A				:   Any			      =    False

   # No `output_type`.
   _A				:   Tuple			      =    frozenset(
       [
           '''num_inference_steps''',
           '''generator''',
           '''latents''',
           '''return_dict''',
           '''callback''',
           '''callback_steps''',
       ]  )
   def 			__UpperCamelCase  (       self	:					str							):
     torch.manual_seed(0							)
     SCREAMING_SNAKE_CASE:int  					=    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   ,)
     SCREAMING_SNAKE_CASE:int  					=    DDIMScheduler(
         beta_start=0.00_085   ,beta_end=0.012   ,beta_schedule="scaled_linear"   ,clip_sample=_a   ,set_alpha_to_one=_a   ,)
     torch.manual_seed(0							)
     SCREAMING_SNAKE_CASE:List[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							)
     SCREAMING_SNAKE_CASE:Tuple  					=    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=1_000   ,hidden_act="gelu"   ,projection_dim=512   ,)
     SCREAMING_SNAKE_CASE:Tuple  					=    CLIPTextModel(_a							)
     SCREAMING_SNAKE_CASE:Union[str, Any]  					=    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							)

     SCREAMING_SNAKE_CASE:List[str]  					=    {
         "unet": unet,
         "scheduler": scheduler,
         "vae": vae,
         "text_encoder": text_encoder,
         "tokenizer": tokenizer,
     }
     return components
   def 			__UpperCamelCase  (       self	:					List[str]   ,SCREAMING_SNAKE_CASE__	:					List[str]   ,SCREAMING_SNAKE_CASE__	:					Optional[int]=0							):
     # 3 frames
     SCREAMING_SNAKE_CASE:Optional[int]  					=    floats_tensor((1, 3, 3, 32, 32)   ,rng=random.Random(_a							)							).to(_a							)

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

     SCREAMING_SNAKE_CASE:int  					=    self.get_dummy_inputs(_a							)
     SCREAMING_SNAKE_CASE:int  					=    "np"
     SCREAMING_SNAKE_CASE:int  					=    sd_pipe(**_a							).frames
     SCREAMING_SNAKE_CASE:List[Any]  					=    frames[0][-3:, -3:, -1]

     assert frames[0].shape == (32, 32, 3)
     SCREAMING_SNAKE_CASE:str  					=    np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]							)

     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 			__UpperCamelCase  (       self	:					str							):
     self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_a   ,expected_max_diff=5e-3							)
   @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."							)
   def 			__UpperCamelCase  (       self	:					Any							):
     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	:					Optional[Any]							):
     pass







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





@slow
@skip_mps
class 							_snake_case       (				unittest.TestCase  ):
   def 			__UpperCamelCase  (       self	:					Optional[int]							):
     SCREAMING_SNAKE_CASE:Optional[Any]  					=    VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"   ,torch_dtype=torch.floataa							)
     pipe.enable_model_cpu_offload()

     # 10 frames
     SCREAMING_SNAKE_CASE:Union[str, Any]  					=    torch.Generator(device="cpu"							).manual_seed(0							)
     SCREAMING_SNAKE_CASE:Dict  					=    torch.randn((1, 10, 3, 1_024, 576)   ,generator=_a							)
     SCREAMING_SNAKE_CASE:str  					=    video.to("cuda"							)

     SCREAMING_SNAKE_CASE:Dict  					=    "Spiderman is surfing"

     SCREAMING_SNAKE_CASE:Optional[int]  					=    pipe(_a   ,video=_a   ,generator=_a   ,num_inference_steps=3   ,output_type="pt"							).frames

     SCREAMING_SNAKE_CASE:Optional[int]  					=    np.array([-1.0_458_984, -1.1_279_297, -0.9_663_086, -0.91_503_906, -0.75_097_656]							)
     assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array							).sum() < 1e-2








"""simple docstring"""




from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_tokenizers_available,
    is_torch_available,
)


_A			:       Optional[int]   		=       {
    """configuration_llama""": ["""LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP""", """LlamaConfig"""],
}

try:
   if not is_sentencepiece_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _A			:       Union[str, Any]   		=       ["""LlamaTokenizer"""]

try:
   if not is_tokenizers_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _A			:       Dict   		=       ["""LlamaTokenizerFast"""]

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _A			:       int   		=       [
       """LlamaForCausalLM""",
       """LlamaModel""",
       """LlamaPreTrainedModel""",
       """LlamaForSequenceClassification""",
   ]


if TYPE_CHECKING:
   from .configuration_llama import LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP, LlamaConfig

   try:
      if not is_sentencepiece_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .tokenization_llama import LlamaTokenizer

   try:
      if not is_tokenizers_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .tokenization_llama_fast import LlamaTokenizerFast

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_llama import LlamaForCausalLM, LlamaForSequenceClassification, LlamaModel, LlamaPreTrainedModel


else:
   import sys

   _A			:       str   		=       _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


