


import inspect
import unittest

import numpy as np

from tests.test_modeling_common import floats_tensor
from transformers import DetrConfig, MaskFormerConfig, SwinConfig, is_torch_available, is_vision_available
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
from transformers.utils import cached_property

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
   import torch

   from transformers import MaskFormerForInstanceSegmentation, MaskFormerModel

   if is_vision_available():
      from transformers import MaskFormerImageProcessor

if is_vision_available():
   from PIL import Image

class     a		:




       """simple docstring"""


       def __init__(							self	:  Tuple   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  int=2   ,     lowerCamelCase	:  List[str]=True   ,     lowerCamelCase	:  Optional[int]=False   ,     lowerCamelCase	:  Union[str, Any]=10   ,     lowerCamelCase	:  Optional[Any]=3   ,     lowerCamelCase	:  int=32 * 4   ,     lowerCamelCase	:  Any=32 * 6   ,     lowerCamelCase	:  int=4   ,     lowerCamelCase	:  List[str]=32   ,     )     ->   List[str]:
              __snake_case		:      Optional[Any]										=       parent
              __snake_case		:      int										=       batch_size
              __snake_case		:      List[Any]										=       is_training
              __snake_case		:      Dict										=       use_auxiliary_loss
              __snake_case		:      Optional[int]										=       num_queries
              __snake_case		:      Union[str, Any]										=       num_channels
              __snake_case		:      Tuple										=       min_size
              __snake_case		:      int										=       max_size
              __snake_case		:      Union[str, Any]										=       num_labels
              __snake_case		:      int										=       mask_feature_size

       def    __snake_case (							self	:  int    )     ->   List[Any]:
              __snake_case		:      Optional[Any]										=       floats_tensor([self.batch_size, self.num_channels, self.min_size, self.max_size]    ).to(
                  lowercase_    )

              __snake_case		:      List[Any]										=       torch.ones([self.batch_size, self.min_size, self.max_size]   ,     device=lowercase_    )

              __snake_case		:      int										=       (
                  torch.rand([self.batch_size, self.num_labels, self.min_size, self.max_size]   ,     device=lowercase_    ) > 0.5
              ).float()
              __snake_case		:      Union[str, Any]										=       (torch.rand((self.batch_size, self.num_labels)   ,     device=lowercase_    ) > 0.5).long()

              __snake_case		:      List[Any]										=       self.get_config()
              return config, pixel_values, pixel_mask, mask_labels, class_labels

       def    __snake_case (							self	:  str    )     ->   Optional[int]:
              return MaskFormerConfig.from_backbone_and_decoder_configs(
                  backbone_config=SwinConfig(
                      depths=[1, 1, 1, 1]   ,     )   ,     decoder_config=DetrConfig(
                      decoder_ffn_dim=128   ,     num_queries=self.num_queries   ,     decoder_attention_heads=2   ,     d_model=self.mask_feature_size   ,     )   ,     mask_feature_size=self.mask_feature_size   ,     fpn_feature_size=self.mask_feature_size   ,     num_channels=self.num_channels   ,     num_labels=self.num_labels   ,     )

       def    __snake_case (							self	:  Optional[Any]    )     ->   List[str]:
              __snake_case			,      __snake_case			,      __snake_case			,      __snake_case			,      __snake_case		:      Any										=       self.prepare_config_and_inputs()
              __snake_case		:      List[str]										=       {"pixel_values": pixel_values, "pixel_mask": pixel_mask}
              return config, inputs_dict

       def    __snake_case (							self	:  str   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  List[Any]    )     ->   Optional[Any]:
              __snake_case		:      Dict										=       output.encoder_hidden_states
              __snake_case		:      Tuple										=       output.pixel_decoder_hidden_states
              __snake_case		:      Union[str, Any]										=       output.transformer_decoder_hidden_states

              self.parent.assertTrue(len(lowercase_    )   ,     len(config.backbone_config.depths    )    )
              self.parent.assertTrue(len(lowercase_    )   ,     len(config.backbone_config.depths    )    )
              self.parent.assertTrue(len(lowercase_    )   ,     config.decoder_config.decoder_layers    )

       def    __snake_case (							self	:  Optional[int]   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  List[str]   ,     lowerCamelCase	:  Optional[Any]   ,     lowerCamelCase	:  Any=False    )     ->   Optional[Any]:
              with torch.no_grad():
                     __snake_case		:      Union[str, Any]										=       MaskFormerModel(config=lowercase_    )
                     model.to(lowercase_    )
                     model.eval()

                     __snake_case		:      Optional[int]										=       model(pixel_values=lowercase_   ,     pixel_mask=lowercase_    )
                     __snake_case		:      Optional[int]										=       model(lowercase_   ,     output_hidden_states=lowercase_    )
              # the correct shape of output.transformer_decoder_hidden_states ensure the correcteness of the
              # encoder and pixel decoder
              self.parent.assertEqual(
                  output.transformer_decoder_last_hidden_state.shape   ,     (self.batch_size, self.num_queries, self.mask_feature_size)   ,     )
              # let's ensure the other two hidden state exists
              self.parent.assertTrue(output.pixel_decoder_last_hidden_state is not None    )
              self.parent.assertTrue(output.encoder_last_hidden_state is not None    )

              if output_hidden_states:
                     self.check_output_hidden_state(lowercase_   ,     lowercase_    )






       def    __snake_case (							self	:  Optional[int]   ,     lowerCamelCase	:  List[Any]   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Tuple   ,     lowerCamelCase	:  Optional[Any]    )     ->   str:
              __snake_case		:      int										=       MaskFormerForInstanceSegmentation(config=lowercase_    )
              model.to(lowercase_    )
              model.eval()

              def comm_check_on_output(lowerCamelCase	:  Any    ):
                     # let's still check that all the required stuff is there
                     self.parent.assertTrue(result.transformer_decoder_last_hidden_state is not None    )
                     self.parent.assertTrue(result.pixel_decoder_last_hidden_state is not None    )
                     self.parent.assertTrue(result.encoder_last_hidden_state is not None    )
                     # okay, now we need to check the logits shape
                     # due to the encoder compression, masks have a //4 spatial size
                     self.parent.assertEqual(
                         result.masks_queries_logits.shape   ,     (self.batch_size, self.num_queries, self.min_size // 4, self.max_size // 4)   ,     )
                     # + 1 for null class
                     self.parent.assertEqual(
                         result.class_queries_logits.shape   ,     (self.batch_size, self.num_queries, self.num_labels + 1)    )

              with torch.no_grad():
                     __snake_case		:      List[Any]										=       model(pixel_values=lowercase_   ,     pixel_mask=lowercase_    )
                     __snake_case		:      Dict										=       model(lowercase_    )

                     comm_check_on_output(lowercase_    )

                     __snake_case		:      Tuple										=       model(
                         pixel_values=lowercase_   ,     pixel_mask=lowercase_   ,     mask_labels=lowercase_   ,     class_labels=lowercase_    )

              comm_check_on_output(lowercase_    )

              self.parent.assertTrue(result.loss is not None    )
              self.parent.assertEqual(result.loss.shape   ,     torch.Size([1]    )    )

@require_torch
class     a		(_lowerCamelCase					,    _lowerCamelCase					,    unittest.TestCase				):




       """simple docstring"""


       __UpperCAmelCase   :							Dict					  =  (MaskFormerModel, MaskFormerForInstanceSegmentation) if is_torch_available() else ()
       __UpperCAmelCase   :							Optional[int]					  =  (
           {"feature-extraction": MaskFormerModel, "image-segmentation": MaskFormerForInstanceSegmentation}
           if is_torch_available()
           else {}
       )

       __UpperCAmelCase   :							Optional[int]					  =  False
       __UpperCAmelCase   :							Dict					  =  False
       __UpperCAmelCase   :							List[Any]					  =  False
       __UpperCAmelCase   :							str					  =  False

       def    __snake_case (							self	:  Any    )     ->   Any:
              __snake_case		:      str										=       MaskFormerModelTester(self    )
              __snake_case		:      int										=       ConfigTester(self   ,     config_class=lowercase_   ,     has_text_modality=lowercase_    )

       def    __snake_case (							self	:  Optional[Any]    )     ->   Optional[int]:
              self.config_tester.run_common_tests()

       def    __snake_case (							self	:  List[Any]    )     ->   Dict:
              __snake_case			,      __snake_case		:      Union[str, Any]										=       self.model_tester.prepare_config_and_inputs_for_common()
              self.model_tester.create_and_check_maskformer_model(lowercase_   ,     **lowercase_   ,     output_hidden_states=lowercase_    )

       def    __snake_case (							self	:  Union[str, Any]    )     ->   Tuple:
              __snake_case		:      Optional[int]										=       self.model_tester.prepare_config_and_inputs()
              self.model_tester.create_and_check_maskformer_instance_segmentation_head_model(*lowercase_    )

       @unittest.skip(reason="MaskFormer does not use inputs_embeds"    )
       def    __snake_case (							self	:  Optional[Any]    )     ->   Dict:
              pass

       @unittest.skip(reason="MaskFormer does not have a get_input_embeddings method"    )
       def    __snake_case (							self	:  str    )     ->   Tuple:
              pass

       @unittest.skip(reason="MaskFormer is not a generative model"    )
       def    __snake_case (							self	:  Any    )     ->   Dict:
              pass

       @unittest.skip(reason="MaskFormer does not use token embeddings"    )
       def    __snake_case (							self	:  List[str]    )     ->   Optional[Any]:
              pass

       @require_torch_multi_gpu
       @unittest.skip(
           reason="MaskFormer has some layers using `add_module` which doesn\'t work well with `nn.DataParallel`"    )
       def    __snake_case (							self	:  Optional[Any]    )     ->   str:
              pass

       @unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."    )
       def    __snake_case (							self	:  List[Any]    )     ->   Any:
              pass

       def    __snake_case (							self	:  str    )     ->   Optional[int]:
              __snake_case			,      __snake_case		:      str										=       self.model_tester.prepare_config_and_inputs_for_common()

              for model_class in self.all_model_classes:
                     __snake_case		:      Optional[Any]										=       model_class(lowercase_    )
                     __snake_case		:      Union[str, Any]										=       inspect.signature(model.forward    )
                     # signature.parameters is an OrderedDict => so arg_names order is deterministic
                     __snake_case		:      Dict										=       [*signature.parameters.keys()]

                     __snake_case		:      Tuple										=       ["pixel_values"]
                     self.assertListEqual(arg_names[:1]   ,     lowercase_    )

       @slow
       def    __snake_case (							self	:  List[str]    )     ->   int:
              for model_name in ["facebook/maskformer-swin-small-coco"]:
                     __snake_case		:      Tuple										=       MaskFormerModel.from_pretrained(lowercase_    )
                     self.assertIsNotNone(lowercase_    )

       def    __snake_case (							self	:  int    )     ->   Any:
              __snake_case		:      str										=       (self.model_tester.min_size,) * 2
              __snake_case		:      Optional[Any]										=       {
                  "pixel_values": torch.randn((2, 3, *size)   ,     device=lowercase_    ),
                  "mask_labels": torch.randn((2, 10, *size)   ,     device=lowercase_    ),
                  "class_labels": torch.zeros(2   ,     10   ,     device=lowercase_    ).long(),
              }

              __snake_case		:      Union[str, Any]										=       MaskFormerForInstanceSegmentation(MaskFormerConfig()    ).to(lowercase_    )
              __snake_case		:      Tuple										=       model(**lowercase_    )
              self.assertTrue(outputs.loss is not None    )

       def    __snake_case (							self	:  Optional[Any]    )     ->   Optional[int]:
              __snake_case			,      __snake_case		:      List[str]										=       self.model_tester.prepare_config_and_inputs_for_common()
              self.model_tester.create_and_check_maskformer_model(lowercase_   ,     **lowercase_   ,     output_hidden_states=lowercase_    )

       def    __snake_case (							self	:  str    )     ->   Tuple:
              __snake_case			,      __snake_case		:      Any										=       self.model_tester.prepare_config_and_inputs_for_common()

              for model_class in self.all_model_classes:
                     __snake_case		:      Tuple										=       model_class(lowercase_    ).to(lowercase_    )
                     __snake_case		:      int										=       model(**lowercase_   ,     output_attentions=lowercase_    )
                     self.assertTrue(outputs.attentions is not None    )

       def    __snake_case (							self	:  Union[str, Any]    )     ->   Any:
              if not self.model_tester.is_training:
                     return
              # only MaskFormerForInstanceSegmentation has the loss
              __snake_case		:      str										=       self.all_model_classes[1]
              __snake_case			,      __snake_case			,      __snake_case			,      __snake_case			,      __snake_case		:      Union[str, Any]										=       self.model_tester.prepare_config_and_inputs()

              __snake_case		:      Union[str, Any]										=       model_class(lowercase_    )
              model.to(lowercase_    )
              model.train()

              __snake_case		:      Tuple										=       model(lowercase_   ,     mask_labels=lowercase_   ,     class_labels=lowercase_    ).loss
              loss.backward()






       def    __snake_case (							self	:  List[Any]    )     ->   List[str]:
              # only MaskFormerForInstanceSegmentation has the loss
              __snake_case		:      List[Any]										=       self.all_model_classes[1]
              __snake_case			,      __snake_case			,      __snake_case			,      __snake_case			,      __snake_case		:      Union[str, Any]										=       self.model_tester.prepare_config_and_inputs()
              __snake_case		:      Dict										=       True
              __snake_case		:      Dict										=       True

              __snake_case		:      Optional[Any]										=       model_class(lowercase_    )
              model.to(lowercase_    )
              model.train()

              __snake_case		:      int										=       model(lowercase_   ,     mask_labels=lowercase_   ,     class_labels=lowercase_    )

              __snake_case		:      List[str]										=       outputs.encoder_hidden_states[0]
              encoder_hidden_states.retain_grad()

              __snake_case		:      int										=       outputs.pixel_decoder_hidden_states[0]
              pixel_decoder_hidden_states.retain_grad()
              # we requires_grad=True in inputs_embeds (line 2152), the original implementation don't
              __snake_case		:      Optional[int]										=       outputs.transformer_decoder_hidden_states[0]
              transformer_decoder_hidden_states.retain_grad()

              __snake_case		:      List[str]										=       outputs.attentions[0]
              attentions.retain_grad()

              outputs.loss.backward(retain_graph=lowercase_    )

              self.assertIsNotNone(encoder_hidden_states.grad    )
              self.assertIsNotNone(pixel_decoder_hidden_states.grad    )
              self.assertIsNotNone(transformer_decoder_hidden_states.grad    )
              self.assertIsNotNone(attentions.grad    )
_snake_case					:		str      =    1E-4







def     lowerCAmelCase_					( ):
       __snake_case		:      str										=       Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"     )
       return image

@require_vision
@slow
class     a		(unittest.TestCase				):




       """simple docstring"""


       @cached_property
       def    __snake_case (							self	:  List[str]    )     ->   Optional[int]:
              return (
                  MaskFormerImageProcessor.from_pretrained("facebook/maskformer-swin-small-coco"    )
                  if is_vision_available()
                  else None
              )

       def    __snake_case (							self	:  str    )     ->   Optional[int]:
              __snake_case		:      Dict										=       MaskFormerModel.from_pretrained("facebook/maskformer-swin-small-coco"    ).to(lowercase_    )
              __snake_case		:      List[Any]										=       self.default_image_processor
              __snake_case		:      str										=       prepare_img()
              __snake_case		:      Dict										=       image_processor(lowercase_   ,     return_tensors="pt"    ).to(lowercase_    )
              __snake_case		:      int										=       inputs["pixel_values"].shape
              # check size is divisible by 32
              self.assertTrue((inputs_shape[-1] % 32) == 0 and (inputs_shape[-2] % 32) == 0    )
              # check size
              self.assertEqual(lowercase_   ,     (1, 3, 800, 1088)    )

              with torch.no_grad():
                     __snake_case		:      int										=       model(**lowercase_    )

              __snake_case		:      Any										=       torch.tensor(
                  [[-0.04_82, 0.92_28, 0.49_51], [-0.25_47, 0.80_17, 0.85_27], [-0.00_69, 0.33_85, -0.00_89]]    ).to(lowercase_    )
              self.assertTrue(
                  torch.allclose(
                      outputs.encoder_last_hidden_state[0, 0, :3, :3]   ,     lowercase_   ,     atol=lowercase_    )    )

              __snake_case		:      Optional[Any]										=       torch.tensor(
                  [[-0.84_22, -0.84_34, -0.97_18], [-1.01_44, -0.55_65, -0.41_95], [-1.00_38, -0.44_84, -0.19_61]]    ).to(lowercase_    )
              self.assertTrue(
                  torch.allclose(
                      outputs.pixel_decoder_last_hidden_state[0, 0, :3, :3]   ,     lowercase_   ,     atol=lowercase_    )    )

              __snake_case		:      Dict										=       torch.tensor(
                  [[0.28_52, -0.01_59, 0.97_35], [0.62_54, 0.18_58, 0.85_29], [-0.06_80, -0.41_16, 1.84_13]]    ).to(lowercase_    )
              self.assertTrue(
                  torch.allclose(
                      outputs.transformer_decoder_last_hidden_state[0, :3, :3]   ,     lowercase_   ,     atol=lowercase_    )    )

       def    __snake_case (							self	:  str    )     ->   str:
              __snake_case		:      str										=       (
                  MaskFormerForInstanceSegmentation.from_pretrained("facebook/maskformer-swin-small-coco"    )
                  .to(lowercase_    )
                  .eval()
              )
              __snake_case		:      Union[str, Any]										=       self.default_image_processor
              __snake_case		:      Dict										=       prepare_img()
              __snake_case		:      Optional[Any]										=       image_processor(lowercase_   ,     return_tensors="pt"    ).to(lowercase_    )
              __snake_case		:      List[str]										=       inputs["pixel_values"].shape
              # check size is divisible by 32
              self.assertTrue((inputs_shape[-1] % 32) == 0 and (inputs_shape[-2] % 32) == 0    )
              # check size
              self.assertEqual(lowercase_   ,     (1, 3, 800, 1088)    )

              with torch.no_grad():
                     __snake_case		:      Dict										=       model(**lowercase_    )
              # masks_queries_logits
              __snake_case		:      List[str]										=       outputs.masks_queries_logits
              self.assertEqual(
                  masks_queries_logits.shape   ,     (1, model.config.decoder_config.num_queries, inputs_shape[-2] // 4, inputs_shape[-1] // 4)   ,     )
              __snake_case		:      Union[str, Any]										=       [
                  [-1.3_73_71_24, -1.7_72_49_37, -1.9_36_42_33],
                  [-1.5_97_72_81, -1.9_86_79_39, -2.1_52_36_95],
                  [-1.5_79_53_98, -1.9_26_98_32, -2.09_39_42],
              ]
              __snake_case		:      List[Any]										=       torch.tensor(lowercase_    ).to(lowercase_    )
              self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]   ,     lowercase_   ,     atol=lowercase_    )    )
              # class_queries_logits
              __snake_case		:      str										=       outputs.class_queries_logits
              self.assertEqual(
                  class_queries_logits.shape   ,     (1, model.config.decoder_config.num_queries, model.config.num_labels + 1)    )
              __snake_case		:      List[Any]										=       torch.tensor(
                  [
                      [1.6512E00, -5.2572E00, -3.3519E00],
                      [3.6169E-02, -5.9025E00, -2.9313E00],
                      [1.0766E-04, -7.7630E00, -5.1263E00],
                  ]    ).to(lowercase_    )
              self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]   ,     lowercase_   ,     atol=lowercase_    )    )

       def    __snake_case (							self	:  Dict    )     ->   Union[str, Any]:
              __snake_case		:      Union[str, Any]										=       (
                  MaskFormerForInstanceSegmentation.from_pretrained("facebook/maskformer-resnet101-coco-stuff"    )
                  .to(lowercase_    )
                  .eval()
              )
              __snake_case		:      List[str]										=       self.default_image_processor
              __snake_case		:      Optional[int]										=       prepare_img()
              __snake_case		:      Union[str, Any]										=       image_processor(lowercase_   ,     return_tensors="pt"    ).to(lowercase_    )
              __snake_case		:      Optional[Any]										=       inputs["pixel_values"].shape
              # check size is divisible by 32
              self.assertTrue((inputs_shape[-1] % 32) == 0 and (inputs_shape[-2] % 32) == 0    )
              # check size
              self.assertEqual(lowercase_   ,     (1, 3, 800, 1088)    )

              with torch.no_grad():
                     __snake_case		:      Tuple										=       model(**lowercase_    )
              # masks_queries_logits
              __snake_case		:      List[str]										=       outputs.masks_queries_logits
              self.assertEqual(
                  masks_queries_logits.shape   ,     (1, model.config.decoder_config.num_queries, inputs_shape[-2] // 4, inputs_shape[-1] // 4)   ,     )
              __snake_case		:      Optional[int]										=       [[-0.90_46, -2.63_66, -4.60_62], [-3.41_79, -5.78_90, -8.80_57], [-4.91_79, -7.65_60, -10.77_11]]
              __snake_case		:      Union[str, Any]										=       torch.tensor(lowercase_    ).to(lowercase_    )
              self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]   ,     lowercase_   ,     atol=lowercase_    )    )
              # class_queries_logits
              __snake_case		:      Tuple										=       outputs.class_queries_logits
              self.assertEqual(
                  class_queries_logits.shape   ,     (1, model.config.decoder_config.num_queries, model.config.num_labels + 1)    )
              __snake_case		:      Tuple										=       torch.tensor(
                  [[4.71_88, -3.25_85, -2.88_57], [6.68_71, -2.91_81, -1.24_87], [7.24_49, -2.27_64, -2.18_74]]    ).to(lowercase_    )
              self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]   ,     lowercase_   ,     atol=lowercase_    )    )






       def    __snake_case (							self	:  Optional[Any]    )     ->   Any:
              __snake_case		:      Optional[int]										=       (
                  MaskFormerForInstanceSegmentation.from_pretrained("facebook/maskformer-swin-small-coco"    )
                  .to(lowercase_    )
                  .eval()
              )
              __snake_case		:      str										=       self.default_image_processor

              __snake_case		:      Optional[int]										=       image_processor(
                  [np.zeros((3, 800, 1333)    ), np.zeros((3, 800, 1333)    )]   ,     segmentation_maps=[np.zeros((384, 384)    ).astype(np.floataa    ), np.zeros((384, 384)    ).astype(np.floataa    )]   ,     return_tensors="pt"   ,     )

              __snake_case		:      List[str]										=       inputs["pixel_values"].to(lowercase_    )
              __snake_case		:      Optional[Any]										=       [el.to(lowercase_    ) for el in inputs["mask_labels"]]
              __snake_case		:      Optional[Any]										=       [el.to(lowercase_    ) for el in inputs["class_labels"]]

              with torch.no_grad():
                     __snake_case		:      List[Any]										=       model(**lowercase_    )

              self.assertTrue(outputs.loss is not None    )





