


import json
import os
from collections.abc import Mapping
from typing import Dict, List, Optional, Union

import numpy as np
import torch

from .imports import is_safetensors_available
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :str      ,						lowercase__  :Dict      ,						lowercase__  :Optional[int]      ,						lowercase__  :Dict=None				)       ->  Union[str, Any]:
						__magic_name__				:							Optional[int]				=      None
						# Check the string instead of the dtype to be compatible with versions of PyTorch that don't have bfloat16.
						if str(weight.dtype				) == "torch.bfloat16":
												# Need to reinterpret the underlined data as int16 since NumPy does not handle bfloat16s.
												__magic_name__				:							Dict				=      weight.view(torch.intaa				)
												__magic_name__				:							Tuple				=      'bfloat16'
						__magic_name__				:							List[Any]				=      weight.cpu().numpy()
						__magic_name__				:							int				=      os.path.join(lowercase__      ,						f"{weight_name}.dat"				)
						if index is not None:
												if dtype is None:
																		__magic_name__				:							Optional[int]				=      str(array.dtype				)
												__magic_name__				:							Optional[int]				=      {'dtype': dtype, 'shape': list(array.shape				)}
						if array.ndim == 0:
												__magic_name__				:							List[Any]				=      array[None]
						__magic_name__				:							Optional[int]				=      np.memmap(lowercase__      ,						dtype=array.dtype      ,						mode='w+'      ,						shape=array.shape				)
						__magic_name__				:							int				=      array[:]
						file_array.flush()
						return index
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Optional[int]      ,						lowercase__  :Dict				)       ->  List[str]:
						__magic_name__				:							Optional[Any]				=      tuple(weight_info['shape']				)
						if shape == ():
												# NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
												__magic_name__				:							str				=      (1,)

						__magic_name__				:							str				=      weight_info['dtype']
						if dtype == "bfloat16":
												# NumPy does not support bfloat16 so this was saved as a int16
												__magic_name__				:							str				=      'int16'

						__magic_name__				:							List[Any]				=      np.memmap(lowercase__      ,						dtype=lowercase__      ,						shape=lowercase__      ,						mode='r'				)

						if len(weight_info['shape']				) == 0:
												__magic_name__				:							Tuple				=      weight[0]
						__magic_name__				:							Optional[int]				=      torch.tensor(lowercase__				)
						if weight_info["dtype"] == "bfloat16":
												__magic_name__				:							Union[str, Any]				=      weight.view(torch.bfloataa				)

						return weight
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Optional[int]      ,						lowercase__  :Optional[int]				)       ->  Tuple:
						if index is None or len(lowercase__				) == 0:
												# Nothing to save
												return

						__magic_name__				:							Optional[int]				=      os.path.join(lowercase__      ,						'index.json'				)
						if os.path.isfile(lowercase__				):
												with open(lowercase__      ,						'r'      ,						encoding='utf-8'				) as f:
																		__magic_name__				:							str				=      json.load(lowercase__				)
						else:
												__magic_name__				:							List[Any]				=      {}
						current_index.update(lowercase__				)

						with open(lowercase__      ,						'w'      ,						encoding='utf-8'				) as f:
												json.dump(lowercase__      ,						lowercase__      ,						indent=2				)
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Union[str, os.PathLike]      ,						lowercase__  :Dict[str, torch.Tensor]				)       ->  List[Any]:
						os.makedirs(lowercase__      ,						exist_ok=lowercase__				)
						__magic_name__				:							str				=      {}
						for name, parameter in state_dict.items():
												__magic_name__				:							Any				=      offload_weight(lowercase__      ,						lowercase__      ,						lowercase__      ,						index=lowercase__				)

						# Update index
						save_offload_index(lowercase__      ,						lowercase__				)



