





from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


_UpperCAmelCase             =							logging.get_logger(__name__)  # pylint: disable=invalid-name

_UpperCAmelCase             =							'\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")\n        >>> pipe_prior.to("cuda")\n        >>> prompt = "red cat, 4k photo"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> zero_image_emb = out.negative_image_embeds\n        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")\n        >>> pipe.to("cuda")\n        >>> image = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=50,\n        ... ).images\n        >>> image[0].save("cat.png")\n        ```\n'






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:Dict   ,   SCREAMING_SNAKE_CASE		:Optional[Any]   ,   SCREAMING_SNAKE_CASE		:Optional[Any]=8   )      ->   Optional[int]:
 __lowerCAmelCase					:	List[str]      =       height // scale_factor**2
 if height % scale_factor**2 != 0:
  new_height += 1
 __lowerCAmelCase					:	Optional[Any]      =       width // scale_factor**2
 if width % scale_factor**2 != 0:
  new_width += 1
 return new_height * scale_factor, new_width * scale_factor







class 							snake_case_    (		snake_case_						):






  def __init__(							self			:					Optional[int]       ,  _snake_case			:					UNetaDConditionModel       ,  _snake_case			:					DDPMScheduler       ,  _snake_case			:					VQModel       ,  )->str:






   '''simple docstring'''

   super().__init__()

   self.register_modules(
       unet=_snake_case       ,  scheduler=_snake_case       ,  movq=_snake_case       ,  )
   __lowerCAmelCase					:	Optional[Any]      =       2 ** (len(self.movq.config.block_out_channels      ) - 1)






  def 				UpperCAmelCase__						(							self			:					Optional[Any]       ,  _snake_case			:					Tuple       ,  _snake_case			:					int       ,  _snake_case			:					Any       ,  _snake_case			:					Optional[int]       ,  _snake_case			:					Union[str, Any]       ,  _snake_case			:					Optional[int]      )->Tuple:






   '''simple docstring'''

   if latents is None:
    __lowerCAmelCase					:	Optional[Any]      =       randn_tensor(_snake_case       ,  generator=_snake_case       ,  device=_snake_case       ,  dtype=_snake_case      )
   else:
    if latents.shape != shape:
     raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {shape}'''      )
    __lowerCAmelCase					:	int      =       latents.to(_snake_case      )

   __lowerCAmelCase					:	Optional[Any]      =       latents * scheduler.init_noise_sigma
   return latents






  def 				UpperCAmelCase__						(							self			:					List[Any]       ,  _snake_case			:					List[str]=0      )->Dict:






   '''simple docstring'''

   if is_accelerate_available():
    from accelerate import cpu_offload
   else:
    raise ImportError("""Please install accelerate via `pip install accelerate`"""      )

   __lowerCAmelCase					:	str      =       torch.device(F'''cuda:{gpu_id}'''      )

   __lowerCAmelCase					:	Dict      =       [
       self.unet,
       self.movq,
   ]
   for cpu_offloaded_model in models:
    if cpu_offloaded_model is not None:
     cpu_offload(_snake_case       ,  _snake_case      )






  def 				UpperCAmelCase__						(							self			:					Union[str, Any]       ,  _snake_case			:					Any=0      )->Tuple:






   '''simple docstring'''

   if is_accelerate_available() and is_accelerate_version(""">="""       ,  """0.17.0.dev0"""      ):
    from accelerate import cpu_offload_with_hook
   else:
    raise ImportError("""`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher."""      )

   __lowerCAmelCase					:	Optional[int]      =       torch.device(F'''cuda:{gpu_id}'''      )

   if self.device.type != "cpu":
    self.to("""cpu"""       ,  silence_dtype_warnings=_snake_case      )
    torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

   __lowerCAmelCase					:	List[Any]      =       None
   for cpu_offloaded_model in [self.unet, self.movq]:
    __lowerCAmelCase					:	Optional[int]      =       cpu_offload_with_hook(_snake_case       ,  _snake_case       ,  prev_module_hook=_snake_case      )

   # We'll offload the last model manually.
   __lowerCAmelCase					:	Tuple      =       hook






  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def 				UpperCAmelCase__						(							self			:					str      )->Optional[int]:






   '''simple docstring'''

   if not hasattr(self.unet       ,  """_hf_hook"""      ):
    return self.device
   for module in self.unet.modules():
    if (
        hasattr(_snake_case       ,  """_hf_hook"""      )
        and hasattr(module._hf_hook       ,  """execution_device"""      )
        and module._hf_hook.execution_device is not None
    ):
     return torch.device(module._hf_hook.execution_device      )
   return self.device






  @torch.no_grad()
  @replace_example_docstring(_snake_case      )
  def __call__(							self			:					int       ,  _snake_case			:					Union[torch.FloatTensor, List[torch.FloatTensor]]       ,  _snake_case			:					Union[torch.FloatTensor, List[torch.FloatTensor]]       ,  _snake_case			:					int = 512       ,  _snake_case			:					int = 512       ,  _snake_case			:					int = 100       ,  _snake_case			:					float = 4.0       ,  _snake_case			:					int = 1       ,  _snake_case			:					Optional[Union[torch.Generator, List[torch.Generator]]] = None       ,  _snake_case			:					Optional[torch.FloatTensor] = None       ,  _snake_case			:					Optional[str] = "pil"       ,  _snake_case			:					bool = True       ,  )->Dict:






   '''simple docstring'''

   __lowerCAmelCase					:	Optional[Any]      =       self._execution_device

   __lowerCAmelCase					:	str      =       guidance_scale > 1.0

   if isinstance(_snake_case       ,  _snake_case      ):
    __lowerCAmelCase					:	Tuple      =       torch.cat(_snake_case       ,  dim=0      )
   __lowerCAmelCase					:	str      =       image_embeds.shape[0] * num_images_per_prompt
   if isinstance(_snake_case       ,  _snake_case      ):
    __lowerCAmelCase					:	Any      =       torch.cat(_snake_case       ,  dim=0      )

   if do_classifier_free_guidance:
    __lowerCAmelCase					:	int      =       image_embeds.repeat_interleave(_snake_case       ,  dim=0      )
    __lowerCAmelCase					:	str      =       negative_image_embeds.repeat_interleave(_snake_case       ,  dim=0      )

   __lowerCAmelCase					:	str      =       torch.cat([negative_image_embeds, image_embeds]       ,  dim=0      ).to(dtype=self.unet.dtype       ,  device=_snake_case      )

   self.scheduler.set_timesteps(_snake_case       ,  device=_snake_case      )
   __lowerCAmelCase					:	int      =       self.scheduler.timesteps

   __lowerCAmelCase					:	Union[str, Any]      =       self.unet.config.in_channels

   __lowerCAmelCase					:	Union[str, Any]      =       downscale_height_and_width(_snake_case       ,  _snake_case       ,  self.movq_scale_factor      )

   # create initial latent
   __lowerCAmelCase					:	Tuple      =       self.prepare_latents(
       (batch_size, num_channels_latents, height, width)       ,  image_embeds.dtype       ,  _snake_case       ,  _snake_case       ,  _snake_case       ,  self.scheduler       ,  )

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

    __lowerCAmelCase					:	int      =       {'image_embeds': image_embeds}
    __lowerCAmelCase					:	Dict      =       self.unet(
        sample=_snake_case       ,  timestep=_snake_case       ,  encoder_hidden_states=_snake_case       ,  added_cond_kwargs=_snake_case       ,  return_dict=_snake_case       ,  )[0]

    if do_classifier_free_guidance:
     __lowerCAmelCase					:	Tuple      =       noise_pred.split(latents.shape[1]       ,  dim=1      )
     __lowerCAmelCase					:	List[Any]      =       noise_pred.chunk(2      )
     __lowerCAmelCase					:	Union[str, Any]      =       variance_pred.chunk(2      )
     __lowerCAmelCase					:	Optional[int]      =       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
     __lowerCAmelCase					:	Any      =       torch.cat([noise_pred, variance_pred_text]       ,  dim=1      )

    if not (
        hasattr(self.scheduler.config       ,  """variance_type"""      )
        and self.scheduler.config.variance_type in ["learned", "learned_range"]
    ):
     __lowerCAmelCase					:	List[str]      =       noise_pred.split(latents.shape[1]       ,  dim=1      )

    # compute the previous noisy sample x_t -> x_t-1
    __lowerCAmelCase					:	Dict      =       self.scheduler.step(
        _snake_case       ,  _snake_case       ,  _snake_case       ,  generator=_snake_case       ,  )[0]
   # post-processing
   __lowerCAmelCase					:	int      =       self.movq.decode(_snake_case       ,  force_not_quantize=_snake_case      )['sample']

   if output_type not in ["pt", "np", "pil"]:
    raise ValueError(F'''Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'''      )

   if output_type in ["np", "pil"]:
    __lowerCAmelCase					:	Tuple      =       image * 0.5 + 0.5
    __lowerCAmelCase					:	Union[str, Any]      =       image.clamp(0       ,  1      )
    __lowerCAmelCase					:	List[Any]      =       image.cpu().permute(0       ,  2       ,  3       ,  1      ).float().numpy()

   if output_type == "pil":
    __lowerCAmelCase					:	List[Any]      =       self.numpy_to_pil(_snake_case      )

   if not return_dict:
    return (image,)

   return ImagePipelineOutput(images=_snake_case      )






from __future__ import annotations






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:list[list[int]]   )      ->   bool:
 __lowerCAmelCase					:	Tuple      =       len(SCREAMING_SNAKE_CASE   )
 # We need to create solution object to save path.
 __lowerCAmelCase					:	str      =       [[0 for _ in range(SCREAMING_SNAKE_CASE   )] for _ in range(SCREAMING_SNAKE_CASE   )]
 __lowerCAmelCase					:	str      =       run_maze(SCREAMING_SNAKE_CASE   ,   0   ,   0   ,   SCREAMING_SNAKE_CASE   )
 if solved:
  print("""\n""".join(str(SCREAMING_SNAKE_CASE   ) for row in solutions   )   )
 else:
  print("""No solution exists!"""   )
 return solved






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:list[list[int]]   ,   SCREAMING_SNAKE_CASE		:int   ,   SCREAMING_SNAKE_CASE		:int   ,   SCREAMING_SNAKE_CASE		:list[list[int]]   )      ->   bool:
 __lowerCAmelCase					:	List[str]      =       len(SCREAMING_SNAKE_CASE   )
 # Final check point.
 if i == j == (size - 1):
  __lowerCAmelCase					:	str      =       1
  return True

 __lowerCAmelCase					:	Optional[Any]      =       (not i < 0) and (not j < 0)  # Check lower bounds
 __lowerCAmelCase					:	Optional[Any]      =       (i < size) and (j < size)  # Check upper bounds

 if lower_flag and upper_flag:
  # check for already visited and block points.
  __lowerCAmelCase					:	int      =       (not solutions[i][j]) and (not maze[i][j])
  if block_flag:
   # check visited
   __lowerCAmelCase					:	Tuple      =       1

   # check for directions
   if (
       run_maze(SCREAMING_SNAKE_CASE   ,   i + 1   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )
       or run_maze(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   j + 1   ,   SCREAMING_SNAKE_CASE   )
       or run_maze(SCREAMING_SNAKE_CASE   ,   i - 1   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )
       or run_maze(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   j - 1   ,   SCREAMING_SNAKE_CASE   )
   ):
    return True

   __lowerCAmelCase					:	Tuple      =       0
   return False
 return False


if __name__ == "__main__":
      import doctest

      doctest.testmod()