

"""simple docstring"""




import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, 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      _UpperCAmelCase   (							_lowerCAmelCase		,    unittest.TestCase       ):
	a__		: int					=					KandinskyImgaImgPipeline
	a__		: Optional[int]					=					["prompt", "image_embeds", "negative_image_embeds", "image"]
	a__		: Optional[Any]					=					[
	    "prompt",
	    "negative_prompt",
	    "image_embeds",
	    "negative_image_embeds",
	    "image",
	]
	a__		: Tuple					=					[
	    "generator",
	    "height",
	    "width",
	    "strength",
	    "guidance_scale",
	    "negative_prompt",
	    "num_inference_steps",
	    "return_dict",
	    "guidance_scale",
	    "num_images_per_prompt",
	    "output_type",
	    "return_dict",
	]
	a__		: Optional[Any]					=					False



	@property
	def        a  (  self     :       Tuple						):
		return 32



	@property
	def        a  (  self     :       int						):
		return 32



	@property
	def        a  (  self     :       Optional[Any]						):
		return self.time_input_dim



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



	@property
	def        a  (  self     :       Union[str, Any]						):
		return 1_00



	@property
	def        a  (  self     :       List[Any]						):
		__UpperCAmelCase      		= XLMRobertaTokenizerFast.from_pretrained('''YiYiXu/tiny-random-mclip-base'''						)
		return tokenizer



	@property
	def        a  (  self     :       Union[str, Any]						):
		torch.manual_seed(0						)
		__UpperCAmelCase      		= MCLIPConfig(
		    numDims=self.cross_attention_dim			,    transformerDimensions=self.text_embedder_hidden_size			,    hidden_size=self.text_embedder_hidden_size			,    intermediate_size=37			,    num_attention_heads=4			,    num_hidden_layers=5			,    vocab_size=10_05			,    )

		__UpperCAmelCase      		= MultilingualCLIP(_lowercase						)
		__UpperCAmelCase      		= text_encoder.eval()

		return text_encoder



	@property
	def        a  (  self     :       Any						):
		torch.manual_seed(0						)

		__UpperCAmelCase      		= {
		    '''in_channels''': 4,
		    # Out channels is double in channels because predicts mean and variance
		    '''out_channels''': 8,
		    '''addition_embed_type''': '''text_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''': '''text_image_proj''',
		    '''cross_attention_dim''': self.cross_attention_dim,
		    '''attention_head_dim''': 4,
		    '''resnet_time_scale_shift''': '''scale_shift''',
		    '''class_embed_type''': None,
		}

		__UpperCAmelCase      		= UNetaDConditionModel(**_lowercase						)
		return model



	@property
	def        a  (  self     :       List[str]						):
		return {
		    "block_out_channels": [32, 64],
		    "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": 12,
		    "out_channels": 3,
		    "up_block_types": [
		        "AttnUpDecoderBlock2D",
		        "UpDecoderBlock2D",
		    ],
		    "vq_embed_dim": 4,
		}



	@property
	def        a  (  self     :       Tuple						):
		torch.manual_seed(0						)
		__UpperCAmelCase      		= VQModel(**self.dummy_movq_kwargs						)
		return model



	def        a  (  self     :       Optional[int]						):
		__UpperCAmelCase      		= self.dummy_text_encoder
		__UpperCAmelCase      		= self.dummy_tokenizer
		__UpperCAmelCase      		= self.dummy_unet
		__UpperCAmelCase      		= self.dummy_movq

		__UpperCAmelCase      		= {
		    '''num_train_timesteps''': 10_00,
		    '''beta_schedule''': '''linear''',
		    '''beta_start''': 0.00_085,
		    '''beta_end''': 0.012,
		    '''clip_sample''': False,
		    '''set_alpha_to_one''': False,
		    '''steps_offset''': 0,
		    '''prediction_type''': '''epsilon''',
		    '''thresholding''': False,
		}

		__UpperCAmelCase      		= DDIMScheduler(**_lowercase						)

		__UpperCAmelCase      		= {
		    '''text_encoder''': text_encoder,
		    '''tokenizer''': tokenizer,
		    '''unet''': unet,
		    '''scheduler''': scheduler,
		    '''movq''': movq,
		}

		return components



	def        a  (  self     :       Optional[int]			,    _lowercase     :       List[str]			,    _lowercase     :       List[str]=0						):
		__UpperCAmelCase      		= floats_tensor((1, self.cross_attention_dim)			,    rng=random.Random(_lowercase						)						).to(_lowercase						)
		__UpperCAmelCase      		= floats_tensor((1, self.cross_attention_dim)			,    rng=random.Random(seed + 1						)						).to(_lowercase						)
		# create init_image
		__UpperCAmelCase      		= floats_tensor((1, 3, 64, 64)			,    rng=random.Random(_lowercase						)						).to(_lowercase						)
		__UpperCAmelCase      		= image.cpu().permute(0			,    2			,    3			,    1						)[0]
		__UpperCAmelCase      		= Image.fromarray(np.uinta(_lowercase						)						).convert('''RGB'''						).resize((2_56, 2_56)						)

		if str(_lowercase						).startswith('''mps'''						):
			__UpperCAmelCase      		= torch.manual_seed(_lowercase						)
		else:
			__UpperCAmelCase      		= torch.Generator(device=_lowercase						).manual_seed(_lowercase						)
		__UpperCAmelCase      		= {
		    '''prompt''': '''horse''',
		    '''image''': init_image,
		    '''image_embeds''': image_embeds,
		    '''negative_image_embeds''': negative_image_embeds,
		    '''generator''': generator,
		    '''height''': 64,
		    '''width''': 64,
		    '''num_inference_steps''': 10,
		    '''guidance_scale''': 7.0,
		    '''strength''': 0.2,
		    '''output_type''': '''np''',
		}
		return inputs



	def        a  (  self     :       List[Any]						):
		__UpperCAmelCase      		= '''cpu'''

		__UpperCAmelCase      		= self.get_dummy_components()

		__UpperCAmelCase      		= self.pipeline_class(**_lowercase						)
		__UpperCAmelCase      		= pipe.to(_lowercase						)

		pipe.set_progress_bar_config(disable=_lowercase						)

		__UpperCAmelCase      		= pipe(**self.get_dummy_inputs(_lowercase						)						)
		__UpperCAmelCase      		= output.images

		__UpperCAmelCase      		= pipe(
		    **self.get_dummy_inputs(_lowercase						)			,    return_dict=_lowercase			,    )[0]

		__UpperCAmelCase      		= image[0, -3:, -3:, -1]
		__UpperCAmelCase      		= image_from_tuple[0, -3:, -3:, -1]

		assert image.shape == (1, 64, 64, 3)

		__UpperCAmelCase      		= np.array(
		    [0.61_474_943, 0.6_073_539, 0.43_308_544, 0.5_928_269, 0.47_493_595, 0.46_755_973, 0.4_613_838, 0.45_368_797, 0.50_119_233]						)
		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      _UpperCAmelCase   (							unittest.TestCase       ):



	def        a  (  self     :       List[Any]						):
		# clean up the VRAM after each test
		super().tearDown()
		gc.collect()
		torch.cuda.empty_cache()



	def        a  (  self     :       List[str]						):
		__UpperCAmelCase      		= load_numpy(
		    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
		    '''/kandinsky/kandinsky_img2img_frog.npy'''						)

		__UpperCAmelCase      		= load_image(
		    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png'''						)
		__UpperCAmelCase      		= '''A red cartoon frog, 4k'''

		__UpperCAmelCase      		= KandinskyPriorPipeline.from_pretrained(
		    '''kandinsky-community/kandinsky-2-1-prior'''			,    torch_dtype=torch.floataa						)
		pipe_prior.to(_lowercase						)

		__UpperCAmelCase      		= KandinskyImgaImgPipeline.from_pretrained(
		    '''kandinsky-community/kandinsky-2-1'''			,    torch_dtype=torch.floataa						)
		__UpperCAmelCase      		= pipeline.to(_lowercase						)

		pipeline.set_progress_bar_config(disable=_lowercase						)

		__UpperCAmelCase      		= torch.Generator(device='''cpu'''						).manual_seed(0						)
		__UpperCAmelCase		,							__UpperCAmelCase      		= pipe_prior(
		    _lowercase			,    generator=_lowercase			,    num_inference_steps=5			,    negative_prompt=''''''			,    ).to_tuple()

		__UpperCAmelCase      		= pipeline(
		    _lowercase			,    image=_lowercase			,    image_embeds=_lowercase			,    negative_image_embeds=_lowercase			,    generator=_lowercase			,    num_inference_steps=1_00			,    height=7_68			,    width=7_68			,    strength=0.2			,    output_type='''np'''			,    )

		__UpperCAmelCase      		= output.images[0]

		assert image.shape == (7_68, 7_68, 3)

		assert_mean_pixel_difference(_lowercase			,    _lowercase						)




'''simple docstring'''
import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import PaddingStrategy, logging
from .tokenization_realm import RealmTokenizer


UpperCAmelCase_			:  int										=   logging.get_logger(__name__)

UpperCAmelCase_			:  int										=   {'vocab_file': 'vocab.txt', 'tokenizer_file': 'tokenizer.json'}

UpperCAmelCase_			:  List[str]										=   {
    'vocab_file': {
        'google/realm-cc-news-pretrained-embedder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/vocab.txt'
        ),
        'google/realm-cc-news-pretrained-encoder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/vocab.txt'
        ),
        'google/realm-cc-news-pretrained-scorer': (
            'https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/vocab.txt'
        ),
        'google/realm-cc-news-pretrained-openqa': (
            'https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/vocab.txt'
        ),
        'google/realm-orqa-nq-openqa': 'https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/vocab.txt',
        'google/realm-orqa-nq-reader': 'https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/vocab.txt',
        'google/realm-orqa-wq-openqa': 'https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/vocab.txt',
        'google/realm-orqa-wq-reader': 'https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/vocab.txt',
    },
    'tokenizer_file': {
        'google/realm-cc-news-pretrained-embedder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/tokenizer.jsont'
        ),
        'google/realm-cc-news-pretrained-encoder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/tokenizer.json'
        ),
        'google/realm-cc-news-pretrained-scorer': (
            'https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/tokenizer.json'
        ),
        'google/realm-cc-news-pretrained-openqa': (
            'https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/tokenizer.json'
        ),
        'google/realm-orqa-nq-openqa': (
            'https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/tokenizer.json'
        ),
        'google/realm-orqa-nq-reader': (
            'https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/tokenizer.json'
        ),
        'google/realm-orqa-wq-openqa': (
            'https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/tokenizer.json'
        ),
        'google/realm-orqa-wq-reader': (
            'https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/tokenizer.json'
        ),
    },
}

UpperCAmelCase_			:  str										=   {
    'google/realm-cc-news-pretrained-embedder': 5_1_2,
    'google/realm-cc-news-pretrained-encoder': 5_1_2,
    'google/realm-cc-news-pretrained-scorer': 5_1_2,
    'google/realm-cc-news-pretrained-openqa': 5_1_2,
    'google/realm-orqa-nq-openqa': 5_1_2,
    'google/realm-orqa-nq-reader': 5_1_2,
    'google/realm-orqa-wq-openqa': 5_1_2,
    'google/realm-orqa-wq-reader': 5_1_2,
}

UpperCAmelCase_			:  str										=   {
    'google/realm-cc-news-pretrained-embedder': {'do_lower_case': True},
    'google/realm-cc-news-pretrained-encoder': {'do_lower_case': True},
    'google/realm-cc-news-pretrained-scorer': {'do_lower_case': True},
    'google/realm-cc-news-pretrained-openqa': {'do_lower_case': True},
    'google/realm-orqa-nq-openqa': {'do_lower_case': True},
    'google/realm-orqa-nq-reader': {'do_lower_case': True},
    'google/realm-orqa-wq-openqa': {'do_lower_case': True},
    'google/realm-orqa-wq-reader': {'do_lower_case': True},
}

class     _lowerCamelCase       (     snake_case_					):




 '''simple docstring'''
 __lowercase					:		Union[str, Any]     				=    VOCAB_FILES_NAMES
 __lowercase					:		int     				=    PRETRAINED_VOCAB_FILES_MAP
 __lowercase					:		Union[str, Any]     				=    PRETRAINED_INIT_CONFIGURATION
 __lowercase					:		Union[str, Any]     				=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __lowercase					:		List[str]     				=    RealmTokenizer


 def __init__(   self   ,   __lowercase=None   ,   __lowercase=None   ,   __lowercase=True   ,   __lowercase="[UNK]"   ,   __lowercase="[SEP]"   ,   __lowercase="[PAD]"   ,   __lowercase="[CLS]"   ,   __lowercase="[MASK]"   ,   __lowercase=True   ,   __lowercase=None   ,   **__lowercase   ,   ):

  """simple docstring"""







  super().__init__(
      __lowercase   ,   tokenizer_file=__lowercase   ,   do_lower_case=__lowercase   ,   unk_token=__lowercase   ,   sep_token=__lowercase   ,   pad_token=__lowercase   ,   cls_token=__lowercase   ,   mask_token=__lowercase   ,   tokenize_chinese_chars=__lowercase   ,   strip_accents=__lowercase   ,   **__lowercase   ,   )

  __A	:		Optional[int]	  =      json.loads(self.backend_tokenizer.normalizer.__getstate__()    )
  if (
      normalizer_state.get('lowercase'   ,   __lowercase    ) != do_lower_case
      or normalizer_state.get('strip_accents'   ,   __lowercase    ) != strip_accents
      or normalizer_state.get('handle_chinese_chars'   ,   __lowercase    ) != tokenize_chinese_chars
  ):
   __A	:		Tuple	  =      getattr(__lowercase   ,   normalizer_state.pop('type'    )    )
   __A	:		Optional[int]	  =      do_lower_case
   __A	:		List[str]	  =      strip_accents
   __A	:		Dict	  =      tokenize_chinese_chars
   __A	:		List[Any]	  =      normalizer_class(**__lowercase    )

  __A	:		int	  =      do_lower_case


 def 			snake_case__ (   self   ,   __lowercase   ,   **__lowercase    ):

  """simple docstring"""







  __A	:		Dict	  =      PaddingStrategy.MAX_LENGTH

  __A	:		Optional[int]	  =      text
  __A	:		Union[str, Any]	  =      kwargs.pop('text_pair'   ,   __lowercase    )
  __A	:		List[str]	  =      kwargs.pop('return_tensors'   ,   __lowercase    )

  __A	:		int	  =      {
      'input_ids': [],
      'attention_mask': [],
      'token_type_ids': [],
  }

  for idx, candidate_text in enumerate(__lowercase    ):
   if batch_text_pair is not None:
    __A	:		Tuple	  =      batch_text_pair[idx]
   else:
    __A	:		Union[str, Any]	  =      None

   __A	:		Optional[int]	  =      super().__call__(__lowercase   ,   __lowercase   ,   return_tensors=__lowercase   ,   **__lowercase    )

   __A	:		str	  =      encoded_candidates.get('input_ids'    )
   __A	:		Union[str, Any]	  =      encoded_candidates.get('attention_mask'    )
   __A	:		Optional[int]	  =      encoded_candidates.get('token_type_ids'    )

   if encoded_input_ids is not None:
    output_data["input_ids"].append(__lowercase    )
   if encoded_attention_mask is not None:
    output_data["attention_mask"].append(__lowercase    )
   if encoded_token_type_ids is not None:
    output_data["token_type_ids"].append(__lowercase    )

  __A	:		List[str]	  =      {key: item for key, item in output_data.items() if len(__lowercase    ) != 0}

  return BatchEncoding(__lowercase   ,   tensor_type=__lowercase    )


 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase=None    ):

  """simple docstring"""







  __A	:		Union[str, Any]	  =      [self.cls_token_id] + token_ids_a + [self.sep_token_id]

  if token_ids_a:
   output += token_ids_a + [self.sep_token_id]

  return output


 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase = None    ):

  """simple docstring"""







  __A	:		Union[str, Any]	  =      [self.sep_token_id]
  __A	:		Tuple	  =      [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    ) * [0] + len(token_ids_a + sep    ) * [1]







 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase = None    ):

  """simple docstring"""







  __A	:		int	  =      self._tokenizer.model.save(__lowercase   ,   name=__lowercase    )
  return tuple(__lowercase    )






