
import unittest

import numpy as np
import timeout_decorator  # noqa

from transformers import OPTConfig, is_flax_available
from transformers.testing_utils import require_flax, require_sentencepiece, slow

from ...generation.test_flax_utils import FlaxGenerationTesterMixin
from ...test_modeling_flax_common import FlaxModelTesterMixin, ids_tensor


if is_flax_available():
							import os

							# The slow tests are often failing with OOM error on GPU
							# This makes JAX allocate exactly what is needed on demand, and deallocate memory that is no longer needed
							# but will be slower as stated here https://jax.readthedocs.io/en/latest/gpu_memory_allocation.html
							SCREAMING_SNAKE_CASE__							=    """platform"""

							import jax
							import jax.numpy as jnp

							from transformers import FlaxOPTForCausalLM, FlaxOPTModel, GPTaTokenizer




def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				int   ,							SCREAMING_SNAKE_CASE	:				List[str]   ,							SCREAMING_SNAKE_CASE	:				Any=None   ,							SCREAMING_SNAKE_CASE	:				Dict=None      )					->						Any:
				if attention_mask is None:
								__lowercase								= np.where(input_ids != config.pad_token_id   ,							1   ,							0      )
				return {
				    "input_ids": input_ids,
				    "attention_mask": attention_mask,
				}




