






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

__UpperCamelCase					:		Optional[Any]	  =						logging.get_logger(__name__)


__UpperCamelCase					:		List[str]	  =						{
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
    "tokenizer_config_file": "tokenizer_config.json",
}

__UpperCamelCase					:		List[str]	  =						{
    "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"
    },
}

__UpperCamelCase					:		Any	  =						{"facebook/blenderbot-3B": 128}
@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def      a_ (      )			->			Any:







				"""simple docstring"""
				snake_case__									=     (
				    list(range(ord('!'  )      ,   ord('~'  ) + 1  )  ) + list(range(ord('¡'  )      ,   ord('¬'  ) + 1  )  ) + list(range(ord('®'  )      ,   ord('ÿ'  ) + 1  )  )
				)
				snake_case__									=     bs[:]
				snake_case__									=     0
				for b in range(2**8  ):
								if b not in bs:
												bs.append(__A  )
												cs.append(2**8 + n  )
												n += 1
				snake_case__									=     [chr(__A  ) for n in cs]
				return dict(zip(__A      ,   __A  )  )
def      a_ (      _A  )			->			Dict:







				"""simple docstring"""
				snake_case__									=     set()
				snake_case__									=     word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)  )
								snake_case__									=     char
				return pairs


class   __SCREAMING_SNAKE_CASE(	__lowerCAmelCase	):
							_UpperCAmelCase     =       VOCAB_FILES_NAMES
							_UpperCAmelCase     =       PRETRAINED_VOCAB_FILES_MAP
							_UpperCAmelCase     =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							_UpperCAmelCase     =       ["input_ids", "attention_mask"]
							def __init__(     self:      Optional[Any]       ,	UpperCamelCase:      Tuple       ,	UpperCamelCase:      Dict       ,	UpperCamelCase:      int="replace"       ,	UpperCamelCase:      Tuple="<s>"       ,	UpperCamelCase:      Dict="</s>"       ,	UpperCamelCase:      List[Any]="</s>"       ,	UpperCamelCase:      Optional[Any]="<s>"       ,	UpperCamelCase:      List[Any]="<unk>"       ,	UpperCamelCase:      Any="<pad>"       ,	UpperCamelCase:      Optional[Any]="<mask>"       ,	UpperCamelCase:      Union[str, Any]=False       ,	**UpperCamelCase:      str       ,	)		->		List[Any]:
											snake_case__									=     AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase						) if isinstance(_UpperCamelCase       ,	_UpperCamelCase						) else bos_token
											snake_case__									=     AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase						) if isinstance(_UpperCamelCase       ,	_UpperCamelCase						) else eos_token
											snake_case__									=     AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase						) if isinstance(_UpperCamelCase       ,	_UpperCamelCase						) else sep_token
											snake_case__									=     AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase						) if isinstance(_UpperCamelCase       ,	_UpperCamelCase						) else cls_token
											snake_case__									=     AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase						) if isinstance(_UpperCamelCase       ,	_UpperCamelCase						) else unk_token
											snake_case__									=     AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase						) if isinstance(_UpperCamelCase       ,	_UpperCamelCase						) else pad_token

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

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

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

											# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
											snake_case__									=     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						)		->		str:
											return len(self.encoder						)
							def lowerCAmelCase_ (     self:      List[str]						)		->		Optional[Any]:
											return dict(self.encoder       ,	**self.added_tokens_encoder						)
							def lowerCAmelCase_ (     self:      Union[str, Any]       ,	UpperCamelCase:      Tuple						)		->		Optional[Any]:


											if token in self.cache:
															return self.cache[token]
											snake_case__									=     tuple(_UpperCamelCase						)
											snake_case__									=     get_pairs(_UpperCamelCase						)

											if not pairs:
															return token

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

																			if word[i] == first and i < len(_UpperCamelCase						) - 1 and word[i + 1] == second:
																							new_word.append(first + second						)
																							i += 2
																			else:
																							new_word.append(word[i]						)
																							i += 1
															snake_case__									=     tuple(_UpperCamelCase						)
															snake_case__									=     new_word
															if len(_UpperCamelCase						) == 1:
																			break
															else:
																			snake_case__									=     get_pairs(_UpperCamelCase						)
											snake_case__									=     """ """.join(_UpperCamelCase						)
											snake_case__									=     word
											return word
							def lowerCAmelCase_ (     self:      Union[str, Any]       ,	UpperCamelCase:      Optional[Any]						)		->		str:
											snake_case__									=     []
											for token in re.findall(self.pat       ,	_UpperCamelCase						):
															snake_case__									=     """""".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(_UpperCamelCase						).split(' '						)						)
											return bpe_tokens
							def lowerCAmelCase_ (     self:      Tuple       ,	UpperCamelCase:      Optional[int]						)		->		Optional[int]:
											return self.encoder.get(_UpperCamelCase       ,	self.encoder.get(self.unk_token						)						)
							def lowerCAmelCase_ (     self:      int       ,	UpperCamelCase:      int						)		->		Union[str, Any]:
											return self.decoder.get(_UpperCamelCase						)
							def lowerCAmelCase_ (     self:      Union[str, Any]       ,	UpperCamelCase:      Any						)		->		List[Any]:
											snake_case__									=     """""".join(_UpperCamelCase						)
											snake_case__									=     bytearray([self.byte_decoder[c] for c in text]						).decode('utf-8'       ,	errors=self.errors						)
											return text
							def lowerCAmelCase_ (     self:      Union[str, Any]       ,	UpperCamelCase:      str       ,	UpperCamelCase:      Optional[str] = None						)		->		Tuple[str]:


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

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

											snake_case__									=     0
											with open(_UpperCamelCase       ,	'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 UpperCamelCase						: 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!'						)
																							snake_case__									=     token_index
																			writer.write(' '.join(_UpperCamelCase						) + '\n'						)
																			index += 1

											return vocab_file, merge_file
							def lowerCAmelCase_ (     self:      List[str]       ,	UpperCamelCase:      List[int]       ,	UpperCamelCase:      Optional[List[int]] = None       ,	UpperCamelCase:      bool = False						)		->		List[int]:


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

											if token_ids_a is None:
															return [1] + ([0] * len(_UpperCamelCase						)) + [1]
											return [1] + ([0] * len(_UpperCamelCase						)) + [1, 1] + ([0] * len(_UpperCamelCase						)) + [1]
							def lowerCAmelCase_ (     self:      List[Any]       ,	UpperCamelCase:      List[int]       ,	UpperCamelCase:      Optional[List[int]] = None						)		->		List[int]:
											snake_case__									=     [self.sep_token_id]
											snake_case__									=     [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:      int       ,	UpperCamelCase:      str       ,	UpperCamelCase:      Optional[Any]=False       ,	**UpperCamelCase:      Optional[Any]						)		->		Optional[Any]:
											snake_case__									=     kwargs.pop('add_prefix_space'       ,	self.add_prefix_space						)
											if (is_split_into_words or add_prefix_space) and (len(_UpperCamelCase						) > 0 and not text[0].isspace()):
															snake_case__									=     """ """ + text
											return (text, kwargs)
							def lowerCAmelCase_ (     self:      Dict       ,	UpperCamelCase:      List[int]       ,	UpperCamelCase:      Optional[List[int]] = None						)		->		Any:
											return token_ids_a + [self.eos_token_id]


							def lowerCAmelCase_ (     self:      Optional[Any]       ,	UpperCamelCase:      "Conversation"						)		->		List[int]:
											snake_case__									=     []
											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(_UpperCamelCase						)

											snake_case__									=     """  """.join(_UpperCamelCase						)
											snake_case__									=     self.encode(_UpperCamelCase						)
											if len(_UpperCamelCase						) > self.model_max_length:
															snake_case__									=     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











