


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_      : Tuple  					=       logging.get_logger(__name__)  # pylint: disable=invalid-name
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      def __init__(       self				:       Any     ,							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     ,							)		->			List[str]:
             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				:       int     ,							SCREAMING_SNAKE_CASE__				:       Optional[Union[str, int]] = "auto"     )		->			Optional[int]:
             if slice_size == "auto":
                    # half the attention head size is usually a good trade-off between
                    # speed and memory
                    a_					:		List[Any]							=  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[str]     ,							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__				:       Optional[Any]     ,							)		->			List[str]:

             if isinstance(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ):
                    a_					:		Optional[int]							=  1
             elif isinstance(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ):
                    a_					:		str							=  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_					:		List[Any]							=  self.tokenizer(
                 SCREAMING_SNAKE_CASE__     ,							padding='max_length'     ,							max_length=self.tokenizer.model_max_length     ,							return_tensors='pt'     ,							)
             a_					:		List[str]							=  text_inputs.input_ids

             if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                    a_					:		Tuple							=  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_					:		Any							=  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[int]							=  text_embeddings.repeat(1     ,							SCREAMING_SNAKE_CASE__     ,							1     )
             a_					:		Union[str, Any]							=  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_					:		List[str]							=  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_					:		str							=  ['']
                    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_					:		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_					:		List[Any]							=  negative_prompt

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

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

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

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

                    # predict the noise residual
                    a_					:		Optional[int]							=  self.unet(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ,							encoder_hidden_states=SCREAMING_SNAKE_CASE__     ).sample

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

                    # compute the previous noisy sample x_t -> x_t-1
                    a_					:		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_					:		Dict							=  self.vae.decode(SCREAMING_SNAKE_CASE__     ).sample

             a_					:		str							=  (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_					:		Union[str, Any]							=  image.cpu().permute(0     ,							2     ,							3     ,							1     ).float().numpy()

             if self.safety_checker is not None:
                    a_					:		int							=  self.feature_extractor(self.numpy_to_pil(SCREAMING_SNAKE_CASE__     )     ,							return_tensors='pt'     ).to(
                        self.device     )
                    a_							,		a_					:		str							=  self.safety_checker(
                        images=SCREAMING_SNAKE_CASE__     ,							clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype     )     )
             else:
                    a_					:		List[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__     )






UpperCAmelCase_      : Optional[int]  					=       [4, 1, 7, 4, 2, 6, 4, 1, 5, 3, 7, 5]
UpperCAmelCase_      : str  					=       [3, 7, 7, 4, 2, 6, 4, 1, 5, 3, 7, 5]
UpperCAmelCase_      : str  					=       {
    0: 'Sunday',
    1: 'Monday',
    2: 'Tuesday',
    3: 'Wednesday',
    4: 'Thursday',
    5: 'Friday',
    6: 'Saturday',
}







def        SCREAMING_SNAKE_CASE_						(			__A :			int  ,    __A :			int  ,    __A :			int				)	->		str:

       """simple docstring"""

       assert len(str(__A				)				) > 2, "year should be in YYYY format"
       assert 1 <= month <= 12, "month should be between 1 to 12"
       assert 1 <= day <= 31, "day should be between 1 to 31"

       # Doomsday algorithm:
       a_					:		List[str]							=  year // 1_00
       a_					:		Optional[int]							=  (5 * (century % 4) + 2) % 7
       a_					:		List[str]							=  year % 1_00
       a_					:		str							=  centurian % 12
       a_					:		List[str]							=  (
           (centurian // 12) + centurian_m + (centurian_m // 4) + century_anchor
       ) % 7
       a_					:		Any							=  (
           DOOMSDAY_NOT_LEAP[month - 1]
           if (year % 4 != 0) or (centurian == 0 and (year % 4_00) == 0)
           else DOOMSDAY_LEAP[month - 1]
       )
       a_					:		Any							=  (dooms_day + day - day_anchor) % 7
       return WEEK_DAY_NAMES[week_day]


if __name__ == "__main__":
   import doctest

   doctest.testmod()


