





'''simple docstring'''

import math
from collections import defaultdict
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin, SchedulerOutput





def lowerCamelCase						(     __lowerCamelCase :    Tuple					,       __lowerCamelCase :    Union[str, Any]=0.999					,       __lowerCamelCase :    Dict="cosine"					,       )						->Tuple:
    if alpha_transform_type == "cosine":

        def alpha_bar_fn(__lowerCamelCase :    int    ):
            return math.cos((t + 0.008) / 1.008 * math.pi / 2    ) ** 2

    elif alpha_transform_type == "exp":

        def alpha_bar_fn(__lowerCamelCase :    List[str]    ):
            return math.exp(t * -12.0    )

    else:
        raise ValueError(F'Unsupported alpha_tranform_type: {alpha_transform_type}'    )

    _SCREAMING_SNAKE_CASE									=			[]
    for i in range(__lowerCamelCase    ):
        _SCREAMING_SNAKE_CASE									=			i / num_diffusion_timesteps
        _SCREAMING_SNAKE_CASE									=			(i + 1) / num_diffusion_timesteps
        betas.append(min(1 - alpha_bar_fn(__lowerCamelCase    ) / alpha_bar_fn(__lowerCamelCase    )					,       __lowerCamelCase    )    )
    return torch.tensor(__lowerCamelCase					,       dtype=torch.floataa    )