@require_flax
class      A__       :





							def __init__(			self  :       Any					,			_UpperCAmelCase  :       List[str]					,			_UpperCAmelCase  :       Dict=13					,			_UpperCAmelCase  :       List[Any]=7					,			_UpperCAmelCase  :       Optional[Any]=True					,			_UpperCAmelCase  :       List[str]=False					,			_UpperCAmelCase  :       int=99					,			_UpperCAmelCase  :       List[Any]=16					,			_UpperCAmelCase  :       List[Any]=2					,			_UpperCAmelCase  :       Optional[Any]=4					,			_UpperCAmelCase  :       Tuple=4					,			_UpperCAmelCase  :       Optional[int]="gelu"					,			_UpperCAmelCase  :       Any=0.1					,			_UpperCAmelCase  :       List[Any]=0.1					,			_UpperCAmelCase  :       Union[str, Any]=20					,			_UpperCAmelCase  :       int=2					,			_UpperCAmelCase  :       str=1					,			_UpperCAmelCase  :       Any=0					,			_UpperCAmelCase  :       Optional[Any]=16					,			_UpperCAmelCase  :       Optional[Any]=16					,			_UpperCAmelCase  :       List[Any]=0.02					,			)						->		Any:



											"""simple docstring"""


											__lowercase								= parent
											__lowercase								= batch_size
											__lowercase								= seq_length
											__lowercase								= is_training
											__lowercase								= use_labels
											__lowercase								= vocab_size
											__lowercase								= hidden_size
											__lowercase								= num_hidden_layers
											__lowercase								= num_attention_heads
											__lowercase								= intermediate_size
											__lowercase								= hidden_act
											__lowercase								= hidden_dropout_prob
											__lowercase								= attention_probs_dropout_prob
											__lowercase								= max_position_embeddings
											__lowercase								= eos_token_id
											__lowercase								= pad_token_id
											__lowercase								= bos_token_id
											__lowercase								= embed_dim
											__lowercase								= word_embed_proj_dim
											__lowercase								= initializer_range
											__lowercase								= False





							def       a__			(			self  :       int    )						->		List[Any]:



											"""simple docstring"""


											__lowercase								= np.clip(ids_tensor([self.batch_size, self.seq_length - 1]					,			self.vocab_size    )					,			3					,			self.vocab_size    )
											__lowercase								= np.concatenate((input_ids, 2 * np.ones((self.batch_size, 1)					,			dtype=np.intaa    ))					,			-1    )

											__lowercase								= OPTConfig(
											    vocab_size=self.vocab_size					,			hidden_size=self.hidden_size					,			num_hidden_layers=self.num_hidden_layers					,			num_attention_heads=self.num_attention_heads					,			ffn_dim=self.intermediate_size					,			dropout=self.hidden_dropout_prob					,			attention_dropout=self.attention_probs_dropout_prob					,			max_position_embeddings=self.max_position_embeddings					,			eos_token_id=self.eos_token_id					,			bos_token_id=self.bos_token_id					,			pad_token_id=self.pad_token_id					,			embed_dim=self.embed_dim					,			is_encoder_decoder=_UpperCAmelCase					,			word_embed_proj_dim=self.word_embed_proj_dim					,			initializer_range=self.initializer_range					,			use_cache=_UpperCAmelCase					,			)
											__lowercase								= prepare_opt_inputs_dict(_UpperCAmelCase					,			_UpperCAmelCase    )
											return config, inputs_dict





							def       a__			(			self  :       Tuple    )						->		Optional[Any]:



											"""simple docstring"""


											__lowercase		, __lowercase								= self.prepare_config_and_inputs()
											return config, inputs_dict





							def       a__			(			self  :       Optional[Any]					,			_UpperCAmelCase  :       Union[str, Any]					,			_UpperCAmelCase  :       Dict					,			_UpperCAmelCase  :       str    )						->		List[Any]:



											"""simple docstring"""


											__lowercase								= 20
											__lowercase								= model_class_name(_UpperCAmelCase    )

											__lowercase								= inputs_dict['input_ids']
											__lowercase								= inputs_dict['attention_mask']

											__lowercase								= model.init_cache(input_ids.shape[0]					,			_UpperCAmelCase    )
											__lowercase								= jnp.ones((input_ids.shape[0], max_length)					,			dtype='i4'    )

											__lowercase								= jnp.broadcast_to(
											    jnp.arange(input_ids.shape[-1] - 1    )[None, :]					,			(input_ids.shape[0], input_ids.shape[-1] - 1)					,			)
											__lowercase								= model(
											    input_ids[:, :-1]					,			attention_mask=_UpperCAmelCase					,			past_key_values=_UpperCAmelCase					,			position_ids=_UpperCAmelCase					,			)

											__lowercase								= jnp.array(input_ids.shape[0] * [[input_ids.shape[-1] - 1]]					,			dtype='i4'    )
											__lowercase								= model(
											    input_ids[:, -1:]					,			attention_mask=_UpperCAmelCase					,			past_key_values=outputs_cache.past_key_values					,			position_ids=_UpperCAmelCase					,			)

											__lowercase								= model(_UpperCAmelCase    )

											__lowercase								= np.max(np.abs((outputs_cache_next[0][:, -1, :5] - outputs[0][:, -1, :5])    )    )
											self.parent.assertTrue(diff < 1e-3					,			msg=f"""Max diff is {diff}"""    )





							def       a__			(			self  :       Optional[int]					,			_UpperCAmelCase  :       List[Any]					,			_UpperCAmelCase  :       Optional[int]					,			_UpperCAmelCase  :       Any    )						->		Tuple:



											"""simple docstring"""


											__lowercase								= 20
											__lowercase								= model_class_name(_UpperCAmelCase    )

											__lowercase		, __lowercase								= (
											    inputs_dict['input_ids'],
											    inputs_dict['attention_mask'],
											)

											__lowercase								= jnp.concatenate(
											    [
											        attention_mask,
											        jnp.zeros((attention_mask.shape[0], max_length - attention_mask.shape[1])    ),
											    ]					,			axis=-1					,			)

											__lowercase								= model.init_cache(input_ids.shape[0]					,			_UpperCAmelCase    )
											__lowercase								= jnp.broadcast_to(
											    jnp.arange(input_ids.shape[-1] - 1    )[None, :]					,			(input_ids.shape[0], input_ids.shape[-1] - 1)					,			)

											__lowercase								= model(
											    input_ids[:, :-1]					,			attention_mask=_UpperCAmelCase					,			past_key_values=_UpperCAmelCase					,			position_ids=_UpperCAmelCase					,			)
											__lowercase								= jnp.array(input_ids.shape[0] * [[input_ids.shape[-1] - 1]]					,			dtype='i4'    )
											__lowercase								= model(
											    input_ids[:, -1:]					,			past_key_values=outputs_cache.past_key_values					,			attention_mask=_UpperCAmelCase					,			position_ids=_UpperCAmelCase					,			)

											__lowercase								= model(_UpperCAmelCase					,			attention_mask=_UpperCAmelCase    )

											__lowercase								= np.max(np.abs((outputs_cache_next[0][:, -1, :5] - outputs[0][:, -1, :5])    )    )
											self.parent.assertTrue(diff < 1e-3					,			msg=f"""Max diff is {diff}"""    )






