



"""simple docstring"""

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

import regex as re

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


lowercase__			   =  logging.get_logger(__name__)


lowercase__			   =  {"vocab_file": "vocab.json", "merges_file": "merges.txt"}

# See all BART models at https://huggingface.co/models?filter=bart
lowercase__			   =  {
    "vocab_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/vocab.json",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/vocab.json",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json",
    },
    "merges_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/merges.txt",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/merges.txt",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt",
    },
}

lowercase__			   =  {
    "facebook/bart-base": 1024,
    "facebook/bart-large": 1024,
    "facebook/bart-large-mnli": 1024,
    "facebook/bart-large-cnn": 1024,
    "facebook/bart-large-xsum": 1024,
    "yjernite/bart_eli5": 1024,
}





@lru_cache()
def       __magic_name__    (   ):
							__a		:		Optional[int]									=       (
							    list(range(ord("""!"""					)  ,       ord("""~"""					) + 1					)					) + list(range(ord("""¡"""					)  ,       ord("""¬"""					) + 1					)					) + list(range(ord("""®"""					)  ,       ord("""ÿ"""					) + 1					)					)
							)
							__a		:		List[str]									=       bs[:]
							__a		:		Optional[int]									=       0
							for b in range(2**8					):
														if b not in bs:
																					bs.append(_lowerCamelCase					)
																					cs.append(2**8 + n					)
																					n += 1
							__a		:		Tuple									=       [chr(_lowerCamelCase					) for n in cs]
							return dict(zip(_lowerCamelCase  ,       _lowerCamelCase					)					)





def       __magic_name__    (   _lowerCamelCase				: Dict					):
							__a		:		Tuple									=       set()
							__a		:		Union[str, Any]									=       word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)					)
														__a		:		Union[str, Any]									=       char
							return pairs