from typing import List, Optional, Union

import torch
from transformers import (
    XLMRobertaTokenizer,
)

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .text_encoder import MultilingualCLIP


_snake_case					:		Optional[int]      =    logging.get_logger(__name__)  # pylint: disable=invalid-name

_snake_case					:		Union[str, Any]      =    "\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained(\"kandinsky-community/Kandinsky-2-1-prior\")\n        >>> pipe_prior.to(\"cuda\")\n\n        >>> prompt = \"red cat, 4k photo\"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> negative_image_emb = out.negative_image_embeds\n\n        >>> pipe = KandinskyPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-1\")\n        >>> pipe.to(\"cuda\")\n\n        >>> image = pipe(\n        ...     prompt,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=negative_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ... ).images\n\n        >>> image[0].save(\"cat.png\")\n        ```\n"







def     lowerCAmelCase_					( __lowerCamelCase							,     __lowerCamelCase							,     __lowerCamelCase=8     ):
							__snake_case		:      List[Any]										=       h // scale_factor**2
							if h % scale_factor**2 != 0:
														new_h += 1
							__snake_case		:      Optional[int]										=       w // scale_factor**2
							if w % scale_factor**2 != 0:
														new_w += 1
							return new_h * scale_factor, new_w * scale_factor

class     a		(_lowerCAmelCase				):




							"""simple docstring"""


							def __init__(							self	:  Union[str, Any]   ,     lowerCamelCase	:  MultilingualCLIP   ,     lowerCamelCase	:  XLMRobertaTokenizer   ,     lowerCamelCase	:  UNetaDConditionModel   ,     lowerCamelCase	:  Union[DDIMScheduler, DDPMScheduler]   ,     lowerCamelCase	:  VQModel   ,     )     ->   Optional[int]:
														super().__init__()

														self.register_modules(
														    text_encoder=lowerCamelCase   ,     tokenizer=lowerCamelCase   ,     unet=lowerCamelCase   ,     scheduler=lowerCamelCase   ,     movq=lowerCamelCase   ,     )
														__snake_case		:      List[Any]										=       2 ** (len(self.movq.config.block_out_channels    ) - 1)

							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  List[Any]   ,     lowerCamelCase	:  Optional[int]   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  Optional[Any]   ,     lowerCamelCase	:  int    )     ->   Any:
														if latents is None:
																					__snake_case		:      str										=       randn_tensor(lowerCamelCase   ,     generator=lowerCamelCase   ,     device=lowerCamelCase   ,     dtype=lowerCamelCase    )
														else:
																					if latents.shape != shape:
																												raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {shape}'    )
																					__snake_case		:      Optional[int]										=       latents.to(lowerCamelCase    )

														__snake_case		:      List[Any]										=       latents * scheduler.init_noise_sigma
														return latents

							def    __snake_case (							self	:  Optional[int]   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  List[Any]   ,     lowerCamelCase	:  str=None   ,     )     ->   List[str]:
														__snake_case		:      Tuple										=       len(lowerCamelCase    ) if isinstance(lowerCamelCase   ,     lowerCamelCase    ) else 1
														# get prompt text embeddings
														__snake_case		:      Optional[int]										=       self.tokenizer(
														    lowerCamelCase   ,     padding="max_length"   ,     truncation=lowerCamelCase   ,     max_length=77   ,     return_attention_mask=lowerCamelCase   ,     add_special_tokens=lowerCamelCase   ,     return_tensors="pt"   ,     )

														__snake_case		:      List[str]										=       text_inputs.input_ids
														__snake_case		:      List[Any]										=       self.tokenizer(lowerCamelCase   ,     padding="longest"   ,     return_tensors="pt"    ).input_ids

														if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      Optional[Any]										=       self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]    )
																					logger.warning(
																					    "The following part of your input was truncated because CLIP can only handle sequences up to"
																					    F' {self.tokenizer.model_max_length} tokens: {removed_text}'    )

														__snake_case		:      Any										=       text_input_ids.to(lowerCamelCase    )
														__snake_case		:      List[str]										=       text_inputs.attention_mask.to(lowerCamelCase    )

														__snake_case			,      __snake_case		:      List[str]										=       self.text_encoder(
														    input_ids=lowerCamelCase   ,     attention_mask=lowerCamelCase    )

														__snake_case		:      List[Any]										=       prompt_embeds.repeat_interleave(lowerCamelCase   ,     dim=0    )
														__snake_case		:      List[str]										=       text_encoder_hidden_states.repeat_interleave(lowerCamelCase   ,     dim=0    )
														__snake_case		:      Optional[int]										=       text_mask.repeat_interleave(lowerCamelCase   ,     dim=0    )

														if do_classifier_free_guidance:
																					__snake_case		:      List[str]
																					if negative_prompt is None:
																												__snake_case		:      Any										=       [""] * batch_size
																					elif type(lowerCamelCase    ) is not type(lowerCamelCase    ):
																												raise TypeError(
																												    F'`negative_prompt` should be the same type to `prompt`, but got {type(lowerCamelCase    )} !='
																												    F' {type(lowerCamelCase    )}.'    )
																					elif isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																												__snake_case		:      List[Any]										=       [negative_prompt]
																					elif batch_size != len(lowerCamelCase    ):
																												raise ValueError(
																												    F'`negative_prompt`: {negative_prompt} has batch size {len(lowerCamelCase    )}, but `prompt`:'
																												    F' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
																												    " the batch size of `prompt`."    )
																					else:
																												__snake_case		:      int										=       negative_prompt

																					__snake_case		:      Dict										=       self.tokenizer(
																					    lowerCamelCase   ,     padding="max_length"   ,     max_length=77   ,     truncation=lowerCamelCase   ,     return_attention_mask=lowerCamelCase   ,     add_special_tokens=lowerCamelCase   ,     return_tensors="pt"   ,     )
																					__snake_case		:      Dict										=       uncond_input.input_ids.to(lowerCamelCase    )
																					__snake_case		:      List[Any]										=       uncond_input.attention_mask.to(lowerCamelCase    )

																					__snake_case			,      __snake_case		:      Tuple										=       self.text_encoder(
																					    input_ids=lowerCamelCase   ,     attention_mask=lowerCamelCase    )

																					# duplicate unconditional embeddings for each generation per prompt, using mps friendly method

																					__snake_case		:      Dict										=       negative_prompt_embeds.shape[1]
																					__snake_case		:      int										=       negative_prompt_embeds.repeat(1   ,     lowerCamelCase    )
																					__snake_case		:      List[str]										=       negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,     lowerCamelCase    )

																					__snake_case		:      Union[str, Any]										=       uncond_text_encoder_hidden_states.shape[1]
																					__snake_case		:      Tuple										=       uncond_text_encoder_hidden_states.repeat(1   ,     lowerCamelCase   ,     1    )
																					__snake_case		:      str										=       uncond_text_encoder_hidden_states.view(
																					    batch_size * num_images_per_prompt   ,     lowerCamelCase   ,     -1    )
																					__snake_case		:      Optional[int]										=       uncond_text_mask.repeat_interleave(lowerCamelCase   ,     dim=0    )

																					# done duplicates

																					# For classifier free guidance, we need to do two forward passes.
																					# Here we concatenate the unconditional and text embeddings into a single batch
																					# to avoid doing two forward passes
																					__snake_case		:      Optional[int]										=       torch.cat([negative_prompt_embeds, prompt_embeds]    )
																					__snake_case		:      List[Any]										=       torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]    )

																					__snake_case		:      Any										=       torch.cat([uncond_text_mask, text_mask]    )

														return prompt_embeds, text_encoder_hidden_states, text_mask

							def    __snake_case (							self	:  List[str]   ,     lowerCamelCase	:  Dict=0    )     ->   Tuple:
														if is_accelerate_available():
																					from accelerate import cpu_offload
														else:
																					raise ImportError("Please install accelerate via `pip install accelerate`"    )

														__snake_case		:      Optional[int]										=       torch.device(F'cuda:{gpu_id}'    )

														__snake_case		:      Optional[Any]										=       [
														    self.unet,
														    self.text_encoder,
														    self.movq,
														]
														for cpu_offloaded_model in models:
																					if cpu_offloaded_model is not None:
																												cpu_offload(lowerCamelCase   ,     lowerCamelCase    )

							def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  int=0    )     ->   Optional[int]:
														if is_accelerate_available() and is_accelerate_version(">="   ,     "0.17.0.dev0"    ):
																					from accelerate import cpu_offload_with_hook
														else:
																					raise ImportError("`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher."    )

														__snake_case		:      Optional[Any]										=       torch.device(F'cuda:{gpu_id}'    )

														if self.device.type != "cpu":
																					self.to("cpu"   ,     silence_dtype_warnings=lowerCamelCase    )
																					torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

														__snake_case		:      List[str]										=       None
														for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
																					__snake_case			,      __snake_case		:      List[Any]										=       cpu_offload_with_hook(lowerCamelCase   ,     lowerCamelCase   ,     prev_module_hook=lowerCamelCase    )

														if self.safety_checker is not None:
																					__snake_case			,      __snake_case		:      Optional[int]										=       cpu_offload_with_hook(self.safety_checker   ,     lowerCamelCase   ,     prev_module_hook=lowerCamelCase    )

														# We'll offload the last model manually.
														__snake_case		:      str										=       hook

							@property
							# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
							def    __snake_case (							self	:  List[Any]    )     ->   Optional[int]:
														if not hasattr(self.unet   ,     "_hf_hook"    ):
																					return self.device
														for module in self.unet.modules():
																					if (
																					    hasattr(lowerCamelCase   ,     "_hf_hook"    )
																					    and hasattr(module._hf_hook   ,     "execution_device"    )
																					    and module._hf_hook.execution_device is not None
																					):
																												return torch.device(module._hf_hook.execution_device    )
														return self.device






							@torch.no_grad()
							@replace_example_docstring(lowerCamelCase    )
							def __call__(							self	:  Dict   ,     lowerCamelCase	:  Union[str, List[str]]   ,     lowerCamelCase	:  Union[torch.FloatTensor, List[torch.FloatTensor]]   ,     lowerCamelCase	:  Union[torch.FloatTensor, List[torch.FloatTensor]]   ,     lowerCamelCase	:  Optional[Union[str, List[str]]] = None   ,     lowerCamelCase	:  int = 512   ,     lowerCamelCase	:  int = 512   ,     lowerCamelCase	:  int = 100   ,     lowerCamelCase	:  float = 4.0   ,     lowerCamelCase	:  int = 1   ,     lowerCamelCase	:  Optional[Union[torch.Generator, List[torch.Generator]]] = None   ,     lowerCamelCase	:  Optional[torch.FloatTensor] = None   ,     lowerCamelCase	:  Optional[str] = "pil"   ,     lowerCamelCase	:  bool = True   ,     )     ->   List[Any]:

														if isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      Optional[int]										=       1
														elif isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      List[Any]										=       len(lowerCamelCase    )
														else:
																					raise ValueError(F'`prompt` has to be of type `str` or `list` but is {type(lowerCamelCase    )}'    )

														__snake_case		:      Any										=       self._execution_device

														__snake_case		:      Any										=       batch_size * num_images_per_prompt
														__snake_case		:      Any										=       guidance_scale > 1.0

														__snake_case			,      __snake_case			,      __snake_case		:      Optional[Any]										=       self._encode_prompt(
														    lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase    )

														if isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      List[Any]										=       torch.cat(lowerCamelCase   ,     dim=0    )
														if isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      str										=       torch.cat(lowerCamelCase   ,     dim=0    )

														if do_classifier_free_guidance:
																					__snake_case		:      Dict										=       image_embeds.repeat_interleave(lowerCamelCase   ,     dim=0    )
																					__snake_case		:      Optional[Any]										=       negative_image_embeds.repeat_interleave(lowerCamelCase   ,     dim=0    )

														__snake_case		:      str										=       torch.cat([negative_image_embeds, image_embeds]   ,     dim=0    ).to(
														    dtype=prompt_embeds.dtype   ,     device=lowerCamelCase    )

														self.scheduler.set_timesteps(lowerCamelCase   ,     device=lowerCamelCase    )
														__snake_case		:      Tuple										=       self.scheduler.timesteps

														__snake_case		:      Union[str, Any]										=       self.unet.config.in_channels

														__snake_case			,      __snake_case		:      Tuple										=       get_new_h_w(lowerCamelCase   ,     lowerCamelCase   ,     self.movq_scale_factor    )

														# create initial latent
														__snake_case		:      Any										=       self.prepare_latents(
														    (batch_size, num_channels_latents, height, width)   ,     text_encoder_hidden_states.dtype   ,     lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     self.scheduler   ,     )

														for i, t in enumerate(self.progress_bar(lowerCamelCase    )    ):
																					# expand the latents if we are doing classifier free guidance
																					__snake_case		:      Any										=       torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents

																					__snake_case		:      int										=       {"text_embeds": prompt_embeds, "image_embeds": image_embeds}
																					__snake_case		:      Optional[Any]										=       self.unet(
																					    sample=lowerCamelCase   ,     timestep=lowerCamelCase   ,     encoder_hidden_states=lowerCamelCase   ,     added_cond_kwargs=lowerCamelCase   ,     return_dict=lowerCamelCase   ,     )[0]

																					if do_classifier_free_guidance:
																												__snake_case			,      __snake_case		:      Any										=       noise_pred.split(latents.shape[1]   ,     dim=1    )
																												__snake_case			,      __snake_case		:      Union[str, Any]										=       noise_pred.chunk(2    )
																												__snake_case			,      __snake_case		:      str										=       variance_pred.chunk(2    )
																												__snake_case		:      Optional[Any]										=       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
																												__snake_case		:      Union[str, Any]										=       torch.cat([noise_pred, variance_pred_text]   ,     dim=1    )

																					if not (
																					    hasattr(self.scheduler.config   ,     "variance_type"    )
																					    and self.scheduler.config.variance_type in ["learned", "learned_range"]
																					):
																												__snake_case			,      __snake_case		:      Union[str, Any]										=       noise_pred.split(latents.shape[1]   ,     dim=1    )

																					# compute the previous noisy sample x_t -> x_t-1
																					__snake_case		:      str										=       self.scheduler.step(
																					    lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     generator=lowerCamelCase   ,     ).prev_sample
														# post-processing
														__snake_case		:      str										=       self.movq.decode(lowerCamelCase   ,     force_not_quantize=lowerCamelCase    )["sample"]

														if output_type not in ["pt", "np", "pil"]:
																					raise ValueError(F'Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'    )

														if output_type in ["np", "pil"]:
																					__snake_case		:      Union[str, Any]										=       image * 0.5 + 0.5
																					__snake_case		:      Union[str, Any]										=       image.clamp(0   ,     1    )
																					__snake_case		:      Optional[int]										=       image.cpu().permute(0   ,     2   ,     3   ,     1    ).float().numpy()

														if output_type == "pil":
																					__snake_case		:      str										=       self.numpy_to_pil(lowerCamelCase    )

														if not return_dict:
																					return (image,)

														return ImagePipelineOutput(images=lowerCamelCase    )

