




"""simple docstring"""

from typing import Any, Callable, Dict, List, Optional, Union

import PIL.Image
import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionImgaImgPipeline,
    StableDiffusionInpaintPipelineLegacy,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.configuration_utils import FrozenDict
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import deprecate, logging


_lowerCamelCase					:     List[Any]				       =    logging.get_logger(__name__)  # pylint: disable=invalid-name

class     lowercase (      __UpperCAmelCase):
				__lowerCAmelCase      :      Dict					    =				["""safety_checker""", """feature_extractor"""]







				def __init__(							self				:   Tuple ,       _lowerCamelCase				:   AutoencoderKL ,       _lowerCamelCase				:   CLIPTextModel ,       _lowerCamelCase				:   CLIPTokenizer ,       _lowerCamelCase				:   UNetaDConditionModel ,       _lowerCamelCase				:   Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler] ,       _lowerCamelCase				:   StableDiffusionSafetyChecker ,       _lowerCamelCase				:   CLIPImageProcessor ,       _lowerCamelCase				:   bool = True ,       ):



					"""simple docstring"""

					super().__init__()
					if hasattr(scheduler.config ,       '''steps_offset'''    ) and scheduler.config.steps_offset != 1:
						A_					:			int						=						(
						    F"""The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"""
						    F""" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure """
						    '''to update the config accordingly as leaving `steps_offset` might led to incorrect results'''
						    ''' in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,'''
						    ''' it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`'''
						    ''' file'''
						)
						deprecate('''steps_offset!=1''' ,       '''1.0.0''' ,       _lowerCamelCase ,       standard_warn=_lowerCamelCase    )
						A_					:			Dict						=						dict(scheduler.config    )
						A_					:			Union[str, Any]						=						1
						A_					:			Any						=						FrozenDict(_lowerCamelCase    )

					self.register_modules(
					    vae=_lowerCamelCase ,       text_encoder=_lowerCamelCase ,       tokenizer=_lowerCamelCase ,       unet=_lowerCamelCase ,       scheduler=_lowerCamelCase ,       safety_checker=_lowerCamelCase ,       feature_extractor=_lowerCamelCase ,       )
					self.register_to_config(requires_safety_checker=_lowerCamelCase    )







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



					"""simple docstring"""

					return {k: getattr(self ,       _lowerCamelCase    ) for k in self.config.keys() if not k.startswith('''_'''    )}







				def    a_					(							self				:   Tuple ,       _lowerCamelCase				:   Optional[Union[str, int]] = "auto"    ):



					"""simple docstring"""

					if slice_size == "auto":
						# half the attention head size is usually a good trade-off between
						# speed and memory
						A_					:			Optional[Any]						=						self.unet.config.attention_head_dim // 2
					self.unet.set_attention_slice(_lowerCamelCase    )







				def    a_					(							self				:   str    ):



					"""simple docstring"""

					self.enable_attention_slicing(_lowerCamelCase    )







				@torch.no_grad()
				def    a_					(							self				:   List[Any] ,       _lowerCamelCase				:   Union[str, List[str]] ,       _lowerCamelCase				:   Union[torch.FloatTensor, PIL.Image.Image] ,       _lowerCamelCase				:   Union[torch.FloatTensor, PIL.Image.Image] ,       _lowerCamelCase				:   float = 0.8 ,       _lowerCamelCase				:   Optional[int] = 50 ,       _lowerCamelCase				:   Optional[float] = 7.5 ,       _lowerCamelCase				:   Optional[Union[str, List[str]]] = None ,       _lowerCamelCase				:   Optional[int] = 1 ,       _lowerCamelCase				:   Optional[float] = 0.0 ,       _lowerCamelCase				:   Optional[torch.Generator] = None ,       _lowerCamelCase				:   Optional[str] = "pil" ,       _lowerCamelCase				:   bool = True ,       _lowerCamelCase				:   Optional[Callable[[int, int, torch.FloatTensor], None]] = None ,       _lowerCamelCase				:   int = 1 ,       ):



					"""simple docstring"""

					return StableDiffusionInpaintPipelineLegacy(**self.components    )(
					    prompt=_lowerCamelCase ,       image=_lowerCamelCase ,       mask_image=_lowerCamelCase ,       strength=_lowerCamelCase ,       num_inference_steps=_lowerCamelCase ,       guidance_scale=_lowerCamelCase ,       negative_prompt=_lowerCamelCase ,       num_images_per_prompt=_lowerCamelCase ,       eta=_lowerCamelCase ,       generator=_lowerCamelCase ,       output_type=_lowerCamelCase ,       return_dict=_lowerCamelCase ,       callback=_lowerCamelCase ,       )







				@torch.no_grad()
				def    a_					(							self				:   str ,       _lowerCamelCase				:   Union[str, List[str]] ,       _lowerCamelCase				:   Union[torch.FloatTensor, PIL.Image.Image] ,       _lowerCamelCase				:   float = 0.8 ,       _lowerCamelCase				:   Optional[int] = 50 ,       _lowerCamelCase				:   Optional[float] = 7.5 ,       _lowerCamelCase				:   Optional[Union[str, List[str]]] = None ,       _lowerCamelCase				:   Optional[int] = 1 ,       _lowerCamelCase				:   Optional[float] = 0.0 ,       _lowerCamelCase				:   Optional[torch.Generator] = None ,       _lowerCamelCase				:   Optional[str] = "pil" ,       _lowerCamelCase				:   bool = True ,       _lowerCamelCase				:   Optional[Callable[[int, int, torch.FloatTensor], None]] = None ,       _lowerCamelCase				:   int = 1 ,       **_lowerCamelCase				:   str ,       ):



					"""simple docstring"""

					return StableDiffusionImgaImgPipeline(**self.components    )(
					    prompt=_lowerCamelCase ,       image=_lowerCamelCase ,       strength=_lowerCamelCase ,       num_inference_steps=_lowerCamelCase ,       guidance_scale=_lowerCamelCase ,       negative_prompt=_lowerCamelCase ,       num_images_per_prompt=_lowerCamelCase ,       eta=_lowerCamelCase ,       generator=_lowerCamelCase ,       output_type=_lowerCamelCase ,       return_dict=_lowerCamelCase ,       callback=_lowerCamelCase ,       callback_steps=_lowerCamelCase ,       )







				@torch.no_grad()
				def    a_					(							self				:   List[str] ,       _lowerCamelCase				:   Union[str, List[str]] ,       _lowerCamelCase				:   int = 5_12 ,       _lowerCamelCase				:   int = 5_12 ,       _lowerCamelCase				:   int = 50 ,       _lowerCamelCase				:   float = 7.5 ,       _lowerCamelCase				:   Optional[Union[str, List[str]]] = None ,       _lowerCamelCase				:   Optional[int] = 1 ,       _lowerCamelCase				:   float = 0.0 ,       _lowerCamelCase				:   Optional[torch.Generator] = None ,       _lowerCamelCase				:   Optional[torch.FloatTensor] = None ,       _lowerCamelCase				:   Optional[str] = "pil" ,       _lowerCamelCase				:   bool = True ,       _lowerCamelCase				:   Optional[Callable[[int, int, torch.FloatTensor], None]] = None ,       _lowerCamelCase				:   int = 1 ,       ):



					"""simple docstring"""

					return StableDiffusionPipeline(**self.components    )(
					    prompt=_lowerCamelCase ,       height=_lowerCamelCase ,       width=_lowerCamelCase ,       num_inference_steps=_lowerCamelCase ,       guidance_scale=_lowerCamelCase ,       negative_prompt=_lowerCamelCase ,       num_images_per_prompt=_lowerCamelCase ,       eta=_lowerCamelCase ,       generator=_lowerCamelCase ,       latents=_lowerCamelCase ,       output_type=_lowerCamelCase ,       return_dict=_lowerCamelCase ,       callback=_lowerCamelCase ,       callback_steps=_lowerCamelCase ,       )











