


from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block





@dataclass
class 							lowercase_					(			a__):






							"""simple docstring"""







							snake_case_										=		42





class 							lowercase_					(			nn.Module):






							"""simple docstring"""







							def __init__( self      ,     _UpperCAmelCase=3      ,     _UpperCAmelCase=3      ,     _UpperCAmelCase=("DownEncoderBlock2D",)      ,     _UpperCAmelCase=(64,)      ,     _UpperCAmelCase=2      ,     _UpperCAmelCase=32      ,     _UpperCAmelCase="silu"      ,     _UpperCAmelCase=True      ,     ):







													"""simple docstring"""


													super().__init__()
													a_		      =		layers_per_block

													a_		      =		torch.nn.Convad(
													    __a      ,     block_out_channels[0]      ,     kernel_size=3      ,     stride=1      ,     padding=1      ,     )

													a_		      =		None
													a_		      =		nn.ModuleList([]		)

													# down
													a_		      =		block_out_channels[0]
													for i, down_block_type in enumerate(__a		):
																			a_		      =		output_channel
																			a_		      =		block_out_channels[i]
																			a_		      =		i == len(__a		) - 1

																			a_		      =		get_down_block(
																			    __a      ,     num_layers=self.layers_per_block      ,     in_channels=__a      ,     out_channels=__a      ,     add_downsample=not is_final_block      ,     resnet_eps=1e-6      ,     downsample_padding=0      ,     resnet_act_fn=__a      ,     resnet_groups=__a      ,     attention_head_dim=__a      ,     temb_channels=__a      ,     )
																			self.down_blocks.append(__a		)

													# mid
													a_		      =		UNetMidBlockaD(
													    in_channels=block_out_channels[-1]      ,     resnet_eps=1e-6      ,     resnet_act_fn=__a      ,     output_scale_factor=1      ,     resnet_time_scale_shift="""default"""      ,     attention_head_dim=block_out_channels[-1]      ,     resnet_groups=__a      ,     temb_channels=__a      ,     )

													# out
													a_		      =		nn.GroupNorm(num_channels=block_out_channels[-1]      ,     num_groups=__a      ,     eps=1e-6		)
													a_		      =		nn.SiLU()

													a_		      =		2 * out_channels if double_z else out_channels
													a_		      =		nn.Convad(block_out_channels[-1]      ,     __a      ,     3      ,     padding=1		)

													a_		      =		False






							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		x
													a_		      =		self.conv_in(__a		)

													if self.training and self.gradient_checkpointing:

																			def create_custom_forward(_UpperCAmelCase		):
																									def custom_forward(*_UpperCAmelCase		):
																															return module(*__a		)

																									return custom_forward

																			# down
																			if is_torch_version(""">="""      ,     """1.11.0"""		):
																									for down_block in self.down_blocks:
																															a_		      =		torch.utils.checkpoint.checkpoint(
																															    create_custom_forward(__a		)      ,     __a      ,     use_reentrant=__a		)
																									# middle
																									a_		      =		torch.utils.checkpoint.checkpoint(
																									    create_custom_forward(self.mid_block		)      ,     __a      ,     use_reentrant=__a		)
																			else:
																									for down_block in self.down_blocks:
																															a_		      =		torch.utils.checkpoint.checkpoint(create_custom_forward(__a		)      ,     __a		)
																									# middle
																									a_		      =		torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block		)      ,     __a		)

													else:
																			# down
																			for down_block in self.down_blocks:
																									a_		      =		down_block(__a		)

																			# middle
																			a_		      =		self.mid_block(__a		)

													# post-process
													a_		      =		self.conv_norm_out(__a		)
													a_		      =		self.conv_act(__a		)
													a_		      =		self.conv_out(__a		)

													return sample





