






from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from transformers import CLIPImageProcessor, CLIPVisionModel

from ...models import PriorTransformer
from ...pipelines import DiffusionPipeline
from ...schedulers import HeunDiscreteScheduler
from ...utils import (
    BaseOutput,
    is_accelerate_available,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .renderer import ShapERenderer


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

SCREAMING_SNAKE_CASE__				     =			'''\n    Examples:\n        ```py\n        >>> from PIL import Image\n        >>> import torch\n        >>> from diffusers import DiffusionPipeline\n        >>> from diffusers.utils import export_to_gif, load_image\n\n        >>> device = torch.device("cuda" if torch.cuda.is_available() else "cpu")\n\n        >>> repo = "openai/shap-e-img2img"\n        >>> pipe = DiffusionPipeline.from_pretrained(repo, torch_dtype=torch.float16)\n        >>> pipe = pipe.to(device)\n\n        >>> guidance_scale = 3.0\n        >>> image_url = "https://hf.co/datasets/diffusers/docs-images/resolve/main/shap-e/corgi.png"\n        >>> image = load_image(image_url).convert("RGB")\n\n        >>> images = pipe(\n        ...     image,\n        ...     guidance_scale=guidance_scale,\n        ...     num_inference_steps=64,\n        ...     frame_size=256,\n        ... ).images\n\n        >>> gif_path = export_to_gif(images[0], "corgi_3d.gif")\n        ```\n'''






@dataclass
class 			_UpperCamelCase(       __lowerCamelCase		):
		__SCREAMING_SNAKE_CASE  :       int							=							42






class 			_UpperCamelCase(       __lowerCamelCase		):

		def __init__(					self		:    Any    ,       SCREAMING_SNAKE_CASE__		:    Dict    ,       SCREAMING_SNAKE_CASE__		:    Optional[int]    ,       SCREAMING_SNAKE_CASE__		:    List[Any]    ,       SCREAMING_SNAKE_CASE__		:    Union[str, Any]    ,       SCREAMING_SNAKE_CASE__		:    int    ,       ):



					'''simple docstring'''



					super().__init__()

					self.register_modules(
					    prior=A_    ,       image_encoder=A_    ,       image_processor=A_    ,       scheduler=A_    ,       renderer=A_    ,       )

		def    __lowerCAmelCase  (					self		:    int    ,       SCREAMING_SNAKE_CASE__		:    Optional[Any]    ,       SCREAMING_SNAKE_CASE__		:    str    ,       SCREAMING_SNAKE_CASE__		:    Any    ,       SCREAMING_SNAKE_CASE__		:    List[Any]    ,       SCREAMING_SNAKE_CASE__		:    Union[str, Any]    ,       SCREAMING_SNAKE_CASE__		:    Union[str, Any]    ):



					'''simple docstring'''



					if latents is None:
								__a						:  Optional[Any]									=     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						:  str									=     latents.to(A_    )

					__a						:  Optional[int]									=     latents * scheduler.init_noise_sigma
					return latents

		def    __lowerCAmelCase  (					self		:    str    ,       SCREAMING_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`'    )

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

					__a						:  Any									=     [self.image_encoder, self.prior]
					for cpu_offloaded_model in models:
								if cpu_offloaded_model is not None:
											cpu_offload(A_    ,       A_    )

		@property
		def    __lowerCAmelCase  (					self		:    Dict    ):



					'''simple docstring'''



					if self.device != torch.device('meta'    ) or not hasattr(self.image_encoder    ,       '_hf_hook'    ):
								return self.device
					for module in self.image_encoder.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

		def    __lowerCAmelCase  (					self		:    Any    ,       SCREAMING_SNAKE_CASE__		:    Union[str, Any]    ,       SCREAMING_SNAKE_CASE__		:    Any    ,       SCREAMING_SNAKE_CASE__		:    List[Any]    ,       SCREAMING_SNAKE_CASE__		:    str    ,       ):



					'''simple docstring'''



					if isinstance(A_    ,       A_    ) and isinstance(image[0]    ,       torch.Tensor    ):
								__a						:  str									=     torch.cat(A_    ,       axis=0    ) if image[0].ndim == 4 else torch.stack(A_    ,       axis=0    )

					if not isinstance(A_    ,       torch.Tensor    ):
								__a						:  Any									=     self.image_processor(A_    ,       return_tensors='pt'    ).pixel_values[0].unsqueeze(0    )

					__a						:  int									=     image.to(dtype=self.image_encoder.dtype    ,       device=A_    )

					__a						:  Tuple									=     self.image_encoder(A_    )['last_hidden_state']
					__a						:  Union[str, Any]									=     image_embeds[:, 1:, :].contiguous()  # batch_size, dim, 256

					__a						:  int									=     image_embeds.repeat_interleave(A_    ,       dim=0    )

					if do_classifier_free_guidance:
								__a						:  Optional[Any]									=     torch.zeros_like(A_    )

								# For classifier free guidance, we need to do two forward passes.
								# Here we concatenate the unconditional and text embeddings into a single batch
								# to avoid doing two forward passes
								__a						:  Optional[int]									=     torch.cat([negative_image_embeds, image_embeds]    )

					return image_embeds



		@torch.no_grad()
		@replace_example_docstring(A_    )
		def __call__(					self		:    Dict    ,       SCREAMING_SNAKE_CASE__		:    Optional[Any]    ,       SCREAMING_SNAKE_CASE__		:    int = 1    ,       SCREAMING_SNAKE_CASE__		:    str = 2_5    ,       SCREAMING_SNAKE_CASE__		:    Dict = None    ,       SCREAMING_SNAKE_CASE__		:    Tuple = None    ,       SCREAMING_SNAKE_CASE__		:    Optional[Any] = 4.0    ,       SCREAMING_SNAKE_CASE__		:    int = 6_4    ,       SCREAMING_SNAKE_CASE__		:    str = "pil"    ,       SCREAMING_SNAKE_CASE__		:    Optional[Any] = True    ,       ):



					'''simple docstring'''



					if isinstance(A_    ,       PIL.Image.Image    ):
								__a						:  str									=     1
					elif isinstance(A_    ,       torch.Tensor    ):
								__a						:  List[str]									=     image.shape[0]
					elif isinstance(A_    ,       A_    ) and isinstance(image[0]    ,       (torch.Tensor, PIL.Image.Image)    ):
								__a						:  Dict									=     len(A_    )
					else:
								raise ValueError(
								    f'''`image` has to be of type `PIL.Image.Image`, `torch.Tensor`, `List[PIL.Image.Image]` or `List[torch.Tensor]` but is {type(A_    )}'''    )

					__a						:  str									=     self._execution_device

					__a						:  int									=     batch_size * num_images_per_prompt

					__a						:  Any									=     guidance_scale > 1.0
					__a						:  Dict									=     self._encode_image(A_    ,       A_    ,       A_    ,       A_    )

					# prior

					self.scheduler.set_timesteps(A_    ,       device=A_    )
					__a						:  Tuple									=     self.scheduler.timesteps

					__a						:  List[Any]									=     self.prior.config.num_embeddings
					__a						:  Any									=     self.prior.config.embedding_dim

					__a						:  int									=     self.prepare_latents(
					    (batch_size, num_embeddings * embedding_dim)    ,       image_embeds.dtype    ,       A_    ,       A_    ,       A_    ,       self.scheduler    ,       )

					# YiYi notes: for testing only to match ldm, we can directly create a latents with desired shape: batch_size, num_embeddings, embedding_dim
					__a						:  Any									=     latents.reshape(latents.shape[0]    ,       A_    ,       A_    )

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

								__a						:  Dict									=     self.prior(
								    A_    ,       timestep=A_    ,       proj_embedding=A_    ,       ).predicted_image_embedding

								# remove the variance
								__a						,					__a						:  Any									=     noise_pred.split(
								    scaled_model_input.shape[2]    ,       dim=2    )  # batch_size, num_embeddings, embedding_dim

								if do_classifier_free_guidance is not None:
											__a						,					__a						:  Dict									=     noise_pred.chunk(2    )
											__a						:  Dict									=     noise_pred_uncond + guidance_scale * (noise_pred - noise_pred_uncond)

								__a						:  Optional[int]									=     self.scheduler.step(
								    A_    ,       timestep=A_    ,       sample=A_    ,       ).prev_sample

					if output_type == "latent":
								return ShapEPipelineOutput(images=A_    )

					__a						:  str									=     []
					for i, latent in enumerate(A_    ):
								print()
								__a						:  int									=     self.renderer.decode(
								    latent[None, :]    ,       A_    ,       size=A_    ,       ray_batch_size=4_0_9_6    ,       n_coarse_samples=6_4    ,       n_fine_samples=1_2_8    ,       )

								images.append(A_    )

					__a						:  Optional[Any]									=     torch.stack(A_    )

					if output_type not in ["np", "pil"]:
								raise ValueError(f'''Only the output types `pil` and `np` are supported not output_type={output_type}'''    )

					__a						:  List[str]									=     images.cpu().numpy()

					if output_type == "pil":
								__a						:  List[str]									=     [self.numpy_to_pil(A_    ) for image in images]

					# Offload last model to CPU
					if hasattr(self    ,       'final_offload_hook'    ) and self.final_offload_hook is not None:
								self.final_offload_hook.offload()

					if not return_dict:
								return (images,)

					return ShapEPipelineOutput(images=A_    )












from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
)

@flax.struct.dataclass
class     snake_case_		(   a							):






 '''simple docstring'''





 __UpperCamelCase       =      42
 __UpperCamelCase       =      42

class     snake_case_		(   nn.Module							):






 '''simple docstring'''





 __UpperCamelCase       =      42
 __UpperCamelCase       =      (1_6, 3_2, 9_6, 2_5_6)
 __UpperCamelCase       =      jnp.floataa

 def  __UpperCAmelCase						(  self		)	->       Any:
    UpperCAmelCase__						=nn.Conv(
        self.block_out_channels[0],		kernel_size=(3, 3),		padding=((1, 1), (1, 1)),		dtype=self.dtype,		)

    UpperCAmelCase__						=[]
    for i in range(len(self.block_out_channels		) - 1		):
       UpperCAmelCase__						=self.block_out_channels[i]
       UpperCAmelCase__						=self.block_out_channels[i + 1]
       UpperCAmelCase__						=nn.Conv(
           A_,		kernel_size=(3, 3),		padding=((1, 1), (1, 1)),		dtype=self.dtype,		)
       blocks.append(A_		)
       UpperCAmelCase__						=nn.Conv(
           A_,		kernel_size=(3, 3),		strides=(2, 2),		padding=((1, 1), (1, 1)),		dtype=self.dtype,		)
       blocks.append(A_		)
    UpperCAmelCase__						=blocks

    UpperCAmelCase__						=nn.Conv(
        self.conditioning_embedding_channels,		kernel_size=(3, 3),		padding=((1, 1), (1, 1)),		kernel_init=nn.initializers.zeros_init(),		bias_init=nn.initializers.zeros_init(),		dtype=self.dtype,		)






 def __call__(  self,		A_		)	->       Any:
    UpperCAmelCase__						=self.conv_in(A_		)
    UpperCAmelCase__						=nn.silu(A_		)

    for block in self.blocks:
       UpperCAmelCase__						=block(A_		)
       UpperCAmelCase__						=nn.silu(A_		)

    UpperCAmelCase__						=self.conv_out(A_		)

    return embedding

@flax_register_to_config
class     snake_case_		(   nn.Module,		a,		a							):






 '''simple docstring'''





 __UpperCamelCase       =      3_2
 __UpperCamelCase       =      4
 __UpperCamelCase       =      (
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "DownBlock2D",
 )
 __UpperCamelCase       =      False
 __UpperCamelCase       =      (3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
 __UpperCamelCase       =      2
 __UpperCamelCase       =      8
 __UpperCamelCase       =      None
 __UpperCamelCase       =      1_2_8_0
 __UpperCamelCase       =      0.0
 __UpperCamelCase       =      False
 __UpperCamelCase       =      jnp.floataa
 __UpperCamelCase       =      True
 __UpperCamelCase       =      0
 __UpperCamelCase       =      "rgb"
 __UpperCamelCase       =      (1_6, 3_2, 9_6, 2_5_6)

 def  __UpperCAmelCase						(  self,		A_		)	->       FrozenDict:
    # init input tensors
    UpperCAmelCase__						=(1, self.in_channels, self.sample_size, self.sample_size)
    UpperCAmelCase__						=jnp.zeros(A_,		dtype=jnp.floataa		)
    UpperCAmelCase__						=jnp.ones((1,),		dtype=jnp.intaa		)
    UpperCAmelCase__						=jnp.zeros((1, 1, self.cross_attention_dim),		dtype=jnp.floataa		)
    UpperCAmelCase__						=(1, 3, self.sample_size * 8, self.sample_size * 8)
    UpperCAmelCase__						=jnp.zeros(A_,		dtype=jnp.floataa		)

    UpperCAmelCase__			,							UpperCAmelCase__						=jax.random.split(A_		)
    UpperCAmelCase__						={"params": params_rng, "dropout": dropout_rng}

    return self.init(A_,		A_,		A_,		A_,		A_		)["params"]

 def  __UpperCAmelCase						(  self		)	->       int:
    UpperCAmelCase__						=self.block_out_channels
    UpperCAmelCase__						=block_out_channels[0] * 4

    # If `num_attention_heads` is not defined (which is the case for most models)
    # it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
    # The reason for this behavior is to correct for incorrectly named variables that were introduced
    # when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
    # Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
    # which is why we correct for the naming here.
    UpperCAmelCase__						=self.num_attention_heads or self.attention_head_dim

    # input
    UpperCAmelCase__						=nn.Conv(
        block_out_channels[0],		kernel_size=(3, 3),		strides=(1, 1),		padding=((1, 1), (1, 1)),		dtype=self.dtype,		)

    # time
    UpperCAmelCase__						=FlaxTimesteps(
        block_out_channels[0],		flip_sin_to_cos=self.flip_sin_to_cos,		freq_shift=self.config.freq_shift		)
    UpperCAmelCase__						=FlaxTimestepEmbedding(A_,		dtype=self.dtype		)

    UpperCAmelCase__						=FlaxControlNetConditioningEmbedding(
        conditioning_embedding_channels=block_out_channels[0],		block_out_channels=self.conditioning_embedding_out_channels,		)

    UpperCAmelCase__						=self.only_cross_attention
    if isinstance(A_,		A_		):
       UpperCAmelCase__						=(only_cross_attention,) * len(self.down_block_types		)

    if isinstance(A_,		A_		):
       UpperCAmelCase__						=(num_attention_heads,) * len(self.down_block_types		)

    # down
    UpperCAmelCase__						=[]
    UpperCAmelCase__						=[]

    UpperCAmelCase__						=block_out_channels[0]

    UpperCAmelCase__						=nn.Conv(
        A_,		kernel_size=(1, 1),		padding="VALID",		kernel_init=nn.initializers.zeros_init(),		bias_init=nn.initializers.zeros_init(),		dtype=self.dtype,		)
    controlnet_down_blocks.append(A_		)

    for i, down_block_type in enumerate(self.down_block_types		):
       UpperCAmelCase__						=output_channel
       UpperCAmelCase__						=block_out_channels[i]
       UpperCAmelCase__						=i == len(A_		) - 1

       if down_block_type == "CrossAttnDownBlock2D":
          UpperCAmelCase__						=FlaxCrossAttnDownBlockaD(
              in_channels=A_,		out_channels=A_,		dropout=self.dropout,		num_layers=self.layers_per_block,		num_attention_heads=num_attention_heads[i],		add_downsample=not is_final_block,		use_linear_projection=self.use_linear_projection,		only_cross_attention=only_cross_attention[i],		dtype=self.dtype,		)
       else:
          UpperCAmelCase__						=FlaxDownBlockaD(
              in_channels=A_,		out_channels=A_,		dropout=self.dropout,		num_layers=self.layers_per_block,		add_downsample=not is_final_block,		dtype=self.dtype,		)

       down_blocks.append(A_		)

       for _ in range(self.layers_per_block		):
          UpperCAmelCase__						=nn.Conv(
              A_,		kernel_size=(1, 1),		padding="VALID",		kernel_init=nn.initializers.zeros_init(),		bias_init=nn.initializers.zeros_init(),		dtype=self.dtype,		)
          controlnet_down_blocks.append(A_		)

       if not is_final_block:
          UpperCAmelCase__						=nn.Conv(
              A_,		kernel_size=(1, 1),		padding="VALID",		kernel_init=nn.initializers.zeros_init(),		bias_init=nn.initializers.zeros_init(),		dtype=self.dtype,		)
          controlnet_down_blocks.append(A_		)

    UpperCAmelCase__						=down_blocks
    UpperCAmelCase__						=controlnet_down_blocks

    # mid
    UpperCAmelCase__						=block_out_channels[-1]
    UpperCAmelCase__						=FlaxUNetMidBlockaDCrossAttn(
        in_channels=A_,		dropout=self.dropout,		num_attention_heads=num_attention_heads[-1],		use_linear_projection=self.use_linear_projection,		dtype=self.dtype,		)

    UpperCAmelCase__						=nn.Conv(
        A_,		kernel_size=(1, 1),		padding="VALID",		kernel_init=nn.initializers.zeros_init(),		bias_init=nn.initializers.zeros_init(),		dtype=self.dtype,		)






 def __call__(  self,		A_,		A_,		A_,		A_,		A_ = 1.0,		A_ = True,		A_ = False,		)	->       Union[FlaxControlNetOutput, Tuple]:
    UpperCAmelCase__						=self.controlnet_conditioning_channel_order
    if channel_order == "bgr":
       UpperCAmelCase__						=jnp.flip(A_,		axis=1		)

    # 1. time
    if not isinstance(A_,		jnp.ndarray		):
       UpperCAmelCase__						=jnp.array([timesteps],		dtype=jnp.intaa		)
    elif isinstance(A_,		jnp.ndarray		) and len(timesteps.shape		) == 0:
       UpperCAmelCase__						=timesteps.astype(dtype=jnp.floataa		)
       UpperCAmelCase__						=jnp.expand_dims(A_,		0		)

    UpperCAmelCase__						=self.time_proj(A_		)
    UpperCAmelCase__						=self.time_embedding(A_		)

    # 2. pre-process
    UpperCAmelCase__						=jnp.transpose(A_,		(0, 2, 3, 1)		)
    UpperCAmelCase__						=self.conv_in(A_		)

    UpperCAmelCase__						=jnp.transpose(A_,		(0, 2, 3, 1)		)
    UpperCAmelCase__						=self.controlnet_cond_embedding(A_		)
    sample += controlnet_cond

    # 3. down
    UpperCAmelCase__						=(sample,)
    for down_block in self.down_blocks:
       if isinstance(A_,		A_		):
          UpperCAmelCase__			,							UpperCAmelCase__						=down_block(A_,		A_,		A_,		deterministic=not train		)
       else:
          UpperCAmelCase__			,							UpperCAmelCase__						=down_block(A_,		A_,		deterministic=not train		)
       down_block_res_samples += res_samples

    # 4. mid
    UpperCAmelCase__						=self.mid_block(A_,		A_,		A_,		deterministic=not train		)

    # 5. contronet blocks
    UpperCAmelCase__						=()
    for down_block_res_sample, controlnet_block in zip(A_,		self.controlnet_down_blocks		):
       UpperCAmelCase__						=controlnet_block(A_		)
       controlnet_down_block_res_samples += (down_block_res_sample,)

    UpperCAmelCase__						=controlnet_down_block_res_samples

    UpperCAmelCase__						=self.controlnet_mid_block(A_		)

    # 6. scaling
    UpperCAmelCase__						=[sample * conditioning_scale for sample in down_block_res_samples]
    mid_block_res_sample *= conditioning_scale

    if not return_dict:
       return (down_block_res_samples, mid_block_res_sample)

    return FlaxControlNetOutput(
        down_block_res_samples=A_,		mid_block_res_sample=A_		)