class SCREAMING_SNAKE_CASE__					(		__snake_case      ):
						_lowerCAmelCase							=					VOCAB_FILES_NAMES
						_lowerCAmelCase							=					PRETRAINED_VOCAB_FILES_MAP
						_lowerCAmelCase							=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						_lowerCAmelCase							=					["input_ids", "attention_mask"]



						def __init__(self						,     _lowercase						,     _lowercase						,     _lowercase="replace"						,     _lowercase="<s>"						,     _lowercase="</s>"						,     _lowercase="</s>"						,     _lowercase="<s>"						,     _lowercase="<unk>"						,     _lowercase="<pad>"						,     _lowercase="<mask>"						,     _lowercase=False						,     **_lowercase						,     ):





													'''simple docstring'''

													__a		:		Optional[Any]									=       AddedToken(_lowercase						,     lstrip=_lowercase						,     rstrip=_lowercase							) if isinstance(_lowercase						,     _lowercase							) else bos_token
													__a		:		str									=       AddedToken(_lowercase						,     lstrip=_lowercase						,     rstrip=_lowercase							) if isinstance(_lowercase						,     _lowercase							) else eos_token
													__a		:		str									=       AddedToken(_lowercase						,     lstrip=_lowercase						,     rstrip=_lowercase							) if isinstance(_lowercase						,     _lowercase							) else sep_token
													__a		:		str									=       AddedToken(_lowercase						,     lstrip=_lowercase						,     rstrip=_lowercase							) if isinstance(_lowercase						,     _lowercase							) else cls_token
													__a		:		int									=       AddedToken(_lowercase						,     lstrip=_lowercase						,     rstrip=_lowercase							) if isinstance(_lowercase						,     _lowercase							) else unk_token
													__a		:		Any									=       AddedToken(_lowercase						,     lstrip=_lowercase						,     rstrip=_lowercase							) if isinstance(_lowercase						,     _lowercase							) else pad_token

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

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

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

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



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													return len(self.encoder							)



						def     lowerCAmelCase__(self							):





													'''simple docstring'''

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



						def     lowerCAmelCase__(self						,     _lowercase							):





													'''simple docstring'''

													if token in self.cache:
																				return self.cache[token]
													__a		:		int									=       tuple(_lowercase							)
													__a		:		str									=       get_pairs(_lowercase							)

													if not pairs:
																				return token

													while True:
																				__a		:		Any									=       min(_lowercase						,     key=lambda _lowercase							: self.bpe_ranks.get(_lowercase						,     float("""inf"""							)							)							)
																				if bigram not in self.bpe_ranks:
																											break
																				__a		:		Tuple									=       bigram
																				__a		:		int									=       []
																				__a		:		Union[str, Any]									=       0
																				while i < len(_lowercase							):
																											try:
																																		__a		:		Union[str, Any]									=       word.index(_lowercase						,     _lowercase							)
																											except ValueError:
																																		new_word.extend(word[i:]							)
																																		break
																											else:
																																		new_word.extend(word[i:j]							)
																																		__a		:		List[str]									=       j

																											if word[i] == first and i < len(_lowercase							) - 1 and word[i + 1] == second:
																																		new_word.append(first + second							)
																																		i += 2
																											else:
																																		new_word.append(word[i]							)
																																		i += 1
																				__a		:		Any									=       tuple(_lowercase							)
																				__a		:		List[Any]									=       new_word
																				if len(_lowercase							) == 1:
																											break
																				else:
																											__a		:		Optional[int]									=       get_pairs(_lowercase							)
													__a		:		List[Any]									=       """ """.join(_lowercase							)
													__a		:		Union[str, Any]									=       word
													return word



						def     lowerCAmelCase__(self						,     _lowercase							):





													'''simple docstring'''

													__a		:		str									=       []
													for token in re.findall(self.pat						,     _lowercase							):
																				__a		:		Union[str, 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(_lowercase							).split(""" """							)							)
													return bpe_tokens



						def     lowerCAmelCase__(self						,     _lowercase							):





													'''simple docstring'''

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



						def     lowerCAmelCase__(self						,     _lowercase							):





													'''simple docstring'''

													return self.decoder.get(_lowercase							)



						def     lowerCAmelCase__(self						,     _lowercase							):





													'''simple docstring'''

													__a		:		Optional[int]									=       """""".join(_lowercase							)
													__a		:		Any									=       bytearray([self.byte_decoder[c] for c in text]							).decode("""utf-8"""						,     errors=self.errors							)
													return text



						def     lowerCAmelCase__(self						,     _lowercase						,     _lowercase = None							):





													'''simple docstring'''

													if not os.path.isdir(_lowercase							):
																				logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''							)
																				return
													__a		:		Optional[Any]									=       os.path.join(
													    _lowercase						,     (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]							)
													__a		:		Union[str, Any]									=       os.path.join(
													    _lowercase						,     (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]							)

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

													__a		:		List[Any]									=       0
													with open(_lowercase						,     """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 _lowercase							: 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!"""							)
																																		__a		:		Tuple									=       token_index
																											writer.write(""" """.join(_lowercase							) + """\n"""							)
																											index += 1

													return vocab_file, merge_file



						def     lowerCAmelCase__(self						,     _lowercase						,     _lowercase = None							):





													'''simple docstring'''

													if token_ids_a is None:
																				return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
													__a		:		Optional[Any]									=       [self.cls_token_id]
													__a		:		Tuple									=       [self.sep_token_id]
													return cls + token_ids_a + sep + sep + token_ids_a + sep



						def     lowerCAmelCase__(self						,     _lowercase						,     _lowercase = None						,     _lowercase = False							):





													'''simple docstring'''

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

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



						def     lowerCAmelCase__(self						,     _lowercase						,     _lowercase = None							):





													'''simple docstring'''

													__a		:		Union[str, Any]									=       [self.sep_token_id]
													__a		:		Dict									=       [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						,     _lowercase						,     _lowercase=False						,     **_lowercase							):





													'''simple docstring'''

													__a		:		Any									=       kwargs.pop("""add_prefix_space"""						,     self.add_prefix_space							)
													if (is_split_into_words or add_prefix_space) and (len(_lowercase							) > 0 and not text[0].isspace()):
																				__a		:		Optional[int]									=       """ """ + text
													return (text, kwargs)






"""simple docstring"""

import unittest

