

import unittest
from functools import partial

import torch

from accelerate import Accelerator, debug_launcher
from accelerate.state import AcceleratorState, GradientState
from accelerate.test_utils import require_cpu, require_huggingface_suite
from accelerate.utils import GradientAccumulationPlugin





def 	snake_case_      (__A     :						Tuple=2					,   __A     :						Tuple=True					,   __A     :						Optional[Any]=False     )			->	str:
   __lowerCAmelCase  :			Optional[int]					    =						Accelerator(step_scheduler_with_optimizer=__A					,   split_batches=__A     )
   __lowerCAmelCase  :			List[str]					    =						torch.nn.Linear(2					,   4     )
   __lowerCAmelCase  :			Dict					    =						torch.optim.AdamW(model.parameters()					,   lr=1.0     )
   __lowerCAmelCase  :			str					    =						torch.optim.lr_scheduler.OneCycleLR(__A					,   max_lr=0.01					,   steps_per_epoch=2					,   epochs=1     )
   __lowerCAmelCase    ,__lowerCAmelCase    ,__lowerCAmelCase  :			Optional[Any]					    =						accelerator.prepare(__A					,   __A					,   __A     )

   # Optimizer has stepped
   scheduler.step()
   if step_scheduler_with_optimizer or (num_processes == 1):
      assert (
          scheduler.scheduler.last_epoch == num_processes
      ), f'''Last Epoch ({scheduler.scheduler.last_epoch}) != Num Processes ({num_processes})'''
   else:
      assert (
          scheduler.scheduler.last_epoch != num_processes
      ), f'''Last Epoch ({scheduler.scheduler.last_epoch}) == Num Processes ({num_processes})'''





def 	snake_case_      (__A     :						List[Any]=2					,   __A     :						str=True					,   __A     :						List[Any]=False     )			->	Union[str, Any]:
   __lowerCAmelCase  :			int					    =						Accelerator(step_scheduler_with_optimizer=__A					,   split_batches=__A     )
   __lowerCAmelCase  :			str					    =						torch.nn.Linear(2					,   4     )
   __lowerCAmelCase  :			Optional[int]					    =						torch.optim.AdamW(model.parameters()					,   lr=1.0     )
   __lowerCAmelCase  :			Dict					    =						torch.optim.lr_scheduler.LambdaLR(__A					,   lr_lambda=lambda __A     : 1 - n / 1_0     )
   __lowerCAmelCase    ,__lowerCAmelCase    ,__lowerCAmelCase  :			Union[str, Any]					    =						accelerator.prepare(__A					,   __A					,   __A     )

   # Optimizer has stepped
   __lowerCAmelCase  :			List[str]					    =						False
   scheduler.step()
   __lowerCAmelCase  :			List[str]					    =						1 - (num_processes if (step_scheduler_with_optimizer and not split_batches) else 1) / 1_0
   assert (
       scheduler.get_last_lr()[0] == expected_lr
   ), f'''Wrong lr found at first step, expected {expected_lr}, got {scheduler.get_last_lr()[0]}'''

   # Optimizer has not stepped
   __lowerCAmelCase  :			int					    =						True
   scheduler.step()
   if not step_scheduler_with_optimizer:
      __lowerCAmelCase  :			Union[str, Any]					    =						1 - 2 / 1_0
   assert (
       scheduler.get_last_lr()[0] == expected_lr
   ), f'''Wrong lr found at second step, expected {expected_lr}, got {scheduler.get_last_lr()[0]}'''





