



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 	__UpperCamelCase	(  _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase		)       ->  Dict:



    """simple docstring"""






    if isinstance(_lowerCAmelCase      ,     torch.Tensor		):
        return image
    elif isinstance(_lowerCAmelCase      ,     PIL.Image.Image		):
        A :       Optional[Any]          =   [image]

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




def 	__UpperCamelCase	(  _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase      ,     _lowerCAmelCase=0.9_995		)       ->  Optional[int]:



    """simple docstring"""






    if not isinstance(_lowerCAmelCase      ,     np.ndarray		):
        A :       Tuple          =   True
        A :       Optional[int]          =   va.device
        A :       Tuple          =   va.cpu().numpy()
        A :       List[str]          =   va.cpu().numpy()

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

    if inputs_are_torch:
        A :       List[Any]          =   torch.from_numpy(_lowerCAmelCase		).to(_lowerCAmelCase		)

    return va




def 	__UpperCamelCase	(  _lowerCAmelCase      ,     _lowerCAmelCase		)       ->  Optional[Any]:



    """simple docstring"""






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




def 	__UpperCamelCase	(  _lowerCAmelCase      ,     _lowerCAmelCase		)       ->  Dict:



    """simple docstring"""






    for param in model.parameters():
        A :       Any          =   value






