


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 				SCREAMING_SNAKE_CASE__						:
      def __init__( self       : Optional[Any]  ,							SCREAMING_SNAKE_CASE__       : Any  ,							SCREAMING_SNAKE_CASE__       : str=1_3  ,							SCREAMING_SNAKE_CASE__       : Tuple=3_2  ,							SCREAMING_SNAKE_CASE__       : Dict=3  ,							SCREAMING_SNAKE_CASE__       : str=4  ,							SCREAMING_SNAKE_CASE__       : Tuple=[1_0, 2_0, 3_0, 4_0]  ,							SCREAMING_SNAKE_CASE__       : Dict=[2, 2, 3, 2]  ,							SCREAMING_SNAKE_CASE__       : List[str]=True  ,							SCREAMING_SNAKE_CASE__       : List[str]=True  ,							SCREAMING_SNAKE_CASE__       : str=3_7  ,							SCREAMING_SNAKE_CASE__       : Union[str, Any]="gelu"  ,							SCREAMING_SNAKE_CASE__       : Optional[int]=1_0  ,							SCREAMING_SNAKE_CASE__       : Tuple=0.02  ,							SCREAMING_SNAKE_CASE__       : List[str]=["stage2", "stage3", "stage4"]  ,							SCREAMING_SNAKE_CASE__       : List[str]=3  ,							SCREAMING_SNAKE_CASE__       : str=None  ,							)	->    Optional[Any]:
             a_      :	List[str]							=						parent
             a_      :	Any							=						batch_size
             a_      :	Any							=						image_size
             a_      :	Dict							=						num_channels
             a_      :	Tuple							=						num_stages
             a_      :	Optional[int]							=						hidden_sizes
             a_      :	Dict							=						depths
             a_      :	Dict							=						is_training
             a_      :	Dict							=						use_labels
             a_      :	Union[str, Any]							=						intermediate_size
             a_      :	Any							=						hidden_act
             a_      :	Tuple							=						type_sequence_label_size
             a_      :	Union[str, Any]							=						initializer_range
             a_      :	int							=						out_features
             a_      :	str							=						num_labels
             a_      :	Optional[Any]							=						scope
             a_      :	List[Any]							=						num_stages
      def  SCREAMING_SNAKE_CASE							( self       : Dict  )	->    Tuple:
             a_      :	Optional[int]							=						floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]  )

             a_      :	Dict							=						None
             if self.use_labels:
                    a_      :	List[Any]							=						ids_tensor([self.batch_size]  ,							self.type_sequence_label_size  )

             a_      :	Any							=						self.get_config()

             return config, pixel_values, labels
      def  SCREAMING_SNAKE_CASE							( self       : Dict  )	->    List[Any]:
             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  SCREAMING_SNAKE_CASE							( self       : Any  )	->    int:
             return UperNetConfig(
                 backbone_config=self.get_backbone_config()  ,							hidden_size=5_1_2  ,							pool_scales=[1, 2, 3, 6]  ,							use_auxiliary_head=SCREAMING_SNAKE_CASE__  ,							auxiliary_loss_weight=0.4  ,							auxiliary_in_channels=4_0  ,							auxiliary_channels=2_5_6  ,							auxiliary_num_convs=1  ,							auxiliary_concat_input=SCREAMING_SNAKE_CASE__  ,							loss_ignore_index=2_5_5  ,							num_labels=self.num_labels  ,							)
      def  SCREAMING_SNAKE_CASE							( self       : Dict  ,							SCREAMING_SNAKE_CASE__       : str  ,							SCREAMING_SNAKE_CASE__       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : List[Any]  )	->    Tuple:
             a_      :	Dict							=						UperNetForSemanticSegmentation(config=SCREAMING_SNAKE_CASE__  )
             model.to(SCREAMING_SNAKE_CASE__  )
             model.eval()
             a_      :	str							=						model(SCREAMING_SNAKE_CASE__  )
             self.parent.assertEqual(
                 result.logits.shape  ,							(self.batch_size, self.num_labels, self.image_size, self.image_size)  )

      def  SCREAMING_SNAKE_CASE							( self       : int  )	->    Any:
             a_      :	str							=						self.prepare_config_and_inputs()
             (
                 (
                 a_
             )							,	(
                 a_
             )							,	(
                 a_
             )							,	
             )      :	List[Any]							=						config_and_inputs
             a_      :	Union[str, Any]							=						{'pixel_values': pixel_values}
             return config, inputs_dict
