

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       ( SCREAMING_SNAKE_CASE__    ):
				__SCREAMING_SNAKE_CASE      :				Union[str, Any]     =    42







class 				lowercase       ( nn.Module    ):




				def __init__(      self   ,     snake_case=3   ,     snake_case=3   ,     snake_case=("DownEncoderBlock2D",)   ,     snake_case=(64,)   ,     snake_case=2   ,     snake_case=32   ,     snake_case="silu"   ,     snake_case=True   ,     ):
											super().__init__()
											snake_case_           =					layers_per_block

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

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

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

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

											snake_case_           =					False




				def 					a							(      self   ,     snake_case ):
											snake_case_           =					x
											snake_case_           =					self.conv_in(snake_case__ )

											if self.training and self.gradient_checkpointing:

																		def create_custom_forward(snake_case ):
																									def custom_forward(*snake_case ):
																																return module(*snake_case__ )

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

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

																		# middle
																		snake_case_           =					self.mid_block(snake_case__ )

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

											return sample







class 				lowercase       ( nn.Module    ):




				def __init__(      self   ,     snake_case=3   ,     snake_case=3   ,     snake_case=("UpDecoderBlock2D",)   ,     snake_case=(64,)   ,     snake_case=2   ,     snake_case=32   ,     snake_case="silu"   ,     snake_case="group"   ,     ):
											super().__init__()
											snake_case_           =					layers_per_block

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

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

																		snake_case_           =					i == len(snake_case__ ) - 1

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

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

											snake_case_           =					False




				def 					a							(      self   ,     snake_case   ,     snake_case=None ):
											snake_case_           =					z
											snake_case_           =					self.conv_in(snake_case__ )

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

																		def create_custom_forward(snake_case ):
																									def custom_forward(*snake_case ):
																																return module(*snake_case__ )

																									return custom_forward

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

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

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

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

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

											return sample







class 				lowercase       ( nn.Module    ):




				def __init__(      self   ,     snake_case   ,     snake_case   ,     snake_case   ,     snake_case=None   ,     snake_case="random"   ,     snake_case=False   ,     snake_case=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   ,     snake_case ):
											snake_case_           =					inds.shape
											assert len(snake_case__ ) > 1
											snake_case_           =					inds.reshape(ishape[0]   ,     -1 )
											snake_case_           =					self.used.to(snake_case__ )
											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(snake_case__ )




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




				def 					a							(      self   ,     snake_case ):
											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(snake_case__   ,     self.embedding.weight )   ,     dim=1 )

											snake_case_           =					self.embedding(snake_case__ ).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(snake_case__ )
																		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   ,     snake_case   ,     snake_case ):
											# 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(snake_case__ )
																		snake_case_           =					indices.reshape(-1 )  # flatten again

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

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

											return z_q







