





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 					__A    (    lowerCAmelCase			,				unittest.TestCase ):

			'''simple docstring'''







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

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





			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				torch.manual_seed(0							)
				lowerCamelCase__								=		UNetaDConditionModel(
				    block_out_channels=(3_2, 6_4, 6_4, 6_4)       ,  layers_per_block=2       ,  sample_size=3_2       ,  in_channels=4       ,  out_channels=4       ,  down_block_types=('''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''CrossAttnDownBlock3D''', '''DownBlock3D''')       ,  up_block_types=('''UpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''', '''CrossAttnUpBlock3D''')       ,  cross_attention_dim=3_2       ,  attention_head_dim=4       ,  )
				lowerCamelCase__								=		DDIMScheduler(
				    beta_start=0.0_0085       ,  beta_end=0.012       ,  beta_schedule='''scaled_linear'''       ,  clip_sample=__lowerCAmelCase       ,  set_alpha_to_one=__lowerCAmelCase       ,  )
				torch.manual_seed(0							)
				lowerCamelCase__								=		AutoencoderKL(
				    block_out_channels=[3_2, 6_4]       ,  in_channels=3       ,  out_channels=3       ,  down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']       ,  up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']       ,  latent_channels=4       ,  sample_size=1_2_8       ,  )
				torch.manual_seed(0							)
				lowerCamelCase__								=		CLIPTextConfig(
				    bos_token_id=0       ,  eos_token_id=2       ,  hidden_size=3_2       ,  intermediate_size=3_7       ,  layer_norm_eps=1E-05       ,  num_attention_heads=4       ,  num_hidden_layers=5       ,  pad_token_id=1       ,  vocab_size=1_0_0_0       ,  hidden_act='''gelu'''       ,  projection_dim=5_1_2       ,  )
				lowerCamelCase__								=		CLIPTextModel(__lowerCAmelCase							)
				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 		__lowerCamelCase				(    self       ,  __lowerCAmelCase       ,  __lowerCAmelCase=0							):



				'''simple docstring'''







				lowerCamelCase__								=		floats_tensor((1, 3, 3, 3_2, 3_2)       ,  rng=random.Random(__lowerCAmelCase							)							).to(__lowerCAmelCase							)

				if str(__lowerCAmelCase							).startswith('''mps'''							):
					lowerCamelCase__								=		torch.manual_seed(__lowerCAmelCase							)
				else:
					lowerCamelCase__								=		torch.Generator(device=__lowerCAmelCase							).manual_seed(__lowerCAmelCase							)
				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 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				lowerCamelCase__								=		'''cpu'''  # ensure determinism for the device-dependent torch.Generator
				lowerCamelCase__								=		self.get_dummy_components()
				lowerCamelCase__								=		VideoToVideoSDPipeline(**__lowerCAmelCase							)
				lowerCamelCase__								=		sd_pipe.to(__lowerCAmelCase							)
				sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase							)

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

				assert frames[0].shape == (3_2, 3_2, 3)
				lowerCamelCase__								=		np.array([1_0_6, 1_1_7, 1_1_3, 1_7_4, 1_3_7, 1_1_2, 1_4_8, 1_5_1, 1_3_1]							)

				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 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__lowerCAmelCase       ,  expected_max_diff=5E-3							)





			@unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''							)
			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				pass





			@unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''							)
			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				pass





			@unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''							)
			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				pass





			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				return super().test_progress_bar()






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

			'''simple docstring'''







			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				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, 1_0, 3, 1_0_2_4, 5_7_6)       ,  generator=__lowerCAmelCase							)
				lowerCamelCase__								=		video.to('''cuda'''							)

				lowerCamelCase__								=		'''Spiderman is surfing'''

				lowerCamelCase__								=		pipe(__lowerCAmelCase       ,  video=__lowerCAmelCase       ,  generator=__lowerCAmelCase       ,  num_inference_steps=3       ,  output_type='''pt'''							).frames

				lowerCamelCase__								=		np.array([-1.045_8984, -1.127_9297, -0.966_3086, -0.9150_3906, -0.7509_7656]							)
				assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array							).sum() < 1E-2










import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin






class 					__A    (    lowerCAmelCase ):

			'''simple docstring'''







			lowerCAmelCase_							     =						"""Speech2TextFeatureExtractor"""
			lowerCAmelCase_							     =						"""Speech2TextTokenizer"""





			def __init__(    self       ,  __lowerCAmelCase       ,  __lowerCAmelCase							):



				'''simple docstring'''







				super().__init__(__lowerCAmelCase       ,  __lowerCAmelCase							)
				lowerCamelCase__								=		self.feature_extractor
				lowerCamelCase__								=		False





			def __call__(    self       ,  *__lowerCAmelCase       ,  **__lowerCAmelCase							):



				'''simple docstring'''







				if self._in_target_context_manager:
					return self.current_processor(*__lowerCAmelCase       ,  **__lowerCAmelCase							)

				if "raw_speech" in kwargs:
					warnings.warn('''Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'''							)
					lowerCamelCase__								=		kwargs.pop('''raw_speech'''							)
				else:
					lowerCamelCase__								=		kwargs.pop('''audio'''       ,  __lowerCAmelCase							)
				lowerCamelCase__								=		kwargs.pop('''sampling_rate'''       ,  __lowerCAmelCase							)
				lowerCamelCase__								=		kwargs.pop('''text'''       ,  __lowerCAmelCase							)
				if len(__lowerCAmelCase							) > 0:
					lowerCamelCase__								=		args[0]
					lowerCamelCase__								=		args[1:]

				if audio is None and text is None:
					raise ValueError('''You need to specify either an `audio` or `text` input to process.'''							)

				if audio is not None:
					lowerCamelCase__								=		self.feature_extractor(__lowerCAmelCase       ,  *__lowerCAmelCase       ,  sampling_rate=__lowerCAmelCase       ,  **__lowerCAmelCase							)
				if text is not None:
					lowerCamelCase__								=		self.tokenizer(__lowerCAmelCase       ,  **__lowerCAmelCase							)

				if text is None:
					return inputs
				elif audio is None:
					return encodings
				else:
					lowerCamelCase__								=		encodings['''input_ids''']
					return inputs





			def 		__lowerCamelCase				(    self       ,  *__lowerCAmelCase       ,  **__lowerCAmelCase							):



				'''simple docstring'''







				return self.tokenizer.batch_decode(*__lowerCAmelCase       ,  **__lowerCAmelCase							)





			def 		__lowerCamelCase				(    self       ,  *__lowerCAmelCase       ,  **__lowerCAmelCase							):



				'''simple docstring'''







				return self.tokenizer.decode(*__lowerCAmelCase       ,  **__lowerCAmelCase							)





			@contextmanager
			def 		__lowerCamelCase				(    self							):



				'''simple docstring'''







				warnings.warn(
				    '''`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '''
				    '''labels by using the argument `text` of the regular `__call__` method (either in the same call as '''
				    '''your audio inputs, or in a separate call.'''							)
				lowerCamelCase__								=		True
				lowerCamelCase__								=		self.tokenizer
				yield
				lowerCamelCase__								=		self.feature_extractor
				lowerCamelCase__								=		False



