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



				"""simple docstring"""
				_lowerCAmelCase     :    Optional[int]			  =						42





class   lowerCAmelCase_    (			_A     ,    _A					):



				"""simple docstring"""
				_lowerCAmelCase     :    List[Any]			  =						True

				@register_to_config
				def __init__(				self		,    lowerCAmelCase = 3		,    lowerCAmelCase = 3		,    lowerCAmelCase = ("DownEncoderBlock2D",)		,    lowerCAmelCase = ("UpDecoderBlock2D",)		,    lowerCAmelCase = (64,)		,    lowerCAmelCase = 1		,    lowerCAmelCase = "silu"		,    lowerCAmelCase = 4		,    lowerCAmelCase = 32		,    lowerCAmelCase = 32		,    lowerCAmelCase = 0.1_82_15		,    ):






										"""simple docstring"""

										super().__init__()

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

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

										snake_case								=			nn.Convad(2 * latent_channels		,    2 * latent_channels		,    1    )
										snake_case								=			nn.Convad(lowerCAmelCase		,    lowerCAmelCase		,    1    )

										snake_case								=			False
										snake_case								=			False

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

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase=False    ):






										"""simple docstring"""

										if isinstance(lowerCAmelCase		,    (Encoder, Decoder)    ):
																snake_case								=			value

				def 		snake_case (				self		,    lowerCAmelCase = True    ):






										"""simple docstring"""

										snake_case								=			use_tiling

				def 		snake_case (				self    ):






										"""simple docstring"""

										self.enable_tiling(lowerCAmelCase    )

				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			True

				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			False

				@property
				# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
				def 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			{}

										def fn_recursive_add_processors(lowerCAmelCase		,    lowerCAmelCase		,    lowerCAmelCase    ):
																if hasattr(lowerCAmelCase		,    'set_processor'    ):
																						snake_case								=			module.processor

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

																return processors

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

										return processors

				def 		snake_case (				self		,    lowerCAmelCase    ):






										"""simple docstring"""

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

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

										def fn_recursive_attn_processor(lowerCAmelCase		,    lowerCAmelCase		,    lowerCAmelCase    ):
																if hasattr(lowerCAmelCase		,    'set_processor'    ):
																						if not isinstance(lowerCAmelCase		,    lowerCAmelCase    ):
																												module.set_processor(lowerCAmelCase    )
																						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}"""		,    lowerCAmelCase		,    lowerCAmelCase    )

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

				def 		snake_case (				self    ):






										"""simple docstring"""

										self.set_attn_processor(AttnProcessor()    )

				@apply_forward_hook
				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = True    ):






										"""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(lowerCAmelCase		,    return_dict=lowerCAmelCase    )

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

										snake_case								=			self.quant_conv(lowerCAmelCase    )
										snake_case								=			DiagonalGaussianDistribution(lowerCAmelCase    )

										if not return_dict:
																return (posterior,)

										return AutoencoderKLOutput(latent_dist=lowerCAmelCase    )

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = True    ):






										"""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(lowerCAmelCase		,    return_dict=lowerCAmelCase    )

										snake_case								=			self.post_quant_conv(lowerCAmelCase    )
										snake_case								=			self.decoder(lowerCAmelCase    )

										if not return_dict:
																return (dec,)

										return DecoderOutput(sample=lowerCAmelCase    )

				@apply_forward_hook
				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = True    ):






										"""simple docstring"""

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

										if not return_dict:
																return (decoded,)

										return DecoderOutput(sample=lowerCAmelCase    )

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase		,    lowerCAmelCase    ):






										"""simple docstring"""

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

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase		,    lowerCAmelCase    ):






										"""simple docstring"""

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

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = True    ):






										"""simple docstring"""

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

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

										snake_case								=			torch.cat(lowerCAmelCase		,    dim=2    )
										snake_case								=			DiagonalGaussianDistribution(lowerCAmelCase    )

										if not return_dict:
																return (posterior,)

										return AutoencoderKLOutput(latent_dist=lowerCAmelCase    )

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = True    ):






										"""simple docstring"""

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

										snake_case								=			torch.cat(lowerCAmelCase		,    dim=2    )
										if not return_dict:
																return (dec,)

										return DecoderOutput(sample=lowerCAmelCase    )

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase = False		,    lowerCAmelCase = True		,    lowerCAmelCase = None		,    ):






										"""simple docstring"""

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

										if not return_dict:
																return (dec,)

										return DecoderOutput(sample=lowerCAmelCase    )
"""simple docstring"""
import enum
import os
from hashlib import shaaaa
from typing import Optional

from .. import config
from .logging import get_logger


SCREAMING_SNAKE_CASE__      						=  get_logger(__name__)



class   lowerCAmelCase_    (			enum.Enum					):



				"""simple docstring"""
				_lowerCAmelCase     :    Union[str, Any]			  =						"""all_checks"""
				_lowerCAmelCase     :    Optional[Any]			  =						"""basic_checks"""
				_lowerCAmelCase     :    Optional[int]			  =						"""no_checks"""





class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





def    lowerCAmelCase__    (			_UpperCamelCase       :				Optional[dict] ,			_UpperCamelCase       :				dict ,			_UpperCamelCase       :				str=None					)					->				Dict:





						"""simple docstring"""






						if expected_checksums is None:
												logger.info('Unable to verify checksums.'					)
												return
						if len(set(_UpperCamelCase					) - set(_UpperCamelCase					)					) > 0:
												raise ExpectedMoreDownloadedFiles(str(set(_UpperCamelCase					) - set(_UpperCamelCase					)					)					)
						if len(set(_UpperCamelCase					) - set(_UpperCamelCase					)					) > 0:
												raise UnexpectedDownloadedFile(str(set(_UpperCamelCase					) - set(_UpperCamelCase					)					)					)
						snake_case								=			[url for url in expected_checksums if expected_checksums[url] != recorded_checksums[url]]
						snake_case								=			' for ' + verification_name if verification_name is not None else ''
						if len(_UpperCamelCase					) > 0:
												raise NonMatchingChecksumError(
												    f"""Checksums didn't match{for_verification_name}:\n"""
												    f"""{bad_urls}\n"""
												    'Set `verification_mode=\'no_checks\'` to skip checksums verification and ignore this error'					)
						logger.info('All the checksums matched successfully' + for_verification_name					)