import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    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   __SCREAMING_SNAKE_CASE(	a_    ,  a_    ,  unittest.TestCase	):
							_UpperCAmelCase     =       StableDiffusionSAGPipeline
							_UpperCAmelCase     =       TEXT_TO_IMAGE_PARAMS
							_UpperCAmelCase     =       TEXT_TO_IMAGE_BATCH_PARAMS
							_UpperCAmelCase     =       TEXT_TO_IMAGE_IMAGE_PARAMS
							_UpperCAmelCase     =       TEXT_TO_IMAGE_IMAGE_PARAMS
							_UpperCAmelCase     =       False
							def lowerCAmelCase_ (     self:      int						)		->		Any:
											torch.manual_seed(0						)
											snake_case__									=     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       ,	)
											snake_case__									=     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						)
											snake_case__									=     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       ,	)
											torch.manual_seed(0						)
											snake_case__									=     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=10_00       ,	)
											snake_case__									=     CLIPTextModel(UpperCamelCase						)
											snake_case__									=     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'						)

											snake_case__									=     {
											    'unet': unet,
											    'scheduler': scheduler,
											    'vae': vae,
											    'text_encoder': text_encoder,
											    'tokenizer': tokenizer,
											    'safety_checker': None,
											    'feature_extractor': None,
											}
											return components
							def lowerCAmelCase_ (     self:      Optional[int]       ,	UpperCamelCase:      Dict       ,	UpperCamelCase:      Optional[int]=0						)		->		Union[str, Any]:
											if str(UpperCamelCase						).startswith('mps'						):
															snake_case__									=     torch.manual_seed(UpperCamelCase						)
											else:
															snake_case__									=     torch.Generator(device=UpperCamelCase						).manual_seed(UpperCamelCase						)
											snake_case__									=     {
											    'prompt': '.',
											    'generator': generator,
											    'num_inference_steps': 2,
											    'guidance_scale': 1.0,
											    'sag_scale': 1.0,
											    'output_type': 'numpy',
											}
											return inputs


							def lowerCAmelCase_ (     self:      Union[str, Any]						)		->		Dict:
											super().test_inference_batch_single_identical(expected_max_diff=3e-3						)


