





import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, ByTaTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
      __UpperCAmelCase	:       List[str]    	=							"pt"
elif is_tf_available():
      __UpperCAmelCase	:       int    	=							"tf"
else:
      __UpperCAmelCase	:       List[Any]    	=							"jax"







class    __snake_case	(							__lowerCamelCase       ,      unittest.TestCase      ):





       '''simple docstring'''







       lowerCAmelCase__						=					ByTaTokenizer
       lowerCAmelCase__						=					False




       def 							UpperCAmelCase__			(  self			:   Any		):
         super().setUp()
         __snake_case:						str			  =						ByTaTokenizer()
         tokenizer.save_pretrained(self.tmpdirname		)




       @cached_property
       def 							UpperCAmelCase__			(  self			:   Any		):
         return ByTaTokenizer.from_pretrained("""google/byt5-small"""		)




       def 							UpperCAmelCase__			(  self			:   str							,    **A			:   List[str]		):
         return self.tokenizer_class.from_pretrained(self.tmpdirname							,    **A		)




       def 							UpperCAmelCase__			(  self			:   List[Any]							,    A			:   Dict							,    A			:   List[str]=False							,    A			:   Union[str, Any]=20							,    A			:   int=5		):
         # XXX The default common tokenizer tests assume that every ID is decodable on its own.
         # This assumption is invalid for ByT5 because single bytes might not be
         # valid utf-8 (byte 128 for instance).
         # Here we're overriding the smallest possible method to provide
         # a clean sequence without making the same assumption.

         __snake_case:						str			  =						[]
         for i in range(len(A		)		):
           try:
             __snake_case:						List[Any]			  =						tokenizer.decode([i]							,    clean_up_tokenization_spaces=A		)
           except UnicodeDecodeError:
             pass
           toks.append((i, tok)		)

         __snake_case:						str			  =						list(filter(lambda A		: re.match(r"""^[ a-zA-Z]+$"""							,    t[1]		)							,    A		)		)
         __snake_case:						Dict			  =						list(filter(lambda A		: [t[0]] == tokenizer.encode(t[1]							,    add_special_tokens=A		)							,    A		)		)
         if max_length is not None and len(A		) > max_length:
           __snake_case:						Any			  =						toks[:max_length]
         if min_length is not None and len(A		) < min_length and len(A		) > 0:
           while len(A		) < min_length:
             __snake_case:						List[Any]			  =						toks + toks
        # toks_str = [t[1] for t in toks]
         __snake_case:						Optional[Any]			  =						[t[0] for t in toks]

         # Ensure consistency
         __snake_case:						Dict			  =						tokenizer.decode(A							,    clean_up_tokenization_spaces=A		)
         if " " not in output_txt and len(A		) > 1:
           __snake_case:						Dict			  =						(
               tokenizer.decode([toks_ids[0]]							,    clean_up_tokenization_spaces=A		)
               + """ """
               + tokenizer.decode(toks_ids[1:]							,    clean_up_tokenization_spaces=A		)
           )
         if with_prefix_space:
           __snake_case:						Any			  =						""" """ + output_txt
         __snake_case:						Union[str, Any]			  =						tokenizer.encode(A							,    add_special_tokens=A		)
         return output_txt, output_ids




       def 							UpperCAmelCase__			(  self			:   Optional[int]		):
         __snake_case:						Any			  =						self.ta_base_tokenizer
         __snake_case:						Union[str, Any]			  =						tokenizer(["""hi</s>""", """I went to the gym</s>""", """</s>"""]		)
         __snake_case:						List[str]			  =						tokenizer(["""hi""", """I went to the gym""", """"""]		)
         self.assertListEqual(batch_with_eos_added["""input_ids"""]							,    batch_without_eos_added["""input_ids"""]		)




       def 							UpperCAmelCase__			(  self			:   str		):
         __snake_case:						Optional[int]			  =						self.ta_base_tokenizer
         __snake_case:						Dict			  =						"""Unicode €."""
         __snake_case:						Optional[Any]			  =						tokenizer(A		)
         __snake_case:						List[str]			  =						[88, 113, 108, 102, 114, 103, 104, 35, 229, 133, 175, 49, 1]
         self.assertEqual(encoded["""input_ids"""]							,    A		)

         # decoding
         __snake_case:						str			  =						tokenizer.decode(A		)
         self.assertEqual(A							,    """Unicode €.</s>"""		)

         __snake_case:						str			  =						tokenizer("""e è é ê ë"""		)
         __snake_case:						List[Any]			  =						[104, 35, 198, 171, 35, 198, 172, 35, 198, 173, 35, 198, 174, 1]
         self.assertEqual(encoded["""input_ids"""]							,    A		)
         # decoding
         __snake_case:						Optional[int]			  =						tokenizer.decode(A		)
         self.assertEqual(A							,    """e è é ê ë</s>"""		)

         # encode/decode, but with `encode` instead of `__call__`
         self.assertEqual(tokenizer.decode(tokenizer.encode("""e è é ê ë"""		)		)							,    """e è é ê ë</s>"""		)




       def 							UpperCAmelCase__			(  self			:   Optional[int]		):
         __snake_case:						Any			  =						self.ta_base_tokenizer
         __snake_case:						Union[str, Any]			  =						["""A long paragraph for summarization.""", """Another paragraph for summarization."""]
         # fmt: off
         __snake_case:						List[str]			  =						[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 1, 0]
         # fmt: on
         __snake_case:						Union[str, Any]			  =						tokenizer(A							,    padding=A							,    return_tensors=A		)
         self.assertIsInstance(A							,    A		)

         if FRAMEWORK != "jax":
           __snake_case:						int			  =						list(batch.input_ids.numpy()[0]		)
         else:
           __snake_case:						List[str]			  =						list(batch.input_ids.tolist()[0]		)

         self.assertListEqual(A							,    A		)

         self.assertEqual((2, 37)							,    batch.input_ids.shape		)
         self.assertEqual((2, 37)							,    batch.attention_mask.shape		)




       def 							UpperCAmelCase__			(  self			:   Optional[int]		):
         __snake_case:						Union[str, Any]			  =						self.ta_base_tokenizer
         __snake_case:						Tuple			  =						["""A long paragraph for summarization.""", """Another paragraph for summarization."""]
         __snake_case:						Dict			  =						tokenizer(A							,    padding=A							,    return_tensors=A		)
         # check if input_ids are returned and no decoder_input_ids
         self.assertIn("""input_ids"""							,    A		)
         self.assertIn("""attention_mask"""							,    A		)
         self.assertNotIn("""decoder_input_ids"""							,    A		)
         self.assertNotIn("""decoder_attention_mask"""							,    A		)




       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         __snake_case:						Optional[Any]			  =						self.ta_base_tokenizer
         __snake_case:						str			  =						[
             """Summary of the text.""",
             """Another summary.""",
         ]
         __snake_case:						Optional[int]			  =						tokenizer(
             text_target=A							,    max_length=32							,    padding="""max_length"""							,    truncation=A							,    return_tensors=A		)
         self.assertEqual(32							,    targets["""input_ids"""].shape[1]		)




       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         __snake_case:						Tuple			  =						self.ta_base_tokenizer
         __snake_case:						List[str]			  =						["""A long paragraph for summarization. </s>"""]
         __snake_case:						Optional[Any]			  =						["""Summary of the text. </s>"""]
         # fmt: off
         __snake_case:						str			  =						[68, 35, 111, 114, 113, 106, 35, 115, 100, 117, 100, 106, 117, 100, 115, 107, 35, 105, 114, 117, 35, 118, 120, 112, 112, 100, 117, 108, 125, 100, 119, 108, 114, 113, 49, 35, 1]
         __snake_case:						int			  =						[86, 120, 112, 112, 100, 117, 124, 35, 114, 105, 35, 119, 107, 104, 35, 119, 104, 123, 119, 49, 35, 1]
         # fmt: on

         __snake_case:						Optional[Any]			  =						tokenizer(A							,    text_target=A		)

         self.assertEqual(A							,    batch["""input_ids"""][0]		)
         self.assertEqual(A							,    batch["""labels"""][0]		)




       def 							UpperCAmelCase__			(  self			:   List[str]		):
         # safety check on max_len default value so we are sure the test works
         __snake_case:						int			  =						self.get_tokenizers()
         for tokenizer in tokenizers:
           with self.subTest(f'''{tokenizer.__class__.__name__}'''		):
             self.assertNotEqual(tokenizer.model_max_length							,    42		)

        # Now let's start the test
         __snake_case:						Tuple			  =						self.get_tokenizers()
         for tokenizer in tokenizers:
           with self.subTest(f'''{tokenizer.__class__.__name__}'''		):
             # Isolate this from the other tests because we save additional tokens/etc
             __snake_case:						Tuple			  =						tempfile.mkdtemp()

             __snake_case:						List[str]			  =						""" He is very happy, UNwant\u00E9d,running"""
             __snake_case:						int			  =						tokenizer.encode(A							,    add_special_tokens=A		)
             tokenizer.save_pretrained(A		)

             __snake_case:						List[Any]			  =						tokenizer.__class__.from_pretrained(A		)
             __snake_case:						Optional[Any]			  =						after_tokenizer.encode(A							,    add_special_tokens=A		)
             self.assertListEqual(A							,    A		)

             shutil.rmtree(A		)

         __snake_case:						Optional[Any]			  =						self.get_tokenizers(model_max_length=42		)
         for tokenizer in tokenizers:
           with self.subTest(f'''{tokenizer.__class__.__name__}'''		):
             # Isolate this from the other tests because we save additional tokens/etc
             __snake_case:						List[Any]			  =						tempfile.mkdtemp()

             __snake_case:						str			  =						""" He is very happy, UNwant\u00E9d,running"""
             tokenizer.add_tokens(["""bim""", """bambam"""]		)
             __snake_case:						str			  =						tokenizer.additional_special_tokens
             additional_special_tokens.append("""new_additional_special_token"""		)
             tokenizer.add_special_tokens({"""additional_special_tokens""": additional_special_tokens}		)
             __snake_case:						str			  =						tokenizer.encode(A							,    add_special_tokens=A		)
             tokenizer.save_pretrained(A		)

             __snake_case:						str			  =						tokenizer.__class__.from_pretrained(A		)
             __snake_case:						List[Any]			  =						after_tokenizer.encode(A							,    add_special_tokens=A		)
             self.assertListEqual(A							,    A		)
             self.assertIn("""new_additional_special_token"""							,    after_tokenizer.additional_special_tokens		)
             self.assertEqual(after_tokenizer.model_max_length							,    42		)

             __snake_case:						Dict			  =						tokenizer.__class__.from_pretrained(A							,    model_max_length=43		)
             self.assertEqual(tokenizer.model_max_length							,    43		)

             shutil.rmtree(A		)




       def 							UpperCAmelCase__			(  self			:   Union[str, Any]		):
         __snake_case:						Dict			  =						[]
         if self.test_slow_tokenizer:
           tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())		)

         if self.test_rust_tokenizer:
           tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())		)

         for tokenizer_class, tokenizer_utils in tokenizer_list:
           with tempfile.TemporaryDirectory() as tmp_dir:
             tokenizer_utils.save_pretrained(A		)

             with open(os.path.join(A							,    """special_tokens_map.json"""		)							,    encoding="""utf-8"""		) as json_file:
               __snake_case:						int			  =						json.load(A		)

             with open(os.path.join(A							,    """tokenizer_config.json"""		)							,    encoding="""utf-8"""		) as json_file:
               __snake_case:						Dict			  =						json.load(A		)

             __snake_case:						List[str]			  =						[f'''<extra_id_{i}>''' for i in range(125		)]

             __snake_case:						Any			  =						added_tokens_extra_ids + [
                 """an_additional_special_token"""
             ]
             __snake_case:						Dict			  =						added_tokens_extra_ids + [
                 """an_additional_special_token"""
             ]

             with open(os.path.join(A							,    """special_tokens_map.json"""		)							,    """w"""							,    encoding="""utf-8"""		) as outfile:
               json.dump(A							,    A		)
             with open(os.path.join(A							,    """tokenizer_config.json"""		)							,    """w"""							,    encoding="""utf-8"""		) as outfile:
               json.dump(A							,    A		)

             # the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
             # into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
             # "special_tokens_map.json" files
             __snake_case:						Tuple			  =						tokenizer_class.from_pretrained(
                 A							,    )
             self.assertIn(
                 """an_additional_special_token"""							,    tokenizer_without_change_in_init.additional_special_tokens		)
             # self.assertIn("an_additional_special_token",tokenizer_without_change_in_init.get_vocab()) # ByT5Tokenization no vocab
             self.assertEqual(
                 ["""an_additional_special_token"""]							,    tokenizer_without_change_in_init.convert_ids_to_tokens(
                     tokenizer_without_change_in_init.convert_tokens_to_ids(["""an_additional_special_token"""]		)		)							,    )

             # Now we test that we can change the value of additional_special_tokens in the from_pretrained
             __snake_case:						Union[str, Any]			  =						added_tokens_extra_ids + [AddedToken("""a_new_additional_special_token"""							,    lstrip=A		)]
             __snake_case:						str			  =						tokenizer_class.from_pretrained(
                 A							,    additional_special_tokens=A							,    )

             self.assertIn("""a_new_additional_special_token"""							,    tokenizer.additional_special_tokens		)
             self.assertEqual(
                 ["""a_new_additional_special_token"""]							,    tokenizer.convert_ids_to_tokens(
                     tokenizer.convert_tokens_to_ids(["""a_new_additional_special_token"""]		)		)							,    )




       def 							UpperCAmelCase__			(  self			:   Union[str, Any]		):
         __snake_case:						Any			  =						[]
         if self.test_slow_tokenizer:
           tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())		)

         if self.test_rust_tokenizer:
           tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())		)

         for tokenizer_class, tokenizer_utils in tokenizer_list:
           with tempfile.TemporaryDirectory() as tmp_dir:
             tokenizer_utils.save_pretrained(A		)

             __snake_case:						str			  =						tokenizer_class.from_pretrained(A		)

             self.assertTrue(tokenizer.decode([255]		) == """"""		)




       def 							UpperCAmelCase__			(  self			:   Union[str, Any]		):
         pass




       def 							UpperCAmelCase__			(  self			:   Dict		):
         pass




       def 							UpperCAmelCase__			(  self			:   Any		):
         pass




       def 							UpperCAmelCase__			(  self			:   Optional[Any]		):
         pass




       def 							UpperCAmelCase__			(  self			:   Tuple		):
         # The default common tokenizer tests uses invalid tokens for ByT5 that can only accept one-character strings
         # and special added tokens as tokens
         __snake_case:						Any			  =						self.get_tokenizers(fast=A							,    do_lower_case=A		)
         for tokenizer in tokenizers:
           with self.subTest(f'''{tokenizer.__class__.__name__}'''		):
             __snake_case:						Optional[int]			  =						["""t""", """h""", """i""", """s""", """ """, """i""", """s""", """ """, """a""", """ """, """t""", """e""", """x""", """t""", """</s>"""]
             __snake_case:						Union[str, Any]			  =						tokenizer.convert_tokens_to_string(A		)

             self.assertIsInstance(A							,    A		)






       def 							UpperCAmelCase__			(  self			:   Optional[Any]		):
         __snake_case:						Union[str, Any]			  =						self.get_tokenizers()
         for tokenizer in tokenizers:
           with self.subTest(f'''{tokenizer.__class__.__name__}'''		):
             __snake_case:						List[Any]			  =						[
                 """bos_token""",
                 """eos_token""",
                 """unk_token""",
                 """sep_token""",
                 """pad_token""",
                 """cls_token""",
                 """mask_token""",
             ]

             __snake_case:						Optional[int]			  =						0
             __snake_case:						List[Any]			  =						tokenizer.convert_ids_to_tokens(
                 A							,    skip_special_tokens=A		)

             for attr in attributes_list:
               setattr(A							,    attr + """_id"""							,    A		)
               self.assertEqual(getattr(A							,    A		)							,    A		)
               self.assertEqual(getattr(A							,    attr + """_id"""		)							,    A		)

               setattr(A							,    attr + """_id"""							,    A		)
               self.assertEqual(getattr(A							,    A		)							,    A		)
               self.assertEqual(getattr(A							,    attr + """_id"""		)							,    A		)

             setattr(A							,    """additional_special_tokens_ids"""							,    []		)
             self.assertListEqual(getattr(A							,    """additional_special_tokens"""		)							,    []		)
             self.assertListEqual(getattr(A							,    """additional_special_tokens_ids"""		)							,    []		)

             setattr(A							,    """additional_special_tokens_ids"""							,    [token_id_to_test_setters]		)
             self.assertListEqual(getattr(A							,    """additional_special_tokens"""		)							,    [token_to_test_setters]		)
             self.assertListEqual(getattr(A							,    """additional_special_tokens_ids"""		)							,    [token_id_to_test_setters]		)













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    __snake_case	(							__lowerCamelCase      ):





       '''simple docstring'''







       lowerCAmelCase__						=					42







