





'''simple docstring'''


from logging import getLogger
from typing import Any, Callable, List, Optional, Union

import numpy as np
import PIL
import torch

from ...schedulers import DDPMScheduler
from ..onnx_utils import ORT_TO_NP_TYPE, OnnxRuntimeModel
from ..pipeline_utils import ImagePipelineOutput
from . import StableDiffusionUpscalePipeline


lowerCAmelCase      :  Optional[int]      							=getLogger(__name__)


lowerCAmelCase      :  Optional[Any]      							=4
lowerCAmelCase      :  List[Any]      							=7

lowerCAmelCase      :  List[Any]      							={
    '''float16''': torch.floataa,
    '''float32''': torch.floataa,
}
def 							UpperCAmelCase_						(		__lowerCamelCase      :      Any				):
   if isinstance(__lowerCamelCase			,torch.Tensor				):
      return image
   elif isinstance(__lowerCamelCase			,PIL.Image.Image				):
      lowercase_    :Dict					    =				[image]

   if isinstance(image[0]			,PIL.Image.Image				):
      lowercase_    ,							lowercase_    :Tuple					    =				image[0].size
      lowercase_    ,							lowercase_    :List[Any]					    =				(x - x % 64 for x in (w, h))  # resize to integer multiple of 32

      lowercase_    :int					    =				[np.array(i.resize((w, h)				)				)[None, :] for i in image]
      lowercase_    :Optional[int]					    =				np.concatenate(__lowerCamelCase			,axis=0				)
      lowercase_    :int					    =				np.array(__lowerCamelCase				).astype(np.floataa				) / 255.0
      lowercase_    :Union[str, Any]					    =				image.transpose(0			,3			,1			,2				)
      lowercase_    :str					    =				2.0 * image - 1.0
      lowercase_    :int					    =				torch.from_numpy(__lowerCamelCase				)
   elif isinstance(image[0]			,torch.Tensor				):
      lowercase_    :Optional[Any]					    =				torch.cat(__lowerCamelCase			,dim=0				)

   return image


