





"""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


a__    :   Any    						= [nn.LayerNorm]

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

a__    :   str    						= version.parse(version.parse(torch.__version__).base_version)

a__    :   Tuple    						= parsed_torch_version_base >= version.parse("""2.0""")
a__    :   str    						= parsed_torch_version_base >= version.parse("""1.12""")
a__    :   Optional[int]    						= parsed_torch_version_base >= version.parse("""1.11""")
a__    :   Any    						= parsed_torch_version_base < version.parse("""1.11""")





def 							a__	(     _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase		,   self		)			->     Dict:

 '''simple docstring'''






 from torch import _softmax_backward_data

 if is_torch_less_than_1_11:
  return _softmax_backward_data(_UpperCAmelCase		,   _UpperCAmelCase		,   parent.dim		,   self		)
 else:
  return _softmax_backward_data(_UpperCAmelCase		,   _UpperCAmelCase		,   parent.dim		,   self.dtype		)





def 							a__	(     _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase = 0		)			->     nn.Linear:

 '''simple docstring'''






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



class    lowercase_ (   nn.Module							):

       def __init__(	self      ,    __SCREAMING_SNAKE_CASE      ,    __SCREAMING_SNAKE_CASE   ):






        '''simple docstring'''

        super().__init__()
        snake_case__										= nf
        snake_case__										= nn.Parameter(torch.empty(__SCREAMING_SNAKE_CASE      ,    __SCREAMING_SNAKE_CASE   )   )
        snake_case__										= nn.Parameter(torch.zeros(__SCREAMING_SNAKE_CASE   )   )
        nn.init.normal_(self.weight      ,    std=0.02   )

       def 						_SCREAMING_SNAKE_CASE(	self      ,    __SCREAMING_SNAKE_CASE   ):






        '''simple docstring'''

        snake_case__										= x.size()[:-1] + (self.nf,)
        snake_case__										= torch.addmm(self.bias      ,    x.view(-1      ,    x.size(-1   )   )      ,    self.weight   )
        snake_case__										= x.view(__SCREAMING_SNAKE_CASE   )
        return x





def 							a__	(     _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase = 1		)			->     ConvaD:

 '''simple docstring'''






 snake_case__										= index.to(layer.weight.device		)
 snake_case__										= layer.weight.index_select(_UpperCAmelCase		,   _UpperCAmelCase		).clone().detach()
 if dim == 0:
  snake_case__										= layer.bias.clone().detach()
 else:
  snake_case__										= layer.bias[index].clone().detach()
 snake_case__										= list(layer.weight.size()		)
 snake_case__										= len(_UpperCAmelCase		)
 snake_case__										= ConvaD(new_size[1]		,   new_size[0]		).to(layer.weight.device		)
 snake_case__										= False
 new_layer.weight.copy_(W.contiguous()		)
 snake_case__										= True
 snake_case__										= False
 new_layer.bias.copy_(b.contiguous()		)
 snake_case__										= True
 return new_layer





def 							a__	(     _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase = None		)			->     Union[nn.Linear, ConvaD]:

 '''simple docstring'''






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





def 							a__	(     _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase		,   *_UpperCAmelCase		)			->     torch.Tensor:

 '''simple docstring'''






 assert len(_UpperCAmelCase		) > 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
 snake_case__										= len(inspect.signature(_UpperCAmelCase		).parameters		)
 if num_args_in_forward_chunk_fn != len(_UpperCAmelCase		):
  raise ValueError(
      f"forward_chunk_fn expects {num_args_in_forward_chunk_fn} arguments, but only {len(_UpperCAmelCase		)} input "
      "tensors are given"		)

 if chunk_size > 0:
  snake_case__										= 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}"		)

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

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

 return forward_fn(*_UpperCAmelCase		)





def 							a__	(     _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase		,   _UpperCAmelCase		)			->     Tuple[Set[int], torch.LongTensor]:

 '''simple docstring'''






 snake_case__										= torch.ones(_UpperCAmelCase		,   _UpperCAmelCase		)
 snake_case__										= set(_UpperCAmelCase		) - 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
  snake_case__										= head - sum(1 if h < head else 0 for h in already_pruned_heads		)
  snake_case__										= 0
 snake_case__										= mask.view(-1		).contiguous().eq(1		)
 snake_case__										= torch.arange(len(_UpperCAmelCase		)		)[mask].long()
 return heads, index





def 							a__	(     *_UpperCAmelCase		,   _UpperCAmelCase = None		)			->     Tuple[torch.Tensor, ...]:

 '''simple docstring'''






 return torch.meshgrid(*_UpperCAmelCase		,   indexing=_UpperCAmelCase		)





def 							a__	(     _UpperCAmelCase		)			->     Tuple[torch.device, int, int]:

 '''simple docstring'''






 return tensor.device, storage_ptr(_UpperCAmelCase		), storage_size(_UpperCAmelCase		)








"""simple docstring"""


from typing import List

import PIL
import torch
from PIL import Image

from ...configuration_utils import ConfigMixin
from ...models.modeling_utils import ModelMixin
from ...utils import PIL_INTERPOLATION



class    lowercase_ (   __lowerCAmelCase ,__lowerCAmelCase							):

       def __init__(	self   ):






        '''simple docstring'''

        super().__init__()

        self.register_buffer("watermark_image"      ,    torch.zeros((62, 62, 4)   )   )
        snake_case__										= None

       def 						_SCREAMING_SNAKE_CASE(	self      ,    __SCREAMING_SNAKE_CASE      ,    __SCREAMING_SNAKE_CASE=None   ):






        '''simple docstring'''

        snake_case__										= images[0].height
        snake_case__										= images[0].width

        snake_case__										= sample_size or h

        snake_case__										= min(h / sample_size      ,    w / sample_size   )
        snake_case__							,      snake_case__										= (int(h / coef   ), int(w / coef   )) if coef < 1 else (h, w)

        snake_case__							,      snake_case__										= 1024**2, img_w * img_h
        snake_case__										= (Sa / Sa) ** 0.5
        snake_case__							,      snake_case__							,      snake_case__										= int(K * 62   ), img_w - int(14 * K   ), img_h - int(14 * K   )

        if self.watermark_image_as_pil is None:
         snake_case__										= self.watermark_image.to(torch.uinta   ).cpu().numpy()
         snake_case__										= Image.fromarray(__SCREAMING_SNAKE_CASE      ,    mode="RGBA"   )
         snake_case__										= watermark_image

        snake_case__										= self.watermark_image_as_pil.resize(
            (wm_size, wm_size)      ,    PIL_INTERPOLATION["bicubic"]      ,    reducing_gap=__SCREAMING_SNAKE_CASE   )

        for pil_img in images:
         pil_img.paste(__SCREAMING_SNAKE_CASE      ,    box=(wm_x - wm_size, wm_y - wm_size, wm_x, wm_y)      ,    mask=wm_img.split()[-1]   )

        return images

