



'''simple docstring'''







import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


SCREAMING_SNAKE_CASE__     : str				=   logging.get_logger(__name__)







def  SCREAMING_SNAKE_CASE__ (				SCREAMING_SNAKE_CASE ,		SCREAMING_SNAKE_CASE     ):
		SCREAMING_SNAKE_CASE_						:int										=      nn.functional.normalize(lowerCAmelCase__     )
		SCREAMING_SNAKE_CASE_						:Optional[int]										=      nn.functional.normalize(lowerCAmelCase__     )
		return torch.mm(lowerCAmelCase__ ,		normalized_text_embeds.t()     )





class 				__lowerCAmelCase(							lowerCAmelCase__      ):
			__snake_case  :		int		=      CLIPConfig

			__snake_case  :		Dict		=      ['CLIPEncoderLayer']






			def __init__(   self  :							int      ,       SCREAMING_SNAKE_CASE  :							CLIPConfig	):




					"""simple docstring"""







					super().__init__(lowercase__	)

					SCREAMING_SNAKE_CASE_						:Optional[int]										=      CLIPVisionModel(config.vision_config	)
					SCREAMING_SNAKE_CASE_						:int										=      nn.Linear(config.vision_config.hidden_size      ,       config.projection_dim      ,       bias=lowercase__	)

					SCREAMING_SNAKE_CASE_						:int										=      nn.Parameter(torch.ones(17      ,       config.projection_dim	)      ,       requires_grad=lowercase__	)
					SCREAMING_SNAKE_CASE_						:List[Any]										=      nn.Parameter(torch.ones(3      ,       config.projection_dim	)      ,       requires_grad=lowercase__	)

					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      nn.Parameter(torch.ones(17	)      ,       requires_grad=lowercase__	)
					SCREAMING_SNAKE_CASE_						:str										=      nn.Parameter(torch.ones(3	)      ,       requires_grad=lowercase__	)






			@torch.no_grad()
			def 				_lowercase	(   self  :							int      ,       SCREAMING_SNAKE_CASE  :							List[str]      ,       SCREAMING_SNAKE_CASE  :							Union[str, Any]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:str										=      self.vision_model(lowercase__	)[1]  # pooled_output
					SCREAMING_SNAKE_CASE_						:Tuple										=      self.visual_projection(lowercase__	)

					# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
					SCREAMING_SNAKE_CASE_						:Any										=      cosine_distance(lowercase__      ,       self.special_care_embeds	).cpu().float().numpy()
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      cosine_distance(lowercase__      ,       self.concept_embeds	).cpu().float().numpy()

					SCREAMING_SNAKE_CASE_						:List[str]										=      []
					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      image_embeds.shape[0]
					for i in range(lowercase__	):
							SCREAMING_SNAKE_CASE_						:Optional[Any]										=      {'special_scores': {}, 'special_care': [], 'concept_scores': {}, 'bad_concepts': []}

							# increase this value to create a stronger `nfsw` filter
							# at the cost of increasing the possibility of filtering benign images
							SCREAMING_SNAKE_CASE_						:str										=      0.0

							for concept_idx in range(len(special_cos_dist[0]	)	):
									SCREAMING_SNAKE_CASE_						:List[str]										=      special_cos_dist[i][concept_idx]
									SCREAMING_SNAKE_CASE_						:Any										=      self.special_care_embeds_weights[concept_idx].item()
									SCREAMING_SNAKE_CASE_						:List[str]										=      round(concept_cos - concept_threshold + adjustment      ,       3	)
									if result_img["special_scores"][concept_idx] > 0:
											result_img["special_care"].append({concept_idx, result_img['special_scores'][concept_idx]}	)
											SCREAMING_SNAKE_CASE_						:List[Any]										=      0.01

							for concept_idx in range(len(cos_dist[0]	)	):
									SCREAMING_SNAKE_CASE_						:int										=      cos_dist[i][concept_idx]
									SCREAMING_SNAKE_CASE_						:int										=      self.concept_embeds_weights[concept_idx].item()
									SCREAMING_SNAKE_CASE_						:Dict										=      round(concept_cos - concept_threshold + adjustment      ,       3	)
									if result_img["concept_scores"][concept_idx] > 0:
											result_img["bad_concepts"].append(lowercase__	)

							result.append(lowercase__	)

					SCREAMING_SNAKE_CASE_						:Any										=      [len(res['bad_concepts']	) > 0 for res in result]

					for idx, has_nsfw_concept in enumerate(lowercase__	):
							if has_nsfw_concept:
									if torch.is_tensor(lowercase__	) or torch.is_tensor(images[0]	):
											SCREAMING_SNAKE_CASE_						:Dict										=      torch.zeros_like(images[idx]	)  # black image
									else:
											SCREAMING_SNAKE_CASE_						:Dict										=      np.zeros(images[idx].shape	)  # black image

					if any(lowercase__	):
							logger.warning(
							    'Potential NSFW content was detected in one or more images. A black image will be returned instead.'
							    ' Try again with a different prompt and/or seed.'	)

					return images, has_nsfw_concepts






			@torch.no_grad()
			def 				_lowercase	(   self  :							Dict      ,       SCREAMING_SNAKE_CASE  :							torch.FloatTensor      ,       SCREAMING_SNAKE_CASE  :							torch.FloatTensor	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Any										=      self.vision_model(lowercase__	)[1]  # pooled_output
					SCREAMING_SNAKE_CASE_						:Tuple										=      self.visual_projection(lowercase__	)

					SCREAMING_SNAKE_CASE_						:Any										=      cosine_distance(lowercase__      ,       self.special_care_embeds	)
					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      cosine_distance(lowercase__      ,       self.concept_embeds	)

					# increase this value to create a stronger `nsfw` filter
					# at the cost of increasing the possibility of filtering benign images
					SCREAMING_SNAKE_CASE_						:Tuple										=      0.0

					SCREAMING_SNAKE_CASE_						:List[Any]										=      special_cos_dist - self.special_care_embeds_weights + adjustment
					# special_scores = special_scores.round(decimals=3)
					SCREAMING_SNAKE_CASE_						:Dict										=      torch.any(special_scores > 0      ,       dim=1	)
					SCREAMING_SNAKE_CASE_						:Any										=      special_care * 0.01
					SCREAMING_SNAKE_CASE_						:List[Any]										=      special_adjustment.unsqueeze(1	).expand(-1      ,       cos_dist.shape[1]	)

					SCREAMING_SNAKE_CASE_						:List[str]										=      (cos_dist - self.concept_embeds_weights) + special_adjustment
					# concept_scores = concept_scores.round(decimals=3)
					SCREAMING_SNAKE_CASE_						:Dict										=      torch.any(concept_scores > 0      ,       dim=1	)

					SCREAMING_SNAKE_CASE_						:List[Any]										=      0.0  # black image

					return images, has_nsfw_concepts

from typing import Union

import torch
from PIL import Image
from torchvision import transforms as tfms
from tqdm.auto import tqdm
from transformers import CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
class 							a_					(		SCREAMING_SNAKE_CASE		):






 '''simple docstring'''






 def __init__(     self   :    List[Any]     ,       lowercase__   :    AutoencoderKL     ,       lowercase__   :    CLIPTextModel     ,       lowercase__   :    CLIPTokenizer     ,       lowercase__   :    UNetaDConditionModel     ,       lowercase__   :    Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler]     ,       ):







        '''simple docstring'''



        super().__init__()

        self.register_modules(vae=lowercase__     ,       text_encoder=lowercase__     ,       tokenizer=lowercase__     ,       unet=lowercase__     ,       scheduler=lowercase__)




 def 						__snake_case		(     self   :    List[Any]     ,       lowercase__   :    str):







        '''simple docstring'''



        with torch.no_grad():
               lowerCAmelCase__	      =   self.vae.encode(tfms.ToTensor()(lowercase__).unsqueeze(0).to(self.device) * 2 - 1)
               lowerCAmelCase__	      =   0.18_215 * latent.latent_dist.sample()
        return latent




 def 						__snake_case		(     self   :    Tuple     ,       lowercase__   :    str):







        '''simple docstring'''



        lowerCAmelCase__	      =   (1 / 0.18_215) * latent
        with torch.no_grad():
               lowerCAmelCase__	      =   self.vae.decode(lowercase__).sample
        lowerCAmelCase__	      =   (img / 2 + 0.5).clamp(0     ,       1)
        lowerCAmelCase__	      =   img.detach().cpu().permute(0     ,       2     ,       3     ,       1).numpy()
        lowerCAmelCase__	      =   (img * 255).round().astype('uint8')
        return Image.fromarray(img[0])




 def 						__snake_case		(     self   :    Union[str, Any]     ,       lowercase__   :    Optional[int]):







        '''simple docstring'''



        lowerCAmelCase__	      =   self.tokenizer(
            lowercase__     ,       padding='max_length'     ,       max_length=self.tokenizer.model_max_length     ,       truncation=lowercase__     ,       return_tensors='pt'     ,       )

        lowerCAmelCase__	      =   self.text_encoder(text_input.input_ids.to(self.device))[0]

        lowerCAmelCase__	      =   self.tokenizer(
            ''     ,       padding='max_length'     ,       max_length=self.tokenizer.model_max_length     ,       truncation=lowercase__     ,       return_tensors='pt'     ,       )

        lowerCAmelCase__	      =   self.text_encoder(uncond_input.input_ids.to(self.device))[0]

        return torch.cat([uncond_embedding, text_embedding])




 def __call__(     self   :    List[Any]     ,       lowercase__   :    Image.Image     ,       lowercase__   :    str     ,       lowercase__   :    float = 0.3     ,       lowercase__   :    float = 0.6     ,       lowercase__   :    float = 0.5     ,       lowercase__   :    int = 42     ,       lowercase__   :    int = 50     ,       lowercase__   :    float = 7.5     ,       ):







        '''simple docstring'''



        lowerCAmelCase__	      =   steps - int(kmin * steps)
        lowerCAmelCase__	      =   steps - int(kmax * steps)

        lowerCAmelCase__	      =   self.prep_text(lowercase__)

        self.scheduler.set_timesteps(lowercase__)

        lowerCAmelCase__      ,   lowerCAmelCase__	      =   img.size
        lowerCAmelCase__	      =   self.encode(lowercase__)

        torch.manual_seed(lowercase__)
        lowerCAmelCase__	      =   torch.randn(
            (1, self.unet.config.in_channels, height // 8, width // 8)     ,       ).to(self.device)

        lowerCAmelCase__	      =   self.scheduler.add_noise(
            lowercase__     ,       lowercase__     ,       timesteps=self.scheduler.timesteps[tmax]     ,       )

        lowerCAmelCase__	      =   torch.cat([latents] * 2)

        lowerCAmelCase__	      =   self.scheduler.scale_model_input(lowercase__     ,       self.scheduler.timesteps[tmax])

        with torch.no_grad():
               lowerCAmelCase__	      =   self.unet(
                   lowercase__     ,       self.scheduler.timesteps[tmax]     ,       encoder_hidden_states=lowercase__     ,       ).sample

        lowerCAmelCase__      ,   lowerCAmelCase__	      =   pred.chunk(2)
        lowerCAmelCase__	      =   pred_uncond + guidance_scale * (pred_text - pred_uncond)

        lowerCAmelCase__	      =   self.scheduler.step(lowercase__     ,       self.scheduler.timesteps[tmax]     ,       lowercase__).prev_sample

        for i, t in enumerate(tqdm(self.scheduler.timesteps)):
               if i > tmax:
                      if i < tmin:  # layout generation phase
                             lowerCAmelCase__	      =   self.scheduler.add_noise(
                                 lowercase__     ,       lowercase__     ,       timesteps=lowercase__     ,       )

                             lowerCAmelCase__	      =   (mix_factor * latents) + (
                                 1 - mix_factor
                             ) * orig_latents  # interpolating between layout noise and conditionally generated noise to preserve layout sematics
                             lowerCAmelCase__	      =   torch.cat([input] * 2)

                      else:  # content generation phase
                             lowerCAmelCase__	      =   torch.cat([latents] * 2)

                      lowerCAmelCase__	      =   self.scheduler.scale_model_input(lowercase__     ,       lowercase__)

                      with torch.no_grad():
                             lowerCAmelCase__	      =   self.unet(
                                 lowercase__     ,       lowercase__     ,       encoder_hidden_states=lowercase__     ,       ).sample

                      lowerCAmelCase__      ,   lowerCAmelCase__	      =   pred.chunk(2)
                      lowerCAmelCase__	      =   pred_uncond + guidance_scale * (pred_text - pred_uncond)

                      lowerCAmelCase__	      =   self.scheduler.step(lowercase__     ,       lowercase__     ,       lowercase__).prev_sample

        return self.decode(lowercase__)
