






'''simple docstring'''
import time
import warnings
from abc import ABC
from copy import deepcopy
from typing import Optional

import torch

from ..utils import add_start_docstrings, logging


UpperCamelCase__ =			logging.get_logger(__name__)


UpperCamelCase__ =			r'\n    Args:\n        input_ids (`torch.LongTensor` of shape `(batch_size, sequence_length)`):\n            Indices of input sequence tokens in the vocabulary.\n\n            Indices can be obtained using [`AutoTokenizer`]. See [`PreTrainedTokenizer.encode`] and\n            [`PreTrainedTokenizer.__call__`] for details.\n\n            [What are input IDs?](../glossary#input-ids)\n        scores (`torch.FloatTensor` of shape `(batch_size, config.vocab_size)`):\n            Prediction scores of a language modeling head. These can be scores for each vocabulary token before SoftMax\n            or scores for each vocabulary token after SoftMax.\n        kwargs (`Dict[str, Any]`, *optional*):\n            Additional stopping criteria specific kwargs.\n\n    Return:\n        `bool`. `False` indicates we should continue, `True` indicates we should stop.\n\n'


class        _UpperCAmelCase						(		snake_case			):




    @add_start_docstrings(a			)
    def __call__(			self  :       Any  ,     a  :       torch.LongTensor  ,     a  :       torch.FloatTensor  ,     **a  :       Dict			):






       '''simple docstring'''






       raise NotImplementedError("StoppingCriteria needs to be subclassed"			)


class        _UpperCAmelCase						(		snake_case			):




    def __init__(			self  :       str  ,     a  :       int  ,     a  :       Optional[int] = None			):






       '''simple docstring'''






       lowercase_	:    List[Any]           =     max_length
       lowercase_	:    List[Any]           =     max_position_embeddings





    @add_start_docstrings(a			)
    def __call__(			self  :       Union[str, Any]  ,     a  :       torch.LongTensor  ,     a  :       torch.FloatTensor  ,     **a  :       List[str]			):






       '''simple docstring'''






       lowercase_	:    Tuple           =     input_ids.shape[-1]
       lowercase_	:    str           =     cur_len >= self.max_length
       if self.max_position_embeddings is not None and not is_done and cur_len >= self.max_position_embeddings:
          logger.warning_once(
              "This is a friendly reminder - the current text generation call will exceed the model's predefined "
              f"""maximum length ({self.max_position_embeddings}). Depending on the model, you may observe """
              "exceptions, performance degradation, or nothing at all."			)
       return is_done


class        _UpperCAmelCase						(		snake_case			):




    def __init__(			self  :       List[Any]  ,     a  :       int  ,     a  :       int			):






       '''simple docstring'''






       warnings.warn(
           "The class `MaxNewTokensCriteria` is deprecated. "
           f"""Please use `MaxLengthCriteria(max_length={start_length + max_new_tokens})` """
           "with `max_length = start_length + max_new_tokens` instead."  ,     a  ,     )
       lowercase_	:    List[Any]           =     start_length
       lowercase_	:    Tuple           =     max_new_tokens
       lowercase_	:    str           =     start_length + max_new_tokens





    @add_start_docstrings(a			)
    def __call__(			self  :       int  ,     a  :       torch.LongTensor  ,     a  :       torch.FloatTensor  ,     **a  :       Optional[int]			):






       '''simple docstring'''






       return input_ids.shape[-1] >= self.max_length


class        _UpperCAmelCase						(		snake_case			):




    def __init__(			self  :       Optional[int]  ,     a  :       float  ,     a  :       Optional[float] = None			):






       '''simple docstring'''






       lowercase_	:    Tuple           =     max_time
       lowercase_	:    Optional[int]           =     time.time() if initial_timestamp is None else initial_timestamp





    @add_start_docstrings(a			)
    def __call__(			self  :       int  ,     a  :       torch.LongTensor  ,     a  :       torch.FloatTensor  ,     **a  :       Optional[Any]			):






       '''simple docstring'''






       return time.time() - self.initial_timestamp > self.max_time


class        _UpperCAmelCase						(		snake_case			):




    @add_start_docstrings(a			)
    def __call__(			self  :       List[Any]  ,     a  :       torch.LongTensor  ,     a  :       torch.FloatTensor  ,     **a  :       List[Any]			):






       '''simple docstring'''






       return any(criteria(a  ,     a			) for criteria in self			)





    @property
    def 				lowerCAmelCase__				(			self  :       Optional[int]			):






       '''simple docstring'''






       for stopping_criterium in self:
          if isinstance(a  ,     a			):
             return stopping_criterium.max_length
          elif isinstance(a  ,     a			):
             return stopping_criterium.max_length
       return None




def 		__SCREAMING_SNAKE_CASE					(  _UpperCamelCase						,		_UpperCamelCase							):


   """simple docstring"""




   lowercase_	:    str           =     stopping_criteria.max_length
   lowercase_	:    List[str]           =     deepcopy(_UpperCamelCase							)
   if stopping_max_length is not None and stopping_max_length != max_length:
      warnings.warn("You set different `max_length` for stopping criteria and `max_length` parameter"						,		_UpperCamelCase							)
   elif stopping_max_length is None:
      new_stopping_criteria.append(MaxLengthCriteria(max_length=_UpperCamelCase							)							)
   return new_stopping_criteria












