import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    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       __lowerCAmelCase     (				a_,		a_,		unittest.TestCase  ):



 '''simple docstring'''
 _lowerCamelCase  :		Optional[Any] 			=       StableDiffusionXLPipeline
 _lowerCamelCase  :		Tuple 			=       TEXT_TO_IMAGE_PARAMS
 _lowerCamelCase  :		Union[str, Any] 			=       TEXT_TO_IMAGE_BATCH_PARAMS
 _lowerCamelCase  :		Tuple 			=       TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowerCamelCase  :		str 			=       TEXT_TO_IMAGE_IMAGE_PARAMS

 def     __snake_case    (	self				:						Dict     )       ->							Optional[int]:
        torch.manual_seed(0     )
        _UpperCamelCase   :				List[str]       =     UNetaDConditionModel(
            block_out_channels=(3_2, 6_4)				,	layers_per_block=2				,	sample_size=3_2				,	in_channels=4				,	out_channels=4				,	down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')				,	up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')				,	attention_head_dim=(2, 4)				,	use_linear_projection=lowerCamelCase				,	addition_embed_type='text_time'				,	addition_time_embed_dim=8				,	transformer_layers_per_block=(1, 2)				,	projection_class_embeddings_input_dim=8_0				,	cross_attention_dim=6_4				,	)
        _UpperCamelCase   :				List[Any]       =     EulerDiscreteScheduler(
            beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	steps_offset=1				,	beta_schedule='scaled_linear'				,	timestep_spacing='leading'				,	)
        torch.manual_seed(0     )
        _UpperCamelCase   :				Any       =     AutoencoderKL(
            block_out_channels=[3_2, 6_4]				,	in_channels=3				,	out_channels=3				,	down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']				,	up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']				,	latent_channels=4				,	sample_size=1_2_8				,	)
        torch.manual_seed(0     )
        _UpperCamelCase   :				Tuple       =     CLIPTextConfig(
            bos_token_id=0				,	eos_token_id=2				,	hidden_size=3_2				,	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				,	hidden_act='gelu'				,	projection_dim=3_2				,	)
        _UpperCamelCase   :				List[Any]       =     CLIPTextModel(lowerCamelCase     )
        _UpperCamelCase   :				Tuple       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				,	local_files_only=lowerCamelCase     )

        _UpperCamelCase   :				Union[str, Any]       =     CLIPTextModelWithProjection(lowerCamelCase     )
        _UpperCamelCase   :				str       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				,	local_files_only=lowerCamelCase     )

        _UpperCamelCase   :				List[Any]       =     {
            'unet': unet,
            'scheduler': scheduler,
            'vae': vae,
            'text_encoder': text_encoder,
            'tokenizer': tokenizer,
            'text_encoder_2': text_encoder_a,
            'tokenizer_2': tokenizer_a,
            # "safety_checker": None,
            # "feature_extractor": None,
        }
        return components

 def     __snake_case    (	self				:						Union[str, Any]				,	lowerCamelCase				:						Optional[Any]				,	lowerCamelCase				:						Union[str, Any]=0     )       ->							int:
        if str(lowerCamelCase     ).startswith('mps'     ):
               _UpperCamelCase   :				Optional[Any]       =     torch.manual_seed(lowerCamelCase     )
        else:
               _UpperCamelCase   :				Any       =     torch.Generator(device=lowerCamelCase     ).manual_seed(lowerCamelCase     )
        _UpperCamelCase   :				Optional[Any]       =     {
            'prompt': 'A painting of a squirrel eating a burger',
            'generator': generator,
            'num_inference_steps': 2,
            'guidance_scale': 5.0,
            'output_type': 'numpy',
        }
        return inputs

 def     __snake_case    (	self				:						List[Any]     )       ->							Optional[int]:
        _UpperCamelCase   :				List[Any]       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
        _UpperCamelCase   :				int       =     self.get_dummy_components()
        _UpperCamelCase   :				Tuple       =     StableDiffusionXLPipeline(**lowerCamelCase     )
        _UpperCamelCase   :				Union[str, Any]       =     sd_pipe.to(lowerCamelCase     )
        sd_pipe.set_progress_bar_config(disable=lowerCamelCase     )

        _UpperCamelCase   :				List[Any]       =     self.get_dummy_inputs(lowerCamelCase     )
        _UpperCamelCase   :				Tuple       =     sd_pipe(**lowerCamelCase     ).images
        _UpperCamelCase   :				Union[str, Any]       =     image[0, -3:, -3:, -1]

        assert image.shape == (1, 6_4, 6_4, 3)
        _UpperCamelCase   :				Union[str, Any]       =     np.array([0.5_8_7_3, 0.6_1_2_8, 0.4_7_9_7, 0.5_1_2_2, 0.5_6_7_4, 0.4_6_3_9, 0.5_2_2_7, 0.5_1_4_9, 0.4_7_4_7]     )

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

 def     __snake_case    (	self				:						Optional[int]     )       ->							Dict:
        _UpperCamelCase   :				Dict       =     self.get_dummy_components()
        _UpperCamelCase   :				str       =     StableDiffusionXLPipeline(**lowerCamelCase     )
        _UpperCamelCase   :				Optional[Any]       =     sd_pipe.to(lowerCamelCase     )
        _UpperCamelCase   :				str       =     sd_pipe.to(lowerCamelCase     )
        sd_pipe.set_progress_bar_config(disable=lowerCamelCase     )

        # forward without prompt embeds
        _UpperCamelCase   :				Union[str, Any]       =     self.get_dummy_inputs(lowerCamelCase     )
        _UpperCamelCase   :				List[str]       =     3 * ['this is a negative prompt']
        _UpperCamelCase   :				Any       =     negative_prompt
        _UpperCamelCase   :				List[Any]       =     3 * [inputs['prompt']]

        _UpperCamelCase   :				List[Any]       =     sd_pipe(**lowerCamelCase     )
        _UpperCamelCase   :				Any       =     output.images[0, -3:, -3:, -1]

        # forward with prompt embeds
        _UpperCamelCase   :				Optional[int]       =     self.get_dummy_inputs(lowerCamelCase     )
        _UpperCamelCase   :				Dict       =     3 * ['this is a negative prompt']
        _UpperCamelCase   :				int       =     3 * [inputs.pop('prompt'     )]

        (
            (
            _UpperCamelCase
        )  ,					(
            _UpperCamelCase
        )  ,					(
            _UpperCamelCase
        )  ,					(
            _UpperCamelCase
        )  ,					
        )   :				Tuple       =     sd_pipe.encode_prompt(lowerCamelCase				,	negative_prompt=lowerCamelCase     )

        _UpperCamelCase   :				Optional[int]       =     sd_pipe(
            **lowerCamelCase				,	prompt_embeds=lowerCamelCase				,	negative_prompt_embeds=lowerCamelCase				,	pooled_prompt_embeds=lowerCamelCase				,	negative_pooled_prompt_embeds=lowerCamelCase				,	)
        _UpperCamelCase   :				str       =     output.images[0, -3:, -3:, -1]

        # make sure that it's equal
        assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()     ).max() < 1e-4

 def     __snake_case    (	self				:						List[str]     )       ->							int:
        super().test_attention_slicing_forward_pass(expected_max_diff=3e-3     )

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







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



 '''simple docstring'''

 def     __snake_case    (	self				:						Optional[Any]     )       ->							Optional[Any]:
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()

 def     __snake_case    (	self				:						str				,	lowerCamelCase				:						List[str]				,	lowerCamelCase				:						Union[str, Any]="cpu"				,	lowerCamelCase				:						Tuple=torch.floataa				,	lowerCamelCase				:						int=0     )       ->							Optional[Any]:
        _UpperCamelCase   :				Optional[Any]       =     torch.Generator(device=lowerCamelCase     ).manual_seed(lowerCamelCase     )
        _UpperCamelCase   :				str       =     np.random.RandomState(lowerCamelCase     ).standard_normal((1, 4, 6_4, 6_4)     )
        _UpperCamelCase   :				Optional[int]       =     torch.from_numpy(lowerCamelCase     ).to(device=lowerCamelCase				,	dtype=lowerCamelCase     )
        _UpperCamelCase   :				str       =     {
            'prompt': 'a photograph of an astronaut riding a horse',
            'latents': latents,
            'generator': generator,
            'num_inference_steps': 3,
            'guidance_scale': 7.5,
            'output_type': 'numpy',
        }
        return inputs

 def     __snake_case    (	self				:						int     )       ->							Union[str, Any]:
        _UpperCamelCase   :				int       =     DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base'     )
        pipe.to(lowerCamelCase     )
        pipe.set_progress_bar_config(disable=lowerCamelCase     )

        _UpperCamelCase   :				Union[str, Any]       =     self.get_inputs(lowerCamelCase     )
        _UpperCamelCase   :				List[str]       =     pipe(**lowerCamelCase     ).images
        _UpperCamelCase   :				List[str]       =     image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 5_1_2, 5_1_2, 3)
        _UpperCamelCase   :				Any       =     np.array([0.4_9_4_9_3, 0.4_7_8_9_6, 0.4_0_7_9_8, 0.5_4_2_1_4, 0.5_3_2_1_2, 0.4_8_2_0_2, 0.4_7_6_5_6, 0.4_6_3_2_9, 0.4_8_5_0_6]     )
        assert np.abs(image_slice - expected_slice     ).max() < 7e-3




