


"""simple docstring"""




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 	A_ (	unittest.TestCase   ):




      '''simple docstring'''


      def 					_snake_case     (    self:	Optional[int]    ):
       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=a						,				)
       assert hasattr(self						,				'env'    )




      def 					_snake_case     (    self:	Optional[Any]						,				a:	Union[str, Any]    ):
       __lowerCamelCase		:							Tuple		 =						F'{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'
       # distributed data settings
       __lowerCamelCase		:							List[str]		 =						{'''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=a						,				instance_count=a						,				instance_type=self.instance_type						,				debugger_hook_config=a						,				hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path}						,				metric_definitions=self.env.metric_definitions						,				distribution=a						,				py_version='py36'						,				)




      def 					_snake_case     (    self:	Union[str, Any]						,				a:	Union[str, Any]    ):
       TrainingJobAnalytics(a    ).export_csv(F'{self.env.test_path}/{job_name}_metrics.csv'    )




      @parameterized.expand([(2,)]    )
      def 					_snake_case     (    self:	Dict						,				a:	Any    ):
       # create estimator
       __lowerCamelCase		:							Optional[int]		 =						self.create_estimator(a    )

       # run training
       estimator.fit()

       # result dataframe
       __lowerCamelCase		:							Tuple		 =						TrainingJobAnalytics(estimator.latest_training_job.name    ).dataframe()

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

       # 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}						,				a    )





import copy
from typing import TYPE_CHECKING, Any, Mapping, Optional, OrderedDict

from packaging import version

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig


if TYPE_CHECKING:
       from ... import PreTrainedTokenizerBase, TensorType

lowercase_	   =     logging.get_logger(__name__)


class 	A_ (	__UpperCamelCase   ):




      '''simple docstring'''


      __snake_case 	=       """vision-encoder-decoder"""
      __snake_case 	=       True




      def __init__(    self:	Union[str, Any]						,				**a:	Optional[Any]    ):
       super().__init__(**a    )
       if "encoder" not in kwargs or "decoder" not in kwargs:
        raise ValueError(
            F'A configuraton of type {self.model_type} cannot be instantiated because '
            F'not both `encoder` and `decoder` sub-configurations are passed, but only {kwargs}'    )

       __lowerCamelCase		:							Dict		 =						kwargs.pop('encoder'    )
       __lowerCamelCase		:							int		 =						encoder_config.pop('model_type'    )
       __lowerCamelCase		:							Any		 =						kwargs.pop('decoder'    )
       __lowerCamelCase		:							Union[str, Any]		 =						decoder_config.pop('model_type'    )

       __lowerCamelCase		:							Optional[Any]		 =						AutoConfig.for_model(a						,				**a    )
       __lowerCamelCase		:							List[Any]		 =						AutoConfig.for_model(a						,				**a    )
       __lowerCamelCase		:							Tuple		 =						True




      @classmethod
      def 					_snake_case     (    cls:	Optional[Any]						,				a:	PretrainedConfig						,				a:	PretrainedConfig						,				**a:	Dict    ):
       logger.info('Setting `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config'    )
       __lowerCamelCase		:							Optional[Any]		 =						True
       __lowerCamelCase		:							Optional[Any]		 =						True

       return cls(encoder=encoder_config.to_dict()						,				decoder=decoder_config.to_dict()						,				**a    )




      def 					_snake_case     (    self:	str    ):
       __lowerCamelCase		:							int		 =						copy.deepcopy(self.__dict__    )
       __lowerCamelCase		:							Dict		 =						self.encoder.to_dict()
       __lowerCamelCase		:							Union[str, Any]		 =						self.decoder.to_dict()
       __lowerCamelCase		:							Optional[int]		 =						self.__class__.model_type
       return output



class 	A_ (	__UpperCamelCase   ):




      '''simple docstring'''


      __snake_case 	=       version.parse("""1.11"""   )




      @property
      def 					_snake_case     (    self:	Union[str, Any]    ):
       return OrderedDict(
           [
               ('pixel_values', {0: 'batch', 1: 'num_channels', 2: 'height', 3: 'width'}),
           ]    )




      @property
      def 					_snake_case     (    self:	Optional[Any]    ):
       return 1e-4




      @property
      def 					_snake_case     (    self:	List[str]    ):
       return OrderedDict({'last_hidden_state': {0: 'batch', 1: 'encoder_sequence'}}    )



class 	A_ (	__UpperCamelCase   ):




      '''simple docstring'''


      @property
      def 					_snake_case     (    self:	Tuple    ):
       __lowerCamelCase		:							Union[str, Any]		 =						OrderedDict()
       __lowerCamelCase		:							Union[str, Any]		 =						{0: 'batch', 1: 'past_decoder_sequence + sequence'}
       __lowerCamelCase		:							Optional[Any]		 =						{0: 'batch', 1: 'past_decoder_sequence + sequence'}
       __lowerCamelCase		:							Dict		 =						{0: 'batch', 1: 'encoder_sequence'}

       return common_inputs




      def 					_snake_case     (    self:	List[Any]						,				a:	"PreTrainedTokenizerBase"						,				a:	int = -1						,				a:	int = -1						,				a:	bool = False						,				a:	Optional["TensorType"] = None						,				):
       import torch

       __lowerCamelCase		:							str		 =						OrderedDict()

       __lowerCamelCase		:							List[Any]		 =						super().generate_dummy_inputs(
           a						,				batch_size=a						,				seq_length=a						,				is_pair=a						,				framework=a    )

       __lowerCamelCase ,					__lowerCamelCase		:							Dict		 =						dummy_input['input_ids'].shape
       __lowerCamelCase		:							str		 =						(batch, encoder_sequence, self._config.encoder_hidden_size)
       __lowerCamelCase		:							str		 =						dummy_input.pop('input_ids'    )
       __lowerCamelCase		:							Union[str, Any]		 =						dummy_input.pop('attention_mask'    )
       __lowerCamelCase		:							str		 =						torch.zeros(a    )

       return common_inputs



class 	A_ (	__UpperCamelCase   ):




      '''simple docstring'''


      @property
      def 					_snake_case     (    self:	List[Any]    ):
       pass




      def 					_snake_case     (    self:	List[str]						,				a:	PretrainedConfig    ):
       return VisionEncoderDecoderEncoderOnnxConfig(a    )




      def 					_snake_case     (    self:	Optional[int]						,				a:	PretrainedConfig						,				a:	PretrainedConfig						,				a:	str = "default"    ):
       __lowerCamelCase		:							List[Any]		 =						encoder_config.hidden_size
       return VisionEncoderDecoderDecoderOnnxConfig(a						,				a    )

