



from typing import List, Optional, Union

import torch

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


a_      :      Tuple    =     logging.get_logger(__name__)  # pylint: disable=invalid-name

a_      :      Optional[Any]    =     '''
    Examples:
        ```py
        >>> import torch
        >>> import numpy as np

        >>> from diffusers import KandinskyV22PriorPipeline, KandinskyV22ControlnetPipeline
        >>> from transformers import pipeline
        >>> from diffusers.utils import load_image


        >>> def make_hint(image, depth_estimator):
        ...     image = depth_estimator(image)["depth"]
        ...     image = np.array(image)
        ...     image = image[:, :, None]
        ...     image = np.concatenate([image, image, image], axis=2)
        ...     detected_map = torch.from_numpy(image).float() / 255.0
        ...     hint = detected_map.permute(2, 0, 1)
        ...     return hint


        >>> depth_estimator = pipeline("depth-estimation")

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16
        ... )
        >>> pipe_prior = pipe_prior.to("cuda")

        >>> pipe = KandinskyV22ControlnetPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-controlnet-depth", torch_dtype=torch.float16
        ... )
        >>> pipe = pipe.to("cuda")


        >>> img = load_image(
        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
        ...     "/kandinsky/cat.png"
        ... ).resize((768, 768))

        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to("cuda")

        >>> prompt = "A robot, 4k photo"
        >>> negative_prior_prompt = "lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature"

        >>> generator = torch.Generator(device="cuda").manual_seed(43)

        >>> image_emb, zero_image_emb = pipe_prior(
        ...     prompt=prompt, negative_prompt=negative_prior_prompt, generator=generator
        ... ).to_tuple()

        >>> images = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     hint=hint,
        ...     num_inference_steps=50,
        ...     generator=generator,
        ...     height=768,
        ...     width=768,
        ... ).images

        >>> images[0].save("robot_cat.png")
        ```
'''






def     __a					(		__UpperCAmelCase							,  __UpperCAmelCase							,  __UpperCAmelCase=8       ):
 a__												= height // scale_factor**2
 if height % scale_factor**2 != 0:
  new_height += 1
 a__												= width // scale_factor**2
 if width % scale_factor**2 != 0:
  new_width += 1
 return new_height * scale_factor, new_width * scale_factor

