




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  :







						'''simple docstring'''





						def __init__(		self				:	List[str]			,      __lowercase				:	Optional[Any]			,      __lowercase				:	Tuple=13			,      __lowercase				:	Any=32			,      __lowercase				:	List[Any]=3			,      __lowercase				:	int=4			,      __lowercase				:	Any=[10, 20, 30, 40]			,      __lowercase				:	List[str]=[2, 2, 3, 2]			,      __lowercase				:	Union[str, Any]=True			,      __lowercase				:	str=True			,      __lowercase				:	List[str]=37			,      __lowercase				:	Dict="gelu"			,      __lowercase				:	int=10			,      __lowercase				:	Dict=0.02			,      __lowercase				:	List[Any]=["stage2", "stage3", "stage4"]			,      __lowercase				:	Optional[int]=3			,      __lowercase				:	Optional[int]=None			,      ):






							"""simple docstring"""






							snake_case_						       =	parent
							snake_case_						       =	batch_size
							snake_case_						       =	image_size
							snake_case_						       =	num_channels
							snake_case_						       =	num_stages
							snake_case_						       =	hidden_sizes
							snake_case_						       =	depths
							snake_case_						       =	is_training
							snake_case_						       =	use_labels
							snake_case_						       =	intermediate_size
							snake_case_						       =	hidden_act
							snake_case_						       =	type_sequence_label_size
							snake_case_						       =	initializer_range
							snake_case_						       =	out_features
							snake_case_						       =	num_labels
							snake_case_						       =	scope
							snake_case_						       =	num_stages






						def 			snake_case__		(		self				:	Dict		):






							"""simple docstring"""






							snake_case_						       =	floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]		)

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

							snake_case_						       =	self.get_config()

							return config, pixel_values, labels






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






							"""simple docstring"""






							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				:	Union[str, Any]		):






							"""simple docstring"""






							return UperNetConfig(
							    backbone_config=self.get_backbone_config()			,      hidden_size=5_12			,      pool_scales=[1, 2, 3, 6]			,      use_auxiliary_head=__lowercase			,      auxiliary_loss_weight=0.4			,      auxiliary_in_channels=40			,      auxiliary_channels=2_56			,      auxiliary_num_convs=1			,      auxiliary_concat_input=__lowercase			,      loss_ignore_index=2_55			,      num_labels=self.num_labels			,      )






						def 			snake_case__		(		self				:	Optional[int]			,      __lowercase				:	Tuple			,      __lowercase				:	Union[str, Any]			,      __lowercase				:	int		):






							"""simple docstring"""






							snake_case_						       =	UperNetForSemanticSegmentation(config=__lowercase		)
							model.to(__lowercase		)
							model.eval()
							snake_case_						       =	model(__lowercase		)
							self.parent.assertEqual(
							    result.logits.shape			,      (self.batch_size, self.num_labels, self.image_size, self.image_size)		)






						def 			snake_case__		(		self				:	Optional[Any]		):






							"""simple docstring"""






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







