


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     a		(_lowerCAmelCase				):




							"""simple docstring"""


							__UpperCAmelCase   :							"DiagonalGaussianDistribution"

class     a		(_lowerCAmelCase					,    _lowerCAmelCase				):




							"""simple docstring"""


							__UpperCAmelCase   :							Any					  =  True

							@register_to_config
							def __init__(							self	:  Optional[int]   ,     lowerCamelCase	:  int = 3   ,     lowerCamelCase	:  int = 3   ,     lowerCamelCase	:  Tuple[str] = ("DownEncoderBlock2D",)   ,     lowerCamelCase	:  Tuple[str] = ("UpDecoderBlock2D",)   ,     lowerCamelCase	:  Tuple[int] = (64,)   ,     lowerCamelCase	:  int = 1   ,     lowerCamelCase	:  str = "silu"   ,     lowerCamelCase	:  int = 4   ,     lowerCamelCase	:  int = 32   ,     lowerCamelCase	:  int = 32   ,     lowerCamelCase	:  float = 0.1_82_15   ,     )     ->   Union[str, Any]:
														super().__init__()

														# pass init params to Encoder
														__snake_case		:      Dict										=       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		:      List[str]										=       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		:      List[str]										=       nn.Convad(2 * latent_channels   ,     2 * latent_channels   ,     1    )
														__snake_case		:      int										=       nn.Convad(lowerCamelCase   ,     lowerCamelCase   ,     1    )

														__snake_case		:      str										=       False
														__snake_case		:      List[Any]										=       False

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

							def    __snake_case (							self	:  Union[str, Any]   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  Dict=False    )     ->   Tuple:
														if isinstance(lowerCamelCase   ,     (Encoder, Decoder)    ):
																					__snake_case		:      int										=       value

							def    __snake_case (							self	:  Tuple   ,     lowerCamelCase	:  bool = True    )     ->   Optional[int]:
														__snake_case		:      int										=       use_tiling

							def    __snake_case (							self	:  Dict    )     ->   Tuple:
														self.enable_tiling(lowerCamelCase    )

							def    __snake_case (							self	:  Optional[Any]    )     ->   List[str]:
														__snake_case		:      List[str]										=       True

							def    __snake_case (							self	:  Union[str, Any]    )     ->   Union[str, Any]:
														__snake_case		:      Dict										=       False

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

														def fn_recursive_add_processors(lowerCamelCase	:  str   ,     lowerCamelCase	:  torch.nn.Module   ,     lowerCamelCase	:  Dict[str, AttentionProcessor]    ):
																					if hasattr(lowerCamelCase   ,     "set_processor"    ):
																												__snake_case		:      Optional[Any]										=       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	:  Tuple   ,     lowerCamelCase	:  Union[AttentionProcessor, Dict[str, AttentionProcessor]]    )     ->   Tuple:
														__snake_case		:      int										=       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	:  str   ,     lowerCamelCase	:  torch.nn.Module   ,     lowerCamelCase	:  Optional[Any]    ):
																					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	:  Optional[int]    )     ->   Union[str, Any]:
														self.set_attn_processor(AttnProcessor()    )

							@apply_forward_hook
							def    __snake_case (							self	:  Tuple   ,     lowerCamelCase	:  torch.FloatTensor   ,     lowerCamelCase	:  bool = 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(lowerCamelCase   ,     return_dict=lowerCamelCase    )

														if self.use_slicing and x.shape[0] > 1:
																					__snake_case		:      Dict										=       [self.encoder(lowerCamelCase    ) for x_slice in x.split(1    )]
																					__snake_case		:      Tuple										=       torch.cat(lowerCamelCase    )
														else:
																					__snake_case		:      List[str]										=       self.encoder(lowerCamelCase    )

														__snake_case		:      Optional[int]										=       self.quant_conv(lowerCamelCase    )
														__snake_case		:      Tuple										=       DiagonalGaussianDistribution(lowerCamelCase    )

														if not return_dict:
																					return (posterior,)

														return AutoencoderKLOutput(latent_dist=lowerCamelCase    )

							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  torch.FloatTensor   ,     lowerCamelCase	:  bool = 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(lowerCamelCase   ,     return_dict=lowerCamelCase    )

														__snake_case		:      Dict										=       self.post_quant_conv(lowerCamelCase    )
														__snake_case		:      int										=       self.decoder(lowerCamelCase    )

														if not return_dict:
																					return (dec,)

														return DecoderOutput(sample=lowerCamelCase    )

							@apply_forward_hook
							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  torch.FloatTensor   ,     lowerCamelCase	:  bool = True    )     ->   Union[DecoderOutput, torch.FloatTensor]:
														if self.use_slicing and z.shape[0] > 1:
																					__snake_case		:      Dict										=       [self._decode(lowerCamelCase    ).sample for z_slice in z.split(1    )]
																					__snake_case		:      List[Any]										=       torch.cat(lowerCamelCase    )
														else:
																					__snake_case		:      Any										=       self._decode(lowerCamelCase    ).sample

														if not return_dict:
																					return (decoded,)

														return DecoderOutput(sample=lowerCamelCase    )

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

							def    __snake_case (							self	:  str   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  Optional[Any]    )     ->   List[str]:
														__snake_case		:      Union[str, Any]										=       min(a.shape[3]   ,     b.shape[3]   ,     lowerCamelCase    )
														for x in range(lowerCamelCase    ):
																					__snake_case		:      Optional[int]										=       a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
														return b

							def    __snake_case (							self	:  Dict   ,     lowerCamelCase	:  torch.FloatTensor   ,     lowerCamelCase	:  bool = True    )     ->   AutoencoderKLOutput:
														__snake_case		:      List[Any]										=       int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)    )
														__snake_case		:      List[Any]										=       int(self.tile_latent_min_size * self.tile_overlap_factor    )
														__snake_case		:      Union[str, Any]										=       self.tile_latent_min_size - blend_extent

														# Split the image into 512x512 tiles and encode them separately.
														__snake_case		:      int										=       []
														for i in range(0   ,     x.shape[2]   ,     lowerCamelCase    ):
																					__snake_case		:      Tuple										=       []
																					for j in range(0   ,     x.shape[3]   ,     lowerCamelCase    ):
																												__snake_case		:      Any										=       x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
																												__snake_case		:      Tuple										=       self.encoder(lowerCamelCase    )
																												__snake_case		:      Union[str, Any]										=       self.quant_conv(lowerCamelCase    )
																												row.append(lowerCamelCase    )
																					rows.append(lowerCamelCase    )
														__snake_case		:      Optional[int]										=       []
														for i, row in enumerate(lowerCamelCase    ):
																					__snake_case		:      Optional[int]										=       []
																					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		:      Optional[int]										=       self.blend_v(rows[i - 1][j]   ,     lowerCamelCase   ,     lowerCamelCase    )
																												if j > 0:
																																			__snake_case		:      List[Any]										=       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		:      str										=       torch.cat(lowerCamelCase   ,     dim=2    )
														__snake_case		:      Union[str, Any]										=       DiagonalGaussianDistribution(lowerCamelCase    )

														if not return_dict:
																					return (posterior,)

														return AutoencoderKLOutput(latent_dist=lowerCamelCase    )

							def    __snake_case (							self	:  Optional[Any]   ,     lowerCamelCase	:  torch.FloatTensor   ,     lowerCamelCase	:  bool = True    )     ->   Union[DecoderOutput, torch.FloatTensor]:
														__snake_case		:      Optional[int]										=       int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)    )
														__snake_case		:      List[Any]										=       int(self.tile_sample_min_size * self.tile_overlap_factor    )
														__snake_case		:      Optional[int]										=       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		:      Tuple										=       []
														for i in range(0   ,     z.shape[2]   ,     lowerCamelCase    ):
																					__snake_case		:      List[Any]										=       []
																					for j in range(0   ,     z.shape[3]   ,     lowerCamelCase    ):
																												__snake_case		:      List[Any]										=       z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
																												__snake_case		:      List[str]										=       self.post_quant_conv(lowerCamelCase    )
																												__snake_case		:      Dict										=       self.decoder(lowerCamelCase    )
																												row.append(lowerCamelCase    )
																					rows.append(lowerCamelCase    )
														__snake_case		:      Optional[Any]										=       []
														for i, row in enumerate(lowerCamelCase    ):
																					__snake_case		:      List[Any]										=       []
																					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		:      Tuple										=       self.blend_v(rows[i - 1][j]   ,     lowerCamelCase   ,     lowerCamelCase    )
																												if j > 0:
																																			__snake_case		:      Optional[Any]										=       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		:      List[str]										=       torch.cat(lowerCamelCase   ,     dim=2    )
														if not return_dict:
																					return (dec,)

														return DecoderOutput(sample=lowerCamelCase    )






							def    __snake_case (							self	:  Tuple   ,     lowerCamelCase	:  torch.FloatTensor   ,     lowerCamelCase	:  bool = False   ,     lowerCamelCase	:  bool = True   ,     lowerCamelCase	:  Optional[torch.Generator] = None   ,     )     ->   Union[DecoderOutput, torch.FloatTensor]:
														__snake_case		:      int										=       sample
														__snake_case		:      List[Any]										=       self.encode(lowerCamelCase    ).latent_dist
														if sample_posterior:
																					__snake_case		:      List[Any]										=       posterior.sample(generator=lowerCamelCase    )
														else:
																					__snake_case		:      Union[str, Any]										=       posterior.mode()
														__snake_case		:      Any										=       self.decode(lowerCamelCase    ).sample

														if not return_dict:
																					return (dec,)

														return DecoderOutput(sample=lowerCamelCase    )





