


import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
   import jieba

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


UpperCAmelCase_      : Any  					=       logging.get_logger(__name__)

UpperCAmelCase_      : Dict  					=       {'vocab_file': 'vocab.txt'}

UpperCAmelCase_      : Optional[int]  					=       {
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

UpperCAmelCase_      : Tuple  					=       {
    'openbmb/cpm-ant-10b': 1024,
}







def        SCREAMING_SNAKE_CASE_						(			__A :			Tuple				)	->		Tuple:

       """simple docstring"""

       a_					:		Union[str, Any]							=  collections.OrderedDict()
       with open(__A  ,    'r'  ,    encoding='utf-8'				) as reader:
              a_					:		int							=  reader.readlines()
       for index, token in enumerate(__A				):
              a_					:		Union[str, Any]							=  token.rstrip('\n'				)
              a_					:		Union[str, Any]							=  index
       return vocab
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      def __init__(       self				:       int     ,							SCREAMING_SNAKE_CASE__				:       int     ,							SCREAMING_SNAKE_CASE__				:       List[str]="<unk>"     ,							SCREAMING_SNAKE_CASE__				:       Optional[Any]=2_0_0     )		->			List[str]:
             a_					:		List[Any]							=  vocab
             a_					:		Tuple							=  unk_token
             a_					:		Tuple							=  max_input_chars_per_word

      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]     )		->			Dict:
             a_					:		Any							=  list(SCREAMING_SNAKE_CASE__     )
             if len(SCREAMING_SNAKE_CASE__     ) > self.max_input_chars_per_word:
                    return [self.unk_token]

             a_					:		Tuple							=  0
             a_					:		Union[str, Any]							=  []
             while start < len(SCREAMING_SNAKE_CASE__     ):
                    a_					:		List[Any]							=  len(SCREAMING_SNAKE_CASE__     )
                    a_					:		List[Any]							=  None
                    while start < end:
                           a_					:		Dict							=  ''.join(chars[start:end]     )
                           if substr in self.vocab:
                                  a_					:		int							=  substr
                                  break
                           end -= 1
                    if cur_substr is None:
                           sub_tokens.append(self.unk_token     )
                           start += 1
                    else:
                           sub_tokens.append(SCREAMING_SNAKE_CASE__     )
                           a_					:		Union[str, Any]							=  end

             return sub_tokens
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     Optional[int]        =						VOCAB_FILES_NAMES
      snake_case__				:     Dict        =						PRETRAINED_VOCAB_FILES_MAP
      snake_case__				:     List[Any]        =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      snake_case__				:     str        =						['''input_ids''', '''attention_mask''']
      snake_case__				:     Union[str, Any]        =						False
      def __init__(       self				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       Dict="<d>"     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]="</d>"     ,							SCREAMING_SNAKE_CASE__				:       List[Any]="<s>"     ,							SCREAMING_SNAKE_CASE__				:       Optional[int]="</s>"     ,							SCREAMING_SNAKE_CASE__				:       Tuple="<pad>"     ,							SCREAMING_SNAKE_CASE__				:       str="<unk>"     ,							SCREAMING_SNAKE_CASE__				:       str="</n>"     ,							SCREAMING_SNAKE_CASE__				:       Any="</_>"     ,							SCREAMING_SNAKE_CASE__				:       Tuple="left"     ,							**SCREAMING_SNAKE_CASE__				:       Dict     ,							)		->			Union[str, Any]:
             requires_backends(self     ,							['jieba']     )
             super().__init__(
                 bod_token=SCREAMING_SNAKE_CASE__     ,							eod_token=SCREAMING_SNAKE_CASE__     ,							bos_token=SCREAMING_SNAKE_CASE__     ,							eos_token=SCREAMING_SNAKE_CASE__     ,							pad_token=SCREAMING_SNAKE_CASE__     ,							unk_token=SCREAMING_SNAKE_CASE__     ,							line_token=SCREAMING_SNAKE_CASE__     ,							space_token=SCREAMING_SNAKE_CASE__     ,							padding_side=SCREAMING_SNAKE_CASE__     ,							**SCREAMING_SNAKE_CASE__     ,							)
             a_					:		Tuple							=  bod_token
             a_					:		str							=  eod_token
             a_					:		Optional[int]							=  load_vocab(SCREAMING_SNAKE_CASE__     )
             a_					:		Tuple							=  self.encoder[space_token]
             a_					:		Any							=  self.encoder[line_token]

             del self.encoder[space_token]
             del self.encoder[line_token]

             a_					:		str							=  collections.OrderedDict(sorted(self.encoder.items()     ,							key=lambda SCREAMING_SNAKE_CASE__     : x[1]     )     )
             a_					:		List[Any]							=  {v: k for k, v in self.encoder.items()}

             a_					:		str							=  WordpieceTokenizer(vocab=self.encoder     ,							unk_token=self.unk_token     )
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			int:
             return self.encoder[self.bod_token]
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     )		->			Optional[Any]:
             return self.encoder[self.eod_token]
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     )		->			List[Any]:
             return self.encoder["\n"]
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     )		->			int:
             return len(self.encoder     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Union[str, Any]:
             return dict(self.encoder     ,							**self.added_tokens_encoder     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     ,							SCREAMING_SNAKE_CASE__				:       Optional[Any]     )		->			Dict:
             a_					:		Union[str, Any]							=  []
             for x in jieba.cut(SCREAMING_SNAKE_CASE__     ,							cut_all=SCREAMING_SNAKE_CASE__     ):
                    output_tokens.extend(self.wordpiece_tokenizer.tokenize(SCREAMING_SNAKE_CASE__     )     )
             return output_tokens
      def        SCREAMING_SNAKE_CASE							(       self				:       str     ,							SCREAMING_SNAKE_CASE__				:       List[str]     ,							**SCREAMING_SNAKE_CASE__				:       str     )		->			List[Any]:
             a_					:		Optional[Any]							=  [i for i in token_ids if i >= 0]
             a_					:		int							=  [
                 x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
             ]
             return super()._decode(SCREAMING_SNAKE_CASE__     ,							**SCREAMING_SNAKE_CASE__     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       List[Any]     )		->			int:
             return token in self.encoder
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     ,							SCREAMING_SNAKE_CASE__				:       List[str]     )		->			str:
             return "".join(SCREAMING_SNAKE_CASE__     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     ,							SCREAMING_SNAKE_CASE__				:       Dict     )		->			Union[str, Any]:
             return self.encoder.get(SCREAMING_SNAKE_CASE__     ,							self.encoder.get(self.unk_token     )     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     ,							SCREAMING_SNAKE_CASE__				:       List[str]     )		->			Dict:
             return self.decoder.get(SCREAMING_SNAKE_CASE__     ,							self.unk_token     )
      def        SCREAMING_SNAKE_CASE							(       self				:       int     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       Optional[str] = None     )		->			Tuple[str]:
             if os.path.isdir(SCREAMING_SNAKE_CASE__     ):
                    a_					:		str							=  os.path.join(
                        SCREAMING_SNAKE_CASE__     ,							(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']     )
             else:
                    a_					:		str							=  (filename_prefix + '-' if filename_prefix else '') + save_directory
             a_					:		int							=  0
             if " " in self.encoder:
                    a_					:		List[str]							=  self.encoder[' ']
                    del self.encoder[" "]
             if "\n" in self.encoder:
                    a_					:		Union[str, Any]							=  self.encoder['\n']
                    del self.encoder["\n"]
             a_					:		List[Any]							=  collections.OrderedDict(sorted(self.encoder.items()     ,							key=lambda SCREAMING_SNAKE_CASE__     : x[1]     )     )
             with open(SCREAMING_SNAKE_CASE__     ,							'w'     ,							encoding='utf-8'     ) as writer:
                    for token, token_index in self.encoder.items():
                           if index != token_index:
                                  logger.warning(
                                      F"""Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."""
                                      ' Please check that the vocabulary is not corrupted!'     )
                                  a_					:		Optional[Any]							=  token_index
                           writer.write(token + '\n'     )
                           index += 1
             return (vocab_file,)
      def        SCREAMING_SNAKE_CASE							(       self				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       List[int]     ,							SCREAMING_SNAKE_CASE__				:       List[int] = None     )		->			List[int]:
             if token_ids_a is None:
                    return [self.bos_token_id] + token_ids_a
             return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a

      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       List[int]     ,							SCREAMING_SNAKE_CASE__				:       Optional[List[int]] = None     ,							SCREAMING_SNAKE_CASE__				:       bool = False     )		->			List[int]:

             if already_has_special_tokens:
                    return super().get_special_tokens_mask(
                        token_ids_a=SCREAMING_SNAKE_CASE__     ,							token_ids_a=SCREAMING_SNAKE_CASE__     ,							already_has_special_tokens=SCREAMING_SNAKE_CASE__     )

             if token_ids_a is not None:
                    return [1] + ([0] * len(SCREAMING_SNAKE_CASE__     )) + [1] + ([0] * len(SCREAMING_SNAKE_CASE__     ))
             return [1] + ([0] * len(SCREAMING_SNAKE_CASE__     ))








'''simple docstring'''

import gc
import unittest

import torch
from parameterized import parameterized

from diffusers import AutoencoderKL
from diffusers.utils import floats_tensor, load_hf_numpy, require_torch_gpu, slow, torch_all_close, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()






class 						a				(   _lowerCamelCase   ,     _lowerCamelCase   ,     unittest.TestCase     ):
				snake_case_      =		AutoencoderKL
				snake_case_      =		"sample"
				snake_case_      =		1e-2


				@property
				def   A_ (		self    :					Dict					):
									snake_case_									=				4
									snake_case_									=				3
									snake_case_									=				(32, 32)

									snake_case_									=				floats_tensor((batch_size, num_channels) + sizes					).to(lowercase_					)

									return {"sample": image}


				@property
				def   A_ (		self    :					List[Any]					):
									return (3, 32, 32)


				@property
				def   A_ (		self    :					Dict					):
									return (3, 32, 32)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				{
									    '''block_out_channels''': [32, 64],
									    '''in_channels''': 3,
									    '''out_channels''': 3,
									    '''down_block_types''': ['''DownEncoderBlock2D''', '''DownEncoderBlock2D'''],
									    '''up_block_types''': ['''UpDecoderBlock2D''', '''UpDecoderBlock2D'''],
									    '''latent_channels''': 4,
									}
									snake_case_									=				self.dummy_input
									return init_dict, inputs_dict


				def   A_ (		self    :					Any					):
									pass


				def   A_ (		self    :					str					):
									pass


				@unittest.skipIf(torch_device == '''mps'''       ,				'''Gradient checkpointing skipped on MPS'''					)
				def   A_ (		self    :					Dict					):
									# enable deterministic behavior for gradient checkpointing
									snake_case_			,snake_case_									=				self.prepare_init_args_and_inputs_for_common()
									snake_case_									=				self.model_class(**lowercase_					)
									model.to(lowercase_					)

									assert not model.is_gradient_checkpointing and model.training

									snake_case_									=				model(**lowercase_					).sample
									# run the backwards pass on the model. For backwards pass, for simplicity purpose,
									# we won't calculate the loss and rather backprop on out.sum()
									model.zero_grad()

									snake_case_									=				torch.randn_like(lowercase_					)
									snake_case_									=				(out - labels).mean()
									loss.backward()

									# re-instantiate the model now enabling gradient checkpointing
									snake_case_									=				self.model_class(**lowercase_					)
									# clone model
									model_a.load_state_dict(model.state_dict()					)
									model_a.to(lowercase_					)
									model_a.enable_gradient_checkpointing()

									assert model_a.is_gradient_checkpointing and model_a.training

									snake_case_									=				model_a(**lowercase_					).sample
									# run the backwards pass on the model. For backwards pass, for simplicity purpose,
									# we won't calculate the loss and rather backprop on out.sum()
									model_a.zero_grad()
									snake_case_									=				(out_a - labels).mean()
									loss_a.backward()

									# compare the output and parameters gradients
									self.assertTrue((loss - loss_a).abs() < 1e-5					)
									snake_case_									=				dict(model.named_parameters()					)
									snake_case_									=				dict(model_a.named_parameters()					)
									for name, param in named_params.items():
														self.assertTrue(torch_all_close(param.grad.data       ,				named_params_a[name].grad.data       ,				atol=5e-5					)					)


				def   A_ (		self    :					Tuple					):
									snake_case_			,snake_case_									=				AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''       ,				output_loading_info=lowercase_					)
									self.assertIsNotNone(lowercase_					)
									self.assertEqual(len(loading_info['''missing_keys''']					)       ,				0					)

									model.to(lowercase_					)
									snake_case_									=				model(**self.dummy_input					)

									assert image is not None, "Make sure output is not None"


				def   A_ (		self    :					Tuple					):
									snake_case_									=				AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''					)
									snake_case_									=				model.to(lowercase_					)
									model.eval()

									if torch_device == "mps":
														snake_case_									=				torch.manual_seed(0					)
									else:
														snake_case_									=				torch.Generator(device=lowercase_					).manual_seed(0					)

									snake_case_									=				torch.randn(
									    1       ,				model.config.in_channels       ,				model.config.sample_size       ,				model.config.sample_size       ,				generator=torch.manual_seed(0					)       ,				)
									snake_case_									=				image.to(lowercase_					)
									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				sample_posterior=lowercase_       ,				generator=lowercase_					).sample

									snake_case_									=				output[0, -1, -3:, -3:].flatten().cpu()

									# Since the VAE Gaussian prior's generator is seeded on the appropriate device,
									# the expected output slices are not the same for CPU and GPU.
									if torch_device == "mps":
														snake_case_									=				torch.tensor(
														    [
														        -4.0_078e-01,
														        -3.8_323e-04,
														        -1.2_681e-01,
														        -1.1_462e-01,
														        2.0_095e-01,
														        1.0_893e-01,
														        -8.8_247e-02,
														        -3.0_361e-01,
														        -9.8_644e-03,
														    ]					)
									elif torch_device == "cpu":
														snake_case_									=				torch.tensor(
														    [-0.1352, 0.0878, 0.0419, -0.0818, -0.1069, 0.0688, -0.1458, -0.4446, -0.0026]					)
									else:
														snake_case_									=				torch.tensor(
														    [-0.2421, 0.4642, 0.2507, -0.0438, 0.0682, 0.3160, -0.2018, -0.0727, 0.2485]					)

									self.assertTrue(torch_all_close(lowercase_       ,				lowercase_       ,				rtol=1e-2					)					)







