'''simple docstring'''




import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


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






class 							a__(   lowerCAmelCase__      ):


			'''simple docstring'''






			def __init__(  self			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase			,       ):




					"""simple docstring"""


					super().__init__()
					self.register_modules(
					    vae=__lowerCAmelCase			,       text_encoder=__lowerCAmelCase			,       tokenizer=__lowerCAmelCase			,       unet=__lowerCAmelCase			,       scheduler=__lowerCAmelCase			,       safety_checker=__lowerCAmelCase			,       feature_extractor=__lowerCAmelCase			,       )



			def 				a_			(  self			,       __lowerCAmelCase = "auto"):




					"""simple docstring"""


					if slice_size == "auto":
							# half the attention head size is usually a good trade-off between
							# speed and memory
							lowerCAmelCase     =    self.unet.config.attention_head_dim // 2
					self.unet.set_attention_slice(__lowerCAmelCase)



			def 				a_			(  self):




					"""simple docstring"""


					self.enable_attention_slicing(__lowerCAmelCase)







			@torch.no_grad()
			def __call__(  self			,       __lowerCAmelCase			,       __lowerCAmelCase = 512			,       __lowerCAmelCase = 512			,       __lowerCAmelCase = 50			,       __lowerCAmelCase = 7.5			,       __lowerCAmelCase = None			,       __lowerCAmelCase = 1			,       __lowerCAmelCase = 0.0			,       __lowerCAmelCase = None			,       __lowerCAmelCase = None			,       __lowerCAmelCase = "pil"			,       __lowerCAmelCase = True			,       __lowerCAmelCase = None			,       __lowerCAmelCase = 1			,       __lowerCAmelCase = None			,       **__lowerCAmelCase			,       ):




					"""simple docstring"""


					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)}")

					if height % 8 != 0 or width % 8 != 0:
							raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}.")

					if (callback_steps is None) or (
					    callback_steps is not None and (not isinstance(__lowerCAmelCase			,       __lowerCAmelCase) or callback_steps <= 0)
					):
							raise ValueError(
							    f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
							    f" {type(__lowerCAmelCase)}.")

					# get prompt text embeddings
					lowerCAmelCase     =    self.tokenizer(
					    __lowerCAmelCase			,       padding="""max_length"""			,       max_length=self.tokenizer.model_max_length			,       return_tensors="""pt"""			,       )
					lowerCAmelCase     =    text_inputs.input_ids

					if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
							lowerCAmelCase     =    self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :])
							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[:, : self.tokenizer.model_max_length]

					if text_embeddings is None:
							lowerCAmelCase     =    self.text_encoder(text_input_ids.to(self.device))[0]

					# duplicate text embeddings for each generation per prompt, using mps friendly method
					lowerCAmelCase   ,	lowerCAmelCase   ,	lowerCAmelCase     =    text_embeddings.shape
					lowerCAmelCase     =    text_embeddings.repeat(1			,       __lowerCAmelCase			,       1)
					lowerCAmelCase     =    text_embeddings.view(bs_embed * num_images_per_prompt			,       __lowerCAmelCase			,       -1)

					# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
					# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
					# corresponds to doing no classifier free guidance.
					lowerCAmelCase     =    guidance_scale > 1.0
					# get unconditional embeddings for classifier free guidance
					if do_classifier_free_guidance:
							lowerCAmelCase     =    42
							if negative_prompt is None:
									lowerCAmelCase     =    [""""""]
							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     =    text_input_ids.shape[-1]
							lowerCAmelCase     =    self.tokenizer(
							    __lowerCAmelCase			,       padding="""max_length"""			,       max_length=__lowerCAmelCase			,       truncation=__lowerCAmelCase			,       return_tensors="""pt"""			,       )
							lowerCAmelCase     =    self.text_encoder(uncond_input.input_ids.to(self.device))[0]

							# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
							lowerCAmelCase     =    uncond_embeddings.shape[1]
							lowerCAmelCase     =    uncond_embeddings.repeat(__lowerCAmelCase			,       __lowerCAmelCase			,       1)
							lowerCAmelCase     =    uncond_embeddings.view(batch_size * num_images_per_prompt			,       __lowerCAmelCase			,       -1)

							# 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([uncond_embeddings, text_embeddings])

					# get the initial random noise unless the user supplied it

					# Unlike in other pipelines, latents need to be generated in the target device
					# for 1-to-1 results reproducibility with the CompVis implementation.
					# However this currently doesn't work in `mps`.
					lowerCAmelCase     =    (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
					lowerCAmelCase     =    (batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
					lowerCAmelCase     =    text_embeddings.dtype
					if latents is None:
							if self.device.type == "mps":
									# randn does not exist on mps
									lowerCAmelCase     =    torch.randn(
									    __lowerCAmelCase			,       generator=__lowerCAmelCase			,       device="""cpu"""			,       dtype=__lowerCAmelCase).to(self.device)
									lowerCAmelCase     =    torch.randn(__lowerCAmelCase			,       generator=__lowerCAmelCase			,       device="""cpu"""			,       dtype=__lowerCAmelCase).to(
									    self.device)
							else:
									lowerCAmelCase     =    torch.randn(
									    __lowerCAmelCase			,       generator=__lowerCAmelCase			,       device=self.device			,       dtype=__lowerCAmelCase)
									lowerCAmelCase     =    torch.randn(__lowerCAmelCase			,       generator=__lowerCAmelCase			,       device=self.device			,       dtype=__lowerCAmelCase)
					else:
							if latents_reference.shape != latents_shape:
									raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}")
							lowerCAmelCase     =    latents_reference.to(self.device)
							lowerCAmelCase     =    latents.to(self.device)

					# This is the key part of the pipeline where we
					# try to ensure that the generated images w/ the same seed
					# but different sizes actually result in similar images
					lowerCAmelCase     =    (latents_shape[3] - latents_shape_reference[3]) // 2
					lowerCAmelCase     =    (latents_shape[2] - latents_shape_reference[2]) // 2
					lowerCAmelCase     =    latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
					lowerCAmelCase     =    latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
					lowerCAmelCase     =    0 if dx < 0 else dx
					lowerCAmelCase     =    0 if dy < 0 else dy
					lowerCAmelCase     =    max(-dx			,       0)
					lowerCAmelCase     =    max(-dy			,       0)
					# import pdb
					# pdb.set_trace()
					lowerCAmelCase     =    latents_reference[:, :, dy : dy + h, dx : dx + w]

					# set timesteps
					self.scheduler.set_timesteps(__lowerCAmelCase)

					# Some schedulers like PNDM have timesteps as arrays
					# It's more optimized to move all timesteps to correct device beforehand
					lowerCAmelCase     =    self.scheduler.timesteps.to(self.device)

					# scale the initial noise by the standard deviation required by the scheduler
					lowerCAmelCase     =    latents * self.scheduler.init_noise_sigma

					# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
					# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
					# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
					# and should be between [0, 1]
					lowerCAmelCase     =    """eta""" in set(inspect.signature(self.scheduler.step).parameters.keys())
					lowerCAmelCase     =    {}
					if accepts_eta:
							lowerCAmelCase     =    eta

					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     =    self.scheduler.scale_model_input(__lowerCAmelCase			,       __lowerCAmelCase)

							# predict the noise residual
							lowerCAmelCase     =    self.unet(__lowerCAmelCase			,       __lowerCAmelCase			,       encoder_hidden_states=__lowerCAmelCase).sample

							# perform guidance
							if do_classifier_free_guidance:
									lowerCAmelCase   ,	lowerCAmelCase     =    noise_pred.chunk(2)
									lowerCAmelCase     =    noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

							# compute the previous noisy sample x_t -> x_t-1
							lowerCAmelCase     =    self.scheduler.step(__lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase			,       **__lowerCAmelCase).prev_sample

							# call the callback, if provided
							if callback is not None and i % callback_steps == 0:
									callback(__lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase)

					lowerCAmelCase     =    1 / 0.18215 * latents
					lowerCAmelCase     =    self.vae.decode(__lowerCAmelCase).sample

					lowerCAmelCase     =    (image / 2 + 0.5).clamp(0			,       1)

					# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
					lowerCAmelCase     =    image.cpu().permute(0			,       2			,       3			,       1).float().numpy()

					if self.safety_checker is not None:
							lowerCAmelCase     =    self.feature_extractor(self.numpy_to_pil(__lowerCAmelCase)			,       return_tensors="""pt""").to(
							    self.device)
							lowerCAmelCase   ,	lowerCAmelCase     =    self.safety_checker(
							    images=__lowerCAmelCase			,       clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype))
					else:
							lowerCAmelCase     =    None

					if output_type == "pil":
							lowerCAmelCase     =    self.numpy_to_pil(__lowerCAmelCase)

					if not return_dict:
							return (image, has_nsfw_concept)

					return StableDiffusionPipelineOutput(images=__lowerCAmelCase			,       nsfw_content_detected=__lowerCAmelCase)