@require_flax
class      A__       (						lowerCAmelCase__   ,						unittest.TestCase   ,						lowerCAmelCase__    ):
							lowerCAmelCase__		:						str         =							(FlaxOPTModel, FlaxOPTForCausalLM) if is_flax_available() else ()
							lowerCAmelCase__		:						Optional[Any]         =							() if is_flax_available() else ()





							def       a__			(			self  :       Tuple    )						->		Optional[int]:



											"""simple docstring"""


											__lowercase								= FlaxOPTModelTester(self    )





							def       a__			(			self  :       List[Any]    )						->		Union[str, Any]:



											"""simple docstring"""


											__lowercase		, __lowercase								= self.model_tester.prepare_config_and_inputs()
											for model_class in self.all_model_classes:
															self.model_tester.check_use_cache_forward(_UpperCAmelCase					,			_UpperCAmelCase					,			_UpperCAmelCase    )





							def       a__			(			self  :       Any    )						->		Union[str, Any]:



											"""simple docstring"""


											__lowercase		, __lowercase								= self.model_tester.prepare_config_and_inputs()
											for model_class in self.all_model_classes:
															self.model_tester.check_use_cache_forward_with_attn_mask(_UpperCAmelCase					,			_UpperCAmelCase					,			_UpperCAmelCase    )





							@slow
							def       a__			(			self  :       int    )						->		Tuple:



											"""simple docstring"""


											for model_class_name in self.all_model_classes:
															__lowercase								= model_class_name.from_pretrained('facebook/opt-125m'    )
															__lowercase								= np.ones((1, 1)    ) * model.config.eos_token_id
															__lowercase								= model(_UpperCAmelCase    )
															self.assertIsNotNone(_UpperCAmelCase    )






@require_sentencepiece
@require_flax
class      A__       (						unittest.TestCase    ):





							@slow
							def       a__			(			self  :       Any    )						->		List[Any]:



											"""simple docstring"""


											__lowercase								= FlaxOPTModel.from_pretrained('facebook/opt-350m'    )
											__lowercase								= jnp.array([[0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69, 4_60_78, 15_88, 2]]    )
											__lowercase								= model(input_ids=_UpperCAmelCase    ).last_hidden_state
											__lowercase								= (1, 11, 5_12)
											self.assertEqual(output.shape					,			_UpperCAmelCase    )
											__lowercase								= jnp.array(
											    [[-0.2_867, -1.9_256, -0.3_062], [-1.2_711, -0.1_337, -0.1_897], [0.4_109, 0.1_187, -1.3_142]]    )
											self.assertTrue(jnp.allclose(output[:, :3, :3]					,			_UpperCAmelCase					,			atol=4e-2    )    )






