




'''simple docstring'''







from typing import Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
    from PIL import Image

    from ..image_utils import load_image

if is_torch_available():
    from ..models.auto.modeling_auto import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING

lowerCAmelCase_     						=				logging.get_logger(__name__)

@add_end_docstrings(UpperCAmelCase      )
class 	_snake_case(			UpperCAmelCase      ):






 def __init__(self       :  List[Any]    ,		*a       :  Dict    ,		**a       :  int    )       ->   int:






    """simple docstring"""


    super().__init__(*a    ,		**a    )
    self.check_model_type(a    )






 def 						_UpperCamelCase (self       :  List[Any]    ,		a       :  List[Any]=None    ,		a       :  Optional[int]=None    ,		a       :  Any=None    ,		**a       :  str    )       ->   Optional[Any]:






    """simple docstring"""


    A__  ,			A__       =     {}, {}
    if padding is not None:
       A__       =     padding
    if truncation is not None:
       A__       =     truncation
    if top_k is not None:
       A__       =     top_k
    return preprocess_params, {}, postprocess_params






 def __call__(self       :  Any    ,		a       :  Union["Image.Image", str]    ,		a       :  str = None    ,		**a       :  str    )       ->   Optional[int]:






    """simple docstring"""


    if isinstance(a    ,		(Image.Image, str)    ) and isinstance(a    ,		a    ):
       A__       =     {'image': image, 'question': question}
    else:
       A__       =     image
    A__       =     super().__call__(a    ,		**a    )
    return results






 def 						_UpperCamelCase (self       :  List[str]    ,		a       :  Optional[int]    ,		a       :  Union[str, Any]=False    ,		a       :  Tuple=False    )       ->   List[Any]:






    """simple docstring"""


    A__       =     load_image(inputs['image']    )
    A__       =     self.tokenizer(
        inputs['question']    ,		return_tensors=self.framework    ,		padding=a    ,		truncation=a    )
    A__       =     self.image_processor(images=a    ,		return_tensors=self.framework    )
    model_inputs.update(a    )
    return model_inputs






 def 						_UpperCamelCase (self       :  str    ,		a       :  Any    )       ->   List[str]:






    """simple docstring"""


    A__       =     self.model(**a    )
    return model_outputs






 def 						_UpperCamelCase (self       :  Tuple    ,		a       :  str    ,		a       :  Tuple=5    )       ->   int:






    """simple docstring"""


    if top_k > self.model.config.num_labels:
       A__       =     self.model.config.num_labels

    if self.framework == "pt":
       A__       =     model_outputs.logits.sigmoid()[0]
       A__  ,			A__       =     probs.topk(a    )
    else:
       raise ValueError(f"""Unsupported framework: {self.framework}"""    )

    A__       =     scores.tolist()
    A__       =     ids.tolist()
    return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(a    ,		a    )]








'''simple docstring'''







import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()

