





from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

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


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

SCREAMING_SNAKE_CASE_   :      int			=   '''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Img2ImgPipeline, KandinskyV22PriorPipeline
        >>> from diffusers.utils import load_image
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16
        ... )
        >>> pipe_prior.to("cuda")

        >>> prompt = "A red cartoon frog, 4k"
        >>> image_emb, zero_image_emb = pipe_prior(prompt, return_dict=False)

        >>> pipe = KandinskyV22Img2ImgPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-decoder", torch_dtype=torch.float16
        ... )
        >>> pipe.to("cuda")

        >>> init_image = load_image(
        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
        ...     "/kandinsky/frog.png"
        ... )

        >>> image = pipe(
        ...     image=init_image,
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=100,
        ...     strength=0.2,
        ... ).images

        >>> image[0].save("red_frog.png")
        ```
'''



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case						,   snake_case=8							)		->    Optional[int]:
		__lowercase											=     height // scale_factor**2
		if height % scale_factor**2 != 0:
				new_height += 1
		__lowercase											=     width // scale_factor**2
		if width % scale_factor**2 != 0:
				new_width += 1
		return new_height * scale_factor, new_width * scale_factor



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case=512						,   snake_case=512							)		->    Optional[Any]:
		__lowercase											=     pil_image.resize((w, h)						,   resample=Image.BICUBIC						,   reducing_gap=1							)
		__lowercase											=     np.array(pil_image.convert('RGB'							)							)
		__lowercase											=     arr.astype(np.floataa							) / 127.5 - 1
		__lowercase											=     np.transpose(snake_case						,   [2, 0, 1]							)
		__lowercase											=     torch.from_numpy(snake_case							).unsqueeze(0							)
		return image


