import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
      from test_module.custom_tokenization_fast import CustomTokenizerFast






class 		_lowerCamelCase		(   unittest.TestCase    ):
    def 						UpperCamelCase_				(  self   )       ->					str:
     # A mock response for an HTTP head request to emulate server down
     SCREAMING_SNAKE_CASE__:   Any=      mock.Mock()
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      500
     SCREAMING_SNAKE_CASE__:   Optional[int]=      {}
     SCREAMING_SNAKE_CASE__:   Tuple=      HTTPError
     SCREAMING_SNAKE_CASE__:   Any=      {}

     # Download this model to make sure it's in the cache.
     SCREAMING_SNAKE_CASE__:   Optional[int]=      BertTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert'''   )

     # Under the mock environment we get a 500 error when trying to reach the tokenizer.
     with mock.patch('''requests.Session.request''' , return_value=lowerCAmelCase   ) as mock_head:
      SCREAMING_SNAKE_CASE__:   Tuple=      BertTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert'''   )
      # This check we did call the fake head request
      mock_head.assert_called()
    @require_tokenizers
    def 						UpperCamelCase_				(  self   )       ->					int:
     # A mock response for an HTTP head request to emulate server down
     SCREAMING_SNAKE_CASE__:   List[str]=      mock.Mock()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      500
     SCREAMING_SNAKE_CASE__:   Dict=      {}
     SCREAMING_SNAKE_CASE__:   List[str]=      HTTPError
     SCREAMING_SNAKE_CASE__:   Tuple=      {}

     # Download this model to make sure it's in the cache.
     SCREAMING_SNAKE_CASE__:   str=      GPTaTokenizerFast.from_pretrained('''gpt2'''   )

     # Under the mock environment we get a 500 error when trying to reach the tokenizer.
     with mock.patch('''requests.Session.request''' , return_value=lowerCAmelCase   ) as mock_head:
      SCREAMING_SNAKE_CASE__:   int=      GPTaTokenizerFast.from_pretrained('''gpt2'''   )
      # This check we did call the fake head request
      mock_head.assert_called()
    def 						UpperCamelCase_				(  self   )       ->					str:
     # This test is for deprecated behavior and can be removed in v5
     try:
      SCREAMING_SNAKE_CASE__:   List[str]=      tempfile.mktemp()
      with open(lowerCAmelCase , '''wb'''   ) as f:
       http_get('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model''' , lowerCAmelCase   )

      SCREAMING_SNAKE_CASE__:   Any=      AlbertTokenizer.from_pretrained(lowerCAmelCase   )
     finally:
      os.remove(lowerCAmelCase   )

     # Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
     # the current folder and have the right name.
     if os.path.isfile('''tokenizer.json'''   ):
      # We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
      return
     try:
      with open('''tokenizer.json''' , '''wb'''   ) as f:
       http_get('''https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json''' , lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   List[str]=      AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   )
      # The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
      self.assertEqual(tokenizer.vocab_size , 1000   )
      # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

     finally:
      os.remove('''tokenizer.json'''   )



    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     # This test is for deprecated behavior and can be removed in v5
     SCREAMING_SNAKE_CASE__:   List[str]=      AlbertTokenizer.from_pretrained('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model'''   )






