






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







class       __lowercase					(nn.Module    ):

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

			UpperCamelCase__				:					str					=					torch.nn.Convad(
			    UpperCAmelCase_  ,					block_out_channels[0]  ,					kernel_size=3  ,					stride=1  ,					padding=1  ,					)

			UpperCamelCase__				:					str					=					None
			UpperCamelCase__				:					List[str]					=					nn.ModuleList([])

			# down
			UpperCamelCase__				:					Union[str, Any]					=					block_out_channels[0]
			for i, down_block_type in enumerate(UpperCAmelCase_):
					UpperCamelCase__				:					str					=					output_channel
					UpperCamelCase__				:					Any					=					block_out_channels[i]
					UpperCamelCase__				:					Any					=					i == len(UpperCAmelCase_) - 1

					UpperCamelCase__				:					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
			UpperCamelCase__				:					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
			UpperCamelCase__				:					Union[str, Any]					=					nn.GroupNorm(num_channels=block_out_channels[-1]  ,					num_groups=UpperCAmelCase_  ,					eps=1e-6)
			UpperCamelCase__				:					Dict					=					nn.SiLU()

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

			UpperCamelCase__				:					Dict					=					False





	def      __UpperCamelCase							(      self   :						Any  ,					UpperCAmelCase_   :						List[str]):
			UpperCamelCase__				:					List[Any]					=					x
			UpperCamelCase__				:					Union[str, Any]					=					self.conv_in(UpperCAmelCase_)

			if self.training and self.gradient_checkpointing:

					def create_custom_forward(UpperCAmelCase_   :						Tuple):
							def custom_forward(*UpperCAmelCase_   :						Optional[Any]):
									return module(*UpperCAmelCase_)

							return custom_forward

					# down
					if is_torch_version('>='  ,					'1.11.0'):
							for down_block in self.down_blocks:
									UpperCamelCase__				:					int					=					torch.utils.checkpoint.checkpoint(
									    create_custom_forward(UpperCAmelCase_)  ,					UpperCAmelCase_  ,					use_reentrant=UpperCAmelCase_)
							# middle
							UpperCamelCase__				:					int					=					torch.utils.checkpoint.checkpoint(
							    create_custom_forward(self.mid_block)  ,					UpperCAmelCase_  ,					use_reentrant=UpperCAmelCase_)
					else:
							for down_block in self.down_blocks:
									UpperCamelCase__				:					Union[str, Any]					=					torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCAmelCase_)  ,					UpperCAmelCase_)
							# middle
							UpperCamelCase__				:					Any					=					torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block)  ,					UpperCAmelCase_)

			else:
					# down
					for down_block in self.down_blocks:
							UpperCamelCase__				:					Optional[Any]					=					down_block(UpperCAmelCase_)

					# middle
					UpperCamelCase__				:					str					=					self.mid_block(UpperCAmelCase_)

			# post-process
			UpperCamelCase__				:					Optional[int]					=					self.conv_norm_out(UpperCAmelCase_)
			UpperCamelCase__				:					List[str]					=					self.conv_act(UpperCAmelCase_)
			UpperCamelCase__				:					Optional[int]					=					self.conv_out(UpperCAmelCase_)

			return sample







