

'''simple docstring'''

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

from ...utils import logging


a       :      List[Any]  	=			logging.get_logger(__name__)





def 				lowercase (					__magic_name__			,   __magic_name__							):



					'''simple docstring'''







					UpperCAmelCase		:							Union[str, Any]        =       nn.functional.normalize(__magic_name__							)
					UpperCAmelCase		:							Union[str, Any]        =       nn.functional.normalize(__magic_name__							)
					return torch.mm(__magic_name__			,   normalized_text_embeds.t()							)







class   UpperCamelCase__   (       lowercase__       ):


		"""simple docstring"""

		SCREAMING_SNAKE_CASE__  :     List[str]       =							CLIPConfig

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







		def __init__(			self   ,      snake_case      ):


							'''simple docstring'''

							super().__init__(snake_case      )

							UpperCAmelCase		:							str        =       CLIPVisionModel(config.vision_config      )
							UpperCAmelCase		:							int        =       nn.Linear(config.vision_config.hidden_size   ,      config.projection_dim   ,      bias=snake_case      )

							UpperCAmelCase		:							List[str]        =       nn.Parameter(torch.ones(1_7   ,      config.projection_dim      )   ,      requires_grad=snake_case      )
							UpperCAmelCase		:							str        =       nn.Parameter(torch.ones(3   ,      config.projection_dim      )   ,      requires_grad=snake_case      )

							UpperCAmelCase		:							Union[str, Any]        =       nn.Parameter(torch.ones(1_7      )   ,      requires_grad=snake_case      )
							UpperCAmelCase		:							Optional[int]        =       nn.Parameter(torch.ones(3      )   ,      requires_grad=snake_case      )







		@torch.no_grad()
		def 				A_	(			self   ,      snake_case   ,      snake_case      ):


							'''simple docstring'''

							UpperCAmelCase		:							int        =       self.vision_model(snake_case      )[1]  # pooled_output
							UpperCAmelCase		:							Optional[Any]        =       self.visual_projection(snake_case      )

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

							UpperCAmelCase		:							Optional[int]        =       []
							UpperCAmelCase		:							Tuple        =       image_embeds.shape[0]
							for i in range(snake_case      ):
												UpperCAmelCase		:							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
												UpperCAmelCase		:							List[str]        =       0.0

												for concept_idx in range(len(special_cos_dist[0]      )      ):
																	UpperCAmelCase		:							int        =       special_cos_dist[i][concept_idx]
																	UpperCAmelCase		:							int        =       self.special_care_embeds_weights[concept_idx].item()
																	UpperCAmelCase		:							List[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]}      )
																						UpperCAmelCase		:							Any        =       0.01

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

												result.append(snake_case      )

							UpperCAmelCase		:							int        =       [len(res["bad_concepts"]      ) > 0 for res in result]

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

							if any(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 				A_	(			self   ,      snake_case   ,      snake_case      ):


							'''simple docstring'''

							UpperCAmelCase		:							Optional[Any]        =       self.vision_model(snake_case      )[1]  # pooled_output
							UpperCAmelCase		:							Tuple        =       self.visual_projection(snake_case      )

							UpperCAmelCase		:							List[Any]        =       cosine_distance(snake_case   ,      self.special_care_embeds      )
							UpperCAmelCase		:							Optional[Any]        =       cosine_distance(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
							UpperCAmelCase		:							Dict        =       0.0

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

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

							UpperCAmelCase		:							Optional[int]        =       0.0  # black image

							return images, has_nsfw_concepts



'''simple docstring'''

import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only

from utils import save_json





def 				lowercase (					__magic_name__							):



					'''simple docstring'''







					UpperCAmelCase		:							Any        =       filter(lambda __magic_name__							: p.requires_grad			,   model.parameters()							)
					UpperCAmelCase		:							Union[str, Any]        =       sum([np.prod(p.size()							) for p in model_parameters]							)
					return params


a       :      int  	=			logging.getLogger(__name__)







class   UpperCamelCase__   (       pl.Callback       ):


		"""simple docstring"""

		def 				A_	(			self   ,      snake_case   ,      snake_case      ):


							'''simple docstring'''

							UpperCAmelCase		:							Union[str, Any]        =       {f"lr_group_{i}": param["lr"] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups      )}
							pl_module.logger.log_metrics(snake_case      )







		@rank_zero_only
		def 				A_	(			self   ,      snake_case   ,      snake_case   ,      snake_case   ,      snake_case=True      ):


							'''simple docstring'''

							logger.info(f"***** {type_path} results at step {trainer.global_step:05d} *****"      )
							UpperCAmelCase		:							Optional[Any]        =       trainer.callback_metrics
							trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ["log", "progress_bar", "preds"]}      )
							# Log results
							UpperCAmelCase		:							List[str]        =       Path(pl_module.hparams.output_dir      )
							if type_path == "test":
												UpperCAmelCase		:							Dict        =       od / "test_results.txt"
												UpperCAmelCase		:							Dict        =       od / "test_generations.txt"
							else:
												# this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
												# If people want this it will be easy enough to add back.
												UpperCAmelCase		:							Union[str, Any]        =       od / f"{type_path}_results/{trainer.global_step:05d}.txt"
												UpperCAmelCase		:							int        =       od / f"{type_path}_generations/{trainer.global_step:05d}.txt"
												results_file.parent.mkdir(exist_ok=snake_case      )
												generations_file.parent.mkdir(exist_ok=snake_case      )
							with open(snake_case   ,      "a+"      ) as writer:
												for key in sorted(snake_case      ):
																	if key in ["log", "progress_bar", "preds"]:
																						continue
																	UpperCAmelCase		:							Optional[Any]        =       metrics[key]
																	if isinstance(snake_case   ,      torch.Tensor      ):
																						UpperCAmelCase		:							Union[str, Any]        =       val.item()
																	UpperCAmelCase		:							List[str]        =       f"{key}: {val:.6f}\n"
																	writer.write(snake_case      )

							if not save_generations:
												return

							if "preds" in metrics:
												UpperCAmelCase		:							List[str]        =       "\n".join(metrics["preds"]      )
												generations_file.open("w+"      ).write(snake_case      )







		@rank_zero_only
		def 				A_	(			self   ,      snake_case   ,      snake_case      ):


							'''simple docstring'''

							try:
												UpperCAmelCase		:							Tuple        =       pl_module.model.model.num_parameters()
							except AttributeError:
												UpperCAmelCase		:							int        =       pl_module.model.num_parameters()

							UpperCAmelCase		:							List[Any]        =       count_trainable_parameters(snake_case      )
							# mp stands for million parameters
							trainer.logger.log_metrics({"n_params": npars, "mp": npars / 1e6, "grad_mp": n_trainable_pars / 1e6}      )







		@rank_zero_only
		def 				A_	(			self   ,      snake_case   ,      snake_case      ):


							'''simple docstring'''

							save_json(pl_module.metrics   ,      pl_module.metrics_save_path      )
							return self._write_logs(snake_case   ,      snake_case   ,      "test"      )







		@rank_zero_only
		def 				A_	(			self   ,      snake_case   ,      snake_case      ):


							'''simple docstring'''

							save_json(pl_module.metrics   ,      pl_module.metrics_save_path      )
							# Uncommenting this will save val generations
							# return self._write_logs(trainer, pl_module, "valid")





