
import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


lowercase__	:Union[str, Any]										=					logging.get_logger(__name__)







class        lowercase	(      SCREAMING_SNAKE_CASE__	):
							lowercase_ : Dict       ='''linear'''
							lowercase_ : int       ='''cosine'''
							lowercase_ : List[Any]       ='''cosine_with_restarts'''
							lowercase_ : Tuple       ='''polynomial'''
							lowercase_ : Union[str, Any]       ='''constant'''
							lowercase_ : Union[str, Any]       ='''constant_with_warmup'''
							lowercase_ : List[Any]       ='''piecewise_constant'''



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__ = -1	):
					'''simple docstring'''
					return LambdaLR(lowerCAmelCase__				,    lambda lowerCAmelCase__	: 1				,    last_epoch=lowerCAmelCase__	)



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__ = -1	):
					'''simple docstring'''

					def lr_lambda(lowerCAmelCase__	):
										if current_step < num_warmup_steps:
															return float(lowerCAmelCase__	) / float(max(1.0				,    lowerCAmelCase__	)	)
										return 1.0

					return LambdaLR(lowerCAmelCase__				,    lowerCAmelCase__				,    last_epoch=lowerCAmelCase__	)



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__ = -1	):
					'''simple docstring'''
					lowercase					=		{}
					lowercase					=		step_rules.split(''','''	)
					for rule_str in rule_list[:-1]:
										lowercase			,		lowercase					=		rule_str.split(''':'''	)
										lowercase					=		int(lowerCAmelCase__	)
										lowercase					=		float(lowerCAmelCase__	)
										lowercase					=		value
					lowercase					=		float(rule_list[-1]	)

					def create_rules_function(lowerCAmelCase__				,    lowerCAmelCase__	):
										def rule_func(lowerCAmelCase__	) -> float:
															lowercase					=		sorted(rules_dict.keys()	)
															for i, sorted_step in enumerate(lowerCAmelCase__	):
																				if steps < sorted_step:
																									return rules_dict[sorted_steps[i]]
															return last_lr_multiple

										return rule_func

					lowercase					=		create_rules_function(lowerCAmelCase__				,    lowerCAmelCase__	)

					return LambdaLR(lowerCAmelCase__				,    lowerCAmelCase__				,    last_epoch=lowerCAmelCase__	)



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__=-1	):
					'''simple docstring'''

					def lr_lambda(lowerCAmelCase__	):
										if current_step < num_warmup_steps:
															return float(lowerCAmelCase__	) / float(max(1				,    lowerCAmelCase__	)	)
										return max(
										    0.0				,    float(num_training_steps - current_step	) / float(max(1				,    num_training_steps - num_warmup_steps	)	)	)

					return LambdaLR(lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__	)



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__ = 0.5				,    lowerCAmelCase__ = -1	):
					'''simple docstring'''

					def lr_lambda(lowerCAmelCase__	):
										if current_step < num_warmup_steps:
															return float(lowerCAmelCase__	) / float(max(1				,    lowerCAmelCase__	)	)
										lowercase					=		float(current_step - num_warmup_steps	) / float(max(1				,    num_training_steps - num_warmup_steps	)	)
										return max(0.0				,    0.5 * (1.0 + math.cos(math.pi * float(lowerCAmelCase__	) * 2.0 * progress	))	)

					return LambdaLR(lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__	)



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__ = 1				,    lowerCAmelCase__ = -1	):
					'''simple docstring'''

					def lr_lambda(lowerCAmelCase__	):
										if current_step < num_warmup_steps:
															return float(lowerCAmelCase__	) / float(max(1				,    lowerCAmelCase__	)	)
										lowercase					=		float(current_step - num_warmup_steps	) / float(max(1				,    num_training_steps - num_warmup_steps	)	)
										if progress >= 1.0:
															return 0.0
										return max(0.0				,    0.5 * (1.0 + math.cos(math.pi * ((float(lowerCAmelCase__	) * progress) % 1.0)	))	)

					return LambdaLR(lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__	)



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__=1E-7				,    lowerCAmelCase__=1.0				,    lowerCAmelCase__=-1	):
					'''simple docstring'''
					lowercase					=		optimizer.defaults['''lr''']
					if not (lr_init > lr_end):
										raise ValueError(f'lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})'	)

					def lr_lambda(lowerCAmelCase__	):
										if current_step < num_warmup_steps:
															return float(lowerCAmelCase__	) / float(max(1				,    lowerCAmelCase__	)	)
										elif current_step > num_training_steps:
															return lr_end / lr_init  # as LambdaLR multiplies by lr_init
										else:
															lowercase					=		lr_init - lr_end
															lowercase					=		num_training_steps - num_warmup_steps
															lowercase					=		1 - (current_step - num_warmup_steps) / decay_steps
															lowercase					=		lr_range * pct_remaining**power + lr_end
															return decay / lr_init  # as LambdaLR multiplies by lr_init

					return LambdaLR(lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__	)