@require_torch
class 				SCREAMING_SNAKE_CASE__						(       lowercase__		,					lowercase__		,					unittest.TestCase    ):
      snake_case__       :  Any       =		(UperNetForSemanticSegmentation,) if is_torch_available() else ()
      snake_case__       :  Dict       =		{'''image-segmentation''': UperNetForSemanticSegmentation} if is_torch_available() else {}
      snake_case__       :  str       =		False
      snake_case__       :  List[Any]       =		False
      snake_case__       :  str       =		False
      snake_case__       :  str       =		False
      snake_case__       :  str       =		False
      snake_case__       :  Union[str, Any]       =		False
      def  SCREAMING_SNAKE_CASE							( self       : str  )	->    Union[str, Any]:
             a_      :	List[Any]							=						UperNetModelTester(self  )
             a_      :	Optional[Any]							=						ConfigTester(self  ,							config_class=SCREAMING_SNAKE_CASE__  ,							has_text_modality=SCREAMING_SNAKE_CASE__  ,							hidden_size=3_7  )
      def  SCREAMING_SNAKE_CASE							( self       : List[str]  )	->    int:
             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  SCREAMING_SNAKE_CASE							( self       : int  )	->    Union[str, Any]:
             return
      def  SCREAMING_SNAKE_CASE							( self       : Dict  )	->    Optional[int]:
             a_							,	a_      :	int							=						self.model_tester.prepare_config_and_inputs_for_common()

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

                    a_      :	Union[str, Any]							=						['pixel_values']
                    self.assertListEqual(arg_names[:1]  ,							SCREAMING_SNAKE_CASE__  )
      def  SCREAMING_SNAKE_CASE							( self       : Dict  )	->    int:
             a_      :	Tuple							=						self.model_tester.prepare_config_and_inputs()
             self.model_tester.create_and_check_for_semantic_segmentation(*SCREAMING_SNAKE_CASE__  )
      @unittest.skip(reason='UperNet does not use inputs_embeds'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[int]  )	->    int:
             pass
      @unittest.skip(reason='UperNet does not support input and output embeddings'  )
      def  SCREAMING_SNAKE_CASE							( self       : str  )	->    List[Any]:
             pass
      @unittest.skip(reason='UperNet does not have a base model'  )
      def  SCREAMING_SNAKE_CASE							( self       : Tuple  )	->    Dict:
             pass
      @unittest.skip(reason='UperNet does not have a base model'  )
      def  SCREAMING_SNAKE_CASE							( self       : Tuple  )	->    Optional[Any]:
             pass
      @require_torch_multi_gpu
      @unittest.skip(reason='UperNet has some layers using `add_module` which doesn\'t work well with `nn.DataParallel`'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[Any]  )	->    List[str]:
             pass
      @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'  )
      def  SCREAMING_SNAKE_CASE							( self       : List[Any]  )	->    List[str]:
             pass
      def  SCREAMING_SNAKE_CASE							( self       : str  )	->    str:
             def check_hidden_states_output(SCREAMING_SNAKE_CASE__       : List[Any]  ,							SCREAMING_SNAKE_CASE__       : Tuple  ,							SCREAMING_SNAKE_CASE__       : str  ):
                    a_      :	int							=						model_class(SCREAMING_SNAKE_CASE__  )
                    model.to(SCREAMING_SNAKE_CASE__  )
                    model.eval()

                    with torch.no_grad():
                           a_      :	str							=						model(**self._prepare_for_class(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  )  )

                    a_      :	int							=						outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

                    a_      :	str							=						self.model_tester.num_stages
                    self.assertEqual(len(SCREAMING_SNAKE_CASE__  )  ,							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]  ,							)

             a_							,	a_      :	Optional[int]							=						self.model_tester.prepare_config_and_inputs_for_common()

             for model_class in self.all_model_classes:
                    a_      :	Optional[int]							=						True
                    check_hidden_states_output(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  )

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

                    check_hidden_states_output(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  )
      def  SCREAMING_SNAKE_CASE							( self       : int  )	->    str:
             a_							,	a_      :	Tuple							=						self.model_tester.prepare_config_and_inputs_for_common()

             a_      :	Tuple							=						_config_zero_init(SCREAMING_SNAKE_CASE__  )
             a_      :	Optional[Any]							=						_config_zero_init(configs_no_init.backbone_config  )
             for model_class in self.all_model_classes:
                    a_      :	Optional[Any]							=						model_class(config=SCREAMING_SNAKE_CASE__  )
                    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  SCREAMING_SNAKE_CASE							( self       : Optional[int]  )	->    Optional[Any]:
             pass

      @slow
      def  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  )	->    Tuple:
             for model_name in UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                    a_      :	Optional[Any]							=						UperNetForSemanticSegmentation.from_pretrained(SCREAMING_SNAKE_CASE__  )
                    self.assertIsNotNone(SCREAMING_SNAKE_CASE__  )







