





"""simple docstring"""






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	(	_UpperCamelCase							):







					'''simple docstring'''




					__lowerCAmelCase     =     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		,	):
									super().__init__()
									__a       :  Optional[Any]					=     layers_per_block

									__a       :  Optional[int]					=     torch.nn.Convad(
									    _UpperCAmelCase		,	block_out_channels[0]		,	kernel_size=3		,	stride=1		,	padding=1		,	)

									__a       :  str					=     None
									__a       :  Union[str, Any]					=     nn.ModuleList([]   )

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

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

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

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

									__a       :  Optional[int]					=     2 * out_channels if double_z else out_channels
									__a       :  Tuple					=     nn.Convad(block_out_channels[-1]		,	_UpperCAmelCase		,	3		,	padding=1   )

									__a       :  List[str]					=     False







					def        _lowerCamelCase						(  self		,	_UpperCAmelCase   ):
									__a       :  int					=     x
									__a       :  Tuple					=     self.conv_in(_UpperCAmelCase   )

									if self.training and self.gradient_checkpointing:

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

																	return custom_forward

													# down
													if is_torch_version('''>='''		,	'''1.11.0'''   ):
																	for down_block in self.down_blocks:
																					__a       :  str					=     torch.utils.checkpoint.checkpoint(
																					    create_custom_forward(_UpperCAmelCase   )		,	_UpperCAmelCase		,	use_reentrant=_UpperCAmelCase   )
																	# middle
																	__a       :  Optional[int]					=     torch.utils.checkpoint.checkpoint(
																	    create_custom_forward(self.mid_block   )		,	_UpperCAmelCase		,	use_reentrant=_UpperCAmelCase   )
													else:
																	for down_block in self.down_blocks:
																					__a       :  Optional[int]					=     torch.utils.checkpoint.checkpoint(create_custom_forward(_UpperCAmelCase   )		,	_UpperCAmelCase   )
																	# middle
																	__a       :  List[str]					=     torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block   )		,	_UpperCAmelCase   )

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

													# middle
													__a       :  Optional[Any]					=     self.mid_block(_UpperCAmelCase   )

									# post-process
									__a       :  Tuple					=     self.conv_norm_out(_UpperCAmelCase   )
									__a       :  List[Any]					=     self.conv_act(_UpperCAmelCase   )
									__a       :  Dict					=     self.conv_out(_UpperCAmelCase   )

									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"		,	):
									super().__init__()
									__a       :  int					=     layers_per_block

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

									__a       :  Optional[Any]					=     None
									__a       :  Optional[int]					=     nn.ModuleList([]   )

									__a       :  Tuple					=     in_channels if norm_type == '''spatial''' else None

									# mid
									__a       :  Union[str, Any]					=     UNetMidBlockaD(
									    in_channels=block_out_channels[-1]		,	resnet_eps=1e-6		,	resnet_act_fn=_UpperCAmelCase		,	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=_UpperCAmelCase		,	temb_channels=_UpperCAmelCase		,	)

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

													__a       :  Tuple					=     i == len(_UpperCAmelCase   ) - 1

													__a       :  int					=     get_up_block(
													    _UpperCAmelCase		,	num_layers=self.layers_per_block + 1		,	in_channels=_UpperCAmelCase		,	out_channels=_UpperCAmelCase		,	prev_output_channel=_UpperCAmelCase		,	add_upsample=not is_final_block		,	resnet_eps=1e-6		,	resnet_act_fn=_UpperCAmelCase		,	resnet_groups=_UpperCAmelCase		,	attention_head_dim=_UpperCAmelCase		,	temb_channels=_UpperCAmelCase		,	resnet_time_scale_shift=_UpperCAmelCase		,	)
													self.up_blocks.append(_UpperCAmelCase   )
													__a       :  Optional[Any]					=     output_channel

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

									__a       :  Optional[int]					=     False







					def        _lowerCamelCase						(  self		,	_UpperCAmelCase		,	_UpperCAmelCase=None   ):
									__a       :  str					=     z
									__a       :  Tuple					=     self.conv_in(_UpperCAmelCase   )

									__a       :  Dict					=     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(*_UpperCAmelCase   )

																	return custom_forward

													if is_torch_version('''>='''		,	'''1.11.0'''   ):
																	# middle
																	__a       :  List[str]					=     torch.utils.checkpoint.checkpoint(
																	    create_custom_forward(self.mid_block   )		,	_UpperCAmelCase		,	_UpperCAmelCase		,	use_reentrant=_UpperCAmelCase   )
																	__a       :  str					=     sample.to(_UpperCAmelCase   )

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

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

													# up
													for up_block in self.up_blocks:
																	__a       :  List[Any]					=     up_block(_UpperCAmelCase		,	_UpperCAmelCase   )

        # post-process
									if latent_embeds is None:
													__a       :  str					=     self.conv_norm_out(_UpperCAmelCase   )
									else:
													__a       :  Any					=     self.conv_norm_out(_UpperCAmelCase		,	_UpperCAmelCase   )
									__a       :  Optional[int]					=     self.conv_act(_UpperCAmelCase   )
									__a       :  List[str]					=     self.conv_out(_UpperCAmelCase   )

									return sample






