


'''simple docstring'''






import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizer

from diffusers import (
    AltDiffusionImgaImgPipeline,
    AutoencoderKL,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()






class    __magic_name__				(						unittest.TestCase							):






       def      lowerCAmelCase						(     self)		->    Any:




          '''simple docstring'''
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()






       @property
       def      lowerCAmelCase						(     self)		->    Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[Any]				       =1
          _UpperCAmelCase  :  Tuple				       =3
          _UpperCAmelCase  :  int				       =(3_2, 3_2)

          _UpperCAmelCase  :  List[Any]				       =floats_tensor((batch_size, num_channels) + sizes    ,				rng=random.Random(0)).to(snake_case)
          return image






       @property
       def      lowerCAmelCase						(     self)		->    Optional[Any]:




          '''simple docstring'''
          torch.manual_seed(0)
          _UpperCAmelCase  :  int				       =UNetaDConditionModel(
              block_out_channels=(3_2, 6_4)    ,				layers_per_block=2    ,				sample_size=3_2    ,				in_channels=4    ,				out_channels=4    ,				down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')    ,				up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')    ,				cross_attention_dim=3_2    ,				)
          return model






       @property
       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          torch.manual_seed(0)
          _UpperCAmelCase  :  Dict				       =AutoencoderKL(
              block_out_channels=[3_2, 6_4]    ,				in_channels=3    ,				out_channels=3    ,				down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']    ,				up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']    ,				latent_channels=4    ,				)
          return model






       @property
       def      lowerCAmelCase						(     self)		->    List[str]:




          '''simple docstring'''
          torch.manual_seed(0)
          _UpperCAmelCase  :  int				       =RobertaSeriesConfig(
              hidden_size=3_2    ,				project_dim=3_2    ,				intermediate_size=3_7    ,				layer_norm_eps=1E-0_5    ,				num_attention_heads=4    ,				num_hidden_layers=5    ,				pad_token_id=1    ,				vocab_size=5_0_0_6    ,				)
          return RobertaSeriesModelWithTransformation(snake_case)






       @property
       def      lowerCAmelCase						(     self)		->    Optional[Any]:




          '''simple docstring'''
          def extract(*snake_case    ,				**snake_case):






             class    __magic_name__				:






                    def __init__(     self)		->    str:




                       '''simple docstring'''
                       _UpperCAmelCase  :  Optional[int]				       =torch.ones([0])






                    def      lowerCAmelCase						(     self    ,				snake_case)		->    Optional[int]:




                       '''simple docstring'''
                       self.pixel_values.to(snake_case)
                       return self






             return Out()

          return extract






       def      lowerCAmelCase						(     self)		->    Tuple:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[int]				       ='cpu'  # ensure determinism for the device-dependent torch.Generator
          _UpperCAmelCase  :  Optional[Any]				       =self.dummy_cond_unet
          _UpperCAmelCase  :  Any				       =PNDMScheduler(skip_prk_steps=snake_case)
          _UpperCAmelCase  :  Any				       =self.dummy_vae
          _UpperCAmelCase  :  Optional[int]				       =self.dummy_text_encoder
          _UpperCAmelCase  :  Optional[int]				       =XLMRobertaTokenizer.from_pretrained('hf-internal-testing/tiny-xlm-roberta')
          _UpperCAmelCase  :  List[Any]				       =7_7

          _UpperCAmelCase  :  int				       =self.dummy_image.to(snake_case)
          _UpperCAmelCase  :  Any				       =init_image / 2 + 0.5

          # make sure here that pndm scheduler skips prk
          _UpperCAmelCase  :  Tuple				       =AltDiffusionImgaImgPipeline(
              unet=snake_case    ,				scheduler=snake_case    ,				vae=snake_case    ,				text_encoder=snake_case    ,				tokenizer=snake_case    ,				safety_checker=snake_case    ,				feature_extractor=self.dummy_extractor    ,				)
          _UpperCAmelCase  :  int				       =VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor    ,				do_normalize=snake_case)
          _UpperCAmelCase  :  Tuple				       =alt_pipe.to(snake_case)
          alt_pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  Union[str, Any]				       ='A painting of a squirrel eating a burger'
          _UpperCAmelCase  :  Optional[Any]				       =torch.Generator(device=snake_case).manual_seed(0)
          _UpperCAmelCase  :  int				       =alt_pipe(
              [prompt]    ,				generator=snake_case    ,				guidance_scale=6.0    ,				num_inference_steps=2    ,				output_type='np'    ,				image=snake_case    ,				)

          _UpperCAmelCase  :  Union[str, Any]				       =output.images

          _UpperCAmelCase  :  Union[str, Any]				       =torch.Generator(device=snake_case).manual_seed(0)
          _UpperCAmelCase  :  int				       =alt_pipe(
              [prompt]    ,				generator=snake_case    ,				guidance_scale=6.0    ,				num_inference_steps=2    ,				output_type='np'    ,				image=snake_case    ,				return_dict=snake_case    ,				)[0]

          _UpperCAmelCase  :  Optional[int]				       =image[0, -3:, -3:, -1]
          _UpperCAmelCase  :  Tuple				       =image_from_tuple[0, -3:, -3:, -1]

          assert image.shape == (1, 3_2, 3_2, 3)
          _UpperCAmelCase  :  List[str]				       =np.array([0.44_27, 0.37_31, 0.42_49, 0.49_41, 0.45_46, 0.41_48, 0.41_93, 0.46_66, 0.44_99])

          assert np.abs(image_slice.flatten() - expected_slice).max() < 5E-3
          assert np.abs(image_from_tuple_slice.flatten() - expected_slice).max() < 5E-3






       @unittest.skipIf(torch_device != 'cuda'    ,				'This test requires a GPU')
       def      lowerCAmelCase						(     self)		->    Dict:




          '''simple docstring'''
          _UpperCAmelCase  :  Dict				       =self.dummy_cond_unet
          _UpperCAmelCase  :  Dict				       =PNDMScheduler(skip_prk_steps=snake_case)
          _UpperCAmelCase  :  Dict				       =self.dummy_vae
          _UpperCAmelCase  :  str				       =self.dummy_text_encoder
          _UpperCAmelCase  :  str				       =XLMRobertaTokenizer.from_pretrained('hf-internal-testing/tiny-xlm-roberta')
          _UpperCAmelCase  :  Union[str, Any]				       =7_7

          _UpperCAmelCase  :  List[Any]				       =self.dummy_image.to(snake_case)

          # put models in fp16
          _UpperCAmelCase  :  List[str]				       =unet.half()
          _UpperCAmelCase  :  Union[str, Any]				       =vae.half()
          _UpperCAmelCase  :  Optional[Any]				       =bert.half()

          # make sure here that pndm scheduler skips prk
          _UpperCAmelCase  :  List[str]				       =AltDiffusionImgaImgPipeline(
              unet=snake_case    ,				scheduler=snake_case    ,				vae=snake_case    ,				text_encoder=snake_case    ,				tokenizer=snake_case    ,				safety_checker=snake_case    ,				feature_extractor=self.dummy_extractor    ,				)
          _UpperCAmelCase  :  List[Any]				       =VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor    ,				do_normalize=snake_case)
          _UpperCAmelCase  :  Optional[int]				       =alt_pipe.to(snake_case)
          alt_pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  List[str]				       ='A painting of a squirrel eating a burger'
          _UpperCAmelCase  :  Optional[Any]				       =torch.manual_seed(0)
          _UpperCAmelCase  :  int				       =alt_pipe(
              [prompt]    ,				generator=snake_case    ,				num_inference_steps=2    ,				output_type='np'    ,				image=snake_case    ,				).images

          assert image.shape == (1, 3_2, 3_2, 3)






       @unittest.skipIf(torch_device != 'cuda'    ,				'This test requires a GPU')
       def      lowerCAmelCase						(     self)		->    Union[str, Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  Union[str, Any]				       =load_image(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
              '/img2img/sketch-mountains-input.jpg')
          # resize to resolution that is divisible by 8 but not 16 or 32
          _UpperCAmelCase  :  str				       =init_image.resize((7_6_0, 5_0_4))

          _UpperCAmelCase  :  Dict				       ='BAAI/AltDiffusion'
          _UpperCAmelCase  :  List[Any]				       =AltDiffusionImgaImgPipeline.from_pretrained(
              snake_case    ,				safety_checker=snake_case    ,				)
          pipe.to(snake_case)
          pipe.set_progress_bar_config(disable=snake_case)
          pipe.enable_attention_slicing()

          _UpperCAmelCase  :  Any				       ='A fantasy landscape, trending on artstation'

          _UpperCAmelCase  :  List[str]				       =torch.manual_seed(0)
          _UpperCAmelCase  :  Dict				       =pipe(
              prompt=snake_case    ,				image=snake_case    ,				strength=0.75    ,				guidance_scale=7.5    ,				generator=snake_case    ,				output_type='np'    ,				)
          _UpperCAmelCase  :  int				       =output.images[0]

          _UpperCAmelCase  :  Optional[int]				       =image[2_5_5:2_5_8, 3_8_3:3_8_6, -1]

          assert image.shape == (5_0_4, 7_6_0, 3)
          _UpperCAmelCase  :  Tuple				       =np.array([0.93_58, 0.93_97, 0.95_99, 0.99_01, 1.00_00, 1.00_00, 0.98_82, 1.00_00, 1.00_00])

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






@slow
@require_torch_gpu
class    __magic_name__				(						unittest.TestCase							):






       def      lowerCAmelCase						(     self)		->    Tuple:




          '''simple docstring'''
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()






       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  List[Any]				       =load_image(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
              '/img2img/sketch-mountains-input.jpg')
          _UpperCAmelCase  :  Optional[int]				       =init_image.resize((7_6_8, 5_1_2))
          _UpperCAmelCase  :  Union[str, Any]				       =load_numpy(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/img2img/fantasy_landscape_alt.npy')

          _UpperCAmelCase  :  Optional[Any]				       ='BAAI/AltDiffusion'
          _UpperCAmelCase  :  Tuple				       =AltDiffusionImgaImgPipeline.from_pretrained(
              snake_case    ,				safety_checker=snake_case    ,				)
          pipe.to(snake_case)
          pipe.set_progress_bar_config(disable=snake_case)
          pipe.enable_attention_slicing()

          _UpperCAmelCase  :  int				       ='A fantasy landscape, trending on artstation'

          _UpperCAmelCase  :  Optional[int]				       =torch.manual_seed(0)
          _UpperCAmelCase  :  Any				       =pipe(
              prompt=snake_case    ,				image=snake_case    ,				strength=0.75    ,				guidance_scale=7.5    ,				generator=snake_case    ,				output_type='np'    ,				)
          _UpperCAmelCase  :  List[str]				       =output.images[0]

          assert image.shape == (5_1_2, 7_6_8, 3)
          # img2img is flaky across GPUs even in fp32, so using MAE here
          assert np.abs(expected_image - image).max() < 1E-2





'''simple docstring'''






import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding






class    __magic_name__				(						lowerCAmelCase							):
       UpperCAmelCase		 =["image_processor", "tokenizer"]
       UpperCAmelCase		 ="CLIPImageProcessor"
       UpperCAmelCase		 =("CLIPTokenizer", "CLIPTokenizerFast")






       def __init__(     self    ,				snake_case=None    ,				snake_case=None    ,				**snake_case)		->    Dict:




          '''simple docstring'''
          _UpperCAmelCase  :  Union[str, Any]				       =None
          if "feature_extractor" in kwargs:
             warnings.warn(
                 'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
                 ' instead.'    ,				snake_case    ,				)
             _UpperCAmelCase  :  str				       =kwargs.pop('feature_extractor')

          _UpperCAmelCase  :  List[str]				       =image_processor if image_processor is not None else feature_extractor
          if image_processor is None:
             raise ValueError('You need to specify an `image_processor`.')
          if tokenizer is None:
             raise ValueError('You need to specify a `tokenizer`.')

          super().__init__(snake_case    ,				snake_case)






       def __call__(     self    ,				snake_case=None    ,				snake_case=None    ,				snake_case=None    ,				**snake_case)		->    Optional[Any]:




          '''simple docstring'''

          if text is None and images is None:
             raise ValueError('You have to specify either text or images. Both cannot be none.')

          if text is not None:
             _UpperCAmelCase  :  List[Any]				       =self.tokenizer(snake_case    ,				return_tensors=snake_case    ,				**snake_case)

          if images is not None:
             _UpperCAmelCase  :  List[str]				       =self.image_processor(snake_case    ,				return_tensors=snake_case    ,				**snake_case)

          if text is not None and images is not None:
             _UpperCAmelCase  :  Any				       =image_features.pixel_values
             return encoding
          elif text is not None:
             return encoding
          else:
             return BatchEncoding(data=dict(**snake_case)    ,				tensor_type=snake_case)






       def      lowerCAmelCase						(     self    ,				*snake_case    ,				**snake_case)		->    int:




          '''simple docstring'''
          return self.tokenizer.batch_decode(*snake_case    ,				**snake_case)






       def      lowerCAmelCase						(     self    ,				*snake_case    ,				**snake_case)		->    Optional[int]:




          '''simple docstring'''
          return self.tokenizer.decode(*snake_case    ,				**snake_case)






       @property
       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  List[Any]				       =self.tokenizer.model_input_names
          _UpperCAmelCase  :  Any				       =self.image_processor.model_input_names
          return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))






       @property
       def      lowerCAmelCase						(     self)		->    Any:




          '''simple docstring'''
          warnings.warn(
              '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'    ,				snake_case    ,				)
          return self.image_processor_class






       @property
       def      lowerCAmelCase						(     self)		->    Optional[int]:




          '''simple docstring'''
          warnings.warn(
              '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'    ,				snake_case    ,				)
          return self.image_processor