class    __UpperCamelCase			(       _a				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				) ->			Tuple:
  super().__init__()

  self.register_modules(
      unet=__UpperCamelCase		,				scheduler=__UpperCamelCase		,				movq=__UpperCamelCase		,				)
  a__												= 2 ** (len(self.movq.config.block_out_channels      ) - 1)

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			Optional[Any]:
  if latents is None:
   a__												= randn_tensor(__UpperCamelCase		,				generator=__UpperCamelCase		,				device=__UpperCamelCase		,				dtype=__UpperCamelCase      )
  else:
   if latents.shape != shape:
    raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {shape}"      )
   a__												= latents.to(__UpperCamelCase      )

  a__												= latents * scheduler.init_noise_sigma
  return latents

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE=0      ) ->			int:
  if is_accelerate_available():
   from accelerate import cpu_offload
  else:
   raise ImportError('''Please install accelerate via `pip install accelerate`'''      )

  a__												= torch.device(f"cuda:{gpu_id}"      )

  a__												= [
      self.unet,
      self.movq,
  ]
  for cpu_offloaded_model in models:
   if cpu_offloaded_model is not None:
    cpu_offload(__UpperCamelCase		,				__UpperCamelCase      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE=0      ) ->			Union[str, Any]:
  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.'''      )

  a__												= torch.device(f"cuda:{gpu_id}"      )

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

  a__												= None
  for cpu_offloaded_model in [self.unet, self.movq]:
   a__							,			a__												= cpu_offload_with_hook(__UpperCamelCase		,				__UpperCamelCase		,				prev_module_hook=__UpperCamelCase      )

  # We'll offload the last model manually.
  a__												= hook

 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def 						_UpperCAmelCase       (	self      ) ->			str:
  if not hasattr(self.unet		,				'''_hf_hook'''      ):
   return self.device
  for module in self.unet.modules():
   if (
       hasattr(__UpperCamelCase		,				'''_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(__UpperCamelCase      )
 def __call__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = 5_1_2		,				SCREAMING_SNAKE_CASE = 5_1_2		,				SCREAMING_SNAKE_CASE = 1_0_0		,				SCREAMING_SNAKE_CASE = 4.0		,				SCREAMING_SNAKE_CASE = 1		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = "pil"		,				SCREAMING_SNAKE_CASE = True		,				) ->			str:
  a__												= self._execution_device

  a__												= guidance_scale > 1.0

  if isinstance(__UpperCamelCase		,				__UpperCamelCase      ):
   a__												= torch.cat(__UpperCamelCase		,				dim=0      )
  if isinstance(__UpperCamelCase		,				__UpperCamelCase      ):
   a__												= torch.cat(__UpperCamelCase		,				dim=0      )
  if isinstance(__UpperCamelCase		,				__UpperCamelCase      ):
   a__												= torch.cat(__UpperCamelCase		,				dim=0      )

  a__												= image_embeds.shape[0] * num_images_per_prompt

  if do_classifier_free_guidance:
   a__												= image_embeds.repeat_interleave(__UpperCamelCase		,				dim=0      )
   a__												= negative_image_embeds.repeat_interleave(__UpperCamelCase		,				dim=0      )
   a__												= hint.repeat_interleave(__UpperCamelCase		,				dim=0      )

  a__												= torch.cat([negative_image_embeds, image_embeds]		,				dim=0      ).to(dtype=self.unet.dtype		,				device=__UpperCamelCase      )
  a__												= torch.cat([hint, hint]		,				dim=0      ).to(dtype=self.unet.dtype		,				device=__UpperCamelCase      )
  self.scheduler.set_timesteps(__UpperCamelCase		,				device=__UpperCamelCase      )
  a__												= self.scheduler.timesteps

  a__												= self.movq.config.latent_channels

  a__							,			a__												= downscale_height_and_width(__UpperCamelCase		,				__UpperCamelCase		,				self.movq_scale_factor      )

  # create initial latent
  a__												= self.prepare_latents(
      (batch_size, num_channels_latents, height, width)		,				image_embeds.dtype		,				__UpperCamelCase		,				__UpperCamelCase		,				__UpperCamelCase		,				self.scheduler		,				)

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

   a__												= {'''image_embeds''': image_embeds, '''hint''': hint}
   a__												= self.unet(
       sample=__UpperCamelCase		,				timestep=__UpperCamelCase		,				encoder_hidden_states=__UpperCamelCase		,				added_cond_kwargs=__UpperCamelCase		,				return_dict=__UpperCamelCase		,				)[0]

   if do_classifier_free_guidance:
    a__							,			a__												= noise_pred.split(latents.shape[1]		,				dim=1      )
    a__							,			a__												= noise_pred.chunk(2      )
    a__							,			a__												= variance_pred.chunk(2      )
    a__												= noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
    a__												= 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"]
   ):
    a__							,			a__												= noise_pred.split(latents.shape[1]		,				dim=1      )

   # compute the previous noisy sample x_t -> x_t-1
   a__												= self.scheduler.step(
       __UpperCamelCase		,				__UpperCamelCase		,				__UpperCamelCase		,				generator=__UpperCamelCase		,				)[0]
  # post-processing
  a__												= self.movq.decode(__UpperCamelCase		,				force_not_quantize=__UpperCamelCase      )['''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"]:
   a__												= image * 0.5 + 0.5
   a__												= image.clamp(0		,				1      )
   a__												= image.cpu().permute(0		,				2		,				3		,				1      ).float().numpy()

  if output_type == "pil":
   a__												= self.numpy_to_pil(__UpperCamelCase      )

  if not return_dict:
   return (image,)

  return ImagePipelineOutput(images=__UpperCamelCase      )







import inspect
import unittest

from huggingface_hub import hf_hub_download

from transformers import ConvNextConfig, UperNetConfig
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
from transformers.utils import is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import UperNetForSemanticSegmentation
		from transformers.models.upernet.modeling_upernet import UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
		from PIL import Image

		from transformers import AutoImageProcessor



class    a_ :






			def __init__(      self				,			__UpperCamelCase				,			__UpperCamelCase=13				,			__UpperCamelCase=32				,			__UpperCamelCase=3				,			__UpperCamelCase=4				,			__UpperCamelCase=[10, 20, 30, 40]				,			__UpperCamelCase=[2, 2, 3, 2]				,			__UpperCamelCase=True				,			__UpperCamelCase=True				,			__UpperCamelCase=37				,			__UpperCamelCase="gelu"				,			__UpperCamelCase=10				,			__UpperCamelCase=0.02				,			__UpperCamelCase=["stage2", "stage3", "stage4"]				,			__UpperCamelCase=3				,			__UpperCamelCase=None				,			):
									_lowercase					       =					parent
									_lowercase					       =					batch_size
									_lowercase					       =					image_size
									_lowercase					       =					num_channels
									_lowercase					       =					num_stages
									_lowercase					       =					hidden_sizes
									_lowercase					       =					depths
									_lowercase					       =					is_training
									_lowercase					       =					use_labels
									_lowercase					       =					intermediate_size
									_lowercase					       =					hidden_act
									_lowercase					       =					type_sequence_label_size
									_lowercase					       =					initializer_range
									_lowercase					       =					out_features
									_lowercase					       =					num_labels
									_lowercase					       =					scope
									_lowercase					       =					num_stages






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]  )

									_lowercase					       =					None
									if self.use_labels:
															_lowercase					       =					ids_tensor([self.batch_size]				,			self.type_sequence_label_size  )

									_lowercase					       =					self.get_config()

									return config, pixel_values, labels






			def    UpperCamelCase_				(      self  ):
									return ConvNextConfig(
									    num_channels=self.num_channels				,			num_stages=self.num_stages				,			hidden_sizes=self.hidden_sizes				,			depths=self.depths				,			is_training=self.is_training				,			intermediate_size=self.intermediate_size				,			hidden_act=self.hidden_act				,			out_features=self.out_features				,			)






			def    UpperCamelCase_				(      self  ):
									return UperNetConfig(
									    backbone_config=self.get_backbone_config()				,			hidden_size=512				,			pool_scales=[1, 2, 3, 6]				,			use_auxiliary_head=__UpperCamelCase				,			auxiliary_loss_weight=0.4				,			auxiliary_in_channels=40				,			auxiliary_channels=256				,			auxiliary_num_convs=1				,			auxiliary_concat_input=__UpperCamelCase				,			loss_ignore_index=255				,			num_labels=self.num_labels				,			)






			def    UpperCamelCase_				(      self				,			__UpperCamelCase				,			__UpperCamelCase				,			__UpperCamelCase  ):
									_lowercase					       =					UperNetForSemanticSegmentation(config=__UpperCamelCase  )
									model.to(__UpperCamelCase  )
									model.eval()
									_lowercase					       =					model(__UpperCamelCase  )
									self.parent.assertEqual(
									    result.logits.shape				,			(self.batch_size, self.num_labels, self.image_size, self.image_size)  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.prepare_config_and_inputs()
									(
									    (
									    _lowercase
									)       ,						(
									    _lowercase
									)       ,						(
									    _lowercase
									)       ,						
									)					       =					config_and_inputs
									_lowercase					       =					{"""pixel_values""": pixel_values}
									return config, inputs_dict







@require_torch
class    a_ (			_a					,			_a					,			unittest.TestCase    ):
			a	:			Optional[int]					   =							(UperNetForSemanticSegmentation,) if is_torch_available() else ()
			a	:			Optional[Any]					   =							{'''image-segmentation''': UperNetForSemanticSegmentation} if is_torch_available() else {}
			a	:			Any					   =							False
			a	:			Union[str, Any]					   =							False
			a	:			Optional[Any]					   =							False
			a	:			int					   =							False
			a	:			Optional[int]					   =							False
			a	:			List[str]					   =							False






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					UperNetModelTester(self  )
									_lowercase					       =					ConfigTester(self				,			config_class=__UpperCamelCase				,			has_text_modality=__UpperCamelCase				,			hidden_size=37  )






			def    UpperCamelCase_				(      self  ):
									self.create_and_test_config_common_properties()
									self.config_tester.create_and_test_config_to_json_string()
									self.config_tester.create_and_test_config_to_json_file()
									self.config_tester.create_and_test_config_from_and_save_pretrained()
									self.config_tester.create_and_test_config_with_num_labels()
									self.config_tester.check_config_can_be_init_without_params()
									self.config_tester.check_config_arguments_init()






			def    UpperCamelCase_				(      self  ):
									return






			def    UpperCamelCase_				(      self  ):
									_lowercase       ,						_lowercase					       =					self.model_tester.prepare_config_and_inputs_for_common()

									for model_class in self.all_model_classes:
															_lowercase					       =					model_class(__UpperCamelCase  )
															_lowercase					       =					inspect.signature(model.forward  )
															# signature.parameters is an OrderedDict => so arg_names order is deterministic
															_lowercase					       =					[*signature.parameters.keys()]

															_lowercase					       =					["""pixel_values"""]
															self.assertListEqual(arg_names[:1]				,			__UpperCamelCase  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_semantic_segmentation(*__UpperCamelCase  )






			@unittest.skip(reason="""UperNet does not use inputs_embeds"""  )
			def    UpperCamelCase_				(      self  ):
									pass






			@unittest.skip(reason="""UperNet does not support input and output embeddings"""  )
			def    UpperCamelCase_				(      self  ):
									pass






			@unittest.skip(reason="""UperNet does not have a base model"""  )
			def    UpperCamelCase_				(      self  ):
									pass






			@unittest.skip(reason="""UperNet does not have a base model"""  )
			def    UpperCamelCase_				(      self  ):
									pass






			@require_torch_multi_gpu
			@unittest.skip(reason="""UperNet has some layers using `add_module` which doesn't work well with `nn.DataParallel`"""  )
			def    UpperCamelCase_				(      self  ):
									pass






			@unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""  )
			def    UpperCamelCase_				(      self  ):
									pass






			def    UpperCamelCase_				(      self  ):
									def check_hidden_states_output(__UpperCamelCase				,			__UpperCamelCase				,			__UpperCamelCase  ):
															_lowercase					       =					model_class(__UpperCamelCase  )
															model.to(__UpperCamelCase  )
															model.eval()

															with torch.no_grad():
																					_lowercase					       =					model(**self._prepare_for_class(__UpperCamelCase				,			__UpperCamelCase  )  )

															_lowercase					       =					outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

															_lowercase					       =					self.model_tester.num_stages
															self.assertEqual(len(__UpperCamelCase  )				,			expected_num_stages + 1  )

															# ConvNext's feature maps are of shape (batch_size, num_channels, height, width)
															self.assertListEqual(
															    list(hidden_states[0].shape[-2:]  )				,			[self.model_tester.image_size // 4, self.model_tester.image_size // 4]				,			)

									_lowercase       ,						_lowercase					       =					self.model_tester.prepare_config_and_inputs_for_common()

									for model_class in self.all_model_classes:
															_lowercase					       =					True
															check_hidden_states_output(__UpperCamelCase				,			__UpperCamelCase				,			__UpperCamelCase  )

															# check that output_hidden_states also work using config
															del inputs_dict["output_hidden_states"]
															_lowercase					       =					True

															check_hidden_states_output(__UpperCamelCase				,			__UpperCamelCase				,			__UpperCamelCase  )






			def    UpperCamelCase_				(      self  ):
									_lowercase       ,						_lowercase					       =					self.model_tester.prepare_config_and_inputs_for_common()

									_lowercase					       =					_config_zero_init(__UpperCamelCase  )
									_lowercase					       =					_config_zero_init(configs_no_init.backbone_config  )
									for model_class in self.all_model_classes:
															_lowercase					       =					model_class(config=__UpperCamelCase  )
															for name, param in model.named_parameters():
																					if param.requires_grad:
																											self.assertIn(
																											    ((param.data.mean() * 1E9).round() / 1E9).item()				,			[0.0, 1.0]				,			msg=f"""Parameter {name} of model {model_class} seems not properly initialized"""				,			)






			@unittest.skip(reason="""UperNet does not have tied weights"""  )
			def    UpperCamelCase_				(      self  ):
									pass






			@slow
			def    UpperCamelCase_				(      self  ):
									for model_name in UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
															_lowercase					       =					UperNetForSemanticSegmentation.from_pretrained(__UpperCamelCase  )
															self.assertIsNotNone(__UpperCamelCase  )







def 	UpperCamelCase__  (     )  ->     str:
						_lowercase					       =					hf_hub_download(
						    repo_id="""hf-internal-testing/fixtures_ade20k"""		,   repo_type="""dataset"""		,   filename="""ADE_val_00000001.jpg"""       )
						_lowercase					       =					Image.open(SCREAMING_SNAKE_CASE_       ).convert("""RGB"""       )
						return image



@require_torch
@require_vision
@slow
class    a_ (			unittest.TestCase    ):






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					AutoImageProcessor.from_pretrained("""openmmlab/upernet-swin-tiny"""  )
									_lowercase					       =					UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-swin-tiny"""  ).to(__UpperCamelCase  )

									_lowercase					       =					prepare_img()
									_lowercase					       =					processor(images=__UpperCamelCase				,			return_tensors="""pt"""  ).to(__UpperCamelCase  )

									with torch.no_grad():
															_lowercase					       =					model(**__UpperCamelCase  )

									_lowercase					       =					torch.Size((1, model.config.num_labels, 512, 512)  )
									self.assertEqual(outputs.logits.shape				,			__UpperCamelCase  )

									_lowercase					       =					torch.tensor(
									    [[-7.59_58, -7.59_58, -7.43_02], [-7.59_58, -7.59_58, -7.43_02], [-7.47_97, -7.47_97, -7.30_68]]  ).to(__UpperCamelCase  )
									self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]				,			__UpperCamelCase				,			atol=1E-4  )  )






			def    UpperCamelCase_				(      self  ):
									_lowercase					       =					AutoImageProcessor.from_pretrained("""openmmlab/upernet-convnext-tiny"""  )
									_lowercase					       =					UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-convnext-tiny"""  ).to(__UpperCamelCase  )

									_lowercase					       =					prepare_img()
									_lowercase					       =					processor(images=__UpperCamelCase				,			return_tensors="""pt"""  ).to(__UpperCamelCase  )

									with torch.no_grad():
															_lowercase					       =					model(**__UpperCamelCase  )

									_lowercase					       =					torch.Size((1, model.config.num_labels, 512, 512)  )
									self.assertEqual(outputs.logits.shape				,			__UpperCamelCase  )

									_lowercase					       =					torch.tensor(
									    [[-8.81_10, -8.81_10, -8.65_21], [-8.81_10, -8.81_10, -8.65_21], [-8.77_46, -8.77_46, -8.61_30]]  ).to(__UpperCamelCase  )
									self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]				,			__UpperCamelCase				,			atol=1E-4  )  )