class       __lowercase					(nn.Module    ):

	def __init__(      self   :						List[Any]  ,					UpperCAmelCase_   :						str=3  ,					UpperCAmelCase_   :						Tuple=3  ,					UpperCAmelCase_   :						Union[str, Any]=("UpDecoderBlock2D",)  ,					UpperCAmelCase_   :						str=(64,)  ,					UpperCAmelCase_   :						Union[str, Any]=2  ,					UpperCAmelCase_   :						Union[str, Any]=32  ,					UpperCAmelCase_   :						Optional[Any]="silu"  ,					UpperCAmelCase_   :						Any="group"  ,					):
			super().__init__()
			UpperCamelCase__				:					List[str]					=					layers_per_block

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

			UpperCamelCase__				:					Optional[int]					=					None
			UpperCamelCase__				:					Dict					=					nn.ModuleList([])

			UpperCamelCase__				:					Any					=					in_channels if norm_type == 'spatial' else None

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

					UpperCamelCase__				:					Tuple					=					i == len(UpperCAmelCase_) - 1

					UpperCamelCase__				:					List[Any]					=					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_)
					UpperCamelCase__				:					Union[str, Any]					=					output_channel

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

			UpperCamelCase__				:					Optional[int]					=					False





	def      __UpperCamelCase							(      self   :						Tuple  ,					UpperCAmelCase_   :						str  ,					UpperCAmelCase_   :						str=None):
			UpperCamelCase__				:					str					=					z
			UpperCamelCase__				:					str					=					self.conv_in(UpperCAmelCase_)

			UpperCamelCase__				:					List[str]					=					next(iter(self.up_blocks.parameters())).dtype
			if self.training and self.gradient_checkpointing:

					def create_custom_forward(UpperCAmelCase_   :						Optional[int]):
							def custom_forward(*UpperCAmelCase_   :						Tuple):
									return module(*UpperCAmelCase_)

							return custom_forward

					if is_torch_version('>='  ,					'1.11.0'):
							# middle
							UpperCamelCase__				:					Tuple					=					torch.utils.checkpoint.checkpoint(
							    create_custom_forward(self.mid_block)  ,					UpperCAmelCase_  ,					UpperCAmelCase_  ,					use_reentrant=UpperCAmelCase_)
							UpperCamelCase__				:					str					=					sample.to(UpperCAmelCase_)

							# up
							for up_block in self.up_blocks:
									UpperCamelCase__				:					Dict					=					torch.utils.checkpoint.checkpoint(
									    create_custom_forward(UpperCAmelCase_)  ,					UpperCAmelCase_  ,					UpperCAmelCase_  ,					use_reentrant=UpperCAmelCase_)
					else:
							# middle
							UpperCamelCase__				:					Dict					=					torch.utils.checkpoint.checkpoint(
							    create_custom_forward(self.mid_block)  ,					UpperCAmelCase_  ,					UpperCAmelCase_)
							UpperCamelCase__				:					Optional[Any]					=					sample.to(UpperCAmelCase_)

							# up
							for up_block in self.up_blocks:
									UpperCamelCase__				:					Dict					=					torch.utils.checkpoint.checkpoint(create_custom_forward(UpperCAmelCase_)  ,					UpperCAmelCase_  ,					UpperCAmelCase_)
			else:
					# middle
					UpperCamelCase__				:					Optional[Any]					=					self.mid_block(UpperCAmelCase_  ,					UpperCAmelCase_)
					UpperCamelCase__				:					Dict					=					sample.to(UpperCAmelCase_)

					# up
					for up_block in self.up_blocks:
							UpperCamelCase__				:					Dict					=					up_block(UpperCAmelCase_  ,					UpperCAmelCase_)

        # post-process
			if latent_embeds is None:
					UpperCamelCase__				:					str					=					self.conv_norm_out(UpperCAmelCase_)
			else:
					UpperCamelCase__				:					Tuple					=					self.conv_norm_out(UpperCAmelCase_  ,					UpperCAmelCase_)
			UpperCamelCase__				:					List[str]					=					self.conv_act(UpperCAmelCase_)
			UpperCamelCase__				:					List[str]					=					self.conv_out(UpperCAmelCase_)

			return sample







