

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


lowerCAmelCase__						:       Optional[Any]										=						logging.get_logger(__name__)  # pylint: disable=invalid-name

lowerCAmelCase__						:       int										=						"""
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

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




def 	_a       (				__lowerCAmelCase		:      Any   ,							__lowerCAmelCase		:      Tuple   ,							__lowerCAmelCase		:      List[str]=8     ):

		"""simple docstring"""

		snake_case__						:       str       						=  height // scale_factor**2
		if height % scale_factor**2 != 0:
				new_height += 1
		snake_case__						:       int       						=  width // scale_factor**2
		if width % scale_factor**2 != 0:
				new_width += 1
		return new_height * scale_factor, new_width * scale_factor





class     a			(				_UpperCamelCase       ):






						"""simple docstring"""




						def __init__(							self					:      Optional[Any]			,			snake_case_					:      UNetaDConditionModel			,			snake_case_					:      DDPMScheduler			,			snake_case_					:      VQModel			,			):



								'''simple docstring'''



								super().__init__()

								self.register_modules(
								    unet=snake_case_			,			scheduler=snake_case_			,			movq=snake_case_			,			)
								snake_case__						:       List[Any]       						=  2 ** (len(self.movq.config.block_out_channels  ) - 1)







						def       __magic_name__							(							self					:      Union[str, Any]			,			snake_case_					:      Any			,			snake_case_					:      Union[str, Any]			,			snake_case_					:      Tuple			,			snake_case_					:      Optional[Any]			,			snake_case_					:      List[str]			,			snake_case_					:      Optional[Any]  ):



								'''simple docstring'''



								if latents is None:
										snake_case__						:       Optional[int]       						=  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}"""  )
										snake_case__						:       Dict       						=  latents.to(snake_case_  )

								snake_case__						:       Tuple       						=  latents * scheduler.init_noise_sigma
								return latents







						def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      Optional[Any]=0  ):



								'''simple docstring'''



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

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

								snake_case__						:       Union[str, Any]       						=  [
								    self.unet,
								    self.movq,
								]
								for cpu_offloaded_model in models:
										if cpu_offloaded_model is not None:
												cpu_offload(snake_case_			,			snake_case_  )







						def       __magic_name__							(							self					:      Optional[int]			,			snake_case_					:      int=0  ):



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

								snake_case__						:       List[str]       						=  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)

								snake_case__						:       str       						=  None
								for cpu_offloaded_model in [self.unet, self.movq]:
										snake_case__						:       Tuple       						=  cpu_offload_with_hook(snake_case_			,			snake_case_			,			prev_module_hook=snake_case_  )

								# We'll offload the last model manually.
								snake_case__						:       Any       						=  hook







						@property
						# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
						def       __magic_name__							(							self					:      str  ):



								'''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					:      Optional[int]			,			snake_case_					:      Union[torch.FloatTensor, List[torch.FloatTensor]]			,			snake_case_					:      Union[torch.FloatTensor, List[torch.FloatTensor]]			,			snake_case_					:      int = 5_1_2			,			snake_case_					:      int = 5_1_2			,			snake_case_					:      int = 1_0_0			,			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			,			):



								'''simple docstring'''



								snake_case__						:       Optional[int]       						=  self._execution_device

								snake_case__						:       Union[str, Any]       						=  guidance_scale > 1.0

								if isinstance(snake_case_			,			snake_case_  ):
										snake_case__						:       Optional[Any]       						=  torch.cat(snake_case_			,			dim=0  )
								snake_case__						:       Optional[int]       						=  image_embeds.shape[0] * num_images_per_prompt
								if isinstance(snake_case_			,			snake_case_  ):
										snake_case__						:       Dict       						=  torch.cat(snake_case_			,			dim=0  )

								if do_classifier_free_guidance:
										snake_case__						:       Union[str, Any]       						=  image_embeds.repeat_interleave(snake_case_			,			dim=0  )
										snake_case__						:       int       						=  negative_image_embeds.repeat_interleave(snake_case_			,			dim=0  )

								snake_case__						:       Union[str, Any]       						=  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_  )
								snake_case__						:       Tuple       						=  self.scheduler.timesteps

								snake_case__						:       Dict       						=  self.unet.config.in_channels

								snake_case__						:       List[str]       						=  downscale_height_and_width(snake_case_			,			snake_case_			,			self.movq_scale_factor  )

								# create initial latent
								snake_case__						:       Optional[Any]       						=  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
										snake_case__						:       List[Any]       						=  torch.cat([latents] * 2  ) if do_classifier_free_guidance else latents

										snake_case__						:       List[str]       						=  {"image_embeds": image_embeds}
										snake_case__						:       Union[str, Any]       						=  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:
												snake_case__						:       Optional[int]       						=  noise_pred.split(latents.shape[1]			,			dim=1  )
												snake_case__						:       Optional[int]       						=  noise_pred.chunk(2  )
												snake_case__						:       int       						=  variance_pred.chunk(2  )
												snake_case__						:       Union[str, Any]       						=  noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
												snake_case__						:       Dict       						=  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"]
										):
												snake_case__						:       Dict       						=  noise_pred.split(latents.shape[1]			,			dim=1  )

										# compute the previous noisy sample x_t -> x_t-1
										snake_case__						:       int       						=  self.scheduler.step(
										    snake_case_			,			snake_case_			,			snake_case_			,			generator=snake_case_			,			)[0]
								# post-processing
								snake_case__						:       Any       						=  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"]:
										snake_case__						:       Dict       						=  image * 0.5 + 0.5
										snake_case__						:       Optional[int]       						=  image.clamp(0			,			1  )
										snake_case__						:       Optional[int]       						=  image.cpu().permute(0			,			2			,			3			,			1  ).float().numpy()

								if output_type == "pil":
										snake_case__						:       int       						=  self.numpy_to_pil(snake_case_  )

								if not return_dict:
										return (image,)

								return ImagePipelineOutput(images=snake_case_  )






'''simple docstring'''


def 	_a       (				__lowerCAmelCase		:      int     ):

		"""simple docstring"""

		if p < 2:
				raise ValueError('''p should not be less than 2!'''     )
		elif p == 2:
				return True

		snake_case__						:       Any       						=  4
		snake_case__						:       int       						=  (1 << p) - 1
		for _ in range(p - 2     ):
				snake_case__						:       Tuple       						=  ((s * s) - 2) % m
		return s == 0


if __name__ == "__main__":
					print(lucas_lehmer_test(7))
					print(lucas_lehmer_test(11))



