

import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

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


enable_full_determinism()


class      lowerCamelCase_ (			_lowerCAmelCase	,				_lowerCAmelCase	,				unittest.TestCase):
 _lowerCAmelCase									=	StableDiffusionParadigmsPipeline
 _lowerCAmelCase									=	TEXT_TO_IMAGE_PARAMS
 _lowerCAmelCase									=	TEXT_TO_IMAGE_BATCH_PARAMS
 _lowerCAmelCase									=	TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowerCAmelCase									=	TEXT_TO_IMAGE_IMAGE_PARAMS




 def     a       (			self							):
   torch.manual_seed(0							)
   __UpperCAmelCase	:  Dict		=     UNetaDConditionModel(
       block_out_channels=(32, 64)  ,  layers_per_block=2  ,  sample_size=32  ,  in_channels=4  ,  out_channels=4  ,  down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')  ,  up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')  ,  cross_attention_dim=32  ,  attention_head_dim=(2, 4)  ,  use_linear_projection=_UpperCAmelCase  ,  )
   __UpperCAmelCase	:  List[str]		=     DDIMParallelScheduler(
       beta_start=0.0_0_0_8_5  ,  beta_end=0.0_1_2  ,  beta_schedule='scaled_linear'  ,  clip_sample=_UpperCAmelCase  ,  set_alpha_to_one=_UpperCAmelCase  ,  )
   torch.manual_seed(0							)
   __UpperCAmelCase	:  List[str]		=     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[int]		=     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  ,  )
   __UpperCAmelCase	:  Optional[int]		=     CLIPTextModel(_UpperCAmelCase							)
   __UpperCAmelCase	:  int		=     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'							)

   __UpperCAmelCase	:  int		=     {
       'unet': unet,
       'scheduler': scheduler,
       'vae': vae,
       'text_encoder': text_encoder,
       'tokenizer': tokenizer,
       'safety_checker': None,
       'feature_extractor': None,
   }
   return components




 def     a       (			self  ,  _UpperCAmelCase  ,  _UpperCAmelCase=0							):
   if str(_UpperCAmelCase							).startswith('mps'							):
     __UpperCAmelCase	:  Optional[int]		=     torch.manual_seed(_UpperCAmelCase							)
   else:
     __UpperCAmelCase	:  str		=     torch.Generator(device=_UpperCAmelCase							).manual_seed(_UpperCAmelCase							)
   __UpperCAmelCase	:  Any		=     {
       'prompt': 'a photograph of an astronaut riding a horse',
       'generator': generator,
       'num_inference_steps': 10,
       'guidance_scale': 6.0,
       'output_type': 'numpy',
       'parallel': 3,
       'debug': True,
   }
   return inputs




 def     a       (			self							):
   __UpperCAmelCase	:  Optional[int]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  List[str]		=     self.get_dummy_components()
   __UpperCAmelCase	:  List[Any]		=     StableDiffusionParadigmsPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  List[str]		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  Optional[int]		=     self.get_dummy_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  List[str]		=     sd_pipe(**_UpperCAmelCase							).images
   __UpperCAmelCase	:  List[str]		=     image[0, -3:, -3:, -1]
   assert image.shape == (1, 64, 64, 3)

   __UpperCAmelCase	:  Optional[Any]		=     np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]							)

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




 def     a       (			self							):
   __UpperCAmelCase	:  List[str]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  Any		=     self.get_dummy_components()
   torch.manual_seed(0							)
   __UpperCAmelCase	:  List[Any]		=     DDPMParallelScheduler()
   torch.manual_seed(0							)
   __UpperCAmelCase	:  Dict		=     StableDiffusionParadigmsPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  Union[str, Any]		=     self.get_dummy_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  int		=     sd_pipe(**_UpperCAmelCase							).images
   __UpperCAmelCase	:  Tuple		=     image[0, -3:, -3:, -1]
   assert image.shape == (1, 64, 64, 3)

   __UpperCAmelCase	:  List[Any]		=     np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]							)

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




 def     a       (			self							):
   super().test_inference_batch_consistent(batch_sizes=[1, 2]							)




 def     a       (			self							):
   super().test_inference_batch_single_identical(batch_size=2  ,  expected_max_diff=3E-3							)




 def     a       (			self							):
   __UpperCAmelCase	:  Optional[Any]		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
   __UpperCAmelCase	:  List[Any]		=     self.get_dummy_components()
   __UpperCAmelCase	:  List[Any]		=     StableDiffusionParadigmsPipeline(**_UpperCAmelCase							)
   __UpperCAmelCase	:  Optional[int]		=     sd_pipe.to(_UpperCAmelCase							)
   sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase							)

   __UpperCAmelCase	:  List[str]		=     self.get_dummy_inputs(_UpperCAmelCase							)
   __UpperCAmelCase	:  int		=     'french fries'
   __UpperCAmelCase	:  str		=     sd_pipe(**_UpperCAmelCase  ,  negative_prompt=_UpperCAmelCase							)
   __UpperCAmelCase	:  Union[str, Any]		=     output.images
   __UpperCAmelCase	:  Optional[int]		=     image[0, -3:, -3:, -1]

   assert image.shape == (1, 64, 64, 3)

   __UpperCAmelCase	:  Optional[Any]		=     np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]							)

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