class       __lowercase					(nn.Module    ):

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

			UpperCamelCase__				:					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)

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

			UpperCamelCase__				:					Dict					=					sane_index_shape

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

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

	def      __UpperCamelCase							(      self   :						Dict  ,					UpperCAmelCase_   :						Optional[int]):
			# reshape z -> (batch, height, width, channel) and flatten
			UpperCamelCase__				:					str					=					z.permute(0  ,					2  ,					3  ,					1).contiguous()
			UpperCamelCase__				:					Any					=					z.view(-1  ,					self.vq_embed_dim)

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

			UpperCamelCase__				:					Optional[int]					=					self.embedding(UpperCAmelCase_).view(z.shape)
			UpperCamelCase__				:					int					=					None
			UpperCamelCase__				:					Optional[Any]					=					None

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

			# preserve gradients
			UpperCamelCase__				:					Union[str, Any]					=					z + (z_q - z).detach()

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

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

			if self.sane_index_shape:
					UpperCamelCase__				:					str					=					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      __UpperCamelCase							(      self   :						int  ,					UpperCAmelCase_   :						List[Any]  ,					UpperCAmelCase_   :						Any):
			# shape specifying (batch, height, width, channel)
			if self.remap is not None:
					UpperCamelCase__				:					Any					=					indices.reshape(shape[0]  ,					-1)  # add batch axis
					UpperCamelCase__				:					Dict					=					self.unmap_to_all(UpperCAmelCase_)
					UpperCamelCase__				:					Optional[int]					=					indices.reshape(-1)  # flatten again

			# get quantized latent vectors
			UpperCamelCase__				:					int					=					self.embedding(UpperCAmelCase_)

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

			return z_q







class       __lowercase					(__lowerCamelCase    ):

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

	def      __UpperCamelCase							(      self   :						Union[str, Any]  ,					UpperCAmelCase_   :						Optional[torch.Generator] = None):
			# make sure sample is on the same device as the parameters and has same dtype
			UpperCamelCase__				:					List[Any]					=					randn_tensor(
			    self.mean.shape  ,					generator=UpperCAmelCase_  ,					device=self.parameters.device  ,					dtype=self.parameters.dtype)
			UpperCamelCase__				:					List[str]					=					self.mean + self.std * sample
			return x

	def      __UpperCamelCase							(      self   :						Tuple  ,					UpperCAmelCase_   :						Dict=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      __UpperCamelCase							(      self   :						Dict  ,					UpperCAmelCase_   :						Union[str, Any]  ,					UpperCAmelCase_   :						Optional[int]=[1, 2, 3]):
			if self.deterministic:
					return torch.Tensor([0.0])
			UpperCamelCase__				:					int					=					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      __UpperCamelCase							(      self   :						Optional[int]):
			return self.mean












'''simple docstring'''




import unittest

from transformers import is_torch_available, is_vision_available
from transformers.testing_utils import require_torch, require_vision, slow, torch_device


if is_torch_available():
		import torch

		from transformers import AutoModelForImageClassification

if is_vision_available():
		from transformers import AutoImageProcessor







@require_torch
@require_vision
class       __lowercase					(unittest.TestCase    ):

	@slow
	def      __UpperCamelCase							(      self   :						int):
			UpperCamelCase__				:					Union[str, Any]					=					AutoImageProcessor.from_pretrained('microsoft/dit-base-finetuned-rvlcdip')
			UpperCamelCase__				:					List[str]					=					AutoModelForImageClassification.from_pretrained('microsoft/dit-base-finetuned-rvlcdip')
			model.to(UpperCAmelCase_)

			from datasets import load_dataset

			UpperCamelCase__				:					Optional[Any]					=					load_dataset('nielsr/rvlcdip-demo')

			UpperCamelCase__				:					int					=					dataset['train'][0]['image'].convert('RGB')

			UpperCamelCase__				:					Union[str, Any]					=					image_processor(UpperCAmelCase_  ,					return_tensors='pt').to(UpperCAmelCase_)

			# forward pass
			with torch.no_grad():
					UpperCamelCase__				:					Optional[Any]					=					model(**UpperCAmelCase_)
					UpperCamelCase__				:					Tuple					=					outputs.logits

			UpperCamelCase__				:					str					=					torch.Size((1, 16))
			self.assertEqual(logits.shape  ,					UpperCAmelCase_)

			UpperCamelCase__				:					Tuple					=					torch.tensor(
			    [-0.41_58, -0.40_92, -0.43_47]  ,					device=UpperCAmelCase_  ,					dtype=torch.float  ,					)
			self.assertTrue(torch.allclose(logits[0, :3]  ,					UpperCAmelCase_  ,					atol=1e-4))