class 				lowercase       ( SCREAMING_SNAKE_CASE__    ):




				def __init__(      self   ,     snake_case   ,     snake_case=False ):
											snake_case_           =					parameters
											snake_case_           =					torch.chunk(snake_case__   ,     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_           =					torch.zeros_like(
																		    self.mean   ,     device=self.parameters.device   ,     dtype=self.parameters.dtype )




				def 					a							(      self   ,     snake_case = None ):
											snake_case_           =					randn_tensor(
											    self.mean.shape   ,     generator=snake_case__   ,     device=self.parameters.device   ,     dtype=self.parameters.dtype )
											snake_case_           =					self.mean + self.std * sample
											return x




				def 					a							(      self   ,     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 					a							(      self   ,     snake_case   ,     snake_case=[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=snake_case__ )




				def 					a							(      self ):
											return self.mean





from __future__ import annotations







class 				lowercase       :




				def __init__(      self   ,     snake_case ):
											snake_case_           =					TypeError(
											    'Matrices must be formed from a list of zero or more lists containing at '
											    'least one and the same number of values, each of which must be of type '
											    'int or float.' )
											if len(snake_case ) != 0:
																		snake_case_           =					len(rows[0] )
																		if cols == 0:
																									raise error
																		for row in rows:
																									if len(snake_case ) != cols:
																																raise error
																									for value in row:
																																if not isinstance(snake_case   ,     (int, float) ):
																																							raise error
																		snake_case_           =					rows
											else:
																		snake_case_           =					[]




				def 					a							(      self ):
											return [[row[i] for row in self.rows] for i in range(len(self.rows[0] ) )]




				@property
				def 					a							(      self ):
											return len(self.rows )




				@property
				def 					a							(      self ):
											return len(self.rows[0] )




				@property
				def 					a							(      self ):
											return (self.num_rows, self.num_columns)




				@property
				def 					a							(      self ):
											return self.order[0] == self.order[1]




				def 					a							(      self ):
											snake_case_           =					[
											    [0 if column_num != row_num else 1 for column_num in range(self.num_rows )]
											    for row_num in range(self.num_rows )
											]
											return Matrix(snake_case )




				def 					a							(      self ):
											if not self.is_square:
																		return 0
											if self.order == (0, 0):
																		return 1
											if self.order == (1, 1):
																		return int(self.rows[0][0] )
											if self.order == (2, 2):
																		return int(
																		    (self.rows[0][0] * self.rows[1][1])
																		    - (self.rows[0][1] * self.rows[1][0]) )
											else:
																		return sum(
																		    self.rows[0][column] * self.cofactors().rows[0][column]
																		    for column in range(self.num_columns ) )




				def 					a							(      self ):
											return bool(self.determinant() )




				def 					a							(      self   ,     snake_case   ,     snake_case ):
											snake_case_           =					[
											    [
											        self.rows[other_row][other_column]
											        for other_column in range(self.num_columns )
											        if other_column != column
											    ]
											    for other_row in range(self.num_rows )
											    if other_row != row
											]
											return Matrix(snake_case ).determinant()




				def 					a							(      self   ,     snake_case   ,     snake_case ):
											if (row + column) % 2 == 0:
																		return self.get_minor(snake_case   ,     snake_case )
											return -1 * self.get_minor(snake_case   ,     snake_case )




				def 					a							(      self ):
											return Matrix(
											    [
											        [self.get_minor(snake_case   ,     snake_case ) for column in range(self.num_columns )]
											        for row in range(self.num_rows )
											    ] )




				def 					a							(      self ):
											return Matrix(
											    [
											        [
											            self.minors().rows[row][column]
											            if (row + column) % 2 == 0
											            else self.minors().rows[row][column] * -1
											            for column in range(self.minors().num_columns )
											        ]
											        for row in range(self.minors().num_rows )
											    ] )




				def 					a							(      self ):
											snake_case_           =					[
											    [self.cofactors().rows[column][row] for column in range(self.num_columns )]
											    for row in range(self.num_rows )
											]
											return Matrix(snake_case )




				def 					a							(      self ):
											snake_case_           =					self.determinant()
											if not determinant:
																		raise TypeError('Only matrices with a non-zero determinant have an inverse' )
											return self.adjugate() * (1 / determinant)




				def __repr__(      self ):
											return str(self.rows )




				def __str__(      self ):
											if self.num_rows == 0:
																		return "[]"
											if self.num_rows == 1:
																		return "[[" + ". ".join(str(self.rows[0] ) ) + "]]"
											return (
											    "["
											    + "\n ".join(
											        [
											            '[' + '. '.join([str(snake_case ) for value in row] ) + '.]'
											            for row in self.rows
											        ] )
											    + "]"
											)




				def 					a							(      self   ,     snake_case   ,     snake_case = None ):
											snake_case_           =					TypeError('Row must be a list containing all ints and/or floats' )
											if not isinstance(snake_case   ,     snake_case ):
																		raise type_error
											for value in row:
																		if not isinstance(snake_case   ,     (int, float) ):
																									raise type_error
											if len(snake_case ) != self.num_columns:
																		raise ValueError(
																		    'Row must be equal in length to the other rows in the matrix' )
											if position is None:
																		self.rows.append(snake_case )
											else:
																		snake_case_           =					self.rows[0:position] + [row] + self.rows[position:]




				def 					a							(      self   ,     snake_case   ,     snake_case = None ):
											snake_case_           =					TypeError(
											    'Column must be a list containing all ints and/or floats' )
											if not isinstance(snake_case   ,     snake_case ):
																		raise type_error
											for value in column:
																		if not isinstance(snake_case   ,     (int, float) ):
																									raise type_error
											if len(snake_case ) != self.num_rows:
																		raise ValueError(
																		    'Column must be equal in length to the other columns in the matrix' )
											if position is None:
																		snake_case_           =					[self.rows[i] + [column[i]] for i in range(self.num_rows )]
											else:
																		snake_case_           =					[
																		    self.rows[i][0:position] + [column[i]] + self.rows[i][position:]
																		    for i in range(self.num_rows )
																		]




				def __eq__(      self   ,     snake_case ):
											if not isinstance(snake_case   ,     snake_case ):
																		return NotImplemented
											return self.rows == other.rows




				def __ne__(      self   ,     snake_case ):
											return not self == other




				def __neg__(      self ):
											return self * -1




				def __add__(      self   ,     snake_case ):
											if self.order != other.order:
																		raise ValueError('Addition requires matrices of the same order' )
											return Matrix(
											    [
											        [self.rows[i][j] + other.rows[i][j] for j in range(self.num_columns )]
											        for i in range(self.num_rows )
											    ] )




				def __sub__(      self   ,     snake_case ):
											if self.order != other.order:
																		raise ValueError('Subtraction requires matrices of the same order' )
											return Matrix(
											    [
											        [self.rows[i][j] - other.rows[i][j] for j in range(self.num_columns )]
											        for i in range(self.num_rows )
											    ] )




				def __mul__(      self   ,     snake_case ):
											if isinstance(snake_case   ,     (int, float) ):
																		return Matrix(
																		    [[int(element * other ) for element in row] for row in self.rows] )
											elif isinstance(snake_case   ,     snake_case ):
																		if self.num_columns != other.num_rows:
																									raise ValueError(
																									    'The number of columns in the first matrix must '
																									    'be equal to the number of rows in the second' )
																		return Matrix(
																		    [
																		        [Matrix.dot_product(snake_case   ,     snake_case ) for column in other.columns()]
																		        for row in self.rows
																		    ] )
											else:
																		raise TypeError(
																		    'A Matrix can only be multiplied by an int, float, or another matrix' )




				def __pow__(      self   ,     snake_case ):
											if not isinstance(snake_case   ,     snake_case ):
																		raise TypeError('A Matrix can only be raised to the power of an int' )
											if not self.is_square:
																		raise ValueError('Only square matrices can be raised to a power' )
											if other == 0:
																		return self.identity()
											if other < 0:
																		if self.is_invertable():
																									return self.inverse() ** (-other)
																		raise ValueError(
																		    'Only invertable matrices can be raised to a negative power' )
											snake_case_           =					self
											for _ in range(other - 1 ):
																		result *= self
											return result




				@classmethod
				def 					a							(      cls   ,     snake_case   ,     snake_case ):
											return sum(row[i] * column[i] for i in range(len(snake_case ) ) )


if __name__ == "__main__":
					import doctest

					doctest.testmod()




