


"""simple docstring"""







import inspect
from typing import Callable, List, Optional, Set, Tuple, Union

import torch
from packaging import version
from safetensors.torch import storage_ptr, storage_size
from torch import nn

from .utils import logging


lowerCamelCase_  :  Dict     		=    [nn.LayerNorm]

lowerCamelCase_  :  Optional[Any]     		=    logging.get_logger(__name__)

lowerCamelCase_  :  List[str]     		=    version.parse(version.parse(torch.__version__).base_version)

lowerCamelCase_  :  List[Any]     		=    parsed_torch_version_base >= version.parse("""2.0""")
lowerCamelCase_  :  Union[str, Any]     		=    parsed_torch_version_base >= version.parse("""1.12""")
lowerCamelCase_  :  Optional[Any]     		=    parsed_torch_version_base >= version.parse("""1.11""")
lowerCamelCase_  :  List[Any]     		=    parsed_torch_version_base < version.parse("""1.11""")
def _A		(		lowercase			,       lowercase			,       lowercase			,       lowercase			,       self						):







    """simple docstring"""
    from torch import _softmax_backward_data

    if is_torch_less_than_1_11:
        return _softmax_backward_data(lowercase			,       lowercase			,       parent.dim			,       self						)
    else:
        return _softmax_backward_data(lowercase			,       lowercase			,       parent.dim			,       self.dtype						)
def _A		(		lowercase			,       lowercase			,       lowercase = 0						):







    """simple docstring"""
    a											=index.to(layer.weight.device						)
    a											=layer.weight.index_select(lowercase			,       lowercase						).clone().detach()
    if layer.bias is not None:
        if dim == 1:
            a											=layer.bias.clone().detach()
        else:
            a											=layer.bias[index].clone().detach()
    a											=list(layer.weight.size()						)
    a											=len(lowercase						)
    a											=nn.Linear(new_size[1]			,       new_size[0]			,       bias=layer.bias is not None						).to(layer.weight.device						)
    a											=False
    new_layer.weight.copy_(W.contiguous()						)
    a											=True
    if layer.bias is not None:
        a											=False
        new_layer.bias.copy_(b.contiguous()						)
        a											=True
    return new_layer






class  __A							(			nn.Module     ):






 """simple docstring"""




 def __init__(						self					,       __A					,       __A		)   ->	Optional[int]:
     super().__init__()
     a											=nf
     a											=nn.Parameter(torch.empty(__A					,       __A		)		)
     a											=nn.Parameter(torch.zeros(__A		)		)
     nn.init.normal_(self.weight					,       std=0.02		)





 def  SCREAMING_SNAKE_CASE				(						self					,       __A		)   ->	int:
     a											=x.size()[:-1] + (self.nf,)
     a											=torch.addmm(self.bias					,       x.view(-1					,       x.size(-1		)		)					,       self.weight		)
     a											=x.view(__A		)
     return x

def _A		(		lowercase			,       lowercase			,       lowercase = 1						):







    """simple docstring"""
    a											=index.to(layer.weight.device						)
    a											=layer.weight.index_select(lowercase			,       lowercase						).clone().detach()
    if dim == 0:
        a											=layer.bias.clone().detach()
    else:
        a											=layer.bias[index].clone().detach()
    a											=list(layer.weight.size()						)
    a											=len(lowercase						)
    a											=ConvaD(new_size[1]			,       new_size[0]						).to(layer.weight.device						)
    a											=False
    new_layer.weight.copy_(W.contiguous()						)
    a											=True
    a											=False
    new_layer.bias.copy_(b.contiguous()						)
    a											=True
    return new_layer
def _A		(		lowercase			,       lowercase			,       lowercase = None						):







    """simple docstring"""
    if isinstance(lowercase			,       nn.Linear						):
        return prune_linear_layer(lowercase			,       lowercase			,       dim=0 if dim is None else dim						)
    elif isinstance(lowercase			,       lowercase						):
        return prune_convad_layer(lowercase			,       lowercase			,       dim=1 if dim is None else dim						)
    else:
        raise ValueError(f'''Can\'t prune layer of class {layer.__class__}'''						)
