



'''simple docstring'''

import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)







def a_     (    __UpperCAmelCase				,							__UpperCAmelCase				,							__UpperCAmelCase					)				->					List[Any]:





			"""simple docstring"""







			if isinstance(__UpperCAmelCase				,							torch.Tensor					):
						return image
			elif isinstance(__UpperCAmelCase				,							PIL.Image.Image					):
						snake_case:    Tuple      							=[image]

			if isinstance(image[0]				,							PIL.Image.Image					):
						snake_case:    int      							=[np.array(i.resize((w, h)				,							resample=PIL_INTERPOLATION['lanczos']					)					)[None, :] for i in image]
						snake_case:    Tuple      							=np.concatenate(__UpperCAmelCase				,							axis=0					)
						snake_case:    int      							=np.array(__UpperCAmelCase					).astype(np.floataa					) / 255.0
						snake_case:    Optional[Any]      							=image.transpose(0				,							3				,							1				,							2					)
						snake_case:    Optional[int]      							=2.0 * image - 1.0
						snake_case:    Dict      							=torch.from_numpy(__UpperCAmelCase					)
			elif isinstance(image[0]				,							torch.Tensor					):
						snake_case:    Optional[Any]      							=torch.cat(__UpperCAmelCase				,							dim=0					)
			return image







def a_     (    __UpperCAmelCase				,							__UpperCAmelCase				,							__UpperCAmelCase				,							__UpperCAmelCase=0.9995					)				->					Optional[Any]:





			"""simple docstring"""







			if not isinstance(__UpperCAmelCase				,							np.ndarray					):
						snake_case:    int      							=True
						snake_case:    List[Any]      							=va.device
						snake_case:    List[Any]      							=va.cpu().numpy()
						snake_case:    Union[str, Any]      							=va.cpu().numpy()

			snake_case:    List[Any]      							=np.sum(va * va / (np.linalg.norm(__UpperCAmelCase					) * np.linalg.norm(__UpperCAmelCase					))					)
			if np.abs(__UpperCAmelCase					) > DOT_THRESHOLD:
						snake_case:    str      							=(1 - t) * va + t * va
			else:
						snake_case:    List[Any]      							=np.arccos(__UpperCAmelCase					)
						snake_case:    str      							=np.sin(__UpperCAmelCase					)
						snake_case:    List[str]      							=theta_a * t
						snake_case:    Union[str, Any]      							=np.sin(__UpperCAmelCase					)
						snake_case:    Optional[Any]      							=np.sin(theta_a - theta_t					) / sin_theta_a
						snake_case:    Any      							=sin_theta_t / sin_theta_a
						snake_case:    Optional[int]      							=sa * va + sa * va

			if inputs_are_torch:
						snake_case:    List[str]      							=torch.from_numpy(__UpperCAmelCase					).to(__UpperCAmelCase					)

			return va







def a_     (    __UpperCAmelCase				,							__UpperCAmelCase					)				->					Dict:





			"""simple docstring"""







			snake_case:    Optional[int]      							=F.normalize(__UpperCAmelCase				,							dim=-1					)
			snake_case:    Any      							=F.normalize(__UpperCAmelCase				,							dim=-1					)
			return (x - y).norm(dim=-1					).div(2					).arcsin().pow(2					).mul(2					)







def a_     (    __UpperCAmelCase				,							__UpperCAmelCase					)				->					List[Any]:





			"""simple docstring"""







			for param in model.parameters():
						snake_case:    Tuple      							=value