class 							lowercase_					(			nn.Module):






							"""simple docstring"""







							def __init__( self      ,     _UpperCAmelCase=3      ,     _UpperCAmelCase=3      ,     _UpperCAmelCase=("UpDecoderBlock2D",)      ,     _UpperCAmelCase=(64,)      ,     _UpperCAmelCase=2      ,     _UpperCAmelCase=32      ,     _UpperCAmelCase="silu"      ,     _UpperCAmelCase="group"      ,     ):







													"""simple docstring"""


													super().__init__()
													a_		      =		layers_per_block

													a_		      =		nn.Convad(
													    __a      ,     block_out_channels[-1]      ,     kernel_size=3      ,     stride=1      ,     padding=1      ,     )

													a_		      =		None
													a_		      =		nn.ModuleList([]		)

													a_		      =		in_channels if norm_type == """spatial""" else None

													# mid
													a_		      =		UNetMidBlockaD(
													    in_channels=block_out_channels[-1]      ,     resnet_eps=1e-6      ,     resnet_act_fn=__a      ,     output_scale_factor=1      ,     resnet_time_scale_shift="""default""" if norm_type == """group""" else norm_type      ,     attention_head_dim=block_out_channels[-1]      ,     resnet_groups=__a      ,     temb_channels=__a      ,     )

													# up
													a_		      =		list(reversed(__a		)		)
													a_		      =		reversed_block_out_channels[0]
													for i, up_block_type in enumerate(__a		):
																			a_		      =		output_channel
																			a_		      =		reversed_block_out_channels[i]

																			a_		      =		i == len(__a		) - 1

																			a_		      =		get_up_block(
																			    __a      ,     num_layers=self.layers_per_block + 1      ,     in_channels=__a      ,     out_channels=__a      ,     prev_output_channel=__a      ,     add_upsample=not is_final_block      ,     resnet_eps=1e-6      ,     resnet_act_fn=__a      ,     resnet_groups=__a      ,     attention_head_dim=__a      ,     temb_channels=__a      ,     resnet_time_scale_shift=__a      ,     )
																			self.up_blocks.append(__a		)
																			a_		      =		output_channel

													# out
													if norm_type == "spatial":
																			a_		      =		SpatialNorm(block_out_channels[0]      ,     __a		)
													else:
																			a_		      =		nn.GroupNorm(num_channels=block_out_channels[0]      ,     num_groups=__a      ,     eps=1e-6		)
													a_		      =		nn.SiLU()
													a_		      =		nn.Convad(block_out_channels[0]      ,     __a      ,     3      ,     padding=1		)

													a_		      =		False






							def     lowercase__  ( self      ,     _UpperCAmelCase      ,     _UpperCAmelCase=None		):







													"""simple docstring"""


													a_		      =		z
													a_		      =		self.conv_in(__a		)

													a_		      =		next(iter(self.up_blocks.parameters()		)		).dtype
													if self.training and self.gradient_checkpointing:

																			def create_custom_forward(_UpperCAmelCase		):
																									def custom_forward(*_UpperCAmelCase		):
																															return module(*__a		)

																									return custom_forward

																			if is_torch_version(""">="""      ,     """1.11.0"""		):
																									# middle
																									a_		      =		torch.utils.checkpoint.checkpoint(
																									    create_custom_forward(self.mid_block		)      ,     __a      ,     __a      ,     use_reentrant=__a		)
																									a_		      =		sample.to(__a		)

																									# up
																									for up_block in self.up_blocks:
																															a_		      =		torch.utils.checkpoint.checkpoint(
																															    create_custom_forward(__a		)      ,     __a      ,     __a      ,     use_reentrant=__a		)
																			else:
																									# middle
																									a_		      =		torch.utils.checkpoint.checkpoint(
																									    create_custom_forward(self.mid_block		)      ,     __a      ,     __a		)
																									a_		      =		sample.to(__a		)

																									# up
																									for up_block in self.up_blocks:
																															a_		      =		torch.utils.checkpoint.checkpoint(create_custom_forward(__a		)      ,     __a      ,     __a		)
													else:
																			# middle
																			a_		      =		self.mid_block(__a      ,     __a		)
																			a_		      =		sample.to(__a		)

																			# up
																			for up_block in self.up_blocks:
																									a_		      =		up_block(__a      ,     __a		)

        # post-process
													if latent_embeds is None:
																			a_		      =		self.conv_norm_out(__a		)
													else:
																			a_		      =		self.conv_norm_out(__a      ,     __a		)
													a_		      =		self.conv_act(__a		)
													a_		      =		self.conv_out(__a		)

													return sample





