
'''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


A           =   getLogger(__name__)


A           =   4
A           =   7

A           =   {
    "float16": torch.floataa,
    "float32": torch.floataa,
}
def   snake_case_       (							_a     :							List[str]			):
      if isinstance(_a ,					torch.Tensor			):
            return image
      elif isinstance(_a ,					PIL.Image.Image			):
            snake_case_						       =							[image]

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

            snake_case_						       =							[np.array(i.resize((w, h)			)			)[None, :] for i in image]
            snake_case_						       =							np.concatenate(_a ,					axis=0			)
            snake_case_						       =							np.array(_a			).astype(np.floataa			) / 2_5_5.0
            snake_case_						       =							image.transpose(0 ,					3 ,					1 ,					2			)
            snake_case_						       =							2.0 * image - 1.0
            snake_case_						       =							torch.from_numpy(_a			)
      elif isinstance(image[0] ,					torch.Tensor			):
            snake_case_						       =							torch.cat(_a ,					dim=0			)

      return image




class  lowerCAmelCase__   (       UpperCamelCase       ):



  '''simple docstring'''






  def __init__(self  : int	, lowerCAmelCase_  : OnnxRuntimeModel	, lowerCAmelCase_  : OnnxRuntimeModel	, lowerCAmelCase_  : Any	, lowerCAmelCase_  : OnnxRuntimeModel	, lowerCAmelCase_  : DDPMScheduler	, lowerCAmelCase_  : Any	, lowerCAmelCase_  : int = 3_50	, )   ->       Union[str, Any]:
        super().__init__(
            vae=lowerCAmelCase_	, text_encoder=lowerCAmelCase_	, tokenizer=lowerCAmelCase_	, unet=lowerCAmelCase_	, low_res_scheduler=lowerCAmelCase_	, scheduler=lowerCAmelCase_	, safety_checker=lowerCAmelCase_	, feature_extractor=lowerCAmelCase_	, watermarker=lowerCAmelCase_	, max_noise_level=lowerCAmelCase_	, )

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

        # 2. Define call parameters
        if prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
              snake_case_						       =							1
        elif prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
              snake_case_						       =							len(lowerCAmelCase_    )
        else:
              snake_case_						       =							prompt_embeds.shape[0]

        snake_case_						       =							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.
        snake_case_						       =							guidance_scale > 1.0

        # 3. Encode input prompt
        snake_case_						       =							self._encode_prompt(
            lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, prompt_embeds=lowerCAmelCase_	, negative_prompt_embeds=lowerCAmelCase_	, )

        snake_case_						       =							ORT_TO_PT_TYPE[str(text_embeddings.dtype    )]

        # 4. Preprocess image
        snake_case_						       =							preprocess(lowerCAmelCase_    )
        snake_case_						       =							image.cpu()

        # 5. set timesteps
        self.scheduler.set_timesteps(lowerCAmelCase_	, device=lowerCAmelCase_    )
        snake_case_						       =							self.scheduler.timesteps

        # 5. Add noise to image
        snake_case_						       =							torch.tensor([noise_level]	, dtype=torch.long	, device=lowerCAmelCase_    )
        snake_case_						       =							torch.randn(image.shape	, generator=lowerCAmelCase_	, device=lowerCAmelCase_	, dtype=lowerCAmelCase_    )
        snake_case_						       =							self.low_res_scheduler.add_noise(lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_    )

        snake_case_						       =							2 if do_classifier_free_guidance else 1
        snake_case_						       =							np.concatenate([image] * batch_multiplier * num_images_per_prompt    )
        snake_case_						       =							np.concatenate([noise_level] * image.shape[0]    )

        # 6. Prepare latent variables
        snake_case_       ,	snake_case_						       =							image.shape[2:]
        snake_case_						       =							self.prepare_latents(
            batch_size * num_images_per_prompt	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_	, )

        # 7. Check that sizes of image and latents match
        snake_case_						       =							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
        snake_case_						       =							self.prepare_extra_step_kwargs(lowerCAmelCase_	, lowerCAmelCase_    )

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

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

                    # concat latents, mask, masked_image_latents in the channel dimension
                    snake_case_						       =							self.scheduler.scale_model_input(lowerCAmelCase_	, lowerCAmelCase_    )
                    snake_case_						       =							np.concatenate([latent_model_input, image]	, axis=1    )

                    # timestep to tensor
                    snake_case_						       =							np.array([t]	, dtype=lowerCAmelCase_    )

                    # predict the noise residual
                    snake_case_						       =							self.unet(
                        sample=lowerCAmelCase_	, timestep=lowerCAmelCase_	, encoder_hidden_states=lowerCAmelCase_	, class_labels=noise_level.astype(np.intaa    )	, )[0]

                    # perform guidance
                    if do_classifier_free_guidance:
                          snake_case_       ,	snake_case_						       =							np.split(lowerCAmelCase_	, 2    )
                    snake_case_						       =							noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                    # compute the previous noisy sample x_t -> x_t-1
                    snake_case_						       =							self.scheduler.step(
                        torch.from_numpy(lowerCAmelCase_    )	, lowerCAmelCase_	, lowerCAmelCase_	, **lowerCAmelCase_    ).prev_sample

                    # call the callback, if provided
                    if i == len(lowerCAmelCase_    ) - 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(lowerCAmelCase_	, lowerCAmelCase_	, lowerCAmelCase_    )

        # 10. Post-processing
        snake_case_						       =							self.decode_latents(latents.float()    )

        # 11. Convert to PIL
        if output_type == "pil":
              snake_case_						       =							self.numpy_to_pil(lowerCAmelCase_    )

        if not return_dict:
              return (image,)

        return ImagePipelineOutput(images=lowerCAmelCase_    )

  def 		_lowerCamelCase	(self  : int	, lowerCAmelCase_  : str    )   ->       int:
        snake_case_						       =							1 / 0.08_333 * latents
        snake_case_						       =							self.vae(latent_sample=lowerCAmelCase_    )[0]
        snake_case_						       =							np.clip(image / 2 + 0.5	, 0	, 1    )
        snake_case_						       =							image.transpose((0, 2, 3, 1)    )
        return image


  def 		_lowerCamelCase	(self  : List[Any]	, lowerCAmelCase_  : Union[str, List[str]]	, lowerCAmelCase_  : str	, lowerCAmelCase_  : Optional[int]	, lowerCAmelCase_  : bool	, lowerCAmelCase_  : Optional[str]	, lowerCAmelCase_  : Optional[torch.FloatTensor] = None	, lowerCAmelCase_  : Optional[torch.FloatTensor] = None	, )   ->       Tuple:
        if prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
              snake_case_						       =							1
        elif prompt is not None and isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
              snake_case_						       =							len(lowerCAmelCase_    )
        else:
              snake_case_						       =							prompt_embeds.shape[0]

        if prompt_embeds is None:
              snake_case_						       =							self.tokenizer(
                  lowerCAmelCase_	, padding="max_length"	, max_length=self.tokenizer.model_max_length	, truncation=lowerCAmelCase_	, return_tensors="pt"	, )
              snake_case_						       =							text_inputs.input_ids
              snake_case_						       =							self.tokenizer(lowerCAmelCase_	, padding="longest"	, return_tensors="pt"    ).input_ids

              if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
                  lowerCAmelCase_	, lowerCAmelCase_    ):
                    snake_case_						       =							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
              snake_case_						       =							self.text_encoder(
                  input_ids=text_input_ids.int().to(lowerCAmelCase_    )	, )
              snake_case_						       =							prompt_embeds[0]

        snake_case_       ,	snake_case_       ,	snake_case_						       =							prompt_embeds.shape
        # duplicate text embeddings for each generation per prompt, using mps friendly method
        snake_case_						       =							prompt_embeds.repeat(1	, lowerCAmelCase_    )
        snake_case_						       =							prompt_embeds.reshape(bs_embed * num_images_per_prompt	, lowerCAmelCase_	, -1    )

        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance and negative_prompt_embeds is None:
              snake_case_						       =							42
              if negative_prompt is None:
                    snake_case_						       =							[""] * batch_size
              elif type(lowerCAmelCase_    ) is not type(lowerCAmelCase_    ):
                    raise TypeError(
                        f'''`negative_prompt` should be the same type to `prompt`, but got {type(lowerCAmelCase_    )} !='''
                        f''' {type(lowerCAmelCase_    )}.'''    )
              elif isinstance(lowerCAmelCase_	, lowerCAmelCase_    ):
                    snake_case_						       =							[negative_prompt]
              elif batch_size != len(lowerCAmelCase_    ):
                    raise ValueError(
                        f'''`negative_prompt`: {negative_prompt} has batch size {len(lowerCAmelCase_    )}, but `prompt`:'''
                        f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                        " the batch size of `prompt`."    )
              else:
                    snake_case_						       =							negative_prompt

              snake_case_						       =							text_input_ids.shape[-1]
              snake_case_						       =							self.tokenizer(
                  lowerCAmelCase_	, padding="max_length"	, max_length=lowerCAmelCase_	, truncation=lowerCAmelCase_	, return_tensors="pt"	, )

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

              snake_case_						       =							self.text_encoder(
                  input_ids=uncond_input.input_ids.int().to(lowerCAmelCase_    )	, )
              snake_case_						       =							uncond_embeddings[0]

        if do_classifier_free_guidance:
              snake_case_						       =							uncond_embeddings.shape[1]
              # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
              snake_case_						       =							uncond_embeddings.repeat(1	, lowerCAmelCase_    )
              snake_case_						       =							uncond_embeddings.reshape(batch_size * num_images_per_prompt	, lowerCAmelCase_	, -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
              snake_case_						       =							np.concatenate([uncond_embeddings, prompt_embeds]    )

        return prompt_embeds



'''simple docstring'''





def   snake_case_       (							_a     :							int = 1_000			):
      return sum(i for i in range(_a			) if i % 3 == 0 or i % 5 == 0			)


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