def  SCREAMING_SNAKE_CASE_		(    )			->	List[Any]:

       """simple docstring"""

       a_      :	str							=						hf_hub_download(
           repo_id='hf-internal-testing/fixtures_ade20k'						,					repo_type='dataset'						,					filename='ADE_val_00000001.jpg'       )
       a_      :	Union[str, Any]							=						Image.open(__A       ).convert('RGB'       )
       return image
@require_torch
@require_vision
@slow
class 				SCREAMING_SNAKE_CASE__						(       unittest.TestCase    ):
      def  SCREAMING_SNAKE_CASE							( self       : Dict  )	->    Dict:
             a_      :	Tuple							=						AutoImageProcessor.from_pretrained('openmmlab/upernet-swin-tiny'  )
             a_      :	Optional[Any]							=						UperNetForSemanticSegmentation.from_pretrained('openmmlab/upernet-swin-tiny'  ).to(SCREAMING_SNAKE_CASE__  )

             a_      :	Dict							=						prepare_img()
             a_      :	Tuple							=						processor(images=SCREAMING_SNAKE_CASE__  ,							return_tensors='pt'  ).to(SCREAMING_SNAKE_CASE__  )

             with torch.no_grad():
                    a_      :	Union[str, Any]							=						model(**SCREAMING_SNAKE_CASE__  )

             a_      :	Dict							=						torch.Size((1, model.config.num_labels, 5_1_2, 5_1_2)  )
             self.assertEqual(outputs.logits.shape  ,							SCREAMING_SNAKE_CASE__  )

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

      def  SCREAMING_SNAKE_CASE							( self       : Optional[int]  )	->    Dict:
             a_      :	Dict							=						AutoImageProcessor.from_pretrained('openmmlab/upernet-convnext-tiny'  )
             a_      :	Any							=						UperNetForSemanticSegmentation.from_pretrained('openmmlab/upernet-convnext-tiny'  ).to(SCREAMING_SNAKE_CASE__  )

             a_      :	int							=						prepare_img()
             a_      :	Dict							=						processor(images=SCREAMING_SNAKE_CASE__  ,							return_tensors='pt'  ).to(SCREAMING_SNAKE_CASE__  )

             with torch.no_grad():
                    a_      :	Optional[Any]							=						model(**SCREAMING_SNAKE_CASE__  )

             a_      :	Union[str, Any]							=						torch.Size((1, model.config.num_labels, 5_1_2, 5_1_2)  )
             self.assertEqual(outputs.logits.shape  ,							SCREAMING_SNAKE_CASE__  )

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






# This code is adapted from OpenAI's release
# https://github.com/openai/human-eval/blob/master/human_eval/execution.py

import contextlib
import faulthandler
import io
import multiprocessing
import os
import platform
import signal
import tempfile







def  SCREAMING_SNAKE_CASE_		(    __A				:    List[Any]						,					__A				:    List[Any]						,					__A				:    List[Any]						,					__A				:    Optional[int]       )			->	Dict:

       """simple docstring"""

       a_      :	Optional[Any]							=						multiprocessing.Manager()
       a_      :	Tuple							=						manager.list()

       a_      :	Optional[int]							=						multiprocessing.Process(target=__A						,					args=(check_program, result, timeout)       )
       p.start()
       p.join(timeout=timeout + 1       )
       if p.is_alive():
              p.kill()

       if not result:
              result.append('timed out'       )

       return {
           "task_id": task_id,
           "passed": result[0] == "passed",
           "result": result[0],
           "completion_id": completion_id,
       }







def  SCREAMING_SNAKE_CASE_		(    __A				:    Tuple						,					__A				:    Optional[Any]						,					__A				:    Optional[Any]       )			->	Dict:

       """simple docstring"""

       with create_tempdir():
              # These system calls are needed when cleaning up tempdir.
              import os
              import shutil

              a_      :	Any							=						shutil.rmtree
              a_      :	List[str]							=						os.rmdir
              a_      :	List[Any]							=						os.chdir

              # Disable functionalities that can make destructive changes to the test.
              reliability_guard()

              # Run program.
              try:
                     a_      :	Any							=						{}
                     with swallow_io():
                            with time_limit(__A       ):
                                   exec(__A						,					__A       )
                     result.append('passed'       )
              except TimeoutException:
                     result.append('timed out'       )
              except BaseException as e:
                     result.append(F"""failed: {e}"""       )

              # Needed for cleaning up.
              a_      :	Tuple							=						rmtree
              a_      :	Any							=						rmdir
              a_      :	Optional[int]							=						chdir







@contextlib.contextmanager
def  SCREAMING_SNAKE_CASE_		(    __A				:    List[str]       )			->	str:

       """simple docstring"""

       def signal_handler(__A				:    List[str]						,					__A				:    Dict       ):
              raise TimeoutException('Timed out!'       )

       signal.setitimer(signal.ITIMER_REAL						,					__A       )
       signal.signal(signal.SIGALRM						,					__A       )
       try:
              yield
       finally:
              signal.setitimer(signal.ITIMER_REAL						,					0       )







