

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


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

A_	:			List[Any]  	=      '\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    _a						(__magic_name__    ):


					'''simple docstring'''






					UpperCAmelCase__:       Union[PIL.Image.Image, np.ndarray]



class    _a						(__magic_name__    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	):
						super().__init__()

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






					def   __A      (			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__      ):
						if latents is None:
							A__							:   Union[str, 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__							:   Dict					       =		latents.to(A__      )

						A__							:   Dict					       =		latents * scheduler.init_noise_sigma
						return latents






					def   __A      (			self   ,	A__=0      ):
						if is_accelerate_available():
							from accelerate import cpu_offload
						else:
							raise ImportError("""Please install accelerate via `pip install accelerate`"""      )

						A__							:   Optional[int]					       =		torch.device(F"""cuda:{gpu_id}"""      )

						A__							:   List[str]					       =		[self.image_encoder, self.prior]
						for cpu_offloaded_model in models:
							if cpu_offloaded_model is not None:
								cpu_offload(A__   ,	A__      )






					@property
					def   __A      (			self      ):
						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   __A      (			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	):
						if isinstance(A__   ,	A__      ) and isinstance(image[0]   ,	torch.Tensor      ):
							A__							:   List[Any]					       =		torch.cat(A__   ,	axis=0      ) if image[0].ndim == 4 else torch.stack(A__   ,	axis=0      )

						if not isinstance(A__   ,	torch.Tensor      ):
							A__							:   Union[str, Any]					       =		self.image_processor(A__   ,	return_tensors="""pt"""      ).pixel_values[0].unsqueeze(0      )

						A__							:   Optional[int]					       =		image.to(dtype=self.image_encoder.dtype   ,	device=A__      )

						A__							:   Tuple					       =		self.image_encoder(A__      )["""last_hidden_state"""]
						A__							:   Optional[int]					       =		image_embeds[:, 1:, :].contiguous()  # batch_size, dim, 256

						A__							:   List[Any]					       =		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__							:   List[str]					       =		torch.cat([negative_image_embeds, image_embeds]      )

						return image_embeds






					@torch.no_grad()
					@replace_example_docstring(A__      )
					def __call__(			self   ,	A__   ,	A__ = 1   ,	A__ = 25   ,	A__ = None   ,	A__ = None   ,	A__ = 4.0   ,	A__ = 64   ,	A__ = "pil"   ,	A__ = True   ,	):

						if isinstance(A__   ,	PIL.Image.Image      ):
							A__							:   Dict					       =		1
						elif isinstance(A__   ,	torch.Tensor      ):
							A__							:   str					       =		image.shape[0]
						elif isinstance(A__   ,	A__      ) and isinstance(image[0]   ,	(torch.Tensor, PIL.Image.Image)      ):
							A__							:   int					       =		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__							:   Any					       =		self._execution_device

						A__							:   Tuple					       =		batch_size * num_images_per_prompt

						A__							:   Optional[Any]					       =		guidance_scale > 1.0
						A__							:   int					       =		self._encode_image(A__   ,	A__   ,	A__   ,	A__      )

						# prior

						self.scheduler.set_timesteps(A__   ,	device=A__      )
						A__							:   List[str]					       =		self.scheduler.timesteps

						A__							:   Union[str, Any]					       =		self.prior.config.num_embeddings
						A__							:   Dict					       =		self.prior.config.embedding_dim

						A__							:   Optional[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__							:   Dict					       =		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__							:   Optional[int]					       =		torch.cat([latents] * 2      ) if do_classifier_free_guidance else latents
							A__							:   str					       =		self.scheduler.scale_model_input(A__   ,	A__      )

							A__							:   str					       =		self.prior(
							    A__   ,	timestep=A__   ,	proj_embedding=A__   ,	).predicted_image_embedding

							# remove the variance
							A__       ,       A__							:   str					       =		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__							:   Optional[Any]					       =		noise_pred.chunk(2      )
								A__							:   Optional[Any]					       =		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__							:   List[str]					       =		[]
						for i, latent in enumerate(A__      ):
							print()
							A__							:   int					       =		self.renderer.decode(
							    latent[None, :]   ,	A__   ,	size=A__   ,	ray_batch_size=4096   ,	n_coarse_samples=64   ,	n_fine_samples=128   ,	)

							images.append(A__      )

						A__							:   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__							:   Dict					       =		[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 __future__ import annotations

import random
import unittest

from transformers import TransfoXLConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
	import tensorflow as tf

	from transformers import (
	    TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST,
	    TFTransfoXLForSequenceClassification,
	    TFTransfoXLLMHeadModel,
	    TFTransfoXLModel,
	)



class    _a						:


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	):
						A__							:   Any					       =		parent
						A__							:   Any					       =		13
						A__							:   Optional[Any]					       =		7
						A__							:   Union[str, Any]					       =		30
						A__							:   str					       =		self.seq_length + self.mem_len
						A__							:   Dict					       =		15
						A__							:   int					       =		True
						A__							:   Tuple					       =		True
						A__							:   Union[str, Any]					       =		99
						A__							:   Optional[Any]					       =		[10, 50, 80]
						A__							:   str					       =		32
						A__							:   Tuple					       =		32
						A__							:   Union[str, Any]					       =		4
						A__							:   Optional[Any]					       =		8
						A__							:   int					       =		128
						A__							:   List[Any]					       =		2
						A__							:   List[str]					       =		2
						A__							:   int					       =		None
						A__							:   List[str]					       =		1
						A__							:   Union[str, Any]					       =		0
						A__							:   List[str]					       =		3
						A__							:   int					       =		self.vocab_size - 1
						A__							:   Optional[Any]					       =		0.0_1






					def   __A      (			self      ):
						A__							:   List[str]					       =		ids_tensor([self.batch_size, self.seq_length]   ,	self.vocab_size      )
						A__							:   Tuple					       =		ids_tensor([self.batch_size, self.seq_length]   ,	self.vocab_size      )

						A__							:   Optional[Any]					       =		None
						if self.use_labels:
							A__							:   Union[str, Any]					       =		ids_tensor([self.batch_size, self.seq_length]   ,	self.vocab_size      )

						A__							:   Any					       =		TransfoXLConfig(
						    vocab_size=self.vocab_size   ,	mem_len=self.mem_len   ,	clamp_len=self.clamp_len   ,	cutoffs=self.cutoffs   ,	d_model=self.hidden_size   ,	d_embed=self.d_embed   ,	n_head=self.num_attention_heads   ,	d_head=self.d_head   ,	d_inner=self.d_inner   ,	div_val=self.div_val   ,	n_layer=self.num_hidden_layers   ,	eos_token_id=self.eos_token_id   ,	pad_token_id=self.vocab_size - 1   ,	init_range=self.init_range   ,	num_labels=self.num_labels   ,	)

						return (config, input_ids_a, input_ids_a, lm_labels)






					def   __A      (			self      ):
						random.seed(self.seed      )
						tf.random.set_seed(self.seed      )






					def   __A      (			self   ,	A__   ,	A__   ,	A__   ,	A__      ):
						A__							:   Dict					       =		TFTransfoXLModel(A__      )

						A__       ,       A__							:   Tuple					       =		model(A__      ).to_tuple()

						A__							:   List[str]					       =		{"""input_ids""": input_ids_a, """mems""": mems_a}

						A__       ,       A__							:   str					       =		model(A__      ).to_tuple()

						self.parent.assertEqual(hidden_states_a.shape   ,	(self.batch_size, self.seq_length, self.hidden_size)      )
						self.parent.assertEqual(hidden_states_a.shape   ,	(self.batch_size, self.seq_length, self.hidden_size)      )
						self.parent.assertListEqual(
						    [mem.shape for mem in mems_a]   ,	[(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers   ,	)
						self.parent.assertListEqual(
						    [mem.shape for mem in mems_a]   ,	[(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers   ,	)






					def   __A      (			self   ,	A__   ,	A__   ,	A__   ,	A__      ):
						A__							:   Optional[int]					       =		TFTransfoXLLMHeadModel(A__      )

						A__       ,       A__							:   int					       =		model(A__      ).to_tuple()

						A__							:   int					       =		{"""input_ids""": input_ids_a, """labels""": lm_labels}
						A__       ,       A__							:   Optional[Any]					       =		model(A__      ).to_tuple()

						A__       ,       A__							:   Union[str, Any]					       =		model([input_ids_a, mems_a]      ).to_tuple()

						A__							:   Any					       =		{"""input_ids""": input_ids_a, """mems""": mems_a, """labels""": lm_labels}

						A__       ,       A__							:   Tuple					       =		model(A__      ).to_tuple()

						self.parent.assertEqual(lm_logits_a.shape   ,	(self.batch_size, self.seq_length, self.vocab_size)      )
						self.parent.assertListEqual(
						    [mem.shape for mem in mems_a]   ,	[(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers   ,	)

						self.parent.assertEqual(lm_logits_a.shape   ,	(self.batch_size, self.seq_length, self.vocab_size)      )
						self.parent.assertListEqual(
						    [mem.shape for mem in mems_a]   ,	[(self.mem_len, self.batch_size, self.hidden_size)] * self.num_hidden_layers   ,	)






					def   __A      (			self   ,	A__   ,	A__   ,	A__   ,	A__      ):
						A__							:   Any					       =		TFTransfoXLForSequenceClassification(A__      )
						A__							:   Optional[Any]					       =		model(A__      )
						self.parent.assertEqual(result.logits.shape   ,	(self.batch_size, self.num_labels)      )






					def   __A      (			self      ):
						A__							:   Optional[Any]					       =		self.prepare_config_and_inputs()
						((A__)       ,       (A__)       ,       (A__)       ,       (A__))							:   List[Any]					       =		config_and_inputs
						A__							:   int					       =		{"""input_ids""": input_ids_a}
						return config, inputs_dict



@require_tf
class    _a						(__magic_name__		,       __magic_name__		,       unittest.TestCase    ):


					'''simple docstring'''






					UpperCAmelCase__:       List[Any]  							=	(
					    (TFTransfoXLModel, TFTransfoXLLMHeadModel, TFTransfoXLForSequenceClassification) if is_tf_available() else ()
					)
					UpperCAmelCase__:       Optional[Any]  							=	() if is_tf_available() else ()
					UpperCAmelCase__:       int  							=	(
					    {
					        '''feature-extraction''': TFTransfoXLModel,
					        '''text-classification''': TFTransfoXLForSequenceClassification,
					        '''text-generation''': TFTransfoXLLMHeadModel,
					        '''zero-shot''': TFTransfoXLForSequenceClassification,
					    }
					    if is_tf_available()
					    else {}
					)
					# TODO: add this test when TFTransfoXLLMHead has a linear output layer implemented
					UpperCAmelCase__:       Optional[int]  							=	False
					UpperCAmelCase__:       Optional[int]  							=	False
					UpperCAmelCase__:       Tuple  							=	False
					UpperCAmelCase__:       List[str]  							=	False






					def   __A      (			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__      ):
						if pipeline_test_casse_name == "TextGenerationPipelineTests":
							# Get `ValueError: AttributeError: 'NoneType' object has no attribute 'new_ones'` or `AssertionError`.
							# `TransfoXLConfig` was never used in pipeline tests: cannot create a simple
							# tokenizer.
							return True

						return False






					def   __A      (			self      ):
						A__							:   Tuple					       =		TFTransfoXLModelTester(self      )
						A__							:   List[Any]					       =		ConfigTester(self   ,	config_class=A__   ,	d_embed=37      )






					def   __A      (			self      ):
						self.config_tester.run_common_tests()






					def   __A      (			self      ):
						self.model_tester.set_seed()
						A__							:   List[str]					       =		self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_transfo_xl_model(*A__      )






					def   __A      (			self      ):
						self.model_tester.set_seed()
						A__							:   Any					       =		self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_transfo_xl_lm_head(*A__      )






					def   __A      (			self      ):
						A__							:   List[str]					       =		self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_transfo_xl_for_sequence_classification(*A__      )






					def   __A      (			self      ):
						A__       ,       A__							:   int					       =		self.model_tester.prepare_config_and_inputs_for_common()
						A__							:   Union[str, Any]					       =		[TFTransfoXLForSequenceClassification]

						for model_class in self.all_model_classes:
							A__							:   Any					       =		model_class(A__      )
							assert isinstance(model.get_input_embeddings()   ,	tf.keras.layers.Layer      )
							if model_class in list_other_models_with_output_ebd:
								A__							:   Optional[Any]					       =		model.get_output_embeddings()
								assert isinstance(A__   ,	tf.keras.layers.Layer      )
								A__							:   Tuple					       =		model.get_bias()
								assert name is None
							else:
								A__							:   Dict					       =		model.get_output_embeddings()
								assert x is None
								A__							:   int					       =		model.get_bias()
								assert name is None






					def   __A      (			self      ):
						# TODO JP: Make TransfoXL XLA compliant
						pass






					@slow
					def   __A      (			self      ):
						for model_name in TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							A__							:   List[Any]					       =		TFTransfoXLModel.from_pretrained(A__      )
							self.assertIsNotNone(A__      )






					@unittest.skip(reason="""This model doesn't play well with fit() due to not returning a single loss."""      )
					def   __A      (			self      ):
						pass



@require_tf
class    _a						(unittest.TestCase    ):


					'''simple docstring'''






					@unittest.skip("""Skip test until #12651 is resolved."""      )
					@slow
					def   __A      (			self      ):
						A__							:   List[Any]					       =		TFTransfoXLLMHeadModel.from_pretrained("""transfo-xl-wt103"""      )
						# fmt: off
						A__							:   Tuple					       =		tf.convert_to_tensor([[33,1297,2,1,1009,4,1109,1_1739,4762,358,5,25,245,22,1706,17,2_0098,5,3215,21,37,1110,3,13,1041,4,24,603,490,2,7_1477,2_0098,10_4447,2,2_0961,1,2604,4,1,329,3,6224,831,1_6002,2,8,603,7_8967,2_9546,23,803,20,25,416,5,8,232,4,277,6,1855,4601,3,2_9546,54,8,3609,5,5_7211,49,4,1,277,18,8,1755,1_5691,3,341,25,416,693,4_2573,71,17,401,94,31,1_7919,2,2_9546,7873,18,1,435,23,1_1011,755,5,5167,3,7983,98,84,2,2_9546,3267,8,3609,4,1,4865,1075,2,6087,71,6,346,8,5854,3,2_9546,824,1400,1868,2,19,160,2,311,8,5496,2,2_0920,17,25,1_5097,3,24,24,0]]   ,	dtype=tf.intaa      )  # noqa: E231
						# fmt: on
						#  In 1991 , the remains of Russian Tsar Nicholas II and his family
						#  ( except for Alexei and Maria ) are discovered .
						#  The voice of Nicholas's young son , Tsarevich Alexei Nikolaevich , narrates the
						#  remainder of the story . 1883 Western Siberia ,
						#  a young Grigori Rasputin is asked by his father and a group of men to perform magic .
						#  Rasputin has a vision and denounces one of the men as a horse thief . Although his
						#  father initially slaps him for making such an accusation , Rasputin watches as the
						#  man is chased outside and beaten . Twenty years later , Rasputin sees a vision of
						#  the Virgin Mary , prompting him to become a priest . Rasputin quickly becomes famous ,
						#  with people , even a bishop , begging for his blessing . <eod> </s> <eos>

						# fmt: off
						A__							:   Dict					       =		[33,1297,2,1,1009,4,1109,1_1739,4762,358,5,25,245,22,1706,17,2_0098,5,3215,21,37,1110,3,13,1041,4,24,603,490,2,7_1477,2_0098,10_4447,2,2_0961,1,2604,4,1,329,3,6224,831,1_6002,2,8,603,7_8967,2_9546,23,803,20,25,416,5,8,232,4,277,6,1855,4601,3,2_9546,54,8,3609,5,5_7211,49,4,1,277,18,8,1755,1_5691,3,341,25,416,693,4_2573,71,17,401,94,31,1_7919,2,2_9546,7873,18,1,435,23,1_1011,755,5,5167,3,7983,98,84,2,2_9546,3267,8,3609,4,1,4865,1075,2,6087,71,6,346,8,5854,3,2_9546,824,1400,1868,2,19,160,2,311,8,5496,2,2_0920,17,25,1_5097,3,24,24,0,33,1,1857,2,1,1009,4,1109,1_1739,4762,358,5,25,245,28,1110,3,13,1041,4,24,603,490,2,7_1477,2_0098,10_4447,2,2_0961,1,2604,4,1,329,3,0]  # noqa: E231
						# fmt: on
						#  In 1991, the remains of Russian Tsar Nicholas II and his family (
						#  except for Alexei and Maria ) are discovered. The voice of young son,
						#  Tsarevich Alexei Nikolaevich, narrates the remainder of the story.
						#  1883 Western Siberia, a young Grigori Rasputin is asked by his father
						#  and a group of men to perform magic. Rasputin has a vision and
						#  denounces one of the men as a horse thief. Although his father initially
						#  slaps him for making such an accusation, Rasputin watches as the man
						#  is chased outside and beaten. Twenty years later, Rasputin sees a vision
						#  of the Virgin Mary, prompting him to become a priest.
						#  Rasputin quickly becomes famous, with people, even a bishop, begging for
						#  his blessing. <unk> <unk> <eos> In the 1990s, the remains of Russian Tsar
						# Nicholas II and his family were discovered. The voice of <unk> young son,
						# Tsarevich Alexei Nikolaevich, narrates the remainder of the story.<eos>

						A__							:   Any					       =		model.generate(A__   ,	max_length=200   ,	do_sample=A__      )
						self.assertListEqual(output_ids[0].numpy().tolist()   ,	A__      )



