



'''simple docstring'''







import json
import os
import unittest

from transformers.models.fsmt.tokenization_fsmt import VOCAB_FILES_NAMES, FSMTTokenizer
from transformers.testing_utils import slow
from transformers.utils import cached_property

from ...test_tokenization_common import TokenizerTesterMixin


# using a different tiny model than the one used for default params defined in init to ensure proper testing
SCREAMING_SNAKE_CASE__     : Tuple				=   """stas/tiny-wmt19-en-ru"""




class 				__lowerCAmelCase(							lowerCAmelCase__		,			unittest.TestCase      ):
			__snake_case  :		Any		=      FSMTTokenizer
			__snake_case  :		Any		=      False






			def 				_lowercase	(   self  :							Tuple	):




					"""simple docstring"""







					super().setUp()

					# Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      [
					    'l',
					    'o',
					    'w',
					    'e',
					    'r',
					    's',
					    't',
					    'i',
					    'd',
					    'n',
					    'w</w>',
					    'r</w>',
					    't</w>',
					    'lo',
					    'low',
					    'er</w>',
					    'low</w>',
					    'lowest</w>',
					    'newer</w>',
					    'wider</w>',
					    '<unk>',
					]
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      dict(zip(SCREAMING_SNAKE_CASE      ,       range(len(SCREAMING_SNAKE_CASE	)	)	)	)
					SCREAMING_SNAKE_CASE_						:Any										=      ['l o 123', 'lo w 1456', 'e r</w> 1789', '']

					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      ['en', 'ru']
					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      {
					    'langs': self.langs,
					    'src_vocab_size': 10,
					    'tgt_vocab_size': 20,
					}

					SCREAMING_SNAKE_CASE_						:Dict										=      os.path.join(self.tmpdirname      ,       VOCAB_FILES_NAMES['src_vocab_file']	)
					SCREAMING_SNAKE_CASE_						:Dict										=      os.path.join(self.tmpdirname      ,       VOCAB_FILES_NAMES['tgt_vocab_file']	)
					SCREAMING_SNAKE_CASE_						:List[str]										=      os.path.join(self.tmpdirname      ,       'tokenizer_config.json'	)
					SCREAMING_SNAKE_CASE_						:Tuple										=      os.path.join(self.tmpdirname      ,       VOCAB_FILES_NAMES['merges_file']	)
					with open(self.src_vocab_file      ,       'w'	) as fp:
							fp.write(json.dumps(SCREAMING_SNAKE_CASE	)	)
					with open(self.tgt_vocab_file      ,       'w'	) as fp:
							fp.write(json.dumps(SCREAMING_SNAKE_CASE	)	)
					with open(self.merges_file      ,       'w'	) as fp:
							fp.write('\n'.join(SCREAMING_SNAKE_CASE	)	)
					with open(SCREAMING_SNAKE_CASE      ,       'w'	) as fp:
							fp.write(json.dumps(SCREAMING_SNAKE_CASE	)	)






			@cached_property
			def 				_lowercase	(   self  :							List[Any]	):




					"""simple docstring"""







					return FSMTTokenizer.from_pretrained('facebook/wmt19-ru-en'	)






			@cached_property
			def 				_lowercase	(   self  :							str	):




					"""simple docstring"""







					return FSMTTokenizer.from_pretrained('facebook/wmt19-en-ru'	)






			def 				_lowercase	(   self  :							Optional[int]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:int										=      FSMTTokenizer.from_pretrained(SCREAMING_SNAKE_CASE	)
					self.assertListEqual([tokenizer.src_lang, tokenizer.tgt_lang]      ,       ['en', 'ru']	)
					self.assertEqual(tokenizer.src_vocab_size      ,       21	)
					self.assertEqual(tokenizer.tgt_vocab_size      ,       21	)






			def 				_lowercase	(   self  :							Dict	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[Any]										=      FSMTTokenizer(self.langs      ,       self.src_vocab_file      ,       self.tgt_vocab_file      ,       self.merges_file	)

					SCREAMING_SNAKE_CASE_						:Optional[int]										=      'lower'
					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      ['low', 'er</w>']
					SCREAMING_SNAKE_CASE_						:List[str]										=      tokenizer.tokenize(SCREAMING_SNAKE_CASE	)
					self.assertListEqual(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)

					SCREAMING_SNAKE_CASE_						:List[str]										=      tokens + ['<unk>']
					SCREAMING_SNAKE_CASE_						:str										=      [14, 15, 20]
					self.assertListEqual(tokenizer.convert_tokens_to_ids(SCREAMING_SNAKE_CASE	)      ,       SCREAMING_SNAKE_CASE	)






			@slow
			def 				_lowercase	(   self  :							Optional[Any]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Dict										=      self.tokenizer_ru_en

					SCREAMING_SNAKE_CASE_						:List[str]										=      tokenizer.encode('sequence builders'      ,       add_special_tokens=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:Tuple										=      tokenizer.encode('multi-sequence build'      ,       add_special_tokens=SCREAMING_SNAKE_CASE	)

					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      tokenizer.build_inputs_with_special_tokens(SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:int										=      tokenizer.build_inputs_with_special_tokens(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)

					assert encoded_sentence == text + [2]
					assert encoded_pair == text + [2] + text_a + [2]






			@slow
			def 				_lowercase	(   self  :							Dict	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:str										=      self.tokenizer_en_ru
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      self.tokenizer_ru_en

					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      [
					    [
					        'Here\'s a little song I wrote. Don\'t worry, be happy.',
					        [2_470, 39, 11, 2_349, 7_222, 70, 5_979, 7, 8_450, 1_050, 13_160, 5, 26, 6_445, 7, 2],
					    ],
					    ['This is it. No more. I\'m done!', [132, 21, 37, 7, 1_434, 86, 7, 70, 6_476, 1_305, 427, 2]],
					]

					# if data needs to be recreated or added, run:
					# import torch
					# model = torch.hub.load("pytorch/fairseq", "transformer.wmt19.en-ru", checkpoint_file="model4.pt", tokenizer="moses", bpe="fastbpe")
					# for src_text, _ in targets: print(f"""[\n"{src_text}",\n {model.encode(src_text).tolist()}\n],""")

					for src_text, tgt_input_ids in targets:
							SCREAMING_SNAKE_CASE_						:List[Any]										=      tokenizer_enc.encode(SCREAMING_SNAKE_CASE      ,       return_tensors=SCREAMING_SNAKE_CASE	)
							self.assertListEqual(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)

							# and decode backward, using the reversed languages model
							SCREAMING_SNAKE_CASE_						:int										=      tokenizer_dec.decode(SCREAMING_SNAKE_CASE      ,       skip_special_tokens=SCREAMING_SNAKE_CASE	)
							self.assertEqual(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)






			@slow
			def 				_lowercase	(   self  :							str	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[Any]										=      FSMTTokenizer.from_pretrained('facebook/wmt19-ru-en'      ,       do_lower_case=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:Dict										=      tokenizer.tokenize('USA is United States of America'	)
					SCREAMING_SNAKE_CASE_						:Dict										=      ['us', 'a</w>', 'is</w>', 'un', 'i', 'ted</w>', 'st', 'ates</w>', 'of</w>', 'am', 'er', 'ica</w>']
					self.assertListEqual(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)






			@unittest.skip('FSMTConfig.__init__  requires non-optional args'	)
			def 				_lowercase	(   self  :							Optional[Any]	):




					"""simple docstring"""







					pass






			@unittest.skip('FSMTConfig.__init__  requires non-optional args'	)
			def 				_lowercase	(   self  :							List[Any]	):




					"""simple docstring"""







					pass





'''simple docstring'''







import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


SCREAMING_SNAKE_CASE__     : Optional[int]				=   logging.get_logger(__name__)







def  SCREAMING_SNAKE_CASE__ (				SCREAMING_SNAKE_CASE ,		SCREAMING_SNAKE_CASE     ):
		SCREAMING_SNAKE_CASE_						:List[Any]										=      nn.functional.normalize(SCREAMING_SNAKE_CASE     )
		SCREAMING_SNAKE_CASE_						:str										=      nn.functional.normalize(SCREAMING_SNAKE_CASE     )
		return torch.mm(SCREAMING_SNAKE_CASE ,		normalized_text_embeds.t()     )





class 				__lowerCAmelCase(							lowerCAmelCase__      ):
			__snake_case  :		int		=      CLIPConfig

			__snake_case  :		Optional[Any]		=      ['CLIPEncoderLayer']






			def __init__(   self  :							int      ,       SCREAMING_SNAKE_CASE  :							CLIPConfig	):




					"""simple docstring"""







					super().__init__(SCREAMING_SNAKE_CASE	)

					SCREAMING_SNAKE_CASE_						:List[Any]										=      CLIPVisionModel(config.vision_config	)
					SCREAMING_SNAKE_CASE_						:Dict										=      nn.Linear(config.vision_config.hidden_size      ,       config.projection_dim      ,       bias=SCREAMING_SNAKE_CASE	)

					SCREAMING_SNAKE_CASE_						:List[Any]										=      nn.Parameter(torch.ones(17      ,       config.projection_dim	)      ,       requires_grad=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      nn.Parameter(torch.ones(3      ,       config.projection_dim	)      ,       requires_grad=SCREAMING_SNAKE_CASE	)

					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      nn.Parameter(torch.ones(17	)      ,       requires_grad=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:str										=      nn.Parameter(torch.ones(3	)      ,       requires_grad=SCREAMING_SNAKE_CASE	)






			@torch.no_grad()
			def 				_lowercase	(   self  :							Tuple      ,       SCREAMING_SNAKE_CASE  :							Tuple      ,       SCREAMING_SNAKE_CASE  :							Tuple	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Optional[int]										=      self.vision_model(SCREAMING_SNAKE_CASE	)[1]  # pooled_output
					SCREAMING_SNAKE_CASE_						:Dict										=      self.visual_projection(SCREAMING_SNAKE_CASE	)

					# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      cosine_distance(SCREAMING_SNAKE_CASE      ,       self.special_care_embeds	).cpu().float().numpy()
					SCREAMING_SNAKE_CASE_						:List[Any]										=      cosine_distance(SCREAMING_SNAKE_CASE      ,       self.concept_embeds	).cpu().float().numpy()

					SCREAMING_SNAKE_CASE_						:str										=      []
					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      image_embeds.shape[0]
					for i in range(SCREAMING_SNAKE_CASE	):
							SCREAMING_SNAKE_CASE_						:List[Any]										=      {'special_scores': {}, 'special_care': [], 'concept_scores': {}, 'bad_concepts': []}

							# increase this value to create a stronger `nfsw` filter
							# at the cost of increasing the possibility of filtering benign images
							SCREAMING_SNAKE_CASE_						:Optional[int]										=      0.0

							for concept_idx in range(len(special_cos_dist[0]	)	):
									SCREAMING_SNAKE_CASE_						:Dict										=      special_cos_dist[i][concept_idx]
									SCREAMING_SNAKE_CASE_						:Tuple										=      self.special_care_embeds_weights[concept_idx].item()
									SCREAMING_SNAKE_CASE_						:Any										=      round(concept_cos - concept_threshold + adjustment      ,       3	)
									if result_img["special_scores"][concept_idx] > 0:
											result_img["special_care"].append({concept_idx, result_img['special_scores'][concept_idx]}	)
											SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      0.01

							for concept_idx in range(len(cos_dist[0]	)	):
									SCREAMING_SNAKE_CASE_						:int										=      cos_dist[i][concept_idx]
									SCREAMING_SNAKE_CASE_						:Tuple										=      self.concept_embeds_weights[concept_idx].item()
									SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      round(concept_cos - concept_threshold + adjustment      ,       3	)
									if result_img["concept_scores"][concept_idx] > 0:
											result_img["bad_concepts"].append(SCREAMING_SNAKE_CASE	)

							result.append(SCREAMING_SNAKE_CASE	)

					SCREAMING_SNAKE_CASE_						:Any										=      [len(res['bad_concepts']	) > 0 for res in result]

					for idx, has_nsfw_concept in enumerate(SCREAMING_SNAKE_CASE	):
							if has_nsfw_concept:
									if torch.is_tensor(SCREAMING_SNAKE_CASE	) or torch.is_tensor(images[0]	):
											SCREAMING_SNAKE_CASE_						:Tuple										=      torch.zeros_like(images[idx]	)  # black image
									else:
											SCREAMING_SNAKE_CASE_						:str										=      np.zeros(images[idx].shape	)  # black image

					if any(SCREAMING_SNAKE_CASE	):
							logger.warning(
							    'Potential NSFW content was detected in one or more images. A black image will be returned instead.'
							    ' Try again with a different prompt and/or seed.'	)

					return images, has_nsfw_concepts






			@torch.no_grad()
			def 				_lowercase	(   self  :							str      ,       SCREAMING_SNAKE_CASE  :							torch.FloatTensor      ,       SCREAMING_SNAKE_CASE  :							torch.FloatTensor	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Tuple										=      self.vision_model(SCREAMING_SNAKE_CASE	)[1]  # pooled_output
					SCREAMING_SNAKE_CASE_						:Tuple										=      self.visual_projection(SCREAMING_SNAKE_CASE	)

					SCREAMING_SNAKE_CASE_						:List[str]										=      cosine_distance(SCREAMING_SNAKE_CASE      ,       self.special_care_embeds	)
					SCREAMING_SNAKE_CASE_						:int										=      cosine_distance(SCREAMING_SNAKE_CASE      ,       self.concept_embeds	)

					# increase this value to create a stronger `nsfw` filter
					# at the cost of increasing the possibility of filtering benign images
					SCREAMING_SNAKE_CASE_						:Dict										=      0.0

					SCREAMING_SNAKE_CASE_						:List[str]										=      special_cos_dist - self.special_care_embeds_weights + adjustment
					# special_scores = special_scores.round(decimals=3)
					SCREAMING_SNAKE_CASE_						:Dict										=      torch.any(special_scores > 0      ,       dim=1	)
					SCREAMING_SNAKE_CASE_						:Dict										=      special_care * 0.01
					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      special_adjustment.unsqueeze(1	).expand(-1      ,       cos_dist.shape[1]	)

					SCREAMING_SNAKE_CASE_						:str										=      (cos_dist - self.concept_embeds_weights) + special_adjustment
					# concept_scores = concept_scores.round(decimals=3)
					SCREAMING_SNAKE_CASE_						:str										=      torch.any(concept_scores > 0      ,       dim=1	)

					SCREAMING_SNAKE_CASE_						:Optional[int]										=      0.0  # black image

					return images, has_nsfw_concepts
