






import gc
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, KandinskyVaaPipeline, KandinskyVaaPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()

class 				lowerCAmelCase_	(		__snake_case   ,		unittest.TestCase			):
				_UpperCamelCase					:       List[Any]							       = KandinskyVaaPipeline
				_UpperCamelCase					:       int							       = [
				    "image_embeds",
				    "negative_image_embeds",
				]
				_UpperCamelCase					:       Optional[int]							       = ["image_embeds", "negative_image_embeds"]
				_UpperCamelCase					:       Optional[Any]							       = [
				    "generator",
				    "height",
				    "width",
				    "latents",
				    "guidance_scale",
				    "num_inference_steps",
				    "return_dict",
				    "guidance_scale",
				    "num_images_per_prompt",
				    "output_type",
				    "return_dict",
				]
				_UpperCamelCase					:       Dict							       = False


				@property
				def 						__a					(					self      ):
											return 3_2


				@property
				def 						__a					(					self      ):
											return 3_2


				@property
				def 						__a					(					self      ):
											return self.time_input_dim


				@property
				def 						__a					(					self      ):
											return self.time_input_dim * 4


				@property
				def 						__a					(					self      ):
											return 1_0_0


				@property
				def 						__a					(					self      ):
											torch.manual_seed(0      )

											_lowercase							:   Tuple 							=   {
											    'in_channels': 4,
											    # Out channels is double in channels because predicts mean and variance
											    'out_channels': 8,
											    'addition_embed_type': 'image',
											    'down_block_types': ('ResnetDownsampleBlock2D', 'SimpleCrossAttnDownBlock2D'),
											    'up_block_types': ('SimpleCrossAttnUpBlock2D', 'ResnetUpsampleBlock2D'),
											    'mid_block_type': 'UNetMidBlock2DSimpleCrossAttn',
											    'block_out_channels': (self.block_out_channels_a, self.block_out_channels_a * 2),
											    'layers_per_block': 1,
											    'encoder_hid_dim': self.text_embedder_hidden_size,
											    'encoder_hid_dim_type': 'image_proj',
											    'cross_attention_dim': self.cross_attention_dim,
											    'attention_head_dim': 4,
											    'resnet_time_scale_shift': 'scale_shift',
											    'class_embed_type': None,
											}

											_lowercase							:   Union[str, Any] 							=   UNetaDConditionModel(**_lowerCAmelCase      )
											return model


				@property
				def 						__a					(					self      ):
											return {
											    "block_out_channels": [3_2, 6_4],
											    "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
											    "in_channels": 3,
											    "latent_channels": 4,
											    "layers_per_block": 1,
											    "norm_num_groups": 8,
											    "norm_type": "spatial",
											    "num_vq_embeddings": 1_2,
											    "out_channels": 3,
											    "up_block_types": [
											        "AttnUpDecoderBlock2D",
											        "UpDecoderBlock2D",
											    ],
											    "vq_embed_dim": 4,
											}


				@property
				def 						__a					(					self      ):
											torch.manual_seed(0      )
											_lowercase							:   Union[str, Any] 							=   VQModel(**self.dummy_movq_kwargs      )
											return model


				def 						__a					(					self      ):
											_lowercase							:   List[str] 							=   self.dummy_unet
											_lowercase							:   Optional[int] 							=   self.dummy_movq

											_lowercase							:   Tuple 							=   DDIMScheduler(
											    num_train_timesteps=1_0_0_0 ,						beta_schedule='linear' ,						beta_start=0.0_00_85 ,						beta_end=0.0_12 ,						clip_sample=_lowerCAmelCase ,						set_alpha_to_one=_lowerCAmelCase ,						steps_offset=1 ,						prediction_type='epsilon' ,						thresholding=_lowerCAmelCase ,						)

											_lowercase							:   str 							=   {
											    'unet': unet,
											    'scheduler': scheduler,
											    'movq': movq,
											}
											return components


				def 						__a					(					self ,						_lowerCAmelCase ,						_lowerCAmelCase=0      ):
											_lowercase							:   Optional[int] 							=   floats_tensor((1, self.text_embedder_hidden_size) ,						rng=random.Random(_lowerCAmelCase      )      ).to(_lowerCAmelCase      )
											_lowercase							:   Tuple 							=   floats_tensor((1, self.text_embedder_hidden_size) ,						rng=random.Random(seed + 1      )      ).to(
											    _lowerCAmelCase      )
											if str(_lowerCAmelCase      ).startswith('mps'      ):
																		_lowercase							:   Any 							=   torch.manual_seed(_lowerCAmelCase      )
											else:
																		_lowercase							:   Tuple 							=   torch.Generator(device=_lowerCAmelCase      ).manual_seed(_lowerCAmelCase      )
											_lowercase							:   Optional[int] 							=   {
											    'image_embeds': image_embeds,
											    'negative_image_embeds': negative_image_embeds,
											    'generator': generator,
											    'height': 6_4,
											    'width': 6_4,
											    'guidance_scale': 4.0,
											    'num_inference_steps': 2,
											    'output_type': 'np',
											}
											return inputs


				def 						__a					(					self      ):
											_lowercase							:   Union[str, Any] 							=   'cpu'

											_lowercase							:   Any 							=   self.get_dummy_components()

											_lowercase							:   Any 							=   self.pipeline_class(**_lowerCAmelCase      )
											_lowercase							:   str 							=   pipe.to(_lowerCAmelCase      )

											pipe.set_progress_bar_config(disable=_lowerCAmelCase      )

											_lowercase							:   Tuple 							=   pipe(**self.get_dummy_inputs(_lowerCAmelCase      )      )
											_lowercase							:   Dict 							=   output.images

											_lowercase							:   str 							=   pipe(
											    **self.get_dummy_inputs(_lowerCAmelCase      ) ,						return_dict=_lowerCAmelCase ,						)[0]

											_lowercase							:   str 							=   image[0, -3:, -3:, -1]
											_lowercase							:   List[Any] 							=   image_from_tuple[0, -3:, -3:, -1]

											assert image.shape == (1, 6_4, 6_4, 3)

											_lowercase							:   Any 							=   np.array(
											    [0.6_23_79_76, 1.0, 0.36_44_13_32, 1.0, 0.70_63_96_34, 0.29_87_71_86, 0.85_65_21_25, 0.5_21_68_43, 0.54_45_40_46]      )

											assert (
											    np.abs(image_slice.flatten() - expected_slice      ).max() < 1E-2
											), F""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""

											assert (
											    np.abs(image_from_tuple_slice.flatten() - expected_slice      ).max() < 1E-2
											), F""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""




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


				def 						__a					(					self      ):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()


				def 						__a					(					self      ):
											_lowercase							:   List[str] 							=   load_numpy(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
											    '/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy'      )

											_lowercase							:   Optional[Any] 							=   KandinskyVaaPriorPipeline.from_pretrained(
											    'kandinsky-community/kandinsky-2-2-prior' ,						torch_dtype=torch.floataa      )
											pipe_prior.to(_lowerCAmelCase      )

											_lowercase							:   int 							=   KandinskyVaaPipeline.from_pretrained(
											    'kandinsky-community/kandinsky-2-2-decoder' ,						torch_dtype=torch.floataa      )
											_lowercase							:   List[Any] 							=   pipeline.to(_lowerCAmelCase      )
											pipeline.set_progress_bar_config(disable=_lowerCAmelCase      )

											_lowercase							:   List[Any] 							=   'red cat, 4k photo'

											_lowercase							:   List[str] 							=   torch.Generator(device='cuda'      ).manual_seed(0      )
											_lowercase							,							_lowercase							:   Union[str, Any] 							=   pipe_prior(
											    _lowerCAmelCase ,						generator=_lowerCAmelCase ,						num_inference_steps=5 ,						negative_prompt='' ,						).to_tuple()

											_lowercase							:   Any 							=   torch.Generator(device='cuda'      ).manual_seed(0      )
											_lowercase							:   Union[str, Any] 							=   pipeline(
											    image_embeds=_lowerCAmelCase ,						negative_image_embeds=_lowerCAmelCase ,						generator=_lowerCAmelCase ,						num_inference_steps=1_0_0 ,						output_type='np' ,						)

											_lowercase							:   Optional[int] 							=   output.images[0]

											assert image.shape == (5_1_2, 5_1_2, 3)

											assert_mean_pixel_difference(_lowerCAmelCase ,						_lowerCAmelCase      )