def _A		(		lowercase			,       lowercase			,       lowercase			,       *lowercase						):







    """simple docstring"""
    assert len(lowercase						) > 0, f'''{input_tensors} has to be a tuple/list of tensors'''

    # inspect.signature exist since python 3.5 and is a python method -> no problem with backward compatibility
    a											=len(inspect.signature(lowercase						).parameters						)
    if num_args_in_forward_chunk_fn != len(lowercase						):
        raise ValueError(
            f'''forward_chunk_fn expects {num_args_in_forward_chunk_fn} arguments, but only {len(lowercase						)} input '''
            '''tensors are given'''						)

    if chunk_size > 0:
        a											=input_tensors[0].shape[chunk_dim]
        for input_tensor in input_tensors:
            if input_tensor.shape[chunk_dim] != tensor_shape:
                raise ValueError(
                    f'''All input tenors have to be of the same shape: {tensor_shape}, '''
                    f'''found shape {input_tensor.shape[chunk_dim]}'''						)

        if input_tensors[0].shape[chunk_dim] % chunk_size != 0:
            raise ValueError(
                f'''The dimension to be chunked {input_tensors[0].shape[chunk_dim]} has to be a multiple of the chunk '''
                f'''size {chunk_size}'''						)

        a											=input_tensors[0].shape[chunk_dim] // chunk_size

        # chunk input tensor into tuples
        a											=tuple(input_tensor.chunk(lowercase			,       dim=lowercase						) for input_tensor in input_tensors						)
        # apply forward fn to every tuple
        a											=tuple(forward_fn(*lowercase						) for input_tensors_chunk in zip(*lowercase						)						)
        # concatenate output at same dimension
        return torch.cat(lowercase			,       dim=lowercase						)

    return forward_fn(*lowercase						)
def _A		(		lowercase			,       lowercase			,       lowercase			,       lowercase						):







    """simple docstring"""
    a											=torch.ones(lowercase			,       lowercase						)
    a											=set(lowercase						) - already_pruned_heads  # Convert to set and remove already pruned heads
    for head in heads:
        # Compute how many pruned heads are before the head and move the index accordingly
        a											=head - sum(1 if h < head else 0 for h in already_pruned_heads						)
        a											=0
    a											=mask.view(-1						).contiguous().eq(1						)
    a											=torch.arange(len(lowercase						)						)[mask].long()
    return heads, index
def _A		(		*lowercase			,       lowercase = None						):







    """simple docstring"""
    return torch.meshgrid(*lowercase			,       indexing=lowercase						)





def _A		(		lowercase						):







    """simple docstring"""
    return tensor.device, storage_ptr(lowercase						), storage_size(lowercase						)



"""simple docstring"""







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

from ...utils import logging


lowerCamelCase_  :  Tuple     		=    logging.get_logger(__name__)
def _A		(		lowercase			,       lowercase						):







    """simple docstring"""
    a											=nn.functional.normalize(lowercase						)
    a											=nn.functional.normalize(lowercase						)
    return torch.mm(lowercase			,       normalized_text_embeds.t()						)






class  __A							(			_SCREAMING_SNAKE_CASE     ):






 """simple docstring"""




 __lowerCAmelCase					   =    CLIPConfig

 __lowerCAmelCase					   =    ["CLIPEncoderLayer"]


 def __init__(						self					,       __A		)   ->	int:
     super().__init__(__A		)

     a											=CLIPVisionModel(config.vision_config		)
     a											=nn.Linear(config.vision_config.hidden_size					,       config.projection_dim					,       bias=__A		)

     a											=nn.Parameter(torch.ones(17					,       config.projection_dim		)					,       requires_grad=__A		)
     a											=nn.Parameter(torch.ones(3					,       config.projection_dim		)					,       requires_grad=__A		)

     a											=nn.Parameter(torch.ones(17		)					,       requires_grad=__A		)
     a											=nn.Parameter(torch.ones(3		)					,       requires_grad=__A		)


 @torch.no_grad()
 def  SCREAMING_SNAKE_CASE				(						self					,       __A					,       __A		)   ->	Union[str, Any]:
     a											=self.vision_model(__A		)[1]  # pooled_output
     a											=self.visual_projection(__A		)

     # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
     a											=cosine_distance(__A					,       self.special_care_embeds		).cpu().float().numpy()
     a											=cosine_distance(__A					,       self.concept_embeds		).cpu().float().numpy()

     a											=[]
     a											=image_embeds.shape[0]
     for i in range(__A		):
         a											={'''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
         a											=0.0

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

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

         result.append(__A		)

     a											=[len(res['''bad_concepts''']		) > 0 for res in result]

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

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

     return images, has_nsfw_concepts





 @torch.no_grad()
 def  SCREAMING_SNAKE_CASE				(						self					,       __A					,       __A		)   ->	str:
     a											=self.vision_model(__A		)[1]  # pooled_output
     a											=self.visual_projection(__A		)

     a											=cosine_distance(__A					,       self.special_care_embeds		)
     a											=cosine_distance(__A					,       self.concept_embeds		)

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

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

     a											=(cos_dist - self.concept_embeds_weights) + special_adjustment
     # concept_scores = concept_scores.round(decimals=3)
     a											=torch.any(concept_scores > 0					,       dim=1		)

     a											=0.0  # black image

     return images, has_nsfw_concepts