@require_flax
@slow
class      A__       (						unittest.TestCase    ):





							def       a__			(			self  :       List[Any]    )						->		List[str]:



											"""simple docstring"""


											super().setUp()
											__lowercase								= 'facebook/opt-350m'





							def       a__			(			self  :       List[Any]    )						->		str:



											"""simple docstring"""


											__lowercase								= FlaxOPTForCausalLM.from_pretrained(self.path_model    )
											__lowercase								= GPTaTokenizer.from_pretrained(self.path_model    )

											__lowercase								= [
											    'Today is a beautiful day and I want to',
											    'In the city of',
											    'Paris is the capital of France and',
											    'Computers and mobile phones have taken',
											]
											# verify that prompt without BOS token is identical to Metaseq -> add_special_tokens=False
											__lowercase								= tokenizer(_UpperCAmelCase					,			return_tensors='jax'					,			padding=_UpperCAmelCase					,			add_special_tokens=_UpperCAmelCase    )
											__lowercase								= model(inputs.input_ids					,			attention_mask=inputs.attention_mask    )[0].mean(axis=-1    )
											__lowercase								= jnp.array(
											    [
											        [1.3_851, -13.8_923, -10.5_229, -10.7_533, -0.2_309, -10.2_384, -0.5_365, -9.0_947, -5.1_670],
											        [-4.7_073, -10.6_276, -3.9_415, -21.5_242, -0.2_822, -0.2_822, -0.2_822, -0.2_822, -0.2_822],
											        [0.6_247, -3.4_229, -8.9_179, -1.4_297, -14.1_650, 1.4_146, -9.0_218, -0.2_703, -0.2_703],
											        [6.4_783, -1.9_913, -10.7_926, -2.3_336, 1.5_092, -0.9_974, -6.8_213, 1.3_477, 1.3_477],
											    ]    )
											self.assertTrue(jnp.allclose(_UpperCAmelCase					,			_UpperCAmelCase					,			atol=4e-2    )    )

											__lowercase								= jax.jit(_UpperCAmelCase    )
											__lowercase								= model(inputs.input_ids					,			attention_mask=inputs.attention_mask    )[0].mean(axis=-1    )
											self.assertTrue(jnp.allclose(_UpperCAmelCase					,			_UpperCAmelCase					,			atol=4e-2    )    )