class 							lowercase_					(			nn.Module):






							"""simple docstring"""







							def __init__( self      ,     _UpperCAmelCase      ,     _UpperCAmelCase      ,     _UpperCAmelCase      ,     _UpperCAmelCase=None      ,     _UpperCAmelCase="random"      ,     _UpperCAmelCase=False      ,     _UpperCAmelCase=True		):







													"""simple docstring"""


													super().__init__()
													a_		      =		n_e
													a_		      =		vq_embed_dim
													a_		      =		beta
													a_		      =		legacy

													a_		      =		nn.Embedding(self.n_e      ,     self.vq_embed_dim		)
													self.embedding.weight.data.uniform_(-1.0 / self.n_e      ,     1.0 / self.n_e		)

													a_		      =		remap
													if self.remap is not None:
																			self.register_buffer("""used"""      ,     torch.tensor(np.load(self.remap		)		)		)
																			a_		      =		self.used.shape[0]
																			a_		      =		unknown_index  # "random" or "extra" or integer
																			if self.unknown_index == "extra":
																									a_		      =		self.re_embed
																									a_		      =		self.re_embed + 1
																			print(
																			    f"Remapping {self.n_e} indices to {self.re_embed} indices. "
																			    f"Using {self.unknown_index} for unknown indices."		)
													else:
																			a_		      =		n_e

													a_		      =		sane_index_shape


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		inds.shape
													assert len(__a		) > 1
													a_		      =		inds.reshape(ishape[0]      ,     -1		)
													a_		      =		self.used.to(__a		)
													a_		      =		(inds[:, :, None] == used[None, None, ...]).long()
													a_		      =		match.argmax(-1		)
													a_		      =		match.sum(2		) < 1
													if self.unknown_index == "random":
																			a_		      =		torch.randint(0      ,     self.re_embed      ,     size=new[unknown].shape		).to(device=new.device		)
													else:
																			a_		      =		self.unknown_index
													return new.reshape(__a		)


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		inds.shape
													assert len(__a		) > 1
													a_		      =		inds.reshape(ishape[0]      ,     -1		)
													a_		      =		self.used.to(__a		)
													if self.re_embed > self.used.shape[0]:  # extra token
																			a_		      =		0  # simply set to zero
													a_		      =		torch.gather(used[None, :][inds.shape[0] * [0], :]      ,     1      ,     __a		)
													return back.reshape(__a		)


							def     lowercase__  ( self      ,     _UpperCAmelCase		):







													"""simple docstring"""


													a_		      =		z.permute(0      ,     2      ,     3      ,     1		).contiguous()
													a_		      =		z.view(-1      ,     self.vq_embed_dim		)

													# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
													a_		      =		torch.argmin(torch.cdist(__a      ,     self.embedding.weight		)      ,     dim=1		)

													a_		      =		self.embedding(__a		).view(z.shape		)
													a_		      =		None
													a_		      =		None

													# compute loss for embedding
													if not self.legacy:
																			a_		      =		self.beta * torch.mean((z_q.detach() - z) ** 2		) + torch.mean((z_q - z.detach()) ** 2		)
													else:
																			a_		      =		torch.mean((z_q.detach() - z) ** 2		) + self.beta * torch.mean((z_q - z.detach()) ** 2		)

													# preserve gradients
													a_		      =		z + (z_q - z).detach()

													# reshape back to match original input shape
													a_		      =		z_q.permute(0      ,     3      ,     1      ,     2		).contiguous()

													if self.remap is not None:
																			a_		      =		min_encoding_indices.reshape(z.shape[0]      ,     -1		)  # add batch axis
																			a_		      =		self.remap_to_used(__a		)
																			a_		      =		min_encoding_indices.reshape(-1      ,     1		)  # flatten

													if self.sane_index_shape:
																			a_		      =		min_encoding_indices.reshape(z_q.shape[0]      ,     z_q.shape[2]      ,     z_q.shape[3]		)

													return z_q, loss, (perplexity, min_encodings, min_encoding_indices)






							def     lowercase__  ( self      ,     _UpperCAmelCase      ,     _UpperCAmelCase		):







													"""simple docstring"""


													if self.remap is not None:
																			a_		      =		indices.reshape(shape[0]      ,     -1		)  # add batch axis
																			a_		      =		self.unmap_to_all(__a		)
																			a_		      =		indices.reshape(-1		)  # flatten again

													# get quantized latent vectors
													a_		      =		self.embedding(__a		)

													if shape is not None:
																			a_		      =		z_q.view(__a		)
																			# reshape back to match original input shape
																			a_		      =		z_q.permute(0      ,     3      ,     1      ,     2		).contiguous()

													return z_q