@require_torch
class 						UpperCAmelCase  (							UpperCAmelCase__			,  UpperCAmelCase__			,  unittest.TestCase     ):







						'''simple docstring'''





						lowerCAmelCase_							=  (UperNetForSemanticSegmentation,) if is_torch_available() else ()
						lowerCAmelCase_							=  {'''image-segmentation''': UperNetForSemanticSegmentation} if is_torch_available() else {}
						lowerCAmelCase_							=  False
						lowerCAmelCase_							=  False
						lowerCAmelCase_							=  False
						lowerCAmelCase_							=  False
						lowerCAmelCase_							=  False
						lowerCAmelCase_							=  False






						def 			snake_case__		(		self				:	int		):






							"""simple docstring"""






							snake_case_						       =	UperNetModelTester(self		)
							snake_case_						       =	ConfigTester(self			,      config_class=__lowercase			,      has_text_modality=__lowercase			,      hidden_size=37		)






						def 			snake_case__		(		self				:	Optional[Any]		):






							"""simple docstring"""






							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				:	List[str]		):






							"""simple docstring"""






							return






						def 			snake_case__		(		self				:	Any		):






							"""simple docstring"""






							snake_case_       , snake_case_						       =	self.model_tester.prepare_config_and_inputs_for_common()

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

								snake_case_						       =	["pixel_values"]
								self.assertListEqual(arg_names[:1]			,      __lowercase		)






						def 			snake_case__		(		self				:	Optional[Any]		):






							"""simple docstring"""






							snake_case_						       =	self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_semantic_segmentation(*__lowercase		)






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






							"""simple docstring"""






							pass






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






							"""simple docstring"""






							pass






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






							"""simple docstring"""






							pass






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






							"""simple docstring"""






							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				:	Tuple		):






							"""simple docstring"""






							pass






						@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."		)
						def 			snake_case__		(		self				:	Union[str, Any]		):






							"""simple docstring"""






							pass






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






							"""simple docstring"""






							def check_hidden_states_output(__lowercase				:	int			,      __lowercase				:	Dict			,      __lowercase				:	Dict		):
								snake_case_						       =	model_class(__lowercase		)
								model.to(__lowercase		)
								model.eval()

								with torch.no_grad():
									snake_case_						       =	model(**self._prepare_for_class(__lowercase			,      __lowercase		)		)

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

								snake_case_						       =	self.model_tester.num_stages
								self.assertEqual(len(__lowercase		)			,      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]			,      )

							snake_case_       , snake_case_						       =	self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
								snake_case_						       =	True
								check_hidden_states_output(__lowercase			,      __lowercase			,      __lowercase		)

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

								check_hidden_states_output(__lowercase			,      __lowercase			,      __lowercase		)






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






							"""simple docstring"""






							snake_case_       , snake_case_						       =	self.model_tester.prepare_config_and_inputs_for_common()

							snake_case_						       =	_config_zero_init(__lowercase		)
							snake_case_						       =	_config_zero_init(configs_no_init.backbone_config		)
							for model_class in self.all_model_classes:
								snake_case_						       =	model_class(config=__lowercase		)
								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				:	Optional[int]		):






							"""simple docstring"""






							pass






						@slow
						def 			snake_case__		(		self				:	Tuple		):






							"""simple docstring"""






							for model_name in UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
								snake_case_						       =	UperNetForSemanticSegmentation.from_pretrained(__lowercase		)
								self.assertIsNotNone(__lowercase		)







def 		lowerCamelCase__       (						):







	'''simple docstring'''





	snake_case_						       =	hf_hub_download(
	    repo_id="hf-internal-testing/fixtures_ade20k"	,      repo_type="dataset"	,      filename="ADE_val_00000001.jpg"				)
	snake_case_						       =	Image.open(_A				).convert("RGB"				)
	return image






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







						'''simple docstring'''





						def 			snake_case__		(		self				:	Tuple		):






							"""simple docstring"""






							snake_case_						       =	AutoImageProcessor.from_pretrained("openmmlab/upernet-swin-tiny"		)
							snake_case_						       =	UperNetForSemanticSegmentation.from_pretrained("openmmlab/upernet-swin-tiny"		).to(__lowercase		)

							snake_case_						       =	prepare_img()
							snake_case_						       =	processor(images=__lowercase			,      return_tensors="pt"		).to(__lowercase		)

							with torch.no_grad():
								snake_case_						       =	model(**__lowercase		)

							snake_case_						       =	torch.Size((1, model.config.num_labels, 5_12, 5_12)		)
							self.assertEqual(outputs.logits.shape			,      __lowercase		)

							snake_case_						       =	torch.tensor(
							    [[-7.5958, -7.5958, -7.4302], [-7.5958, -7.5958, -7.4302], [-7.4797, -7.4797, -7.3068]]		).to(__lowercase		)
							self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]			,      __lowercase			,      atol=1E-4		)		)






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






							"""simple docstring"""






							snake_case_						       =	AutoImageProcessor.from_pretrained("openmmlab/upernet-convnext-tiny"		)
							snake_case_						       =	UperNetForSemanticSegmentation.from_pretrained("openmmlab/upernet-convnext-tiny"		).to(__lowercase		)

							snake_case_						       =	prepare_img()
							snake_case_						       =	processor(images=__lowercase			,      return_tensors="pt"		).to(__lowercase		)

							with torch.no_grad():
								snake_case_						       =	model(**__lowercase		)

							snake_case_						       =	torch.Size((1, model.config.num_labels, 5_12, 5_12)		)
							self.assertEqual(outputs.logits.shape			,      __lowercase		)

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