class 	_snake_case(			UpperCAmelCase       , UpperCAmelCase       , UpperCAmelCase       , unittest.TestCase      ):
 __snake_case:							Union[str, Any]						=					StableUnCLIPPipeline
 __snake_case:							List[str]						=					TEXT_TO_IMAGE_PARAMS
 __snake_case:							str						=					TEXT_TO_IMAGE_BATCH_PARAMS
 __snake_case:							Dict						=					TEXT_TO_IMAGE_IMAGE_PARAMS
 __snake_case:							List[str]						=					TEXT_TO_IMAGE_IMAGE_PARAMS

 # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
 __snake_case:							str						=					False






 def 						_UpperCamelCase (self       :  Optional[int]    )       ->   int:






    """simple docstring"""


    A__       =     32
    A__       =     embedder_hidden_size

    # prior components

    torch.manual_seed(0    )
    A__       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

    torch.manual_seed(0    )
    A__       =     CLIPTextModelWithProjection(
        CLIPTextConfig(
            bos_token_id=0    ,		eos_token_id=2    ,		hidden_size=a    ,		projection_dim=a    ,		intermediate_size=37    ,		layer_norm_eps=1e-05    ,		num_attention_heads=4    ,		num_hidden_layers=5    ,		pad_token_id=1    ,		vocab_size=10_00    ,		)    )

    torch.manual_seed(0    )
    A__       =     PriorTransformer(
        num_attention_heads=2    ,		attention_head_dim=12    ,		embedding_dim=a    ,		num_layers=1    ,		)

    torch.manual_seed(0    )
    A__       =     DDPMScheduler(
        variance_type='fixed_small_log'    ,		prediction_type='sample'    ,		num_train_timesteps=10_00    ,		clip_sample=a    ,		clip_sample_range=5.0    ,		beta_schedule='squaredcos_cap_v2'    ,		)

    # regular denoising components

    torch.manual_seed(0    )
    A__       =     StableUnCLIPImageNormalizer(embedding_dim=a    )
    A__       =     DDPMScheduler(beta_schedule='squaredcos_cap_v2'    )

    torch.manual_seed(0    )
    A__       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

    torch.manual_seed(0    )
    A__       =     CLIPTextModel(
        CLIPTextConfig(
            bos_token_id=0    ,		eos_token_id=2    ,		hidden_size=a    ,		projection_dim=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    ,		)    )

    torch.manual_seed(0    )
    A__       =     UNetaDConditionModel(
        sample_size=32    ,		in_channels=4    ,		out_channels=4    ,		down_block_types=('CrossAttnDownBlock2D', 'DownBlock2D')    ,		up_block_types=('UpBlock2D', 'CrossAttnUpBlock2D')    ,		block_out_channels=(32, 64)    ,		attention_head_dim=(2, 4)    ,		class_embed_type='projection'    ,		projection_class_embeddings_input_dim=embedder_projection_dim * 2    ,		cross_attention_dim=a    ,		layers_per_block=1    ,		upcast_attention=a    ,		use_linear_projection=a    ,		)

    torch.manual_seed(0    )
    A__       =     DDIMScheduler(
        beta_schedule='scaled_linear'    ,		beta_start=0.0_0085    ,		beta_end=0.012    ,		prediction_type='v_prediction'    ,		set_alpha_to_one=a    ,		steps_offset=1    ,		)

    torch.manual_seed(0    )
    A__       =     AutoencoderKL()

    A__       =     {
        # prior components
        'prior_tokenizer': prior_tokenizer,
        'prior_text_encoder': prior_text_encoder,
        'prior': prior,
        'prior_scheduler': prior_scheduler,
        # image noising components
        'image_normalizer': image_normalizer,
        'image_noising_scheduler': image_noising_scheduler,
        # regular denoising components
        'tokenizer': tokenizer,
        'text_encoder': text_encoder,
        'unet': unet,
        'scheduler': scheduler,
        'vae': vae,
    }

    return components






 def 						_UpperCamelCase (self       :  List[Any]    ,		a       :  Optional[Any]    ,		a       :  str=0    )       ->   Dict:






    """simple docstring"""


    if str(a    ).startswith('mps'    ):
       A__       =     torch.manual_seed(a    )
    else:
       A__       =     torch.Generator(device=a    ).manual_seed(a    )
    A__       =     {
        'prompt': 'A painting of a squirrel eating a burger',
        'generator': generator,
        'num_inference_steps': 2,
        'prior_num_inference_steps': 2,
        'output_type': 'numpy',
    }
    return inputs






 def 						_UpperCamelCase (self       :  List[str]    )       ->   Optional[int]:






    """simple docstring"""


    A__       =     torch_device == 'cpu'

    self._test_attention_slicing_forward_pass(test_max_difference=a    )






 def 						_UpperCamelCase (self       :  Dict    )       ->   str:






    """simple docstring"""


    A__       =     torch_device in ['cpu', 'mps']

    self._test_inference_batch_single_identical(test_max_difference=a    )




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






 def 						_UpperCamelCase (self       :  Any    )       ->   Optional[Any]:






    """simple docstring"""


    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()






 def 						_UpperCamelCase (self       :  Dict    )       ->   Union[str, Any]:






    """simple docstring"""


    A__       =     load_numpy(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy'    )

    A__       =     StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l'    ,		torch_dtype=torch.floataa    )
    pipe.to(a    )
    pipe.set_progress_bar_config(disable=a    )
    # stable unclip will oom when integration tests are run on a V100,
    # so turn on memory savings
    pipe.enable_attention_slicing()
    pipe.enable_sequential_cpu_offload()

    A__       =     torch.Generator(device='cpu'    ).manual_seed(0    )
    A__       =     pipe('anime turle'    ,		generator=a    ,		output_type='np'    )

    A__       =     output.images[0]

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

    assert_mean_pixel_difference(a    ,		a    )






 def 						_UpperCamelCase (self       :  Optional[Any]    )       ->   str:






    """simple docstring"""


    torch.cuda.empty_cache()
    torch.cuda.reset_max_memory_allocated()
    torch.cuda.reset_peak_memory_stats()

    A__       =     StableUnCLIPPipeline.from_pretrained('fusing/stable-unclip-2-1-l'    ,		torch_dtype=torch.floataa    )
    A__       =     pipe.to(a    )
    pipe.set_progress_bar_config(disable=a    )
    pipe.enable_attention_slicing()
    pipe.enable_sequential_cpu_offload()

    A__       =     pipe(
        'anime turtle'    ,		prior_num_inference_steps=2    ,		num_inference_steps=2    ,		output_type='np'    ,		)

    A__       =     torch.cuda.max_memory_allocated()
    # make sure that less than 7 GB is allocated
    assert mem_bytes < 7 * 10**9


