





'''simple docstring'''


import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


lowerCAmelCase :					List[Any] 							=logging.get_logger(__name__)  # pylint: disable=invalid-name
class 				a_       (    _lowerCAmelCase							):





     def __init__(			self		: Optional[Any]     ,				lowercase		: AutoencoderKL     ,				lowercase		: CLIPTextModel     ,				lowercase		: CLIPTokenizer     ,				lowercase		: UNetaDConditionModel     ,				lowercase		: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]     ,				lowercase		: StableDiffusionSafetyChecker     ,				lowercase		: CLIPImageProcessor     ,				):





        """simple docstring"""




        super().__init__()
        self.register_modules(
            vae=lowercase     ,				text_encoder=lowercase     ,				tokenizer=lowercase     ,				unet=lowercase     ,				scheduler=lowercase     ,				safety_checker=lowercase     ,				feature_extractor=lowercase     ,				)





     def 							lowercase__				(			self		: Tuple     ,				lowercase		: 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
           lowercase_      :Optional[int]         =			self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(lowercase						)





     def 							lowercase__				(			self		: int						):





        """simple docstring"""




        self.enable_attention_slicing(lowercase						)







     @torch.no_grad()
     def __call__(			self		: Union[str, Any]     ,				lowercase		: Union[str, List[str]]     ,				lowercase		: int = 512     ,				lowercase		: int = 512     ,				lowercase		: int = 50     ,				lowercase		: float = 7.5     ,				lowercase		: Optional[Union[str, List[str]]] = None     ,				lowercase		: Optional[int] = 1     ,				lowercase		: float = 0.0     ,				lowercase		: Optional[torch.Generator] = None     ,				lowercase		: Optional[torch.FloatTensor] = None     ,				lowercase		: Optional[str] = "pil"     ,				lowercase		: bool = True     ,				lowercase		: Optional[Callable[[int, int, torch.FloatTensor], None]] = None     ,				lowercase		: int = 1     ,				lowercase		: Optional[torch.FloatTensor] = None     ,				**lowercase		: int     ,				):





        """simple docstring"""




        if isinstance(lowercase     ,				lowercase						):
           lowercase_      :Tuple         =			1
        elif isinstance(lowercase     ,				lowercase						):
           lowercase_      :str         =			len(lowercase						)
        else:
           raise ValueError(F'`prompt` has to be of type `str` or `list` but is {type(lowercase						)}'						)

        if height % 8 != 0 or width % 8 != 0:
           raise ValueError(F'`height` and `width` have to be divisible by 8 but are {height} and {width}.'						)

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(lowercase     ,				lowercase						) or callback_steps <= 0)
        ):
           raise ValueError(
               F'`callback_steps` has to be a positive integer but is {callback_steps} of type'
               F' {type(lowercase						)}.'						)

        # get prompt text embeddings
        lowercase_      :int         =			self.tokenizer(
            lowercase     ,				padding="max_length"     ,				max_length=self.tokenizer.model_max_length     ,				return_tensors="pt"     ,				)
        lowercase_      :Optional[int]         =			text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
           lowercase_      :Dict         =			self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]						)
           logger.warning(
               "The following part of your input was truncated because CLIP can only handle sequences up to"
               F' {self.tokenizer.model_max_length} tokens: {removed_text}'						)
           lowercase_      :List[Any]         =			text_input_ids[:, : self.tokenizer.model_max_length]

        if text_embeddings is None:
           lowercase_      :Optional[int]         =			self.text_encoder(text_input_ids.to(self.device						)						)[0]

        # duplicate text embeddings for each generation per prompt, using mps friendly method
        lowercase_      ,							lowercase_      ,							lowercase_      :Optional[int]         =			text_embeddings.shape
        lowercase_      :str         =			text_embeddings.repeat(1     ,				lowercase     ,				1						)
        lowercase_      :Union[str, Any]         =			text_embeddings.view(bs_embed * num_images_per_prompt     ,				lowercase     ,				-1						)

        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
        # corresponds to doing no classifier free guidance.
        lowercase_      :int         =			guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
           lowercase_      :List[str]
           if negative_prompt is None:
              lowercase_      :Tuple         =			[""]
           elif type(lowercase						) is not type(lowercase						):
              raise TypeError(
                  F'`negative_prompt` should be the same type to `prompt`, but got {type(lowercase						)} !='
                  F' {type(lowercase						)}.'						)
           elif isinstance(lowercase     ,				lowercase						):
              lowercase_      :Optional[Any]         =			[negative_prompt]
           elif batch_size != len(lowercase						):
              raise ValueError(
                  F'`negative_prompt`: {negative_prompt} has batch size {len(lowercase						)}, but `prompt`:'
                  F' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
                  " the batch size of `prompt`."						)
           else:
              lowercase_      :Union[str, Any]         =			negative_prompt

           lowercase_      :Optional[int]         =			text_input_ids.shape[-1]
           lowercase_      :Any         =			self.tokenizer(
               lowercase     ,				padding="max_length"     ,				max_length=lowercase     ,				truncation=lowercase     ,				return_tensors="pt"     ,				)
           lowercase_      :Tuple         =			self.text_encoder(uncond_input.input_ids.to(self.device						)						)[0]

           # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
           lowercase_      :List[Any]         =			uncond_embeddings.shape[1]
           lowercase_      :str         =			uncond_embeddings.repeat(lowercase     ,				lowercase     ,				1						)
           lowercase_      :Optional[Any]         =			uncond_embeddings.view(batch_size * num_images_per_prompt     ,				lowercase     ,				-1						)

           # For classifier free guidance, we need to do two forward passes.
           # Here we concatenate the unconditional and text embeddings into a single batch
           # to avoid doing two forward passes
           lowercase_      :Dict         =			torch.cat([uncond_embeddings, text_embeddings]						)

        # get the initial random noise unless the user supplied it

        # Unlike in other pipelines, latents need to be generated in the target device
        # for 1-to-1 results reproducibility with the CompVis implementation.
        # However this currently doesn't work in `mps`.
        lowercase_      :Union[str, Any]         =			(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        lowercase_      :Optional[Any]         =			(batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
        lowercase_      :int         =			text_embeddings.dtype
        if latents is None:
           if self.device.type == "mps":
              # randn does not exist on mps
              lowercase_      :Optional[Any]         =			torch.randn(
                  lowercase     ,				generator=lowercase     ,				device="cpu"     ,				dtype=lowercase						).to(self.device						)
              lowercase_      :Union[str, Any]         =			torch.randn(lowercase     ,				generator=lowercase     ,				device="cpu"     ,				dtype=lowercase						).to(
                  self.device						)
           else:
              lowercase_      :Any         =			torch.randn(
                  lowercase     ,				generator=lowercase     ,				device=self.device     ,				dtype=lowercase						)
              lowercase_      :int         =			torch.randn(lowercase     ,				generator=lowercase     ,				device=self.device     ,				dtype=lowercase						)
        else:
           if latents_reference.shape != latents_shape:
              raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'						)
           lowercase_      :Optional[int]         =			latents_reference.to(self.device						)
           lowercase_      :int         =			latents.to(self.device						)

        # This is the key part of the pipeline where we
        # try to ensure that the generated images w/ the same seed
        # but different sizes actually result in similar images
        lowercase_      :List[str]         =			(latents_shape[3] - latents_shape_reference[3]) // 2
        lowercase_      :List[Any]         =			(latents_shape[2] - latents_shape_reference[2]) // 2
        lowercase_      :str         =			latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
        lowercase_      :str         =			latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
        lowercase_      :Dict         =			0 if dx < 0 else dx
        lowercase_      :str         =			0 if dy < 0 else dy
        lowercase_      :Optional[Any]         =			max(-dx     ,				0						)
        lowercase_      :Tuple         =			max(-dy     ,				0						)
        # import pdb
        # pdb.set_trace()
        lowercase_      :List[str]         =			latents_reference[:, :, dy : dy + h, dx : dx + w]

        # set timesteps
        self.scheduler.set_timesteps(lowercase						)

        # Some schedulers like PNDM have timesteps as arrays
        # It's more optimized to move all timesteps to correct device beforehand
        lowercase_      :List[Any]         =			self.scheduler.timesteps.to(self.device						)

        # scale the initial noise by the standard deviation required by the scheduler
        lowercase_      :Dict         =			latents * self.scheduler.init_noise_sigma

        # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
        # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
        # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
        # and should be between [0, 1]
        lowercase_      :Tuple         =			"eta" in set(inspect.signature(self.scheduler.step						).parameters.keys()						)
        lowercase_      :Optional[int]         =			{}
        if accepts_eta:
           lowercase_      :Tuple         =			eta

        for i, t in enumerate(self.progress_bar(lowercase						)						):
           # expand the latents if we are doing classifier free guidance
           lowercase_      :int         =			torch.cat([latents] * 2						) if do_classifier_free_guidance else latents
           lowercase_      :Any         =			self.scheduler.scale_model_input(lowercase     ,				lowercase						)

           # predict the noise residual
           lowercase_      :Tuple         =			self.unet(lowercase     ,				lowercase     ,				encoder_hidden_states=lowercase						).sample

           # perform guidance
           if do_classifier_free_guidance:
              lowercase_      ,							lowercase_      :Union[str, Any]         =			noise_pred.chunk(2						)
              lowercase_      :List[str]         =			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

           # compute the previous noisy sample x_t -> x_t-1
           lowercase_      :Tuple         =			self.scheduler.step(lowercase     ,				lowercase     ,				lowercase     ,				**lowercase						).prev_sample

           # call the callback, if provided
           if callback is not None and i % callback_steps == 0:
              callback(lowercase     ,				lowercase     ,				lowercase						)

        lowercase_      :str         =			1 / 0.1_82_15 * latents
        lowercase_      :Dict         =			self.vae.decode(lowercase						).sample

        lowercase_      :Tuple         =			(image / 2 + 0.5).clamp(0     ,				1						)

        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
        lowercase_      :Tuple         =			image.cpu().permute(0     ,				2     ,				3     ,				1						).float().numpy()

        if self.safety_checker is not None:
           lowercase_      :int         =			self.feature_extractor(self.numpy_to_pil(lowercase						)     ,				return_tensors="pt"						).to(
               self.device						)
           lowercase_      ,							lowercase_      :Union[str, Any]         =			self.safety_checker(
               images=lowercase     ,				clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype						)						)
        else:
           lowercase_      :Tuple         =			None

        if output_type == "pil":
           lowercase_      :List[Any]         =			self.numpy_to_pil(lowercase						)

        if not return_dict:
           return (image, has_nsfw_concept)

        return StableDiffusionPipelineOutput(images=lowercase     ,				nsfw_content_detected=lowercase						)









'''simple docstring'''


import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType
class 				a_       (    _lowerCAmelCase							):
     __A								=      ["image_processor", "tokenizer"]
     __A								=      "LayoutLMv3ImageProcessor"
     __A								=      ("LayoutLMv3Tokenizer", "LayoutLMv3TokenizerFast")





     def __init__(			self		: int     ,				lowercase		: Optional[Any]=None     ,				lowercase		: List[str]=None     ,				**lowercase		: Optional[int]						):





        """simple docstring"""




        lowercase_      :int         =			None
        if "feature_extractor" in kwargs:
           warnings.warn(
               "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
               " instead."     ,				lowercase     ,				)
           lowercase_      :Optional[int]         =			kwargs.pop("feature_extractor"						)

        lowercase_      :Union[str, Any]         =			image_processor if image_processor is not None else feature_extractor
        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`."						)

        super().__init__(lowercase     ,				lowercase						)





     def __call__(			self		: Optional[Any]     ,				lowercase		: List[str]     ,				lowercase		: Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None     ,				lowercase		: Optional[Union[PreTokenizedInput, List[PreTokenizedInput]]] = None     ,				lowercase		: Union[List[List[int]], List[List[List[int]]]] = None     ,				lowercase		: Optional[Union[List[int], List[List[int]]]] = None     ,				lowercase		: bool = True     ,				lowercase		: Union[bool, str, PaddingStrategy] = False     ,				lowercase		: Union[bool, str, TruncationStrategy] = None     ,				lowercase		: Optional[int] = None     ,				lowercase		: int = 0     ,				lowercase		: Optional[int] = None     ,				lowercase		: Optional[bool] = None     ,				lowercase		: Optional[bool] = None     ,				lowercase		: bool = False     ,				lowercase		: bool = False     ,				lowercase		: bool = False     ,				lowercase		: bool = False     ,				lowercase		: bool = True     ,				lowercase		: Optional[Union[str, TensorType]] = None     ,				**lowercase		: List[Any]     ,				):





        """simple docstring"""




        if self.image_processor.apply_ocr and (boxes is not None):
           raise ValueError(
               "You cannot provide bounding boxes if you initialized the image processor with apply_ocr set to True."						)

        if self.image_processor.apply_ocr and (word_labels is not None):
           raise ValueError(
               "You cannot provide word labels if you initialized the image processor with apply_ocr set to True."						)

        # first, apply the image processor
        lowercase_      :Dict         =			self.image_processor(images=lowercase     ,				return_tensors=lowercase						)

        # second, apply the tokenizer
        if text is not None and self.image_processor.apply_ocr and text_pair is None:
           if isinstance(lowercase     ,				lowercase						):
              lowercase_      :str         =			[text]  # add batch dimension (as the image processor always adds a batch dimension)
           lowercase_      :Union[str, Any]         =			features["words"]

        lowercase_      :Optional[Any]         =			self.tokenizer(
            text=text if text is not None else features["words"]     ,				text_pair=text_pair if text_pair is not None else None     ,				boxes=boxes if boxes is not None else features["boxes"]     ,				word_labels=lowercase     ,				add_special_tokens=lowercase     ,				padding=lowercase     ,				truncation=lowercase     ,				max_length=lowercase     ,				stride=lowercase     ,				pad_to_multiple_of=lowercase     ,				return_token_type_ids=lowercase     ,				return_attention_mask=lowercase     ,				return_overflowing_tokens=lowercase     ,				return_special_tokens_mask=lowercase     ,				return_offsets_mapping=lowercase     ,				return_length=lowercase     ,				verbose=lowercase     ,				return_tensors=lowercase     ,				**lowercase     ,				)

        # add pixel values
        lowercase_      :Any         =			features.pop("pixel_values"						)
        if return_overflowing_tokens is True:
           lowercase_      :Any         =			self.get_overflowing_images(lowercase     ,				encoded_inputs["overflow_to_sample_mapping"]						)
        lowercase_      :Any         =			images

        return encoded_inputs





     def 							lowercase__				(			self		: List[Any]     ,				lowercase		: Any     ,				lowercase		: Optional[Any]						):





        """simple docstring"""




        lowercase_      :Union[str, Any]         =			[]
        for sample_idx in overflow_to_sample_mapping:
           images_with_overflow.append(images[sample_idx]						)

        if len(lowercase						) != len(lowercase						):
           raise ValueError(
               "Expected length of images to be the same as the length of `overflow_to_sample_mapping`, but got"
               F' {len(lowercase						)} and {len(lowercase						)}'						)

        return images_with_overflow





     def 							lowercase__				(			self		: Union[str, Any]     ,				*lowercase		: List[Any]     ,				**lowercase		: Optional[Any]						):





        """simple docstring"""




        return self.tokenizer.batch_decode(*lowercase     ,				**lowercase						)





     def 							lowercase__				(			self		: List[Any]     ,				*lowercase		: Any     ,				**lowercase		: str						):





        """simple docstring"""




        return self.tokenizer.decode(*lowercase     ,				**lowercase						)





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





        """simple docstring"""




        return ["input_ids", "bbox", "attention_mask", "pixel_values"]





     @property
     def 							lowercase__				(			self		: str						):





        """simple docstring"""




        warnings.warn(
            "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."     ,				lowercase     ,				)
        return self.image_processor_class







     @property
     def 							lowercase__				(			self		: Tuple						):





        """simple docstring"""




        warnings.warn(
            "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."     ,				lowercase     ,				)
        return self.image_processor