@require_flax
@slow
class      A__       (						unittest.TestCase    ):





							@property
							def       a__			(			self  :       List[str]    )						->		Optional[int]:



											"""simple docstring"""


											return [
											    "Today is a beautiful day and I want",
											    "In the city of",
											    "Paris is the capital of France and",
											    "Computers and mobile phones have taken",
											]





							def       a__			(			self  :       Union[str, Any]    )						->		int:



											"""simple docstring"""


											__lowercase								= 'facebook/opt-125m'

											__lowercase								= [
											    'Today is a beautiful day and I want to',
											    'In the city of New York, the city',
											    'Paris is the capital of France and the capital',
											    'Computers and mobile phones have taken over the',
											]

											__lowercase								= []

											__lowercase								= FlaxOPTForCausalLM.from_pretrained(_UpperCAmelCase    )
											__lowercase								= GPTaTokenizer.from_pretrained(_UpperCAmelCase    )

											for prompt in self.prompts:
															__lowercase								= tokenizer(_UpperCAmelCase					,			return_tensors='jax'    ).input_ids

															__lowercase								= model.generate(_UpperCAmelCase					,			max_length=10    )
															__lowercase								= generated_ids[0]

															__lowercase								= tokenizer.batch_decode(_UpperCAmelCase					,			skip_special_tokens=_UpperCAmelCase    )
															predicted_outputs += generated_string

											self.assertListEqual(_UpperCAmelCase					,			_UpperCAmelCase    )





							def       a__			(			self  :       Union[str, Any]    )						->		Dict:



											"""simple docstring"""


											__lowercase								= 'facebook/opt-350m'

											__lowercase								= [
											    'Today is a beautiful day and I want to',
											    'In the city of San Francisco, the city',
											    'Paris is the capital of France and the capital',
											    'Computers and mobile phones have taken over the',
											]

											__lowercase								= []
											__lowercase								= FlaxOPTForCausalLM.from_pretrained(_UpperCAmelCase    )
											__lowercase								= GPTaTokenizer.from_pretrained(_UpperCAmelCase    )

											for prompt in self.prompts:
															__lowercase								= tokenizer(_UpperCAmelCase					,			return_tensors='jax'    ).input_ids

															__lowercase								= model.generate(_UpperCAmelCase					,			max_length=10    )
															__lowercase								= generated_ids[0]

															__lowercase								= tokenizer.batch_decode(_UpperCAmelCase					,			skip_special_tokens=_UpperCAmelCase    )
															predicted_outputs += generated_string

											self.assertListEqual(_UpperCAmelCase					,			_UpperCAmelCase    )





							def       a__			(			self  :       Dict    )						->		int:



											"""simple docstring"""


											__lowercase								= 'facebook/opt-125m'
											__lowercase								= [
											    'Today is a beautiful day and I want to thank',
											    'In the city of Rome Canaver Canaver Canaver Canaver',
											]
											__lowercase								= FlaxOPTForCausalLM.from_pretrained(_UpperCAmelCase    )
											__lowercase								= GPTaTokenizer.from_pretrained(_UpperCAmelCase    )
											__lowercase								= tokenizer(
											    [
											        'Today is a beautiful day and I want to',
											        'In the city of',
											    ]					,			return_tensors='jax'					,			padding=_UpperCAmelCase					,			)

											__lowercase								= jax.jit(model.generate    )

											__lowercase								= jit_generate(inputs['input_ids']					,			attention_mask=inputs['attention_mask']    ).sequences

											__lowercase								= tokenizer.batch_decode(_UpperCAmelCase					,			skip_special_tokens=_UpperCAmelCase    )

											self.assertIsNotNone(_UpperCAmelCase					,			_UpperCAmelCase    )





							def       a__			(			self  :       Union[str, Any]    )						->		Any:



											"""simple docstring"""


											__lowercase								= 'facebook/opt-350m'

											__lowercase								= GPTaTokenizer.from_pretrained(_UpperCAmelCase    )
											__lowercase								= FlaxOPTForCausalLM.from_pretrained(_UpperCAmelCase    )

											__lowercase								= 'left'

											# use different length sentences to test batching
											__lowercase								= [
											    'Hello, my dog is a little',
											    'Today, I',
											]

											__lowercase								= tokenizer(_UpperCAmelCase					,			return_tensors='jax'					,			padding=_UpperCAmelCase    )
											__lowercase								= inputs['input_ids']

											__lowercase								= model.generate(input_ids=_UpperCAmelCase					,			attention_mask=inputs['attention_mask']					,			trace=_UpperCAmelCase    )

											__lowercase								= tokenizer(sentences[0]					,			return_tensors='jax'    ).input_ids
											__lowercase								= model.generate(input_ids=_UpperCAmelCase    )

											__lowercase								= inputs_non_padded.shape[-1] - inputs['attention_mask'][-1].sum()
											__lowercase								= tokenizer(sentences[1]					,			return_tensors='jax'    ).input_ids
											__lowercase								= model.generate(input_ids=_UpperCAmelCase					,			max_length=model.config.max_length - num_paddings    )

											__lowercase								= tokenizer.batch_decode(outputs[0]					,			skip_special_tokens=_UpperCAmelCase    )
											__lowercase								= tokenizer.decode(output_non_padded[0][0]					,			skip_special_tokens=_UpperCAmelCase    )
											__lowercase								= tokenizer.decode(output_padded[0][0]					,			skip_special_tokens=_UpperCAmelCase    )

											__lowercase								= [
											    'Hello, my dog is a little bit of a dork.\nI\'m a little bit',
											    'Today, I was in the middle of a conversation with a friend about the',
											]
											self.assertListEqual(_UpperCAmelCase					,			_UpperCAmelCase    )
											self.assertListEqual(_UpperCAmelCase					,			[non_padded_sentence, padded_sentence]    )







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__							=    logging.get_logger(__name__)  # pylint: disable=invalid-name

