


'''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 						_snake_case   (       snake_case_		):
					lowerCAmelCase :torch.FloatTensor


class 						_snake_case   (       nn.Module		):




					def __init__( self							,							_lowerCamelCase=3							,							_lowerCamelCase=3							,							_lowerCamelCase=("DownEncoderBlock2D",)							,							_lowerCamelCase=(64,)							,							_lowerCamelCase=2							,							_lowerCamelCase=32							,							_lowerCamelCase="silu"							,							_lowerCamelCase=True							,							):
								super().__init__()
								UpperCAmelCase__					:   List[str]       				=       layers_per_block

								UpperCAmelCase__					:   Optional[int]       				=       torch.nn.Convad(
								    _A							,							block_out_channels[0]							,							kernel_size=3							,							stride=1							,							padding=1							,							)

								UpperCAmelCase__					:   str       				=       None
								UpperCAmelCase__					:   List[str]       				=       nn.ModuleList([])

								# down
								UpperCAmelCase__					:   List[str]       				=       block_out_channels[0]
								for i, down_block_type in enumerate(_A):
											UpperCAmelCase__					:   List[str]       				=       output_channel
											UpperCAmelCase__					:   Any       				=       block_out_channels[i]
											UpperCAmelCase__					:   Any       				=       i == len(_A) - 1

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

								# mid
								UpperCAmelCase__					:   str       				=       UNetMidBlockaD(
								    in_channels=block_out_channels[-1]							,							resnet_eps=1e-6							,							resnet_act_fn=_A							,							output_scale_factor=1							,							resnet_time_scale_shift="""default"""							,							attention_head_dim=block_out_channels[-1]							,							resnet_groups=_A							,							temb_channels=_A							,							)

								# out
								UpperCAmelCase__					:   Optional[int]       				=       nn.GroupNorm(num_channels=block_out_channels[-1]							,							num_groups=_A							,							eps=1e-6)
								UpperCAmelCase__					:   int       				=       nn.SiLU()

								UpperCAmelCase__					:   Optional[int]       				=       2 * out_channels if double_z else out_channels
								UpperCAmelCase__					:   Any       				=       nn.Convad(block_out_channels[-1]							,							_A							,							3							,							padding=1)

								UpperCAmelCase__					:   List[str]       				=       False




					def 			snake_case__  ( self							,							_lowerCamelCase):
								UpperCAmelCase__					:   Any       				=       x
								UpperCAmelCase__					:   List[str]       				=       self.conv_in(_A)

								if self.training and self.gradient_checkpointing:

											def create_custom_forward(_lowerCamelCase):
														def custom_forward(*_lowerCamelCase):
																	return module(*_A)

														return custom_forward

											# down
											if is_torch_version(""">="""							,							"""1.11.0"""):
														for down_block in self.down_blocks:
																	UpperCAmelCase__					:   List[str]       				=       torch.utils.checkpoint.checkpoint(
																	    create_custom_forward(_A)							,							_A							,							use_reentrant=_A)
														# middle
														UpperCAmelCase__					:   Union[str, Any]       				=       torch.utils.checkpoint.checkpoint(
														    create_custom_forward(self.mid_block)							,							_A							,							use_reentrant=_A)
											else:
														for down_block in self.down_blocks:
																	UpperCAmelCase__					:   Tuple       				=       torch.utils.checkpoint.checkpoint(create_custom_forward(_A)							,							_A)
														# middle
														UpperCAmelCase__					:   Any       				=       torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block)							,							_A)

								else:
											# down
											for down_block in self.down_blocks:
														UpperCAmelCase__					:   Tuple       				=       down_block(_A)

											# middle
											UpperCAmelCase__					:   Optional[int]       				=       self.mid_block(_A)

								# post-process
								UpperCAmelCase__					:   str       				=       self.conv_norm_out(_A)
								UpperCAmelCase__					:   Dict       				=       self.conv_act(_A)
								UpperCAmelCase__					:   List[Any]       				=       self.conv_out(_A)

								return sample


class 						_snake_case   (       nn.Module		):




					def __init__( self							,							_lowerCamelCase=3							,							_lowerCamelCase=3							,							_lowerCamelCase=("UpDecoderBlock2D",)							,							_lowerCamelCase=(64,)							,							_lowerCamelCase=2							,							_lowerCamelCase=32							,							_lowerCamelCase="silu"							,							_lowerCamelCase="group"							,							):
								super().__init__()
								UpperCAmelCase__					:   Any       				=       layers_per_block

								UpperCAmelCase__					:   Any       				=       nn.Convad(
								    _A							,							block_out_channels[-1]							,							kernel_size=3							,							stride=1							,							padding=1							,							)

								UpperCAmelCase__					:   List[Any]       				=       None
								UpperCAmelCase__					:   List[str]       				=       nn.ModuleList([])

								UpperCAmelCase__					:   Dict       				=       in_channels if norm_type == 'spatial' else None

								# mid
								UpperCAmelCase__					:   int       				=       UNetMidBlockaD(
								    in_channels=block_out_channels[-1]							,							resnet_eps=1e-6							,							resnet_act_fn=_A							,							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=_A							,							temb_channels=_A							,							)

								# up
								UpperCAmelCase__					:   Union[str, Any]       				=       list(reversed(_A))
								UpperCAmelCase__					:   int       				=       reversed_block_out_channels[0]
								for i, up_block_type in enumerate(_A):
											UpperCAmelCase__					:   List[Any]       				=       output_channel
											UpperCAmelCase__					:   Optional[int]       				=       reversed_block_out_channels[i]

											UpperCAmelCase__					:   Dict       				=       i == len(_A) - 1

											UpperCAmelCase__					:   Optional[Any]       				=       get_up_block(
											    _A							,							num_layers=self.layers_per_block + 1							,							in_channels=_A							,							out_channels=_A							,							prev_output_channel=_A							,							add_upsample=not is_final_block							,							resnet_eps=1e-6							,							resnet_act_fn=_A							,							resnet_groups=_A							,							attention_head_dim=_A							,							temb_channels=_A							,							resnet_time_scale_shift=_A							,							)
											self.up_blocks.append(_A)
											UpperCAmelCase__					:   int       				=       output_channel

								# out
								if norm_type == "spatial":
											UpperCAmelCase__					:   Union[str, Any]       				=       SpatialNorm(block_out_channels[0]							,							_A)
								else:
											UpperCAmelCase__					:   Dict       				=       nn.GroupNorm(num_channels=block_out_channels[0]							,							num_groups=_A							,							eps=1e-6)
								UpperCAmelCase__					:   List[str]       				=       nn.SiLU()
								UpperCAmelCase__					:   Optional[int]       				=       nn.Convad(block_out_channels[0]							,							_A							,							3							,							padding=1)

								UpperCAmelCase__					:   Optional[int]       				=       False




					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase=None):
								UpperCAmelCase__					:   Any       				=       z
								UpperCAmelCase__					:   List[str]       				=       self.conv_in(_A)

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

											def create_custom_forward(_lowerCamelCase):
														def custom_forward(*_lowerCamelCase):
																	return module(*_A)

														return custom_forward

											if is_torch_version(""">="""							,							"""1.11.0"""):
														# middle
														UpperCAmelCase__					:   Dict       				=       torch.utils.checkpoint.checkpoint(
														    create_custom_forward(self.mid_block)							,							_A							,							_A							,							use_reentrant=_A)
														UpperCAmelCase__					:   Optional[Any]       				=       sample.to(_A)

														# up
														for up_block in self.up_blocks:
																	UpperCAmelCase__					:   int       				=       torch.utils.checkpoint.checkpoint(
																	    create_custom_forward(_A)							,							_A							,							_A							,							use_reentrant=_A)
											else:
														# middle
														UpperCAmelCase__					:   int       				=       torch.utils.checkpoint.checkpoint(
														    create_custom_forward(self.mid_block)							,							_A							,							_A)
														UpperCAmelCase__					:   str       				=       sample.to(_A)

														# up
														for up_block in self.up_blocks:
																	UpperCAmelCase__					:   str       				=       torch.utils.checkpoint.checkpoint(create_custom_forward(_A)							,							_A							,							_A)
								else:
											# middle
											UpperCAmelCase__					:   str       				=       self.mid_block(_A							,							_A)
											UpperCAmelCase__					:   Any       				=       sample.to(_A)

											# up
											for up_block in self.up_blocks:
														UpperCAmelCase__					:   int       				=       up_block(_A							,							_A)

        # post-process
								if latent_embeds is None:
											UpperCAmelCase__					:   List[Any]       				=       self.conv_norm_out(_A)
								else:
											UpperCAmelCase__					:   Optional[int]       				=       self.conv_norm_out(_A							,							_A)
								UpperCAmelCase__					:   Any       				=       self.conv_act(_A)
								UpperCAmelCase__					:   Dict       				=       self.conv_out(_A)

								return sample


class 						_snake_case   (       nn.Module		):




					def __init__( self							,							_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase=None							,							_lowerCamelCase="random"							,							_lowerCamelCase=False							,							_lowerCamelCase=True):
								super().__init__()
								UpperCAmelCase__					:   Optional[Any]       				=       n_e
								UpperCAmelCase__					:   List[Any]       				=       vq_embed_dim
								UpperCAmelCase__					:   Dict       				=       beta
								UpperCAmelCase__					:   List[Any]       				=       legacy

								UpperCAmelCase__					:   str       				=       nn.Embedding(self.n_e							,							self.vq_embed_dim)
								self.embedding.weight.data.uniform_(-1.0 / self.n_e							,							1.0 / self.n_e)

								UpperCAmelCase__					:   str       				=       remap
								if self.remap is not None:
											self.register_buffer("""used"""							,							torch.tensor(np.load(self.remap)))
											UpperCAmelCase__					:   List[Any]       				=       self.used.shape[0]
											UpperCAmelCase__					:   Dict       				=       unknown_index  # "random" or "extra" or integer
											if self.unknown_index == "extra":
														UpperCAmelCase__					:   Union[str, Any]       				=       self.re_embed
														UpperCAmelCase__					:   List[Any]       				=       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:
											UpperCAmelCase__					:   Optional[Any]       				=       n_e

								UpperCAmelCase__					:   Tuple       				=       sane_index_shape




					def 			snake_case__  ( self							,							_lowerCamelCase):
								UpperCAmelCase__					:   str       				=       inds.shape
								assert len(_A) > 1
								UpperCAmelCase__					:   Union[str, Any]       				=       inds.reshape(ishape[0]							,							-1)
								UpperCAmelCase__					:   Tuple       				=       self.used.to(_A)
								UpperCAmelCase__					:   Dict       				=       (inds[:, :, None] == used[None, None, ...]).long()
								UpperCAmelCase__					:   List[Any]       				=       match.argmax(-1)
								UpperCAmelCase__					:   Optional[int]       				=       match.sum(2) < 1
								if self.unknown_index == "random":
											UpperCAmelCase__					:   Union[str, Any]       				=       torch.randint(0							,							self.re_embed							,							size=new[unknown].shape).to(device=new.device)
								else:
											UpperCAmelCase__					:   Tuple       				=       self.unknown_index
								return new.reshape(_A)




					def 			snake_case__  ( self							,							_lowerCamelCase):
								UpperCAmelCase__					:   Tuple       				=       inds.shape
								assert len(_A) > 1
								UpperCAmelCase__					:   str       				=       inds.reshape(ishape[0]							,							-1)
								UpperCAmelCase__					:   Optional[Any]       				=       self.used.to(_A)
								if self.re_embed > self.used.shape[0]:  # extra token
											UpperCAmelCase__					:   Optional[Any]       				=       0  # simply set to zero
								UpperCAmelCase__					:   List[Any]       				=       torch.gather(used[None, :][inds.shape[0] * [0], :]							,							1							,							_A)
								return back.reshape(_A)




					def 			snake_case__  ( self							,							_lowerCamelCase):
								UpperCAmelCase__					:   str       				=       z.permute(0							,							2							,							3							,							1).contiguous()
								UpperCAmelCase__					:   Optional[Any]       				=       z.view(-1							,							self.vq_embed_dim)

								# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
								UpperCAmelCase__					:   int       				=       torch.argmin(torch.cdist(_A							,							self.embedding.weight)							,							dim=1)

								UpperCAmelCase__					:   Optional[int]       				=       self.embedding(_A).view(z.shape)
								UpperCAmelCase__					:   int       				=       None
								UpperCAmelCase__					:   List[Any]       				=       None

								# compute loss for embedding
								if not self.legacy:
											UpperCAmelCase__					:   int       				=       self.beta * torch.mean((z_q.detach() - z) ** 2) + torch.mean((z_q - z.detach()) ** 2)
								else:
											UpperCAmelCase__					:   int       				=       torch.mean((z_q.detach() - z) ** 2) + self.beta * torch.mean((z_q - z.detach()) ** 2)

								# preserve gradients
								UpperCAmelCase__					:   Tuple       				=       z + (z_q - z).detach()

								# reshape back to match original input shape
								UpperCAmelCase__					:   Union[str, Any]       				=       z_q.permute(0							,							3							,							1							,							2).contiguous()

								if self.remap is not None:
											UpperCAmelCase__					:   int       				=       min_encoding_indices.reshape(z.shape[0]							,							-1)  # add batch axis
											UpperCAmelCase__					:   Optional[Any]       				=       self.remap_to_used(_A)
											UpperCAmelCase__					:   List[str]       				=       min_encoding_indices.reshape(-1							,							1)  # flatten

								if self.sane_index_shape:
											UpperCAmelCase__					:   Any       				=       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 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase):


								if self.remap is not None:
											UpperCAmelCase__					:   Optional[int]       				=       indices.reshape(shape[0]							,							-1)  # add batch axis
											UpperCAmelCase__					:   Optional[int]       				=       self.unmap_to_all(_A)
											UpperCAmelCase__					:   Optional[Any]       				=       indices.reshape(-1)  # flatten again

								# get quantized latent vectors
								UpperCAmelCase__					:   List[Any]       				=       self.embedding(_A)

								if shape is not None:
											UpperCAmelCase__					:   Union[str, Any]       				=       z_q.view(_A)
											# reshape back to match original input shape
											UpperCAmelCase__					:   Tuple       				=       z_q.permute(0							,							3							,							1							,							2).contiguous()

								return z_q


class 						_snake_case   (       snake_case_		):




					def __init__( self							,							_lowerCamelCase							,							_lowerCamelCase=False):
								UpperCAmelCase__					:   int       				=       parameters
								UpperCAmelCase__					:   Optional[int]       				=       torch.chunk(_A							,							2							,							dim=1)
								UpperCAmelCase__					:   str       				=       torch.clamp(self.logvar							,							-30.0							,							20.0)
								UpperCAmelCase__					:   Any       				=       deterministic
								UpperCAmelCase__					:   int       				=       torch.exp(0.5 * self.logvar)
								UpperCAmelCase__					:   Optional[int]       				=       torch.exp(self.logvar)
								if self.deterministic:
											UpperCAmelCase__					:   Union[str, Any]       				=       torch.zeros_like(
											    self.mean							,							device=self.parameters.device							,							dtype=self.parameters.dtype)




					def 			snake_case__  ( self							,							_lowerCamelCase = None):
								UpperCAmelCase__					:   Union[str, Any]       				=       randn_tensor(
								    self.mean.shape							,							generator=_A							,							device=self.parameters.device							,							dtype=self.parameters.dtype)
								UpperCAmelCase__					:   Optional[int]       				=       self.mean + self.std * sample
								return x




					def 			snake_case__  ( self							,							_lowerCamelCase=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 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase=[1, 2, 3]):


								if self.deterministic:
											return torch.Tensor([0.0])
								UpperCAmelCase__					:   int       				=       np.log(2.0 * np.pi)
								return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean							,							2) / self.var							,							dim=_A)




					def 			snake_case__  ( self):
								return self.mean



'''simple docstring'''



import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE


class 						_snake_case   (       a__		):
					lowerCAmelCase :Optional[int]						=   ''''''
					lowerCAmelCase :str						=   (
					    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
					)
					lowerCAmelCase :str						=   None  # compression type in fsspec. ex: "gzip"
					lowerCAmelCase :str						=   None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz




					def __init__( self							,							_lowerCamelCase = ""							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							**_lowerCamelCase):
								super().__init__(self							,							**_lowerCamelCase)
								# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
								UpperCAmelCase__					:   Optional[Any]       				=       fsspec.open(
								    _lowerCamelCase							,							mode="""rb"""							,							protocol=_lowerCamelCase							,							compression=self.compression							,							client_kwargs={
								        """requote_redirect_url""": False,  # see https://github.com/huggingface/datasets/pull/5459
								        """trust_env""": True,  # Enable reading proxy env variables.
								        **(target_options or {}).pop("""client_kwargs"""							,							{}),  # To avoid issues if it was already passed.
								    }							,							**(target_options or {})							,							)
								UpperCAmelCase__					:   List[Any]       				=       os.path.basename(self.file.path.split("""::""")[0])
								UpperCAmelCase__					:   Dict       				=       (
								    self.compressed_name[: self.compressed_name.rindex(""".""")]
								    if """.""" in self.compressed_name
								    else self.compressed_name
								)
								UpperCAmelCase__					:   Tuple       				=       None




					@classmethod
					def 			snake_case__  ( cls							,							_lowerCamelCase):
								# compressed file paths are always relative to the archive root
								return super()._strip_protocol(_lowerCamelCase).lstrip("""/""")




					def 			snake_case__  ( self):
								if self.dir_cache is None:
											UpperCAmelCase__					:   Optional[Any]       				=       {**self.file.fs.info(self.file.path), """name""": self.uncompressed_name}
											UpperCAmelCase__					:   Union[str, Any]       				=       {f["""name"""]: f}




					def 			snake_case__  ( self							,							_lowerCamelCase):
								return self.file.open().read()




					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase = "rb"							,							_lowerCamelCase=None							,							_lowerCamelCase=True							,							_lowerCamelCase=None							,							**_lowerCamelCase							,							):
								UpperCAmelCase__					:   List[str]       				=       self._strip_protocol(_lowerCamelCase)
								if mode != "rb":
											raise ValueError(f'''Tried to read with mode {mode} on file {self.file.path} opened with mode \'rb\'''')
								return self.file.open()