@contextlib.contextmanager
def  SCREAMING_SNAKE_CASE_		(    )			->	Dict:

       """simple docstring"""

       a_      :	int							=						WriteOnlyStringIO()
       with contextlib.redirect_stdout(__A       ):
              with contextlib.redirect_stderr(__A       ):
                     with redirect_stdin(__A       ):
                            yield







@contextlib.contextmanager
def  SCREAMING_SNAKE_CASE_		(    )			->	Optional[int]:

       """simple docstring"""

       with tempfile.TemporaryDirectory() as dirname:
              with chdir(__A       ):
                     yield dirname
class 				SCREAMING_SNAKE_CASE__						(       lowercase__    ):
      pass
class 				SCREAMING_SNAKE_CASE__						(       io.StringIO    ):
      def  SCREAMING_SNAKE_CASE							( self       : List[Any]  ,							*SCREAMING_SNAKE_CASE__       : List[str]  ,							**SCREAMING_SNAKE_CASE__       : List[Any]  )	->    int:
             raise OSError
      def  SCREAMING_SNAKE_CASE							( self       : Tuple  ,							*SCREAMING_SNAKE_CASE__       : Tuple  ,							**SCREAMING_SNAKE_CASE__       : Dict  )	->    List[str]:
             raise OSError
      def  SCREAMING_SNAKE_CASE							( self       : int  ,							*SCREAMING_SNAKE_CASE__       : Dict  ,							**SCREAMING_SNAKE_CASE__       : List[Any]  )	->    Optional[Any]:
             raise OSError

      def  SCREAMING_SNAKE_CASE							( self       : Any  ,							*SCREAMING_SNAKE_CASE__       : Union[str, Any]  ,							**SCREAMING_SNAKE_CASE__       : List[Any]  )	->    List[Any]:
             return False
class 				SCREAMING_SNAKE_CASE__						(       contextlib._RedirectStream    ):  # type: ignore
      snake_case__       :  Optional[int]       =		'''stdin'''







@contextlib.contextmanager
def  SCREAMING_SNAKE_CASE_		(    __A				:    Tuple       )			->	Union[str, Any]:

       """simple docstring"""

       if root == ".":
              yield
              return
       a_      :	str							=						os.getcwd()
       os.chdir(__A       )
       try:
              yield
       except BaseException as exc:
              raise exc
       finally:
              os.chdir(__A       )







def  SCREAMING_SNAKE_CASE_		(    __A				:    Dict=None       )			->	Dict:

       """simple docstring"""

       if maximum_memory_bytes is not None:
              import resource

              resource.setrlimit(resource.RLIMIT_AS						,					(maximum_memory_bytes, maximum_memory_bytes)       )
              resource.setrlimit(resource.RLIMIT_DATA						,					(maximum_memory_bytes, maximum_memory_bytes)       )
              if not platform.uname().system == "Darwin":
                     resource.setrlimit(resource.RLIMIT_STACK						,					(maximum_memory_bytes, maximum_memory_bytes)       )

       faulthandler.disable()

       import builtins

       a_      :	str							=						None
       a_      :	str							=						None

       import os

       a_      :	Optional[int]							=						'1'

       a_      :	Any							=						None
       a_      :	int							=						None
       a_      :	int							=						None
       a_      :	Tuple							=						None
       a_      :	Dict							=						None
       a_      :	Any							=						None
       a_      :	Dict							=						None
       a_      :	Dict							=						None
       a_      :	List[str]							=						None
       a_      :	Optional[Any]							=						None
       a_      :	Tuple							=						None
       a_      :	Union[str, Any]							=						None
       a_      :	Dict							=						None
       a_      :	Dict							=						None
       a_      :	int							=						None
       a_      :	str							=						None
       a_      :	Union[str, Any]							=						None
       a_      :	str							=						None
       a_      :	Optional[int]							=						None
       a_      :	List[Any]							=						None
       a_      :	Union[str, Any]							=						None
       a_      :	Dict							=						None
       a_      :	Tuple							=						None
       a_      :	List[str]							=						None
       a_      :	str							=						None
       a_      :	str							=						None
       a_      :	str							=						None

       import shutil

       a_      :	Dict							=						None
       a_      :	Tuple							=						None
       a_      :	str							=						None

       import subprocess

       a_      :	str							=						None  # type: ignore

       a_      :	List[Any]							=						None

       import sys

       a_      :	Optional[int]							=						None
       a_      :	int							=						None
       a_      :	Any							=						None
       a_      :	str							=						None
       a_      :	str							=						None


