


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


__UpperCamelCase					:	List[Any]          =				getLogger(__name__)


__UpperCamelCase					:	Union[str, Any]          =				4
__UpperCamelCase					:	List[Any]          =				7

__UpperCamelCase					:	Optional[int]          =				{
    'float16': torch.floataa,
    'float32': torch.floataa,
}
def   snake_case_    (			__lowercase							):
   if isinstance(__lowercase      ,							torch.Tensor							):
      return image
   elif isinstance(__lowercase      ,							PIL.Image.Image							):
      UpperCAmelCase_				:   Dict							= [image]

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

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

   return image





class 			lowerCAmelCase__(     snake_case__    ):






    '''simple docstring'''





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





       '''simple docstring'''
       super().__init__(
           vae=__snake_case							, text_encoder=__snake_case							, tokenizer=__snake_case							, unet=__snake_case							, low_res_scheduler=__snake_case							, scheduler=__snake_case							, safety_checker=__snake_case							, feature_extractor=__snake_case							, watermarker=__snake_case							, max_noise_level=__snake_case							, )





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





       '''simple docstring'''
       self.check_inputs(__snake_case							, __snake_case							, __snake_case							, __snake_case						)

       # 2. Define call parameters
       if prompt is not None and isinstance(__snake_case							, __snake_case						):
          UpperCAmelCase_				:   Optional[int]							= 1
       elif prompt is not None and isinstance(__snake_case							, __snake_case						):
          UpperCAmelCase_				:   Union[str, Any]							= len(__snake_case						)
       else:
          UpperCAmelCase_				:   List[str]							= prompt_embeds.shape[0]

       UpperCAmelCase_				:   List[Any]							= 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.
       UpperCAmelCase_				:   int							= guidance_scale > 1.0

       # 3. Encode input prompt
       UpperCAmelCase_				:   int							= self._encode_prompt(
           __snake_case							, __snake_case							, __snake_case							, __snake_case							, __snake_case							, prompt_embeds=__snake_case							, negative_prompt_embeds=__snake_case							, )

       UpperCAmelCase_				:   List[Any]							= ORT_TO_PT_TYPE[str(text_embeddings.dtype						)]

       # 4. Preprocess image
       UpperCAmelCase_				:   Optional[int]							= preprocess(__snake_case						)
       UpperCAmelCase_				:   Dict							= image.cpu()

       # 5. set timesteps
       self.scheduler.set_timesteps(__snake_case							, device=__snake_case						)
       UpperCAmelCase_				:   List[str]							= self.scheduler.timesteps

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

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

       # 6. Prepare latent variables
       UpperCAmelCase_		,   UpperCAmelCase_				:   Tuple							= image.shape[2:]
       UpperCAmelCase_				:   Optional[int]							= self.prepare_latents(
           batch_size * num_images_per_prompt							, __snake_case							, __snake_case							, __snake_case							, __snake_case							, __snake_case							, __snake_case							, __snake_case							, )

       # 7. Check that sizes of image and latents match
       UpperCAmelCase_				:   Optional[int]							= 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
       UpperCAmelCase_				:   List[str]							= self.prepare_extra_step_kwargs(__snake_case							, __snake_case						)

       UpperCAmelCase_				:   int							= next(
           (input.type for input in self.unet.model.get_inputs() if input.name == '''timestep''')							, '''tensor(float)'''						)
       UpperCAmelCase_				:   List[str]							= ORT_TO_NP_TYPE[timestep_dtype]

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

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

             # timestep to tensor
             UpperCAmelCase_				:   List[Any]							= np.array([t]							, dtype=__snake_case						)

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

             # perform guidance
             if do_classifier_free_guidance:
                UpperCAmelCase_		,   UpperCAmelCase_				:   Optional[Any]							= np.split(__snake_case							, 2						)
             UpperCAmelCase_				:   List[str]							= noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

             # compute the previous noisy sample x_t -> x_t-1
             UpperCAmelCase_				:   int							= self.scheduler.step(
                 torch.from_numpy(__snake_case						)							, __snake_case							, __snake_case							, **__snake_case						).prev_sample

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

        # 10. Post-processing
       UpperCAmelCase_				:   Any							= self.decode_latents(latents.float()						)

       # 11. Convert to PIL
       if output_type == "pil":
          UpperCAmelCase_				:   Tuple							= self.numpy_to_pil(__snake_case						)

       if not return_dict:
          return (image,)

       return ImagePipelineOutput(images=__snake_case						)





    def        _lowerCamelCase			(						self					: List[Any]							, __snake_case					: Optional[Any]						):





       '''simple docstring'''
       UpperCAmelCase_				:   int							= 1 / 0.08_333 * latents
       UpperCAmelCase_				:   Optional[Any]							= self.vae(latent_sample=__snake_case						)[0]
       UpperCAmelCase_				:   Tuple							= np.clip(image / 2 + 0.5							, 0							, 1						)
       UpperCAmelCase_				:   Union[str, Any]							= image.transpose((0, 2, 3, 1)						)
       return image





    def        _lowerCamelCase			(						self					: str							, __snake_case					: Union[str, List[str]]							, __snake_case					: Optional[int]							, __snake_case					: Optional[int]							, __snake_case					: bool							, __snake_case					: Optional[str]							, __snake_case					: Optional[torch.FloatTensor] = None							, __snake_case					: Optional[torch.FloatTensor] = None							, ):





       '''simple docstring'''
       if prompt is not None and isinstance(__snake_case							, __snake_case						):
          UpperCAmelCase_				:   Any							= 1
       elif prompt is not None and isinstance(__snake_case							, __snake_case						):
          UpperCAmelCase_				:   Union[str, Any]							= len(__snake_case						)
       else:
          UpperCAmelCase_				:   List[str]							= prompt_embeds.shape[0]

       if prompt_embeds is None:
          UpperCAmelCase_				:   Dict							= self.tokenizer(
              __snake_case							, padding='''max_length'''							, max_length=self.tokenizer.model_max_length							, truncation=__snake_case							, return_tensors='''pt'''							, )
          UpperCAmelCase_				:   Any							= text_inputs.input_ids
          UpperCAmelCase_				:   str							= self.tokenizer(__snake_case							, padding='''longest'''							, return_tensors='''pt'''						).input_ids

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

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

       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance and negative_prompt_embeds is None:
          UpperCAmelCase_				:   List[str]
          if negative_prompt is None:
             UpperCAmelCase_				:   Union[str, Any]							= [''''''] * batch_size
          elif type(__snake_case						) is not type(__snake_case						):
             raise TypeError(
                 f'''`negative_prompt` should be the same type to `prompt`, but got {type(__snake_case						)} !='''
                 f''' {type(__snake_case						)}.'''						)
          elif isinstance(__snake_case							, __snake_case						):
             UpperCAmelCase_				:   Optional[Any]							= [negative_prompt]
          elif batch_size != len(__snake_case						):
             raise ValueError(
                 f'''`negative_prompt`: {negative_prompt} has batch size {len(__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:
             UpperCAmelCase_				:   List[Any]							= negative_prompt

          UpperCAmelCase_				:   Dict							= text_input_ids.shape[-1]
          UpperCAmelCase_				:   Dict							= self.tokenizer(
              __snake_case							, padding='''max_length'''							, max_length=__snake_case							, truncation=__snake_case							, return_tensors='''pt'''							, )

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

          UpperCAmelCase_				:   Union[str, Any]							= self.text_encoder(
              input_ids=uncond_input.input_ids.int().to(__snake_case						)							, )
          UpperCAmelCase_				:   Optional[Any]							= uncond_embeddings[0]

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

       return prompt_embeds



import os
import random
from pathlib import Path
from typing import List

import numpy as np
import torch
from torch.cuda.amp import GradScaler

from .utils import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    get_pretty_name,
    is_tpu_available,
    is_xpu_available,
    save,
)


if is_tpu_available(check_device=False):
       import torch_xla.core.xla_model as xm

from .logging import get_logger
from .state import PartialState


__UpperCamelCase					:	int          =				get_logger(__name__)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase = None      ,							):
   # Model states
   for i, state in enumerate(__lowercase							):
      UpperCAmelCase_				:   Dict							= F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Model weights saved in {output_model_file}'''							)
   # Optimizer states
   for i, opt in enumerate(__lowercase							):
      UpperCAmelCase_				:   Tuple							= opt.state_dict()
      UpperCAmelCase_				:   Any							= F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Optimizer state saved in {output_optimizer_file}'''							)
   # Scheduler states
   for i, scheduler in enumerate(__lowercase							):
      UpperCAmelCase_				:   Tuple							= scheduler.state_dict()
      UpperCAmelCase_				:   Tuple							= F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   Tuple							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Scheduler state saved in {output_scheduler_file}'''							)
   # GradScaler state
   if scaler is not None:
      UpperCAmelCase_				:   Union[str, Any]							= scaler.state_dict()
      UpperCAmelCase_				:   Union[str, Any]							= os.path.join(__lowercase      ,							__lowercase							)
      torch.save(__lowercase      ,							__lowercase							)
      logger.info(F'''Gradient scaler state saved in {output_scaler_file}'''							)
   # Random number generator states
   UpperCAmelCase_				:   Union[str, Any]							= {}
   UpperCAmelCase_				:   Tuple							= F'''{RNG_STATE_NAME}_{process_index}.pkl'''
   UpperCAmelCase_				:   List[Any]							= random.getstate()
   UpperCAmelCase_				:   Any							= np.random.get_state()
   UpperCAmelCase_				:   List[Any]							= torch.get_rng_state()
   if is_xpu_available():
      UpperCAmelCase_				:   Optional[int]							= torch.xpu.get_rng_state_all()
   else:
      UpperCAmelCase_				:   Any							= torch.cuda.get_rng_state_all()
   if is_tpu_available():
      UpperCAmelCase_				:   List[str]							= xm.get_rng_state()
   UpperCAmelCase_				:   Any							= os.path.join(__lowercase      ,							__lowercase							)
   torch.save(__lowercase      ,							__lowercase							)
   logger.info(F'''Random states saved in {output_states_file}'''							)
   return output_dir
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase=None      ,							__lowercase=None      ,							**__lowercase      ,							):
   if map_location not in [None, "cpu", "on_device"]:
      raise TypeError(
          '''Unsupported optimizer map location passed, please choose one of `None`, `\'cpu\'`, or `\'on_device\'`'''							)
   if map_location is None:
      UpperCAmelCase_				:   Optional[Any]							= '''cpu'''
   elif map_location == "on_device":
      UpperCAmelCase_				:   Optional[int]							= PartialState().device
   # Model states
   for i, model in enumerate(__lowercase							):
      UpperCAmelCase_				:   List[Any]							= F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
      UpperCAmelCase_				:   int							= os.path.join(__lowercase      ,							__lowercase							)
      models[i].load_state_dict(torch.load(__lowercase      ,							map_location=__lowercase							)      ,							**__lowercase							)
   logger.info('''All model weights loaded successfully'''							)

   # Optimizer states
   for i, opt in enumerate(__lowercase							):
      UpperCAmelCase_				:   List[str]							= F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      UpperCAmelCase_				:   Tuple							= torch.load(__lowercase      ,							map_location=__lowercase							)
      optimizers[i].load_state_dict(__lowercase							)
   logger.info('''All optimizer states loaded successfully'''							)

   # Scheduler states
   for i, scheduler in enumerate(__lowercase							):
      UpperCAmelCase_				:   Dict							= F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   int							= os.path.join(__lowercase      ,							__lowercase							)
      scheduler.load_state_dict(torch.load(__lowercase							)							)
   logger.info('''All scheduler states loaded successfully'''							)

   # GradScaler state
   if scaler is not None:
      UpperCAmelCase_				:   List[str]							= os.path.join(__lowercase      ,							__lowercase							)
      scaler.load_state_dict(torch.load(__lowercase							)							)
      logger.info('''GradScaler state loaded successfully'''							)

   # Random states
   try:
      UpperCAmelCase_				:   Optional[Any]							= torch.load(os.path.join(__lowercase      ,							F'''{RNG_STATE_NAME}_{process_index}.pkl'''							)							)
      random.setstate(states['''random_state''']							)
      np.random.set_state(states['''numpy_random_seed''']							)
      torch.set_rng_state(states['''torch_manual_seed''']							)
      if is_xpu_available():
         torch.xpu.set_rng_state_all(states['''torch_xpu_manual_seed''']							)
      else:
         torch.cuda.set_rng_state_all(states['''torch_cuda_manual_seed''']							)
      if is_tpu_available():
         xm.set_rng_state(states['''xm_seed''']							)
      logger.info('''All random states loaded successfully'''							)
   except Exception:
      logger.info('''Could not load random states'''							)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase = 0							):
   UpperCAmelCase_				:   Optional[int]							= Path(__lowercase							) / F'''custom_checkpoint_{index}.pkl'''
   logger.info(F'''Saving the state of {get_pretty_name(__lowercase							)} to {save_location}'''							)
   torch.save(obj.state_dict()      ,							__lowercase							)


def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase = 0							):
   UpperCAmelCase_				:   Dict							= F'''{path}/custom_checkpoint_{index}.pkl'''
   logger.info(F'''Loading the state of {get_pretty_name(__lowercase							)} from {load_location}'''							)
   obj.load_state_dict(torch.load(__lowercase      ,							map_location='''cpu'''							)							)