SCREAMING_SNAKE_CASE__							=    """
    Examples:
        ```py
        >>> import torch
        >>> import numpy as np

        >>> from diffusers import KandinskyV22PriorEmb2EmbPipeline, KandinskyV22ControlnetImg2ImgPipeline
        >>> from transformers import pipeline
        >>> from diffusers.utils import load_image


        >>> def make_hint(image, depth_estimator):
        ...     image = depth_estimator(image)[\"depth\"]
        ...     image = np.array(image)
        ...     image = image[:, :, None]
        ...     image = np.concatenate([image, image, image], axis=2)
        ...     detected_map = torch.from_numpy(image).float() / 255.0
        ...     hint = detected_map.permute(2, 0, 1)
        ...     return hint


        >>> depth_estimator = pipeline(\"depth-estimation\")

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

        >>> pipe = KandinskyV22ControlnetImg2ImgPipeline.from_pretrained(
        ...     \"kandinsky-community/kandinsky-2-2-controlnet-depth\", torch_dtype=torch.float16
        ... )
        >>> pipe = pipe.to(\"cuda\")

        >>> img = load_image(
        ...     \"https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main\"
        ...     \"/kandinsky/cat.png\"
        ... ).resize((768, 768))


        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to(\"cuda\")

        >>> prompt = \"A robot, 4k photo\"
        >>> negative_prior_prompt = \"lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature\"

        >>> generator = torch.Generator(device=\"cuda\").manual_seed(43)

        >>> img_emb = pipe_prior(prompt=prompt, image=img, strength=0.85, generator=generator)
        >>> negative_emb = pipe_prior(prompt=negative_prior_prompt, image=img, strength=1, generator=generator)

        >>> images = pipe(
        ...     image=img,
        ...     strength=0.5,
        ...     image_embeds=img_emb.image_embeds,
        ...     negative_image_embeds=negative_emb.image_embeds,
        ...     hint=hint,
        ...     num_inference_steps=50,
        ...     generator=generator,
        ...     height=768,
        ...     width=768,
        ... ).images

        >>> images[0].save(\"robot_cat.png\")
        ```
"""




def 							__SCREAMING_SNAKE_CASE       (		SCREAMING_SNAKE_CASE	:				List[Any]   ,							SCREAMING_SNAKE_CASE	:				Dict   ,							SCREAMING_SNAKE_CASE	:				Any=8      )					->						Any:
				__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       (		SCREAMING_SNAKE_CASE	:				List[Any]   ,							SCREAMING_SNAKE_CASE	:				Any=512   ,							SCREAMING_SNAKE_CASE	:				List[str]=512      )					->						str:
				__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(SCREAMING_SNAKE_CASE   ,							[2, 0, 1]      )
				__lowercase								= torch.from_numpy(SCREAMING_SNAKE_CASE      ).unsqueeze(0      )
				return image
