






"""simple docstring"""
from typing import Callable, List, Optional, Union

import PIL
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPSegForImageSegmentation,
    CLIPSegProcessor,
    CLIPTextModel,
    CLIPTokenizer,
)

from diffusers import DiffusionPipeline
from diffusers.configuration_utils import FrozenDict
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionInpaintPipeline
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import deprecate, is_accelerate_available, logging


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







class      UpperCamelCase_			(SCREAMING_SNAKE_CASE_	):





	def __init__( self   :					Union[str, Any]      ,      lowerCAmelCase_   :					CLIPSegForImageSegmentation      ,      lowerCAmelCase_   :					CLIPSegProcessor      ,      lowerCAmelCase_   :					AutoencoderKL      ,      lowerCAmelCase_   :					CLIPTextModel      ,      lowerCAmelCase_   :					CLIPTokenizer      ,      lowerCAmelCase_   :					UNetaDConditionModel      ,      lowerCAmelCase_   :					Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]      ,      lowerCAmelCase_   :					StableDiffusionSafetyChecker      ,      lowerCAmelCase_   :					CLIPImageProcessor      ,      )		->     int:
				super().__init__()

				if hasattr(scheduler.config      ,      "steps_offset"		) and scheduler.config.steps_offset != 1:
							UpperCAmelCase_	:					int     	=   (
							    f"""The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"""
							    f""" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure """
							    'to update the config accordingly as leaving `steps_offset` might led to incorrect results'
							    ' in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,'
							    ' it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`'
							    ' file'
							)
							deprecate("steps_offset!=1"      ,      "1.0.0"      ,      _a      ,      standard_warn=_a		)
							UpperCAmelCase_	:					int     	=   dict(scheduler.config		)
							UpperCAmelCase_	:					Any     	=   1
							UpperCAmelCase_	:					List[Any]     	=   FrozenDict(_a		)

				if hasattr(scheduler.config      ,      "skip_prk_steps"		) and scheduler.config.skip_prk_steps is False:
							UpperCAmelCase_	:					str     	=   (
							    f"""The configuration file of this scheduler: {scheduler} has not set the configuration"""
							    ' `skip_prk_steps`. `skip_prk_steps` should be set to True in the configuration file. Please make'
							    ' sure to update the config accordingly as not setting `skip_prk_steps` in the config might lead to'
							    ' incorrect results in future versions. If you have downloaded this checkpoint from the Hugging Face'
							    ' Hub, it would be very nice if you could open a Pull request for the'
							    ' `scheduler/scheduler_config.json` file'
							)
							deprecate("skip_prk_steps not set"      ,      "1.0.0"      ,      _a      ,      standard_warn=_a		)
							UpperCAmelCase_	:					Dict     	=   dict(scheduler.config		)
							UpperCAmelCase_	:					Tuple     	=   True
							UpperCAmelCase_	:					Union[str, Any]     	=   FrozenDict(_a		)

				if safety_checker is None:
							logger.warning(
							    f"""You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"""
							    " that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
							    " results in services or applications open to the public. Both the diffusers team and Hugging Face"
							    " strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
							    " it only for use-cases that involve analyzing network behavior or auditing its results. For more"
							    " information, please have a look at https://github.com/huggingface/diffusers/pull/254 ."		)

				self.register_modules(
				    segmentation_model=_a      ,      segmentation_processor=_a      ,      vae=_a      ,      text_encoder=_a      ,      tokenizer=_a      ,      unet=_a      ,      scheduler=_a      ,      safety_checker=_a      ,      feature_extractor=_a      ,      )





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





	def 	_SCREAMING_SNAKE_CASE ( self   :					Optional[Any]		)		->     int:
				self.enable_attention_slicing(_a		)





	def 	_SCREAMING_SNAKE_CASE ( self   :					str		)		->     Any:
				if is_accelerate_available():
							from accelerate import cpu_offload
				else:
							raise ImportError("Please install accelerate via `pip install accelerate`"		)

				UpperCAmelCase_	:					Dict     	=   torch.device("cuda"		)

				for cpu_offloaded_model in [self.unet, self.text_encoder, self.vae, self.safety_checker]:
							if cpu_offloaded_model is not None:
										cpu_offload(_a      ,      _a		)





	@property
	# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
	def 	_SCREAMING_SNAKE_CASE ( self   :					Dict		)		->     int:
				if self.device != torch.device("meta"		) or not hasattr(self.unet      ,      "_hf_hook"		):
							return self.device
				for module in self.unet.modules():
							if (
							    hasattr(_a      ,      "_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()
	def __call__( self   :					List[str]      ,      lowerCAmelCase_   :					Union[str, List[str]]      ,      lowerCAmelCase_   :					Union[torch.FloatTensor, PIL.Image.Image]      ,      lowerCAmelCase_   :					str      ,      lowerCAmelCase_   :					int = 512      ,      lowerCAmelCase_   :					int = 512      ,      lowerCAmelCase_   :					int = 50      ,      lowerCAmelCase_   :					float = 7.5      ,      lowerCAmelCase_   :					Optional[Union[str, List[str]]] = None      ,      lowerCAmelCase_   :					Optional[int] = 1      ,      lowerCAmelCase_   :					float = 0.0      ,      lowerCAmelCase_   :					Optional[torch.Generator] = None      ,      lowerCAmelCase_   :					Optional[torch.FloatTensor] = None      ,      lowerCAmelCase_   :					Optional[str] = "pil"      ,      lowerCAmelCase_   :					bool = True      ,      lowerCAmelCase_   :					Optional[Callable[[int, int, torch.FloatTensor], None]] = None      ,      lowerCAmelCase_   :					int = 1      ,      **lowerCAmelCase_   :					Optional[Any]      ,      )		->     int:
				UpperCAmelCase_	:					Union[str, Any]     	=   self.segmentation_processor(
				    text=[text]      ,      images=[image]      ,      padding="max_length"      ,      return_tensors="pt"		).to(self.device		)
				UpperCAmelCase_	:					Union[str, Any]     	=   self.segmentation_model(**_a		)
				UpperCAmelCase_	:					List[str]     	=   torch.sigmoid(outputs.logits		).cpu().detach().unsqueeze(-1		).numpy()
				UpperCAmelCase_	:					Optional[Any]     	=   self.numpy_to_pil(_a		)[0].resize(image.size		)

				# Run inpainting pipeline with the generated mask
				UpperCAmelCase_	:					Tuple     	=   StableDiffusionInpaintPipeline(
				    vae=self.vae      ,      text_encoder=self.text_encoder      ,      tokenizer=self.tokenizer      ,      unet=self.unet      ,      scheduler=self.scheduler      ,      safety_checker=self.safety_checker      ,      feature_extractor=self.feature_extractor      ,      )
				return inpainting_pipeline(
				    prompt=_a      ,      image=_a      ,      mask_image=_a      ,      height=_a      ,      width=_a      ,      num_inference_steps=_a      ,      guidance_scale=_a      ,      negative_prompt=_a      ,      num_images_per_prompt=_a      ,      eta=_a      ,      generator=_a      ,      latents=_a      ,      output_type=_a      ,      return_dict=_a      ,      callback=_a      ,      callback_steps=_a      ,      )








'''simple docstring'''





from typing import Optional, Tuple

import jax
import jax.numpy as jnp
from flax import linen as nn
from flax.core.frozen_dict import FrozenDict
from transformers import CLIPConfig, FlaxPreTrainedModel
from transformers.models.clip.modeling_flax_clip import FlaxCLIPVisionModule

def 			a_	(				_lowerCAmelCase			,_lowerCAmelCase			,_lowerCAmelCase=1E-12			) -> List[Any]:
  __lowerCamelCase			:			int       			=					jnp.divide(emb_a.T			,jnp.clip(jnp.linalg.norm(_lowerCAmelCase			,axis=1			)			,a_min=_lowerCAmelCase			)			).T
  __lowerCamelCase			:			Tuple       			=					jnp.divide(emb_a.T			,jnp.clip(jnp.linalg.norm(_lowerCAmelCase			,axis=1			)			,a_min=_lowerCAmelCase			)			).T
  return jnp.matmul(_lowerCAmelCase			,norm_emb_a.T			)






class    lowerCamelCase_       (							nn.Module	):
  """simple docstring"""




  a_				   =42
  a_				   =jnp.floataa




  def 		_lowercase	(					self :      Optional[Any]  )   ->   Tuple:
    __lowerCamelCase			:			Optional[int]       			=					FlaxCLIPVisionModule(self.config.vision_config  )
    __lowerCamelCase			:			Any       			=					nn.Dense(self.config.projection_dim     ,      use_bias=_a     ,      dtype=self.dtype  )

    __lowerCamelCase			:			str       			=					self.param('concept_embeds'     ,      jax.nn.initializers.ones     ,      (17, self.config.projection_dim)  )
    __lowerCamelCase			:			str       			=					self.param(
        'special_care_embeds'     ,      jax.nn.initializers.ones     ,      (3, self.config.projection_dim)  )

    __lowerCamelCase			:			Optional[int]       			=					self.param('concept_embeds_weights'     ,      jax.nn.initializers.ones     ,      (17,)  )
    __lowerCamelCase			:			Dict       			=					self.param('special_care_embeds_weights'     ,      jax.nn.initializers.ones     ,      (3,)  )




  def __call__(					self :      Tuple     ,      _a :      Any  )   ->   Optional[int]:
    __lowerCamelCase			:			int       			=					self.vision_model(_a  )[1]
    __lowerCamelCase			:			Optional[int]       			=					self.visual_projection(_a  )

    __lowerCamelCase			:			Optional[Any]       			=					jax_cosine_distance(_a     ,      self.special_care_embeds  )
    __lowerCamelCase			:			Tuple       			=					jax_cosine_distance(_a     ,      self.concept_embeds  )

    # increase this value to create a stronger `nfsw` filter
    # at the cost of increasing the possibility of filtering benign image inputs
    __lowerCamelCase			:			Dict       			=					0.0

    __lowerCamelCase			:			str       			=					special_cos_dist - self.special_care_embeds_weights[None, :] + adjustment
    __lowerCamelCase			:			Optional[int]       			=					jnp.round(_a     ,      3  )
    __lowerCamelCase			:			Dict       			=					jnp.any(special_scores > 0     ,      axis=1     ,      keepdims=_a  )
    # Use a lower threshold if an image has any special care concept
    __lowerCamelCase			:			Dict       			=					is_special_care * 0.01

    __lowerCamelCase			:			Optional[int]       			=					cos_dist - self.concept_embeds_weights[None, :] + special_adjustment
    __lowerCamelCase			:			List[str]       			=					jnp.round(_a     ,      3  )
    __lowerCamelCase			:			List[Any]       			=					jnp.any(concept_scores > 0     ,      axis=1  )

    return has_nsfw_concepts






class    lowerCamelCase_       (							SCREAMING_SNAKE_CASE_	):
  """simple docstring"""




  a_				   =CLIPConfig
  a_				   ="""clip_input"""
  a_				   =FlaxStableDiffusionSafetyCheckerModule




  def __init__(					self :      List[str]     ,      _a :      CLIPConfig     ,      _a :      Optional[Tuple] = None     ,      _a :      int = 0     ,      _a :      jnp.dtype = jnp.floataa     ,      _a :      bool = True     ,      **_a :      List[Any]     ,      )   ->   List[Any]:
    if input_shape is None:
      __lowerCamelCase			:			str       			=					(1, 224, 224, 3)
    __lowerCamelCase			:			List[str]       			=					self.module_class(config=_a     ,      dtype=_a     ,      **_a  )
    super().__init__(_a     ,      _a     ,      input_shape=_a     ,      seed=_a     ,      dtype=_a     ,      _do_init=_do_init  )




  def 		_lowercase	(					self :      int     ,      _a :      jax.random.KeyArray     ,      _a :      Tuple     ,      _a :      FrozenDict = None  )   ->   FrozenDict:
    # init input tensor
    __lowerCamelCase			:			Optional[int]       			=					jax.random.normal(_a     ,      _a  )

    __lowerCamelCase			,__lowerCamelCase			:			Optional[Any]       			=					jax.random.split(_a  )
    __lowerCamelCase			:			int       			=					{'params': params_rng, 'dropout': dropout_rng}

    __lowerCamelCase			:			Tuple       			=					self.module.init(_a     ,      _a  )['params']

    return random_params




  def __call__(					self :      Optional[int]     ,      _a :      int     ,      _a :      dict = None     ,      )   ->   Dict:
    __lowerCamelCase			:			str       			=					jnp.transpose(_a     ,      (0, 2, 3, 1)  )

    return self.module.apply(
        {'params': params or self.params}     ,      jnp.array(_a     ,      dtype=jnp.floataa  )     ,      rngs={}     ,      )