@slow
class 						a				(   unittest.TestCase     ):


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Optional[int]					):
									return F"gaussian_noise_s={seed}_shape={'_'.join([str(lowercase_					) for s in shape]					)}.npy"


				def   A_ (		self    :					Any					):
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()


				def   A_ (		self    :					Dict       ,				lowercase_    :					List[Any]=0       ,				lowercase_    :					Union[str, Any]=(4, 3, 512, 512)       ,				lowercase_    :					Optional[Any]=False					):
									snake_case_									=				torch.floataa if fpaa else torch.floataa
									snake_case_									=				torch.from_numpy(load_hf_numpy(self.get_file_format(lowercase_       ,				lowercase_					)					)					).to(lowercase_					).to(lowercase_					)
									return image


				def   A_ (		self    :					Any       ,				lowercase_    :					Dict="CompVis/stable-diffusion-v1-4"       ,				lowercase_    :					List[str]=False					):
									snake_case_									=				'''fp16''' if fpaa else None
									snake_case_									=				torch.floataa if fpaa else torch.floataa

									snake_case_									=				AutoencoderKL.from_pretrained(
									    lowercase_       ,				subfolder='''vae'''       ,				torch_dtype=lowercase_       ,				revision=lowercase_       ,				)
									model.to(lowercase_					).eval()

									return model


				def   A_ (		self    :					Any       ,				lowercase_    :					int=0					):
									if torch_device == "mps":
														return torch.manual_seed(lowercase_					)
									return torch.Generator(device=lowercase_					).manual_seed(lowercase_					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1603, 0.9878, -0.0495, -0.0790, -0.2709, 0.8375, -0.2060, -0.0824], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2376, 0.1168, 0.1332, -0.4840, -0.2508, -0.0791, -0.0493, -0.4089], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				generator=lowercase_       ,				sample_posterior=lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, -2:, :2].flatten().float().cpu()
									snake_case_									=				torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=3e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.0513, 0.0289, 1.3799, 0.2166, -0.2573, -0.0871, 0.5103, -0.0999]],
				        [47, [-0.4128, -0.1320, -0.3704, 0.1965, -0.4116, -0.2332, -0.3340, 0.2247]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Optional[int]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					Dict					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				fpaa=lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				generator=lowercase_       ,				sample_posterior=lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().float().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-2					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1609, 0.9866, -0.0487, -0.0777, -0.2716, 0.8368, -0.2055, -0.0814], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2377, 0.1147, 0.1333, -0.4841, -0.2506, -0.0805, -0.0491, -0.4085], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					Tuple       ,				lowercase_    :					Dict       ,				lowercase_    :					str       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, -2:, :2].flatten().float().cpu()
									snake_case_									=				torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=3e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [13, [-0.2051, -0.1803, -0.2311, -0.2114, -0.3292, -0.3574, -0.2953, -0.3323]],
				        [37, [-0.2632, -0.2625, -0.2199, -0.2741, -0.4539, -0.4990, -0.3720, -0.4925]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Dict       ,				lowercase_    :					Tuple       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [27, [-0.0369, 0.0207, -0.0776, -0.0682, -0.1747, -0.1930, -0.1465, -0.2039]],
				        [16, [-0.1628, -0.2134, -0.2747, -0.2642, -0.3774, -0.4404, -0.3687, -0.4277]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Any       ,				lowercase_    :					Optional[Any]					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)       ,				fpaa=lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().float().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=5e-3					)


				@parameterized.expand([(13,), (16,), (27,)]					)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()       ,				reason='''xformers is not required when using PyTorch 2.0.'''					)
				def   A_ (		self    :					Optional[Any]       ,				lowercase_    :					List[str]					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)       ,				fpaa=lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									model.enable_xformers_memory_efficient_attention()
									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-1					)


				@parameterized.expand([(13,), (16,), (37,)]					)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()       ,				reason='''xformers is not required when using PyTorch 2.0.'''					)
				def   A_ (		self    :					Optional[Any]       ,				lowercase_    :					Any					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									model.enable_xformers_memory_efficient_attention()
									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-2					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.3001, 0.0918, -2.6984, -3.9720, -3.2099, -5.0353, 1.7338, -0.2065, 3.4267]],
				        [47, [-1.5030, -4.3871, -6.0355, -9.1157, -1.6661, -2.7853, 2.1607, -5.0823, 2.5633]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					str       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					Tuple					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.encode(lowercase_					).latent_dist
														snake_case_									=				dist.sample(generator=lowercase_					)

									assert list(sample.shape					) == [image.shape[0], 4] + [i // 8 for i in image.shape[2:]]

									snake_case_									=				sample[0, -1, -3:, -3:].flatten().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									snake_case_									=				3e-3 if torch_device != '''mps''' else 1e-2
									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=lowercase_					)