class SCREAMING_SNAKE_CASE__     (		SCREAMING_SNAKE_CASE__    ):




    '''simple docstring'''






    def __init__(       self,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__=None,		lowerCamelCase__=None,		lowerCamelCase__=None,		):
        super().__init__()
        self.register_modules(
            vae=lowerCamelCase__,		text_encoder=lowerCamelCase__,		clip_model=lowerCamelCase__,		tokenizer=lowerCamelCase__,		unet=lowerCamelCase__,		scheduler=lowerCamelCase__,		feature_extractor=lowerCamelCase__,		coca_model=lowerCamelCase__,		coca_tokenizer=lowerCamelCase__,		coca_transform=lowerCamelCase__,		)
        A :       Union[str, Any]          =   (
            feature_extractor.size
            if isinstance(feature_extractor.size,		lowerCamelCase__		)
            else feature_extractor.size["""shortest_edge"""]
        )
        A :       Optional[Any]          =   transforms.Normalize(mean=feature_extractor.image_mean,		std=feature_extractor.image_std		)
        set_requires_grad(self.text_encoder,		lowerCamelCase__		)
        set_requires_grad(self.clip_model,		lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__ = "auto"		):
        if slice_size == "auto":
            # half the attention head size is usually a good trade-off between
            # speed and memory
            A :       Optional[Any]          =   self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self		):
        self.enable_attention_slicing(lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self		):
        set_requires_grad(self.vae,		lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self		):
        set_requires_grad(self.vae,		lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self		):
        set_requires_grad(self.unet,		lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self		):
        set_requires_grad(self.unet,		lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__		):
        # get the original timestep using init_timestep
        A :       Dict          =   min(int(num_inference_steps * strength		),		lowerCamelCase__		)

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

        return timesteps, num_inference_steps - t_start






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__=None		):
        if not isinstance(lowerCamelCase__,		torch.Tensor		):
            raise ValueError(f'''`image` has to be of type `torch.Tensor` but is {type(lowerCamelCase__		)}'''		)

        A :       List[Any]          =   image.to(device=lowerCamelCase__,		dtype=lowerCamelCase__		)

        if isinstance(lowerCamelCase__,		lowerCamelCase__		):
            A :       List[Any]          =   [
                self.vae.encode(image[i : i + 1]		).latent_dist.sample(generator[i]		) for i in range(lowerCamelCase__		)
            ]
            A :       int          =   torch.cat(lowerCamelCase__,		dim=0		)
        else:
            A :       Union[str, Any]          =   self.vae.encode(lowerCamelCase__		).latent_dist.sample(lowerCamelCase__		)

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
        A :       Tuple          =   0.1_8215 * init_latents
        A :       Tuple          =   init_latents.repeat_interleave(lowerCamelCase__,		dim=0		)

        A :       Union[str, Any]          =   randn_tensor(init_latents.shape,		generator=lowerCamelCase__,		device=lowerCamelCase__,		dtype=lowerCamelCase__		)

        # get latents
        A :       List[Any]          =   self.scheduler.add_noise(lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__		)
        A :       Dict          =   init_latents

        return latents






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__		):
        A :       Union[str, Any]          =   self.coca_transform(lowerCamelCase__		).unsqueeze(0		)
        with torch.no_grad(), torch.cuda.amp.autocast():
            A :       Union[str, Any]          =   self.coca_model.generate(transformed_image.to(device=self.device,		dtype=self.coca_model.dtype		)		)
        A :       Tuple          =   self.coca_tokenizer.decode(generated[0].cpu().numpy()		)
        return generated.split("""<end_of_text>"""		)[0].replace("""<start_of_text>""",		""""""		).rstrip(""" .,"""		)






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__		):
        A :       List[str]          =   self.feature_extractor.preprocess(lowerCamelCase__		)
        A :       int          =   torch.from_numpy(clip_image_input["""pixel_values"""][0]		).unsqueeze(0		).to(self.device		).half()
        A :       Any          =   self.clip_model.get_image_features(lowerCamelCase__		)
        A :       Union[str, Any]          =   image_embeddings_clip / image_embeddings_clip.norm(p=2,		dim=-1,		keepdim=lowerCamelCase__		)
        A :       Optional[int]          =   image_embeddings_clip.repeat_interleave(lowerCamelCase__,		dim=0		)
        return image_embeddings_clip






    @torch.enable_grad()
    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		):
        A :       int          =   latents.detach().requires_grad_()

        A :       Any          =   self.scheduler.scale_model_input(lowerCamelCase__,		lowerCamelCase__		)

        # predict the noise residual
        A :       Union[str, Any]          =   self.unet(lowerCamelCase__,		lowerCamelCase__,		encoder_hidden_states=lowerCamelCase__		).sample

        if isinstance(self.scheduler,		(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)		):
            A :       Tuple          =   self.scheduler.alphas_cumprod[timestep]
            A :       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
            A :       Union[str, Any]          =   (latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

            A :       Tuple          =   torch.sqrt(lowerCamelCase__		)
            A :       Tuple          =   pred_original_sample * (fac) + latents * (1 - fac)
        elif isinstance(self.scheduler,		lowerCamelCase__		):
            A :       Optional[int]          =   self.scheduler.sigmas[index]
            A :       Any          =   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
        A :       List[str]          =   1 / 0.1_8215 * sample
        A :       Dict          =   self.vae.decode(lowerCamelCase__		).sample
        A :       List[str]          =   (image / 2 + 0.5).clamp(0,		1		)

        A :       List[Any]          =   transforms.Resize(self.feature_extractor_size		)(lowerCamelCase__		)
        A :       Optional[Any]          =   self.normalize(lowerCamelCase__		).to(latents.dtype		)

        A :       int          =   self.clip_model.get_image_features(lowerCamelCase__		)
        A :       str          =   image_embeddings_clip / image_embeddings_clip.norm(p=2,		dim=-1,		keepdim=lowerCamelCase__		)

        A :       List[Any]          =   spherical_dist_loss(lowerCamelCase__,		lowerCamelCase__		).mean() * clip_guidance_scale

        A :       int          =   -torch.autograd.grad(lowerCamelCase__,		lowerCamelCase__		)[0]

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






    @torch.no_grad()
    def __call__(       self,		lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__ = None,		lowerCamelCase__ = None,		lowerCamelCase__ = 512,		lowerCamelCase__ = 512,		lowerCamelCase__ = 0.6,		lowerCamelCase__ = 50,		lowerCamelCase__ = 7.5,		lowerCamelCase__ = 1,		lowerCamelCase__ = 0.0,		lowerCamelCase__ = 100,		lowerCamelCase__ = None,		lowerCamelCase__ = "pil",		lowerCamelCase__ = True,		lowerCamelCase__ = 0.8,		lowerCamelCase__ = 0.1,		lowerCamelCase__ = 0.1,		):
        if isinstance(lowerCamelCase__,		lowerCamelCase__		) and len(lowerCamelCase__		) != batch_size:
            raise ValueError(f'''You have passed {batch_size} batch_size, but only {len(lowerCamelCase__		)} 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(lowerCamelCase__,		torch.Generator		) and batch_size > 1:
            A :       Union[str, Any]          =   [generator] + [None] * (batch_size - 1)

        A :       List[Any]          =   [
            ("""model""", self.coca_model is None),
            ("""tokenizer""", self.coca_tokenizer is None),
            ("""transform""", self.coca_transform is None),
        ]
        A :       List[Any]          =   [x[0] for x in coca_is_none if x[1]]
        A :       List[Any]          =   """, """.join(lowerCamelCase__		)
        # generate prompts with coca model if prompt is None
        if content_prompt is None:
            if len(lowerCamelCase__		):
                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.'''		)
            A :       Any          =   self.get_image_description(lowerCamelCase__		)
        if style_prompt is None:
            if len(lowerCamelCase__		):
                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.'''		)
            A :       Optional[Any]          =   self.get_image_description(lowerCamelCase__		)

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

        A :       Union[str, Any]          =   self.tokenizer(
            lowerCamelCase__,		padding="""max_length""",		max_length=self.tokenizer.model_max_length,		truncation=lowerCamelCase__,		return_tensors="""pt""",		)
        A :       str          =   self.text_encoder(style_text_input.input_ids.to(self.device		)		)[0]

        A :       List[Any]          =   slerp(lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__		)

        # duplicate text embeddings for each generation per prompt
        A :       Optional[Any]          =   text_embeddings.repeat_interleave(lowerCamelCase__,		dim=0		)

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

        self.scheduler.set_timesteps(lowerCamelCase__,		**lowerCamelCase__		)
        # 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		)

        A     ,							A :       int          =   self.get_timesteps(lowerCamelCase__,		lowerCamelCase__,		self.device		)
        A :       Union[str, Any]          =   timesteps[:1].repeat(lowerCamelCase__		)

        # Preprocess image
        A :       int          =   preprocess(lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__		)
        A :       Optional[Any]          =   self.prepare_latents(
            lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		text_embeddings.dtype,		self.device,		lowerCamelCase__		)

        A :       List[Any]          =   preprocess(lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__		)
        A :       List[str]          =   self.prepare_latents(
            lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		text_embeddings.dtype,		self.device,		lowerCamelCase__		)

        A :       str          =   slerp(lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__		)

        if clip_guidance_scale > 0:
            A :       str          =   self.get_clip_image_embeddings(lowerCamelCase__,		lowerCamelCase__		)
            A :       Dict          =   self.get_clip_image_embeddings(lowerCamelCase__,		lowerCamelCase__		)
            A :       List[str]          =   slerp(
                lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__		)

        # 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.
        A :       Tuple          =   guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
            A :       Optional[Any]          =   content_text_input.input_ids.shape[-1]
            A :       List[Any]          =   self.tokenizer([""""""],		padding="""max_length""",		max_length=lowerCamelCase__,		return_tensors="""pt"""		)
            A :       Tuple          =   self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]
            # duplicate unconditional embeddings for each generation per prompt
            A :       Optional[Any]          =   uncond_embeddings.repeat_interleave(lowerCamelCase__,		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
            A :       Tuple          =   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`.
        A :       str          =   (batch_size, self.unet.config.in_channels, height // 8, width // 8)
        A :       int          =   text_embeddings.dtype
        if latents is None:
            if self.device.type == "mps":
                # randn does not work reproducibly on mps
                A :       Union[str, Any]          =   torch.randn(lowerCamelCase__,		generator=lowerCamelCase__,		device="""cpu""",		dtype=lowerCamelCase__		).to(
                    self.device		)
            else:
                A :       str          =   torch.randn(lowerCamelCase__,		generator=lowerCamelCase__,		device=self.device,		dtype=lowerCamelCase__		)
        else:
            if latents.shape != latents_shape:
                raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''		)
            A :       int          =   latents.to(self.device		)

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

        # check if the scheduler accepts generator
        A :       List[str]          =   """generator""" in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
        if accepts_generator:
            A :       List[str]          =   generator

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

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

                # perform classifier free guidance
                if do_classifier_free_guidance:
                    A     ,							A :       Dict          =   noise_pred.chunk(2		)
                    A :       int          =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

                # compute the previous noisy sample x_t -> x_t-1
                A :       Union[str, Any]          =   self.scheduler.step(lowerCamelCase__,		lowerCamelCase__,		lowerCamelCase__,		**lowerCamelCase__		).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
        A :       List[str]          =   1 / 0.1_8215 * latents
        A :       Optional[int]          =   self.vae.decode(lowerCamelCase__		).sample

        A :       Tuple          =   (image / 2 + 0.5).clamp(0,		1		)
        A :       Union[str, Any]          =   image.cpu().permute(0,		2,		3,		1		).numpy()

        if output_type == "pil":
            A :       Union[str, Any]          =   self.numpy_to_pil(lowerCamelCase__		)

        if not return_dict:
            return (image, None)

        return StableDiffusionPipelineOutput(images=lowerCamelCase__,		nsfw_content_detected=lowerCamelCase__		)







from __future__ import annotations

from math import ceil, floor, sqrt




def 	__UpperCamelCase	(  _lowerCAmelCase = 200_0000		)       ->  int:



    """simple docstring"""






    A :       list[int]          =   [0]
    A :       int

    for idx in range(1      ,     ceil(sqrt(target * 2		) * 1.1		)		):
        triangle_numbers.append(triangle_numbers[-1] + idx		)

    # we want this to be as close as possible to target
    A :       int          =   0
    # the area corresponding to the grid that gives the product closest to target
    A :       int          =   0
    # an estimate of b, using the quadratic formula
    A :       float
    # the largest integer less than b_estimate
    A :       int
    # the largest integer less than b_estimate
    A :       int
    # the triangle number corresponding to b_floor
    A :       int
    # the triangle number corresponding to b_ceil
    A :       int

    for idx_a, triangle_a in enumerate(triangle_numbers[1:]      ,     1		):
        A :       Union[str, Any]          =   (-1 + sqrt(1 + 8 * target / triangle_a		)) / 2
        A :       List[Any]          =   floor(_lowerCAmelCase		)
        A :       Tuple          =   ceil(_lowerCAmelCase		)
        A :       int          =   triangle_numbers[b_floor]
        A :       Dict          =   triangle_numbers[b_ceil]

        if abs(target - triangle_b_first_guess * triangle_a		) < abs(
            target - best_product		):
            A :       Optional[int]          =   triangle_b_first_guess * triangle_a
            A :       Optional[int]          =   idx_a * b_floor

        if abs(target - triangle_b_second_guess * triangle_a		) < abs(
            target - best_product		):
            A :       Tuple          =   triangle_b_second_guess * triangle_a
            A :       Tuple          =   idx_a * b_ceil

    return area


if __name__ == "__main__":
     print(F"""{solution() = }""")


