


'''simple docstring'''






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  _snake_case       (					A			,   A			,   A=8	)						->   Tuple:
							lowerCAmelCase__	       =     height // scale_factor**2
							if height % scale_factor**2 != 0:
														new_height += 1
							lowerCAmelCase__	       =     width // scale_factor**2
							if width % scale_factor**2 != 0:
														new_width += 1
							return new_height * scale_factor, new_width * scale_factor






class 		a__    (							__lowerCAmelCase  ):







							def __init__(		self ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_ ,   )	->							str:
														super().__init__()

														self.register_modules(
														    unet=lowerCamelCase__ ,   scheduler=lowerCamelCase__ ,   movq=lowerCamelCase__ ,   )
														lowerCAmelCase__	       =     2 ** (len(self.movq.config.block_out_channels       ) - 1)







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_       )	->							Optional[int]:






														if latents is None:
																					lowerCAmelCase__	       =     randn_tensor(lowerCamelCase__ ,   generator=lowerCamelCase__ ,   device=lowerCamelCase__ ,   dtype=lowerCamelCase__       )
														else:
																					if latents.shape != shape:
																												raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}"""       )
																					lowerCAmelCase__	       =     latents.to(lowerCamelCase__       )

														lowerCAmelCase__	       =     latents * scheduler.init_noise_sigma
														return latents







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_=0       )	->							Optional[int]:






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

														lowerCAmelCase__	       =     torch.device(F"""cuda:{gpu_id}"""       )

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







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_=0       )	->							Tuple:






														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__	       =     torch.device(F"""cuda:{gpu_id}"""       )

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

														lowerCAmelCase__	       =     None
														for cpu_offloaded_model in [self.unet, self.movq]:
																					lowerCAmelCase__	       =     cpu_offload_with_hook(lowerCamelCase__ ,   lowerCamelCase__ ,   prev_module_hook=lowerCamelCase__       )

														# We'll offload the last model manually.
														lowerCAmelCase__	       =     hook







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






														if not hasattr(self.unet ,   '''_hf_hook'''       ):
																					return self.device
														for module in self.unet.modules():
																					if (
																					    hasattr(lowerCamelCase__ ,   '''_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(lowerCamelCase__       )
							def __call__(		self ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_ = 5_12 ,   lowerCamelCase_ = 5_12 ,   lowerCamelCase_ = 1_00 ,   lowerCamelCase_ = 4.0 ,   lowerCamelCase_ = 1 ,   lowerCamelCase_ = None ,   lowerCamelCase_ = None ,   lowerCamelCase_ = "pil" ,   lowerCamelCase_ = True ,   )	->							Dict:
														lowerCAmelCase__	       =     self._execution_device

														lowerCAmelCase__	       =     guidance_scale > 1.0

														if isinstance(lowerCamelCase__ ,   lowerCamelCase__       ):
																					lowerCAmelCase__	       =     torch.cat(lowerCamelCase__ ,   dim=0       )
														lowerCAmelCase__	       =     image_embeds.shape[0] * num_images_per_prompt
														if isinstance(lowerCamelCase__ ,   lowerCamelCase__       ):
																					lowerCAmelCase__	       =     torch.cat(lowerCamelCase__ ,   dim=0       )

														if do_classifier_free_guidance:
																					lowerCAmelCase__	       =     image_embeds.repeat_interleave(lowerCamelCase__ ,   dim=0       )
																					lowerCAmelCase__	       =     negative_image_embeds.repeat_interleave(lowerCamelCase__ ,   dim=0       )

														lowerCAmelCase__	       =     torch.cat([negative_image_embeds, image_embeds] ,   dim=0       ).to(dtype=self.unet.dtype ,   device=lowerCamelCase__       )

														self.scheduler.set_timesteps(lowerCamelCase__ ,   device=lowerCamelCase__       )
														lowerCAmelCase__	       =     self.scheduler.timesteps

														lowerCAmelCase__	       =     self.unet.config.in_channels

														lowerCAmelCase__	       =     downscale_height_and_width(lowerCamelCase__ ,   lowerCamelCase__ ,   self.movq_scale_factor       )

														# create initial latent
														lowerCAmelCase__	       =     self.prepare_latents(
														    (batch_size, num_channels_latents, height, width) ,   image_embeds.dtype ,   lowerCamelCase__ ,   lowerCamelCase__ ,   lowerCamelCase__ ,   self.scheduler ,   )

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

																					lowerCAmelCase__	       =     {'''image_embeds''': image_embeds}
																					lowerCAmelCase__	       =     self.unet(
																					    sample=lowerCamelCase__ ,   timestep=lowerCamelCase__ ,   encoder_hidden_states=lowerCamelCase__ ,   added_cond_kwargs=lowerCamelCase__ ,   return_dict=lowerCamelCase__ ,   )[0]

																					if do_classifier_free_guidance:
																												lowerCAmelCase__	       =     noise_pred.split(latents.shape[1] ,   dim=1       )
																												lowerCAmelCase__	       =     noise_pred.chunk(2       )
																												lowerCAmelCase__	       =     variance_pred.chunk(2       )
																												lowerCAmelCase__	       =     noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
																												lowerCAmelCase__	       =     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__	       =     noise_pred.split(latents.shape[1] ,   dim=1       )

																					# compute the previous noisy sample x_t -> x_t-1
																					lowerCAmelCase__	       =     self.scheduler.step(
																					    lowerCamelCase__ ,   lowerCamelCase__ ,   lowerCamelCase__ ,   generator=lowerCamelCase__ ,   )[0]
														# post-processing
														lowerCAmelCase__	       =     self.movq.decode(lowerCamelCase__ ,   force_not_quantize=lowerCamelCase__       )['''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__	       =     image * 0.5 + 0.5
																					lowerCAmelCase__	       =     image.clamp(0 ,   1       )
																					lowerCAmelCase__	       =     image.cpu().permute(0 ,   2 ,   3 ,   1       ).float().numpy()

														if output_type == "pil":
																					lowerCAmelCase__	       =     self.numpy_to_pil(lowerCamelCase__       )

														if not return_dict:
																					return (image,)

														return ImagePipelineOutput(images=lowerCamelCase__       )



'''simple docstring'''






def  _snake_case       (					A = 10			,   A = 22	)						->   int:
							lowerCAmelCase__	       =     range(1			,   A	)
							lowerCAmelCase__	       =     range(1			,   A	)
							return sum(
							    1 for power in powers for base in bases if len(str(base**power	)	) == power	)


if __name__ == "__main__":
		print(f"""{solution(10, 22) = }""")