






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


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

__A      =				'\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      __A					(     _lowercase					,			_lowercase					,			_lowercase=8 ):

					'''simple docstring'''







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





class 						SCREAMING_SNAKE_CASE				( snake_case					):





							"""simple docstring"""

							def __init__(  self:   List[Any]		,     __A:   UNetaDConditionModel		,     __A:   DDPMScheduler		,     __A:   VQModel		,     )    ->				List[Any]:
												super().__init__()

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

							def       __A (  self:   Union[str, Any]		,     __A:   Optional[Any]		,     __A:   int		,     __A:   int		,     __A:   Tuple		,     __A:   List[str]		,     __A:   Optional[int]   )    ->				Union[str, Any]:
												if latents is None:
																	_A								= randn_tensor(__A		,     generator=__A		,     device=__A		,     dtype=__A   )
												else:
																	if latents.shape != shape:
																						raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {shape}"""   )
																	_A								= latents.to(__A   )

												_A								= latents * scheduler.init_noise_sigma
												return latents

							def       __A (  self:   Optional[Any]		,     __A:   Any=0   )    ->				Dict:
												if is_accelerate_available():
																	from accelerate import cpu_offload
												else:
																	raise ImportError('''Please install accelerate via `pip install accelerate`'''   )

												_A								= torch.device(f"""cuda:{gpu_id}"""   )

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

							def       __A (  self:   List[Any]		,     __A:   Union[str, Any]=0   )    ->				List[Any]:
												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.'''   )

												_A								= torch.device(f"""cuda:{gpu_id}"""   )

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

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

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

							@property
							# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
							def       __A (  self:   Union[str, Any]   )    ->				List[str]:
												if not hasattr(self.unet		,     '''_hf_hook'''   ):
																	return self.device
												for module in self.unet.modules():
																	if (
																	    hasattr(__A		,     '''_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(__A   )
							def __call__(  self:   Union[str, Any]		,     __A:   Union[torch.FloatTensor, List[torch.FloatTensor]]		,     __A:   Union[torch.FloatTensor, List[torch.FloatTensor]]		,     __A:   int = 5_12		,     __A:   int = 5_12		,     __A:   int = 1_00		,     __A:   float = 4.0		,     __A:   int = 1		,     __A:   Optional[Union[torch.Generator, List[torch.Generator]]] = None		,     __A:   Optional[torch.FloatTensor] = None		,     __A:   Optional[str] = "pil"		,     __A:   bool = True		,     )    ->				Any:
												_A								= self._execution_device

												_A								= guidance_scale > 1.0

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

												if do_classifier_free_guidance:
																	_A								= image_embeds.repeat_interleave(__A		,     dim=0   )
																	_A								= negative_image_embeds.repeat_interleave(__A		,     dim=0   )

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

												self.scheduler.set_timesteps(__A		,     device=__A   )
												_A								= self.scheduler.timesteps

												_A								= self.unet.config.in_channels

												_A							,_A								= downscale_height_and_width(__A		,     __A		,     self.movq_scale_factor   )

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

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

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

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

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

												if output_type == "pil":
																	_A								= self.numpy_to_pil(__A   )

												if not return_dict:
																	return (image,)

												return ImagePipelineOutput(images=__A   )











import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


__A      =				'\\n    Text data.\n    Second line of data.'

__A      =				'file'




@pytest.fixture(scope='''session''' )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / (FILE_PATH + '''.zstd''')
					_A								= bytes(_lowercase					,			'''utf-8''' )
					with zstd.open(_lowercase					,			'''wb''' ) as f:
										f.write(_lowercase )
					return path




@pytest.fixture
def      __A					(     _lowercase ):

					'''simple docstring'''







					with open(os.path.join(tmpfs.local_root_dir					,			_lowercase )					,			'''w''' ) as f:
										f.write(_lowercase )
					return FILE_PATH




@pytest.mark.parametrize('''compression_format'''					,			['''gzip''', '''xz''', '''zstd'''] )
def      __A					(     _lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase ):

					'''simple docstring'''







					_A								= {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
					_A								= input_paths[compression_format]
					_A								= tmp_path / '''cache'''
					_A								= DownloadConfig(cache_dir=_lowercase					,			extract_compressed_file=_lowercase )
					_A								= cached_path(_lowercase					,			download_config=_lowercase )
					with open(_lowercase ) as f:
										_A								= f.read()
					with open(_lowercase ) as f:
										_A								= f.read()
					assert extracted_file_content == expected_file_content




@pytest.mark.parametrize('''default_extracted'''					,			[True, False] )
@pytest.mark.parametrize('''default_cache_dir'''					,			[True, False] )
def      __A					(     _lowercase					,			_lowercase					,			_lowercase					,			_lowercase					,			_lowercase ):

					'''simple docstring'''







					_A								= '''custom_cache'''
					_A								= '''custom_extracted_dir'''
					_A								= tmp_path / '''custom_extracted_path'''
					if default_extracted:
										_A								= ('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
					else:
										monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''					,			_lowercase )
										monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''					,			str(_lowercase ) )
										_A								= custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

					_A								= xz_file
					_A								= (
					    DownloadConfig(extract_compressed_file=_lowercase )
					    if default_cache_dir
					    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir					,			extract_compressed_file=_lowercase )
					)
					_A								= cached_path(_lowercase					,			download_config=_lowercase )
					assert Path(_lowercase ).parent.parts[-2:] == expected




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= str(Path(_lowercase ).resolve() )
					assert cached_path(_lowercase ) == text_file
					# relative path
					_A								= str(Path(_lowercase ).resolve().relative_to(Path(os.getcwd() ) ) )
					assert cached_path(_lowercase ) == text_file




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= str(tmp_path.resolve() / '''__missing_file__.txt''' )
					with pytest.raises(_lowercase ):
										cached_path(_lowercase )
					# relative path
					_A								= '''./__missing_file__.txt'''
					with pytest.raises(_lowercase ):
										cached_path(_lowercase )




def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= get_from_cache(f"""tmp://{tmpfs_file}""" )
					with open(_lowercase ) as f:
										_A								= f.read()
					assert output_file_content == FILE_CONTENT




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     ):

					'''simple docstring'''







					with pytest.raises(_lowercase ):
										cached_path('''https://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										http_get('''https://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										http_head('''https://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										ftp_get('''ftp://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										ftp_head('''ftp://huggingface.co''' )




@patch('''datasets.config.HF_DATASETS_OFFLINE'''					,			_lowercase )
def      __A					(     _lowercase ):

					'''simple docstring'''







					_A								= tmp_path_factory.mktemp('''data''' ) / '''file.html'''
					with pytest.raises(_lowercase ):
										fsspec_get('''s3://huggingface.co'''					,			temp_file=_lowercase )
					with pytest.raises(_lowercase ):
										fsspec_head('''s3://huggingface.co''' )