from typing import List

from ...processing_utils import ProcessorMixin
from ...utils import is_torch_available


if is_torch_available():
 import torch




class       __lowerCAmelCase     (				a_  ):



 '''simple docstring'''
 _lowerCamelCase  :		Dict 			=       ['image_processor', 'tokenizer']
 _lowerCamelCase  :		Tuple 			=       'OneFormerImageProcessor'
 _lowerCamelCase  :		Dict 			=       ('CLIPTokenizer', 'CLIPTokenizerFast')

 def __init__(	self				:						str				,	lowerCamelCase				:						Any=None				,	lowerCamelCase				:						List[Any]=None				,	lowerCamelCase				:						int = 7_7				,	lowerCamelCase				:						int = 7_7				,	**lowerCamelCase				:						Union[str, Any]     )       ->							Any:
        if image_processor is None:
               raise ValueError('You need to specify an `image_processor`.'     )
        if tokenizer is None:
               raise ValueError('You need to specify a `tokenizer`.'     )

        _UpperCamelCase   :				int       =     max_seq_length
        _UpperCamelCase   :				Optional[Any]       =     task_seq_length

        super().__init__(lowerCamelCase				,	lowerCamelCase     )

 def     __snake_case    (	self				:						List[str]				,	lowerCamelCase				:						int=None				,	lowerCamelCase				:						Optional[int]=7_7     )       ->							Tuple:
        if text_list is None:
               raise ValueError('tokens cannot be None.'     )

        _UpperCamelCase   :				Dict       =     self.tokenizer(lowerCamelCase				,	padding='max_length'				,	max_length=lowerCamelCase				,	truncation=lowerCamelCase     )

        _UpperCamelCase  ,					_UpperCamelCase   :				List[str]       =     tokens['attention_mask'], tokens['input_ids']

        _UpperCamelCase   :				Optional[int]       =     []
        for attn_mask, input_id in zip(lowerCamelCase				,	lowerCamelCase     ):
               _UpperCamelCase   :				int       =     torch.tensor(lowerCamelCase     ) * torch.tensor(lowerCamelCase     )
               token_inputs.append(token.unsqueeze(0     )     )

        _UpperCamelCase   :				Dict       =     torch.cat(lowerCamelCase				,	dim=0     )
        return token_inputs

 def __call__(	self				:						Union[str, Any]				,	lowerCamelCase				:						int=None				,	lowerCamelCase				:						str=None				,	lowerCamelCase				:						Any=None				,	**lowerCamelCase				:						List[Any]     )       ->							List[Any]:

        if task_inputs is None:
               raise ValueError('You have to specify the task_input. Found None.'     )
        elif images is None:
               raise ValueError('You have to specify the image. Found None.'     )

        if not all(task in ['semantic', 'instance', 'panoptic'] for task in task_inputs     ):
               raise ValueError('task_inputs must be semantic, instance, or panoptic.'     )

        _UpperCamelCase   :				Optional[Any]       =     self.image_processor(lowerCamelCase				,	lowerCamelCase				,	lowerCamelCase				,	**lowerCamelCase     )

        if isinstance(lowerCamelCase				,	lowerCamelCase     ):
               _UpperCamelCase   :				int       =     [task_inputs]

        if isinstance(lowerCamelCase				,	lowerCamelCase     ) and all(isinstance(lowerCamelCase				,	lowerCamelCase     ) for task_input in task_inputs     ):
               _UpperCamelCase   :				str       =     []
               for task in task_inputs:
                      _UpperCamelCase   :				Optional[Any]       =     f"the task is {task}"
                      task_token_inputs.append(lowerCamelCase     )
               _UpperCamelCase   :				Dict       =     self._preprocess_text(lowerCamelCase				,	max_length=self.task_seq_length     )
        else:
               raise TypeError('Task Inputs should be a string or a list of strings.'     )

        if hasattr(lowerCamelCase				,	'text_inputs'     ):
               _UpperCamelCase   :				str       =     encoded_inputs.text_inputs

               _UpperCamelCase   :				Optional[Any]       =     []
               for texts in texts_list:
                      _UpperCamelCase   :				Optional[int]       =     self._preprocess_text(lowerCamelCase				,	max_length=self.max_seq_length     )
                      text_inputs.append(text_input_list.unsqueeze(0     )     )

               _UpperCamelCase   :				Optional[int]       =     torch.cat(lowerCamelCase				,	dim=0     )

        return encoded_inputs

 def     __snake_case    (	self				:						List[str]				,	lowerCamelCase				:						str=None				,	lowerCamelCase				:						str=None				,	lowerCamelCase				:						int=None				,	**lowerCamelCase				:						List[str]     )       ->							Tuple:

        if task_inputs is None:
               raise ValueError('You have to specify the task_input. Found None.'     )
        elif images is None:
               raise ValueError('You have to specify the image. Found None.'     )

        if not all(task in ['semantic', 'instance', 'panoptic'] for task in task_inputs     ):
               raise ValueError('task_inputs must be semantic, instance, or panoptic.'     )

        _UpperCamelCase   :				Any       =     self.image_processor.encode_inputs(lowerCamelCase				,	lowerCamelCase				,	lowerCamelCase				,	**lowerCamelCase     )

        if isinstance(lowerCamelCase				,	lowerCamelCase     ):
               _UpperCamelCase   :				int       =     [task_inputs]

        if isinstance(lowerCamelCase				,	lowerCamelCase     ) and all(isinstance(lowerCamelCase				,	lowerCamelCase     ) for task_input in task_inputs     ):
               _UpperCamelCase   :				int       =     []
               for task in task_inputs:
                      _UpperCamelCase   :				List[str]       =     f"the task is {task}"
                      task_token_inputs.append(lowerCamelCase     )
               _UpperCamelCase   :				int       =     self._preprocess_text(lowerCamelCase				,	max_length=self.task_seq_length     )
        else:
               raise TypeError('Task Inputs should be a string or a list of strings.'     )

        if hasattr(lowerCamelCase				,	'text_inputs'     ):
               _UpperCamelCase   :				Optional[int]       =     encoded_inputs.text_inputs

               _UpperCamelCase   :				Any       =     []
               for texts in texts_list:
                      _UpperCamelCase   :				Any       =     self._preprocess_text(lowerCamelCase				,	max_length=self.max_seq_length     )
                      text_inputs.append(text_input_list.unsqueeze(0     )     )

               _UpperCamelCase   :				Optional[Any]       =     torch.cat(lowerCamelCase				,	dim=0     )

        return encoded_inputs

 def     __snake_case    (	self				:						Dict				,	*lowerCamelCase				:						Tuple				,	**lowerCamelCase				:						Any     )       ->							str:
        return self.image_processor.post_process_semantic_segmentation(*lowerCamelCase				,	**lowerCamelCase     )

 def     __snake_case    (	self				:						Union[str, Any]				,	*lowerCamelCase				:						Optional[int]				,	**lowerCamelCase				:						Any     )       ->							Any:
        return self.image_processor.post_process_instance_segmentation(*lowerCamelCase				,	**lowerCamelCase     )

 def     __snake_case    (	self				:						Dict				,	*lowerCamelCase				:						List[str]				,	**lowerCamelCase				:						int     )       ->							List[Any]:
        return self.image_processor.post_process_panoptic_segmentation(*lowerCamelCase				,	**lowerCamelCase     )



