






"""simple docstring"""







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      UpperCamelCase		:






						def __init__(		self  :List[Any]			, __magic_name__  :Dict			, __magic_name__  :Optional[int]=13			, __magic_name__  :Optional[int]=32			, __magic_name__  :Optional[int]=3			, __magic_name__  :Any=4			, __magic_name__  :Dict=[10, 20, 30, 40]			, __magic_name__  :Tuple=[2, 2, 3, 2]			, __magic_name__  :str=True			, __magic_name__  :Optional[int]=True			, __magic_name__  :Tuple=37			, __magic_name__  :Dict="gelu"			, __magic_name__  :str=10			, __magic_name__  :List[Any]=0.02			, __magic_name__  :Optional[int]=["stage2", "stage3", "stage4"]			, __magic_name__  :Optional[Any]=3			, __magic_name__  :Dict=None			, )      ->Dict:
										lowercase    :      Dict       = parent
										lowercase    :      List[Any]       = batch_size
										lowercase    :      Tuple       = image_size
										lowercase    :      Tuple       = num_channels
										lowercase    :      Tuple       = num_stages
										lowercase    :      List[str]       = hidden_sizes
										lowercase    :      Any       = depths
										lowercase    :      Dict       = is_training
										lowercase    :      List[str]       = use_labels
										lowercase    :      Optional[int]       = intermediate_size
										lowercase    :      Optional[int]       = hidden_act
										lowercase    :      Optional[int]       = type_sequence_label_size
										lowercase    :      int       = initializer_range
										lowercase    :      List[str]       = out_features
										lowercase    :      str       = num_labels
										lowercase    :      Optional[Any]       = scope
										lowercase    :      Optional[Any]       = num_stages






						def      __snake_case      (		self  :List[str]				)      ->int:
										lowercase    :      Union[str, Any]       = floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]				)

										lowercase    :      Union[str, Any]       = None
										if self.use_labels:
														lowercase    :      int       = ids_tensor([self.batch_size]			, self.type_sequence_label_size				)

										lowercase    :      Optional[Any]       = self.get_config()

										return config, pixel_values, labels






						def      __snake_case      (		self  :Optional[int]				)      ->int:
										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      __snake_case      (		self  :List[Any]				)      ->List[Any]:
										return UperNetConfig(
										    backbone_config=self.get_backbone_config()			, hidden_size=512			, pool_scales=[1, 2, 3, 6]			, use_auxiliary_head=__magic_name__			, auxiliary_loss_weight=0.4			, auxiliary_in_channels=40			, auxiliary_channels=256			, auxiliary_num_convs=1			, auxiliary_concat_input=__magic_name__			, loss_ignore_index=255			, num_labels=self.num_labels			, )






						def      __snake_case      (		self  :List[Any]			, __magic_name__  :int			, __magic_name__  :List[str]			, __magic_name__  :int				)      ->Optional[Any]:
										lowercase    :      int       = UperNetForSemanticSegmentation(config=__magic_name__				)
										model.to(__magic_name__				)
										model.eval()
										lowercase    :      List[Any]       = model(__magic_name__				)
										self.parent.assertEqual(
										    result.logits.shape			, (self.batch_size, self.num_labels, self.image_size, self.image_size)				)






						def      __snake_case      (		self  :Any				)      ->Dict:
										lowercase    :      int       = self.prepare_config_and_inputs()
										(
										    (
										    lowercase
										)      ,      (
										    lowercase
										)      ,      (
										    lowercase
										)      ,      
										)    :      Any       = config_and_inputs
										lowercase    :      Any       = {"""pixel_values""": pixel_values}
										return config, inputs_dict