class 		snake_case_   (							UpperCAmelCase_						):







	'''simple docstring'''




	def __init__(						self							:     Dict					,      __lowerCamelCase							:     UNetaDConditionModel					,      __lowerCamelCase							:     DDPMScheduler					,      __lowerCamelCase							:     VQModel					,      )       -> Optional[Any]:

			'''simple docstring'''




			super().__init__()

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




	def    UpperCAmelCase			(						self							:     Optional[int]					,      __lowerCamelCase							:     str					,      __lowerCamelCase							:     Tuple					,      __lowerCamelCase							:     List[str]   )       -> Optional[Any]:

			'''simple docstring'''




			__lowercase											=     min(int(num_inference_steps * strength   )					,      __lowerCamelCase   )

			__lowercase											=     max(num_inference_steps - init_timestep					,      0   )
			__lowercase											=     self.scheduler.timesteps[t_start:]

			return timesteps, num_inference_steps - t_start




	def    UpperCAmelCase			(						self							:     str					,      __lowerCamelCase							:     str					,      __lowerCamelCase							:     Optional[int]					,      __lowerCamelCase							:     List[Any]					,      __lowerCamelCase							:     Union[str, Any]					,      __lowerCamelCase							:     str					,      __lowerCamelCase							:     Optional[Any]					,      __lowerCamelCase							:     Optional[int]=None   )       -> Optional[int]:

			'''simple docstring'''




			if not isinstance(__lowerCamelCase					,      (torch.Tensor, PIL.Image.Image, list)   ):
					raise ValueError(
					    F"`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(__lowerCamelCase   )}"   )

			__lowercase											=     image.to(device=__lowerCamelCase					,      dtype=__lowerCamelCase   )

			__lowercase											=     batch_size * num_images_per_prompt

			if image.shape[1] == 4:
					__lowercase											=     image

			else:
					if isinstance(__lowerCamelCase					,      __lowerCamelCase   ) and len(__lowerCamelCase   ) != batch_size:
							raise ValueError(
							    F"You have passed a list of generators of length {len(__lowerCamelCase   )}, but requested an effective batch"
							    F" size of {batch_size}. Make sure the batch size matches the length of the generators."   )

					elif isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
							__lowercase											=     [
							    self.movq.encode(image[i : i + 1]   ).latent_dist.sample(generator[i]   ) for i in range(__lowerCamelCase   )
							]
							__lowercase											=     torch.cat(__lowerCamelCase					,      dim=0   )
					else:
							__lowercase											=     self.movq.encode(__lowerCamelCase   ).latent_dist.sample(__lowerCamelCase   )

					__lowercase											=     self.movq.config.scaling_factor * init_latents

			__lowercase											=     torch.cat([init_latents]					,      dim=0   )

			__lowercase											=     init_latents.shape
			__lowercase											=     randn_tensor(__lowerCamelCase					,      generator=__lowerCamelCase					,      device=__lowerCamelCase					,      dtype=__lowerCamelCase   )

			# get latents
			__lowercase											=     self.scheduler.add_noise(__lowerCamelCase					,      __lowerCamelCase					,      __lowerCamelCase   )

			__lowercase											=     init_latents

			return latents




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

			'''simple docstring'''




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

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

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




	def    UpperCAmelCase			(						self							:     int					,      __lowerCamelCase							:     Dict=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.'   )

			__lowercase											=     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)

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

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




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

			'''simple docstring'''




			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							:     Any					,      __lowerCamelCase							:     Union[torch.FloatTensor, List[torch.FloatTensor]]					,      __lowerCamelCase							:     Union[torch.FloatTensor, PIL.Image.Image, List[torch.FloatTensor], List[PIL.Image.Image]]					,      __lowerCamelCase							:     Union[torch.FloatTensor, List[torch.FloatTensor]]					,      __lowerCamelCase							:     int = 512					,      __lowerCamelCase							:     int = 512					,      __lowerCamelCase							:     int = 100					,      __lowerCamelCase							:     float = 4.0					,      __lowerCamelCase							:     float = 0.3					,      __lowerCamelCase							:     int = 1					,      __lowerCamelCase							:     Optional[Union[torch.Generator, List[torch.Generator]]] = None					,      __lowerCamelCase							:     Optional[str] = "pil"					,      __lowerCamelCase							:     bool = True					,      )       -> List[Any]:

			'''simple docstring'''




			__lowercase											=     self._execution_device

			__lowercase											=     guidance_scale > 1.0

			if isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					__lowercase											=     torch.cat(__lowerCamelCase					,      dim=0   )
			__lowercase											=     image_embeds.shape[0]
			if isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					__lowercase											=     torch.cat(__lowerCamelCase					,      dim=0   )

			if do_classifier_free_guidance:
					__lowercase											=     image_embeds.repeat_interleave(__lowerCamelCase					,      dim=0   )
					__lowercase											=     negative_image_embeds.repeat_interleave(__lowerCamelCase					,      dim=0   )

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

			if not isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					__lowercase											=     [image]
			if not all(isinstance(__lowerCamelCase					,      (PIL.Image.Image, torch.Tensor)   ) for i in image   ):
					raise ValueError(
					    F"Input is in incorrect format: {[type(__lowerCamelCase   ) for i in image]}. Currently, we only support  PIL image and pytorch tensor"   )

			__lowercase											=     torch.cat([prepare_image(__lowerCamelCase					,      __lowerCamelCase					,      __lowerCamelCase   ) for i in image]					,      dim=0   )
			__lowercase											=     image.to(dtype=image_embeds.dtype					,      device=__lowerCamelCase   )

			__lowercase											=     self.movq.encode(__lowerCamelCase   )['latents']
			__lowercase											=     latents.repeat_interleave(__lowerCamelCase					,      dim=0   )
			self.scheduler.set_timesteps(__lowerCamelCase					,      device=__lowerCamelCase   )
			__lowercase						,      __lowercase											=     self.get_timesteps(__lowerCamelCase					,      __lowerCamelCase					,      __lowerCamelCase   )
			__lowercase											=     timesteps[:1].repeat(batch_size * num_images_per_prompt   )
			__lowercase						,      __lowercase											=     downscale_height_and_width(__lowerCamelCase					,      __lowerCamelCase					,      self.movq_scale_factor   )
			__lowercase											=     self.prepare_latents(
			    __lowerCamelCase					,      __lowerCamelCase					,      __lowerCamelCase					,      __lowerCamelCase					,      image_embeds.dtype					,      __lowerCamelCase					,      __lowerCamelCase   )
			for i, t in enumerate(self.progress_bar(__lowerCamelCase   )   ):
					# expand the latents if we are doing classifier free guidance
					__lowercase											=     torch.cat([latents] * 2   ) if do_classifier_free_guidance else latents

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

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

					# compute the previous noisy sample x_t -> x_t-1
					__lowercase											=     self.scheduler.step(
					    __lowerCamelCase					,      __lowerCamelCase					,      __lowerCamelCase					,      generator=__lowerCamelCase					,      )[0]

			# post-processing
			__lowercase											=     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"]:
					__lowercase											=     image * 0.5 + 0.5
					__lowercase											=     image.clamp(0					,      1   )
					__lowercase											=     image.cpu().permute(0					,      2					,      3					,      1   ).float().numpy()

			if output_type == "pil":
					__lowercase											=     self.numpy_to_pil(__lowerCamelCase   )

			if not return_dict:
					return (image,)

			return ImagePipelineOutput(images=__lowerCamelCase   )











from math import factorial



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case							)		->    int:

		# If either of the conditions are true, the function is being asked
		# to calculate a factorial of a negative number, which is not possible
		if n < k or k < 0:
				raise ValueError('Please enter positive integers for n and k where n >= k'							)
		return factorial(snake_case							) // (factorial(snake_case							) * factorial(n - k							))


if __name__ == "__main__":
		print(
		    '''The number of five-card hands possible from a standard''',
		    F"""fifty-two card deck is: {combinations(52, 5)}\n""",
		)

		print(
		    '''If a class of 40 students must be arranged into groups of''',
		    F"""4 for group projects, there are {combinations(40, 4)} ways""",
		    '''to arrange them.\n''',
		)

		print(
		    '''If 10 teams are competing in a Formula One race, there''',
		    F"""are {combinations(10, 3)} ways that first, second and""",
		    '''third place can be awarded.''',
		)




