
from dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder







@dataclass
class        lowerCAmelCase__      (   __a	):
						UpperCamelCase_							: "DiagonalGaussianDistribution"







class        lowerCAmelCase__      (   __a		,	__a	):
						UpperCamelCase_							: Tuple	=		True

						@register_to_config
						def __init__(			self       ,							a = 3       ,							a = 3       ,							a = ("DownEncoderBlock2D",)       ,							a = ("UpDecoderBlock2D",)       ,							a = (64,)       ,							a = 1       ,							a = "silu"       ,							a = 4       ,							a = 32       ,							a = 32       ,							a = 0.1_8215       ,							)					->      Tuple:


								'''simple docstring'''

								super().__init__()

								# pass init params to Encoder
								_UpperCamelCase						    =	Encoder(
								    in_channels=snake_case__       ,							out_channels=snake_case__       ,							down_block_types=snake_case__       ,							block_out_channels=snake_case__       ,							layers_per_block=snake_case__       ,							act_fn=snake_case__       ,							norm_num_groups=snake_case__       ,							double_z=snake_case__       ,							)

								# pass init params to Decoder
								_UpperCamelCase						    =	Decoder(
								    in_channels=snake_case__       ,							out_channels=snake_case__       ,							up_block_types=snake_case__       ,							block_out_channels=snake_case__       ,							layers_per_block=snake_case__       ,							norm_num_groups=snake_case__       ,							act_fn=snake_case__       ,							)

								_UpperCamelCase						    =	nn.Convad(2 * latent_channels       ,							2 * latent_channels       ,							1				)
								_UpperCamelCase						    =	nn.Convad(snake_case__       ,							snake_case__       ,							1				)

								_UpperCamelCase						    =	False
								_UpperCamelCase						    =	False

								# only relevant if vae tiling is enabled
								_UpperCamelCase						    =	self.config.sample_size
								_UpperCamelCase						    =	(
								    self.config.sample_size[0]
								    if isinstance(self.config.sample_size       ,							(list, tuple)				)
								    else self.config.sample_size
								)
								_UpperCamelCase						    =	int(sample_size / (2 ** (len(self.config.block_out_channels				) - 1))				)
								_UpperCamelCase						    =	0.25

						def  A_				(			self       ,							a       ,							a=False				)					->      Union[str, Any]:


								'''simple docstring'''

								if isinstance(snake_case__       ,							(Encoder, Decoder)				):
										_UpperCamelCase						    =	value

						def  A_				(			self       ,							a = True				)					->      Optional[Any]:


								'''simple docstring'''

								_UpperCamelCase						    =	use_tiling

						def  A_				(			self				)					->      List[Any]:


								'''simple docstring'''

								self.enable_tiling(snake_case__				)

						def  A_				(			self				)					->      int:


								'''simple docstring'''

								_UpperCamelCase						    =	True

						def  A_				(			self				)					->      Optional[Any]:


								'''simple docstring'''

								_UpperCamelCase						    =	False

						@property
						# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
						def  A_				(			self				)					->      Tuple:


								'''simple docstring'''

								_UpperCamelCase						    =	{}

								def fn_recursive_add_processors(a       ,							a       ,							a				):
										if hasattr(snake_case__       ,							"""set_processor"""				):
												_UpperCamelCase						    =	module.processor

										for sub_name, child in module.named_children():
												fn_recursive_add_processors(F'{name}.{sub_name}'       ,							snake_case__       ,							snake_case__				)

										return processors

								for name, module in self.named_children():
										fn_recursive_add_processors(snake_case__       ,							snake_case__       ,							snake_case__				)

								return processors

						def  A_				(			self       ,							a				)					->      Optional[Any]:


								'''simple docstring'''

								_UpperCamelCase						    =	len(self.attn_processors.keys()				)

								if isinstance(snake_case__       ,							snake_case__				) and len(snake_case__				) != count:
										raise ValueError(
										    F'A dict of processors was passed, but the number of processors {len(snake_case__				)} does not match the'
										    F' number of attention layers: {count}. Please make sure to pass {count} processor classes.'				)

								def fn_recursive_attn_processor(a       ,							a       ,							a				):
										if hasattr(snake_case__       ,							"""set_processor"""				):
												if not isinstance(snake_case__       ,							snake_case__				):
														module.set_processor(snake_case__				)
												else:
														module.set_processor(processor.pop(F'{name}.processor'				)				)

										for sub_name, child in module.named_children():
												fn_recursive_attn_processor(F'{name}.{sub_name}'       ,							snake_case__       ,							snake_case__				)

								for name, module in self.named_children():
										fn_recursive_attn_processor(snake_case__       ,							snake_case__       ,							snake_case__				)

						def  A_				(			self				)					->      Union[str, Any]:


								'''simple docstring'''

								self.set_attn_processor(AttnProcessor()				)

						@apply_forward_hook
						def  A_				(			self       ,							a       ,							a = True				)					->      str:


								'''simple docstring'''

								if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
										return self.tiled_encode(snake_case__       ,							return_dict=snake_case__				)

								if self.use_slicing and x.shape[0] > 1:
										_UpperCamelCase						    =	[self.encoder(snake_case__				) for x_slice in x.split(1				)]
										_UpperCamelCase						    =	torch.cat(snake_case__				)
								else:
										_UpperCamelCase						    =	self.encoder(snake_case__				)

								_UpperCamelCase						    =	self.quant_conv(snake_case__				)
								_UpperCamelCase						    =	DiagonalGaussianDistribution(snake_case__				)

								if not return_dict:
										return (posterior,)

								return AutoencoderKLOutput(latent_dist=snake_case__				)

						def  A_				(			self       ,							a       ,							a = True				)					->      Tuple:


								'''simple docstring'''

								if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
										return self.tiled_decode(snake_case__       ,							return_dict=snake_case__				)

								_UpperCamelCase						    =	self.post_quant_conv(snake_case__				)
								_UpperCamelCase						    =	self.decoder(snake_case__				)

								if not return_dict:
										return (dec,)

								return DecoderOutput(sample=snake_case__				)

						@apply_forward_hook
						def  A_				(			self       ,							a       ,							a = True				)					->      int:


								'''simple docstring'''

								if self.use_slicing and z.shape[0] > 1:
										_UpperCamelCase						    =	[self._decode(snake_case__				).sample for z_slice in z.split(1				)]
										_UpperCamelCase						    =	torch.cat(snake_case__				)
								else:
										_UpperCamelCase						    =	self._decode(snake_case__				).sample

								if not return_dict:
										return (decoded,)

								return DecoderOutput(sample=snake_case__				)

						def  A_				(			self       ,							a       ,							a       ,							a				)					->      Union[str, Any]:


								'''simple docstring'''

								_UpperCamelCase						    =	min(a.shape[2]       ,							b.shape[2]       ,							snake_case__				)
								for y in range(snake_case__				):
										_UpperCamelCase						    =	a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
								return b

						def  A_				(			self       ,							a       ,							a       ,							a				)					->      Optional[Any]:


								'''simple docstring'''

								_UpperCamelCase						    =	min(a.shape[3]       ,							b.shape[3]       ,							snake_case__				)
								for x in range(snake_case__				):
										_UpperCamelCase						    =	a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
								return b

						def  A_				(			self       ,							a       ,							a = True				)					->      Tuple:


								'''simple docstring'''

								_UpperCamelCase						    =	int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)				)
								_UpperCamelCase						    =	int(self.tile_latent_min_size * self.tile_overlap_factor				)
								_UpperCamelCase						    =	self.tile_latent_min_size - blend_extent

								# Split the image into 512x512 tiles and encode them separately.
								_UpperCamelCase						    =	[]
								for i in range(0       ,							x.shape[2]       ,							snake_case__				):
										_UpperCamelCase						    =	[]
										for j in range(0       ,							x.shape[3]       ,							snake_case__				):
												_UpperCamelCase						    =	x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
												_UpperCamelCase						    =	self.encoder(snake_case__				)
												_UpperCamelCase						    =	self.quant_conv(snake_case__				)
												row.append(snake_case__				)
										rows.append(snake_case__				)
								_UpperCamelCase						    =	[]
								for i, row in enumerate(snake_case__				):
										_UpperCamelCase						    =	[]
										for j, tile in enumerate(snake_case__				):
												# blend the above tile and the left tile
												# to the current tile and add the current tile to the result row
												if i > 0:
														_UpperCamelCase						    =	self.blend_v(rows[i - 1][j]       ,							snake_case__       ,							snake_case__				)
												if j > 0:
														_UpperCamelCase						    =	self.blend_h(row[j - 1]       ,							snake_case__       ,							snake_case__				)
												result_row.append(tile[:, :, :row_limit, :row_limit]				)
										result_rows.append(torch.cat(snake_case__       ,							dim=3				)				)

								_UpperCamelCase						    =	torch.cat(snake_case__       ,							dim=2				)
								_UpperCamelCase						    =	DiagonalGaussianDistribution(snake_case__				)

								if not return_dict:
										return (posterior,)

								return AutoencoderKLOutput(latent_dist=snake_case__				)

						def  A_				(			self       ,							a       ,							a = True				)					->      Any:


								'''simple docstring'''

								_UpperCamelCase						    =	int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)				)
								_UpperCamelCase						    =	int(self.tile_sample_min_size * self.tile_overlap_factor				)
								_UpperCamelCase						    =	self.tile_sample_min_size - blend_extent

								# Split z into overlapping 64x64 tiles and decode them separately.
								# The tiles have an overlap to avoid seams between tiles.
								_UpperCamelCase						    =	[]
								for i in range(0       ,							z.shape[2]       ,							snake_case__				):
										_UpperCamelCase						    =	[]
										for j in range(0       ,							z.shape[3]       ,							snake_case__				):
												_UpperCamelCase						    =	z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
												_UpperCamelCase						    =	self.post_quant_conv(snake_case__				)
												_UpperCamelCase						    =	self.decoder(snake_case__				)
												row.append(snake_case__				)
										rows.append(snake_case__				)
								_UpperCamelCase						    =	[]
								for i, row in enumerate(snake_case__				):
										_UpperCamelCase						    =	[]
										for j, tile in enumerate(snake_case__				):
												# blend the above tile and the left tile
												# to the current tile and add the current tile to the result row
												if i > 0:
														_UpperCamelCase						    =	self.blend_v(rows[i - 1][j]       ,							snake_case__       ,							snake_case__				)
												if j > 0:
														_UpperCamelCase						    =	self.blend_h(row[j - 1]       ,							snake_case__       ,							snake_case__				)
												result_row.append(tile[:, :, :row_limit, :row_limit]				)
										result_rows.append(torch.cat(snake_case__       ,							dim=3				)				)

								_UpperCamelCase						    =	torch.cat(snake_case__       ,							dim=2				)
								if not return_dict:
										return (dec,)

								return DecoderOutput(sample=snake_case__				)







						def  A_				(			self       ,							a       ,							a = False       ,							a = True       ,							a = None       ,							)					->      int:


								'''simple docstring'''

								_UpperCamelCase						    =	sample
								_UpperCamelCase						    =	self.encode(snake_case__				).latent_dist
								if sample_posterior:
										_UpperCamelCase						    =	posterior.sample(generator=snake_case__				)
								else:
										_UpperCamelCase						    =	posterior.mode()
								_UpperCamelCase						    =	self.decode(snake_case__				).sample

								if not return_dict:
										return (dec,)

								return DecoderOutput(sample=snake_case__				)





from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowerCamelCase__	=							{
    "configuration_mctct": ["MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP", "MCTCTConfig"],
    "feature_extraction_mctct": ["MCTCTFeatureExtractor"],
    "processing_mctct": ["MCTCTProcessor"],
}


try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowerCamelCase__	=							[
		    "MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST",
		    "MCTCTForCTC",
		    "MCTCTModel",
		    "MCTCTPreTrainedModel",
		]


if TYPE_CHECKING:
		from .configuration_mctct import MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP, MCTCTConfig
		from .feature_extraction_mctct import MCTCTFeatureExtractor
		from .processing_mctct import MCTCTProcessor

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_mctct import MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST, MCTCTForCTC, MCTCTModel, MCTCTPreTrainedModel

else:
		import sys

		lowerCamelCase__	=							_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)