def 				lowercase (					__magic_name__			,   __magic_name__			,   __magic_name__=1			,   __magic_name__=False							):



					'''simple docstring'''







					if metric == "rouge2":
										UpperCAmelCase		:							Dict        =       "{val_avg_rouge2:.4f}-{step_count}"
					elif metric == "bleu":
										UpperCAmelCase		:							int        =       "{val_avg_bleu:.4f}-{step_count}"
					elif metric == "loss":
										UpperCAmelCase		:							Tuple        =       "{val_avg_loss:.4f}-{step_count}"
					else:
										raise NotImplementedError(
										    F"seq2seq callbacks only support rouge2, bleu and loss, got {metric}, You can make your own by adding to"
										    " this function."							)

					UpperCAmelCase		:							int        =       ModelCheckpoint(
					    dirpath=__magic_name__			,   filename=__magic_name__			,   monitor=F"val_{metric}"			,   mode="min" if "loss" in metric else "max"			,   save_top_k=__magic_name__			,   )
					return checkpoint_callback





def 				lowercase (					__magic_name__			,   __magic_name__							):



					'''simple docstring'''







					return EarlyStopping(
					    monitor=F"val_{metric}"			,   mode="min" if "loss" in metric else "max"			,   patience=__magic_name__			,   verbose=__magic_name__			,   )