class 						_snake_case   (       a__		):
					lowerCAmelCase :Dict						=   '''bz2'''
					lowerCAmelCase :List[str]						=   '''bz2'''
					lowerCAmelCase :Dict						=   '''.bz2'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :int						=   '''gzip'''
					lowerCAmelCase :Tuple						=   '''gzip'''
					lowerCAmelCase :str						=   '''.gz'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :List[str]						=   '''lz4'''
					lowerCAmelCase :Any						=   '''lz4'''
					lowerCAmelCase :int						=   '''.lz4'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :Union[str, Any]						=   '''xz'''
					lowerCAmelCase :int						=   '''xz'''
					lowerCAmelCase :List[Any]						=   '''.xz'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :Tuple						=   '''zstd'''
					lowerCAmelCase :List[str]						=   '''zstd'''
					lowerCAmelCase :Union[str, Any]						=   '''.zst'''




					def __init__( self							,							_lowerCamelCase							,							_lowerCamelCase = "rb"							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = DEFAULT_BLOCK_SIZE							,							**_lowerCamelCase							,							):
								super().__init__(
								    fo=_lowerCamelCase							,							mode=_lowerCamelCase							,							target_protocol=_lowerCamelCase							,							target_options=_lowerCamelCase							,							block_size=_lowerCamelCase							,							**_lowerCamelCase							,							)
								# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
								#
								# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
								#     out.close = close
								# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
								#
								# see https://github.com/intake/filesystem_spec/issues/725
								UpperCAmelCase__					:   Dict       				=       self.file.__enter__


								class 						_snake_case   :




													def __init__( self							,							_lowerCamelCase):
																UpperCAmelCase__					:   Optional[int]       				=       file_




													def __enter__( self):
																self._file.__enter__()
																return self




													def __exit__( self							,							*_lowerCamelCase							,							**_lowerCamelCase):
																self._file.__exit__(*_lowerCamelCase							,							**_lowerCamelCase)




													def __iter__( self):
																return iter(self._file)




													def 			snake_case__  ( self):
																return next(self._file)




													def __getattr__( self							,							_lowerCamelCase):
																return getattr(self._file							,							_lowerCamelCase)

								def fixed_enter(*_lowerCamelCase							,							**_lowerCamelCase):
											return WrappedFile(_enter(*_lowerCamelCase							,							**_lowerCamelCase))

								UpperCAmelCase__					:   List[Any]       				=       fixed_enter