@require_torch
class      UpperCamelCase		(__snake_case						,						__snake_case						,						unittest.TestCase     ):
						_SCREAMING_SNAKE_CASE  : Optional[Any]												=		(UperNetForSemanticSegmentation,) if is_torch_available() else ()
						_SCREAMING_SNAKE_CASE  : Optional[Any]												=		{"""image-segmentation""": UperNetForSemanticSegmentation} if is_torch_available() else {}
						_SCREAMING_SNAKE_CASE  : Tuple												=		False
						_SCREAMING_SNAKE_CASE  : Optional[Any]												=		False
						_SCREAMING_SNAKE_CASE  : Optional[Any]												=		False
						_SCREAMING_SNAKE_CASE  : Optional[Any]												=		False
						_SCREAMING_SNAKE_CASE  : Union[str, Any]												=		False
						_SCREAMING_SNAKE_CASE  : int												=		False






						def      __snake_case      (		self  :Dict				)      ->Optional[int]:
										lowercase    :      Union[str, Any]       = UperNetModelTester(self				)
										lowercase    :      Union[str, Any]       = ConfigTester(self			, config_class=__magic_name__			, has_text_modality=__magic_name__			, hidden_size=37				)






						def      __snake_case      (		self  :Dict				)      ->Union[str, Any]:
										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      __snake_case      (		self  :Union[str, Any]				)      ->Optional[int]:
										return






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

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

														lowercase    :      Any       = ["""pixel_values"""]
														self.assertListEqual(arg_names[:1]			, __magic_name__				)






						def      __snake_case      (		self  :List[Any]				)      ->Union[str, Any]:
										lowercase    :      Optional[Any]       = self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_semantic_segmentation(*__magic_name__				)






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






						@unittest.skip(reason="""UperNet does not support input and output embeddings"""				)
						def      __snake_case      (		self  :Optional[Any]				)      ->Any:
										pass






						@unittest.skip(reason="""UperNet does not have a base model"""				)
						def      __snake_case      (		self  :Union[str, Any]				)      ->Optional[int]:
										pass






						@unittest.skip(reason="""UperNet does not have a base model"""				)
						def      __snake_case      (		self  :Any				)      ->Tuple:
										pass






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






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






						def      __snake_case      (		self  :Dict				)      ->List[str]:
										def check_hidden_states_output(__magic_name__  :List[Any]			, __magic_name__  :str			, __magic_name__  :Any				):
														lowercase    :      Any       = model_class(__magic_name__				)
														model.to(__magic_name__				)
														model.eval()

														with torch.no_grad():
																		lowercase    :      int       = model(**self._prepare_for_class(__magic_name__			, __magic_name__				)				)

														lowercase    :      List[Any]       = outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

														lowercase    :      List[Any]       = self.model_tester.num_stages
														self.assertEqual(len(__magic_name__				)			, 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    :      List[str]       = self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
														lowercase    :      List[Any]       = True
														check_hidden_states_output(__magic_name__			, __magic_name__			, __magic_name__				)

														# check that output_hidden_states also work using config
														del inputs_dict["output_hidden_states"]
														lowercase    :      Any       = True

														check_hidden_states_output(__magic_name__			, __magic_name__			, __magic_name__				)






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

										lowercase    :      int       = _config_zero_init(__magic_name__				)
										lowercase    :      int       = _config_zero_init(configs_no_init.backbone_config				)
										for model_class in self.all_model_classes:
														lowercase    :      str       = model_class(config=__magic_name__				)
														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      __snake_case      (		self  :List[str]				)      ->List[Any]:
										pass






						@slow
						def      __snake_case      (		self  :Union[str, Any]				)      ->Tuple:
										for model_name in UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														lowercase    :      List[Any]       = UperNetForSemanticSegmentation.from_pretrained(__magic_name__				)
														self.assertIsNotNone(__magic_name__				)


def      UpperCamelCase       ( )     ->       Tuple:
				lowercase    :      Union[str, Any]       = hf_hub_download(
				    repo_id="""hf-internal-testing/fixtures_ade20k"""  ,				repo_type="""dataset"""  ,				filename="""ADE_val_00000001.jpg""" )
				lowercase    :      Dict       = Image.open(_A ).convert("""RGB""" )
				return image


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






						def      __snake_case      (		self  :Optional[Any]				)      ->str:
										lowercase    :      List[str]       = AutoImageProcessor.from_pretrained("""openmmlab/upernet-swin-tiny"""				)
										lowercase    :      int       = UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-swin-tiny"""				).to(__magic_name__				)

										lowercase    :      str       = prepare_img()
										lowercase    :      List[Any]       = processor(images=__magic_name__			, return_tensors="""pt"""				).to(__magic_name__				)

										with torch.no_grad():
														lowercase    :      Optional[Any]       = model(**__magic_name__				)

										lowercase    :      Union[str, Any]       = torch.Size((1, model.config.num_labels, 512, 512)				)
										self.assertEqual(outputs.logits.shape			, __magic_name__				)

										lowercase    :      Union[str, Any]       = torch.tensor(
										    [[-7.5958, -7.5958, -7.4302], [-7.5958, -7.5958, -7.4302], [-7.4797, -7.4797, -7.3068]]				).to(__magic_name__				)
										self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]			, __magic_name__			, atol=1E-4				)				)






						def      __snake_case      (		self  :str				)      ->Union[str, Any]:
										lowercase    :      Union[str, Any]       = AutoImageProcessor.from_pretrained("""openmmlab/upernet-convnext-tiny"""				)
										lowercase    :      List[str]       = UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-convnext-tiny"""				).to(__magic_name__				)

										lowercase    :      Dict       = prepare_img()
										lowercase    :      List[Any]       = processor(images=__magic_name__			, return_tensors="""pt"""				).to(__magic_name__				)

										with torch.no_grad():
														lowercase    :      Dict       = model(**__magic_name__				)

										lowercase    :      str       = torch.Size((1, model.config.num_labels, 512, 512)				)
										self.assertEqual(outputs.logits.shape			, __magic_name__				)

										lowercase    :      Tuple       = torch.tensor(
										    [[-8.8110, -8.8110, -8.6521], [-8.8110, -8.8110, -8.6521], [-8.7746, -8.7746, -8.6130]]				).to(__magic_name__				)
										self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]			, __magic_name__			, atol=1E-4				)				)








