





"""simple docstring"""




import logging
import os
from dataclasses import dataclass
from typing import List, Optional, Union

import tqdm
from filelock import FileLock

from transformers import (
    BartTokenizer,
    BartTokenizerFast,
    DataProcessor,
    PreTrainedTokenizer,
    RobertaTokenizer,
    RobertaTokenizerFast,
    XLMRobertaTokenizer,
    is_tf_available,
    is_torch_available,
)


a		:					Optional[Any]	      =     logging.getLogger(__name__)


@dataclass(frozen=lowercase_		)
class 			__UpperCamelCase							:
							lowerCamelCase  :   str
							lowerCamelCase  :   str
							lowerCamelCase  :   Optional[str]    =None
							lowerCamelCase  :   Optional[str]    =None
							lowerCamelCase  :   Optional[str]    =None


@dataclass(frozen=lowercase_		)
class 			__UpperCamelCase							:
							lowerCamelCase  :   List[int]
							lowerCamelCase  :   Optional[List[int]]    =None
							lowerCamelCase  :   Optional[List[int]]    =None
							lowerCamelCase  :   Optional[Union[int, float]]    =None
							lowerCamelCase  :   Optional[int]    =None


if is_torch_available():
		import torch
		from torch.utils.data import Dataset


		class 			__UpperCamelCase							(		lowercase_		):
									lowerCamelCase  :   List[InputFeatures]






									def __init__(  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ = None ,   lowerCAmelCase__=False ,   lowerCAmelCase__ = False ,   )      ->					int:
										a :						Any      =							hans_processors[task]()

										a :						Tuple      =							os.path.join(
										    lowerCAmelCase__ ,   "cached_{}_{}_{}_{}".format(
										        "dev" if evaluate else "train" ,   tokenizer.__class__.__name__ ,   str(lowerCAmelCase__ ) ,   lowerCAmelCase__ ,   ) ,   )
										a :						List[str]      =							processor.get_labels()
										if tokenizer.__class__ in (
										    RobertaTokenizer,
										    RobertaTokenizerFast,
										    XLMRobertaTokenizer,
										    BartTokenizer,
										    BartTokenizerFast,
										):
											# HACK(label indices are swapped in RoBERTa pretrained model)
											a,			a :						int      =							label_list[2], label_list[1]
										a :						int      =							label_list

										# Make sure only the first process in distributed training processes the dataset,
										# and the others will use the cache.
										a :						Dict      =							cached_features_file + ".lock"
										with FileLock(lowerCAmelCase__ ):
											if os.path.exists(lowerCAmelCase__ ) and not overwrite_cache:
												logger.info(f"""Loading features from cached file {cached_features_file}""" )
												a :						Dict      =							torch.load(lowerCAmelCase__ )
											else:
												logger.info(f"""Creating features from dataset file at {data_dir}""" )

												a :						Union[str, Any]      =							(
												    processor.get_dev_examples(lowerCAmelCase__ ) if evaluate else processor.get_train_examples(lowerCAmelCase__ )
												)

												logger.info("Training examples: %s" ,   len(lowerCAmelCase__ ) )
												a :						Optional[int]      =							hans_convert_examples_to_features(lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )
												logger.info("Saving features into cached file %s" ,   lowerCAmelCase__ )
												torch.save(self.features ,   lowerCAmelCase__ )






									def __len__(  self )      ->					Tuple:
										return len(self.features )






									def __getitem__(  self ,   lowerCAmelCase__ )      ->					Any:
										return self.features[i]






									def        __a  (  self )      ->					Dict:
										return self.label_list


