





"""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_(__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=1e-12  )->    Optional[int]:
							_SCREAMING_SNAKE_CASE		:						Dict			  =    jnp.divide(emb_a.T     ,	jnp.clip(jnp.linalg.norm(__SCREAMING_SNAKE_CASE     ,	axis=1  )     ,	a_min=__SCREAMING_SNAKE_CASE  )  ).T
							_SCREAMING_SNAKE_CASE		:						Any			  =    jnp.divide(emb_a.T     ,	jnp.clip(jnp.linalg.norm(__SCREAMING_SNAKE_CASE     ,	axis=1  )     ,	a_min=__SCREAMING_SNAKE_CASE  )  ).T
							return jnp.matmul(__SCREAMING_SNAKE_CASE     ,	norm_emb_a.T  )




class  _snake_case		(    nn.Module ):







				"""simple docstring"""






				a			=						42
				a			=						jnp.floataa





				def 		_lowerCAmelCase		(      self							:							Union[str, Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    FlaxCLIPVisionModule(self.config.vision_config)
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    nn.Dense(self.config.projection_dim      ,    use_bias=_A      ,    dtype=self.dtype)

											_SCREAMING_SNAKE_CASE		:						Dict			  =    self.param("""concept_embeds"""      ,    jax.nn.initializers.ones      ,    (1_7, self.config.projection_dim))
											_SCREAMING_SNAKE_CASE		:						Any			  =    self.param(
											    """special_care_embeds"""      ,    jax.nn.initializers.ones      ,    (3, self.config.projection_dim))

											_SCREAMING_SNAKE_CASE		:						Dict			  =    self.param("""concept_embeds_weights"""      ,    jax.nn.initializers.ones      ,    (1_7,))
											_SCREAMING_SNAKE_CASE		:						int			  =    self.param("""special_care_embeds_weights"""      ,    jax.nn.initializers.ones      ,    (3,))





				def __call__(      self							:							List[Any]      ,    _A							:							str):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Dict			  =    self.vision_model(_A)[1]
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    self.visual_projection(_A)

											_SCREAMING_SNAKE_CASE		:						str			  =    jax_cosine_distance(_A      ,    self.special_care_embeds)
											_SCREAMING_SNAKE_CASE		:						Any			  =    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
											_SCREAMING_SNAKE_CASE		:						Dict			  =    0.0

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

											_SCREAMING_SNAKE_CASE		:						Dict			  =    cos_dist - self.concept_embeds_weights[None, :] + special_adjustment
											_SCREAMING_SNAKE_CASE		:						int			  =    jnp.round(_A      ,    3)
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    jnp.any(concept_scores > 0      ,    axis=1)

											return has_nsfw_concepts




class  _snake_case		(    __snake_case ):







				"""simple docstring"""






				a			=						CLIPConfig
				a			=						"clip_input"
				a			=						FlaxStableDiffusionSafetyCheckerModule





				def __init__(      self							:							Optional[int]      ,    _A							:							CLIPConfig      ,    _A							:							Optional[Tuple] = None      ,    _A							:							int = 0      ,    _A							:							jnp.dtype = jnp.floataa      ,    _A							:							bool = True      ,    **_A							:							Optional[int]      ,    ):



											"""simple docstring"""







											if input_shape is None:
																		_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    (1, 2_2_4, 2_2_4, 3)
											_SCREAMING_SNAKE_CASE		:						Any			  =    self.module_class(config=_A      ,    dtype=_A      ,    **_A)
											super().__init__(_A      ,    _A      ,    input_shape=_A      ,    seed=_A      ,    dtype=_A      ,    _do_init=_do_init)





				def 		_lowerCAmelCase		(      self							:							Dict      ,    _A							:							jax.random.KeyArray      ,    _A							:							Tuple      ,    _A							:							FrozenDict = None):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Dict			  =    jax.random.normal(_A      ,    _A)

											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    jax.random.split(_A)
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    {"""params""": params_rng, """dropout""": dropout_rng}

											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    self.module.init(_A      ,    _A)["""params"""]

											return random_params





				def __call__(      self							:							Tuple      ,    _A							:							Union[str, Any]      ,    _A							:							dict = None      ,    ):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Tuple			  =    jnp.transpose(_A      ,    (0, 2, 3, 1))

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



"""simple docstring"""





import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
			from sagemaker import Session, TrainingJobAnalytics
			from sagemaker.huggingface import HuggingFace




@pytest.mark.skipif(
    literal_eval(os.getenv("TEST_SAGEMAKER"   ,       "False" ) ) is not True   ,       reason="Skipping test because should only be run when releasing minor transformers version"   ,       )
@pytest.mark.usefixtures("sm_env" )
@parameterized_class(
    [
        {
            "framework": "pytorch",
            "script": "run_glue.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 6_50, "eval_accuracy": 0.6, "eval_loss": 0.9},
        },
        {
            "framework": "tensorflow",
            "script": "run_tf.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 6_00, "eval_accuracy": 0.3, "eval_loss": 0.9},
        },
    ] )
class  _snake_case		(    unittest.TestCase ):







				"""simple docstring"""






				def 		_lowerCAmelCase		(      self							:							Optional[int]):



											"""simple docstring"""







											if self.framework == "pytorch":
																		subprocess.run(
																		    f"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()      ,    encoding="""utf-8"""      ,    check=_A      ,    )
											assert hasattr(self      ,    """env""")





				def 		_lowerCAmelCase		(      self							:							Union[str, Any]      ,    _A							:							str=1):



											"""simple docstring"""







											return HuggingFace(
											    entry_point=self.script      ,    source_dir=self.env.test_path      ,    role=self.env.role      ,    image_uri=self.env.image_uri      ,    base_job_name=f"""{self.env.base_job_name}-single"""      ,    instance_count=_A      ,    instance_type=self.instance_type      ,    debugger_hook_config=_A      ,    hyperparameters={**self.env.hyperparameters, """model_name_or_path""": self.model_name_or_path}      ,    metric_definitions=self.env.metric_definitions      ,    py_version="""py36"""      ,    )





				def 		_lowerCAmelCase		(      self							:							Union[str, Any]      ,    _A							:							Union[str, Any]):



											"""simple docstring"""







											TrainingJobAnalytics(_A).export_csv(f"""{self.env.test_path}/{job_name}_metrics.csv""")





				def 		_lowerCAmelCase		(      self							:							Any):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						str			  =    self.create_estimator()

											# run training
											estimator.fit()

											# result dataframe
											_SCREAMING_SNAKE_CASE		:						Any			  =    TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()

											# extract kpis
											_SCREAMING_SNAKE_CASE		:						Any			  =    list(result_metrics_df[result_metrics_df.metric_name == """eval_accuracy"""]["""value"""])
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    list(result_metrics_df[result_metrics_df.metric_name == """eval_loss"""]["""value"""])
											# get train time from SageMaker job, this includes starting, preprocessing, stopping
											_SCREAMING_SNAKE_CASE		:						int			  =    (
											    Session().describe_training_job(estimator.latest_training_job.name).get("""TrainingTimeInSeconds"""      ,    9_9_9_9_9_9)
											)

											# assert kpis
											assert train_runtime <= self.results["train_runtime"]
											assert all(t >= self.results["""eval_accuracy"""] for t in eval_accuracy)
											assert all(t <= self.results["""eval_loss"""] for t in eval_loss)

											# dump tests result into json file to share in PR
											with open(f"""{estimator.latest_training_job.name}.json"""      ,    """w""") as outfile:
																		json.dump({"""train_time""": train_runtime, """eval_accuracy""": eval_accuracy, """eval_loss""": eval_loss}      ,    _A)