'''simple docstring'''




import math
import time

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
						import torch_xla.core.xla_model as xm
						import torch_xla.debug.metrics as met






class 							a__(   lowerCAmelCase__      ):


			'''simple docstring'''






			def __init__(  self			,       *__lowerCAmelCase			,       __lowerCAmelCase=None			,       __lowerCAmelCase=None			,       **__lowerCAmelCase):




					"""simple docstring"""


					super().__init__(*__lowerCAmelCase			,       **__lowerCAmelCase)
					lowerCAmelCase     =    eval_examples
					lowerCAmelCase     =    post_process_function



			def 				a_			(  self			,       __lowerCAmelCase=None			,       __lowerCAmelCase=None			,       __lowerCAmelCase=None			,       __lowerCAmelCase = "eval"):




					"""simple docstring"""


					lowerCAmelCase     =    self.eval_dataset if eval_dataset is None else eval_dataset
					lowerCAmelCase     =    self.get_eval_dataloader(__lowerCAmelCase)
					lowerCAmelCase     =    self.eval_examples if eval_examples is None else eval_examples

					# Temporarily disable metric computation, we will do it in the loop here.
					lowerCAmelCase     =    self.compute_metrics
					lowerCAmelCase     =    None
					lowerCAmelCase     =    self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					lowerCAmelCase     =    time.time()
					try:
							lowerCAmelCase     =    eval_loop(
							    __lowerCAmelCase			,       description="""Evaluation"""			,       prediction_loss_only=True if compute_metrics is None else None			,       ignore_keys=__lowerCAmelCase			,       metric_key_prefix=__lowerCAmelCase			,       )
					finally:
							lowerCAmelCase     =    compute_metrics
					lowerCAmelCase     =    self.args.eval_batch_size * self.args.world_size
					if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
							start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
					output.metrics.update(
					    speed_metrics(
					        __lowerCAmelCase			,       __lowerCAmelCase			,       num_samples=output.num_samples			,       num_steps=math.ceil(output.num_samples / total_batch_size)			,       ))
					if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
							# Only the main node write the results by default
							lowerCAmelCase     =    self.post_process_function(__lowerCAmelCase			,       __lowerCAmelCase			,       output.predictions)
							lowerCAmelCase     =    self.compute_metrics(__lowerCAmelCase)

							# Prefix all keys with metric_key_prefix + '_'
							for key in list(metrics.keys()):
									if not key.startswith(f"{metric_key_prefix}_"):
											lowerCAmelCase     =    metrics.pop(__lowerCAmelCase)
							metrics.update(output.metrics)
					else:
							lowerCAmelCase     =    output.metrics

					if self.args.should_log:
							# Only the main node log the results by default
							self.log(__lowerCAmelCase)

					if self.args.tpu_metrics_debug or self.args.debug:
							# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
							xm.master_print(met.metrics_report())

					lowerCAmelCase     =    self.callback_handler.on_evaluate(self.args			,       self.state			,       self.control			,       __lowerCAmelCase)
					return metrics







			def 				a_			(  self			,       __lowerCAmelCase			,       __lowerCAmelCase			,       __lowerCAmelCase=None			,       __lowerCAmelCase = "test"):




					"""simple docstring"""


					lowerCAmelCase     =    self.get_test_dataloader(__lowerCAmelCase)

					# Temporarily disable metric computation, we will do it in the loop here.
					lowerCAmelCase     =    self.compute_metrics
					lowerCAmelCase     =    None
					lowerCAmelCase     =    self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					lowerCAmelCase     =    time.time()
					try:
							lowerCAmelCase     =    eval_loop(
							    __lowerCAmelCase			,       description="""Prediction"""			,       prediction_loss_only=True if compute_metrics is None else None			,       ignore_keys=__lowerCAmelCase			,       metric_key_prefix=__lowerCAmelCase			,       )
					finally:
							lowerCAmelCase     =    compute_metrics
					lowerCAmelCase     =    self.args.eval_batch_size * self.args.world_size
					if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
							start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
					output.metrics.update(
					    speed_metrics(
					        __lowerCAmelCase			,       __lowerCAmelCase			,       num_samples=output.num_samples			,       num_steps=math.ceil(output.num_samples / total_batch_size)			,       ))

					if self.post_process_function is None or self.compute_metrics is None:
							return output

					lowerCAmelCase     =    self.post_process_function(__lowerCAmelCase			,       __lowerCAmelCase			,       output.predictions			,       """predict""")
					lowerCAmelCase     =    self.compute_metrics(__lowerCAmelCase)

					# Prefix all keys with metric_key_prefix + '_'
					for key in list(metrics.keys()):
							if not key.startswith(f"{metric_key_prefix}_"):
									lowerCAmelCase     =    metrics.pop(__lowerCAmelCase)
					metrics.update(output.metrics)
					return PredictionOutput(predictions=predictions.predictions			,       label_ids=predictions.label_ids			,       metrics=__lowerCAmelCase)