@slow
@require_torch_gpu
class   __SCREAMING_SNAKE_CASE(	unittest.TestCase	):
							def lowerCAmelCase_ (     self:      Optional[int]						)		->		Optional[int]:
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()
							def lowerCAmelCase_ (     self:      Dict						)		->		Tuple:
											snake_case__									=     StableDiffusionSAGPipeline.from_pretrained('CompVis/stable-diffusion-v1-4'						)
											snake_case__									=     sag_pipe.to(UpperCamelCase						)
											sag_pipe.set_progress_bar_config(disable=UpperCamelCase						)

											snake_case__									=     '.'
											snake_case__									=     torch.manual_seed(0						)
											snake_case__									=     sag_pipe(
											    [prompt]       ,	generator=UpperCamelCase       ,	guidance_scale=7.5       ,	sag_scale=1.0       ,	num_inference_steps=20       ,	output_type='np'						)

											snake_case__									=     output.images

											snake_case__									=     image[0, -3:, -3:, -1]

											assert image.shape == (1, 5_12, 5_12, 3)
											snake_case__									=     np.array([0.1_568, 0.1_738, 0.1_695, 0.1_693, 0.1_507, 0.1_705, 0.1_547, 0.1_751, 0.1_949]						)

											assert np.abs(image_slice.flatten() - expected_slice						).max() < 5e-2
							def lowerCAmelCase_ (     self:      Union[str, Any]						)		->		List[Any]:
											snake_case__									=     StableDiffusionSAGPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base'						)
											snake_case__									=     sag_pipe.to(UpperCamelCase						)
											sag_pipe.set_progress_bar_config(disable=UpperCamelCase						)

											snake_case__									=     '.'
											snake_case__									=     torch.manual_seed(0						)
											snake_case__									=     sag_pipe(
											    [prompt]       ,	generator=UpperCamelCase       ,	guidance_scale=7.5       ,	sag_scale=1.0       ,	num_inference_steps=20       ,	output_type='np'						)

											snake_case__									=     output.images

											snake_case__									=     image[0, -3:, -3:, -1]

											assert image.shape == (1, 5_12, 5_12, 3)
											snake_case__									=     np.array([0.3_459, 0.2_876, 0.2_537, 0.3_002, 0.2_671, 0.2_160, 0.3_026, 0.2_262, 0.2_371]						)

											assert np.abs(image_slice.flatten() - expected_slice						).max() < 5e-2


							def lowerCAmelCase_ (     self:      Union[str, Any]						)		->		Optional[Any]:
											snake_case__									=     StableDiffusionSAGPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base'						)
											snake_case__									=     sag_pipe.to(UpperCamelCase						)
											sag_pipe.set_progress_bar_config(disable=UpperCamelCase						)

											snake_case__									=     '.'
											snake_case__									=     torch.manual_seed(0						)
											snake_case__									=     sag_pipe(
											    [prompt]       ,	width=7_68       ,	height=5_12       ,	generator=UpperCamelCase       ,	guidance_scale=7.5       ,	sag_scale=1.0       ,	num_inference_steps=20       ,	output_type='np'       ,	)

											snake_case__									=     output.images

											assert image.shape == (1, 5_12, 7_68, 3)