class     snake_case__  (      _A  ):
				def __init__(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							int				=      dataset
										__magic_name__				:							Any				=      prefix
				def __getitem__(  self     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										return self.dataset[F"{self.prefix}{key}"]
				def __iter__(  self      ):




										"""simple docstring"""




										return iter([key for key in self.dataset if key.startswith(self.prefix      )]      )





				def __len__(  self      ):




										"""simple docstring"""




										return len(self.dataset      )






class     snake_case__  (      _A  ):
				def __init__(  self     ,      _UpperCAmelCase = None     ,      _UpperCAmelCase = None     ,      _UpperCAmelCase = None     ,      _UpperCAmelCase=None     ,      ):




										"""simple docstring"""




										if state_dict is None and save_folder is None:
																raise ValueError('Need either a `state_dict` or a `save_folder` containing offloaded weights.'      )

										__magic_name__				:							Optional[Any]				=      {} if state_dict is None else state_dict
										__magic_name__				:							Tuple				=      save_folder
										if index is None and save_folder is not None:
																with open(os.path.join(_UpperCAmelCase     ,      'index.json'      )      ) as f:
																						__magic_name__				:							Tuple				=      json.load(_UpperCAmelCase      )
										__magic_name__				:							Union[str, Any]				=      {} if index is None else index
										__magic_name__				:							Any				=      list(self.state_dict.keys()      )
										self.all_keys.extend([key for key in self.index if key not in self.all_keys]      )
										__magic_name__				:							Any				=      device
				def __getitem__(  self     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										if key in self.state_dict:
																return self.state_dict[key]
										__magic_name__				:							List[str]				=      self.index[key]
										if weight_info.get('safetensors_file'      ) is not None:
																if not is_safetensors_available():
																						raise ImportError('These offloaded weights require the use of safetensors: `pip install safetensors`.'      )

																from safetensors import safe_open

																__magic_name__				:							Optional[int]				=      'cpu' if self.device is None else self.device
																with safe_open(weight_info['safetensors_file']     ,      framework='pt'     ,      device=_UpperCAmelCase      ) as f:
																						__magic_name__				:							Optional[int]				=      f.get_tensor(weight_info.get('weight_name'     ,      _UpperCAmelCase      )      )

																if "dtype" in weight_info:
																						return tensor.to(getattr(_UpperCAmelCase     ,      weight_info['dtype']      )      )
																else:
																						return tensor

										__magic_name__				:							str				=      os.path.join(self.save_folder     ,      F"{key}.dat"      )
										return load_offloaded_weight(_UpperCAmelCase     ,      _UpperCAmelCase      )
				def __iter__(  self      ):




										"""simple docstring"""




										return iter(self.all_keys      )





				def __len__(  self      ):




										"""simple docstring"""




										return len(self.all_keys      )






def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Dict[str, torch.Tensor]      ,						lowercase__  :List[str]				)       ->  Any:
						__magic_name__				:							str				=      {}
						for module_name in submodule_names:
												# We want to catch module_name parameter (module_name.xxx) or potentially module_name, but not any of the
												# submodules that could being like module_name (transformers.h.1 and transformers.h.10 for instance)
												result.update(
												    {
												        key: param
												        for key, param in state_dict.items()
												        if key == module_name or key.startswith(module_name + '.'				)
												    }				)
						return result






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

from ...utils import logging


snake_case__     							= logging.get_logger(__name__)
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Optional[Any]      ,						lowercase__  :List[str]				)       ->  Tuple:
						__magic_name__				:							List[str]				=      nn.functional.normalize(lowercase__				)
						__magic_name__				:							Union[str, Any]				=      nn.functional.normalize(lowercase__				)
						return torch.mm(lowercase__      ,						normalized_text_embeds.t()				)



class     snake_case__  (      _A  ):
				__lowerCAmelCase		=      CLIPConfig

				__lowerCAmelCase		=      ['''CLIPEncoderLayer''']
				def __init__(  self     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										super().__init__(_UpperCAmelCase      )

										__magic_name__				:							List[Any]				=      CLIPVisionModel(config.vision_config      )
										__magic_name__				:							Tuple				=      nn.Linear(config.vision_config.hidden_size     ,      config.projection_dim     ,      bias=_UpperCAmelCase      )

										__magic_name__				:							str				=      nn.Parameter(torch.ones(17     ,      config.projection_dim      )     ,      requires_grad=_UpperCAmelCase      )
										__magic_name__				:							Optional[Any]				=      nn.Parameter(torch.ones(3     ,      config.projection_dim      )     ,      requires_grad=_UpperCAmelCase      )

										__magic_name__				:							Optional[Any]				=      nn.Parameter(torch.ones(17      )     ,      requires_grad=_UpperCAmelCase      )
										__magic_name__				:							str				=      nn.Parameter(torch.ones(3      )     ,      requires_grad=_UpperCAmelCase      )
				@torch.no_grad()
				def  UpperCamelCase_						(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							Union[str, Any]				=      self.vision_model(_UpperCAmelCase      )[1]  # pooled_output
										__magic_name__				:							Any				=      self.visual_projection(_UpperCAmelCase      )

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

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

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

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

																result.append(_UpperCAmelCase      )

										__magic_name__				:							Union[str, Any]				=      [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]      ):
																												__magic_name__				:							Union[str, Any]				=      torch.zeros_like(images[idx]      )  # black image
																						else:
																												__magic_name__				:							Optional[int]				=      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  UpperCamelCase_						(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							List[str]				=      self.vision_model(_UpperCAmelCase      )[1]  # pooled_output
										__magic_name__				:							List[str]				=      self.visual_projection(_UpperCAmelCase      )

										__magic_name__				:							Union[str, Any]				=      cosine_distance(_UpperCAmelCase     ,      self.special_care_embeds      )
										__magic_name__				:							Optional[int]				=      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
										__magic_name__				:							int				=      0.0

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

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

										__magic_name__				:							Dict				=      0.0  # black image

										return images, has_nsfw_concepts


