


"""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 			A__	(						__lowercase):


     """simple docstring"""
     snake_case__	:					torch.FloatTensor


class 			A__	(						nn.Module):


     """simple docstring"""

     def __init__(	self:						Dict     ,		__a:						Optional[Any]=3     ,		__a:						Any=3     ,		__a:						List[str]=("DownEncoderBlock2D",)     ,		__a:						List[str]=(64,)     ,		__a:						Union[str, Any]=2     ,		__a:						Union[str, Any]=32     ,		__a:						Union[str, Any]="silu"     ,		__a:						List[str]=True     ,		)->    Dict:
           super().__init__()
           lowerCamelCase					:					Dict        =	layers_per_block

           lowerCamelCase					:					List[Any]        =	torch.nn.Convad(
               __a     ,		block_out_channels[0]     ,		kernel_size=3     ,		stride=1     ,		padding=1     ,		)

           lowerCamelCase					:					str        =	None
           lowerCamelCase					:					str        =	nn.ModuleList([]	)

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

                 lowerCamelCase					:					str        =	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
           lowerCamelCase					:					int        =	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
           lowerCamelCase					:					Tuple        =	nn.GroupNorm(num_channels=block_out_channels[-1]     ,		num_groups=__a     ,		eps=1e-6	)
           lowerCamelCase					:					int        =	nn.SiLU()

           lowerCamelCase					:					Union[str, Any]        =	2 * out_channels if double_z else out_channels
           lowerCamelCase					:					List[str]        =	nn.Convad(block_out_channels[-1]     ,		__a     ,		3     ,		padding=1	)

           lowerCamelCase					:					str        =	False







     def 		a__						(	self:						List[Any]     ,		__a:						Dict	)->    Union[str, Any]:
           lowerCamelCase					:					Tuple        =	x
           lowerCamelCase					:					Optional[int]        =	self.conv_in(__a	)

           if self.training and self.gradient_checkpointing:

                 def create_custom_forward(__a:						Tuple	):
                       def custom_forward(*__a:						int	):
                             return module(*__a	)

                       return custom_forward

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

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

                 # middle
                 lowerCamelCase					:					Optional[int]        =	self.mid_block(__a	)

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

           return sample


class 			A__	(						nn.Module):


     """simple docstring"""

     def __init__(	self:						Union[str, Any]     ,		__a:						Dict=3     ,		__a:						Dict=3     ,		__a:						Union[str, Any]=("UpDecoderBlock2D",)     ,		__a:						List[Any]=(64,)     ,		__a:						List[Any]=2     ,		__a:						List[str]=32     ,		__a:						Dict="silu"     ,		__a:						int="group"     ,		)->    Union[str, Any]:
           super().__init__()
           lowerCamelCase					:					Dict        =	layers_per_block

           lowerCamelCase					:					List[str]        =	nn.Convad(
               __a     ,		block_out_channels[-1]     ,		kernel_size=3     ,		stride=1     ,		padding=1     ,		)

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

           lowerCamelCase					:					Any        =	in_channels if norm_type == """spatial""" else None

           # mid
           lowerCamelCase					:					List[str]        =	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
           lowerCamelCase					:					str        =	list(reversed(__a	)	)
           lowerCamelCase					:					List[Any]        =	reversed_block_out_channels[0]
           for i, up_block_type in enumerate(__a	):
                 lowerCamelCase					:					Optional[Any]        =	output_channel
                 lowerCamelCase					:					str        =	reversed_block_out_channels[i]

                 lowerCamelCase					:					int        =	i == len(__a	) - 1

                 lowerCamelCase					:					str        =	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	)
                 lowerCamelCase					:					Tuple        =	output_channel

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

           lowerCamelCase					:					Union[str, Any]        =	False







     def 		a__						(	self:						List[Any]     ,		__a:						List[Any]     ,		__a:						List[str]=None	)->    List[Any]:
           lowerCamelCase					:					List[str]        =	z
           lowerCamelCase					:					str        =	self.conv_in(__a	)

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

                 def create_custom_forward(__a:						str	):
                       def custom_forward(*__a:						List[str]	):
                             return module(*__a	)

                       return custom_forward

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

                       # up
                       for up_block in self.up_blocks:
                             lowerCamelCase					:					List[Any]        =	torch.utils.checkpoint.checkpoint(
                                 create_custom_forward(__a	)     ,		__a     ,		__a     ,		use_reentrant=__a	)
                 else:
                       # middle
                       lowerCamelCase					:					Optional[int]        =	torch.utils.checkpoint.checkpoint(
                           create_custom_forward(self.mid_block	)     ,		__a     ,		__a	)
                       lowerCamelCase					:					Any        =	sample.to(__a	)

                       # up
                       for up_block in self.up_blocks:
                             lowerCamelCase					:					List[str]        =	torch.utils.checkpoint.checkpoint(create_custom_forward(__a	)     ,		__a     ,		__a	)
           else:
                 # middle
                 lowerCamelCase					:					Tuple        =	self.mid_block(__a     ,		__a	)
                 lowerCamelCase					:					Union[str, Any]        =	sample.to(__a	)

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

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

           return sample


class 			A__	(						nn.Module):


     """simple docstring"""

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

           lowerCamelCase					:					List[Any]        =	nn.Embedding(self.n_e     ,		self.vq_embed_dim	)
           self.embedding.weight.data.uniform_(-1.0 / self.n_e     ,		1.0 / self.n_e	)

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

           lowerCamelCase					:					List[Any]        =	sane_index_shape

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

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

     def 		a__						(	self:						Any     ,		__a:						Optional[int]	)->    Dict:
           # reshape z -> (batch, height, width, channel) and flatten
           lowerCamelCase					:					Dict        =	z.permute(0     ,		2     ,		3     ,		1	).contiguous()
           lowerCamelCase					:					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
           lowerCamelCase					:					Union[str, Any]        =	torch.argmin(torch.cdist(__a     ,		self.embedding.weight	)     ,		dim=1	)

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

           # compute loss for embedding
           if not self.legacy:
                 lowerCamelCase					:					Optional[int]        =	self.beta * torch.mean((z_q.detach() - z) ** 2	) + torch.mean((z_q - z.detach()) ** 2	)
           else:
                 lowerCamelCase					:					Tuple        =	torch.mean((z_q.detach() - z) ** 2	) + self.beta * torch.mean((z_q - z.detach()) ** 2	)

           # preserve gradients
           lowerCamelCase					:					List[str]        =	z + (z_q - z).detach()

           # reshape back to match original input shape
           lowerCamelCase					:					List[Any]        =	z_q.permute(0     ,		3     ,		1     ,		2	).contiguous()

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

           if self.sane_index_shape:
                 lowerCamelCase					:					Tuple        =	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 		a__						(	self:						Tuple     ,		__a:						List[str]     ,		__a:						List[str]	)->    Union[str, Any]:
           # shape specifying (batch, height, width, channel)
           if self.remap is not None:
                 lowerCamelCase					:					List[Any]        =	indices.reshape(shape[0]     ,		-1	)  # add batch axis
                 lowerCamelCase					:					str        =	self.unmap_to_all(__a	)
                 lowerCamelCase					:					List[str]        =	indices.reshape(-1	)  # flatten again

           # get quantized latent vectors
           lowerCamelCase					:					Dict        =	self.embedding(__a	)

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

           return z_q


class 			A__	(						__lowercase):


     """simple docstring"""

     def __init__(	self:						Dict     ,		__a:						Optional[Any]     ,		__a:						List[Any]=False	)->    str:
           lowerCamelCase					:					Optional[Any]        =	parameters
           lowerCamelCase     ,			lowerCamelCase					:					List[Any]        =	torch.chunk(__a     ,		2     ,		dim=1	)
           lowerCamelCase					:					Optional[int]        =	torch.clamp(self.logvar     ,		-30.0     ,		20.0	)
           lowerCamelCase					:					Dict        =	deterministic
           lowerCamelCase					:					Tuple        =	torch.exp(0.5 * self.logvar	)
           lowerCamelCase					:					Union[str, Any]        =	torch.exp(self.logvar	)
           if self.deterministic:
                 lowerCamelCase					:					Dict        =	torch.zeros_like(
                     self.mean     ,		device=self.parameters.device     ,		dtype=self.parameters.dtype	)

     def 		a__						(	self:						Any     ,		__a:						Optional[torch.Generator] = None	)->    torch.FloatTensor:
           # make sure sample is on the same device as the parameters and has same dtype
           lowerCamelCase					:					Optional[Any]        =	randn_tensor(
               self.mean.shape     ,		generator=__a     ,		device=self.parameters.device     ,		dtype=self.parameters.dtype	)
           lowerCamelCase					:					Optional[Any]        =	self.mean + self.std * sample
           return x

     def 		a__						(	self:						Optional[Any]     ,		__a:						Tuple=None	)->    Optional[Any]:
           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 		a__						(	self:						Optional[Any]     ,		__a:						Any     ,		__a:						List[Any]=[1, 2, 3]	)->    str:
           if self.deterministic:
                 return torch.Tensor([0.0]	)
           lowerCamelCase					:					List[str]        =	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 		a__						(	self:						List[Any]	)->    Dict:
           return self.mean








"""simple docstring"""


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


__lowerCamelCase      :Tuple	    =		logging.get_logger(__name__)

__lowerCamelCase      :Any	    =		{
    'vinvino02/glpn-kitti': 'https://huggingface.co/vinvino02/glpn-kitti/resolve/main/config.json',
    # See all GLPN models at https://huggingface.co/models?filter=glpn
}


class 			A__	(						__lowercase):


     """simple docstring"""
     snake_case__	:					Tuple													='''glpn'''

     def __init__(	self:						Dict     ,		__a:						List[str]=3     ,		__a:						Optional[int]=4     ,		__a:						Dict=[2, 2, 2, 2]     ,		__a:						str=[8, 4, 2, 1]     ,		__a:						Optional[int]=[32, 64, 160, 256]     ,		__a:						Dict=[7, 3, 3, 3]     ,		__a:						Dict=[4, 2, 2, 2]     ,		__a:						Optional[Any]=[1, 2, 5, 8]     ,		__a:						Tuple=[4, 4, 4, 4]     ,		__a:						int="gelu"     ,		__a:						Union[str, Any]=0.0     ,		__a:						str=0.0     ,		__a:						Union[str, Any]=0.02     ,		__a:						str=0.1     ,		__a:						Union[str, Any]=1e-6     ,		__a:						Any=64     ,		__a:						Dict=10     ,		__a:						Union[str, Any]=-1     ,		**__a:						Optional[Any]     ,		)->    Dict:
           super().__init__(**__a	)

           lowerCamelCase					:					Dict        =	num_channels
           lowerCamelCase					:					Any        =	num_encoder_blocks
           lowerCamelCase					:					Dict        =	depths
           lowerCamelCase					:					List[str]        =	sr_ratios
           lowerCamelCase					:					Dict        =	hidden_sizes
           lowerCamelCase					:					Tuple        =	patch_sizes
           lowerCamelCase					:					Optional[int]        =	strides
           lowerCamelCase					:					Optional[Any]        =	mlp_ratios
           lowerCamelCase					:					Union[str, Any]        =	num_attention_heads
           lowerCamelCase					:					List[str]        =	hidden_act
           lowerCamelCase					:					Any        =	hidden_dropout_prob
           lowerCamelCase					:					Optional[int]        =	attention_probs_dropout_prob
           lowerCamelCase					:					List[Any]        =	initializer_range
           lowerCamelCase					:					Dict        =	drop_path_rate
           lowerCamelCase					:					Any        =	layer_norm_eps
           lowerCamelCase					:					Optional[Any]        =	decoder_hidden_size
           lowerCamelCase					:					Tuple        =	max_depth
           lowerCamelCase					:					Optional[Any]        =	head_in_index




