






'''simple docstring'''







from typing import List, Optional, Union

import torch
from transformers import (
    XLMRobertaTokenizer,
)

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .text_encoder import MultilingualCLIP


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

UpperCamelCase    :      Tuple  			=		'\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained("kandinsky-community/Kandinsky-2-1-prior")\n        >>> pipe_prior.to("cuda")\n\n        >>> prompt = "red cat, 4k photo"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> negative_image_emb = out.negative_image_embeds\n\n        >>> pipe = KandinskyPipeline.from_pretrained("kandinsky-community/kandinsky-2-1")\n        >>> pipe.to("cuda")\n\n        >>> image = pipe(\n        ...     prompt,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=negative_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ... ).images\n\n        >>> image[0].save("cat.png")\n        ```\n'
def 						A__			(				__lowerCAmelCase		:			Optional[int]   ,    __lowerCAmelCase		:			str   ,    __lowerCAmelCase		:			Tuple=8						):
						lowerCamelCase__         =  h // scale_factor**2
						if h % scale_factor**2 != 0:
												new_h += 1
						lowerCamelCase__         =  w // scale_factor**2
						if w % scale_factor**2 != 0:
												new_w += 1
						return new_h * scale_factor, new_w * scale_factor





class 		UpperCamelCase__    (a			):




		'''simple docstring'''





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

								self.register_modules(
								    text_encoder=_lowerCAmelCase   ,tokenizer=_lowerCAmelCase   ,unet=_lowerCAmelCase   ,scheduler=_lowerCAmelCase   ,movq=_lowerCAmelCase   ,)
								lowerCamelCase__         =  2 ** (len(self.movq.config.block_out_channels       ) - 1)






		def 			UpperCamelCase_					(      self   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase       ):
								if latents is None:
														lowerCamelCase__         =  randn_tensor(_lowerCAmelCase   ,generator=_lowerCAmelCase   ,device=_lowerCAmelCase   ,dtype=_lowerCAmelCase       )
								else:
														if latents.shape != shape:
																				raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {shape}'''       )
														lowerCamelCase__         =  latents.to(_lowerCAmelCase       )

								lowerCamelCase__         =  latents * scheduler.init_noise_sigma
								return latents






		def 			UpperCamelCase_					(      self   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase=None   ,):
								lowerCamelCase__         =  len(_lowerCAmelCase       ) if isinstance(_lowerCAmelCase   ,_lowerCAmelCase       ) else 1
								# get prompt text embeddings
								lowerCamelCase__         =  self.tokenizer(
								    _lowerCAmelCase   ,padding="""max_length"""   ,truncation=_lowerCAmelCase   ,max_length=77   ,return_attention_mask=_lowerCAmelCase   ,add_special_tokens=_lowerCAmelCase   ,return_tensors="""pt"""   ,)

								lowerCamelCase__         =  text_inputs.input_ids
								lowerCamelCase__         =  self.tokenizer(_lowerCAmelCase   ,padding="""longest"""   ,return_tensors="""pt"""       ).input_ids

								if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(_lowerCAmelCase   ,_lowerCAmelCase       ):
														lowerCamelCase__         =  self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]       )
														logger.warning(
														    """The following part of your input was truncated because CLIP can only handle sequences up to"""
														    F''' {self.tokenizer.model_max_length} tokens: {removed_text}'''       )

								lowerCamelCase__         =  text_input_ids.to(_lowerCAmelCase       )
								lowerCamelCase__         =  text_inputs.attention_mask.to(_lowerCAmelCase       )

								lowerCamelCase__       ,			lowerCamelCase__         =  self.text_encoder(
								    input_ids=_lowerCAmelCase   ,attention_mask=_lowerCAmelCase       )

								lowerCamelCase__         =  prompt_embeds.repeat_interleave(_lowerCAmelCase   ,dim=0       )
								lowerCamelCase__         =  text_encoder_hidden_states.repeat_interleave(_lowerCAmelCase   ,dim=0       )
								lowerCamelCase__         =  text_mask.repeat_interleave(_lowerCAmelCase   ,dim=0       )

								if do_classifier_free_guidance:
														lowerCamelCase__         =  42
														if negative_prompt is None:
																				lowerCamelCase__         =  [""""""] * batch_size
														elif type(_lowerCAmelCase       ) is not type(_lowerCAmelCase       ):
																				raise TypeError(
																				    F'''`negative_prompt` should be the same type to `prompt`, but got {type(_lowerCAmelCase       )} !='''
																				    F''' {type(_lowerCAmelCase       )}.'''       )
														elif isinstance(_lowerCAmelCase   ,_lowerCAmelCase       ):
																				lowerCamelCase__         =  [negative_prompt]
														elif batch_size != len(_lowerCAmelCase       ):
																				raise ValueError(
																				    F'''`negative_prompt`: {negative_prompt} has batch size {len(_lowerCAmelCase       )}, but `prompt`:'''
																				    F''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
																				    """ the batch size of `prompt`."""       )
														else:
																				lowerCamelCase__         =  negative_prompt

														lowerCamelCase__         =  self.tokenizer(
														    _lowerCAmelCase   ,padding="""max_length"""   ,max_length=77   ,truncation=_lowerCAmelCase   ,return_attention_mask=_lowerCAmelCase   ,add_special_tokens=_lowerCAmelCase   ,return_tensors="""pt"""   ,)
														lowerCamelCase__         =  uncond_input.input_ids.to(_lowerCAmelCase       )
														lowerCamelCase__         =  uncond_input.attention_mask.to(_lowerCAmelCase       )

														lowerCamelCase__       ,			lowerCamelCase__         =  self.text_encoder(
														    input_ids=_lowerCAmelCase   ,attention_mask=_lowerCAmelCase       )

														# duplicate unconditional embeddings for each generation per prompt, using mps friendly method

														lowerCamelCase__         =  negative_prompt_embeds.shape[1]
														lowerCamelCase__         =  negative_prompt_embeds.repeat(1   ,_lowerCAmelCase       )
														lowerCamelCase__         =  negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,_lowerCAmelCase       )

														lowerCamelCase__         =  uncond_text_encoder_hidden_states.shape[1]
														lowerCamelCase__         =  uncond_text_encoder_hidden_states.repeat(1   ,_lowerCAmelCase   ,1       )
														lowerCamelCase__         =  uncond_text_encoder_hidden_states.view(
														    batch_size * num_images_per_prompt   ,_lowerCAmelCase   ,-1       )
														lowerCamelCase__         =  uncond_text_mask.repeat_interleave(_lowerCAmelCase   ,dim=0       )

														# done duplicates

														# 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
														lowerCamelCase__         =  torch.cat([negative_prompt_embeds, prompt_embeds]       )
														lowerCamelCase__         =  torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]       )

														lowerCamelCase__         =  torch.cat([uncond_text_mask, text_mask]       )

								return prompt_embeds, text_encoder_hidden_states, text_mask






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

								lowerCamelCase__         =  torch.device(F'''cuda:{gpu_id}'''       )

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






		def 			UpperCamelCase_					(      self   ,_lowerCAmelCase=0       ):
								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."""       )

								lowerCamelCase__         =  torch.device(F'''cuda:{gpu_id}'''       )

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

								lowerCamelCase__         =  None
								for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
														lowerCamelCase__       ,			lowerCamelCase__         =  cpu_offload_with_hook(_lowerCAmelCase   ,_lowerCAmelCase   ,prev_module_hook=_lowerCAmelCase       )

								if self.safety_checker is not None:
														lowerCamelCase__       ,			lowerCamelCase__         =  cpu_offload_with_hook(self.safety_checker   ,_lowerCAmelCase   ,prev_module_hook=_lowerCAmelCase       )

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






		@property
		# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
		def 			UpperCamelCase_					(      self       ):
								if not hasattr(self.unet   ,"""_hf_hook"""       ):
														return self.device
								for module in self.unet.modules():
														if (
														    hasattr(_lowerCAmelCase   ,"""_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(_lowerCAmelCase       )
		def __call__(      self   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase = None   ,_lowerCAmelCase = 5_12   ,_lowerCAmelCase = 5_12   ,_lowerCAmelCase = 1_00   ,_lowerCAmelCase = 4.0   ,_lowerCAmelCase = 1   ,_lowerCAmelCase = None   ,_lowerCAmelCase = None   ,_lowerCAmelCase = "pil"   ,_lowerCAmelCase = True   ,):

								if isinstance(_lowerCAmelCase   ,_lowerCAmelCase       ):
														lowerCamelCase__         =  1
								elif isinstance(_lowerCAmelCase   ,_lowerCAmelCase       ):
														lowerCamelCase__         =  len(_lowerCAmelCase       )
								else:
														raise ValueError(F'''`prompt` has to be of type `str` or `list` but is {type(_lowerCAmelCase       )}'''       )

								lowerCamelCase__         =  self._execution_device

								lowerCamelCase__         =  batch_size * num_images_per_prompt
								lowerCamelCase__         =  guidance_scale > 1.0

								lowerCamelCase__       ,			lowerCamelCase__       ,			lowerCamelCase__         =  self._encode_prompt(
								    _lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase       )

								if isinstance(_lowerCAmelCase   ,_lowerCAmelCase       ):
														lowerCamelCase__         =  torch.cat(_lowerCAmelCase   ,dim=0       )
								if isinstance(_lowerCAmelCase   ,_lowerCAmelCase       ):
														lowerCamelCase__         =  torch.cat(_lowerCAmelCase   ,dim=0       )

								if do_classifier_free_guidance:
														lowerCamelCase__         =  image_embeds.repeat_interleave(_lowerCAmelCase   ,dim=0       )
														lowerCamelCase__         =  negative_image_embeds.repeat_interleave(_lowerCAmelCase   ,dim=0       )

								lowerCamelCase__         =  torch.cat([negative_image_embeds, image_embeds]   ,dim=0       ).to(
								    dtype=prompt_embeds.dtype   ,device=_lowerCAmelCase       )

								self.scheduler.set_timesteps(_lowerCAmelCase   ,device=_lowerCAmelCase       )
								lowerCamelCase__         =  self.scheduler.timesteps

								lowerCamelCase__         =  self.unet.config.in_channels

								lowerCamelCase__       ,			lowerCamelCase__         =  get_new_h_w(_lowerCAmelCase   ,_lowerCAmelCase   ,self.movq_scale_factor       )

								# create initial latent
								lowerCamelCase__         =  self.prepare_latents(
								    (batch_size, num_channels_latents, height, width)   ,text_encoder_hidden_states.dtype   ,_lowerCAmelCase   ,_lowerCAmelCase   ,_lowerCAmelCase   ,self.scheduler   ,)

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

														lowerCamelCase__         =  {"""text_embeds""": prompt_embeds, """image_embeds""": image_embeds}
														lowerCamelCase__         =  self.unet(
														    sample=_lowerCAmelCase   ,timestep=_lowerCAmelCase   ,encoder_hidden_states=_lowerCAmelCase   ,added_cond_kwargs=_lowerCAmelCase   ,return_dict=_lowerCAmelCase   ,)[0]

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

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

								if output_type == "pil":
														lowerCamelCase__         =  self.numpy_to_pil(_lowerCAmelCase       )

								if not return_dict:
														return (image,)

								return ImagePipelineOutput(images=_lowerCAmelCase       )









'''simple docstring'''







import numpy

# List of input, output pairs
UpperCamelCase    :      List[Any]  			=		(
    ((5, 2, 3), 15),
    ((6, 5, 9), 25),
    ((11, 12, 13), 41),
    ((1, 1, 1), 8),
    ((11, 12, 13), 41),
)
UpperCamelCase    :      Optional[int]  			=		(((5_15, 22, 13), 5_55), ((61, 35, 49), 1_50))
UpperCamelCase    :      int  			=		[2, 4, 1, 5]
UpperCamelCase    :      int  			=		len(train_data)
UpperCamelCase    :      Dict  			=		0.009
def 						A__			(				__lowerCAmelCase		:			int   ,    __lowerCAmelCase		:			str="train"						):
						return calculate_hypothesis_value(__lowerCAmelCase   ,    __lowerCAmelCase						) - output(
						    __lowerCAmelCase   ,    __lowerCAmelCase						)
def 						A__			(				__lowerCAmelCase		:			Any						):
						lowerCamelCase__         =  0
						for i in range(len(__lowerCAmelCase						) - 1						):
												hyp_val += data_input_tuple[i] * parameter_vector[i + 1]
						hyp_val += parameter_vector[0]
						return hyp_val
def 						A__			(				__lowerCAmelCase		:			Optional[Any]   ,    __lowerCAmelCase		:			Tuple						):
						if data_set == "train":
												return train_data[example_no][1]
						elif data_set == "test":
												return test_data[example_no][1]
						return None
def 						A__			(				__lowerCAmelCase		:			int   ,    __lowerCAmelCase		:			Dict						):
						if data_set == "train":
												return _hypothesis_value(train_data[example_no][0]						)
						elif data_set == "test":
												return _hypothesis_value(test_data[example_no][0]						)
						return None
def 						A__			(				__lowerCAmelCase		:			List[Any]   ,    __lowerCAmelCase		:			Optional[Any]=m						):
						lowerCamelCase__         =  0
						for i in range(__lowerCAmelCase						):
												if index == -1:
																		summation_value += _error(__lowerCAmelCase						)
												else:
																		summation_value += _error(__lowerCAmelCase						) * train_data[i][0][index]
						return summation_value
def 						A__			(				__lowerCAmelCase		:			List[Any]						):
						lowerCamelCase__         =  summation_of_cost_derivative(__lowerCAmelCase   ,    __lowerCAmelCase						) / m
						return cost_derivative_value
def 						A__			(				):
						global parameter_vector
						# Tune these values to set a tolerance value for predicted output
						lowerCamelCase__         =  0.00_0002
						lowerCamelCase__         =  0
						lowerCamelCase__         =  0
						while True:
												j += 1
												lowerCamelCase__         =  [0, 0, 0, 0]
												for i in range(0   ,    len(__lowerCAmelCase						)						):
																		lowerCamelCase__         =  get_cost_derivative(i - 1						)
																		lowerCamelCase__         =  (
																		    parameter_vector[i] - LEARNING_RATE * cost_derivative
																		)
												if numpy.allclose(
												    __lowerCAmelCase   ,    __lowerCAmelCase   ,    atol=__lowerCAmelCase   ,    rtol=__lowerCAmelCase   ,    ):
																		break
												lowerCamelCase__         =  temp_parameter_vector
						print(("""Number of iterations:""", j)						)
def 						A__			(				):
						for i in range(len(__lowerCAmelCase						)						):
												print(("""Actual output value:""", output(__lowerCAmelCase   ,    """test"""						))						)
												print(("""Hypothesis output:""", calculate_hypothesis_value(__lowerCAmelCase   ,    """test"""						))						)


if __name__ == "__main__":
							run_gradient_descent()
							print('\nTesting gradient descent for a linear hypothesis function.\n')
							test_gradient_descent()

