


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'''







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

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






					def A      (				self						)   ->			List[Any]:







								'''simple docstring'''

								torch.manual_seed(0						)
								__lowercase 					=   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 ,       )
								__lowercase 					=   DDIMScheduler(
								    beta_start=0.0_0_0_8_5 ,       beta_end=0.0_1_2 ,       beta_schedule='''scaled_linear''' ,       clip_sample=snake_case_ ,       set_alpha_to_one=snake_case_ ,       )
								torch.manual_seed(0						)
								__lowercase 					=   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						)
								__lowercase 					=   CLIPTextConfig(
								    bos_token_id=0 ,       eos_token_id=2 ,       hidden_size=3_2 ,       intermediate_size=3_7 ,       layer_norm_eps=1e-0_5 ,       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 ,       )
								__lowercase 					=   CLIPTextModel(snake_case_						)
								__lowercase 					=   CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

								__lowercase 					=   {
								    '''unet''': unet,
								    '''scheduler''': scheduler,
								    '''vae''': vae,
								    '''text_encoder''': text_encoder,
								    '''tokenizer''': tokenizer,
								}
								return components






					def A      (				self ,       snake_case_ ,       snake_case_=0						)   ->			Dict:







								'''simple docstring'''

								__lowercase 					=   floats_tensor((1, 3, 3, 3_2, 3_2) ,       rng=random.Random(snake_case_						)						).to(snake_case_						)

								if str(snake_case_						).startswith('''mps'''						):
											__lowercase 					=   torch.manual_seed(snake_case_						)
								else:
											__lowercase 					=   torch.Generator(device=snake_case_						).manual_seed(snake_case_						)
								__lowercase 					=   {
								    '''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						)   ->			List[str]:







								'''simple docstring'''

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

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

								assert frames[0].shape == (3_2, 3_2, 3)
								__lowercase 					=   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 A      (				self						)   ->			Any:







								'''simple docstring'''

								self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case_ ,       expected_max_diff=5e-3						)






					@unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''						)
					def A      (				self						)   ->			List[Any]:







								'''simple docstring'''

								pass






					@unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''						)
					def A      (				self						)   ->			List[Any]:







								'''simple docstring'''

								pass






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







								'''simple docstring'''

								pass






					def A      (				self						)   ->			Optional[int]:







								'''simple docstring'''

								return super().test_progress_bar()




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



					'''simple docstring'''







					def A      (				self						)   ->			List[Any]:







								'''simple docstring'''

								__lowercase 					=   VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL''' ,       torch_dtype=torch.floataa						)
								pipe.enable_model_cpu_offload()

								# 10 frames
								__lowercase 					=   torch.Generator(device='''cpu'''						).manual_seed(0						)
								__lowercase 					=   torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6) ,       generator=snake_case_						)
								__lowercase 					=   video.to('''cuda'''						)

								__lowercase 					=   '''Spiderman is surfing'''

								__lowercase 					=   pipe(snake_case_ ,       video=snake_case_ ,       generator=snake_case_ ,       num_inference_steps=3 ,       output_type='''pt'''						).frames

								__lowercase 					=   np.array([-1.0_4_5_8_9_8_4, -1.1_2_7_9_2_9_7, -0.9_6_6_3_0_8_6, -0.9_1_5_0_3_9_0_6, -0.7_5_0_9_7_6_5_6]						)
								assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array						).sum() < 1e-2






import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


a : int		=				logging.get_logger(__name__)


a : Any		=				{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

# See all LED models at https://huggingface.co/models?filter=LED
a : Dict		=				{
    '''vocab_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt''',
    },
    '''tokenizer_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json''',
    },
}

a : Optional[int]		=				{
    '''allenai/led-base-16384''': 16384,
}


@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def 		lowercase_  (     ):


			'''simple docstring'''







			__lowercase 					=   (
			    list(range(ord('''!'''  )       ,			ord('''~'''  ) + 1  )  ) + list(range(ord('''¡'''  )       ,			ord('''¬'''  ) + 1  )  ) + list(range(ord('''®'''  )       ,			ord('''ÿ'''  ) + 1  )  )
			)
			__lowercase 					=   bs[:]
			__lowercase 					=   0
			for b in range(2**8  ):
						if b not in bs:
									bs.append(_UpperCamelCase  )
									cs.append(2**8 + n  )
									n += 1
			__lowercase 					=   [chr(_UpperCamelCase  ) for n in cs]
			return dict(zip(_UpperCamelCase       ,			_UpperCamelCase  )  )


def 		lowercase_  (     _UpperCamelCase  ):


			'''simple docstring'''







			__lowercase 					=   set()
			__lowercase 					=   word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)  )
						__lowercase 					=   char
			return pairs





class 	lowerCamelCase_       (				lowerCAmelCase__				):



					'''simple docstring'''







					__UpperCAmelCase            =				VOCAB_FILES_NAMES
					__UpperCAmelCase            =				PRETRAINED_VOCAB_FILES_MAP
					__UpperCAmelCase            =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					__UpperCAmelCase            =				["input_ids", "attention_mask"]






					def __init__(				self ,       snake_case_ ,       snake_case_ ,       snake_case_="replace" ,       snake_case_="<s>" ,       snake_case_="</s>" ,       snake_case_="</s>" ,       snake_case_="<s>" ,       snake_case_="<unk>" ,       snake_case_="<pad>" ,       snake_case_="<mask>" ,       snake_case_=False ,       **snake_case_ ,       )   ->			List[Any]:







								'''simple docstring'''

								__lowercase 					=   AddedToken(snake_case_ ,       lstrip=snake_case_ ,       rstrip=snake_case_						) if isinstance(snake_case_ ,       snake_case_						) else bos_token
								__lowercase 					=   AddedToken(snake_case_ ,       lstrip=snake_case_ ,       rstrip=snake_case_						) if isinstance(snake_case_ ,       snake_case_						) else eos_token
								__lowercase 					=   AddedToken(snake_case_ ,       lstrip=snake_case_ ,       rstrip=snake_case_						) if isinstance(snake_case_ ,       snake_case_						) else sep_token
								__lowercase 					=   AddedToken(snake_case_ ,       lstrip=snake_case_ ,       rstrip=snake_case_						) if isinstance(snake_case_ ,       snake_case_						) else cls_token
								__lowercase 					=   AddedToken(snake_case_ ,       lstrip=snake_case_ ,       rstrip=snake_case_						) if isinstance(snake_case_ ,       snake_case_						) else unk_token
								__lowercase 					=   AddedToken(snake_case_ ,       lstrip=snake_case_ ,       rstrip=snake_case_						) if isinstance(snake_case_ ,       snake_case_						) else pad_token

								# Mask token behave like a normal word, i.e. include the space before it
								__lowercase 					=   AddedToken(snake_case_ ,       lstrip=snake_case_ ,       rstrip=snake_case_						) if isinstance(snake_case_ ,       snake_case_						) else mask_token

								super().__init__(
								    errors=snake_case_ ,       bos_token=snake_case_ ,       eos_token=snake_case_ ,       unk_token=snake_case_ ,       sep_token=snake_case_ ,       cls_token=snake_case_ ,       pad_token=snake_case_ ,       mask_token=snake_case_ ,       add_prefix_space=snake_case_ ,       **snake_case_ ,       )

								with open(snake_case_ ,       encoding='''utf-8'''						) as vocab_handle:
											__lowercase 					=   json.load(snake_case_						)
								__lowercase 					=   {v: k for k, v in self.encoder.items()}
								__lowercase 					=   errors  # how to handle errors in decoding
								__lowercase 					=   bytes_to_unicode()
								__lowercase 					=   {v: k for k, v in self.byte_encoder.items()}
								with open(snake_case_ ,       encoding='''utf-8'''						) as merges_handle:
											__lowercase 					=   merges_handle.read().split('''\n'''						)[1:-1]
								__lowercase 					=   [tuple(merge.split()						) for merge in bpe_merges]
								__lowercase 					=   dict(zip(snake_case_ ,       range(len(snake_case_						)						)						)						)
								__lowercase 					=   {}
								__lowercase 					=   add_prefix_space

								# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
								__lowercase 					=   re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''						)






					@property
					# Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
					def A      (				self						)   ->			Union[str, Any]:







								'''simple docstring'''

								return len(self.encoder						)






					def A      (				self						)   ->			str:







								'''simple docstring'''

								return dict(self.encoder ,       **self.added_tokens_encoder						)






					def A      (				self ,       snake_case_						)   ->			Optional[int]:







								'''simple docstring'''

								if token in self.cache:
											return self.cache[token]
								__lowercase 					=   tuple(snake_case_						)
								__lowercase 					=   get_pairs(snake_case_						)

								if not pairs:
											return token

								while True:
											__lowercase 					=   min(snake_case_ ,       key=lambda snake_case_						: self.bpe_ranks.get(snake_case_ ,       float('''inf'''						)						)						)
											if bigram not in self.bpe_ranks:
														break
											__lowercase					, __lowercase 					=   bigram
											__lowercase 					=   []
											__lowercase 					=   0
											while i < len(snake_case_						):
														try:
																	__lowercase 					=   word.index(snake_case_ ,       snake_case_						)
														except ValueError:
																	new_word.extend(word[i:]						)
																	break
														else:
																	new_word.extend(word[i:j]						)
																	__lowercase 					=   j

														if word[i] == first and i < len(snake_case_						) - 1 and word[i + 1] == second:
																	new_word.append(first + second						)
																	i += 2
														else:
																	new_word.append(word[i]						)
																	i += 1
											__lowercase 					=   tuple(snake_case_						)
											__lowercase 					=   new_word
											if len(snake_case_						) == 1:
														break
											else:
														__lowercase 					=   get_pairs(snake_case_						)
								__lowercase 					=   ''' '''.join(snake_case_						)
								__lowercase 					=   word
								return word






					def A      (				self ,       snake_case_						)   ->			Tuple:







								'''simple docstring'''

								__lowercase 					=   []
								for token in re.findall(self.pat ,       snake_case_						):
											__lowercase 					=   ''''''.join(
											    self.byte_encoder[b] for b in token.encode('''utf-8'''						)						)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
											bpe_tokens.extend(bpe_token for bpe_token in self.bpe(snake_case_						).split(''' '''						)						)
								return bpe_tokens






					def A      (				self ,       snake_case_						)   ->			List[Any]:







								'''simple docstring'''

								return self.encoder.get(snake_case_ ,       self.encoder.get(self.unk_token						)						)






					def A      (				self ,       snake_case_						)   ->			Tuple:







								'''simple docstring'''

								return self.decoder.get(snake_case_						)






					def A      (				self ,       snake_case_						)   ->			Any:







								'''simple docstring'''

								__lowercase 					=   ''''''.join(snake_case_						)
								__lowercase 					=   bytearray([self.byte_decoder[c] for c in text]						).decode('''utf-8''' ,       errors=self.errors						)
								return text






					def A      (				self ,       snake_case_ ,       snake_case_ = None						)   ->			Tuple[str]:







								'''simple docstring'''

								if not os.path.isdir(snake_case_						):
											logger.error(F'Vocabulary path ({save_directory}) should be a directory'						)
											return
								__lowercase 					=   os.path.join(
								    snake_case_ ,       (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']						)
								__lowercase 					=   os.path.join(
								    snake_case_ ,       (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']						)

								with open(snake_case_ ,       '''w''' ,       encoding='''utf-8'''						) as f:
											f.write(json.dumps(self.encoder ,       indent=2 ,       sort_keys=snake_case_ ,       ensure_ascii=snake_case_						) + '''\n'''						)

								__lowercase 					=   0
								with open(snake_case_ ,       '''w''' ,       encoding='''utf-8'''						) as writer:
											writer.write('''#version: 0.2\n'''						)
											for bpe_tokens, token_index in sorted(self.bpe_ranks.items() ,       key=lambda snake_case_						: kv[1]						):
														if index != token_index:
																	logger.warning(
																	    F'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
																	    ''' Please check that the tokenizer is not corrupted!'''						)
																	__lowercase 					=   token_index
														writer.write(''' '''.join(snake_case_						) + '''\n'''						)
														index += 1

								return vocab_file, merge_file






					def A      (				self ,       snake_case_ ,       snake_case_ = None						)   ->			List[int]:







								'''simple docstring'''

								if token_ids_a is None:
											return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
								__lowercase 					=   [self.cls_token_id]
								__lowercase 					=   [self.sep_token_id]
								return cls + token_ids_a + sep + sep + token_ids_a + sep






					def A      (				self ,       snake_case_ ,       snake_case_ = None ,       snake_case_ = False						)   ->			List[int]:







								'''simple docstring'''

								if already_has_special_tokens:
											return super().get_special_tokens_mask(
											    token_ids_a=snake_case_ ,       token_ids_a=snake_case_ ,       already_has_special_tokens=snake_case_						)

								if token_ids_a is None:
											return [1] + ([0] * len(snake_case_						)) + [1]
								return [1] + ([0] * len(snake_case_						)) + [1, 1] + ([0] * len(snake_case_						)) + [1]






					def A      (				self ,       snake_case_ ,       snake_case_ = None						)   ->			List[int]:







								'''simple docstring'''

								__lowercase 					=   [self.sep_token_id]
								__lowercase 					=   [self.cls_token_id]

								if token_ids_a is None:
											return len(cls + token_ids_a + sep						) * [0]
								return len(cls + token_ids_a + sep + sep + token_ids_a + sep						) * [0]






					def A      (				self ,       snake_case_ ,       snake_case_=False ,       **snake_case_						)   ->			Any:







								'''simple docstring'''

								__lowercase 					=   kwargs.pop('''add_prefix_space''' ,       self.add_prefix_space						)
								if (is_split_into_words or add_prefix_space) and (len(snake_case_						) > 0 and not text[0].isspace()):
											__lowercase 					=   ''' ''' + text
								return (text, kwargs)






					def A      (				self ,       snake_case_ ,       snake_case_ = None ,       snake_case_ = PaddingStrategy.DO_NOT_PAD ,       snake_case_ = None ,       snake_case_ = None ,       )   ->			dict:







								'''simple docstring'''

								__lowercase 					=   super()._pad(
								    encoded_inputs=snake_case_ ,       max_length=snake_case_ ,       padding_strategy=snake_case_ ,       pad_to_multiple_of=snake_case_ ,       return_attention_mask=snake_case_ ,       )

								# Load from model defaults
								if return_attention_mask is None:
											__lowercase 					=   '''attention_mask''' in self.model_input_names

								if return_attention_mask and "global_attention_mask" in encoded_inputs:
											__lowercase 					=   encoded_inputs[self.model_input_names[0]]
											# `global_attention_mask` need to have the same length as other (sequential) inputs.
											__lowercase 					=   len(encoded_inputs['''global_attention_mask''']						) != len(snake_case_						)

											if needs_to_be_padded:
														__lowercase 					=   len(snake_case_						) - len(encoded_inputs['''global_attention_mask''']						)

														if self.padding_side == "right":
																	# Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
																	__lowercase 					=   (
																	    encoded_inputs['''global_attention_mask'''] + [-1] * difference
																	)
														elif self.padding_side == "left":
																	__lowercase 					=   [-1] * difference + encoded_inputs[
																	    '''global_attention_mask'''
																	]
														else:
																	raise ValueError('''Invalid padding strategy:''' + str(self.padding_side						)						)

								return encoded_inputs