class 						a_		(     snake_case    ):


							def __init__(				self   :						List[str]      ,						a_   :						AutoencoderKL      ,						a_   :						CLIPTextModel      ,						a_   :						CLIPModel      ,						a_   :						CLIPTokenizer      ,						a_   :						UNetaDConditionModel      ,						a_   :						Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]      ,						a_   :						CLIPFeatureExtractor      ,						a_   :						Union[str, Any]=None      ,						a_   :						Optional[int]=None      ,						a_   :						Any=None      ,						)			->					Dict:
										super().__init__()
										self.register_modules(
										    vae=a_      ,						text_encoder=a_      ,						clip_model=a_      ,						tokenizer=a_      ,						unet=a_      ,						scheduler=a_      ,						feature_extractor=a_      ,						coca_model=a_      ,						coca_tokenizer=a_      ,						coca_transform=a_      ,						)
										snake_case:    Optional[Any]      							=(
										    feature_extractor.size
										    if isinstance(feature_extractor.size      ,						a_			)
										    else feature_extractor.size['shortest_edge']
										)
										snake_case:    int      							=transforms.Normalize(mean=feature_extractor.image_mean      ,						std=feature_extractor.image_std			)
										set_requires_grad(self.text_encoder      ,						a_			)
										set_requires_grad(self.clip_model      ,						a_			)


							def    UpperCamelCase     (				self   :						List[Any]      ,						a_   :						Optional[Union[str, int]] = "auto"			)			->					List[Any]:
										if slice_size == "auto":
													# half the attention head size is usually a good trade-off between
													# speed and memory
													snake_case:    str      							=self.unet.config.attention_head_dim // 2
										self.unet.set_attention_slice(a_			)


							def    UpperCamelCase     (				self   :						Optional[int]			)			->					Any:
										self.enable_attention_slicing(a_			)


							def    UpperCamelCase     (				self   :						str			)			->					Any:
										set_requires_grad(self.vae      ,						a_			)


							def    UpperCamelCase     (				self   :						Any			)			->					Optional[Any]:
										set_requires_grad(self.vae      ,						a_			)


							def    UpperCamelCase     (				self   :						Optional[Any]			)			->					Optional[int]:
										set_requires_grad(self.unet      ,						a_			)


							def    UpperCamelCase     (				self   :						Union[str, Any]			)			->					str:
										set_requires_grad(self.unet      ,						a_			)


							def    UpperCamelCase     (				self   :						Optional[int]      ,						a_   :						str      ,						a_   :						Tuple      ,						a_   :						int			)			->					int:
										# get the original timestep using init_timestep
										snake_case:    Tuple      							=min(int(num_inference_steps * strength			)      ,						a_			)

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

										return timesteps, num_inference_steps - t_start


							def    UpperCamelCase     (				self   :						int      ,						a_   :						Any      ,						a_   :						str      ,						a_   :						Optional[Any]      ,						a_   :						Any      ,						a_   :						List[str]      ,						a_   :						Optional[Any]=None			)			->					Optional[Any]:
										if not isinstance(a_      ,						torch.Tensor			):
													raise ValueError(F'''`image` has to be of type `torch.Tensor` but is {type(a_			)}'''			)

										snake_case:    Optional[int]      							=image.to(device=a_      ,						dtype=a_			)

										if isinstance(a_      ,						a_			):
													snake_case:    List[str]      							=[
													    self.vae.encode(image[i : i + 1]			).latent_dist.sample(generator[i]			) for i in range(a_			)
													]
													snake_case:    Dict      							=torch.cat(a_      ,						dim=0			)
										else:
													snake_case:    Dict      							=self.vae.encode(a_			).latent_dist.sample(a_			)

										# Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
										snake_case:    List[str]      							=0.1_8_2_1_5 * init_latents
										snake_case:    Tuple      							=init_latents.repeat_interleave(a_      ,						dim=0			)

										snake_case:    Any      							=randn_tensor(init_latents.shape      ,						generator=a_      ,						device=a_      ,						dtype=a_			)

										# get latents
										snake_case:    Optional[Any]      							=self.scheduler.add_noise(a_      ,						a_      ,						a_			)
										snake_case:    Union[str, Any]      							=init_latents

										return latents


							def    UpperCamelCase     (				self   :						Optional[int]      ,						a_   :						Any			)			->					str:
										snake_case:    Tuple      							=self.coca_transform(a_			).unsqueeze(0			)
										with torch.no_grad(), torch.cuda.amp.autocast():
													snake_case:    Tuple      							=self.coca_model.generate(transformed_image.to(device=self.device      ,						dtype=self.coca_model.dtype			)			)
										snake_case:    List[str]      							=self.coca_tokenizer.decode(generated[0].cpu().numpy()			)
										return generated.split('<end_of_text>'			)[0].replace('<start_of_text>'      ,						''			).rstrip(' .,'			)


							def    UpperCamelCase     (				self   :						Any      ,						a_   :						Optional[Any]      ,						a_   :						Optional[int]			)			->					Dict:
										snake_case:    Union[str, Any]      							=self.feature_extractor.preprocess(a_			)
										snake_case:    Tuple      							=torch.from_numpy(clip_image_input['pixel_values'][0]			).unsqueeze(0			).to(self.device			).half()
										snake_case:    str      							=self.clip_model.get_image_features(a_			)
										snake_case:    Union[str, Any]      							=image_embeddings_clip / image_embeddings_clip.norm(p=2      ,						dim=-1      ,						keepdim=a_			)
										snake_case:    Any      							=image_embeddings_clip.repeat_interleave(a_      ,						dim=0			)
										return image_embeddings_clip


							@torch.enable_grad()
							def    UpperCamelCase     (				self   :						Dict      ,						a_   :						Tuple      ,						a_   :						Dict      ,						a_   :						int      ,						a_   :						Union[str, Any]      ,						a_   :						str      ,						a_   :						List[Any]      ,						a_   :						Tuple      ,						)			->					int:
										snake_case:    Any      							=latents.detach().requires_grad_()

										snake_case:    Any      							=self.scheduler.scale_model_input(a_      ,						a_			)

										# predict the noise residual
										snake_case:    List[Any]      							=self.unet(a_      ,						a_      ,						encoder_hidden_states=a_			).sample

										if isinstance(self.scheduler      ,						(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)			):
													snake_case:    List[Any]      							=self.scheduler.alphas_cumprod[timestep]
													snake_case:    Any      							=1 - alpha_prod_t
													# compute predicted original sample from predicted noise also called
													# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
													snake_case:    str      							=(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

													snake_case:    Optional[Any]      							=torch.sqrt(a_			)
													snake_case:    Optional[int]      							=pred_original_sample * (fac) + latents * (1 - fac)
										elif isinstance(self.scheduler      ,						a_			):
													snake_case:    int      							=self.scheduler.sigmas[index]
													snake_case:    Tuple      							=latents - sigma * noise_pred
										else:
													raise ValueError(F'''scheduler type {type(self.scheduler			)} not supported'''			)

										# Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
										snake_case:    str      							=1 / 0.1_8_2_1_5 * sample
										snake_case:    int      							=self.vae.decode(a_			).sample
										snake_case:    List[Any]      							=(image / 2 + 0.5).clamp(0      ,						1			)

										snake_case:    Optional[int]      							=transforms.Resize(self.feature_extractor_size			)(a_			)
										snake_case:    str      							=self.normalize(a_			).to(latents.dtype			)

										snake_case:    Dict      							=self.clip_model.get_image_features(a_			)
										snake_case:    int      							=image_embeddings_clip / image_embeddings_clip.norm(p=2      ,						dim=-1      ,						keepdim=a_			)

										snake_case:    Optional[Any]      							=spherical_dist_loss(a_      ,						a_			).mean() * clip_guidance_scale

										snake_case:    List[Any]      							=-torch.autograd.grad(a_      ,						a_			)[0]

										if isinstance(self.scheduler      ,						a_			):
													snake_case:    Optional[Any]      							=latents.detach() + grads * (sigma**2)
													snake_case:    List[str]      							=noise_pred_original
										else:
													snake_case:    Tuple      							=noise_pred_original - torch.sqrt(a_			) * grads
										return noise_pred, latents




							@torch.no_grad()
							def __call__(				self   :						Tuple      ,						a_   :						Union[torch.FloatTensor, PIL.Image.Image]      ,						a_   :						Union[torch.FloatTensor, PIL.Image.Image]      ,						a_   :						Optional[str] = None      ,						a_   :						Optional[str] = None      ,						a_   :						Optional[int] = 5_1_2      ,						a_   :						Optional[int] = 5_1_2      ,						a_   :						float = 0.6      ,						a_   :						Optional[int] = 5_0      ,						a_   :						Optional[float] = 7.5      ,						a_   :						Optional[int] = 1      ,						a_   :						float = 0.0      ,						a_   :						Optional[float] = 1_0_0      ,						a_   :						Optional[torch.Generator] = None      ,						a_   :						Optional[str] = "pil"      ,						a_   :						bool = True      ,						a_   :						float = 0.8      ,						a_   :						float = 0.1      ,						a_   :						float = 0.1      ,						)			->					str:
										if isinstance(a_      ,						a_			) and len(a_			) != batch_size:
													raise ValueError(F'''You have passed {batch_size} batch_size, but only {len(a_			)} generators.'''			)

										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 isinstance(a_      ,						torch.Generator			) and batch_size > 1:
													snake_case:    str      							=[generator] + [None] * (batch_size - 1)

										snake_case:    str      							=[
										    ('model', self.coca_model is None),
										    ('tokenizer', self.coca_tokenizer is None),
										    ('transform', self.coca_transform is None),
										]
										snake_case:    Optional[int]      							=[x[0] for x in coca_is_none if x[1]]
										snake_case:    str      							=', '.join(a_			)
										# generate prompts with coca model if prompt is None
										if content_prompt is None:
													if len(a_			):
																raise ValueError(
																    F'''Content prompt is None and CoCa [{coca_is_none_str}] is None.'''
																    F'''Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'''			)
													snake_case:    Optional[int]      							=self.get_image_description(a_			)
										if style_prompt is None:
													if len(a_			):
																raise ValueError(
																    F'''Style prompt is None and CoCa [{coca_is_none_str}] is None.'''
																    F''' Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'''			)
													snake_case:    Tuple      							=self.get_image_description(a_			)

										# get prompt text embeddings for content and style
										snake_case:    Any      							=self.tokenizer(
										    a_      ,						padding='max_length'      ,						max_length=self.tokenizer.model_max_length      ,						truncation=a_      ,						return_tensors='pt'      ,						)
										snake_case:    Tuple      							=self.text_encoder(content_text_input.input_ids.to(self.device			)			)[0]

										snake_case:    Optional[Any]      							=self.tokenizer(
										    a_      ,						padding='max_length'      ,						max_length=self.tokenizer.model_max_length      ,						truncation=a_      ,						return_tensors='pt'      ,						)
										snake_case:    List[str]      							=self.text_encoder(style_text_input.input_ids.to(self.device			)			)[0]

										snake_case:    str      							=slerp(a_      ,						a_      ,						a_			)

										# duplicate text embeddings for each generation per prompt
										snake_case:    List[str]      							=text_embeddings.repeat_interleave(a_      ,						dim=0			)

										# set timesteps
										snake_case:    Optional[int]      							='offset' in set(inspect.signature(self.scheduler.set_timesteps			).parameters.keys()			)
										snake_case:    Optional[int]      							={}
										if accepts_offset:
													snake_case:    Any      							=1

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

										snake_case							,  snake_case:    Dict      							=self.get_timesteps(a_      ,						a_      ,						self.device			)
										snake_case:    int      							=timesteps[:1].repeat(a_			)

										# Preprocess image
										snake_case:    Optional[Any]      							=preprocess(a_      ,						a_      ,						a_			)
										snake_case:    Union[str, Any]      							=self.prepare_latents(
										    a_      ,						a_      ,						a_      ,						text_embeddings.dtype      ,						self.device      ,						a_			)

										snake_case:    str      							=preprocess(a_      ,						a_      ,						a_			)
										snake_case:    int      							=self.prepare_latents(
										    a_      ,						a_      ,						a_      ,						text_embeddings.dtype      ,						self.device      ,						a_			)

										snake_case:    List[Any]      							=slerp(a_      ,						a_      ,						a_			)

										if clip_guidance_scale > 0:
													snake_case:    Optional[int]      							=self.get_clip_image_embeddings(a_      ,						a_			)
													snake_case:    Union[str, Any]      							=self.get_clip_image_embeddings(a_      ,						a_			)
													snake_case:    Optional[int]      							=slerp(
													    a_      ,						a_      ,						a_			)

										# 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.
										snake_case:    Union[str, Any]      							=guidance_scale > 1.0
										# get unconditional embeddings for classifier free guidance
										if do_classifier_free_guidance:
													snake_case:    Optional[Any]      							=content_text_input.input_ids.shape[-1]
													snake_case:    Any      							=self.tokenizer(['']      ,						padding='max_length'      ,						max_length=a_      ,						return_tensors='pt'			)
													snake_case:    Optional[Any]      							=self.text_encoder(uncond_input.input_ids.to(self.device			)			)[0]
													# duplicate unconditional embeddings for each generation per prompt
													snake_case:    Any      							=uncond_embeddings.repeat_interleave(a_      ,						dim=0			)

													# 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
													snake_case:    Union[str, Any]      							=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`.
										snake_case:    Optional[int]      							=(batch_size, self.unet.config.in_channels, height // 8, width // 8)
										snake_case:    List[str]      							=text_embeddings.dtype
										if latents is None:
													if self.device.type == "mps":
																# randn does not work reproducibly on mps
																snake_case:    Tuple      							=torch.randn(a_      ,						generator=a_      ,						device='cpu'      ,						dtype=a_			).to(
																    self.device			)
													else:
																snake_case:    Optional[Any]      							=torch.randn(a_      ,						generator=a_      ,						device=self.device      ,						dtype=a_			)
										else:
													if latents.shape != latents_shape:
																raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''			)
													snake_case:    Union[str, Any]      							=latents.to(self.device			)

										# scale the initial noise by the standard deviation required by the scheduler
										snake_case:    Any      							=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]
										snake_case:    str      							='eta' in set(inspect.signature(self.scheduler.step			).parameters.keys()			)
										snake_case:    Optional[Any]      							={}
										if accepts_eta:
													snake_case:    Any      							=eta

										# check if the scheduler accepts generator
										snake_case:    Union[str, Any]      							='generator' in set(inspect.signature(self.scheduler.step			).parameters.keys()			)
										if accepts_generator:
													snake_case:    Optional[int]      							=generator

										with self.progress_bar(total=a_			):
													for i, t in enumerate(a_			):
																# expand the latents if we are doing classifier free guidance
																snake_case:    Optional[Any]      							=torch.cat([latents] * 2			) if do_classifier_free_guidance else latents
																snake_case:    Tuple      							=self.scheduler.scale_model_input(a_      ,						a_			)

																# predict the noise residual
																snake_case:    Optional[int]      							=self.unet(a_      ,						a_      ,						encoder_hidden_states=a_			).sample

																# perform classifier free guidance
																if do_classifier_free_guidance:
																			snake_case							,  snake_case:    Any      							=noise_pred.chunk(2			)
																			snake_case:    List[Any]      							=noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																# perform clip guidance
																if clip_guidance_scale > 0:
																			snake_case:    Optional[Any]      							=(
																			    text_embeddings.chunk(2			)[1] if do_classifier_free_guidance else text_embeddings
																			)
																			snake_case							,  snake_case:    List[str]      							=self.cond_fn(
																			    a_      ,						a_      ,						a_      ,						a_      ,						a_      ,						a_      ,						a_      ,						)

																# compute the previous noisy sample x_t -> x_t-1
																snake_case:    str      							=self.scheduler.step(a_      ,						a_      ,						a_      ,						**a_			).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
										snake_case:    List[str]      							=1 / 0.1_8_2_1_5 * latents
										snake_case:    str      							=self.vae.decode(a_			).sample

										snake_case:    List[str]      							=(image / 2 + 0.5).clamp(0      ,						1			)
										snake_case:    int      							=image.cpu().permute(0      ,						2      ,						3      ,						1			).numpy()

										if output_type == "pil":
													snake_case:    Optional[int]      							=self.numpy_to_pil(a_			)

										if not return_dict:
													return (image, None)

										return StableDiffusionPipelineOutput(images=a_      ,						nsfw_content_detected=a_			)









'''simple docstring'''

import unittest

from knapsack import knapsack as k






class 						a_		(     unittest.TestCase    ):


							def    UpperCamelCase     (				self   :						str			)			->					List[Any]:
										snake_case:    Optional[Any]      							=0
										snake_case:    List[str]      							=[0]
										snake_case:    List[str]      							=[0]
										snake_case:    Union[str, Any]      							=len(a_			)
										self.assertEqual(k.knapsack(a_      ,						a_      ,						a_      ,						a_			)      ,						0			)

										snake_case:    Union[str, Any]      							=[6_0]
										snake_case:    str      							=[1_0]
										snake_case:    List[str]      							=len(a_			)
										self.assertEqual(k.knapsack(a_      ,						a_      ,						a_      ,						a_			)      ,						0			)


							def    UpperCamelCase     (				self   :						int			)			->					Optional[Any]:
										snake_case:    Optional[int]      							=3
										snake_case:    Dict      							=[1, 2, 3]
										snake_case:    List[Any]      							=[3, 2, 1]
										snake_case:    Tuple      							=len(a_			)
										self.assertEqual(k.knapsack(a_      ,						a_      ,						a_      ,						a_			)      ,						5			)




							def    UpperCamelCase     (				self   :						List[str]			)			->					str:
										snake_case:    Any      							=5_0
										snake_case:    str      							=[6_0, 1_0_0, 1_2_0]
										snake_case:    Any      							=[1_0, 2_0, 3_0]
										snake_case:    Tuple      							=len(a_			)
										self.assertEqual(k.knapsack(a_      ,						a_      ,						a_      ,						a_			)      ,						2_2_0			)


if __name__ == "__main__":
	unittest.main()




