
"""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       lowerCAmelCase__				(						UpperCamelCase__	):
     __a						=					42






class       lowerCAmelCase__				(						nn.Module	):







     def __init__(							self			:			int    ,      _lowerCamelCase			:			int=3    ,      _lowerCamelCase			:			Optional[Any]=3    ,      _lowerCamelCase			:			Union[str, Any]=("DownEncoderBlock2D",)    ,      _lowerCamelCase			:			Optional[int]=(64,)    ,      _lowerCamelCase			:			Tuple=2    ,      _lowerCamelCase			:			Dict=32    ,      _lowerCamelCase			:			Any="silu"    ,      _lowerCamelCase			:			Optional[int]=True    ,      ):
       super().__init__()
       _snake_case					     =       layers_per_block

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

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

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

         _snake_case					     =       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
       _snake_case					     =       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
       _snake_case					     =       nn.GroupNorm(num_channels=block_out_channels[-1]    ,      num_groups=UpperCamelCase_    ,      eps=1e-6			)
       _snake_case					     =       nn.SiLU()

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

       _snake_case					     =       False







     def      lowercase	(							self			:			Optional[Any]    ,      _lowerCamelCase			:			str			):
       _snake_case					     =       x
       _snake_case					     =       self.conv_in(UpperCamelCase_			)

       if self.training and self.gradient_checkpointing:

         def create_custom_forward(_lowerCamelCase			:			Dict			):
           def custom_forward(*_lowerCamelCase			:			List[str]			):
             return module(*UpperCamelCase_			)

           return custom_forward

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

       else:
         # down
         for down_block in self.down_blocks:
           _snake_case					     =       down_block(UpperCamelCase_			)

         # middle
         _snake_case					     =       self.mid_block(UpperCamelCase_			)

       # post-process
       _snake_case					     =       self.conv_norm_out(UpperCamelCase_			)
       _snake_case					     =       self.conv_act(UpperCamelCase_			)
       _snake_case					     =       self.conv_out(UpperCamelCase_			)

       return sample






class       lowerCAmelCase__				(						nn.Module	):







     def __init__(							self			:			Dict    ,      _lowerCamelCase			:			int=3    ,      _lowerCamelCase			:			Optional[Any]=3    ,      _lowerCamelCase			:			int=("UpDecoderBlock2D",)    ,      _lowerCamelCase			:			Any=(64,)    ,      _lowerCamelCase			:			Optional[Any]=2    ,      _lowerCamelCase			:			List[Any]=32    ,      _lowerCamelCase			:			Optional[Any]="silu"    ,      _lowerCamelCase			:			Dict="group"    ,      ):
       super().__init__()
       _snake_case					     =       layers_per_block

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

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

       _snake_case					     =       in_channels if norm_type == '''spatial''' else None

       # mid
       _snake_case					     =       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
       _snake_case					     =       list(reversed(UpperCamelCase_			)			)
       _snake_case					     =       reversed_block_out_channels[0]
       for i, up_block_type in enumerate(UpperCamelCase_			):
         _snake_case					     =       output_channel
         _snake_case					     =       reversed_block_out_channels[i]

         _snake_case					     =       i == len(UpperCamelCase_			) - 1

         _snake_case					     =       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_			)
         _snake_case					     =       output_channel

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

       _snake_case					     =       False







     def      lowercase	(							self			:			Optional[Any]    ,      _lowerCamelCase			:			Union[str, Any]    ,      _lowerCamelCase			:			List[str]=None			):
       _snake_case					     =       z
       _snake_case					     =       self.conv_in(UpperCamelCase_			)

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

         def create_custom_forward(_lowerCamelCase			:			List[str]			):
           def custom_forward(*_lowerCamelCase			:			int			):
             return module(*UpperCamelCase_			)

           return custom_forward

         if is_torch_version('''>='''    ,      '''1.11.0'''			):
           # middle
           _snake_case					     =       torch.utils.checkpoint.checkpoint(
               create_custom_forward(self.mid_block			)    ,      UpperCamelCase_    ,      UpperCamelCase_    ,      use_reentrant=UpperCamelCase_			)
           _snake_case					     =       sample.to(UpperCamelCase_			)

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

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

         # up
         for up_block in self.up_blocks:
           _snake_case					     =       up_block(UpperCamelCase_    ,      UpperCamelCase_			)

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

       return sample






class       lowerCAmelCase__				(						nn.Module	):







     def __init__(							self			:			Tuple    ,      _lowerCamelCase			:			List[str]    ,      _lowerCamelCase			:			Optional[Any]    ,      _lowerCamelCase			:			Optional[Any]    ,      _lowerCamelCase			:			Optional[int]=None    ,      _lowerCamelCase			:			List[str]="random"    ,      _lowerCamelCase			:			Union[str, Any]=False    ,      _lowerCamelCase			:			Dict=True			):
       super().__init__()
       _snake_case					     =       n_e
       _snake_case					     =       vq_embed_dim
       _snake_case					     =       beta
       _snake_case					     =       legacy

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

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

       _snake_case					     =       sane_index_shape







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







     def      lowercase	(							self			:			Union[str, Any]    ,      _lowerCamelCase			:			Tuple			):
       _snake_case					     =       inds.shape
       assert len(UpperCamelCase_			) > 1
       _snake_case					     =       inds.reshape(ishape[0]    ,      -1			)
       _snake_case					     =       self.used.to(UpperCamelCase_			)
       if self.re_embed > self.used.shape[0]:  # extra token
         _snake_case					     =       0  # simply set to zero
       _snake_case					     =       torch.gather(used[None, :][inds.shape[0] * [0], :]    ,      1    ,      UpperCamelCase_			)
       return back.reshape(UpperCamelCase_			)







     def      lowercase	(							self			:			List[Any]    ,      _lowerCamelCase			:			Any			):
       _snake_case					     =       z.permute(0    ,      2    ,      3    ,      1			).contiguous()
       _snake_case					     =       z.view(-1    ,      self.vq_embed_dim			)

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

       _snake_case					     =       self.embedding(UpperCamelCase_			).view(z.shape			)
       _snake_case					     =       None
       _snake_case					     =       None

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

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

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

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

       if self.sane_index_shape:
         _snake_case					     =       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			:			List[Any]    ,      _lowerCamelCase			:			Tuple    ,      _lowerCamelCase			:			List[Any]			):







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

       # get quantized latent vectors
       _snake_case					     =       self.embedding(UpperCamelCase_			)

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

       return z_q






class       lowerCAmelCase__				(						UpperCamelCase__	):







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







     def      lowercase	(							self			:			Any    ,      _lowerCamelCase			:			Optional[torch.Generator] = None			):
       _snake_case					     =       randn_tensor(
           self.mean.shape    ,      generator=UpperCamelCase_    ,      device=self.parameters.device    ,      dtype=self.parameters.dtype			)
       _snake_case					     =       self.mean + self.std * sample
       return x







     def      lowercase	(							self			:			List[str]    ,      _lowerCamelCase			:			str=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      lowercase	(							self			:			int    ,      _lowerCamelCase			:			Any    ,      _lowerCamelCase			:			Any=[1, 2, 3]			):







       if self.deterministic:
         return torch.Tensor([0.0]			)
       _snake_case					     =       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      lowercase	(							self			:			str			):
       return self.mean







from typing import TYPE_CHECKING

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


lowerCAmelCase       =							{'configuration_dpt': ['DPT_PRETRAINED_CONFIG_ARCHIVE_MAP', 'DPTConfig']}

try:
			if not is_vision_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowerCAmelCase       =							['DPTFeatureExtractor']
			lowerCAmelCase       =							['DPTImageProcessor']

try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			lowerCAmelCase       =							[
			    'DPT_PRETRAINED_MODEL_ARCHIVE_LIST',
			    'DPTForDepthEstimation',
			    'DPTForSemanticSegmentation',
			    'DPTModel',
			    'DPTPreTrainedModel',
			]


if TYPE_CHECKING:
			from .configuration_dpt import DPT_PRETRAINED_CONFIG_ARCHIVE_MAP, DPTConfig

			try:
						if not is_vision_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .feature_extraction_dpt import DPTFeatureExtractor
						from .image_processing_dpt import DPTImageProcessor

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_dpt import (
						    DPT_PRETRAINED_MODEL_ARCHIVE_LIST,
						    DPTForDepthEstimation,
						    DPTForSemanticSegmentation,
						    DPTModel,
						    DPTPreTrainedModel,
						)


else:
			import sys

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