import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
		from sagemaker import Session, TrainingJobAnalytics
		from sagemaker.huggingface import HuggingFace
@pytest.mark.skipif(
    literal_eval(os.getenv('''TEST_SAGEMAKER'''			,  '''False'''     )     ) is not True			,  reason='''Skipping test because should only be run when releasing minor transformers version'''			,  )
@pytest.mark.usefixtures('''sm_env'''     )
@parameterized_class(
    [
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 650, '''eval_accuracy''': 0.7, '''eval_loss''': 0.6},
        },
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_ddp.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 600, '''eval_accuracy''': 0.7, '''eval_loss''': 0.6},
        },
        {
            '''framework''': '''tensorflow''',
            '''script''': '''run_tf_dist.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 600, '''eval_accuracy''': 0.6, '''eval_loss''': 0.7},
        },
    ]     )
class 						UpperCAmelCase  (							unittest.TestCase     ):







						'''simple docstring'''





						def 			snake_case__		(		self				:	Optional[Any]		):






							"""simple docstring"""






							if self.framework == "pytorch":
								subprocess.run(
								    f"cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py".split()			,      encoding="utf-8"			,      check=__lowercase			,      )
							assert hasattr(self			,      "env"		)






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	Optional[Any]		):






							"""simple docstring"""






							snake_case_						       =	f"{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"
							# distributed data settings
							snake_case_						       =	{"smdistributed": {"dataparallel": {"enabled": True}}} if self.script != "run_ddp.py" else None

							# creates estimator
							return HuggingFace(
							    entry_point=self.script			,      source_dir=self.env.test_path			,      role=self.env.role			,      image_uri=self.env.image_uri			,      base_job_name=__lowercase			,      instance_count=__lowercase			,      instance_type=self.instance_type			,      debugger_hook_config=__lowercase			,      hyperparameters={**self.env.distributed_hyperparameters, "model_name_or_path": self.model_name_or_path}			,      metric_definitions=self.env.metric_definitions			,      distribution=__lowercase			,      py_version="py36"			,      )






						def 			snake_case__		(		self				:	Union[str, Any]			,      __lowercase				:	Tuple		):






							"""simple docstring"""






							TrainingJobAnalytics(__lowercase		).export_csv(f"{self.env.test_path}/{job_name}_metrics.csv"		)






						@parameterized.expand([(2,)]		)
						def 			snake_case__		(		self				:	Union[str, Any]			,      __lowercase				:	Union[str, Any]		):






							"""simple docstring"""






							snake_case_						       =	self.create_estimator(__lowercase		)

							# run training
							estimator.fit()

							# result dataframe
							snake_case_						       =	TrainingJobAnalytics(estimator.latest_training_job.name		).dataframe()

							# extract kpis
							snake_case_						       =	list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]		)
							snake_case_						       =	list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]		)
							# get train time from SageMaker job, this includes starting, preprocessing, stopping
							snake_case_						       =	(
							    Session().describe_training_job(estimator.latest_training_job.name		).get("TrainingTimeInSeconds"			,      99_99_99		)
							)

							# assert kpis
							assert train_runtime <= self.results["train_runtime"]
							assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy		)
							assert all(t <= self.results["eval_loss"] for t in eval_loss		)

							# dump tests result into json file to share in PR
							with open(f"{estimator.latest_training_job.name}.json"			,      "w"		) as outfile:
								json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}			,      __lowercase		)