class 							lowercase_					(			a__):






							"""simple docstring"""







							def __init__( self      ,     _UpperCAmelCase      ,     _UpperCAmelCase=False		):







													"""simple docstring"""


													a_		      =		parameters
													a_		      =		torch.chunk(__a      ,     2      ,     dim=1		)
													a_		      =		torch.clamp(self.logvar      ,     -3_0.0      ,     2_0.0		)
													a_		      =		deterministic
													a_		      =		torch.exp(0.5 * self.logvar		)
													a_		      =		torch.exp(self.logvar		)
													if self.deterministic:
																			a_		      =		torch.zeros_like(
																			    self.mean      ,     device=self.parameters.device      ,     dtype=self.parameters.dtype		)


							def     lowercase__  ( self      ,     _UpperCAmelCase = None		):







													"""simple docstring"""


													a_		      =		randn_tensor(
													    self.mean.shape      ,     generator=__a      ,     device=self.parameters.device      ,     dtype=self.parameters.dtype		)
													a_		      =		self.mean + self.std * sample
													return x


							def     lowercase__  ( self      ,     _UpperCAmelCase=None		):







													"""simple docstring"""


													if self.deterministic:
																			return torch.Tensor([0.0]		)
													else:
																			if other is None:
																									return 0.5 * torch.sum(torch.pow(self.mean      ,     2		) + self.var - 1.0 - self.logvar      ,     dim=[1, 2, 3]		)
																			else:
																									return 0.5 * torch.sum(
																									    torch.pow(self.mean - other.mean      ,     2		) / other.var
																									    + self.var / other.var
																									    - 1.0
																									    - self.logvar
																									    + other.logvar      ,     dim=[1, 2, 3]      ,     )


							def     lowercase__  ( self      ,     _UpperCAmelCase      ,     _UpperCAmelCase=[1, 2, 3]		):







													"""simple docstring"""


													if self.deterministic:
																			return torch.Tensor([0.0]		)
													a_		      =		np.log(2.0 * np.pi		)
													return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean      ,     2		) / self.var      ,     dim=__a		)






							def     lowercase__  ( self		):







													"""simple docstring"""


													return self.mean





'''simple docstring'''







from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowercase      :   Optional[int]      =       logging.get_logger(__name__)

lowercase      :   Optional[Any]      =       {
    """google/pegasus-large""": """https://huggingface.co/google/pegasus-large/resolve/main/config.json""",
    # See all PEGASUS models at https://huggingface.co/models?filter=pegasus
}







class 					_a (a__	):





     '''simple docstring'''



     lowerCAmelCase_  :     Any          =					"""pegasus"""
     lowerCAmelCase_  :     Dict          =					["""past_key_values"""]
     lowerCAmelCase_  :     str          =					{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}







     def __init__(				self							,__a=50_265							,__a=1_024							,__a=12							,__a=4_096							,__a=16							,__a=12							,__a=4_096							,__a=16							,__a=0.0							,__a=0.0							,__a=True							,__a=True							,__a="gelu"							,__a=1_024							,__a=0.1							,__a=0.0							,__a=0.0							,__a=0.02							,__a=0							,__a=False							,__a=0							,__a=1							,__a=1							,**__a							,)      ->				int:
          snake_case			:      List[Any]        =			vocab_size
          snake_case			:      Optional[Any]        =			max_position_embeddings
          snake_case			:      List[Any]        =			d_model
          snake_case			:      str        =			encoder_ffn_dim
          snake_case			:      List[Any]        =			encoder_layers
          snake_case			:      Optional[Any]        =			encoder_attention_heads
          snake_case			:      Union[str, Any]        =			decoder_ffn_dim
          snake_case			:      Tuple        =			decoder_layers
          snake_case			:      Union[str, Any]        =			decoder_attention_heads
          snake_case			:      Union[str, Any]        =			dropout
          snake_case			:      int        =			attention_dropout
          snake_case			:      int        =			activation_dropout
          snake_case			:      Optional[Any]        =			activation_function
          snake_case			:      Tuple        =			init_std
          snake_case			:      Union[str, Any]        =			encoder_layerdrop
          snake_case			:      int        =			decoder_layerdrop
          snake_case			:      Dict        =			use_cache
          snake_case			:      Dict        =			encoder_layers
          snake_case			:      List[Any]        =			scale_embedding  # scale factor will be sqrt(d_model) if True
          super().__init__(
              pad_token_id=__a							,eos_token_id=__a							,is_encoder_decoder=__a							,decoder_start_token_id=__a							,forced_eos_token_id=__a							,**__a							,)







     @property
     def 			snake_case_      (				self							)      ->				int:
          return self.encoder_attention_heads







     @property
     def 			snake_case_      (				self							)      ->				int:
          return self.d_model