@slow
@require_torch_gpu
class      lowerCamelCase_ (			unittest.TestCase):




 def     a       (			self							):
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()




 def     a       (			self  ,  _UpperCAmelCase=0							):
   __UpperCAmelCase	:  List[Any]		=     torch.Generator(device=_UpperCAmelCase							).manual_seed(_UpperCAmelCase							)
   __UpperCAmelCase	:  Tuple		=     {
       'prompt': 'a photograph of an astronaut riding a horse',
       'generator': generator,
       'num_inference_steps': 10,
       'guidance_scale': 7.5,
       'output_type': 'numpy',
       'parallel': 3,
       'debug': True,
   }
   return inputs




 def     a       (			self							):
   __UpperCAmelCase	:  str		=     'stabilityai/stable-diffusion-2-base'
   __UpperCAmelCase	:  Optional[int]		=     DDIMParallelScheduler.from_pretrained(_UpperCAmelCase  ,  subfolder='scheduler'							)
   __UpperCAmelCase	:  str		=     StableDiffusionParadigmsPipeline.from_pretrained(_UpperCAmelCase  ,  scheduler=_UpperCAmelCase  ,  safety_checker=_UpperCAmelCase							)
   pipe.to(_UpperCAmelCase							)
   pipe.set_progress_bar_config(disable=_UpperCAmelCase							)
   pipe.enable_attention_slicing()

   __UpperCAmelCase	:  List[str]		=     self.get_inputs()
   __UpperCAmelCase	:  Dict		=     pipe(**_UpperCAmelCase							).images
   __UpperCAmelCase	:  Union[str, Any]		=     image[0, -3:, -3:, -1].flatten()

   assert image.shape == (1, 512, 512, 3)

   __UpperCAmelCase	:  Any		=     np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]							)

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













from ...configuration_utils import PretrainedConfig
from ...utils import logging


A										=      logging.get_logger(__name__)

A										=      {
    'ZinengTang/tvlt-base': 'https://huggingface.co/ZinengTang/tvlt-base/blob/main/config.json',
}


class 				__a				(	__A    ):





		'''simple docstring'''



		UpperCAmelCase__   :					str					     =    """tvlt"""


		def __init__(    self			,     UpperCamelCase__=224			,     UpperCamelCase__=2048			,     UpperCamelCase__=128			,     UpperCamelCase__=[16, 16]			,     UpperCamelCase__=[16, 16]			,     UpperCamelCase__=3			,     UpperCamelCase__=1			,     UpperCamelCase__=8			,     UpperCamelCase__=768			,     UpperCamelCase__=12			,     UpperCamelCase__=12			,     UpperCamelCase__=3072			,     UpperCamelCase__="gelu"			,     UpperCamelCase__=0.0			,     UpperCamelCase__=0.0			,     UpperCamelCase__=0.02			,     UpperCamelCase__=1E-6			,     UpperCamelCase__=True			,     UpperCamelCase__=False			,     UpperCamelCase__=16			,     UpperCamelCase__=512			,     UpperCamelCase__=8			,     UpperCamelCase__=2048			,     UpperCamelCase__=0.75			,     UpperCamelCase__=0.15			,     UpperCamelCase__="frame-level"			,     UpperCamelCase__=True			,     UpperCamelCase__=True			,     UpperCamelCase__="classification"			,     **UpperCamelCase__			,     ):
						super().__init__(**UpperCamelCase__							)

						if audio_mask_type not in ("frame-level", "patch_level"):
										raise ValueError(
										    'audio_mask_type must be one of two acceptable strategies - {\'frame_level\', \'patch-level\') '
										    F'''got {audio_mask_type}'''							)

						SCREAMING_SNAKE_CASE_ :						Tuple    				= image_size
						SCREAMING_SNAKE_CASE_ :						int    				= spectrogram_length
						SCREAMING_SNAKE_CASE_ :						Optional[Any]    				= frequency_length
						SCREAMING_SNAKE_CASE_ :						Any    				= image_patch_size
						SCREAMING_SNAKE_CASE_ :						List[Any]    				= audio_patch_size
						SCREAMING_SNAKE_CASE_ :						int    				= num_image_channels
						SCREAMING_SNAKE_CASE_ :						Any    				= num_audio_channels
						SCREAMING_SNAKE_CASE_ :						Any    				= num_frames

						SCREAMING_SNAKE_CASE_ :						Optional[int]    				= hidden_size
						SCREAMING_SNAKE_CASE_ :						str    				= num_hidden_layers
						SCREAMING_SNAKE_CASE_ :						Optional[Any]    				= num_attention_heads
						SCREAMING_SNAKE_CASE_ :						str    				= intermediate_size
						SCREAMING_SNAKE_CASE_ :						Optional[Any]    				= hidden_act
						SCREAMING_SNAKE_CASE_ :						Optional[int]    				= hidden_dropout_prob
						SCREAMING_SNAKE_CASE_ :						int    				= attention_probs_dropout_prob
						SCREAMING_SNAKE_CASE_ :						Dict    				= initializer_range
						SCREAMING_SNAKE_CASE_ :						Dict    				= layer_norm_eps
						SCREAMING_SNAKE_CASE_ :						Union[str, Any]    				= qkv_bias
						SCREAMING_SNAKE_CASE_ :						List[str]    				= use_mean_pooling

						SCREAMING_SNAKE_CASE_ :						str    				= decoder_num_attention_heads
						SCREAMING_SNAKE_CASE_ :						Optional[int]    				= decoder_hidden_size
						SCREAMING_SNAKE_CASE_ :						Any    				= decoder_num_hidden_layers
						SCREAMING_SNAKE_CASE_ :						List[Any]    				= decoder_intermediate_size
						SCREAMING_SNAKE_CASE_ :						List[str]    				= pixel_mask_ratio
						SCREAMING_SNAKE_CASE_ :						Optional[int]    				= audio_mask_ratio
						SCREAMING_SNAKE_CASE_ :						Optional[Any]    				= audio_mask_type

						SCREAMING_SNAKE_CASE_ :						Any    				= task_matching
						SCREAMING_SNAKE_CASE_ :						Any    				= task_mae
						SCREAMING_SNAKE_CASE_ :						int    				= loss_type