class      A__       (						lowerCAmelCase__    ):





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



											"""simple docstring"""


											super().__init__()

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





							def       a__			(			self  :       List[str]					,			_UpperCAmelCase  :       Dict					,			_UpperCAmelCase  :       int					,			_UpperCAmelCase  :       Dict    )						->		List[str]:



											"""simple docstring"""


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

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

											return timesteps, num_inference_steps - t_start





							def       a__			(			self  :       List[str]					,			_UpperCAmelCase  :       int					,			_UpperCAmelCase  :       List[str]					,			_UpperCAmelCase  :       Any					,			_UpperCAmelCase  :       Optional[int]					,			_UpperCAmelCase  :       Optional[Any]					,			_UpperCAmelCase  :       List[str]					,			_UpperCAmelCase  :       List[Any]=None    )						->		Optional[Any]:



											"""simple docstring"""


											if not isinstance(_UpperCAmelCase					,			(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(_UpperCAmelCase    )}"""    )

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

											__lowercase								= batch_size * num_images_per_prompt

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

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

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

															__lowercase								= self.movq.config.scaling_factor * init_latents

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

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

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

											__lowercase								= init_latents

											return latents





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



											"""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(_UpperCAmelCase					,			_UpperCAmelCase    )





							def       a__			(			self  :       str					,			_UpperCAmelCase  :       Tuple=0    )						->		List[str]:



											"""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=_UpperCAmelCase    )
															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(_UpperCAmelCase					,			_UpperCAmelCase					,			prev_module_hook=_UpperCAmelCase    )

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





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



											"""simple docstring"""


											if not hasattr(self.unet					,			'_hf_hook'    ):
															return self.device
											for module in self.unet.modules():
															if (
															    hasattr(_UpperCAmelCase					,			'_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(_UpperCAmelCase    )
							def __call__(			self  :       Optional[Any]					,			_UpperCAmelCase  :       Union[torch.FloatTensor, List[torch.FloatTensor]]					,			_UpperCAmelCase  :       Union[torch.FloatTensor, PIL.Image.Image, List[torch.FloatTensor], List[PIL.Image.Image]]					,			_UpperCAmelCase  :       Union[torch.FloatTensor, List[torch.FloatTensor]]					,			_UpperCAmelCase  :       torch.FloatTensor					,			_UpperCAmelCase  :       int = 5_12					,			_UpperCAmelCase  :       int = 5_12					,			_UpperCAmelCase  :       int = 1_00					,			_UpperCAmelCase  :       float = 4.0					,			_UpperCAmelCase  :       float = 0.3					,			_UpperCAmelCase  :       int = 1					,			_UpperCAmelCase  :       Optional[Union[torch.Generator, List[torch.Generator]]] = None					,			_UpperCAmelCase  :       Optional[str] = "pil"					,			_UpperCAmelCase  :       bool = True					,			)						->		Union[str, Any]:



											"""simple docstring"""


											__lowercase								= self._execution_device

											__lowercase								= guidance_scale > 1.0

											if isinstance(_UpperCAmelCase					,			_UpperCAmelCase    ):
															__lowercase								= torch.cat(_UpperCAmelCase					,			dim=0    )
											if isinstance(_UpperCAmelCase					,			_UpperCAmelCase    ):
															__lowercase								= torch.cat(_UpperCAmelCase					,			dim=0    )
											if isinstance(_UpperCAmelCase					,			_UpperCAmelCase    ):
															__lowercase								= torch.cat(_UpperCAmelCase					,			dim=0    )

											__lowercase								= image_embeds.shape[0]

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

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

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

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

											__lowercase								= self.movq.encode(_UpperCAmelCase    )['latents']
											__lowercase								= latents.repeat_interleave(_UpperCAmelCase					,			dim=0    )
											self.scheduler.set_timesteps(_UpperCAmelCase					,			device=_UpperCAmelCase    )
											__lowercase		, __lowercase								= self.get_timesteps(_UpperCAmelCase					,			_UpperCAmelCase					,			_UpperCAmelCase    )
											__lowercase								= timesteps[:1].repeat(batch_size * num_images_per_prompt    )
											__lowercase		, __lowercase								= downscale_height_and_width(_UpperCAmelCase					,			_UpperCAmelCase					,			self.movq_scale_factor    )
											__lowercase								= self.prepare_latents(
											    _UpperCAmelCase					,			_UpperCAmelCase					,			_UpperCAmelCase					,			_UpperCAmelCase					,			image_embeds.dtype					,			_UpperCAmelCase					,			_UpperCAmelCase    )
											for i, t in enumerate(self.progress_bar(_UpperCAmelCase    )    ):
															# 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, 'hint': hint}
															__lowercase								= self.unet(
															    sample=_UpperCAmelCase					,			timestep=_UpperCAmelCase					,			encoder_hidden_states=_UpperCAmelCase					,			added_cond_kwargs=_UpperCAmelCase					,			return_dict=_UpperCAmelCase					,			)[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(
															    _UpperCAmelCase					,			_UpperCAmelCase					,			_UpperCAmelCase					,			generator=_UpperCAmelCase					,			)[0]

											# post-processing
											__lowercase								= self.movq.decode(_UpperCAmelCase					,			force_not_quantize=_UpperCAmelCase    )['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(_UpperCAmelCase    )

											if not return_dict:
															return (image,)

											return ImagePipelineOutput(images=_UpperCAmelCase    )





