
'''simple docstring'''







from __future__ import annotations







def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A			,				__A			,				__A	)       ->	Tuple:  # noqa: E741
	while r - l > 1:
		_snake_case   						=		(l + r) // 2
		if v[m] >= key:
			_snake_case   						=		m
		else:
			_snake_case   						=		m  # noqa: E741
	return r







def 				SCREAMING_SNAKE_CASE__	(   __A	)       ->	int:
	if len(__A	) == 0:
		return 0

	_snake_case   						=		[0] * len(__A	)
	_snake_case   						=		1

	_snake_case   						=		v[0]

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

	return length


if __name__ == "__main__":
					import doctest

					doctest.testmod()



'''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 						__UpperCAmelCase      (					_lowerCamelCase						):
							__lowercase      		=       42


class 						__UpperCAmelCase      (					nn.Module						):


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





								"""simple docstring"""






								super().__init__()
								_snake_case   						=		layers_per_block

								_snake_case   						=		torch.nn.Convad(
								    lowerCAmelCase_    ,   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(lowerCAmelCase_  ):
									_snake_case   						=		output_channel
									_snake_case   						=		block_out_channels[i]
									_snake_case   						=		i == len(lowerCAmelCase_  ) - 1

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

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

								# out
								_snake_case   						=		nn.GroupNorm(num_channels=block_out_channels[-1]    ,   num_groups=lowerCAmelCase_    ,   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]    ,   lowerCAmelCase_    ,   3    ,   padding=1  )

								_snake_case   						=		False




							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		x
								_snake_case   						=		self.conv_in(lowerCAmelCase_  )

								if self.training and self.gradient_checkpointing:

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

										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(lowerCAmelCase_  )    ,   lowerCAmelCase_    ,   use_reentrant=lowerCAmelCase_  )
										# middle
										_snake_case   						=		torch.utils.checkpoint.checkpoint(
										    create_custom_forward(self.mid_block  )    ,   lowerCAmelCase_    ,   use_reentrant=lowerCAmelCase_  )
									else:
										for down_block in self.down_blocks:
											_snake_case   						=		torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCAmelCase_  )    ,   lowerCAmelCase_  )
										# middle
										_snake_case   						=		torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block  )    ,   lowerCAmelCase_  )

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

									# middle
									_snake_case   						=		self.mid_block(lowerCAmelCase_  )

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

								return sample


class 						__UpperCAmelCase      (					nn.Module						):


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





								"""simple docstring"""






								super().__init__()
								_snake_case   						=		layers_per_block

								_snake_case   						=		nn.Convad(
								    lowerCAmelCase_    ,   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=lowerCAmelCase_    ,   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=lowerCAmelCase_    ,   temb_channels=lowerCAmelCase_    ,   )

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

									_snake_case   						=		i == len(lowerCAmelCase_  ) - 1

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

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

								_snake_case   						=		False




							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_=None  ):





								"""simple docstring"""






								_snake_case   						=		z
								_snake_case   						=		self.conv_in(lowerCAmelCase_  )

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

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

										return custom_forward

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

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

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

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

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

								return sample


class 						__UpperCAmelCase      (					nn.Module						):


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





								"""simple docstring"""






								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  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		inds.shape
								assert len(lowerCAmelCase_  ) > 1
								_snake_case   						=		inds.reshape(ishape[0]    ,   -1  )
								_snake_case   						=		self.used.to(lowerCAmelCase_  )
								_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(lowerCAmelCase_  )


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		inds.shape
								assert len(lowerCAmelCase_  ) > 1
								_snake_case   						=		inds.reshape(ishape[0]    ,   -1  )
								_snake_case   						=		self.used.to(lowerCAmelCase_  )
								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    ,   lowerCAmelCase_  )
								return back.reshape(lowerCAmelCase_  )


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_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(lowerCAmelCase_    ,   self.embedding.weight  )    ,   dim=1  )

								_snake_case   						=		self.embedding(lowerCAmelCase_  ).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(lowerCAmelCase_  )
									_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  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_  ):





								"""simple docstring"""






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

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

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

								return z_q


class 						__UpperCAmelCase      (					_lowerCamelCase						):


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





								"""simple docstring"""






								_snake_case   						=		parameters
								_snake_case						, _snake_case   						=		torch.chunk(lowerCAmelCase_    ,   2    ,   dim=1  )
								_snake_case   						=		torch.clamp(self.logvar    ,   -30.0    ,   20.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  lowerCamelCase				(       self    ,   lowerCAmelCase_ = None  ):





								"""simple docstring"""






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


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





								"""simple docstring"""






								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=lowerCAmelCase_  )




							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								return self.mean

