



'''simple docstring'''






import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCAmelCase: Any							=						logging.get_logger(__name__)

lowerCAmelCase: Union[str, Any]							=						{
    'BAAI/AltCLIP': 'https://huggingface.co/BAAI/AltCLIP/resolve/main/config.json',
    # See all AltCLIP models at https://huggingface.co/models?filter=altclip
}

class 							a__( lowerCamelCase__     ):
    lowercase__			=   """altclip_text_model"""







    def __init__(				self			:  List[Any]						,      __snake_case			:  Tuple=25_00_02						,      __snake_case			:  List[Any]=10_24						,      __snake_case			:  Union[str, Any]=24						,      __snake_case			:  str=16						,      __snake_case			:  str=40_96						,      __snake_case			:  Optional[Any]="gelu"						,      __snake_case			:  List[Any]=0.1						,      __snake_case			:  Any=0.1						,      __snake_case			:  List[str]=5_14						,      __snake_case			:  Tuple=1						,      __snake_case			:  Any=0.02						,      __snake_case			:  List[str]=0.02						,      __snake_case			:  List[Any]=1e-0_5						,      __snake_case			:  Dict=1						,      __snake_case			:  int=0						,      __snake_case			:  Any=2						,      __snake_case			:  Optional[Any]="absolute"						,      __snake_case			:  Optional[Any]=True						,      __snake_case			:  List[str]=7_68						,      **__snake_case			:  Any						,      ):
      super().__init__(pad_token_id=__snake_case						,      bos_token_id=__snake_case						,      eos_token_id=__snake_case						,      **__snake_case     )

      a						: int					       =					vocab_size
      a						: Tuple					       =					hidden_size
      a						: List[Any]					       =					num_hidden_layers
      a						: int					       =					num_attention_heads
      a						: List[Any]					       =					hidden_act
      a						: str					       =					intermediate_size
      a						: Union[str, Any]					       =					hidden_dropout_prob
      a						: Dict					       =					attention_probs_dropout_prob
      a						: Union[str, Any]					       =					max_position_embeddings
      a						: Union[str, Any]					       =					type_vocab_size
      a						: int					       =					initializer_range
      a						: Union[str, Any]					       =					initializer_factor
      a						: Dict					       =					layer_norm_eps
      a						: List[str]					       =					position_embedding_type
      a						: Any					       =					use_cache
      a						: Optional[Any]					       =					project_dim