@is_staging_test
class 		_lowerCamelCase		(   unittest.TestCase    ):
    __a  =      ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]
    @classmethod
    def 						UpperCamelCase_				(  cls   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   str=      TOKEN
     HfFolder.save_token(lowerCAmelCase   )
    @classmethod
    def 						UpperCamelCase_				(  cls   )       ->					List[str]:
     try:
      delete_repo(token=cls._token , repo_id='''test-tokenizer'''   )
     except HTTPError:
      pass

     try:
      delete_repo(token=cls._token , repo_id='''valid_org/test-tokenizer-org'''   )
     except HTTPError:
      pass

     try:
      delete_repo(token=cls._token , repo_id='''test-dynamic-tokenizer'''   )
     except HTTPError:
      pass
    def 						UpperCamelCase_				(  self   )       ->					Any:
     with tempfile.TemporaryDirectory() as tmp_dir:
      SCREAMING_SNAKE_CASE__:   List[Any]=      os.path.join(lowerCAmelCase , '''vocab.txt'''   )
      with open(lowerCAmelCase , '''w''' , encoding='''utf-8'''   ) as vocab_writer:
       vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )
      SCREAMING_SNAKE_CASE__:   Tuple=      BertTokenizer(lowerCAmelCase   )

     tokenizer.push_to_hub('''test-tokenizer''' , use_auth_token=self._token   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      BertTokenizer.from_pretrained(f'{USER}/test-tokenizer'   )
     self.assertDictEqual(new_tokenizer.vocab , tokenizer.vocab   )

     # Reset repo
     delete_repo(token=self._token , repo_id='''test-tokenizer'''   )

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
      tokenizer.save_pretrained(lowerCAmelCase , repo_id='''test-tokenizer''' , push_to_hub=lowerCAmelCase , use_auth_token=self._token   )

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      BertTokenizer.from_pretrained(f'{USER}/test-tokenizer'   )
     self.assertDictEqual(new_tokenizer.vocab , tokenizer.vocab   )
    def 						UpperCamelCase_				(  self   )       ->					Dict:
     with tempfile.TemporaryDirectory() as tmp_dir:
      SCREAMING_SNAKE_CASE__:   int=      os.path.join(lowerCAmelCase , '''vocab.txt'''   )
      with open(lowerCAmelCase , '''w''' , encoding='''utf-8'''   ) as vocab_writer:
       vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )
      SCREAMING_SNAKE_CASE__:   Dict=      BertTokenizer(lowerCAmelCase   )

     tokenizer.push_to_hub('''valid_org/test-tokenizer-org''' , use_auth_token=self._token   )
     SCREAMING_SNAKE_CASE__:   int=      BertTokenizer.from_pretrained('''valid_org/test-tokenizer-org'''   )
     self.assertDictEqual(new_tokenizer.vocab , tokenizer.vocab   )

     # Reset repo
     delete_repo(token=self._token , repo_id='''valid_org/test-tokenizer-org'''   )

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
      tokenizer.save_pretrained(
          lowerCAmelCase , repo_id='''valid_org/test-tokenizer-org''' , push_to_hub=lowerCAmelCase , use_auth_token=self._token   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      BertTokenizer.from_pretrained('''valid_org/test-tokenizer-org'''   )
     self.assertDictEqual(new_tokenizer.vocab , tokenizer.vocab   )



    @require_tokenizers
    def 						UpperCamelCase_				(  self   )       ->					List[Any]:
     CustomTokenizer.register_for_auto_class()
     with tempfile.TemporaryDirectory() as tmp_dir:
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      os.path.join(lowerCAmelCase , '''vocab.txt'''   )
      with open(lowerCAmelCase , '''w''' , encoding='''utf-8'''   ) as vocab_writer:
       vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )
      SCREAMING_SNAKE_CASE__:   List[str]=      CustomTokenizer(lowerCAmelCase   )

     # No fast custom tokenizer
     tokenizer.push_to_hub('''test-dynamic-tokenizer''' , use_auth_token=self._token   )

     SCREAMING_SNAKE_CASE__:   int=      AutoTokenizer.from_pretrained(f'{USER}/test-dynamic-tokenizer' , trust_remote_code=lowerCAmelCase   )
     # Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
     self.assertEqual(tokenizer.__class__.__name__ , '''CustomTokenizer'''   )

     # Fast and slow custom tokenizer
     CustomTokenizerFast.register_for_auto_class()
     with tempfile.TemporaryDirectory() as tmp_dir:
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      os.path.join(lowerCAmelCase , '''vocab.txt'''   )
      with open(lowerCAmelCase , '''w''' , encoding='''utf-8'''   ) as vocab_writer:
       vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]   )   )

      SCREAMING_SNAKE_CASE__:   Tuple=      BertTokenizerFast.from_pretrained(lowerCAmelCase   )
      bert_tokenizer.save_pretrained(lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      CustomTokenizerFast.from_pretrained(lowerCAmelCase   )

     tokenizer.push_to_hub('''test-dynamic-tokenizer''' , use_auth_token=self._token   )

     SCREAMING_SNAKE_CASE__:   str=      AutoTokenizer.from_pretrained(f'{USER}/test-dynamic-tokenizer' , trust_remote_code=lowerCAmelCase   )
     # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
     self.assertEqual(tokenizer.__class__.__name__ , '''CustomTokenizerFast'''   )
     SCREAMING_SNAKE_CASE__:   List[Any]=      AutoTokenizer.from_pretrained(
         f'{USER}/test-dynamic-tokenizer' , use_fast=lowerCAmelCase , trust_remote_code=lowerCAmelCase   )
     # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
     self.assertEqual(tokenizer.__class__.__name__ , '''CustomTokenizer'''   )






class 		_lowerCamelCase		(   unittest.TestCase    ):
    def 						UpperCamelCase_				(  self   )       ->					int:
     SCREAMING_SNAKE_CASE__:   List[str]=      Trie()
     trie.add('''Hello 友達'''   )
     self.assertEqual(trie.data , {'''H''': {'''e''': {'''l''': {'''l''': {'''o''': {''' ''': {'''友''': {'''達''': {'''''': 1}}}}}}}}}   )
     trie.add('''Hello'''   )
     trie.data
     self.assertEqual(trie.data , {'''H''': {'''e''': {'''l''': {'''l''': {'''o''': {'''''': 1, ''' ''': {'''友''': {'''達''': {'''''': 1}}}}}}}}}   )
    def 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     SCREAMING_SNAKE_CASE__:   Any=      Trie()
     self.assertEqual(trie.split('''[CLS] This is a extra_id_100'''   ) , ['''[CLS] This is a extra_id_100''']   )
     trie.add('''[CLS]'''   )
     trie.add('''extra_id_1'''   )
     trie.add('''extra_id_100'''   )
     self.assertEqual(trie.split('''[CLS] This is a extra_id_100'''   ) , ['''[CLS]''', ''' This is a ''', '''extra_id_100''']   )
    def 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      Trie()
     trie.add('''A'''   )
     self.assertEqual(trie.split('''ABC'''   ) , ['''A''', '''BC''']   )
     self.assertEqual(trie.split('''BCA'''   ) , ['''BC''', '''A''']   )
    def 						UpperCamelCase_				(  self   )       ->					Any:
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      Trie()
     trie.add('''TOKEN]'''   )
     trie.add('''[SPECIAL_TOKEN]'''   )
     self.assertEqual(trie.split('''This is something [SPECIAL_TOKEN]'''   ) , ['''This is something ''', '''[SPECIAL_TOKEN]''']   )
    def 						UpperCamelCase_				(  self   )       ->					str:
     SCREAMING_SNAKE_CASE__:   Tuple=      Trie()
     trie.add('''A'''   )
     trie.add('''P'''   )
     trie.add('''[SPECIAL_TOKEN]'''   )
     self.assertEqual(trie.split('''This is something [SPECIAL_TOKEN]'''   ) , ['''This is something ''', '''[SPECIAL_TOKEN]''']   )
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Dict=      Trie()
     trie.add('''AB'''   )
     trie.add('''B'''   )
     trie.add('''C'''   )
     self.assertEqual(trie.split('''ABC'''   ) , ['''AB''', '''C''']   )
    def 						UpperCamelCase_				(  self   )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      Trie()
     trie.add('''ABC'''   )
     trie.add('''B'''   )
     trie.add('''CD'''   )
     self.assertEqual(trie.split('''ABCD'''   ) , ['''ABC''', '''D''']   )



    def 						UpperCamelCase_				(  self   )       ->					int:
     # Even if the offsets are wrong, we necessarily output correct string
     # parts.
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      Trie()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      trie.cut_text('''ABC''' , [0, 0, 2, 1, 2, 3]   )
     self.assertEqual(lowerCAmelCase , ['''AB''', '''C''']   )


import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, XLMRobertaTokenizer

from diffusers import AltDiffusionPipeline, AutoencoderKL, DDIMScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class 		_lowerCamelCase		(   UpperCamelCase_    ,     UpperCamelCase_    ,     UpperCamelCase_    ,     unittest.TestCase    ):
    __a  =      AltDiffusionPipeline
    __a  =      TEXT_TO_IMAGE_PARAMS
    __a  =      TEXT_TO_IMAGE_BATCH_PARAMS
    __a  =      TEXT_TO_IMAGE_IMAGE_PARAMS
    __a  =      TEXT_TO_IMAGE_IMAGE_PARAMS
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     torch.manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   List[Any]=      UNetaDConditionModel(
         block_out_channels=(32, 64) , layers_per_block=2 , sample_size=32 , in_channels=4 , out_channels=4 , down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''') , up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''') , cross_attention_dim=32 , )
     SCREAMING_SNAKE_CASE__:   Tuple=      DDIMScheduler(
         beta_start=0.00085 , beta_end=0.012 , beta_schedule='''scaled_linear''' , clip_sample=lowerCAmelCase , set_alpha_to_one=lowerCAmelCase , )
     torch.manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      AutoencoderKL(
         block_out_channels=[32, 64] , in_channels=3 , out_channels=3 , down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D'''] , up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D'''] , latent_channels=4 , )

     # TODO: address the non-deterministic text encoder (fails for save-load tests)
     # torch.manual_seed(0)
     # text_encoder_config = RobertaSeriesConfig(
     #     hidden_size=32,
     #     project_dim=32,
     #     intermediate_size=37,
     #     layer_norm_eps=1e-05,
     #     num_attention_heads=4,
     #     num_hidden_layers=5,
     #     vocab_size=5002,
     # )
     # text_encoder = RobertaSeriesModelWithTransformation(text_encoder_config)

     torch.manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   Tuple=      CLIPTextConfig(
         bos_token_id=0 , eos_token_id=2 , hidden_size=32 , projection_dim=32 , intermediate_size=37 , layer_norm_eps=1e-05 , num_attention_heads=4 , num_hidden_layers=5 , pad_token_id=1 , vocab_size=5002 , )
     SCREAMING_SNAKE_CASE__:   List[Any]=      CLIPTextModel(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      XLMRobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-xlm-roberta'''   )
     SCREAMING_SNAKE_CASE__:   List[Any]=      77

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      {
         '''unet''': unet,
         '''scheduler''': scheduler,
         '''vae''': vae,
         '''text_encoder''': text_encoder,
         '''tokenizer''': tokenizer,
         '''safety_checker''': None,
         '''feature_extractor''': None,
     }
     return components
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase=0   )       ->					List[Any]:
     if str(lowerCAmelCase   ).startswith('''mps'''   ):
      SCREAMING_SNAKE_CASE__:   Any=      torch.manual_seed(lowerCAmelCase   )
     else:
      SCREAMING_SNAKE_CASE__:   Union[str, Any]=      torch.Generator(device=lowerCAmelCase   ).manual_seed(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Any=      {
         '''prompt''': '''A painting of a squirrel eating a burger''',
         '''generator''': generator,
         '''num_inference_steps''': 2,
         '''guidance_scale''': 6.0,
         '''output_type''': '''numpy''',
     }
     return inputs
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     super().test_attention_slicing_forward_pass(expected_max_diff=3e-3   )
    def 						UpperCamelCase_				(  self   )       ->					Tuple:
     super().test_inference_batch_single_identical(expected_max_diff=3e-3   )
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Optional[int]=      '''cpu'''  # ensure determinism for the device-dependent torch.Generator

     SCREAMING_SNAKE_CASE__:   Tuple=      self.get_dummy_components()
     torch.manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   List[Any]=      RobertaSeriesConfig(
         hidden_size=32 , project_dim=32 , intermediate_size=37 , layer_norm_eps=1e-05 , num_attention_heads=4 , num_hidden_layers=5 , vocab_size=5002 , )
     # TODO: remove after fixing the non-deterministic text encoder
     SCREAMING_SNAKE_CASE__:   List[str]=      RobertaSeriesModelWithTransformation(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      text_encoder

     SCREAMING_SNAKE_CASE__:   List[Any]=      AltDiffusionPipeline(**lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      alt_pipe.to(lowerCAmelCase   )
     alt_pipe.set_progress_bar_config(disable=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   str=      self.get_dummy_inputs(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Dict=      '''A photo of an astronaut'''
     SCREAMING_SNAKE_CASE__:   Any=      alt_pipe(**lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      output.images
     SCREAMING_SNAKE_CASE__:   Tuple=      image[0, -3:, -3:, -1]

     assert image.shape == (1, 64, 64, 3)
     SCREAMING_SNAKE_CASE__:   Any=      np.array(
         [0.5748162, 0.60447145, 0.48821217, 0.50100636, 0.5431185, 0.45763683, 0.49657696, 0.48132733, 0.47573093]   )

     assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2



    def 						UpperCamelCase_				(  self   )       ->					Optional[int]:
     SCREAMING_SNAKE_CASE__:   List[str]=      '''cpu'''  # ensure determinism for the device-dependent torch.Generator

     SCREAMING_SNAKE_CASE__:   Any=      self.get_dummy_components()
     SCREAMING_SNAKE_CASE__:   List[str]=      PNDMScheduler(skip_prk_steps=lowerCAmelCase   )
     torch.manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   Tuple=      RobertaSeriesConfig(
         hidden_size=32 , project_dim=32 , intermediate_size=37 , layer_norm_eps=1e-05 , num_attention_heads=4 , num_hidden_layers=5 , vocab_size=5002 , )
     # TODO: remove after fixing the non-deterministic text encoder
     SCREAMING_SNAKE_CASE__:   str=      RobertaSeriesModelWithTransformation(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      text_encoder
     SCREAMING_SNAKE_CASE__:   Any=      AltDiffusionPipeline(**lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      alt_pipe.to(lowerCAmelCase   )
     alt_pipe.set_progress_bar_config(disable=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      self.get_dummy_inputs(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      alt_pipe(**lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      output.images
     SCREAMING_SNAKE_CASE__:   Tuple=      image[0, -3:, -3:, -1]

     assert image.shape == (1, 64, 64, 3)
     SCREAMING_SNAKE_CASE__:   List[Any]=      np.array(
         [0.51605093, 0.5707241, 0.47365507, 0.50578886, 0.5633877, 0.4642503, 0.5182081, 0.48763484, 0.49084237]   )

     assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2






@slow
@require_torch_gpu
class 		_lowerCamelCase		(   unittest.TestCase    ):
    def 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()
    def 						UpperCamelCase_				(  self   )       ->					Optional[int]:
     # make sure here that pndm scheduler skips prk
     SCREAMING_SNAKE_CASE__:   Any=      AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion''' , safety_checker=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      alt_pipe.to(lowerCAmelCase   )
     alt_pipe.set_progress_bar_config(disable=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Dict=      '''A painting of a squirrel eating a burger'''
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      torch.manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   Tuple=      alt_pipe([prompt] , generator=lowerCAmelCase , guidance_scale=6.0 , num_inference_steps=20 , output_type='''np'''   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      output.images

     SCREAMING_SNAKE_CASE__:   Dict=      image[0, -3:, -3:, -1]

     assert image.shape == (1, 512, 512, 3)
     SCREAMING_SNAKE_CASE__:   Optional[int]=      np.array([0.1010, 0.0800, 0.0794, 0.0885, 0.0843, 0.0762, 0.0769, 0.0729, 0.0586]   )

     assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2



    def 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     SCREAMING_SNAKE_CASE__:   List[Any]=      DDIMScheduler.from_pretrained('''BAAI/AltDiffusion''' , subfolder='''scheduler'''   )

     SCREAMING_SNAKE_CASE__:   List[str]=      AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion''' , scheduler=lowerCAmelCase , safety_checker=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      alt_pipe.to(lowerCAmelCase   )
     alt_pipe.set_progress_bar_config(disable=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      '''A painting of a squirrel eating a burger'''
     SCREAMING_SNAKE_CASE__:   Any=      torch.manual_seed(0   )

     SCREAMING_SNAKE_CASE__:   str=      alt_pipe([prompt] , generator=lowerCAmelCase , num_inference_steps=2 , output_type='''numpy'''   )
     SCREAMING_SNAKE_CASE__:   List[str]=      output.images

     SCREAMING_SNAKE_CASE__:   Dict=      image[0, -3:, -3:, -1]

     assert image.shape == (1, 512, 512, 3)
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      np.array([0.4019, 0.4052, 0.3810, 0.4119, 0.3916, 0.3982, 0.4651, 0.4195, 0.5323]   )

     assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2