"""simple docstring"""







from typing import List, Optional, Tuple, Union

import torch

from ...schedulers import DDIMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


class      UpperCamelCase		(__snake_case     ):






						def __init__(		self  :Any			, __magic_name__  :Union[str, Any]			, __magic_name__  :Optional[Any]				)      ->Dict:
										super().__init__()

										# make sure scheduler can always be converted to DDIM
										lowercase    :      Optional[Any]       = DDIMScheduler.from_config(scheduler.config				)

										self.register_modules(unet=__magic_name__			, scheduler=__magic_name__				)






						@torch.no_grad()
						def __call__(		self  :Optional[int]			, __magic_name__  :int = 1			, __magic_name__  :Optional[Union[torch.Generator, List[torch.Generator]]] = None			, __magic_name__  :float = 0.0			, __magic_name__  :int = 50			, __magic_name__  :Optional[bool] = None			, __magic_name__  :Optional[str] = "pil"			, __magic_name__  :bool = True			, )      ->Union[ImagePipelineOutput, Tuple]:

										# Sample gaussian noise to begin loop
										if isinstance(self.unet.config.sample_size			, __magic_name__				):
														lowercase    :      int       = (
														    batch_size,
														    self.unet.config.in_channels,
														    self.unet.config.sample_size,
														    self.unet.config.sample_size,
														)
										else:
														lowercase    :      Tuple       = (batch_size, self.unet.config.in_channels, *self.unet.config.sample_size)

										if isinstance(__magic_name__			, __magic_name__				) and len(__magic_name__				) != batch_size:
														raise ValueError(
														    f"""You have passed a list of generators of length {len(__magic_name__				)}, but requested an effective batch"""
														    f""" size of {batch_size}. Make sure the batch size matches the length of the generators."""				)

										lowercase    :      Dict       = randn_tensor(__magic_name__			, generator=__magic_name__			, device=self.device			, dtype=self.unet.dtype				)

										# set step values
										self.scheduler.set_timesteps(__magic_name__				)

										for t in self.progress_bar(self.scheduler.timesteps				):
														# 1. predict noise model_output
														lowercase    :      Union[str, Any]       = self.unet(__magic_name__			, __magic_name__				).sample

														# 2. predict previous mean of image x_t-1 and add variance depending on eta
														# eta corresponds to η in paper and should be between [0, 1]
														# do x_t -> x_t-1
														lowercase    :      int       = self.scheduler.step(
														    __magic_name__			, __magic_name__			, __magic_name__			, eta=__magic_name__			, use_clipped_model_output=__magic_name__			, generator=__magic_name__				).prev_sample

										lowercase    :      Optional[Any]       = (image / 2 + 0.5).clamp(0			, 1				)
										lowercase    :      Tuple       = image.cpu().permute(0			, 2			, 3			, 1				).numpy()
										if output_type == "pil":
														lowercase    :      Any       = self.numpy_to_pil(__magic_name__				)

										if not return_dict:
														return (image,)

										return ImagePipelineOutput(images=__magic_name__				)
