

"""simple docstring"""




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			):
			lowerCamelCase__   :		Union[str, Any]							=							4_2





class    lowerCamelCase						(_A			,_A			):
			lowerCamelCase__   :		Any							=							True

			@register_to_config
			def __init__(					self				:						Optional[Any]    ,				__UpperCAmelCase				:						Tuple = 3    ,				__UpperCAmelCase				:						Any = 3    ,				__UpperCAmelCase				:						Optional[Any] = ("DownEncoderBlock2D",)    ,				__UpperCAmelCase				:						List[Any] = ("UpDecoderBlock2D",)    ,				__UpperCAmelCase				:						int = (6_4,)    ,				__UpperCAmelCase				:						Optional[int] = 1    ,				__UpperCAmelCase				:						Any = "silu"    ,				__UpperCAmelCase				:						Dict = 4    ,				__UpperCAmelCase				:						Tuple = 3_2    ,				__UpperCAmelCase				:						Optional[int] = 3_2    ,				__UpperCAmelCase				:						Tuple = 0.18_215    ,				)   ->    int:
						super().__init__()

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

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

						SCREAMING_SNAKE_CASE__								=      nn.Convad(2 * latent_channels    ,				2 * latent_channels    ,				1       )
						SCREAMING_SNAKE_CASE__								=      nn.Convad(__UpperCAmelCase    ,				__UpperCAmelCase    ,				1       )

						SCREAMING_SNAKE_CASE__								=      False
						SCREAMING_SNAKE_CASE__								=      False

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

			def 					SCREAMING_SNAKE_CASE				(					self				:						Any    ,				__UpperCAmelCase				:						Optional[int]    ,				__UpperCAmelCase				:						List[str]=False       )   ->    str:
						if isinstance(__UpperCAmelCase    ,				(Encoder, Decoder)       ):
									SCREAMING_SNAKE_CASE__								=      value

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[int]    ,				__UpperCAmelCase				:						Optional[Any] = True       )   ->    List[str]:
						SCREAMING_SNAKE_CASE__								=      use_tiling

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[int]       )   ->    List[str]:
						self.enable_tiling(__UpperCAmelCase       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						str       )   ->    Dict:
						SCREAMING_SNAKE_CASE__								=      True

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[int]       )   ->    Tuple:
						SCREAMING_SNAKE_CASE__								=      False

			@property
			# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[Any]       )   ->    Dict[str, AttentionProcessor]:
						SCREAMING_SNAKE_CASE__								=      {}

						def fn_recursive_add_processors(__UpperCAmelCase				:						Union[str, Any]    ,				__UpperCAmelCase				:						Union[str, Any]    ,				__UpperCAmelCase				:						List[str]       ):
									if hasattr(__UpperCAmelCase    ,				"""set_processor"""       ):
												SCREAMING_SNAKE_CASE__								=      module.processor

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

									return processors

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

						return processors

			def 					SCREAMING_SNAKE_CASE				(					self				:						int    ,				__UpperCAmelCase				:						Optional[Any]       )   ->    Optional[Any]:
						SCREAMING_SNAKE_CASE__								=      len(self.attn_processors.keys()       )

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

						def fn_recursive_attn_processor(__UpperCAmelCase				:						List[Any]    ,				__UpperCAmelCase				:						Tuple    ,				__UpperCAmelCase				:						Optional[Any]       ):
									if hasattr(__UpperCAmelCase    ,				"""set_processor"""       ):
												if not isinstance(__UpperCAmelCase    ,				__UpperCAmelCase       ):
															module.set_processor(__UpperCAmelCase       )
												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}"""    ,				__UpperCAmelCase    ,				__UpperCAmelCase       )

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

			def 					SCREAMING_SNAKE_CASE				(					self				:						Any       )   ->    Optional[int]:
						self.set_attn_processor(AttnProcessor()       )

			@apply_forward_hook
			def 					SCREAMING_SNAKE_CASE				(					self				:						str    ,				__UpperCAmelCase				:						List[Any]    ,				__UpperCAmelCase				:						List[str] = True       )   ->    AutoencoderKLOutput:
						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(__UpperCAmelCase    ,				return_dict=__UpperCAmelCase       )

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

						SCREAMING_SNAKE_CASE__								=      self.quant_conv(__UpperCAmelCase       )
						SCREAMING_SNAKE_CASE__								=      DiagonalGaussianDistribution(__UpperCAmelCase       )

						if not return_dict:
									return (posterior,)

						return AutoencoderKLOutput(latent_dist=__UpperCAmelCase       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Any    ,				__UpperCAmelCase				:						Union[str, Any]    ,				__UpperCAmelCase				:						str = True       )   ->    Union[DecoderOutput, torch.FloatTensor]:
						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(__UpperCAmelCase    ,				return_dict=__UpperCAmelCase       )

						SCREAMING_SNAKE_CASE__								=      self.post_quant_conv(__UpperCAmelCase       )
						SCREAMING_SNAKE_CASE__								=      self.decoder(__UpperCAmelCase       )

						if not return_dict:
									return (dec,)

						return DecoderOutput(sample=__UpperCAmelCase       )

			@apply_forward_hook
			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[Any]    ,				__UpperCAmelCase				:						Optional[Any]    ,				__UpperCAmelCase				:						Optional[Any] = True       )   ->    Union[DecoderOutput, torch.FloatTensor]:
						if self.use_slicing and z.shape[0] > 1:
									SCREAMING_SNAKE_CASE__								=      [self._decode(__UpperCAmelCase       ).sample for z_slice in z.split(1       )]
									SCREAMING_SNAKE_CASE__								=      torch.cat(__UpperCAmelCase       )
						else:
									SCREAMING_SNAKE_CASE__								=      self._decode(__UpperCAmelCase       ).sample

						if not return_dict:
									return (decoded,)

						return DecoderOutput(sample=__UpperCAmelCase       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Tuple    ,				__UpperCAmelCase				:						int    ,				__UpperCAmelCase				:						Dict    ,				__UpperCAmelCase				:						List[str]       )   ->    Dict:
						SCREAMING_SNAKE_CASE__								=      min(a.shape[2]    ,				b.shape[2]    ,				__UpperCAmelCase       )
						for y in range(__UpperCAmelCase       ):
									SCREAMING_SNAKE_CASE__								=      a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
						return b

			def 					SCREAMING_SNAKE_CASE				(					self				:						str    ,				__UpperCAmelCase				:						Optional[Any]    ,				__UpperCAmelCase				:						str    ,				__UpperCAmelCase				:						str       )   ->    Dict:
						SCREAMING_SNAKE_CASE__								=      min(a.shape[3]    ,				b.shape[3]    ,				__UpperCAmelCase       )
						for x in range(__UpperCAmelCase       ):
									SCREAMING_SNAKE_CASE__								=      a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
						return b

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[Any]    ,				__UpperCAmelCase				:						List[Any]    ,				__UpperCAmelCase				:						Dict = True       )   ->    AutoencoderKLOutput:
						SCREAMING_SNAKE_CASE__								=      int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)       )
						SCREAMING_SNAKE_CASE__								=      int(self.tile_latent_min_size * self.tile_overlap_factor       )
						SCREAMING_SNAKE_CASE__								=      self.tile_latent_min_size - blend_extent

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

						SCREAMING_SNAKE_CASE__								=      torch.cat(__UpperCAmelCase    ,				dim=2       )
						SCREAMING_SNAKE_CASE__								=      DiagonalGaussianDistribution(__UpperCAmelCase       )

						if not return_dict:
									return (posterior,)

						return AutoencoderKLOutput(latent_dist=__UpperCAmelCase       )

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[int]    ,				__UpperCAmelCase				:						Dict    ,				__UpperCAmelCase				:						str = True       )   ->    Union[DecoderOutput, torch.FloatTensor]:
						SCREAMING_SNAKE_CASE__								=      int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)       )
						SCREAMING_SNAKE_CASE__								=      int(self.tile_sample_min_size * self.tile_overlap_factor       )
						SCREAMING_SNAKE_CASE__								=      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.
						SCREAMING_SNAKE_CASE__								=      []
						for i in range(0    ,				z.shape[2]    ,				__UpperCAmelCase       ):
									SCREAMING_SNAKE_CASE__								=      []
									for j in range(0    ,				z.shape[3]    ,				__UpperCAmelCase       ):
												SCREAMING_SNAKE_CASE__								=      z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
												SCREAMING_SNAKE_CASE__								=      self.post_quant_conv(__UpperCAmelCase       )
												SCREAMING_SNAKE_CASE__								=      self.decoder(__UpperCAmelCase       )
												row.append(__UpperCAmelCase       )
									rows.append(__UpperCAmelCase       )
						SCREAMING_SNAKE_CASE__								=      []
						for i, row in enumerate(__UpperCAmelCase       ):
									SCREAMING_SNAKE_CASE__								=      []
									for j, tile in enumerate(__UpperCAmelCase       ):
												# blend the above tile and the left tile
												# to the current tile and add the current tile to the result row
												if i > 0:
															SCREAMING_SNAKE_CASE__								=      self.blend_v(rows[i - 1][j]    ,				__UpperCAmelCase    ,				__UpperCAmelCase       )
												if j > 0:
															SCREAMING_SNAKE_CASE__								=      self.blend_h(row[j - 1]    ,				__UpperCAmelCase    ,				__UpperCAmelCase       )
												result_row.append(tile[:, :, :row_limit, :row_limit]       )
									result_rows.append(torch.cat(__UpperCAmelCase    ,				dim=3       )       )

						SCREAMING_SNAKE_CASE__								=      torch.cat(__UpperCAmelCase    ,				dim=2       )
						if not return_dict:
									return (dec,)

						return DecoderOutput(sample=__UpperCAmelCase       )





			def 					SCREAMING_SNAKE_CASE				(					self				:						Tuple    ,				__UpperCAmelCase				:						Union[str, Any]    ,				__UpperCAmelCase				:						Union[str, Any] = False    ,				__UpperCAmelCase				:						Union[str, Any] = True    ,				__UpperCAmelCase				:						List[Any] = None    ,				)   ->    Union[DecoderOutput, torch.FloatTensor]:
						SCREAMING_SNAKE_CASE__								=      sample
						SCREAMING_SNAKE_CASE__								=      self.encode(__UpperCAmelCase       ).latent_dist
						if sample_posterior:
									SCREAMING_SNAKE_CASE__								=      posterior.sample(generator=__UpperCAmelCase       )
						else:
									SCREAMING_SNAKE_CASE__								=      posterior.mode()
						SCREAMING_SNAKE_CASE__								=      self.decode(__UpperCAmelCase       ).sample

						if not return_dict:
									return (dec,)

						return DecoderOutput(sample=__UpperCAmelCase       )





"""simple docstring"""




from .glue import GlueDataset, GlueDataTrainingArguments
from .language_modeling import (
    LineByLineTextDataset,
    LineByLineWithRefDataset,
    LineByLineWithSOPTextDataset,
    TextDataset,
    TextDatasetForNextSentencePrediction,
)
from .squad import SquadDataset, SquadDataTrainingArguments