class    __snake_case	(							__lowerCamelCase       ,      __lowerCamelCase      ):





       '''simple docstring'''







       lowerCAmelCase__						=					True




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

         # pass init params to Encoder
         __snake_case:						Any			  =						Encoder(
             in_channels=A							,    out_channels=A							,    down_block_types=A							,    block_out_channels=A							,    layers_per_block=A							,    act_fn=A							,    norm_num_groups=A							,    double_z=A							,    )

         # pass init params to Decoder
         __snake_case:						int			  =						Decoder(
             in_channels=A							,    out_channels=A							,    up_block_types=A							,    block_out_channels=A							,    layers_per_block=A							,    norm_num_groups=A							,    act_fn=A							,    )

         __snake_case:						Dict			  =						nn.Convad(2 * latent_channels							,    2 * latent_channels							,    1		)
         __snake_case:						int			  =						nn.Convad(A							,    A							,    1		)

         __snake_case:						List[str]			  =						False
         __snake_case:						Optional[int]			  =						False

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




       def 							UpperCAmelCase__			(  self			:   int							,    A			:   List[str]							,    A			:   Optional[Any]=False		):
         if isinstance(A							,    (Encoder, Decoder)		):
           __snake_case:						str			  =						value




       def 							UpperCAmelCase__			(  self			:   str							,    A			:   bool = True		):
         __snake_case:						Union[str, Any]			  =						use_tiling




       def 							UpperCAmelCase__			(  self			:   Optional[int]		):
         self.enable_tiling(A		)




       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         __snake_case:						List[str]			  =						True




       def 							UpperCAmelCase__			(  self			:   List[str]		):
         __snake_case:						List[str]			  =						False




       @property
       # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
       def 							UpperCAmelCase__			(  self			:   Tuple		):
         __snake_case:						Any			  =						{}

         def fn_recursive_add_processors(A			:   str							,    A			:   torch.nn.Module							,    A			:   Dict[str, AttentionProcessor]		):
           if hasattr(A							,    """set_processor"""		):
             __snake_case:						List[Any]			  =						module.processor

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

           return processors

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

         return processors




       def 							UpperCAmelCase__			(  self			:   Optional[int]							,    A			:   Union[AttentionProcessor, Dict[str, AttentionProcessor]]		):
         __snake_case:						Any			  =						len(self.attn_processors.keys()		)

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

         def fn_recursive_attn_processor(A			:   str							,    A			:   torch.nn.Module							,    A			:   Optional[Any]		):
           if hasattr(A							,    """set_processor"""		):
             if not isinstance(A							,    A		):
               module.set_processor(A		)
             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}'''							,    A							,    A		)

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




       def 							UpperCAmelCase__			(  self			:   List[str]		):
         self.set_attn_processor(AttnProcessor()		)




       @apply_forward_hook
       def 							UpperCAmelCase__			(  self			:   Optional[Any]							,    A			:   torch.FloatTensor							,    A			:   bool = True		):
         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(A							,    return_dict=A		)

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

         __snake_case:						Any			  =						self.quant_conv(A		)
         __snake_case:						Tuple			  =						DiagonalGaussianDistribution(A		)

         if not return_dict:
           return (posterior,)

         return AutoencoderKLOutput(latent_dist=A		)




       def 							UpperCAmelCase__			(  self			:   Tuple							,    A			:   torch.FloatTensor							,    A			:   bool = True		):
         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(A							,    return_dict=A		)

         __snake_case:						Optional[int]			  =						self.post_quant_conv(A		)
         __snake_case:						Union[str, Any]			  =						self.decoder(A		)

         if not return_dict:
           return (dec,)

         return DecoderOutput(sample=A		)




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

         if not return_dict:
           return (decoded,)

         return DecoderOutput(sample=A		)




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




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




       def 							UpperCAmelCase__			(  self			:   int							,    A			:   torch.FloatTensor							,    A			:   bool = True		):
         __snake_case:						List[str]			  =						int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)		)
         __snake_case:						Dict			  =						int(self.tile_latent_min_size * self.tile_overlap_factor		)
         __snake_case:						Dict			  =						self.tile_latent_min_size - blend_extent

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

         __snake_case:						Tuple			  =						torch.cat(A							,    dim=2		)
         __snake_case:						Optional[int]			  =						DiagonalGaussianDistribution(A		)

         if not return_dict:
           return (posterior,)

         return AutoencoderKLOutput(latent_dist=A		)




       def 							UpperCAmelCase__			(  self			:   Union[str, Any]							,    A			:   torch.FloatTensor							,    A			:   bool = True		):
         __snake_case:						Optional[Any]			  =						int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)		)
         __snake_case:						str			  =						int(self.tile_sample_min_size * self.tile_overlap_factor		)
         __snake_case:						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:						List[Any]			  =						[]
         for i in range(0							,    z.shape[2]							,    A		):
           __snake_case:						Optional[Any]			  =						[]
           for j in range(0							,    z.shape[3]							,    A		):
             __snake_case:						Dict			  =						z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
             __snake_case:						Any			  =						self.post_quant_conv(A		)
             __snake_case:						Optional[Any]			  =						self.decoder(A		)
             row.append(A		)
           rows.append(A		)
         __snake_case:						Optional[Any]			  =						[]
         for i, row in enumerate(A		):
           __snake_case:						Optional[Any]			  =						[]
           for j, tile in enumerate(A		):
             # 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]							,    A							,    A		)
             if j > 0:
               __snake_case:						List[str]			  =						self.blend_h(row[j - 1]							,    A							,    A		)
             result_row.append(tile[:, :, :row_limit, :row_limit]		)
           result_rows.append(torch.cat(A							,    dim=3		)		)

         __snake_case:						Dict			  =						torch.cat(A							,    dim=2		)
         if not return_dict:
           return (dec,)

         return DecoderOutput(sample=A		)






       def 							UpperCAmelCase__			(  self			:   List[Any]							,    A			:   torch.FloatTensor							,    A			:   bool = False							,    A			:   bool = True							,    A			:   Optional[torch.Generator] = None							,    ):
         __snake_case:						Optional[Any]			  =						sample
         __snake_case:						Union[str, Any]			  =						self.encode(A		).latent_dist
         if sample_posterior:
           __snake_case:						Optional[Any]			  =						posterior.sample(generator=A		)
         else:
           __snake_case:						Dict			  =						posterior.mode()
         __snake_case:						Any			  =						self.decode(A		).sample

         if not return_dict:
           return (dec,)

         return DecoderOutput(sample=A		)