from scipy.stats import spearmanr

import datasets


_snake_case					:		Optional[Any]      =    "\nThe Spearman rank-order correlation coefficient is a measure of the\nrelationship between two datasets. Like other correlation coefficients,\nthis one varies between -1 and +1 with 0 implying no correlation.\nPositive correlations imply that as data in dataset x increases, so\ndoes data in dataset y. Negative correlations imply that as x increases,\ny decreases. Correlations of -1 or +1 imply an exact monotonic relationship.\n\nUnlike the Pearson correlation, the Spearman correlation does not\nassume that both datasets are normally distributed.\n\nThe p-value roughly indicates the probability of an uncorrelated system\nproducing datasets that have a Spearman correlation at least as extreme\nas the one computed from these datasets. The p-values are not entirely\nreliable but are probably reasonable for datasets larger than 500 or so.\n"

_snake_case					:		Dict      =    "\nArgs:\n    predictions (`List[float]`): Predicted labels, as returned by a model.\n    references (`List[float]`): Ground truth labels.\n    return_pvalue (`bool`): If `True`, returns the p-value. If `False`, returns\n            only the spearmanr score. Defaults to `False`.\nReturns:\n    spearmanr (`float`): Spearman correlation coefficient.\n    p-value (`float`): p-value. **Note**: is only returned if `return_pvalue=True` is input.\nExamples:\n    Example 1:\n        >>> spearmanr_metric = datasets.load_metric(\"spearmanr\")\n        >>> results = spearmanr_metric.compute(references=[1, 2, 3, 4, 5], predictions=[10, 9, 2.5, 6, 4])\n        >>> print(results)\n        {'spearmanr': -0.7}\n\n    Example 2:\n        >>> spearmanr_metric = datasets.load_metric(\"spearmanr\")\n        >>> results = spearmanr_metric.compute(references=[1, 2, 3, 4, 5],\n        ...                                     predictions=[10, 9, 2.5, 6, 4],\n        ...                                     return_pvalue=True)\n        >>> print(results['spearmanr'])\n        -0.7\n        >>> print(round(results['spearmanr_pvalue'], 2))\n        0.19\n"

