


"""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     __lowerCAmelCase		(		lowercase    :    List[str]		,				lowercase    :    Dict		,				lowercase    :    List[str]=1e-1_2					)      ->		str:


    """simple docstring"""





    snake_case       : List[str]    =       jnp.divide(emb_a.T		,				jnp.clip(jnp.linalg.norm(lowercase		,				axis=1					)		,				a_min=lowercase					)					).T
    snake_case       : Any    =       jnp.divide(emb_a.T		,				jnp.clip(jnp.linalg.norm(lowercase		,				axis=1					)		,				a_min=lowercase					)					).T
    return jnp.matmul(lowercase		,				norm_emb_a.T					)







class _lowerCAmelCase							( nn.Module      ):
   __UpperCAmelCase					:			CLIPConfig
   __UpperCAmelCase					:			jnp.dtype				    =    jnp.floataa







   def 					lowerCamelCase						(      self		)      ->						Optional[int]:

       '''simple docstring'''




       snake_case       : int    =       FlaxCLIPVisionModule(self.config.vision_config		)
       snake_case       : str    =       nn.Dense(self.config.projection_dim		,      use_bias=UpperCamelCase__		,      dtype=self.dtype		)

       snake_case       : List[str]    =       self.param("concept_embeds"		,      jax.nn.initializers.ones		,      (17, self.config.projection_dim)		)
       snake_case       : Dict    =       self.param(
           "special_care_embeds"		,      jax.nn.initializers.ones		,      (3, self.config.projection_dim)		)

       snake_case       : List[Any]    =       self.param("concept_embeds_weights"		,      jax.nn.initializers.ones		,      (17,)		)
       snake_case       : Any    =       self.param("special_care_embeds_weights"		,      jax.nn.initializers.ones		,      (3,)		)







   def __call__(      self		,      UpperCamelCase__		)      ->						Tuple:

       '''simple docstring'''




       snake_case       : Dict    =       self.vision_model(UpperCamelCase__		)[1]
       snake_case       : int    =       self.visual_projection(UpperCamelCase__		)

       snake_case       : List[str]    =       jax_cosine_distance(UpperCamelCase__		,      self.special_care_embeds		)
       snake_case       : Optional[Any]    =       jax_cosine_distance(UpperCamelCase__		,      self.concept_embeds		)

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

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

       snake_case       : Optional[Any]    =       cos_dist - self.concept_embeds_weights[None, :] + special_adjustment
       snake_case       : Any    =       jnp.round(UpperCamelCase__		,      3		)
       snake_case       : Any    =       jnp.any(concept_scores > 0		,      axis=1		)

       return has_nsfw_concepts







class _lowerCAmelCase							( snake_case_      ):
   __UpperCAmelCase					:			Dict				    =    CLIPConfig
   __UpperCAmelCase					:			Optional[Any]				    =    '''clip_input'''
   __UpperCAmelCase					:			Optional[int]				    =    FlaxStableDiffusionSafetyCheckerModule







   def __init__(      self		,      UpperCamelCase__		,      UpperCamelCase__ = None		,      UpperCamelCase__ = 0		,      UpperCamelCase__ = jnp.floataa		,      UpperCamelCase__ = True		,      **UpperCamelCase__		,      )      ->						int:

       '''simple docstring'''




       if input_shape is None:
           snake_case       : Union[str, Any]    =       (1, 224, 224, 3)
       snake_case       : Tuple    =       self.module_class(config=UpperCamelCase__		,      dtype=UpperCamelCase__		,      **UpperCamelCase__		)
       super().__init__(UpperCamelCase__		,      UpperCamelCase__		,      input_shape=UpperCamelCase__		,      seed=UpperCamelCase__		,      dtype=UpperCamelCase__		,      _do_init=_do_init		)







   def 					lowerCamelCase						(      self		,      UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__ = None		)      ->						FrozenDict:

       '''simple docstring'''




       snake_case       : Any    =       jax.random.normal(UpperCamelCase__		,      UpperCamelCase__		)

       snake_case   ,snake_case       : str    =       jax.random.split(UpperCamelCase__		)
       snake_case       : int    =       {"params": params_rng, "dropout": dropout_rng}

       snake_case       : str    =       self.module.init(UpperCamelCase__		,      UpperCamelCase__		)["params"]

       return random_params







   def __call__(      self		,      UpperCamelCase__		,      UpperCamelCase__ = None		,      )      ->						Optional[Any]:

       '''simple docstring'''




       snake_case       : Optional[int]    =       jnp.transpose(UpperCamelCase__		,      (0, 2, 3, 1)		)

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









"""simple docstring"""



import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest







class _lowerCAmelCase							( snake_case_      ):
   __UpperCAmelCase					:			List[str]				    =    (KDPMaDiscreteScheduler,)
   __UpperCAmelCase					:			List[Any]				    =    10







   def 					lowerCamelCase						(      self		,      **UpperCamelCase__		)      ->						Optional[int]:

       '''simple docstring'''




       snake_case       : List[str]    =       {
           "num_train_timesteps": 1100,
           "beta_start": 0.0001,
           "beta_end": 0.02,
           "beta_schedule": "linear",
       }

       config.update(**UpperCamelCase__		)
       return config







   def 					lowerCamelCase						(      self		)      ->						Optional[int]:

       '''simple docstring'''




       for timesteps in [10, 50, 100, 1000]:
           self.check_over_configs(num_train_timesteps=UpperCamelCase__		)







   def 					lowerCamelCase						(      self		)      ->						str:

       '''simple docstring'''




       for beta_start, beta_end in zip([0.00001, 0.0001, 0.001]		,      [0.0002, 0.002, 0.02]		):
           self.check_over_configs(beta_start=UpperCamelCase__		,      beta_end=UpperCamelCase__		)







   def 					lowerCamelCase						(      self		)      ->						Optional[Any]:

       '''simple docstring'''




       for schedule in ["linear", "scaled_linear"]:
           self.check_over_configs(beta_schedule=UpperCamelCase__		)







   def 					lowerCamelCase						(      self		)      ->						List[str]:

       '''simple docstring'''




       for prediction_type in ["epsilon", "v_prediction"]:
           self.check_over_configs(prediction_type=UpperCamelCase__		)







   def 					lowerCamelCase						(      self		)      ->						Optional[Any]:

       '''simple docstring'''




       snake_case       : str    =       self.scheduler_classes[0]
       snake_case       : str    =       self.get_scheduler_config(prediction_type="v_prediction"		)
       snake_case       : Dict    =       scheduler_class(**UpperCamelCase__		)

       scheduler.set_timesteps(self.num_inference_steps		)

       snake_case       : List[Any]    =       self.dummy_model()
       snake_case       : List[str]    =       self.dummy_sample_deter * scheduler.init_noise_sigma
       snake_case       : Optional[Any]    =       sample.to(UpperCamelCase__		)

       for i, t in enumerate(scheduler.timesteps		):
           snake_case       : int    =       scheduler.scale_model_input(UpperCamelCase__		,      UpperCamelCase__		)

           snake_case       : int    =       model(UpperCamelCase__		,      UpperCamelCase__		)

           snake_case       : Optional[Any]    =       scheduler.step(UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__		)
           snake_case       : Optional[Any]    =       output.prev_sample

       snake_case       : Optional[Any]    =       torch.sum(torch.abs(UpperCamelCase__		)		)
       snake_case       : Any    =       torch.mean(torch.abs(UpperCamelCase__		)		)

       if torch_device in ["cpu", "mps"]:
           assert abs(result_sum.item() - 4.6934e-07		) < 1e-2
           assert abs(result_mean.item() - 6.1112e-10		) < 1e-3
       else:
           # CUDA
           assert abs(result_sum.item() - 4.693_4286_5017_0972e-07		) < 1e-2
           assert abs(result_mean.item() - 0.0002		) < 1e-3







   def 					lowerCamelCase						(      self		)      ->						Dict:

       '''simple docstring'''




       if torch_device == "mps":
           return
       snake_case       : Optional[Any]    =       self.scheduler_classes[0]
       snake_case       : Optional[int]    =       self.get_scheduler_config()
       snake_case       : Dict    =       scheduler_class(**UpperCamelCase__		)

       scheduler.set_timesteps(self.num_inference_steps		)

       snake_case       : List[Any]    =       self.dummy_model()
       snake_case       : Union[str, Any]    =       self.dummy_sample_deter * scheduler.init_noise_sigma
       snake_case       : List[str]    =       sample.to(UpperCamelCase__		)

       for i, t in enumerate(scheduler.timesteps		):
           snake_case       : List[Any]    =       scheduler.scale_model_input(UpperCamelCase__		,      UpperCamelCase__		)

           snake_case       : Optional[int]    =       model(UpperCamelCase__		,      UpperCamelCase__		)

           snake_case       : Optional[int]    =       scheduler.step(UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__		)
           snake_case       : int    =       output.prev_sample

       snake_case       : int    =       torch.sum(torch.abs(UpperCamelCase__		)		)
       snake_case       : Union[str, Any]    =       torch.mean(torch.abs(UpperCamelCase__		)		)

       if torch_device in ["cpu", "mps"]:
           assert abs(result_sum.item() - 20.4125		) < 1e-2
           assert abs(result_mean.item() - 0.0266		) < 1e-3
       else:
           # CUDA
           assert abs(result_sum.item() - 20.4125		) < 1e-2
           assert abs(result_mean.item() - 0.0266		) < 1e-3







   def 					lowerCamelCase						(      self		)      ->						Optional[int]:

       '''simple docstring'''




       if torch_device == "mps":
           return
       snake_case       : Any    =       self.scheduler_classes[0]
       snake_case       : Any    =       self.get_scheduler_config()
       snake_case       : Union[str, Any]    =       scheduler_class(**UpperCamelCase__		)

       scheduler.set_timesteps(self.num_inference_steps		,      device=UpperCamelCase__		)

       snake_case       : Optional[Any]    =       self.dummy_model()
       snake_case       : Optional[int]    =       self.dummy_sample_deter.to(UpperCamelCase__		) * scheduler.init_noise_sigma

       for t in scheduler.timesteps:
           snake_case       : Optional[Any]    =       scheduler.scale_model_input(UpperCamelCase__		,      UpperCamelCase__		)

           snake_case       : Any    =       model(UpperCamelCase__		,      UpperCamelCase__		)

           snake_case       : List[Any]    =       scheduler.step(UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__		)
           snake_case       : Dict    =       output.prev_sample

       snake_case       : Dict    =       torch.sum(torch.abs(UpperCamelCase__		)		)
       snake_case       : str    =       torch.mean(torch.abs(UpperCamelCase__		)		)

       if str(UpperCamelCase__		).startswith("cpu"		):
           # The following sum varies between 148 and 156 on mps. Why?
           assert abs(result_sum.item() - 20.4125		) < 1e-2
           assert abs(result_mean.item() - 0.0266		) < 1e-3
       else:
           # CUDA
           assert abs(result_sum.item() - 20.4125		) < 1e-2
           assert abs(result_mean.item() - 0.0266		) < 1e-3