class 							a__( lowerCamelCase__     ):
    lowercase__			=   """altclip_vision_model"""







    def __init__(				self			:  Union[str, Any]						,      __snake_case			:  Any=7_68						,      __snake_case			:  str=30_72						,      __snake_case			:  Tuple=5_12						,      __snake_case			:  int=12						,      __snake_case			:  Any=12						,      __snake_case			:  List[Any]=3						,      __snake_case			:  Optional[int]=2_24						,      __snake_case			:  str=32						,      __snake_case			:  Union[str, Any]="quick_gelu"						,      __snake_case			:  Any=1e-5						,      __snake_case			:  Tuple=0.0						,      __snake_case			:  Union[str, Any]=0.02						,      __snake_case			:  List[str]=1.0						,      **__snake_case			:  Union[str, Any]						,      ):
      super().__init__(**__snake_case     )

      a						: str					       =					hidden_size
      a						: Tuple					       =					intermediate_size
      a						: Any					       =					projection_dim
      a						: List[str]					       =					num_hidden_layers
      a						: Optional[Any]					       =					num_attention_heads
      a						: List[str]					       =					num_channels
      a						: List[str]					       =					patch_size
      a						: int					       =					image_size
      a						: Union[str, Any]					       =					initializer_range
      a						: Tuple					       =					initializer_factor
      a						: Union[str, Any]					       =					attention_dropout
      a						: List[str]					       =					layer_norm_eps
      a						: int					       =					hidden_act







    @classmethod
    def       lowercase_	(				cls			:  int						,      __snake_case			:  Union[str, os.PathLike]						,      **__snake_case			:  Dict     ):
      cls._set_token_in_kwargs(__snake_case     )

      a       ,     a						: Optional[Any]					       =					cls.get_config_dict(__snake_case						,      **__snake_case     )

      # get the vision config dict if we are loading from AltCLIPConfig
      if config_dict.get('model_type'     ) == "altclip":
        a						: int					       =					config_dict['vision_config']

      if "model_type" in config_dict and hasattr(cls						,      'model_type'     ) and config_dict["model_type"] != cls.model_type:
        logger.warning(
            F"""You are using a model of type {config_dict["model_type"]} to instantiate a model of type """
            F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors."""     )

      return cls.from_dict(__snake_case						,      **__snake_case     )

class 							a__( lowerCamelCase__     ):
    lowercase__			=   """altclip"""
    lowercase__			=   True







    def __init__(				self			:  str						,      __snake_case			:  Optional[int]=None						,      __snake_case			:  Optional[int]=None						,      __snake_case			:  Tuple=7_68						,      __snake_case			:  Any=2.6592						,      **__snake_case			:  List[Any]     ):
      # If `_config_dict` exist, we use them for the backward compatibility.
      # We pop out these 2 attributes before calling `super().__init__` to avoid them being saved (which causes a lot
      # of confusion!).
      a						: Optional[int]					       =					kwargs.pop('text_config_dict'						,      __snake_case     )
      a						: Dict					       =					kwargs.pop('vision_config_dict'						,      __snake_case     )

      super().__init__(**__snake_case     )

      # Instead of simply assigning `[text|vision]_config_dict` to `[text|vision]_config`, we use the values in
      # `[text|vision]_config_dict` to update the values in `[text|vision]_config`. The values should be same in most
      # cases, but we don't want to break anything regarding `_config_dict` that existed before commit `8827e1b2`.
      if text_config_dict is not None:
        if text_config is None:
          a						: List[Any]					       =					{}

        # This is the complete result when using `text_config_dict`.
        a						: Union[str, Any]					       =					AltCLIPTextConfig(**__snake_case     ).to_dict()

        # Give a warning if the values exist in both `_text_config_dict` and `text_config` but being different.
        for key, value in _text_config_dict.items():
          if key in text_config and value != text_config[key] and key not in ["transformers_version"]:
            # If specified in `text_config_dict`
            if key in text_config_dict:
              a						: List[str]					       =					(
                  F"""`{key}` is found in both `text_config_dict` and `text_config` but with different values. """
                  F"""The value `text_config_dict[\"{key}\"]` will be used instead."""
              )
            # If inferred from default argument values (just to be super careful)
            else:
              a						: Any					       =					(
                  F"""`text_config_dict` is provided which will be used to initialize `AltCLIPTextConfig`. The """
                  F"""value `text_config[\"{key}\"]` will be overriden."""
              )
            logger.warning(__snake_case     )

            # Update all values in `text_config` with the ones in `_text_config_dict`.
        text_config.update(_text_config_dict     )

      if vision_config_dict is not None:
        if vision_config is None:
          a						: Optional[int]					       =					{}

        # This is the complete result when using `vision_config_dict`.
        a						: Optional[int]					       =					AltCLIPVisionConfig(**__snake_case     ).to_dict()
        # convert keys to string instead of integer
        if "id2label" in _vision_config_dict:
          a						: Optional[int]					       =					{
              str(__snake_case     ): value for key, value in _vision_config_dict['id2label'].items()
          }

        # Give a warning if the values exist in both `_vision_config_dict` and `vision_config` but being different.
        for key, value in _vision_config_dict.items():
          if key in vision_config and value != vision_config[key] and key not in ["transformers_version"]:
            # If specified in `vision_config_dict`
            if key in vision_config_dict:
              a						: List[str]					       =					(
                  F"""`{key}` is found in both `vision_config_dict` and `vision_config` but with different """
                  F"""values. The value `vision_config_dict[\"{key}\"]` will be used instead."""
              )
            # If inferred from default argument values (just to be super careful)
            else:
              a						: Optional[Any]					       =					(
                  F"""`vision_config_dict` is provided which will be used to initialize `AltCLIPVisionConfig`. """
                  F"""The value `vision_config[\"{key}\"]` will be overriden."""
              )
            logger.warning(__snake_case     )

            # Update all values in `vision_config` with the ones in `_vision_config_dict`.
        vision_config.update(_vision_config_dict     )

      if text_config is None:
        a						: List[Any]					       =					{}
        logger.info('`text_config` is `None`. Initializing the `AltCLIPTextConfig` with default values.'     )

      if vision_config is None:
        a						: str					       =					{}
        logger.info('`vision_config` is `None`. initializing the `AltCLIPVisionConfig` with default values.'     )

      a						: Any					       =					AltCLIPTextConfig(**__snake_case     )
      a						: List[Any]					       =					AltCLIPVisionConfig(**__snake_case     )

      a						: str					       =					projection_dim
      a						: int					       =					logit_scale_init_value
      a						: Optional[Any]					       =					1.0







    @classmethod
    def       lowercase_	(				cls			:  Any						,      __snake_case			:  AltCLIPTextConfig						,      __snake_case			:  AltCLIPVisionConfig						,      **__snake_case			:  str     ):
      return cls(text_config=text_config.to_dict()						,      vision_config=vision_config.to_dict()						,      **__snake_case     )







    def       lowercase_	(				self			:  Optional[Any]     ):
      a						: Dict					       =					copy.deepcopy(self.__dict__     )
      a						: Union[str, Any]					       =					self.text_config.to_dict()
      a						: Union[str, Any]					       =					self.vision_config.to_dict()
      a						: Union[str, Any]					       =					self.__class__.model_type
      return output




'''simple docstring'''






import gc
import random
import unittest

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

from diffusers import AutoencoderKL, CycleDiffusionPipeline, DDIMScheduler, UNetaDConditionModel
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, skip_mps

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class 							a__( lowerCamelCase__		,			lowerCamelCase__		,			unittest.TestCase     ):
    lowercase__			=   CycleDiffusionPipeline
    lowercase__			=   TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
        """negative_prompt""",
        """height""",
        """width""",
        """negative_prompt_embeds""",
    }
    lowercase__			=   PipelineTesterMixin.required_optional_params - {"""latents"""}
    lowercase__			=   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""source_prompt"""}     )
    lowercase__			=   IMAGE_TO_IMAGE_IMAGE_PARAMS
    lowercase__			=   IMAGE_TO_IMAGE_IMAGE_PARAMS







    def       lowercase_	(				self			:  Any     ):
      torch.manual_seed(0     )
      a						: Union[str, 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						,      )
      a						: str					       =					DDIMScheduler(
          beta_start=0.00085						,      beta_end=0.012						,      beta_schedule='scaled_linear'						,      num_train_timesteps=10_00						,      clip_sample=__snake_case						,      set_alpha_to_one=__snake_case						,      )
      torch.manual_seed(0     )
      a						: List[str]					       =					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						,      )
      torch.manual_seed(0     )
      a						: int					       =					CLIPTextConfig(
          bos_token_id=0						,      eos_token_id=2						,      hidden_size=32						,      intermediate_size=37						,      layer_norm_eps=1e-0_5						,      num_attention_heads=4						,      num_hidden_layers=5						,      pad_token_id=1						,      vocab_size=10_00						,      )
      a						: List[str]					       =					CLIPTextModel(__snake_case     )
      a						: int					       =					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

      a						: Tuple					       =					{
          'unet': unet,
          'scheduler': scheduler,
          'vae': vae,
          'text_encoder': text_encoder,
          'tokenizer': tokenizer,
          'safety_checker': None,
          'feature_extractor': None,
      }
      return components







    def       lowercase_	(				self			:  Optional[int]						,      __snake_case			:  Dict						,      __snake_case			:  Any=0     ):
      a						: str					       =					floats_tensor((1, 3, 32, 32)						,      rng=random.Random(__snake_case     )     ).to(__snake_case     )
      a						: Optional[Any]					       =					image / 2 + 0.5
      if str(__snake_case     ).startswith('mps'     ):
        a						: List[str]					       =					torch.manual_seed(__snake_case     )
      else:
        a						: Union[str, Any]					       =					torch.Generator(device=__snake_case     ).manual_seed(__snake_case     )
      a						: List[Any]					       =					{
          'prompt': 'An astronaut riding an elephant',
          'source_prompt': 'An astronaut riding a horse',
          'image': image,
          'generator': generator,
          'num_inference_steps': 2,
          'eta': 0.1,
          'strength': 0.8,
          'guidance_scale': 3,
          'source_guidance_scale': 1,
          'output_type': 'numpy',
      }
      return inputs







    def       lowercase_	(				self			:  Optional[int]     ):
      a						: Optional[Any]					       =					'cpu'  # ensure determinism for the device-dependent torch.Generator

      a						: int					       =					self.get_dummy_components()
      a						: str					       =					CycleDiffusionPipeline(**__snake_case     )
      a						: List[str]					       =					pipe.to(__snake_case     )
      pipe.set_progress_bar_config(disable=__snake_case     )

      a						: Dict					       =					self.get_dummy_inputs(__snake_case     )
      a						: Union[str, Any]					       =					pipe(**__snake_case     )
      a						: List[Any]					       =					output.images

      a						: Optional[Any]					       =					images[0, -3:, -3:, -1]

      assert images.shape == (1, 32, 32, 3)
      a						: Tuple					       =					np.array([0.4459, 0.4943, 0.4544, 0.6643, 0.5474, 0.4327, 0.5701, 0.5959, 0.5179]     )

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







    @unittest.skipIf(torch_device != 'cuda'						,      'This test requires a GPU'     )
    def       lowercase_	(				self			:  int     ):
      a						: List[Any]					       =					self.get_dummy_components()
      for name, module in components.items():
        if hasattr(__snake_case						,      'half'     ):
          a						: Any					       =					module.half()
      a						: Tuple					       =					CycleDiffusionPipeline(**__snake_case     )
      a						: Any					       =					pipe.to(__snake_case     )
      pipe.set_progress_bar_config(disable=__snake_case     )

      a						: str					       =					self.get_dummy_inputs(__snake_case     )
      a						: int					       =					pipe(**__snake_case     )
      a						: Optional[int]					       =					output.images

      a						: Tuple					       =					images[0, -3:, -3:, -1]

      assert images.shape == (1, 32, 32, 3)
      a						: int					       =					np.array([0.3506, 0.4543, 0.446, 0.4575, 0.5195, 0.4155, 0.5273, 0.518, 0.4116]     )

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







    @skip_mps
    def       lowercase_	(				self			:  List[Any]     ):
      return super().test_save_load_local()







    @unittest.skip('non-deterministic pipeline'     )
    def       lowercase_	(				self			:  Dict     ):
      return super().test_inference_batch_single_identical()







    @skip_mps
    def       lowercase_	(				self			:  int     ):
      return super().test_dict_tuple_outputs_equivalent()







    @skip_mps
    def       lowercase_	(				self			:  Dict     ):
      return super().test_save_load_optional_components()







    @skip_mps
    def       lowercase_	(				self			:  List[Any]     ):
      return super().test_attention_slicing_forward_pass()

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







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







    def       lowercase_	(				self			:  Optional[int]     ):
      a						: Any					       =					load_image(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
          '/cycle-diffusion/black_colored_car.png'     )
      a						: Optional[int]					       =					load_numpy(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'     )
      a						: List[str]					       =					init_image.resize((5_12, 5_12)     )

      a						: Dict					       =					'CompVis/stable-diffusion-v1-4'
      a						: List[str]					       =					DDIMScheduler.from_pretrained(__snake_case						,      subfolder='scheduler'     )
      a						: Any					       =					CycleDiffusionPipeline.from_pretrained(
          __snake_case						,      scheduler=__snake_case						,      safety_checker=__snake_case						,      torch_dtype=torch.floataa						,      revision='fp16'     )

      pipe.to(__snake_case     )
      pipe.set_progress_bar_config(disable=__snake_case     )
      pipe.enable_attention_slicing()

      a						: Union[str, Any]					       =					'A black colored car'
      a						: Optional[Any]					       =					'A blue colored car'

      a						: int					       =					torch.manual_seed(0     )
      a						: Optional[Any]					       =					pipe(
          prompt=__snake_case						,      source_prompt=__snake_case						,      image=__snake_case						,      num_inference_steps=1_00						,      eta=0.1						,      strength=0.85						,      guidance_scale=3						,      source_guidance_scale=1						,      generator=__snake_case						,      output_type='np'						,      )
      a						: Dict					       =					output.images

      # the values aren't exactly equal, but the images look the same visually
      assert np.abs(image - expected_image     ).max() < 5e-1







    def       lowercase_	(				self			:  int     ):
      a						: Any					       =					load_image(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
          '/cycle-diffusion/black_colored_car.png'     )
      a						: List[str]					       =					load_numpy(
          'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'     )
      a						: str					       =					init_image.resize((5_12, 5_12)     )

      a						: Optional[int]					       =					'CompVis/stable-diffusion-v1-4'
      a						: Union[str, Any]					       =					DDIMScheduler.from_pretrained(__snake_case						,      subfolder='scheduler'     )
      a						: str					       =					CycleDiffusionPipeline.from_pretrained(__snake_case						,      scheduler=__snake_case						,      safety_checker=__snake_case     )

      pipe.to(__snake_case     )
      pipe.set_progress_bar_config(disable=__snake_case     )
      pipe.enable_attention_slicing()

      a						: Tuple					       =					'A black colored car'
      a						: Tuple					       =					'A blue colored car'

      a						: List[str]					       =					torch.manual_seed(0     )
      a						: str					       =					pipe(
          prompt=__snake_case						,      source_prompt=__snake_case						,      image=__snake_case						,      num_inference_steps=1_00						,      eta=0.1						,      strength=0.85						,      guidance_scale=3						,      source_guidance_scale=1						,      generator=__snake_case						,      output_type='np'						,      )
      a						: Tuple					       =					output.images

      assert np.abs(image - expected_image     ).max() < 2e-2