if is_tf_available():
		import tensorflow as tf


		class 			__UpperCamelCase							:
									lowerCamelCase  :   List[InputFeatures]






									def __init__(  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ = 128 ,   lowerCAmelCase__=False ,   lowerCAmelCase__ = False ,   )      ->					Any:
										a :						str      =							hans_processors[task]()
										a :						List[Any]      =							processor.get_labels()
										if tokenizer.__class__ in (
										    RobertaTokenizer,
										    RobertaTokenizerFast,
										    XLMRobertaTokenizer,
										    BartTokenizer,
										    BartTokenizerFast,
										):
											# HACK(label indices are swapped in RoBERTa pretrained model)
											a,			a :						List[str]      =							label_list[2], label_list[1]
										a :						List[Any]      =							label_list

										a :						List[str]      =							processor.get_dev_examples(lowerCAmelCase__ ) if evaluate else processor.get_train_examples(lowerCAmelCase__ )
										a :						Optional[int]      =							hans_convert_examples_to_features(lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )

										def gen():
											for ex_index, ex in tqdm.tqdm(enumerate(self.features ) ,   desc="convert examples to features" ):
												if ex_index % 1_0000 == 0:
													logger.info("Writing example %d of %d" % (ex_index, len(lowerCAmelCase__ )) )

												yield (
												    {
												        "example_id": 0,
												        "input_ids": ex.input_ids,
												        "attention_mask": ex.attention_mask,
												        "token_type_ids": ex.token_type_ids,
												    },
												    ex.label,
												)

										a :						Optional[Any]      =							tf.data.Dataset.from_generator(
										    lowerCAmelCase__ ,   (
										        {
										            "example_id": tf.intaa,
										            "input_ids": tf.intaa,
										            "attention_mask": tf.intaa,
										            "token_type_ids": tf.intaa,
										        },
										        tf.intaa,
										    ) ,   (
										        {
										            "example_id": tf.TensorShape([] ),
										            "input_ids": tf.TensorShape([None, None] ),
										            "attention_mask": tf.TensorShape([None, None] ),
										            "token_type_ids": tf.TensorShape([None, None] ),
										        },
										        tf.TensorShape([] ),
										    ) ,   )






									def        __a  (  self )      ->					List[Any]:
										return self.dataset






									def __len__(  self )      ->					Tuple:
										return len(self.features )






									def __getitem__(  self ,   lowerCAmelCase__ )      ->					List[str]:
										return self.features[i]






									def        __a  (  self )      ->					List[Any]:
										return self.label_list


class 			__UpperCamelCase							(		lowercase_		):






							def        __a  (  self ,   lowerCAmelCase__ )      ->					Optional[int]:
								return self._create_examples(self._read_tsv(os.path.join(lowerCAmelCase__ ,   "heuristics_train_set.txt" ) ) ,   "train" )






							def        __a  (  self ,   lowerCAmelCase__ )      ->					List[str]:
								return self._create_examples(self._read_tsv(os.path.join(lowerCAmelCase__ ,   "heuristics_evaluation_set.txt" ) ) ,   "dev" )






							def        __a  (  self )      ->					List[str]:
								return ["contradiction", "entailment", "neutral"]






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ )      ->					Optional[Any]:
								a :						Dict      =							[]
								for i, line in enumerate(lowerCAmelCase__ ):
									if i == 0:
										continue
									a :						Union[str, Any]      =							"%s-%s" % (set_type, line[0])
									a :						List[str]      =							line[5]
									a :						str      =							line[6]
									a :						Optional[int]      =							line[7][2:] if line[7].startswith("ex" ) else line[7]
									a :						Optional[int]      =							line[0]
									examples.append(InputExample(guid=lowerCAmelCase__ ,   text_a=lowerCAmelCase__ ,   text_b=lowerCAmelCase__ ,   label=lowerCAmelCase__ ,   pairID=lowerCAmelCase__ ) )
								return examples