class    a_     (       _lowerCAmelCase							):





     def __init__(				self	:      Any			,   lowercase	:      OnnxRuntimeModel			,   lowercase	:      OnnxRuntimeModel			,   lowercase	:      Any			,   lowercase	:      OnnxRuntimeModel			,   lowercase	:      DDPMScheduler			,   lowercase	:      Any			,   lowercase	:      int = 350			,   ):





        """simple docstring"""




        super().__init__(
            vae=lowercase			,   text_encoder=lowercase			,   tokenizer=lowercase			,   unet=lowercase			,   low_res_scheduler=lowercase			,   scheduler=lowercase			,   safety_checker=lowercase			,   feature_extractor=lowercase			,   watermarker=lowercase			,   max_noise_level=lowercase			,   )





     def __call__(				self	:      Any			,   lowercase	:      Union[str, List[str]]			,   lowercase	:      Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]			,   lowercase	:      int = 75			,   lowercase	:      float = 9.0			,   lowercase	:      int = 20			,   lowercase	:      Optional[Union[str, List[str]]] = None			,   lowercase	:      Optional[int] = 1			,   lowercase	:      float = 0.0			,   lowercase	:      Optional[Union[torch.Generator, List[torch.Generator]]] = None			,   lowercase	:      Optional[torch.FloatTensor] = None			,   lowercase	:      Optional[np.ndarray] = None			,   lowercase	:      Optional[np.ndarray] = None			,   lowercase	:      Optional[str] = "pil"			,   lowercase	:      bool = True			,   lowercase	:      Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,   lowercase	:      Optional[int] = 1			,   ):





        """simple docstring"""




        self.check_inputs(lowercase			,   lowercase			,   lowercase			,   lowercase		)

        # 2. Define call parameters
        if prompt is not None and isinstance(lowercase			,   lowercase		):
           lowercase_    :List[str]					    =				1
        elif prompt is not None and isinstance(lowercase			,   lowercase		):
           lowercase_    :Any					    =				len(lowercase		)
        else:
           lowercase_    :Dict					    =				prompt_embeds.shape[0]

        lowercase_    :int					    =				self._execution_device
        # 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

        # 3. Encode input prompt
        lowercase_    :Union[str, Any]					    =				self._encode_prompt(
            lowercase			,   lowercase			,   lowercase			,   lowercase			,   lowercase			,   prompt_embeds=lowercase			,   negative_prompt_embeds=lowercase			,   )

        lowercase_    :Dict					    =				ORT_TO_PT_TYPE[str(text_embeddings.dtype		)]

        # 4. Preprocess image
        lowercase_    :List[str]					    =				preprocess(lowercase		)
        lowercase_    :int					    =				image.cpu()

        # 5. set timesteps
        self.scheduler.set_timesteps(lowercase			,   device=lowercase		)
        lowercase_    :str					    =				self.scheduler.timesteps

        # 5. Add noise to image
        lowercase_    :str					    =				torch.tensor([noise_level]			,   dtype=torch.long			,   device=lowercase		)
        lowercase_    :List[Any]					    =				torch.randn(image.shape			,   generator=lowercase			,   device=lowercase			,   dtype=lowercase		)
        lowercase_    :List[Any]					    =				self.low_res_scheduler.add_noise(lowercase			,   lowercase			,   lowercase		)

        lowercase_    :Optional[int]					    =				2 if do_classifier_free_guidance else 1
        lowercase_    :Any					    =				np.concatenate([image] * batch_multiplier * num_images_per_prompt		)
        lowercase_    :Union[str, Any]					    =				np.concatenate([noise_level] * image.shape[0]		)

        # 6. Prepare latent variables
        lowercase_    ,							lowercase_    :Any					    =				image.shape[2:]
        lowercase_    :List[str]					    =				self.prepare_latents(
            batch_size * num_images_per_prompt			,   lowercase			,   lowercase			,   lowercase			,   lowercase			,   lowercase			,   lowercase			,   lowercase			,   )

        # 7. Check that sizes of image and latents match
        lowercase_    :Tuple					    =				image.shape[1]
        if NUM_LATENT_CHANNELS + num_channels_image != NUM_UNET_INPUT_CHANNELS:
           raise ValueError(
               "Incorrect configuration settings! The config of `pipeline.unet` expects"
               F' {NUM_UNET_INPUT_CHANNELS} but received `num_channels_latents`: {NUM_LATENT_CHANNELS} +'
               F' `num_channels_image`: {num_channels_image} '
               F' = {NUM_LATENT_CHANNELS+num_channels_image}. Please verify the config of'
               " `pipeline.unet` or your `image` input."		)

        # 8. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
        lowercase_    :Union[str, Any]					    =				self.prepare_extra_step_kwargs(lowercase			,   lowercase		)

        lowercase_    :Tuple					    =				next(
            (input.type for input in self.unet.model.get_inputs() if input.name == "timestep")			,   "tensor(float)"		)
        lowercase_    :Optional[int]					    =				ORT_TO_NP_TYPE[timestep_dtype]

        # 9. Denoising loop
        lowercase_    :Tuple					    =				len(lowercase		) - num_inference_steps * self.scheduler.order
        with self.progress_bar(total=lowercase		) as progress_bar:
           for i, t in enumerate(lowercase		):
              # expand the latents if we are doing classifier free guidance
              lowercase_    :Tuple					    =				np.concatenate([latents] * 2		) if do_classifier_free_guidance else latents

              # concat latents, mask, masked_image_latents in the channel dimension
              lowercase_    :Tuple					    =				self.scheduler.scale_model_input(lowercase			,   lowercase		)
              lowercase_    :List[Any]					    =				np.concatenate([latent_model_input, image]			,   axis=1		)

              # timestep to tensor
              lowercase_    :Optional[int]					    =				np.array([t]			,   dtype=lowercase		)

              # predict the noise residual
              lowercase_    :Union[str, Any]					    =				self.unet(
                  sample=lowercase			,   timestep=lowercase			,   encoder_hidden_states=lowercase			,   class_labels=noise_level.astype(np.intaa		)			,   )[0]

              # perform guidance
              if do_classifier_free_guidance:
                 lowercase_    ,							lowercase_    :int					    =				np.split(lowercase			,   2		)
              lowercase_    :Optional[int]					    =				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

              # compute the previous noisy sample x_t -> x_t-1
              lowercase_    :List[Any]					    =				self.scheduler.step(
                  torch.from_numpy(lowercase		)			,   lowercase			,   lowercase			,   **lowercase		).prev_sample

              # call the callback, if provided
              if i == len(lowercase		) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
                 progress_bar.update()
                 if callback is not None and i % callback_steps == 0:
                    callback(lowercase			,   lowercase			,   lowercase		)

        # 10. Post-processing
        lowercase_    :Union[str, Any]					    =				self.decode_latents(latents.float()		)

        # 11. Convert to PIL
        if output_type == "pil":
           lowercase_    :str					    =				self.numpy_to_pil(lowercase		)

        if not return_dict:
           return (image,)

        return ImagePipelineOutput(images=lowercase		)





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





        """simple docstring"""




        lowercase_    :List[Any]					    =				1 / 0.0_83_33 * latents
        lowercase_    :Optional[int]					    =				self.vae(latent_sample=lowercase		)[0]
        lowercase_    :Dict					    =				np.clip(image / 2 + 0.5			,   0			,   1		)
        lowercase_    :Tuple					    =				image.transpose((0, 2, 3, 1)		)
        return image







     def 							lowercase__   (				self	:      List[Any]			,   lowercase	:      Union[str, List[str]]			,   lowercase	:      str			,   lowercase	:      Optional[int]			,   lowercase	:      bool			,   lowercase	:      Optional[str]			,   lowercase	:      Optional[torch.FloatTensor] = None			,   lowercase	:      Optional[torch.FloatTensor] = None			,   ):





        """simple docstring"""




        if prompt is not None and isinstance(lowercase			,   lowercase		):
           lowercase_    :Any					    =				1
        elif prompt is not None and isinstance(lowercase			,   lowercase		):
           lowercase_    :Optional[int]					    =				len(lowercase		)
        else:
           lowercase_    :int					    =				prompt_embeds.shape[0]

        if prompt_embeds is None:
           lowercase_    :str					    =				self.tokenizer(
               lowercase			,   padding="max_length"			,   max_length=self.tokenizer.model_max_length			,   truncation=lowercase			,   return_tensors="pt"			,   )
           lowercase_    :Optional[Any]					    =				text_inputs.input_ids
           lowercase_    :Tuple					    =				self.tokenizer(lowercase			,   padding="longest"			,   return_tensors="pt"		).input_ids

           if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
               lowercase			,   lowercase		):
              lowercase_    :List[str]					    =				self.tokenizer.batch_decode(
                  untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]		)
              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}'		)

           # no positional arguments to text_encoder
           lowercase_    :Union[str, Any]					    =				self.text_encoder(
               input_ids=text_input_ids.int().to(lowercase		)			,   )
           lowercase_    :List[str]					    =				prompt_embeds[0]

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

        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance and negative_prompt_embeds is None:
           lowercase_    :List[str]
           if negative_prompt is None:
              lowercase_    :Optional[int]					    =				[""] * batch_size
           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_    :List[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_    :Optional[int]					    =				negative_prompt

           lowercase_    :List[Any]					    =				text_input_ids.shape[-1]
           lowercase_    :int					    =				self.tokenizer(
               lowercase			,   padding="max_length"			,   max_length=lowercase			,   truncation=lowercase			,   return_tensors="pt"			,   )

           # if hasattr(uncond_input, "attention_mask"):
           #     attention_mask = uncond_input.attention_mask.to(device)
           # else:
           #     attention_mask = None

           lowercase_    :List[str]					    =				self.text_encoder(
               input_ids=uncond_input.input_ids.int().to(lowercase		)			,   )
           lowercase_    :Tuple					    =				uncond_embeddings[0]

        if do_classifier_free_guidance:
           lowercase_    :Tuple					    =				uncond_embeddings.shape[1]
           # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
           lowercase_    :List[Any]					    =				uncond_embeddings.repeat(1			,   lowercase		)
           lowercase_    :Any					    =				uncond_embeddings.reshape(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_    :Optional[Any]					    =				np.concatenate([uncond_embeddings, prompt_embeds]		)

        return prompt_embeds









'''simple docstring'''


import math
def 							UpperCAmelCase_						(		__lowerCamelCase      :      int				):

   if 1 < number < 4:
      # 2 and 3 are primes
      return True
   elif number < 2 or number % 2 == 0 or number % 3 == 0:
      # Negatives, 0, 1, all even numbers, all multiples of 3 are not primes
      return False

   # All primes number are in format of 6k +/- 1
   for i in range(5			,int(math.sqrt(__lowerCamelCase				) + 1				)			,6				):
      if number % i == 0 or number % (i + 2) == 0:
         return False
   return True
def 							UpperCAmelCase_						(		__lowerCamelCase      :      int = 2_00_00_00				):
   return sum(num for num in range(3			,__lowerCamelCase			,2				) if is_prime(__lowerCamelCase				)				) + 2 if n > 2 else 0


if __name__ == "__main__":
 print(F'''{solution() = }''')


