




import warnings
from collections import OrderedDict
from typing import Mapping

from packaging import version

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


lowercase__							:	Tuple											=		logging.get_logger(__name__)

lowercase__							:	Any											=		{
    "nvidia/segformer-b0-finetuned-ade-512-512": (
        "https://huggingface.co/nvidia/segformer-b0-finetuned-ade-512-512/resolve/main/config.json"
    ),
    # See all SegFormer models at https://huggingface.co/models?filter=segformer
}
class 						UpperCAmelCase  (							UpperCAmelCase__     ):







						'''simple docstring'''





						lowerCAmelCase_							=  '''segformer'''






						def __init__(		self				:	List[Any]			,      __lowercase				:	str=3			,      __lowercase				:	Dict=4			,      __lowercase				:	str=[2, 2, 2, 2]			,      __lowercase				:	Dict=[8, 4, 2, 1]			,      __lowercase				:	List[Any]=[32, 64, 1_60, 2_56]			,      __lowercase				:	Union[str, Any]=[7, 3, 3, 3]			,      __lowercase				:	Optional[Any]=[4, 2, 2, 2]			,      __lowercase				:	List[str]=[1, 2, 5, 8]			,      __lowercase				:	str=[4, 4, 4, 4]			,      __lowercase				:	Dict="gelu"			,      __lowercase				:	Optional[Any]=0.0			,      __lowercase				:	List[str]=0.0			,      __lowercase				:	Optional[int]=0.1			,      __lowercase				:	int=0.02			,      __lowercase				:	Union[str, Any]=0.1			,      __lowercase				:	str=1E-6			,      __lowercase				:	Tuple=2_56			,      __lowercase				:	Optional[Any]=2_55			,      **__lowercase				:	int			,      ):






							"""simple docstring"""






							super().__init__(**__lowercase		)

							if "reshape_last_stage" in kwargs and kwargs["reshape_last_stage"] is False:
								warnings.warn(
								    "Reshape_last_stage is set to False in this config. This argument is deprecated and will soon be"
								    " removed, as the behaviour will default to that of reshape_last_stage = True."			,      __lowercase			,      )

							snake_case_						       =	num_channels
							snake_case_						       =	num_encoder_blocks
							snake_case_						       =	depths
							snake_case_						       =	sr_ratios
							snake_case_						       =	hidden_sizes
							snake_case_						       =	patch_sizes
							snake_case_						       =	strides
							snake_case_						       =	mlp_ratios
							snake_case_						       =	num_attention_heads
							snake_case_						       =	hidden_act
							snake_case_						       =	hidden_dropout_prob
							snake_case_						       =	attention_probs_dropout_prob
							snake_case_						       =	classifier_dropout_prob
							snake_case_						       =	initializer_range
							snake_case_						       =	drop_path_rate
							snake_case_						       =	layer_norm_eps
							snake_case_						       =	decoder_hidden_size
							snake_case_						       =	kwargs.get("reshape_last_stage"			,      __lowercase		)
							snake_case_						       =	semantic_loss_ignore_index







class 						UpperCAmelCase  (							UpperCAmelCase__     ):







						'''simple docstring'''





						lowerCAmelCase_							=  version.parse('''1.11'''     )






						@property
						def 			snake_case__		(		self				:	Optional[Any]		):






							"""simple docstring"""






							return OrderedDict(
							    [
							        ("pixel_values", {0: "batch", 1: "num_channels", 2: "height", 3: "width"}),
							    ]		)






						@property
						def 			snake_case__		(		self				:	List[Any]		):






							"""simple docstring"""






							return 1E-4






						@property
						def 			snake_case__		(		self				:	int		):






							"""simple docstring"""






							return 12






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		)
