



'''simple docstring'''







import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


SCREAMING_SNAKE_CASE__     : List[str]				=   namedtuple("""ExampleNamedTuple""", """a b c""")




class 				__lowerCAmelCase(							unittest.TestCase      ):






			def 				_lowercase	(   self  :							Optional[Any]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.randn(5      ,       2	)
					SCREAMING_SNAKE_CASE_						:int										=      torch.device('cuda'	) if torch.cuda.is_available() else torch.device('cpu'	)

					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      send_to_device(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta.cpu()      ,       SCREAMING_SNAKE_CASE	)	)

					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      send_to_device((tensor, [tensor, tensor], 1)      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta[0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta[1]      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta[1][0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta[1][1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta[2]      ,       1	)

					SCREAMING_SNAKE_CASE_						:Tuple										=      send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['a'].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta['b']      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['b'][0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta['b'][1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta['c']      ,       1	)

					SCREAMING_SNAKE_CASE_						:List[Any]										=      send_to_device(ExampleNamedTuple(a=SCREAMING_SNAKE_CASE      ,       b=[tensor, tensor]      ,       c=1	)      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta.a.cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta.b      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta.b[0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta.b[1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta.c      ,       1	)

					SCREAMING_SNAKE_CASE_						:Tuple										=      send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}	)      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['a'].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta['b']      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['b'][0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta['b'][1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta['c']      ,       1	)






			def 				_lowercase	(   self  :							List[str]	):




					"""simple docstring"""







					with self.assertRaises(SCREAMING_SNAKE_CASE	) as cm:
							SCREAMING_SNAKE_CASE_						:Tuple										=      recursively_apply(torch.tensor      ,       (torch.tensor(1	), 1)      ,       error_on_other_type=SCREAMING_SNAKE_CASE	)
					self.assertEqual(
					    str(cm.exception	)      ,       'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'      ,       )






			def 				_lowercase	(   self  :							str	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[str]										=      torch.tensor([1, 2, 3, 4, 5]	)
					self.assertEqual(listify(SCREAMING_SNAKE_CASE	)      ,       [1, 2, 3, 4, 5]	)

					SCREAMING_SNAKE_CASE_						:Any										=      torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]	)
					self.assertEqual(listify(SCREAMING_SNAKE_CASE	)      ,       [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]	)

					SCREAMING_SNAKE_CASE_						:List[Any]										=      torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]	)
					self.assertEqual(
					    listify(SCREAMING_SNAKE_CASE	)      ,       [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]	)






			def 				_lowercase	(   self  :							Tuple	):




					"""simple docstring"""







					with patch_environment(aa=1      ,       BB=2	):
							self.assertEqual(os.environ.get('AA'	)      ,       '1'	)
							self.assertEqual(os.environ.get('BB'	)      ,       '2'	)

					self.assertNotIn('AA'      ,       os.environ	)
					self.assertNotIn('BB'      ,       os.environ	)






			def 				_lowercase	(   self  :							Optional[int]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:str										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:Tuple										=      model.forward
					SCREAMING_SNAKE_CASE_						:Any										=      convert_outputs_to_fpaa(model.forward	)
					SCREAMING_SNAKE_CASE_						:Tuple										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE      ,       keep_fpaa_wrapper=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:List[str]										=      pickle.dumps(SCREAMING_SNAKE_CASE	)






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




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[str]										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:Tuple										=      model.forward
					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.cuda.amp.autocast(dtype=torch.floataa	)(model.forward	)
					SCREAMING_SNAKE_CASE_						:Any										=      convert_outputs_to_fpaa(model.forward	)
					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE      ,       keep_fpaa_wrapper=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:List[Any]										=      pickle.dumps(SCREAMING_SNAKE_CASE	)






			@require_cuda
			@require_torch_min_version(version='2.0'	)
			def 				_lowercase	(   self  :							Dict	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Dict										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:Tuple										=      model.forward
					SCREAMING_SNAKE_CASE_						:int										=      torch.cuda.amp.autocast(dtype=torch.floataa	)(model.forward	)
					SCREAMING_SNAKE_CASE_						:int										=      convert_outputs_to_fpaa(model.forward	)
					SCREAMING_SNAKE_CASE_						:List[Any]										=      torch.compile(model.forward      ,       backend='inductor'	)
					SCREAMING_SNAKE_CASE_						:str										=      torch.randn(4      ,       10	).cuda()
					SCREAMING_SNAKE_CASE_						:int										=      model(SCREAMING_SNAKE_CASE	)






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




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[str]										=      RegressionModel()
					# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      torch.nn.parallel.DataParallel(SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:int										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE	)

					self.assertEqual(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)






			@require_torch_min_version(version='2.0'	)
			def 				_lowercase	(   self  :							Optional[int]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[Any]										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:int										=      torch.compile(SCREAMING_SNAKE_CASE	)

					# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.nn.parallel.DataParallel(SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.compile(SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE	)

					self.assertEqual(compiled_model._orig_mod      ,       compiled_model_unwrapped._orig_mod	)






			def 				_lowercase	(   self  :							Optional[Any]	):




					"""simple docstring"""







					self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3]	)]	)      ,       torch.device('cpu'	)	)
					self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3]	)}	)      ,       torch.device('cpu'	)	)
					self.assertIsNone(find_device([1, 'a']	)	)





'''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
