
"""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       lowerCAmelCase_		(				_lowerCamelCase				,				unittest.TestCase ):



							'''simple docstring'''





							_snake_case						      =		VideoToVideoSDPipeline
							_snake_case						      =		TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''} ) - {'image', 'width', 'height'}
							_snake_case						      =		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''} ) - {'image'}
							_snake_case						      =		PipelineTesterMixin.required_optional_params - {'latents'}
							_snake_case						      =		False

							# No `output_type`.
							_snake_case						      =		frozenset(
							    [
							        '''num_inference_steps''',
							        '''generator''',
							        '''latents''',
							        '''return_dict''',
							        '''callback''',
							        '''callback_steps''',
							    ] )




							def    A__						( self				)       ->  List[Any]:
													torch.manual_seed(0				)
													__lowerCAmelCase     =  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    ,				)
													__lowerCAmelCase     =  DDIMScheduler(
													    beta_start=0.00_085    ,				beta_end=0.012    ,				beta_schedule="""scaled_linear"""    ,				clip_sample=__UpperCamelCase    ,				set_alpha_to_one=__UpperCamelCase    ,				)
													torch.manual_seed(0				)
													__lowerCAmelCase     =  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				)
													__lowerCAmelCase     =  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=1_000    ,				hidden_act="""gelu"""    ,				projection_dim=512    ,				)
													__lowerCAmelCase     =  CLIPTextModel(__UpperCamelCase				)
													__lowerCAmelCase     =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""				)

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




							def    A__						( self    ,				snake_case_    ,				snake_case_=0				)       ->  Optional[int]:
													# 3 frames
													__lowerCAmelCase     =  floats_tensor((1, 3, 3, 32, 32)    ,				rng=random.Random(__UpperCamelCase				)				).to(__UpperCamelCase				)

													if str(__UpperCamelCase				).startswith("""mps"""				):
																			__lowerCAmelCase     =  torch.manual_seed(__UpperCamelCase				)
													else:
																			__lowerCAmelCase     =  torch.Generator(device=__UpperCamelCase				).manual_seed(__UpperCamelCase				)
													__lowerCAmelCase     =  {
													    """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    A__						( self				)       ->  str:
													__lowerCAmelCase     =  """cpu"""  # ensure determinism for the device-dependent torch.Generator
													__lowerCAmelCase     =  self.get_dummy_components()
													__lowerCAmelCase     =  VideoToVideoSDPipeline(**__UpperCamelCase				)
													__lowerCAmelCase     =  sd_pipe.to(__UpperCamelCase				)
													sd_pipe.set_progress_bar_config(disable=__UpperCamelCase				)

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

													assert frames[0].shape == (32, 32, 3)
													__lowerCAmelCase     =  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    A__						( self				)       ->  Tuple:
													self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__UpperCamelCase    ,				expected_max_diff=5e-3				)




							@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""				)
							def    A__						( self				)       ->  Any:
													pass




							@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""				)
							def    A__						( self				)       ->  Any:
													pass




							@unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""				)
							def    A__						( self				)       ->  Optional[Any]:
													pass




							def    A__						( self				)       ->  Any:
													return super().test_progress_bar()
@slow
@skip_mps
class       lowerCAmelCase_		(				unittest.TestCase ):



							'''simple docstring'''





							def    A__						( self				)       ->  Dict:
													__lowerCAmelCase     =  VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""    ,				torch_dtype=torch.floataa				)
													pipe.enable_model_cpu_offload()

													# 10 frames
													__lowerCAmelCase     =  torch.Generator(device="""cpu"""				).manual_seed(0				)
													__lowerCAmelCase     =  torch.randn((1, 10, 3, 1_024, 576)    ,				generator=__UpperCamelCase				)
													__lowerCAmelCase     =  video.to("""cuda"""				)

													__lowerCAmelCase     =  """Spiderman is surfing"""

													__lowerCAmelCase     =  pipe(__UpperCamelCase    ,				video=__UpperCamelCase    ,				generator=__UpperCamelCase    ,				num_inference_steps=3    ,				output_type="""pt"""				).frames

													__lowerCAmelCase     =  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





import json
import os
import unittest

from transformers.models.ctrl.tokenization_ctrl import VOCAB_FILES_NAMES, CTRLTokenizer

from ...test_tokenization_common import TokenizerTesterMixin



class        lowerCAmelCase__					(  _lowerCamelCase ,				unittest.TestCase       ):
				A_   :     int								=     CTRLTokenizer
				A_   :     List[str]								=     False
				A_   :     str								=     False


				def __UpperCamelCase						(							self      :  Any       )		->     Tuple:
											super().setUp()

											# Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
											A      						=    ['adapt', 're@@', 'a@@', 'apt', 'c@@', 't', '<unk>']
											A      						=    dict(zip(__UpperCamelCase						, range(len(__UpperCamelCase       )       )       )       )
											A      						=    ['#version: 0.2', 'a p', 'ap t</w>', 'r e', 'a d', 'ad apt</w>', '']
											A      						=    {'unk_token': '<unk>'}

											A      						=    os.path.join(self.tmpdirname						, VOCAB_FILES_NAMES['vocab_file']       )
											A      						=    os.path.join(self.tmpdirname						, VOCAB_FILES_NAMES['merges_file']       )
											with open(self.vocab_file						, 'w'						, encoding='utf-8'       ) as fp:
																		fp.write(json.dumps(__UpperCamelCase       ) + '\n'       )
											with open(self.merges_file						, 'w'						, encoding='utf-8'       ) as fp:
																		fp.write('\n'.join(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Optional[Any]						, **__UpperCamelCase      :  List[str]       )		->     List[Any]:
											kwargs.update(self.special_tokens_map       )
											return CTRLTokenizer.from_pretrained(self.tmpdirname						, **__UpperCamelCase       )


				def __UpperCamelCase						(							self      :  int						, __UpperCamelCase      :  Optional[int]       )		->     Optional[Any]:
											A      						=    'adapt react readapt apt'
											A      						=    'adapt react readapt apt'
											return input_text, output_text


				def __UpperCamelCase						(							self      :  List[Any]       )		->     Optional[int]:
											A      						=    CTRLTokenizer(self.vocab_file						, self.merges_file						, **self.special_tokens_map       )
											A      						=    'adapt react readapt apt'
											A      						=    'adapt re@@ a@@ c@@ t re@@ adapt apt'.split()
											A      						=    tokenizer.tokenize(__UpperCamelCase       )
											self.assertListEqual(__UpperCamelCase						, __UpperCamelCase       )

											A      						=    tokens + [tokenizer.unk_token]

											A      						=    [0, 1, 2, 4, 5, 1, 0, 3, 6]
											self.assertListEqual(tokenizer.convert_tokens_to_ids(__UpperCamelCase       )						, __UpperCamelCase       )