
"""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__       (	__a):
		A_						:       torch.FloatTensor




class  A__       (	nn.Module):

		def __init__(      self       ,		_SCREAMING_SNAKE_CASE=3       ,		_SCREAMING_SNAKE_CASE=3       ,		_SCREAMING_SNAKE_CASE=("DownEncoderBlock2D",)       ,		_SCREAMING_SNAKE_CASE=(64,)       ,		_SCREAMING_SNAKE_CASE=2       ,		_SCREAMING_SNAKE_CASE=32       ,		_SCREAMING_SNAKE_CASE="silu"       ,		_SCREAMING_SNAKE_CASE=True       ,		):
			super().__init__()
			__lowerCAmelCase			:						str									=      layers_per_block

			__lowerCAmelCase			:						Dict									=      torch.nn.Convad(
			    _SCREAMING_SNAKE_CASE       ,		block_out_channels[0]       ,		kernel_size=3       ,		stride=1       ,		padding=1       ,		)

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

			# down
			__lowerCAmelCase			:						int									=      block_out_channels[0]
			for i, down_block_type in enumerate(_SCREAMING_SNAKE_CASE ):
				__lowerCAmelCase			:						Optional[int]									=      output_channel
				__lowerCAmelCase			:						Optional[Any]									=      block_out_channels[i]
				__lowerCAmelCase			:						Any									=      i == len(_SCREAMING_SNAKE_CASE ) - 1

				__lowerCAmelCase			:						Tuple									=      get_down_block(
				    _SCREAMING_SNAKE_CASE       ,		num_layers=self.layers_per_block       ,		in_channels=_SCREAMING_SNAKE_CASE       ,		out_channels=_SCREAMING_SNAKE_CASE       ,		add_downsample=not is_final_block       ,		resnet_eps=1E-6       ,		downsample_padding=0       ,		resnet_act_fn=_SCREAMING_SNAKE_CASE       ,		resnet_groups=_SCREAMING_SNAKE_CASE       ,		attention_head_dim=_SCREAMING_SNAKE_CASE       ,		temb_channels=_SCREAMING_SNAKE_CASE       ,		)
				self.down_blocks.append(_SCREAMING_SNAKE_CASE )

			# mid
			__lowerCAmelCase			:						Dict									=      UNetMidBlockaD(
			    in_channels=block_out_channels[-1]       ,		resnet_eps=1E-6       ,		resnet_act_fn=_SCREAMING_SNAKE_CASE       ,		output_scale_factor=1       ,		resnet_time_scale_shift='default'       ,		attention_head_dim=block_out_channels[-1]       ,		resnet_groups=_SCREAMING_SNAKE_CASE       ,		temb_channels=_SCREAMING_SNAKE_CASE       ,		)

			# out
			__lowerCAmelCase			:						Optional[int]									=      nn.GroupNorm(num_channels=block_out_channels[-1]       ,		num_groups=_SCREAMING_SNAKE_CASE       ,		eps=1E-6 )
			__lowerCAmelCase			:						Tuple									=      nn.SiLU()

			__lowerCAmelCase			:						Optional[Any]									=      2 * out_channels if double_z else out_channels
			__lowerCAmelCase			:						List[Any]									=      nn.Convad(block_out_channels[-1]       ,		_SCREAMING_SNAKE_CASE       ,		3       ,		padding=1 )

			__lowerCAmelCase			:						Optional[Any]									=      False







		def 							__lowerCamelCase							(      self       ,		_SCREAMING_SNAKE_CASE ):
			__lowerCAmelCase			:						Any									=      x
			__lowerCAmelCase			:						Any									=      self.conv_in(_SCREAMING_SNAKE_CASE )

			if self.training and self.gradient_checkpointing:

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

					return custom_forward

				# down
				if is_torch_version('>='       ,		'1.11.0' ):
					for down_block in self.down_blocks:
						__lowerCAmelCase			:						str									=      torch.utils.checkpoint.checkpoint(
						    create_custom_forward(_SCREAMING_SNAKE_CASE )       ,		_SCREAMING_SNAKE_CASE       ,		use_reentrant=_SCREAMING_SNAKE_CASE )
					# middle
					__lowerCAmelCase			:						Any									=      torch.utils.checkpoint.checkpoint(
					    create_custom_forward(self.mid_block )       ,		_SCREAMING_SNAKE_CASE       ,		use_reentrant=_SCREAMING_SNAKE_CASE )
				else:
					for down_block in self.down_blocks:
						__lowerCAmelCase			:						List[str]									=      torch.utils.checkpoint.checkpoint(create_custom_forward(_SCREAMING_SNAKE_CASE )       ,		_SCREAMING_SNAKE_CASE )
					# middle
					__lowerCAmelCase			:						Tuple									=      torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block )       ,		_SCREAMING_SNAKE_CASE )

			else:
				# down
				for down_block in self.down_blocks:
					__lowerCAmelCase			:						Any									=      down_block(_SCREAMING_SNAKE_CASE )

				# middle
				__lowerCAmelCase			:						Tuple									=      self.mid_block(_SCREAMING_SNAKE_CASE )

			# post-process
			__lowerCAmelCase			:						List[Any]									=      self.conv_norm_out(_SCREAMING_SNAKE_CASE )
			__lowerCAmelCase			:						List[str]									=      self.conv_act(_SCREAMING_SNAKE_CASE )
			__lowerCAmelCase			:						Optional[Any]									=      self.conv_out(_SCREAMING_SNAKE_CASE )

			return sample




class  A__       (	nn.Module):

		def __init__(      self       ,		_SCREAMING_SNAKE_CASE=3       ,		_SCREAMING_SNAKE_CASE=3       ,		_SCREAMING_SNAKE_CASE=("UpDecoderBlock2D",)       ,		_SCREAMING_SNAKE_CASE=(64,)       ,		_SCREAMING_SNAKE_CASE=2       ,		_SCREAMING_SNAKE_CASE=32       ,		_SCREAMING_SNAKE_CASE="silu"       ,		_SCREAMING_SNAKE_CASE="group"       ,		):
			super().__init__()
			__lowerCAmelCase			:						List[Any]									=      layers_per_block

			__lowerCAmelCase			:						Any									=      nn.Convad(
			    _SCREAMING_SNAKE_CASE       ,		block_out_channels[-1]       ,		kernel_size=3       ,		stride=1       ,		padding=1       ,		)

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

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

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

			# up
			__lowerCAmelCase			:						int									=      list(reversed(_SCREAMING_SNAKE_CASE ) )
			__lowerCAmelCase			:						Dict									=      reversed_block_out_channels[0]
			for i, up_block_type in enumerate(_SCREAMING_SNAKE_CASE ):
				__lowerCAmelCase			:						int									=      output_channel
				__lowerCAmelCase			:						List[str]									=      reversed_block_out_channels[i]

				__lowerCAmelCase			:						Any									=      i == len(_SCREAMING_SNAKE_CASE ) - 1

				__lowerCAmelCase			:						int									=      get_up_block(
				    _SCREAMING_SNAKE_CASE       ,		num_layers=self.layers_per_block + 1       ,		in_channels=_SCREAMING_SNAKE_CASE       ,		out_channels=_SCREAMING_SNAKE_CASE       ,		prev_output_channel=_SCREAMING_SNAKE_CASE       ,		add_upsample=not is_final_block       ,		resnet_eps=1E-6       ,		resnet_act_fn=_SCREAMING_SNAKE_CASE       ,		resnet_groups=_SCREAMING_SNAKE_CASE       ,		attention_head_dim=_SCREAMING_SNAKE_CASE       ,		temb_channels=_SCREAMING_SNAKE_CASE       ,		resnet_time_scale_shift=_SCREAMING_SNAKE_CASE       ,		)
				self.up_blocks.append(_SCREAMING_SNAKE_CASE )
				__lowerCAmelCase			:						List[str]									=      output_channel

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

			__lowerCAmelCase			:						List[Any]									=      False







		def 							__lowerCamelCase							(      self       ,		_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE=None ):
			__lowerCAmelCase			:						int									=      z
			__lowerCAmelCase			:						Any									=      self.conv_in(_SCREAMING_SNAKE_CASE )

			__lowerCAmelCase			:						Union[str, Any]									=      next(iter(self.up_blocks.parameters() ) ).dtype
			if self.training and self.gradient_checkpointing:

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

					return custom_forward

				if is_torch_version('>='       ,		'1.11.0' ):
					# middle
					__lowerCAmelCase			:						Optional[int]									=      torch.utils.checkpoint.checkpoint(
					    create_custom_forward(self.mid_block )       ,		_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE       ,		use_reentrant=_SCREAMING_SNAKE_CASE )
					__lowerCAmelCase			:						str									=      sample.to(_SCREAMING_SNAKE_CASE )

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

					# up
					for up_block in self.up_blocks:
						__lowerCAmelCase			:						str									=      torch.utils.checkpoint.checkpoint(create_custom_forward(_SCREAMING_SNAKE_CASE )       ,		_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE )
			else:
				# middle
				__lowerCAmelCase			:						List[str]									=      self.mid_block(_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE )
				__lowerCAmelCase			:						Dict									=      sample.to(_SCREAMING_SNAKE_CASE )

				# up
				for up_block in self.up_blocks:
					__lowerCAmelCase			:						Union[str, Any]									=      up_block(_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE )

        # post-process
			if latent_embeds is None:
				__lowerCAmelCase			:						Optional[Any]									=      self.conv_norm_out(_SCREAMING_SNAKE_CASE )
			else:
				__lowerCAmelCase			:						Any									=      self.conv_norm_out(_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE )
			__lowerCAmelCase			:						Dict									=      self.conv_act(_SCREAMING_SNAKE_CASE )
			__lowerCAmelCase			:						int									=      self.conv_out(_SCREAMING_SNAKE_CASE )

			return sample




class  A__       (	nn.Module):

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

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

			__lowerCAmelCase			:						Dict									=      sane_index_shape

		def 							__lowerCamelCase							(      self       ,		_SCREAMING_SNAKE_CASE ):
			__lowerCAmelCase			:						Optional[Any]									=      inds.shape
			assert len(_SCREAMING_SNAKE_CASE ) > 1
			__lowerCAmelCase			:						Tuple									=      inds.reshape(ishape[0]       ,		-1 )
			__lowerCAmelCase			:						str									=      self.used.to(_SCREAMING_SNAKE_CASE )
			__lowerCAmelCase			:						str									=      (inds[:, :, None] == used[None, None, ...]).long()
			__lowerCAmelCase			:						Any									=      match.argmax(-1 )
			__lowerCAmelCase			:						Union[str, Any]									=      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[Any]									=      self.unknown_index
			return new.reshape(_SCREAMING_SNAKE_CASE )

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

		def 							__lowerCamelCase							(      self       ,		_SCREAMING_SNAKE_CASE ):
			__lowerCAmelCase			:						Any									=      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			:						List[Any]									=      torch.argmin(torch.cdist(_SCREAMING_SNAKE_CASE       ,		self.embedding.weight )       ,		dim=1 )

			__lowerCAmelCase			:						Any									=      self.embedding(_SCREAMING_SNAKE_CASE ).view(z.shape )
			__lowerCAmelCase			:						int									=      None
			__lowerCAmelCase			:						Dict									=      None

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

			# preserve gradients
			__lowerCAmelCase			:						Dict									=      z + (z_q - z).detach()

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

			if self.remap is not None:
				__lowerCAmelCase			:						int									=      min_encoding_indices.reshape(z.shape[0]       ,		-1 )  # add batch axis
				__lowerCAmelCase			:						Dict									=      self.remap_to_used(_SCREAMING_SNAKE_CASE )
				__lowerCAmelCase			:						Any									=      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 							__lowerCamelCase							(      self       ,		_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE ):

			if self.remap is not None:
				__lowerCAmelCase			:						List[Any]									=      indices.reshape(shape[0]       ,		-1 )  # add batch axis
				__lowerCAmelCase			:						Dict									=      self.unmap_to_all(_SCREAMING_SNAKE_CASE )
				__lowerCAmelCase			:						List[Any]									=      indices.reshape(-1 )  # flatten again

			# get quantized latent vectors
			__lowerCAmelCase			:						List[Any]									=      self.embedding(_SCREAMING_SNAKE_CASE )

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

			return z_q




class  A__       (	__a):

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

		def 							__lowerCamelCase							(      self       ,		_SCREAMING_SNAKE_CASE = None ):
			__lowerCAmelCase			:						List[str]									=      randn_tensor(
			    self.mean.shape       ,		generator=_SCREAMING_SNAKE_CASE       ,		device=self.parameters.device       ,		dtype=self.parameters.dtype )
			__lowerCAmelCase			:						Union[str, Any]									=      self.mean + self.std * sample
			return x

		def 							__lowerCamelCase							(      self       ,		_SCREAMING_SNAKE_CASE=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       ,		_SCREAMING_SNAKE_CASE       ,		_SCREAMING_SNAKE_CASE=[1, 2, 3] ):

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







		def 							__lowerCamelCase							(      self ):
			return self.mean





'''simple docstring'''







from __future__ import annotations





def 						snake_case_							(_a	:							Dict    ,   _a	:							str    ,   _a	:							Optional[Any]    ,   _a	:							List[str]     ):  # noqa: E741
    while r - l > 1:
        UpperCAmelCase			       =		(l + r) // 2
        if v[m] >= key:
            UpperCAmelCase			       =		m
        else:
            UpperCAmelCase			       =		m  # noqa: E741
    return r





def 						snake_case_							(_a	:							list[int]     ):
    if len(_a     ) == 0:
        return 0

    UpperCAmelCase			       =		[0] * len(_a     )
    UpperCAmelCase			       =		1

    UpperCAmelCase			       =		v[0]

    for i in range(1    ,   len(_a     )     ):
        if v[i] < tail[0]:
            UpperCAmelCase			       =		v[i]
        elif v[i] > tail[length - 1]:
            UpperCAmelCase			       =		v[i]
            length += 1
        else:
            UpperCAmelCase			       =		v[i]

    return length


if __name__ == "__main__":
   import doctest

   doctest.testmod()