class 					a_     (				snake_case_      ,    snake_case_       ):

    '''simple docstring'''
    UpperCamelCase     = [e.name for e in KarrasDiffusionSchedulers]
    UpperCamelCase     = 2


    @register_to_config
    def __init__(		self ,		A = 1000 ,		A = 0.0_0085 ,		A = 0.012 ,		A = "linear" ,		A = None ,		A = "epsilon" ,		A = "linspace" ,		A = 0 ,		)		->   Dict:
        if trained_betas is not None:
            _SCREAMING_SNAKE_CASE									=			torch.tensor(A ,		dtype=torch.floataa       )
        elif beta_schedule == "linear":
            _SCREAMING_SNAKE_CASE									=			torch.linspace(A ,		A ,		A ,		dtype=torch.floataa       )
        elif beta_schedule == "scaled_linear":
            # this schedule is very specific to the latent diffusion model.
            _SCREAMING_SNAKE_CASE									=			(
                torch.linspace(beta_start**0.5 ,		beta_end**0.5 ,		A ,		dtype=torch.floataa       ) ** 2
            )
        elif beta_schedule == "squaredcos_cap_v2":
            # Glide cosine schedule
            _SCREAMING_SNAKE_CASE									=			betas_for_alpha_bar(A       )
        else:
            raise NotImplementedError(f'{beta_schedule} does is not implemented for {self.__class__}'       )

        _SCREAMING_SNAKE_CASE									=			1.0 - self.betas
        _SCREAMING_SNAKE_CASE									=			torch.cumprod(self.alphas ,		dim=0       )

        #  set all values
        self.set_timesteps(A ,		A ,		A       )


    def 			snake_case_(		self ,		A ,		A=None       )		->   Union[str, Any]:
        if schedule_timesteps is None:
            _SCREAMING_SNAKE_CASE									=			self.timesteps

        _SCREAMING_SNAKE_CASE									=			(schedule_timesteps == timestep).nonzero()

        # The sigma index that is taken for the **very** first `step`
        # is always the second index (or the last index if there is only 1)
        # This way we can ensure we don't accidentally skip a sigma in
        # case we start in the middle of the denoising schedule (e.g. for image-to-image)
        if len(self._index_counter       ) == 0:
            _SCREAMING_SNAKE_CASE									=			1 if len(A       ) > 1 else 0
        else:
            _SCREAMING_SNAKE_CASE									=			timestep.cpu().item() if torch.is_tensor(A       ) else timestep
            _SCREAMING_SNAKE_CASE									=			self._index_counter[timestep_int]

        return indices[pos].item()


    @property
    def 			snake_case_(		self       )		->   Tuple:
        # standard deviation of the initial noise distribution
        if self.config.timestep_spacing in ["linspace", "trailing"]:
            return self.sigmas.max()

        return (self.sigmas.max() ** 2 + 1) ** 0.5


    def 			snake_case_(		self ,		A ,		A ,		)		->   torch.FloatTensor:
        _SCREAMING_SNAKE_CASE									=			self.index_for_timestep(A       )

        if self.state_in_first_order:
            _SCREAMING_SNAKE_CASE									=			self.sigmas[step_index]
        else:
            _SCREAMING_SNAKE_CASE									=			self.sigmas_interpol[step_index]

        _SCREAMING_SNAKE_CASE									=			sample / ((sigma**2 + 1) ** 0.5)
        return sample


    def 			snake_case_(		self ,		A ,		A = None ,		A = None ,		)		->   Optional[Any]:
        _SCREAMING_SNAKE_CASE									=			num_inference_steps

        _SCREAMING_SNAKE_CASE									=			num_train_timesteps or self.config.num_train_timesteps

        # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
        if self.config.timestep_spacing == "linspace":
            _SCREAMING_SNAKE_CASE									=			np.linspace(0 ,		num_train_timesteps - 1 ,		A ,		dtype=A       )[::-1].copy()
        elif self.config.timestep_spacing == "leading":
            _SCREAMING_SNAKE_CASE									=			num_train_timesteps // self.num_inference_steps
            # creates integer timesteps by multiplying by ratio
            # casting to int to avoid issues when num_inference_step is power of 3
            _SCREAMING_SNAKE_CASE									=			(np.arange(0 ,		A       ) * step_ratio).round()[::-1].copy().astype(A       )
            timesteps += self.config.steps_offset
        elif self.config.timestep_spacing == "trailing":
            _SCREAMING_SNAKE_CASE									=			num_train_timesteps / self.num_inference_steps
            # creates integer timesteps by multiplying by ratio
            # casting to int to avoid issues when num_inference_step is power of 3
            _SCREAMING_SNAKE_CASE									=			(np.arange(A ,		0 ,		-step_ratio       )).round().copy().astype(A       )
            timesteps -= 1
        else:
            raise ValueError(
                f'{self.config.timestep_spacing} is not supported. Please make sure to choose one of \'linspace\', \'leading\' or \'trailing\'.'       )

        _SCREAMING_SNAKE_CASE									=			np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5       )
        _SCREAMING_SNAKE_CASE									=			torch.from_numpy(np.log(A       )       ).to(A       )

        _SCREAMING_SNAKE_CASE									=			np.interp(A ,		np.arange(0 ,		len(A       )       ) ,		A       )
        _SCREAMING_SNAKE_CASE									=			np.concatenate([sigmas, [0.0]]       ).astype(np.floataa       )
        _SCREAMING_SNAKE_CASE									=			torch.from_numpy(A       ).to(device=A       )

        # interpolate sigmas
        _SCREAMING_SNAKE_CASE									=			sigmas.log().lerp(sigmas.roll(1       ).log() ,		0.5       ).exp()

        _SCREAMING_SNAKE_CASE									=			torch.cat([sigmas[:1], sigmas[1:].repeat_interleave(2       ), sigmas[-1:]]       )
        _SCREAMING_SNAKE_CASE									=			torch.cat(
            [sigmas_interpol[:1], sigmas_interpol[1:].repeat_interleave(2       ), sigmas_interpol[-1:]]       )

        if str(A       ).startswith("""mps"""       ):
            # mps does not support float64
            _SCREAMING_SNAKE_CASE									=			torch.from_numpy(A       ).to(A ,		dtype=torch.floataa       )
        else:
            _SCREAMING_SNAKE_CASE									=			torch.from_numpy(A       ).to(A       )

        # interpolate timesteps
        _SCREAMING_SNAKE_CASE									=			self.sigma_to_t(A       ).to(A ,		dtype=timesteps.dtype       )
        _SCREAMING_SNAKE_CASE									=			torch.stack((timesteps_interpol[1:-1, None], timesteps[1:, None]) ,		dim=-1       ).flatten()

        _SCREAMING_SNAKE_CASE									=			torch.cat([timesteps[:1], interleaved_timesteps]       )

        _SCREAMING_SNAKE_CASE									=			None

        # for exp beta schedules, such as the one for `pipeline_shap_e.py`
        # we need an index counter
        _SCREAMING_SNAKE_CASE									=			defaultdict(A       )


    def 			snake_case_(		self ,		A       )		->   Optional[Any]:
        # get log sigma
        _SCREAMING_SNAKE_CASE									=			sigma.log()

        # get distribution
        _SCREAMING_SNAKE_CASE									=			log_sigma - self.log_sigmas[:, None]

        # get sigmas range
        _SCREAMING_SNAKE_CASE									=			dists.ge(0       ).cumsum(dim=0       ).argmax(dim=0       ).clamp(max=self.log_sigmas.shape[0] - 2       )
        _SCREAMING_SNAKE_CASE									=			low_idx + 1

        _SCREAMING_SNAKE_CASE									=			self.log_sigmas[low_idx]
        _SCREAMING_SNAKE_CASE									=			self.log_sigmas[high_idx]

        # interpolate sigmas
        _SCREAMING_SNAKE_CASE									=			(low - log_sigma) / (low - high)
        _SCREAMING_SNAKE_CASE									=			w.clamp(0 ,		1       )

        # transform interpolation to time range
        _SCREAMING_SNAKE_CASE									=			(1 - w) * low_idx + w * high_idx
        _SCREAMING_SNAKE_CASE									=			t.view(sigma.shape       )
        return t


    @property
    def 			snake_case_(		self       )		->   int:
        return self.sample is None


    def 			snake_case_(		self ,		A ,		A ,		A ,		A = True ,		)		->   Union[SchedulerOutput, Tuple]:
        _SCREAMING_SNAKE_CASE									=			self.index_for_timestep(A       )

        # advance index counter by 1
        _SCREAMING_SNAKE_CASE									=			timestep.cpu().item() if torch.is_tensor(A       ) else timestep
        self._index_counter[timestep_int] += 1

        if self.state_in_first_order:
            _SCREAMING_SNAKE_CASE									=			self.sigmas[step_index]
            _SCREAMING_SNAKE_CASE									=			self.sigmas_interpol[step_index + 1]
            _SCREAMING_SNAKE_CASE									=			self.sigmas[step_index + 1]
        else:
            # 2nd order / KDPM2's method
            _SCREAMING_SNAKE_CASE									=			self.sigmas[step_index - 1]
            _SCREAMING_SNAKE_CASE									=			self.sigmas_interpol[step_index]
            _SCREAMING_SNAKE_CASE									=			self.sigmas[step_index]

        # currently only gamma=0 is supported. This usually works best anyways.
        # We can support gamma in the future but then need to scale the timestep before
        # passing it to the model which requires a change in API
        _SCREAMING_SNAKE_CASE									=			0
        _SCREAMING_SNAKE_CASE									=			sigma * (gamma + 1)  # Note: sigma_hat == sigma for now

        # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
        if self.config.prediction_type == "epsilon":
            _SCREAMING_SNAKE_CASE									=			sigma_hat if self.state_in_first_order else sigma_interpol
            _SCREAMING_SNAKE_CASE									=			sample - sigma_input * model_output
        elif self.config.prediction_type == "v_prediction":
            _SCREAMING_SNAKE_CASE									=			sigma_hat if self.state_in_first_order else sigma_interpol
            _SCREAMING_SNAKE_CASE									=			model_output * (-sigma_input / (sigma_input**2 + 1) ** 0.5) + (
                sample / (sigma_input**2 + 1)
            )
        elif self.config.prediction_type == "sample":
            raise NotImplementedError("""prediction_type not implemented yet: sample"""       )
        else:
            raise ValueError(
                f'prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`'       )

        if self.state_in_first_order:
            # 2. Convert to an ODE derivative for 1st order
            _SCREAMING_SNAKE_CASE									=			(sample - pred_original_sample) / sigma_hat
            # 3. delta timestep
            _SCREAMING_SNAKE_CASE									=			sigma_interpol - sigma_hat

            # store for 2nd order step
            _SCREAMING_SNAKE_CASE									=			sample
        else:
            # DPM-Solver-2
            # 2. Convert to an ODE derivative for 2nd order
            _SCREAMING_SNAKE_CASE									=			(sample - pred_original_sample) / sigma_interpol

            # 3. delta timestep
            _SCREAMING_SNAKE_CASE									=			sigma_next - sigma_hat

            _SCREAMING_SNAKE_CASE									=			self.sample
            _SCREAMING_SNAKE_CASE									=			None

        _SCREAMING_SNAKE_CASE									=			sample + derivative * dt

        if not return_dict:
            return (prev_sample,)

        return SchedulerOutput(prev_sample=A       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		)		->   torch.FloatTensor:
        # Make sure sigmas and timesteps have the same device and dtype as original_samples
        _SCREAMING_SNAKE_CASE									=			self.sigmas.to(device=original_samples.device ,		dtype=original_samples.dtype       )
        if original_samples.device.type == "mps" and torch.is_floating_point(A       ):
            # mps does not support float64
            _SCREAMING_SNAKE_CASE									=			self.timesteps.to(original_samples.device ,		dtype=torch.floataa       )
            _SCREAMING_SNAKE_CASE									=			timesteps.to(original_samples.device ,		dtype=torch.floataa       )
        else:
            _SCREAMING_SNAKE_CASE									=			self.timesteps.to(original_samples.device       )
            _SCREAMING_SNAKE_CASE									=			timesteps.to(original_samples.device       )

        _SCREAMING_SNAKE_CASE									=			[self.index_for_timestep(A ,		A       ) for t in timesteps]

        _SCREAMING_SNAKE_CASE									=			sigmas[step_indices].flatten()
        while len(sigma.shape       ) < len(original_samples.shape       ):
            _SCREAMING_SNAKE_CASE									=			sigma.unsqueeze(-1       )

        _SCREAMING_SNAKE_CASE									=			original_samples + noise * sigma
        return noisy_samples





    def __len__(		self       )		->   str:
        return self.config.num_train_timesteps










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


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