def  _SCREAMING_SNAKE_CASE						(      _lowercase		:       List[str]       ,			_lowercase		:       Any       ,			_lowercase		:       Any       ,			_lowercase		:       Any       ,			)					->List[Any]:






	'''simple docstring'''






	a :						Optional[Any]      =							{label: i for i, label in enumerate(_lowercase   )}

	a :						Union[str, Any]      =							[]
	for ex_index, example in tqdm.tqdm(enumerate(_lowercase   )       ,			desc="convert examples to features"   ):
		if ex_index % 1_0000 == 0:
			logger.info("Writing example %d" % (ex_index)   )

		a :						Optional[Any]      =							tokenizer(
		    example.text_a       ,			example.text_b       ,			add_special_tokens=_lowercase       ,			max_length=_lowercase       ,			padding="max_length"       ,			truncation=_lowercase       ,			return_overflowing_tokens=_lowercase       ,			)

		a :						Optional[Any]      =							label_map[example.label] if example.label in label_map else 0

		a :						Dict      =							int(example.pairID   )

		features.append(InputFeatures(**_lowercase       ,			label=_lowercase       ,			pairID=_lowercase   )   )

	for i, example in enumerate(examples[:5]   ):
		logger.info("*** Example ***"   )
		logger.info(F"""guid: {example}"""   )
		logger.info(F"""features: {features[i]}"""   )

	return features


a		:					Dict	      =     {
    'hans': 3,
}







a		:					str	      =     {
    'hans': HansProcessor,
}






'''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 _SCREAMING_SNAKE_CASE (A    ,					A    ,					A=1E-12			)			->			str:






       """simple docstring"""







       lowercase__      =					jnp.divide(emb_a.T    ,					jnp.clip(jnp.linalg.norm(A    ,					axis=1			)    ,					a_min=A			)			).T
       lowercase__      =					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     __lowerCAmelCase		(nn.Module	):





      '''simple docstring'''






      lowerCAmelCase__   :	CLIPConfig
      lowerCAmelCase__   :	jnp.dtype       =				jnp.floataa



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

             '''simple docstring'''







             lowercase__      =					FlaxCLIPVisionModule(self.config.vision_config					)
             lowercase__      =					nn.Dense(self.config.projection_dim  ,		use_bias=UpperCamelCase  ,		dtype=self.dtype					)

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

             lowercase__      =					self.param('''concept_embeds_weights'''  ,		jax.nn.initializers.ones  ,		(17,)					)
             lowercase__      =					self.param('''special_care_embeds_weights'''  ,		jax.nn.initializers.ones  ,		(3,)					)






      def __call__(self       :       Union[str, Any]  ,		UpperCamelCase       :       Tuple					):

             '''simple docstring'''







             lowercase__      =					self.vision_model(UpperCamelCase					)[1]
             lowercase__      =					self.visual_projection(UpperCamelCase					)

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

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

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

             return has_nsfw_concepts






class     __lowerCAmelCase		(lowercase_	):





      '''simple docstring'''






      lowerCAmelCase__   :	Union[str, Any]       =				CLIPConfig
      lowerCAmelCase__   :	Optional[int]       =				"""clip_input"""
      lowerCAmelCase__   :	Any       =				FlaxStableDiffusionSafetyCheckerModule



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

             '''simple docstring'''







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



      def  UpperCamelCase__						(self       :       Any  ,		UpperCamelCase       :       jax.random.KeyArray  ,		UpperCamelCase       :       Tuple  ,		UpperCamelCase       :       FrozenDict = None					):

             '''simple docstring'''







             lowercase__      =					jax.random.normal(UpperCamelCase  ,		UpperCamelCase					)

             lowercase__     ,lowercase__      =					jax.random.split(UpperCamelCase					)
             lowercase__      =					{'''params''': params_rng, '''dropout''': dropout_rng}

             lowercase__      =					self.module.init(UpperCamelCase  ,		UpperCamelCase					)['''params''']

             return random_params






      def __call__(self       :       Optional[Any]  ,		UpperCamelCase       :       Tuple  ,		UpperCamelCase       :       dict = None  ,		):

             '''simple docstring'''







             lowercase__      =					jnp.transpose(UpperCamelCase  ,		(0, 2, 3, 1)					)

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




