





import copy
from typing import Dict, List, Optional

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto import CONFIG_MAPPING


snake_case_           =      {
    'facebook/mask2former-swin-small-coco-instance': (
        'https://huggingface.co/facebook/mask2former-swin-small-coco-instance/blob/main/config.json'
    )
    # See all Mask2Former models at https://huggingface.co/models?filter=mask2former
}

snake_case_           =      logging.get_logger(__name__)

class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):
      A_						:    List[Any]							=   'mask2former'
      A_						:    int							=   ['swin']
      A_						:    Optional[Any]							=   {'hidden_size': 'hidden_dim'}


      def __init__(self	:				Tuple    ,      a__	:				Optional[Dict] = None    ,      a__	:				int = 256    ,      a__	:				int = 256    ,      a__	:				int = 256    ,      a__	:				int = 1024    ,      a__	:				str = "relu"    ,      a__	:				int = 6    ,      a__	:				int = 10    ,      a__	:				int = 8    ,      a__	:				float = 0.0    ,      a__	:				int = 2048    ,      a__	:				bool = False    ,      a__	:				bool = False    ,      a__	:				int = 4    ,      a__	:				int = 255    ,      a__	:				int = 100    ,      a__	:				float = 0.1    ,      a__	:				float = 2.0    ,      a__	:				float = 5.0    ,      a__	:				float = 5.0    ,      a__	:				int = 1_2544    ,      a__	:				float = 3.0    ,      a__	:				float = 0.7_5    ,      a__	:				float = 0.0_2    ,      a__	:				float = 1.0    ,      a__	:				bool = True    ,      a__	:				List[int] = [4, 8, 16, 32]    ,      a__	:				bool = None    ,      **a__	:				Optional[int]    ,      ):





         """simple docstring"""



         if backbone_config is None:
            logger.info('''`backbone_config` is `None`. Initializing the config with the default `Swin` backbone.'''				)
            __snake_case								=       CONFIG_MAPPING['''swin'''](
                image_size=224    ,      in_channels=3    ,      patch_size=4    ,      embed_dim=96    ,      depths=[2, 2, 18, 2]    ,      num_heads=[3, 6, 12, 24]    ,      window_size=7    ,      drop_path_rate=0.3    ,      use_absolute_embeddings=a__    ,      out_features=['''stage1''', '''stage2''', '''stage3''', '''stage4''']    ,      )

         if isinstance(a__    ,      a__				):
            __snake_case								=       backbone_config.pop('''model_type'''				)
            __snake_case								=       CONFIG_MAPPING[backbone_model_type]
            __snake_case								=       config_class.from_dict(a__				)

         # verify that the backbone is supported
         if backbone_config.model_type not in self.backbones_supported:
            logger.warning_once(
                f"""Backbone {backbone_config.model_type} is not a supported model and may not be compatible with Mask2Former. """
                f"""Supported model types: {','.join(self.backbones_supported				)}"""				)

         __snake_case								=       backbone_config
         __snake_case								=       feature_size
         __snake_case								=       mask_feature_size
         __snake_case								=       hidden_dim
         __snake_case								=       encoder_feedforward_dim
         __snake_case								=       activation_function
         __snake_case								=       encoder_layers
         __snake_case								=       decoder_layers
         __snake_case								=       num_attention_heads
         __snake_case								=       dropout
         __snake_case								=       dim_feedforward
         __snake_case								=       pre_norm
         __snake_case								=       enforce_input_projection
         __snake_case								=       common_stride
         __snake_case								=       ignore_value
         __snake_case								=       num_queries
         __snake_case								=       no_object_weight
         __snake_case								=       class_weight
         __snake_case								=       mask_weight
         __snake_case								=       dice_weight
         __snake_case								=       train_num_points
         __snake_case								=       oversample_ratio
         __snake_case								=       importance_sample_ratio
         __snake_case								=       init_std
         __snake_case								=       init_xavier_std
         __snake_case								=       use_auxiliary_loss
         __snake_case								=       feature_strides
         __snake_case								=       output_auxiliary_logits
         __snake_case								=       decoder_layers

         super().__init__(**a__				)


      @classmethod
      def        a			(cls	:				Optional[int]    ,      a__	:				PretrainedConfig    ,      **a__	:				Optional[int]				):





         """simple docstring"""



         return cls(
             backbone_config=a__    ,      **a__    ,      )


      def        a			(self	:				Any				):





         """simple docstring"""



         __snake_case								=       copy.deepcopy(self.__dict__				)
         __snake_case								=       self.backbone_config.to_dict()
         __snake_case								=       self.__class__.model_type
         return output








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 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase						,				_UpperCAmelCase						,				unittest.TestCase  ):
      A_						:    Optional[int]							=   CycleDiffusionPipeline
      A_						:    List[Any]							=   TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
          'negative_prompt',
          'height',
          'width',
          'negative_prompt_embeds',
      }
      A_						:    List[str]							=   PipelineTesterMixin.required_optional_params - {'latents'}
      A_						:    List[str]							=   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'source_prompt'}  )
      A_						:    List[str]							=   IMAGE_TO_IMAGE_IMAGE_PARAMS
      A_						:    Any							=   IMAGE_TO_IMAGE_IMAGE_PARAMS


      def        a			(self	:				Dict				):





         """simple docstring"""



         torch.manual_seed(0				)
         __snake_case								=       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    ,      )
         __snake_case								=       DDIMScheduler(
             beta_start=0.0_0_0_8_5    ,      beta_end=0.0_1_2    ,      beta_schedule='''scaled_linear'''    ,      num_train_timesteps=1000    ,      clip_sample=a__    ,      set_alpha_to_one=a__    ,      )
         torch.manual_seed(0				)
         __snake_case								=       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				)
         __snake_case								=       CLIPTextConfig(
             bos_token_id=0    ,      eos_token_id=2    ,      hidden_size=32    ,      intermediate_size=37    ,      layer_norm_eps=1E-05    ,      num_attention_heads=4    ,      num_hidden_layers=5    ,      pad_token_id=1    ,      vocab_size=1000    ,      )
         __snake_case								=       CLIPTextModel(a__				)
         __snake_case								=       CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				)

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


      def        a			(self	:				List[str]    ,      a__	:				Tuple    ,      a__	:				Optional[Any]=0				):





         """simple docstring"""



         __snake_case								=       floats_tensor((1, 3, 32, 32)    ,      rng=random.Random(a__				)				).to(a__				)
         __snake_case								=       image / 2 + 0.5
         if str(a__				).startswith('''mps'''				):
            __snake_case								=       torch.manual_seed(a__				)
         else:
            __snake_case								=       torch.Generator(device=a__				).manual_seed(a__				)
         __snake_case								=       {
             '''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        a			(self	:				str				):





         """simple docstring"""



         __snake_case								=       '''cpu'''  # ensure determinism for the device-dependent torch.Generator

         __snake_case								=       self.get_dummy_components()
         __snake_case								=       CycleDiffusionPipeline(**a__				)
         __snake_case								=       pipe.to(a__				)
         pipe.set_progress_bar_config(disable=a__				)

         __snake_case								=       self.get_dummy_inputs(a__				)
         __snake_case								=       pipe(**a__				)
         __snake_case								=       output.images

         __snake_case								=       images[0, -3:, -3:, -1]

         assert images.shape == (1, 32, 32, 3)
         __snake_case								=       np.array([0.4_4_5_9, 0.4_9_4_3, 0.4_5_4_4, 0.6_6_4_3, 0.5_4_7_4, 0.4_3_2_7, 0.5_7_0_1, 0.5_9_5_9, 0.5_1_7_9]				)

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


      @unittest.skipIf(torch_device != '''cuda'''    ,      '''This test requires a GPU'''				)
      def        a			(self	:				List[str]				):





         """simple docstring"""



         __snake_case								=       self.get_dummy_components()
         for name, module in components.items():
            if hasattr(a__    ,      '''half'''				):
               __snake_case								=       module.half()
         __snake_case								=       CycleDiffusionPipeline(**a__				)
         __snake_case								=       pipe.to(a__				)
         pipe.set_progress_bar_config(disable=a__				)

         __snake_case								=       self.get_dummy_inputs(a__				)
         __snake_case								=       pipe(**a__				)
         __snake_case								=       output.images

         __snake_case								=       images[0, -3:, -3:, -1]

         assert images.shape == (1, 32, 32, 3)
         __snake_case								=       np.array([0.3_5_0_6, 0.4_5_4_3, 0.4_4_6, 0.4_5_7_5, 0.5_1_9_5, 0.4_1_5_5, 0.5_2_7_3, 0.5_1_8, 0.4_1_1_6]				)

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


      @skip_mps
      def        a			(self	:				Any				):





         """simple docstring"""



         return super().test_save_load_local()


      @unittest.skip('''non-deterministic pipeline'''				)
      def        a			(self	:				Any				):





         """simple docstring"""



         return super().test_inference_batch_single_identical()


      @skip_mps
      def        a			(self	:				Any				):





         """simple docstring"""



         return super().test_dict_tuple_outputs_equivalent()


      @skip_mps
      def        a			(self	:				str				):





         """simple docstring"""



         return super().test_save_load_optional_components()


      @skip_mps
      def        a			(self	:				Dict				):





         """simple docstring"""



         return super().test_attention_slicing_forward_pass()






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


      def        a			(self	:				Union[str, Any]				):





         """simple docstring"""



         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()


      def        a			(self	:				Tuple				):





         """simple docstring"""



         __snake_case								=       load_image(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
             '''/cycle-diffusion/black_colored_car.png'''				)
         __snake_case								=       load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car_fp16.npy'''				)
         __snake_case								=       init_image.resize((512, 512)				)

         __snake_case								=       '''CompVis/stable-diffusion-v1-4'''
         __snake_case								=       DDIMScheduler.from_pretrained(a__    ,      subfolder='''scheduler'''				)
         __snake_case								=       CycleDiffusionPipeline.from_pretrained(
             a__    ,      scheduler=a__    ,      safety_checker=a__    ,      torch_dtype=torch.floataa    ,      revision='''fp16'''				)

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

         __snake_case								=       '''A black colored car'''
         __snake_case								=       '''A blue colored car'''

         __snake_case								=       torch.manual_seed(0				)
         __snake_case								=       pipe(
             prompt=a__    ,      source_prompt=a__    ,      image=a__    ,      num_inference_steps=100    ,      eta=0.1    ,      strength=0.8_5    ,      guidance_scale=3    ,      source_guidance_scale=1    ,      generator=a__    ,      output_type='''np'''    ,      )
         __snake_case								=       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        a			(self	:				Tuple				):





         """simple docstring"""



         __snake_case								=       load_image(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
             '''/cycle-diffusion/black_colored_car.png'''				)
         __snake_case								=       load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/cycle-diffusion/blue_colored_car.npy'''				)
         __snake_case								=       init_image.resize((512, 512)				)

         __snake_case								=       '''CompVis/stable-diffusion-v1-4'''
         __snake_case								=       DDIMScheduler.from_pretrained(a__    ,      subfolder='''scheduler'''				)
         __snake_case								=       CycleDiffusionPipeline.from_pretrained(a__    ,      scheduler=a__    ,      safety_checker=a__				)

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

         __snake_case								=       '''A black colored car'''
         __snake_case								=       '''A blue colored car'''

         __snake_case								=       torch.manual_seed(0				)
         __snake_case								=       pipe(
             prompt=a__    ,      source_prompt=a__    ,      image=a__    ,      num_inference_steps=100    ,      eta=0.1    ,      strength=0.8_5    ,      guidance_scale=3    ,      source_guidance_scale=1    ,      generator=a__    ,      output_type='''np'''    ,      )
         __snake_case								=       output.images

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

