"""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					(		__lowerCAmelCase						,				unittest.TestCase ):
    _snake_case  :	List[str]  				=				TextToVideoSDPipeline
    _snake_case  :	List[str]  				=				TEXT_TO_IMAGE_PARAMS
    _snake_case  :	Optional[int]  				=				TEXT_TO_IMAGE_BATCH_PARAMS
    # No `output_type`.
    _snake_case  :	int  				=				frozenset(
        [
            'num_inference_steps',
            'generator',
            'latents',
            'return_dict',
            'callback',
            'callback_steps',
        ] )


    def   lowerCAmelCase_	(       self						:	int       ):
      torch.manual_seed(0       )
      _UpperCAmelCase    			=   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    			=   DDIMScheduler(
          beta_start=0.00_085 ,					beta_end=0.012 ,					beta_schedule="""scaled_linear""" ,					clip_sample=_SCREAMING_SNAKE_CASE ,					set_alpha_to_one=_SCREAMING_SNAKE_CASE ,					)
      torch.manual_seed(0       )
      _UpperCAmelCase    			=   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    			=   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    			=   CLIPTextModel(_SCREAMING_SNAKE_CASE       )
      _UpperCAmelCase    			=   CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

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


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

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

      assert frames[0].shape == (64, 64, 3)
      _UpperCAmelCase    			=   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   lowerCAmelCase_	(       self						:	List[Any]       ):
      self._test_attention_slicing_forward_pass(test_mean_pixel_difference=_SCREAMING_SNAKE_CASE ,					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   lowerCAmelCase_	(       self						:	Dict       ):
      self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_SCREAMING_SNAKE_CASE ,					expected_max_diff=1e-2       )


    @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""       )
    def   lowerCAmelCase_	(       self						:	Dict       ):
      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						:	int       ):
      pass


    def   lowerCAmelCase_	(       self						:	Optional[Any]       ):
      return super().test_progress_bar()



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


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

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

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

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

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


    def   lowerCAmelCase_	(       self						:	List[str]       ):
      _UpperCAmelCase    			=   load_numpy(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"""       )

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

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

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

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






'''simple docstring'''





from pathlib import Path
from typing import List

from transformers import is_torch_available, is_vision_available
from transformers.testing_utils import get_tests_dir, is_tool_test
from transformers.tools.agent_types import AGENT_TYPE_MAPPING, AgentAudio, AgentImage, AgentText


if is_torch_available():
   import torch

if is_vision_available():
   from PIL import Image


UpperCamelCase_		:     int          =		['''text''', '''image''', '''audio''']





def       __a      (  _UpperCamelCase:       List[str]       )		->					Dict:

     """simple docstring"""




     _snake_case				= []

     for input_type in input_types:
          if input_type == "text":
               inputs.append("Text input"       )
          elif input_type == "image":
               inputs.append(
                   Image.open(Path(get_tests_dir("fixtures/tests_samples/COCO"       )       ) / "000000039769.png"       ).resize((512, 512)       )       )
          elif input_type == "audio":
               inputs.append(torch.ones(3_000       )       )
          elif isinstance(_UpperCamelCase   ,		_UpperCamelCase       ):
               inputs.append(create_inputs(_UpperCamelCase       )       )
          else:
               raise ValueError(F"""Invalid type requested: {input_type}"""       )

     return inputs





def       __a      (  _UpperCamelCase:       List       )		->					Dict:

     """simple docstring"""




     _snake_case				= []

     for output in outputs:
          if isinstance(_UpperCamelCase   ,		(str, AgentText)       ):
               output_types.append("text"       )
          elif isinstance(_UpperCamelCase   ,		(Image.Image, AgentImage)       ):
               output_types.append("image"       )
          elif isinstance(_UpperCamelCase   ,		(torch.Tensor, AgentAudio)       ):
               output_types.append("audio"       )
          else:
               raise ValueError(F"""Invalid output: {output}"""       )

     return output_types





@is_tool_test
class   _a							:






     def 							_lowercase      (					self       )				->     Any:
          self.assertTrue(hasattr(self.tool   ,"inputs"       )       )
          self.assertTrue(hasattr(self.tool   ,"outputs"       )       )

          _snake_case				= self.tool.inputs
          for _input in inputs:
               if isinstance(_input   ,_SCREAMING_SNAKE_CASE       ):
                    for __input in _input:
                         self.assertTrue(__input in authorized_types       )
               else:
                    self.assertTrue(_input in authorized_types       )

          _snake_case				= self.tool.outputs
          for _output in outputs:
               self.assertTrue(_output in authorized_types       )






     def 							_lowercase      (					self       )				->     Any:
          _snake_case				= create_inputs(self.tool.inputs       )
          _snake_case				= self.tool(*_SCREAMING_SNAKE_CASE       )

          # There is a single output
          if len(self.tool.outputs       ) == 1:
               _snake_case				= [outputs]

          self.assertListEqual(output_types(_SCREAMING_SNAKE_CASE       )   ,self.tool.outputs       )






     def 							_lowercase      (					self       )				->     str:
          self.assertTrue(hasattr(self.tool   ,"description"       )       )
          self.assertTrue(hasattr(self.tool   ,"default_checkpoint"       )       )
          self.assertTrue(self.tool.description.startswith("This is a tool that"       )       )






     def 							_lowercase      (					self       )				->     Tuple:
          _snake_case				= create_inputs(self.tool.inputs       )
          _snake_case				= self.tool(*_SCREAMING_SNAKE_CASE       )

          if not isinstance(_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE       ):
               _snake_case				= [outputs]

          self.assertEqual(len(_SCREAMING_SNAKE_CASE       )   ,len(self.tool.outputs       )       )

          for output, output_type in zip(_SCREAMING_SNAKE_CASE   ,self.tool.outputs       ):
               _snake_case				= AGENT_TYPE_MAPPING[output_type]
               self.assertTrue(isinstance(_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE       )       )






     def 							_lowercase      (					self       )				->     Optional[Any]:
          _snake_case				= create_inputs(self.tool.inputs       )

          _snake_case				= []

          for _input, input_type in zip(_SCREAMING_SNAKE_CASE   ,self.tool.inputs       ):
               if isinstance(_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE       ):
                    _inputs.append([AGENT_TYPE_MAPPING[_input_type](_input       ) for _input_type in input_type]       )
               else:
                    _inputs.append(AGENT_TYPE_MAPPING[input_type](_input       )       )

        # Should not raise an error
          _snake_case				= self.tool(*_SCREAMING_SNAKE_CASE       )

          if not isinstance(_SCREAMING_SNAKE_CASE   ,_SCREAMING_SNAKE_CASE       ):
               _snake_case				= [outputs]

          self.assertEqual(len(_SCREAMING_SNAKE_CASE       )   ,len(self.tool.outputs       )       )