_snake_case					:		Optional[Any]      =    R"\\n@book{kokoska2000crc,\n  title={CRC standard probability and statistics tables and formulae},\n  author={Kokoska, Stephen and Zwillinger, Daniel},\n  year={2000},\n  publisher={Crc Press}\n}\n@article{2020SciPy-NMeth,\n  author  = {Virtanen, Pauli and Gommers, Ralf and Oliphant, Travis E. and\n            Haberland, Matt and Reddy, Tyler and Cournapeau, David and\n            Burovski, Evgeni and Peterson, Pearu and Weckesser, Warren and\n            Bright, Jonathan and {van der Walt}, St{\'e}fan J. and\n            Brett, Matthew and Wilson, Joshua and Millman, K. Jarrod and\n            Mayorov, Nikolay and Nelson, Andrew R. J. and Jones, Eric and\n            Kern, Robert and Larson, Eric and Carey, C J and\n            Polat, {\.I}lhan and Feng, Yu and Moore, Eric W. and\n            {VanderPlas}, Jake and Laxalde, Denis and Perktold, Josef and\n            Cimrman, Robert and Henriksen, Ian and Quintero, E. A. and\n            Harris, Charles R. and Archibald, Anne M. and\n            Ribeiro, Ant{\^o}nio H. and Pedregosa, Fabian and\n            {van Mulbregt}, Paul and {SciPy 1.0 Contributors}},\n  title   = {{{SciPy} 1.0: Fundamental Algorithms for Scientific\n            Computing in Python}},\n  journal = {Nature Methods},\n  year    = {2020},\n  volume  = {17},\n  pages   = {261--272},\n  adsurl  = {https://rdcu.be/b08Wh},\n  doi     = {10.1038/s41592-019-0686-2},\n}\n"

@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION					,    _KWARGS_DESCRIPTION				)
class     a		(datasets.Metric				):




							"""simple docstring"""


							def    __snake_case (							self	:  Optional[int]    )     ->   Any:
														return datasets.MetricInfo(
														    description=_DESCRIPTION   ,     citation=_CITATION   ,     inputs_description=_KWARGS_DESCRIPTION   ,     features=datasets.Features(
														        {
														            "predictions": datasets.Value("float"    ),
														            "references": datasets.Value("float"    ),
														        }    )   ,     reference_urls=["https://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.spearmanr.html"]   ,     )






							def    __snake_case (							self	:  Optional[Any]   ,     lowerCamelCase	:  Optional[int]   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  Tuple=False    )     ->   Dict:
														__snake_case		:      Optional[int]										=       spearmanr(lowerCamelCase   ,     lowerCamelCase    )
														if return_pvalue:
																					return {"spearmanr": results[0], "spearmanr_pvalue": results[1]}
														else:
																					return {"spearmanr": results[0]}