lowercase__	:Dict										=					{
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__ = None				,    lowerCAmelCase__ = None				,    lowerCAmelCase__ = None				,    lowerCAmelCase__ = 1				,    lowerCAmelCase__ = 1.0				,    lowerCAmelCase__ = -1				,    ):
					'''simple docstring'''
					lowercase					=		SchedulerType(lowerCAmelCase__	)
					lowercase					=		TYPE_TO_SCHEDULER_FUNCTION[name]
					if name == SchedulerType.CONSTANT:
										return schedule_func(lowerCAmelCase__				,    last_epoch=lowerCAmelCase__	)

					if name == SchedulerType.PIECEWISE_CONSTANT:
										return schedule_func(lowerCAmelCase__				,    step_rules=lowerCAmelCase__				,    last_epoch=lowerCAmelCase__	)

					# All other schedulers require `num_warmup_steps`
					if num_warmup_steps is None:
										raise ValueError(f'{name} requires `num_warmup_steps`, please provide that argument.'	)

					if name == SchedulerType.CONSTANT_WITH_WARMUP:
										return schedule_func(lowerCAmelCase__				,    num_warmup_steps=lowerCAmelCase__				,    last_epoch=lowerCAmelCase__	)

					# All other schedulers require `num_training_steps`
					if num_training_steps is None:
										raise ValueError(f'{name} requires `num_training_steps`, please provide that argument.'	)

					if name == SchedulerType.COSINE_WITH_RESTARTS:
										return schedule_func(
										    lowerCAmelCase__				,    num_warmup_steps=lowerCAmelCase__				,    num_training_steps=lowerCAmelCase__				,    num_cycles=lowerCAmelCase__				,    last_epoch=lowerCAmelCase__				,    )

					if name == SchedulerType.POLYNOMIAL:
										return schedule_func(
										    lowerCAmelCase__				,    num_warmup_steps=lowerCAmelCase__				,    num_training_steps=lowerCAmelCase__				,    power=lowerCAmelCase__				,    last_epoch=lowerCAmelCase__				,    )

					return schedule_func(
					    lowerCAmelCase__				,    num_warmup_steps=lowerCAmelCase__				,    num_training_steps=lowerCAmelCase__				,    last_epoch=lowerCAmelCase__	)






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

from ...utils import logging


lowercase__	:Union[str, Any]										=					logging.get_logger(__name__)







class        lowercase	(      SCREAMING_SNAKE_CASE__	):
							lowercase_ : Union[str, Any]       =CLIPConfig

							lowercase_ : str       =['''CLIPEncoderLayer''']




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

												lowercase					=		CLIPVisionModelWithProjection(config.vision_config)

												lowercase					=		nn.Linear(config.vision_config.projection_dim     ,1)
												lowercase					=		nn.Linear(config.vision_config.projection_dim     ,1)





							@torch.no_grad()
							def 				A__   (       self     ,A__     ,A__     ,A__=0.5     ,A__=0.5):
												lowercase					=		self.vision_model(A__)[0]

												lowercase					=		self.p_head(A__)
												lowercase					=		nsfw_detected.flatten()
												lowercase					=		nsfw_detected > p_threshold
												lowercase					=		nsfw_detected.tolist()

												if any(A__):
																	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.''')

												for idx, nsfw_detected_ in enumerate(A__):
																	if nsfw_detected_:
																						lowercase					=		np.zeros(images[idx].shape)

												lowercase					=		self.w_head(A__)
												lowercase					=		watermark_detected.flatten()
												lowercase					=		watermark_detected > w_threshold
												lowercase					=		watermark_detected.tolist()

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

												for idx, watermark_detected_ in enumerate(A__):
																	if watermark_detected_:
																						lowercase					=		np.zeros(images[idx].shape)

												return images, nsfw_detected, watermark_detected




