

"""simple docstring"""


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

import pytest
from parameterized import parameterized, 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.p3.16xlarge""",
            """results""": {"""train_runtime""": 6_50, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """pytorch""",
            """script""": """run_ddp.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 6_00, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf_dist.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 6_00, """eval_accuracy""": 0.6, """eval_loss""": 0.7},
        },
    ]  )
class     snake_case	(				unittest.TestCase  ):

				def 							lowercase_     (	self							:				Optional[Any])->   Tuple:



						'''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=__lowerCAmelCase     ,  )
						assert hasattr(self     ,  "env")

				def 							lowercase_     (	self							:				Union[str, Any]     ,  UpperCamelCase__							:				str)->   Tuple:



						'''simple docstring'''


						__lowerCAmelCase:       Any          =			f"{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"
						# distributed data settings
						__lowerCAmelCase:       str          =			{"smdistributed": {"dataparallel": {"enabled": True}}} if self.script != "run_ddp.py" else None

						# creates estimator
						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=__lowerCAmelCase     ,  instance_count=__lowerCAmelCase     ,  instance_type=self.instance_type     ,  debugger_hook_config=__lowerCAmelCase     ,  hyperparameters={**self.env.distributed_hyperparameters, "model_name_or_path": self.model_name_or_path}     ,  metric_definitions=self.env.metric_definitions     ,  distribution=__lowerCAmelCase     ,  py_version="py36"     ,  )

				def 							lowercase_     (	self							:				Tuple     ,  UpperCamelCase__							:				Optional[Any])->   List[str]:



						'''simple docstring'''


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




				@parameterized.expand([(2,)])
				def 							lowercase_     (	self							:				List[str]     ,  UpperCamelCase__							:				List[Any])->   int:



						'''simple docstring'''


						__lowerCAmelCase:       str          =			self.create_estimator(__lowerCAmelCase)

						# run training
						estimator.fit()

						# result dataframe
						__lowerCAmelCase:       Optional[int]          =			TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()

						# extract kpis
						__lowerCAmelCase:       Optional[Any]          =			list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"])
						__lowerCAmelCase:       List[str]          =			list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"])
						# get train time from SageMaker job, this includes starting, preprocessing, stopping
						__lowerCAmelCase:       Any          =			(
						    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}     ,  __lowerCAmelCase)



'''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 				snake_case__							(			_A:					Union[str, Any]      ,   _A:					Tuple      ,   _A:					Any=1e-12 )    ->	str:

		'''simple docstring'''


		lowerCAmelCase   					=  jnp.divide(emb_a.T      ,   jnp.clip(jnp.linalg.norm(_A      ,   axis=1 )      ,   a_min=_A ) ).T
		lowerCAmelCase   					=  jnp.divide(emb_a.T      ,   jnp.clip(jnp.linalg.norm(_A      ,   axis=1 )      ,   a_min=_A ) ).T
		return jnp.matmul(_A      ,   norm_emb_a.T )






class 				a__(					nn.Module	):


			'''simple docstring'''






			UpperCAmelCase_			:     CLIPConfig
			UpperCAmelCase_			:     jnp.dtype      =       jnp.floataa



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  FlaxCLIPVisionModule(self.config.vision_config)
					lowerCAmelCase   					=  nn.Dense(self.config.projection_dim       ,    use_bias=__lowerCAmelCase       ,    dtype=self.dtype)

					lowerCAmelCase   					=  self.param("""concept_embeds"""       ,    jax.nn.initializers.ones       ,    (17, self.config.projection_dim))
					lowerCAmelCase   					=  self.param(
					    """special_care_embeds"""       ,    jax.nn.initializers.ones       ,    (3, self.config.projection_dim))

					lowerCAmelCase   					=  self.param("""concept_embeds_weights"""       ,    jax.nn.initializers.ones       ,    (17,))
					lowerCAmelCase   					=  self.param("""special_care_embeds_weights"""       ,    jax.nn.initializers.ones       ,    (3,))







			def __call__(   self       ,    __lowerCAmelCase):




					"""simple docstring"""


					lowerCAmelCase   					=  self.vision_model(__lowerCAmelCase)[1]
					lowerCAmelCase   					=  self.visual_projection(__lowerCAmelCase)

					lowerCAmelCase   					=  jax_cosine_distance(__lowerCAmelCase       ,    self.special_care_embeds)
					lowerCAmelCase   					=  jax_cosine_distance(__lowerCAmelCase       ,    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   					=  0.0

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

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

					return has_nsfw_concepts







class 				a__(					lowerCAmelCase__	):


			'''simple docstring'''






			UpperCAmelCase_			:     int      =       CLIPConfig
			UpperCAmelCase_			:     Any      =       '''clip_input'''
			UpperCAmelCase_			:     List[str]      =       FlaxStableDiffusionSafetyCheckerModule



			def __init__(   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = None       ,    __lowerCAmelCase = 0       ,    __lowerCAmelCase = jnp.floataa       ,    __lowerCAmelCase = True       ,    **__lowerCAmelCase       ,    ):




					"""simple docstring"""


					if input_shape is None:
							lowerCAmelCase   					=  (1, 224, 224, 3)
					lowerCAmelCase   					=  self.module_class(config=__lowerCAmelCase       ,    dtype=__lowerCAmelCase       ,    **__lowerCAmelCase)
					super().__init__(__lowerCAmelCase       ,    __lowerCAmelCase       ,    input_shape=__lowerCAmelCase       ,    seed=__lowerCAmelCase       ,    dtype=__lowerCAmelCase       ,    _do_init=_do_init)



			def 							a_       (   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase = None):




					"""simple docstring"""


					lowerCAmelCase   					=  jax.random.normal(__lowerCAmelCase       ,    __lowerCAmelCase)

					lowerCAmelCase					,     lowerCAmelCase   					=  jax.random.split(__lowerCAmelCase)
					lowerCAmelCase   					=  {"""params""": params_rng, """dropout""": dropout_rng}

					lowerCAmelCase   					=  self.module.init(__lowerCAmelCase       ,    __lowerCAmelCase)["""params"""]

					return random_params







			def __call__(   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase = None       ,    ):




					"""simple docstring"""


					lowerCAmelCase   					=  jnp.transpose(__lowerCAmelCase       ,    (0, 2, 3, 1))

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