class  __lowercase	(	nn.Module							):







					'''simple docstring'''




					def __init__(  self		,	_UpperCAmelCase		,	_UpperCAmelCase		,	_UpperCAmelCase		,	_UpperCAmelCase=None		,	_UpperCAmelCase="random"		,	_UpperCAmelCase=False		,	_UpperCAmelCase=True   ):
									super().__init__()
									__a       :  Optional[Any]					=     n_e
									__a       :  Tuple					=     vq_embed_dim
									__a       :  Any					=     beta
									__a       :  List[Any]					=     legacy

									__a       :  Dict					=     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       :  Optional[int]					=     remap
									if self.remap is not None:
													self.register_buffer('''used'''		,	torch.tensor(np.load(self.remap   )   )   )
													__a       :  Any					=     self.used.shape[0]
													__a       :  List[str]					=     unknown_index  # "random" or "extra" or integer
													if self.unknown_index == "extra":
																	__a       :  Optional[int]					=     self.re_embed
																	__a       :  Optional[Any]					=     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       :  Any					=     n_e

									__a       :  Optional[int]					=     sane_index_shape







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







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







					def        _lowerCamelCase						(  self		,	_UpperCAmelCase   ):
									__a       :  Any					=     z.permute(0		,	2		,	3		,	1   ).contiguous()
									__a       :  int					=     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       :  int					=     torch.argmin(torch.cdist(_UpperCAmelCase		,	self.embedding.weight   )		,	dim=1   )

									__a       :  Union[str, Any]					=     self.embedding(_UpperCAmelCase   ).view(z.shape   )
									__a       :  List[str]					=     None
									__a       :  List[str]					=     None

									# compute loss for embedding
									if not self.legacy:
													__a       :  Union[str, Any]					=     self.beta * torch.mean((z_q.detach() - z) ** 2   ) + torch.mean((z_q - z.detach()) ** 2   )
									else:
													__a       :  Dict					=     torch.mean((z_q.detach() - z) ** 2   ) + self.beta * torch.mean((z_q - z.detach()) ** 2   )

									# preserve gradients
									__a       :  int					=     z + (z_q - z).detach()

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

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

									if self.sane_index_shape:
													__a       :  Any					=     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        _lowerCamelCase						(  self		,	_UpperCAmelCase		,	_UpperCAmelCase   ):






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

									# get quantized latent vectors
									__a       :  Optional[int]					=     self.embedding(_UpperCAmelCase   )

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

									return z_q






class  __lowercase	(	_UpperCamelCase							):







					'''simple docstring'''




					def __init__(  self		,	_UpperCAmelCase		,	_UpperCAmelCase=False   ):
									__a       :  Optional[Any]					=     parameters
									__a				,  __a       :  Any					=     torch.chunk(_UpperCAmelCase		,	2		,	dim=1   )
									__a       :  List[Any]					=     torch.clamp(self.logvar		,	-3_0.0		,	2_0.0   )
									__a       :  Union[str, Any]					=     deterministic
									__a       :  Union[str, Any]					=     torch.exp(0.5 * self.logvar   )
									__a       :  Any					=     torch.exp(self.logvar   )
									if self.deterministic:
													__a       :  Optional[int]					=     torch.zeros_like(
													    self.mean		,	device=self.parameters.device		,	dtype=self.parameters.dtype   )







					def        _lowerCamelCase						(  self		,	_UpperCAmelCase = None   ):
									__a       :  Dict					=     randn_tensor(
									    self.mean.shape		,	generator=_UpperCAmelCase		,	device=self.parameters.device		,	dtype=self.parameters.dtype   )
									__a       :  Tuple					=     self.mean + self.std * sample
									return x







					def        _lowerCamelCase						(  self		,	_UpperCAmelCase=None   ):






									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        _lowerCamelCase						(  self		,	_UpperCAmelCase		,	_UpperCAmelCase=[1, 2, 3]   ):






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







					def        _lowerCamelCase						(  self   ):
									return self.mean


from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


lowercase_									=   {
    'configuration_pix2struct': [
        'PIX2STRUCT_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'Pix2StructConfig',
        'Pix2StructTextConfig',
        'Pix2StructVisionConfig',
    ],
    'processing_pix2struct': ['Pix2StructProcessor'],
}

try:
  if not is_vision_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  lowercase_									=   ['Pix2StructImageProcessor']


try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  lowercase_									=   [
      'PIX2STRUCT_PRETRAINED_MODEL_ARCHIVE_LIST',
      'Pix2StructPreTrainedModel',
      'Pix2StructForConditionalGeneration',
      'Pix2StructVisionModel',
      'Pix2StructTextModel',
  ]

if TYPE_CHECKING:
  from .configuration_pixastruct import (
      PIX2STRUCT_PRETRAINED_CONFIG_ARCHIVE_MAP,
      PixaStructConfig,
      PixaStructTextConfig,
      PixaStructVisionConfig,
  )
  from .processing_pixastruct import PixaStructProcessor

  try:
    if not is_vision_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .image_processing_pixastruct import PixaStructImageProcessor

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_pixastruct import (
        PIX2STRUCT_PRETRAINED_MODEL_ARCHIVE_LIST,
        PixaStructForConditionalGeneration,
        PixaStructPreTrainedModel,
        PixaStructTextModel,
        PixaStructVisionModel,
    )

else:
  import sys

  lowercase_									=   _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)