import numpy as np
import torch
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class SCREAMING_SNAKE_CASE__					(		__snake_case     ,					unittest.TestCase      ):
						_lowerCAmelCase							=					KandinskyVaaPriorPipeline
						_lowerCAmelCase							=					["prompt"]
						_lowerCAmelCase							=					["prompt", "negative_prompt"]
						_lowerCAmelCase							=					[
						    "num_images_per_prompt",
						    "generator",
						    "num_inference_steps",
						    "latents",
						    "negative_prompt",
						    "guidance_scale",
						    "output_type",
						    "return_dict",
						]
						_lowerCAmelCase							=					False



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													return 32



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													return 32



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													return self.time_input_dim



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													return self.time_input_dim * 4



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													return 100



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													__a		:		List[str]									=       CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)
													return tokenizer



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													torch.manual_seed(0							)
													__a		:		str									=       CLIPTextConfig(
													    bos_token_id=0						,     eos_token_id=2						,     hidden_size=self.text_embedder_hidden_size						,     projection_dim=self.text_embedder_hidden_size						,     intermediate_size=37						,     layer_norm_eps=1e-05						,     num_attention_heads=4						,     num_hidden_layers=5						,     pad_token_id=1						,     vocab_size=1000						,     )
													return CLIPTextModelWithProjection(_lowercase							)



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													torch.manual_seed(0							)

													__a		:		Dict									=       {
													    """num_attention_heads""": 2,
													    """attention_head_dim""": 12,
													    """embedding_dim""": self.text_embedder_hidden_size,
													    """num_layers""": 1,
													}

													__a		:		Tuple									=       PriorTransformer(**_lowercase							)
													# clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
													__a		:		int									=       nn.Parameter(torch.ones(model.clip_std.shape							)							)
													return model



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													torch.manual_seed(0							)
													__a		:		List[str]									=       CLIPVisionConfig(
													    hidden_size=self.text_embedder_hidden_size						,     image_size=224						,     projection_dim=self.text_embedder_hidden_size						,     intermediate_size=37						,     num_attention_heads=4						,     num_channels=3						,     num_hidden_layers=5						,     patch_size=14						,     )

													__a		:		Optional[Any]									=       CLIPVisionModelWithProjection(_lowercase							)
													return model



						@property
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													__a		:		Optional[Any]									=       CLIPImageProcessor(
													    crop_size=224						,     do_center_crop=_lowercase						,     do_normalize=_lowercase						,     do_resize=_lowercase						,     image_mean=[0.4814_5466, 0.457_8275, 0.4082_1073]						,     image_std=[0.2686_2954, 0.2613_0258, 0.2757_7711]						,     resample=3						,     size=224						,     )

													return image_processor



						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													__a		:		Union[str, Any]									=       self.dummy_prior
													__a		:		int									=       self.dummy_image_encoder
													__a		:		Any									=       self.dummy_text_encoder
													__a		:		int									=       self.dummy_tokenizer
													__a		:		Optional[Any]									=       self.dummy_image_processor

													__a		:		List[Any]									=       UnCLIPScheduler(
													    variance_type="""fixed_small_log"""						,     prediction_type="""sample"""						,     num_train_timesteps=1000						,     clip_sample=_lowercase						,     clip_sample_range=10.0						,     )

													__a		:		List[Any]									=       {
													    """prior""": prior,
													    """image_encoder""": image_encoder,
													    """text_encoder""": text_encoder,
													    """tokenizer""": tokenizer,
													    """scheduler""": scheduler,
													    """image_processor""": image_processor,
													}

													return components



						def     lowerCAmelCase__(self						,     _lowercase						,     _lowercase=0							):





													'''simple docstring'''

													if str(_lowercase							).startswith("""mps"""							):
																				__a		:		Dict									=       torch.manual_seed(_lowercase							)
													else:
																				__a		:		Union[str, Any]									=       torch.Generator(device=_lowercase							).manual_seed(_lowercase							)
													__a		:		Union[str, Any]									=       {
													    """prompt""": """horse""",
													    """generator""": generator,
													    """guidance_scale""": 4.0,
													    """num_inference_steps""": 2,
													    """output_type""": """np""",
													}
													return inputs



						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													__a		:		Union[str, Any]									=       """cpu"""

													__a		:		Union[str, Any]									=       self.get_dummy_components()

													__a		:		Dict									=       self.pipeline_class(**_lowercase							)
													__a		:		Tuple									=       pipe.to(_lowercase							)

													pipe.set_progress_bar_config(disable=_lowercase							)

													__a		:		Optional[int]									=       pipe(**self.get_dummy_inputs(_lowercase							)							)
													__a		:		str									=       output.image_embeds

													__a		:		Any									=       pipe(
													    **self.get_dummy_inputs(_lowercase							)						,     return_dict=_lowercase						,     )[0]

													__a		:		List[Any]									=       image[0, -10:]
													__a		:		List[Any]									=       image_from_tuple[0, -10:]

													assert image.shape == (1, 32)

													__a		:		Optional[Any]									=       np.array(
													    [-0.0532, 1.7120, 0.3656, -1.0852, -0.8946, -1.1756, 0.4348, 0.2482, 0.5146, -0.1156]							)

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



						@skip_mps
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													__a		:		Any									=       torch_device == """cpu"""
													__a		:		Any									=       True
													__a		:		Any									=       False

													self._test_inference_batch_single_identical(
													    test_max_difference=_lowercase						,     relax_max_difference=_lowercase						,     test_mean_pixel_difference=_lowercase						,     )





						@skip_mps
						def     lowerCAmelCase__(self							):





													'''simple docstring'''

													__a		:		Optional[int]									=       torch_device == """cpu"""
													__a		:		Union[str, Any]									=       False

													self._test_attention_slicing_forward_pass(
													    test_max_difference=_lowercase						,     test_mean_pixel_difference=_lowercase						,     )

