




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







class 						a				(   nn.Module     ):


				def __init__(		self    :					int       ,				lowercase_    :					List[Any]=3       ,				lowercase_    :					str=3       ,				lowercase_    :					Union[str, Any]=("DownEncoderBlock2D",)       ,				lowercase_    :					str=(64,)       ,				lowercase_    :					Optional[Any]=2       ,				lowercase_    :					Optional[Any]=32       ,				lowercase_    :					Optional[int]="silu"       ,				lowercase_    :					Optional[int]=True       ,				):
									super().__init__()
									snake_case_									=				layers_per_block

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

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

									# mid
									snake_case_									=				UNetMidBlockaD(
									    in_channels=block_out_channels[-1]       ,				resnet_eps=1e-6       ,				resnet_act_fn=lowercase_       ,				output_scale_factor=1       ,				resnet_time_scale_shift='''default'''       ,				attention_head_dim=block_out_channels[-1]       ,				resnet_groups=lowercase_       ,				temb_channels=lowercase_       ,				)

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

									snake_case_									=				False


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Tuple					):
									snake_case_									=				x
									snake_case_									=				self.conv_in(lowercase_					)

									if self.training and self.gradient_checkpointing:

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

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

									else:
														# down
														for down_block in self.down_blocks:
																			snake_case_									=				down_block(lowercase_					)

														# middle
														snake_case_									=				self.mid_block(lowercase_					)

									# post-process
									snake_case_									=				self.conv_norm_out(lowercase_					)
									snake_case_									=				self.conv_act(lowercase_					)
									snake_case_									=				self.conv_out(lowercase_					)

									return sample







class 						a				(   nn.Module     ):


				def __init__(		self    :					Any       ,				lowercase_    :					Tuple=3       ,				lowercase_    :					int=3       ,				lowercase_    :					Dict=("UpDecoderBlock2D",)       ,				lowercase_    :					Tuple=(64,)       ,				lowercase_    :					Tuple=2       ,				lowercase_    :					Union[str, Any]=32       ,				lowercase_    :					str="silu"       ,				lowercase_    :					List[str]="group"       ,				):
									super().__init__()
									snake_case_									=				layers_per_block

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

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

														snake_case_									=				i == len(lowercase_					) - 1

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

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

									snake_case_									=				False


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					Optional[int]=None					):
									snake_case_									=				z
									snake_case_									=				self.conv_in(lowercase_					)

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

														def create_custom_forward(lowercase_    :					int					):
																			def custom_forward(*lowercase_    :					Any					):
																								return module(*lowercase_					)

																			return custom_forward

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

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

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

														# up
														for up_block in self.up_blocks:
																			snake_case_									=				up_block(lowercase_       ,				lowercase_					)

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

									return sample







class 						a				(   nn.Module     ):


				def __init__(		self    :					Dict       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					List[Any]       ,				lowercase_    :					str       ,				lowercase_    :					Any=None       ,				lowercase_    :					Optional[int]="random"       ,				lowercase_    :					Optional[int]=False       ,				lowercase_    :					int=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   A_ (		self    :					str       ,				lowercase_    :					List[str]					):
									snake_case_									=				inds.shape
									assert len(lowercase_					) > 1
									snake_case_									=				inds.reshape(ishape[0]       ,				-1					)
									snake_case_									=				self.used.to(lowercase_					)
									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(lowercase_					)


				def   A_ (		self    :					int       ,				lowercase_    :					Tuple					):
									snake_case_									=				inds.shape
									assert len(lowercase_					) > 1
									snake_case_									=				inds.reshape(ishape[0]       ,				-1					)
									snake_case_									=				self.used.to(lowercase_					)
									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       ,				lowercase_					)
									return back.reshape(lowercase_					)


				def   A_ (		self    :					str       ,				lowercase_    :					Any					):
									# reshape z -> (batch, height, width, channel) and flatten
									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(lowercase_       ,				self.embedding.weight					)       ,				dim=1					)

									snake_case_									=				self.embedding(lowercase_					).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(lowercase_					)
														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   A_ (		self    :					Optional[int]       ,				lowercase_    :					Tuple       ,				lowercase_    :					Optional[Any]					):
									# shape specifying (batch, height, width, channel)
									if self.remap is not None:
														snake_case_									=				indices.reshape(shape[0]       ,				-1					)  # add batch axis
														snake_case_									=				self.unmap_to_all(lowercase_					)
														snake_case_									=				indices.reshape(-1					)  # flatten again

									# get quantized latent vectors
									snake_case_									=				self.embedding(lowercase_					)

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

									return z_q







class 						a				(   _lowerCamelCase     ):


				def __init__(		self    :					Tuple       ,				lowercase_    :					List[str]       ,				lowercase_    :					Any=False					):
									snake_case_									=				parameters
									snake_case_			,snake_case_									=				torch.chunk(lowercase_       ,				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   A_ (		self    :					List[Any]       ,				lowercase_    :					Optional[torch.Generator] = None					):
									# make sure sample is on the same device as the parameters and has same dtype
									snake_case_									=				randn_tensor(
									    self.mean.shape       ,				generator=lowercase_       ,				device=self.parameters.device       ,				dtype=self.parameters.dtype					)
									snake_case_									=				self.mean + self.std * sample
									return x


				def   A_ (		self    :					Dict       ,				lowercase_    :					Any=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   A_ (		self    :					List[Any]       ,				lowercase_    :					List[Any]       ,				lowercase_    :					Union[str, 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=lowercase_					)


				def   A_ (		self    :					Dict					):
									return self.mean







'''simple docstring'''

from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_tokenizers_available,
    is_torch_available,
)


a  :				Tuple									=  {
    'configuration_llama': ['LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP', 'LlamaConfig'],
}

try:
							if not is_sentencepiece_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							a  :				Optional[Any]									=  ['LlamaTokenizer']

try:
							if not is_tokenizers_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							a  :				str									=  ['LlamaTokenizerFast']

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							a  :				Optional[Any]									=  [
							    'LlamaForCausalLM',
							    'LlamaModel',
							    'LlamaPreTrainedModel',
							    'LlamaForSequenceClassification',
							]


if TYPE_CHECKING:
							from .configuration_llama import LLAMA_PRETRAINED_CONFIG_ARCHIVE_MAP, LlamaConfig

							try:
														if not is_sentencepiece_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .tokenization_llama import LlamaTokenizer

							try:
														if not is_tokenizers_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .tokenization_llama_fast import LlamaTokenizerFast

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_llama import LlamaForCausalLM, LlamaForSequenceClassification, LlamaModel, LlamaPreTrainedModel


else:
							import sys

							a  :				str									=  _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)