"""simple docstring"""

import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers.testing_utils import require_vision
from transformers.utils import is_vision_available


if is_vision_available():
		from PIL import Image

		from transformers import AutoProcessor, BertTokenizer, CLIPImageProcessor, GitProcessor, PreTrainedTokenizerFast

@require_vision
class     lowercase (      unittest.TestCase):







				def    a_					(							self				:   Any    ):



					"""simple docstring"""

					A_					:			List[str]						=						tempfile.mkdtemp()

					A_					:			List[str]						=						CLIPImageProcessor()
					A_					:			Any						=						BertTokenizer.from_pretrained(
					    '''hf-internal-testing/tiny-random-BertModel''' ,       model_input_names=['''input_ids''', '''attention_mask''']    )

					A_					:			str						=						GitProcessor(_lowerCamelCase ,       _lowerCamelCase    )

					processor.save_pretrained(self.tmpdirname    )







				def    a_					(							self				:   Tuple ,       **_lowerCamelCase				:   List[str]    ):



					"""simple docstring"""

					return AutoProcessor.from_pretrained(self.tmpdirname ,       **_lowerCamelCase    ).tokenizer







				def    a_					(							self				:   List[Any] ,       **_lowerCamelCase				:   int    ):



					"""simple docstring"""

					return AutoProcessor.from_pretrained(self.tmpdirname ,       **_lowerCamelCase    ).image_processor







				def    a_					(							self				:   Optional[Any]    ):



					"""simple docstring"""

					shutil.rmtree(self.tmpdirname    )







				def    a_					(							self				:   Optional[Any]    ):



					"""simple docstring"""

					A_					:			List[str]						=						[np.random.randint(2_55 ,       size=(3, 30, 4_00) ,       dtype=np.uinta    )]

					A_					:			Any						=						[Image.fromarray(np.moveaxis(_lowerCamelCase ,       0 ,       -1    )    ) for x in image_inputs]

					return image_inputs







				def    a_					(							self				:   str    ):



					"""simple docstring"""

					A_					:			Union[str, Any]						=						GitProcessor(tokenizer=self.get_tokenizer() ,       image_processor=self.get_image_processor()    )
					processor.save_pretrained(self.tmpdirname    )

					A_					:			Union[str, Any]						=						self.get_tokenizer(bos_token='''(BOS)''' ,       eos_token='''(EOS)'''    )
					A_					:			List[Any]						=						self.get_image_processor(do_normalize=_lowerCamelCase ,       padding_value=1.0    )

					A_					:			List[str]						=						GitProcessor.from_pretrained(
					    self.tmpdirname ,       bos_token='''(BOS)''' ,       eos_token='''(EOS)''' ,       do_normalize=_lowerCamelCase ,       padding_value=1.0    )

					self.assertEqual(processor.tokenizer.get_vocab() ,       tokenizer_add_kwargs.get_vocab()    )
					self.assertIsInstance(processor.tokenizer ,       _lowerCamelCase    )

					self.assertEqual(processor.image_processor.to_json_string() ,       image_processor_add_kwargs.to_json_string()    )
					self.assertIsInstance(processor.image_processor ,       _lowerCamelCase    )







				def    a_					(							self				:   Optional[Any]    ):



					"""simple docstring"""

					A_					:			Tuple						=						self.get_image_processor()
					A_					:			Optional[Any]						=						self.get_tokenizer()

					A_					:			int						=						GitProcessor(tokenizer=_lowerCamelCase ,       image_processor=_lowerCamelCase    )

					A_					:			Tuple						=						self.prepare_image_inputs()

					A_					:			Any						=						image_processor(_lowerCamelCase ,       return_tensors='''np'''    )
					A_					:			Optional[Any]						=						processor(images=_lowerCamelCase ,       return_tensors='''np'''    )

					for key in input_feat_extract.keys():
						self.assertAlmostEqual(input_feat_extract[key].sum() ,       input_processor[key].sum() ,       delta=1E-2    )







				def    a_					(							self				:   Optional[int]    ):



					"""simple docstring"""

					A_					:			Union[str, Any]						=						self.get_image_processor()
					A_					:			Union[str, Any]						=						self.get_tokenizer()

					A_					:			int						=						GitProcessor(tokenizer=_lowerCamelCase ,       image_processor=_lowerCamelCase    )

					A_					:			List[str]						=						'''lower newer'''

					A_					:			Tuple						=						processor(text=_lowerCamelCase    )

					A_					:			List[str]						=						tokenizer(_lowerCamelCase ,       return_token_type_ids=_lowerCamelCase    )

					for key in encoded_tok.keys():
						self.assertListEqual(encoded_tok[key] ,       encoded_processor[key]    )







				def    a_					(							self				:   Tuple    ):



					"""simple docstring"""

					A_					:			Optional[int]						=						self.get_image_processor()
					A_					:			Optional[Any]						=						self.get_tokenizer()

					A_					:			Optional[Any]						=						GitProcessor(tokenizer=_lowerCamelCase ,       image_processor=_lowerCamelCase    )

					A_					:			Tuple						=						'''lower newer'''
					A_					:			str						=						self.prepare_image_inputs()

					A_					:			Optional[int]						=						processor(text=_lowerCamelCase ,       images=_lowerCamelCase    )

					self.assertListEqual(list(inputs.keys()    ) ,       ['''input_ids''', '''attention_mask''', '''pixel_values''']    )

					# test if it raises when no input is passed
					with pytest.raises(_lowerCamelCase    ):
						processor()







				def    a_					(							self				:   Optional[int]    ):



					"""simple docstring"""

					A_					:			Optional[int]						=						self.get_image_processor()
					A_					:			List[Any]						=						self.get_tokenizer()

					A_					:			Optional[Any]						=						GitProcessor(tokenizer=_lowerCamelCase ,       image_processor=_lowerCamelCase    )

					A_					:			Dict						=						[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

					A_					:			str						=						processor.batch_decode(_lowerCamelCase    )
					A_					:			int						=						tokenizer.batch_decode(_lowerCamelCase    )

					self.assertListEqual(_lowerCamelCase ,       _lowerCamelCase    )







				def    a_					(							self				:   Any    ):



					"""simple docstring"""

					A_					:			Dict						=						self.get_image_processor()
					A_					:			List[str]						=						self.get_tokenizer()

					A_					:			Any						=						GitProcessor(tokenizer=_lowerCamelCase ,       image_processor=_lowerCamelCase    )

					A_					:			int						=						'''lower newer'''
					A_					:			Union[str, Any]						=						self.prepare_image_inputs()

					A_					:			List[str]						=						processor(text=_lowerCamelCase ,       images=_lowerCamelCase    )

					# For now the processor supports only ['input_ids', 'attention_mask', 'pixel_values']
					self.assertListEqual(list(inputs.keys()    ) ,       ['''input_ids''', '''attention_mask''', '''pixel_values''']    )