import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
			from transformers.pipelines.conversational import Conversation

_lowerCamelCase    :		Any		      =    logging.get_logger(__name__)


_lowerCamelCase    :		Union[str, Any]		      =    {
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

_lowerCamelCase    :		Optional[Any]		      =    {
    """vocab_file""": {"""facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json"""},
    """merges_file""": {"""facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt"""},
    """tokenizer_config_file""": {
        """facebook/blenderbot-3B""": """https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json"""
    },
}

_lowerCamelCase    :		Optional[int]		      =    {"""facebook/blenderbot-3B""": 128}

@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def        __a			(   )     ->     int:
					SCREAMING_SNAKE_CASE							:     Dict						=				(
					    list(range(ord('!'   )      ,	ord('~'   ) + 1   )   ) + list(range(ord('¡'   )      ,	ord('¬'   ) + 1   )   ) + list(range(ord('®'   )      ,	ord('ÿ'   ) + 1   )   )
					)
					SCREAMING_SNAKE_CASE							:     Optional[int]						=				bs[:]
					SCREAMING_SNAKE_CASE							:     int						=				0
					for b in range(2**8   ):
										if b not in bs:
															bs.append(__lowerCAmelCase   )
															cs.append(2**8 + n   )
															n += 1
					SCREAMING_SNAKE_CASE							:     Any						=				[chr(__lowerCAmelCase   ) for n in cs]
					return dict(zip(__lowerCAmelCase      ,	__lowerCAmelCase   )   )

def        __a			(   __lowerCAmelCase   )     ->     Optional[Any]:
					SCREAMING_SNAKE_CASE							:     Optional[Any]						=				set()
					SCREAMING_SNAKE_CASE							:     List[str]						=				word[0]
					for char in word[1:]:
										pairs.add((prev_char, char)   )
										SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				char
					return pairs



class 						lowercase    (							SCREAMING_SNAKE_CASE_):
						'''simple docstring'''


						UpperCAmelCase			:		Optional[int]		=		VOCAB_FILES_NAMES
						UpperCAmelCase			:		Any		=		PRETRAINED_VOCAB_FILES_MAP
						UpperCAmelCase			:		str		=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						UpperCAmelCase			:		Dict		=		['input_ids', 'attention_mask']







						def __init__(  self    :				Dict       ,	snake_case    :				Optional[int]       ,	snake_case    :				Tuple       ,	snake_case    :				Dict="replace"       ,	snake_case    :				Optional[Any]="<s>"       ,	snake_case    :				Dict="</s>"       ,	snake_case    :				str="</s>"       ,	snake_case    :				Tuple="<s>"       ,	snake_case    :				List[Any]="<unk>"       ,	snake_case    :				Dict="<pad>"       ,	snake_case    :				int="<mask>"       ,	snake_case    :				Any=False       ,	**snake_case    :				str       ,	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     int						=				AddedToken(snake_case       ,	lstrip=snake_case       ,	rstrip=snake_case	) if isinstance(snake_case       ,	snake_case	) else bos_token
											SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				AddedToken(snake_case       ,	lstrip=snake_case       ,	rstrip=snake_case	) if isinstance(snake_case       ,	snake_case	) else eos_token
											SCREAMING_SNAKE_CASE							:     int						=				AddedToken(snake_case       ,	lstrip=snake_case       ,	rstrip=snake_case	) if isinstance(snake_case       ,	snake_case	) else sep_token
											SCREAMING_SNAKE_CASE							:     Dict						=				AddedToken(snake_case       ,	lstrip=snake_case       ,	rstrip=snake_case	) if isinstance(snake_case       ,	snake_case	) else cls_token
											SCREAMING_SNAKE_CASE							:     Optional[Any]						=				AddedToken(snake_case       ,	lstrip=snake_case       ,	rstrip=snake_case	) if isinstance(snake_case       ,	snake_case	) else unk_token
											SCREAMING_SNAKE_CASE							:     List[Any]						=				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
											SCREAMING_SNAKE_CASE							:     str						=				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:
																SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				json.load(snake_case	)
											SCREAMING_SNAKE_CASE							:     int						=				{v: k for k, v in self.encoder.items()}
											SCREAMING_SNAKE_CASE							:     Optional[int]						=				errors  # how to handle errors in decoding
											SCREAMING_SNAKE_CASE							:     List[Any]						=				bytes_to_unicode()
											SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				{v: k for k, v in self.byte_encoder.items()}
											with open(snake_case       ,	encoding='utf-8'	) as merges_handle:
																SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				merges_handle.read().split('\n'	)[1:-1]
											SCREAMING_SNAKE_CASE							:     str						=				[tuple(merge.split()	) for merge in bpe_merges]
											SCREAMING_SNAKE_CASE							:     List[str]						=				dict(zip(snake_case       ,	range(len(snake_case	)	)	)	)
											SCREAMING_SNAKE_CASE							:     int						=				{}
											SCREAMING_SNAKE_CASE							:     str						=				add_prefix_space

											# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
											SCREAMING_SNAKE_CASE							:     List[str]						=				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.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
						def       lowerCamelCase_				(  self    :				str	):



											'''simple docstring'''


											return len(self.encoder	)







						def       lowerCamelCase_				(  self    :				Optional[Any]	):



											'''simple docstring'''


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







						def       lowerCamelCase_				(  self    :				List[str]       ,	snake_case    :				int	):



											'''simple docstring'''


											if token in self.cache:
																return self.cache[token]
											SCREAMING_SNAKE_CASE							:     int						=				tuple(snake_case	)
											SCREAMING_SNAKE_CASE							:     Dict						=				get_pairs(snake_case	)

											if not pairs:
																return token

											while True:
																SCREAMING_SNAKE_CASE							:     List[str]						=				min(snake_case       ,	key=lambda snake_case	: self.bpe_ranks.get(snake_case       ,	float('inf'	)	)	)
																if bigram not in self.bpe_ranks:
																					break
																SCREAMING_SNAKE_CASE					,  SCREAMING_SNAKE_CASE							:     Dict						=				bigram
																SCREAMING_SNAKE_CASE							:     Optional[int]						=				[]
																SCREAMING_SNAKE_CASE							:     Tuple						=				0
																while i < len(snake_case	):
																					try:
																										SCREAMING_SNAKE_CASE							:     List[str]						=				word.index(snake_case       ,	snake_case	)
																					except ValueError:
																										new_word.extend(word[i:]	)
																										break
																					else:
																										new_word.extend(word[i:j]	)
																										SCREAMING_SNAKE_CASE							:     Dict						=				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
																SCREAMING_SNAKE_CASE							:     Any						=				tuple(snake_case	)
																SCREAMING_SNAKE_CASE							:     Tuple						=				new_word
																if len(snake_case	) == 1:
																					break
																else:
																					SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				get_pairs(snake_case	)
											SCREAMING_SNAKE_CASE							:     List[Any]						=				' '.join(snake_case	)
											SCREAMING_SNAKE_CASE							:     Dict						=				word
											return word







						def       lowerCamelCase_				(  self    :				Any       ,	snake_case    :				Tuple	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Tuple						=				[]
											for token in re.findall(self.pat       ,	snake_case	):
																SCREAMING_SNAKE_CASE							:     Any						=				''.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       lowerCamelCase_				(  self    :				Any       ,	snake_case    :				int	):



											'''simple docstring'''


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







						def       lowerCamelCase_				(  self    :				List[Any]       ,	snake_case    :				Dict	):



											'''simple docstring'''


											return self.decoder.get(snake_case	)







						def       lowerCamelCase_				(  self    :				List[Any]       ,	snake_case    :				int	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     List[Any]						=				''.join(snake_case	)
											SCREAMING_SNAKE_CASE							:     Optional[Any]						=				bytearray([self.byte_decoder[c] for c in text]	).decode('utf-8'       ,	errors=self.errors	)
											return text







						def       lowerCamelCase_				(  self    :				Union[str, Any]       ,	snake_case    :				str       ,	snake_case    :				Optional[str] = None	):



											'''simple docstring'''


											if not os.path.isdir(snake_case	):
																logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''	)
																return
											SCREAMING_SNAKE_CASE							:     List[Any]						=				os.path.join(
											    snake_case       ,	(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']	)
											SCREAMING_SNAKE_CASE							:     List[str]						=				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'	)

											SCREAMING_SNAKE_CASE							:     str						=				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!'	)
																										SCREAMING_SNAKE_CASE							:     List[str]						=				token_index
																					writer.write(' '.join(snake_case	) + '\n'	)
																					index += 1

											return vocab_file, merge_file







						def       lowerCamelCase_				(  self    :				Dict       ,	snake_case    :				List[int]       ,	snake_case    :				Optional[List[int]] = None       ,	snake_case    :				bool = False	):



											'''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       lowerCamelCase_				(  self    :				int       ,	snake_case    :				List[int]       ,	snake_case    :				Optional[List[int]] = None	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Dict						=				[self.sep_token_id]
											SCREAMING_SNAKE_CASE							:     int						=				[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       lowerCamelCase_				(  self    :				Optional[Any]       ,	snake_case    :				List[Any]       ,	snake_case    :				Optional[Any]=False       ,	**snake_case    :				Optional[int]	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Optional[int]						=				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()):
																SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				' ' + text
											return (text, kwargs)







						def       lowerCamelCase_				(  self    :				List[str]       ,	snake_case    :				List[int]       ,	snake_case    :				Optional[List[int]] = None	):



											'''simple docstring'''


											return token_ids_a + [self.eos_token_id]







						def       lowerCamelCase_				(  self    :				List[str]       ,	snake_case    :				"Conversation"	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Tuple						=				[]
											for is_user, text in conversation.iter_texts():
																if is_user:
																					# We need to space prefix as it's being done within blenderbot
																					inputs.append(' ' + text	)
																else:
																					# Generated responses should contain them already.
																					inputs.append(snake_case	)

											SCREAMING_SNAKE_CASE							:     Optional[Any]						=				'  '.join(snake_case	)
											SCREAMING_SNAKE_CASE							:     str						=				self.encode(snake_case	)
											if len(snake_case	) > self.model_max_length:
																SCREAMING_SNAKE_CASE							:     Optional[Any]						=				input_ids[-self.model_max_length :]
																logger.warning(f'''Trimmed input from conversation as it was longer than {self.model_max_length} tokens.'''	)
											return input_ids