def 	snake_case_      (__A     :						int = 2     )			->	Tuple:
   from transformers import get_linear_schedule_with_warmup

   __lowerCAmelCase  :			Any					    =						[1, 2, 4]
   for num_steps in steps:
      __lowerCAmelCase  :			Dict					    =						GradientAccumulationPlugin(num_steps=__A					,   adjust_scheduler=num_steps > 1     )
      __lowerCAmelCase  :			str					    =						Accelerator(gradient_accumulation_plugin=__A     )
      __lowerCAmelCase  :			List[str]					    =						torch.nn.Linear(2					,   4     )
      __lowerCAmelCase  :			List[str]					    =						torch.optim.AdamW(model.parameters()					,   lr=10.0     )
      __lowerCAmelCase  :			Tuple					    =						get_linear_schedule_with_warmup(optimizer=__A					,   num_warmup_steps=0					,   num_training_steps=2_0     )

      __lowerCAmelCase    ,__lowerCAmelCase    ,__lowerCAmelCase  :			Optional[int]					    =						accelerator.prepare(__A					,   __A					,   __A     )

      for i in range(1_0 * num_steps     ):
         with accelerator.accumulate(__A     ):
            optimizer.step()
            scheduler.step()

         if i == (1_0 * num_steps - 2):
            assert (
                scheduler.get_last_lr()[0] != 0
            ), f'''Wrong lr found at second-to-last step, expected non-zero, got {scheduler.get_last_lr()[0]}. num_steps: {num_steps}'''
      assert (
          scheduler.get_last_lr()[0] == 0
      ), f'''Wrong lr found at last step, expected 0, got {scheduler.get_last_lr()[0]}'''
      GradientState._reset_state()



@require_cpu
class SCREAMING_SNAKE_CASE	(  unittest.TestCase ):
  """simple docstring"""






  def        SCREAMING_SNAKE_CASE    (			self :    str    )							->	List[str]:






     """simple docstring"""







     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1    )       ,							num_processes=1    )
     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1       ,							split_batches=lowerCAmelCase    )       ,							num_processes=1    )







  def        SCREAMING_SNAKE_CASE    (			self :    Dict    )							->	List[str]:






     """simple docstring"""







     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1    )       ,							num_processes=1    )
     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1       ,							split_batches=lowerCAmelCase    )       ,							num_processes=1    )







  def        SCREAMING_SNAKE_CASE    (			self :    Dict    )							->	Union[str, Any]:






     """simple docstring"""







     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1       ,							step_scheduler_with_optimizer=lowerCAmelCase    )       ,							num_processes=1    )







  def        SCREAMING_SNAKE_CASE    (			self :    Dict    )							->	str:






     """simple docstring"""







     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1       ,							step_scheduler_with_optimizer=lowerCAmelCase    )       ,							num_processes=1    )







  def        SCREAMING_SNAKE_CASE    (			self :    List[Any]    )							->	List[str]:






     """simple docstring"""







     AcceleratorState._reset_state(lowerCAmelCase    )
     debug_launcher(lowerCAmelCase    )
     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1       ,							split_batches=lowerCAmelCase    )       ,							num_processes=1    )







  def        SCREAMING_SNAKE_CASE    (			self :    Optional[int]    )							->	int:






     """simple docstring"""







     AcceleratorState._reset_state(lowerCAmelCase    )
     debug_launcher(lowerCAmelCase    )
     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1       ,							split_batches=lowerCAmelCase    )       ,							num_processes=1    )







  def        SCREAMING_SNAKE_CASE    (			self :    Dict    )							->	Dict:






     """simple docstring"""







     AcceleratorState._reset_state(lowerCAmelCase    )
     debug_launcher(partial(lowerCAmelCase       ,							step_scheduler_with_optimizer=lowerCAmelCase    )    )







  def        SCREAMING_SNAKE_CASE    (			self :    Union[str, Any]    )							->	str:






     """simple docstring"""







     AcceleratorState._reset_state(lowerCAmelCase    )
     debug_launcher(partial(lowerCAmelCase       ,							step_scheduler_with_optimizer=lowerCAmelCase    )    )







  @require_huggingface_suite
  def        SCREAMING_SNAKE_CASE    (			self :    Union[str, Any]    )							->	Dict:






     """simple docstring"""







     AcceleratorState._reset_state(lowerCAmelCase    )
     debug_launcher(partial(lowerCAmelCase       ,							num_processes=1    )    )
     debug_launcher(lowerCAmelCase    )







"""simple docstring"""







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

from ...utils import logging


UpperCAmelCase:   Dict       			=							logging.get_logger(__name__)


