


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


UpperCAmelCase_   :				int						      = logging.get_logger(__name__)  # pylint: disable=invalid-name
class 				SCREAMING_SNAKE_CASE__						(       lowercase__    ):
      def __init__( self       : Optional[int]  ,							SCREAMING_SNAKE_CASE__       : AutoencoderKL  ,							SCREAMING_SNAKE_CASE__       : CLIPTextModel  ,							SCREAMING_SNAKE_CASE__       : CLIPTokenizer  ,							SCREAMING_SNAKE_CASE__       : UNetaDConditionModel  ,							SCREAMING_SNAKE_CASE__       : Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]  ,							SCREAMING_SNAKE_CASE__       : StableDiffusionSafetyChecker  ,							SCREAMING_SNAKE_CASE__       : CLIPImageProcessor  ,							)	->    Optional[Any]:
             super().__init__()
             self.register_modules(
                 vae=SCREAMING_SNAKE_CASE__  ,							text_encoder=SCREAMING_SNAKE_CASE__  ,							tokenizer=SCREAMING_SNAKE_CASE__  ,							unet=SCREAMING_SNAKE_CASE__  ,							scheduler=SCREAMING_SNAKE_CASE__  ,							safety_checker=SCREAMING_SNAKE_CASE__  ,							feature_extractor=SCREAMING_SNAKE_CASE__  ,							)
      def  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : Optional[Union[str, int]] = "auto"  )	->    str:
             if slice_size == "auto":
                    # half the attention head size is usually a good trade-off between
                    # speed and memory
                    a_      :	str							=						self.unet.config.attention_head_dim // 2
             self.unet.set_attention_slice(SCREAMING_SNAKE_CASE__  )
      def  SCREAMING_SNAKE_CASE							( self       : List[Any]  )	->    str:
             self.enable_attention_slicing(SCREAMING_SNAKE_CASE__  )

      @torch.no_grad()
      def __call__( self       : List[Any]  ,							SCREAMING_SNAKE_CASE__       : Union[str, List[str]]  ,							SCREAMING_SNAKE_CASE__       : int = 5_1_2  ,							SCREAMING_SNAKE_CASE__       : int = 5_1_2  ,							SCREAMING_SNAKE_CASE__       : int = 5_0  ,							SCREAMING_SNAKE_CASE__       : float = 7.5  ,							SCREAMING_SNAKE_CASE__       : Optional[Union[str, List[str]]] = None  ,							SCREAMING_SNAKE_CASE__       : Optional[int] = 1  ,							SCREAMING_SNAKE_CASE__       : float = 0.0  ,							SCREAMING_SNAKE_CASE__       : Optional[torch.Generator] = None  ,							SCREAMING_SNAKE_CASE__       : Optional[torch.FloatTensor] = None  ,							SCREAMING_SNAKE_CASE__       : Optional[str] = "pil"  ,							SCREAMING_SNAKE_CASE__       : bool = True  ,							SCREAMING_SNAKE_CASE__       : Optional[Callable[[int, int, torch.FloatTensor], None]] = None  ,							SCREAMING_SNAKE_CASE__       : int = 1  ,							SCREAMING_SNAKE_CASE__       : Optional[torch.FloatTensor] = None  ,							**SCREAMING_SNAKE_CASE__       : List[str]  ,							)	->    List[str]:

             if isinstance(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ):
                    a_      :	Tuple							=						1
             elif isinstance(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ):
                    a_      :	Optional[int]							=						len(SCREAMING_SNAKE_CASE__  )
             else:
                    raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(SCREAMING_SNAKE_CASE__  )}"""  )

             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(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ) or callback_steps <= 0)
             ):
                    raise ValueError(
                        F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
                        F""" {type(SCREAMING_SNAKE_CASE__  )}."""  )

             # get prompt text embeddings
             a_      :	int							=						self.tokenizer(
                 SCREAMING_SNAKE_CASE__  ,							padding='max_length'  ,							max_length=self.tokenizer.model_max_length  ,							return_tensors='pt'  ,							)
             a_      :	Tuple							=						text_inputs.input_ids

             if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                    a_      :	Any							=						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}"""  )
                    a_      :	Dict							=						text_input_ids[:, : self.tokenizer.model_max_length]

             if text_embeddings is None:
                    a_      :	List[str]							=						self.text_encoder(text_input_ids.to(self.device  )  )[0]

             # duplicate text embeddings for each generation per prompt, using mps friendly method
             a_							,	a_							,	a_      :	Any							=						text_embeddings.shape
             a_      :	Optional[Any]							=						text_embeddings.repeat(1  ,							SCREAMING_SNAKE_CASE__  ,							1  )
             a_      :	Tuple							=						text_embeddings.view(bs_embed * num_images_per_prompt  ,							SCREAMING_SNAKE_CASE__  ,							-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.
             a_      :	Union[str, Any]							=						guidance_scale > 1.0
             # get unconditional embeddings for classifier free guidance
             if do_classifier_free_guidance:
                    a_      :	List[str]
                    if negative_prompt is None:
                           a_      :	Tuple							=						['']
                    elif type(SCREAMING_SNAKE_CASE__  ) is not type(SCREAMING_SNAKE_CASE__  ):
                           raise TypeError(
                               F"""`negative_prompt` should be the same type to `prompt`, but got {type(SCREAMING_SNAKE_CASE__  )} !="""
                               F""" {type(SCREAMING_SNAKE_CASE__  )}."""  )
                    elif isinstance(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ):
                           a_      :	List[Any]							=						[negative_prompt]
                    elif batch_size != len(SCREAMING_SNAKE_CASE__  ):
                           raise ValueError(
                               F"""`negative_prompt`: {negative_prompt} has batch size {len(SCREAMING_SNAKE_CASE__  )}, but `prompt`:"""
                               F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                               ' the batch size of `prompt`.'  )
                    else:
                           a_      :	int							=						negative_prompt

                    a_      :	Tuple							=						text_input_ids.shape[-1]
                    a_      :	str							=						self.tokenizer(
                        SCREAMING_SNAKE_CASE__  ,							padding='max_length'  ,							max_length=SCREAMING_SNAKE_CASE__  ,							truncation=SCREAMING_SNAKE_CASE__  ,							return_tensors='pt'  ,							)
                    a_      :	Union[str, Any]							=						self.text_encoder(uncond_input.input_ids.to(self.device  )  )[0]

                    # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
                    a_      :	Dict							=						uncond_embeddings.shape[1]
                    a_      :	Tuple							=						uncond_embeddings.repeat(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ,							1  )
                    a_      :	List[str]							=						uncond_embeddings.view(batch_size * num_images_per_prompt  ,							SCREAMING_SNAKE_CASE__  ,							-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
                    a_      :	List[str]							=						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`.
             a_      :	Tuple							=						(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
             a_      :	Any							=						(batch_size * num_images_per_prompt, self.unet.config.in_channels, 6_4, 6_4)
             a_      :	Dict							=						text_embeddings.dtype
             if latents is None:
                    if self.device.type == "mps":
                           # randn does not exist on mps
                           a_      :	Union[str, Any]							=						torch.randn(
                               SCREAMING_SNAKE_CASE__  ,							generator=SCREAMING_SNAKE_CASE__  ,							device='cpu'  ,							dtype=SCREAMING_SNAKE_CASE__  ).to(self.device  )
                           a_      :	Any							=						torch.randn(SCREAMING_SNAKE_CASE__  ,							generator=SCREAMING_SNAKE_CASE__  ,							device='cpu'  ,							dtype=SCREAMING_SNAKE_CASE__  ).to(
                               self.device  )
                    else:
                           a_      :	Tuple							=						torch.randn(
                               SCREAMING_SNAKE_CASE__  ,							generator=SCREAMING_SNAKE_CASE__  ,							device=self.device  ,							dtype=SCREAMING_SNAKE_CASE__  )
                           a_      :	List[Any]							=						torch.randn(SCREAMING_SNAKE_CASE__  ,							generator=SCREAMING_SNAKE_CASE__  ,							device=self.device  ,							dtype=SCREAMING_SNAKE_CASE__  )
             else:
                    if latents_reference.shape != latents_shape:
                           raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""  )
                    a_      :	str							=						latents_reference.to(self.device  )
                    a_      :	Optional[Any]							=						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
             a_      :	Union[str, Any]							=						(latents_shape[3] - latents_shape_reference[3]) // 2
             a_      :	Tuple							=						(latents_shape[2] - latents_shape_reference[2]) // 2
             a_      :	Any							=						latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
             a_      :	Union[str, Any]							=						latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
             a_      :	List[str]							=						0 if dx < 0 else dx
             a_      :	Any							=						0 if dy < 0 else dy
             a_      :	List[Any]							=						max(-dx  ,							0  )
             a_      :	Any							=						max(-dy  ,							0  )
             # import pdb
             # pdb.set_trace()
             a_      :	Tuple							=						latents_reference[:, :, dy : dy + h, dx : dx + w]

             # set timesteps
             self.scheduler.set_timesteps(SCREAMING_SNAKE_CASE__  )

             # Some schedulers like PNDM have timesteps as arrays
             # It's more optimized to move all timesteps to correct device beforehand
             a_      :	str							=						self.scheduler.timesteps.to(self.device  )

             # scale the initial noise by the standard deviation required by the scheduler
             a_      :	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]
             a_      :	Optional[Any]							=						'eta' in set(inspect.signature(self.scheduler.step  ).parameters.keys()  )
             a_      :	Tuple							=						{}
             if accepts_eta:
                    a_      :	Optional[int]							=						eta

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

                    # predict the noise residual
                    a_      :	str							=						self.unet(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ,							encoder_hidden_states=SCREAMING_SNAKE_CASE__  ).sample

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

                    # compute the previous noisy sample x_t -> x_t-1
                    a_      :	Union[str, Any]							=						self.scheduler.step(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ,							**SCREAMING_SNAKE_CASE__  ).prev_sample

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

             a_      :	Dict							=						1 / 0.18215 * latents
             a_      :	Optional[int]							=						self.vae.decode(SCREAMING_SNAKE_CASE__  ).sample

             a_      :	int							=						(image / 2 + 0.5).clamp(0  ,							1  )

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

             if self.safety_checker is not None:
                    a_      :	List[Any]							=						self.feature_extractor(self.numpy_to_pil(SCREAMING_SNAKE_CASE__  )  ,							return_tensors='pt'  ).to(
                        self.device  )
                    a_							,	a_      :	List[str]							=						self.safety_checker(
                        images=SCREAMING_SNAKE_CASE__  ,							clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype  )  )
             else:
                    a_      :	Any							=						None

             if output_type == "pil":
                    a_      :	Tuple							=						self.numpy_to_pil(SCREAMING_SNAKE_CASE__  )

             if not return_dict:
                    return (image, has_nsfw_concept)

             return StableDiffusionPipelineOutput(images=SCREAMING_SNAKE_CASE__  ,							nsfw_content_detected=SCREAMING_SNAKE_CASE__  )






from argparse import ArgumentParser

from .env import EnvironmentCommand







def  SCREAMING_SNAKE_CASE_		(    )			->	str:

       """simple docstring"""

       a_      :	str							=						ArgumentParser('Diffusers CLI tool'						,					usage='diffusers-cli <command> [<args>]'       )
       a_      :	Any							=						parser.add_subparsers(help='diffusers-cli command helpers'       )

       # Register commands
       EnvironmentCommand.register_subcommand(__A       )

       # Let's go
       a_      :	int							=						parser.parse_args()

       if not hasattr(__A						,					'func'       ):
              parser.print_help()
              exit(1       )

       # Run
       a_      :	Union[str, Any]							=						args.func(__A       )
       service.run()


if __name__ == "__main__":
   main()