lowercase_					= """
    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 lowerCamelCase						(     __lowerCamelCase :    str					,       __lowerCamelCase :    Optional[Any]					,       __lowerCamelCase :    List[Any]=8    )						->Tuple:
    _SCREAMING_SNAKE_CASE									=			height // scale_factor**2
    if height % scale_factor**2 != 0:
        new_height += 1
    _SCREAMING_SNAKE_CASE									=			width // scale_factor**2
    if width % scale_factor**2 != 0:
        new_width += 1
    return new_height * scale_factor, new_width * scale_factor






class 					a_     (				snake_case_       ):

    '''simple docstring'''


    def __init__(		self ,		A ,		A ,		A ,		)		->   Union[str, Any]:
        super().__init__()

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


    def 			snake_case_(		self ,		A ,		A ,		A ,		A ,		A ,		A       )		->   Union[str, Any]:
        if latents is None:
            _SCREAMING_SNAKE_CASE									=			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}'       )
            _SCREAMING_SNAKE_CASE									=			latents.to(A       )

        _SCREAMING_SNAKE_CASE									=			latents * scheduler.init_noise_sigma
        return latents


    def 			snake_case_(		self ,		A=0       )		->   Dict:
        if is_accelerate_available():
            from accelerate import cpu_offload
        else:
            raise ImportError("""Please install accelerate via `pip install accelerate`"""       )

        _SCREAMING_SNAKE_CASE									=			torch.device(f'cuda:{gpu_id}'       )

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


    def 			snake_case_(		self ,		A=0       )		->   str:
        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."""       )

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

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

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


    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def 			snake_case_(		self       )		->   Tuple:
        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 ,		A ,		A ,		A = 512 ,		A = 512 ,		A = 100 ,		A = 4.0 ,		A = 1 ,		A = None ,		A = None ,		A = "pil" ,		A = True ,		)		->   List[str]:
        _SCREAMING_SNAKE_CASE									=			self._execution_device

        _SCREAMING_SNAKE_CASE									=			guidance_scale > 1.0

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

        if do_classifier_free_guidance:
            _SCREAMING_SNAKE_CASE									=			image_embeds.repeat_interleave(A ,		dim=0       )
            _SCREAMING_SNAKE_CASE									=			negative_image_embeds.repeat_interleave(A ,		dim=0       )

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

        self.scheduler.set_timesteps(A ,		device=A       )
        _SCREAMING_SNAKE_CASE									=			self.scheduler.timesteps

        _SCREAMING_SNAKE_CASE									=			self.unet.config.in_channels

        _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE									=			downscale_height_and_width(A ,		A ,		self.movq_scale_factor       )

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

            _SCREAMING_SNAKE_CASE									=			{"""image_embeds""": image_embeds}
            _SCREAMING_SNAKE_CASE									=			self.unet(
                sample=A ,		timestep=A ,		encoder_hidden_states=A ,		added_cond_kwargs=A ,		return_dict=A ,		)[0]

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

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

        if output_type == "pil":
            _SCREAMING_SNAKE_CASE									=			self.numpy_to_pil(A       )

        if not return_dict:
            return (image,)

        return ImagePipelineOutput(images=A       )