'''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        _UpperCAmelCase						(		snake_case				,					unittest.TestCase			):
    __lowerCamelCase:   Dict		       =    KandinskyVaaPriorPipeline
    __lowerCamelCase:   Optional[int]		       =    ['prompt']
    __lowerCamelCase:   Any		       =    ['prompt', 'negative_prompt']
    __lowerCamelCase:   List[Any]		       =    [
        'num_images_per_prompt',
        'generator',
        'num_inference_steps',
        'latents',
        'negative_prompt',
        'guidance_scale',
        'output_type',
        'return_dict',
    ]
    __lowerCamelCase:   List[Any]		       =    False




    @property
    def 				lowerCAmelCase__				(			self  :       Optional[Any]			):






       '''simple docstring'''






       return 3_2




    @property
    def 				lowerCAmelCase__				(			self  :       Any			):






       '''simple docstring'''






       return 3_2




    @property
    def 				lowerCAmelCase__				(			self  :       Any			):






       '''simple docstring'''






       return self.time_input_dim




    @property
    def 				lowerCAmelCase__				(			self  :       str			):






       '''simple docstring'''






       return self.time_input_dim * 4




    @property
    def 				lowerCAmelCase__				(			self  :       Union[str, Any]			):






       '''simple docstring'''






       return 1_0_0




    @property
    def 				lowerCAmelCase__				(			self  :       List[Any]			):






       '''simple docstring'''






       lowercase_	:    Any           =     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"			)
       return tokenizer




    @property
    def 				lowerCAmelCase__				(			self  :       Optional[Any]			):






       '''simple docstring'''






       torch.manual_seed(0			)
       lowercase_	:    Tuple           =     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=3_7  ,     layer_norm_eps=1e-05  ,     num_attention_heads=4  ,     num_hidden_layers=5  ,     pad_token_id=1  ,     vocab_size=1_0_0_0  ,     )
       return CLIPTextModelWithProjection(a			)




    @property
    def 				lowerCAmelCase__				(			self  :       Tuple			):






       '''simple docstring'''






       torch.manual_seed(0			)

       lowercase_	:    List[str]           =     {
           "num_attention_heads": 2,
           "attention_head_dim": 1_2,
           "embedding_dim": self.text_embedder_hidden_size,
           "num_layers": 1,
       }

       lowercase_	:    Union[str, Any]           =     PriorTransformer(**a			)
       # 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
       lowercase_	:    List[Any]           =     nn.Parameter(torch.ones(model.clip_std.shape			)			)
       return model




    @property
    def 				lowerCAmelCase__				(			self  :       Tuple			):






       '''simple docstring'''






       torch.manual_seed(0			)
       lowercase_	:    Dict           =     CLIPVisionConfig(
           hidden_size=self.text_embedder_hidden_size  ,     image_size=2_2_4  ,     projection_dim=self.text_embedder_hidden_size  ,     intermediate_size=3_7  ,     num_attention_heads=4  ,     num_channels=3  ,     num_hidden_layers=5  ,     patch_size=1_4  ,     )

       lowercase_	:    Optional[Any]           =     CLIPVisionModelWithProjection(a			)
       return model




    @property
    def 				lowerCAmelCase__				(			self  :       List[str]			):






       '''simple docstring'''






       lowercase_	:    List[str]           =     CLIPImageProcessor(
           crop_size=2_2_4  ,     do_center_crop=a  ,     do_normalize=a  ,     do_resize=a  ,     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=2_2_4  ,     )

       return image_processor




    def 				lowerCAmelCase__				(			self  :       List[str]			):






       '''simple docstring'''






       lowercase_	:    Any           =     self.dummy_prior
       lowercase_	:    Optional[Any]           =     self.dummy_image_encoder
       lowercase_	:    List[Any]           =     self.dummy_text_encoder
       lowercase_	:    Any           =     self.dummy_tokenizer
       lowercase_	:    Optional[Any]           =     self.dummy_image_processor

       lowercase_	:    List[str]           =     UnCLIPScheduler(
           variance_type="fixed_small_log"  ,     prediction_type="sample"  ,     num_train_timesteps=1_0_0_0  ,     clip_sample=a  ,     clip_sample_range=10.0  ,     )

       lowercase_	:    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  :       Any  ,     a  :       Dict  ,     a  :       Dict=0			):






       '''simple docstring'''






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




    def 				lowerCAmelCase__				(			self  :       Union[str, Any]			):






       '''simple docstring'''






       lowercase_	:    str           =     "cpu"

       lowercase_	:    Any           =     self.get_dummy_components()

       lowercase_	:    int           =     self.pipeline_class(**a			)
       lowercase_	:    Any           =     pipe.to(a			)

       pipe.set_progress_bar_config(disable=a			)

       lowercase_	:    Any           =     pipe(**self.get_dummy_inputs(a			)			)
       lowercase_	:    List[Any]           =     output.image_embeds

       lowercase_	:    str           =     pipe(
           **self.get_dummy_inputs(a			)  ,     return_dict=a  ,     )[0]

       lowercase_	:    Any           =     image[0, -1_0:]
       lowercase_	:    Dict           =     image_from_tuple[0, -1_0:]

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

       lowercase_	:    int           =     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  :       Optional[Any]			):






       '''simple docstring'''






       lowercase_	:    int           =     torch_device == "cpu"
       lowercase_	:    Tuple           =     True
       lowercase_	:    str           =     False

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





    @skip_mps
    def 				lowerCAmelCase__				(			self  :       Union[str, Any]			):






       '''simple docstring'''






       lowercase_	:    Any           =     torch_device == "cpu"
       lowercase_	:    int           =     False

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