class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""





def    lowerCAmelCase__    (			_UpperCamelCase       :				Optional[dict] ,			_UpperCamelCase       :				dict					)					->				Tuple:





						"""simple docstring"""






						if expected_splits is None:
												logger.info('Unable to verify splits sizes.'					)
												return
						if len(set(_UpperCamelCase					) - set(_UpperCamelCase					)					) > 0:
												raise ExpectedMoreSplits(str(set(_UpperCamelCase					) - set(_UpperCamelCase					)					)					)
						if len(set(_UpperCamelCase					) - set(_UpperCamelCase					)					) > 0:
												raise UnexpectedSplits(str(set(_UpperCamelCase					) - set(_UpperCamelCase					)					)					)
						snake_case								=			[
						    {'expected': expected_splits[name], 'recorded': recorded_splits[name]}
						    for name in expected_splits
						    if expected_splits[name].num_examples != recorded_splits[name].num_examples
						]
						if len(_UpperCamelCase					) > 0:
												raise NonMatchingSplitsSizesError(str(_UpperCamelCase					)					)
						logger.info('All the splits matched successfully.'					)





def    lowerCAmelCase__    (			_UpperCamelCase       :				str ,			_UpperCamelCase       :				bool = True					)					->				dict:





						"""simple docstring"""






						if record_checksum:
												snake_case								=			shaaaa()
												with open(_UpperCamelCase ,			'rb'					) as f:
																		for chunk in iter(lambda: f.read(1 << 2_0					) ,			B''					):
																								m.update(_UpperCamelCase					)
																		snake_case								=			m.hexdigest()
						else:
												snake_case								=			None
						return {"num_bytes": os.path.getsize(_UpperCamelCase					), "checksum": checksum}





def    lowerCAmelCase__    (			_UpperCamelCase       :				Optional[int]					)					->				int:





						"""simple docstring"""






						if dataset_size and config.IN_MEMORY_MAX_SIZE:
												return dataset_size < config.IN_MEMORY_MAX_SIZE
						else:
												return False