def       __SCREAMING_SNAKE_CASE      (   __UpperCAmelCase					,							__UpperCAmelCase	):
		_lowercase							:      List[str]			=					nn.functional.normalize(__UpperCAmelCase	)
		_lowercase							:      int			=					nn.functional.normalize(__UpperCAmelCase	)
		return torch.mm(__UpperCAmelCase					,							normalized_text_embeds.t()	)





class    UpperCamelCase     (			snake_case    ):




	"""simple docstring"""
	SCREAMING_SNAKE_CASE_				:    List[str]		    =   CLIPConfig

	SCREAMING_SNAKE_CASE_				:    List[Any]		    =   ["CLIPEncoderLayer"]






	def __init__(			self							,UpperCAmelCase_					):
			super().__init__(UpperCAmelCase_					)

			_lowercase							:      Any			=					CLIPVisionModel(config.vision_config					)
			_lowercase							:      Any			=					nn.Linear(config.vision_config.hidden_size							,config.projection_dim							,bias=UpperCAmelCase_					)

			_lowercase							:      Dict			=					nn.Parameter(torch.ones(17							,config.projection_dim					)							,requires_grad=UpperCAmelCase_					)
			_lowercase							:      Tuple			=					nn.Parameter(torch.ones(3							,config.projection_dim					)							,requires_grad=UpperCAmelCase_					)

			_lowercase							:      Optional[int]			=					nn.Parameter(torch.ones(17					)							,requires_grad=UpperCAmelCase_					)
			_lowercase							:      List[Any]			=					nn.Parameter(torch.ones(3					)							,requires_grad=UpperCAmelCase_					)






	@torch.no_grad()
	def 					lowerCamelCase__ (			self							,UpperCAmelCase_							,UpperCAmelCase_					):
			_lowercase							:      List[str]			=					self.vision_model(UpperCAmelCase_					)[1]  # pooled_output
			_lowercase							:      str			=					self.visual_projection(UpperCAmelCase_					)

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

			_lowercase							:      List[str]			=					[]
			_lowercase							:      Optional[Any]			=					image_embeds.shape[0]
			for i in range(UpperCAmelCase_					):
					_lowercase							:      int			=					{"""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
					_lowercase							:      str			=					0.0

					for concept_idx in range(len(special_cos_dist[0]					)					):
							_lowercase							:      Optional[Any]			=					special_cos_dist[i][concept_idx]
							_lowercase							:      str			=					self.special_care_embeds_weights[concept_idx].item()
							_lowercase							:      Tuple			=					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]}					)
									_lowercase							:      Union[str, Any]			=					0.01

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

					result.append(UpperCAmelCase_					)

			_lowercase							:      List[str]			=					[len(res["""bad_concepts"""]					) > 0 for res in result]

			for idx, has_nsfw_concept in enumerate(UpperCAmelCase_					):
					if has_nsfw_concept:
							if torch.is_tensor(UpperCAmelCase_					) or torch.is_tensor(images[0]					):
									_lowercase							:      List[Any]			=					torch.zeros_like(images[idx]					)  # black image
							else:
									_lowercase							:      Any			=					np.zeros(images[idx].shape					)  # black image

			if any(UpperCAmelCase_					):
					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 					lowerCamelCase__ (			self							,UpperCAmelCase_							,UpperCAmelCase_					):
			_lowercase							:      List[Any]			=					self.vision_model(UpperCAmelCase_					)[1]  # pooled_output
			_lowercase							:      Optional[int]			=					self.visual_projection(UpperCAmelCase_					)

			_lowercase							:      Union[str, Any]			=					cosine_distance(UpperCAmelCase_							,self.special_care_embeds					)
			_lowercase							:      Tuple			=					cosine_distance(UpperCAmelCase_							,self.concept_embeds					)

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

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

			_lowercase							:      Optional[Any]			=					(cos_dist - self.concept_embeds_weights) + special_adjustment
			# concept_scores = concept_scores.round(decimals=3)
			_lowercase							:      Optional[int]			=					torch.any(concept_scores > 0							,dim=1					)

			_lowercase							:      Tuple			=					0.0  # black image

			return images, has_nsfw_concepts



