




"""simple docstring"""


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

import pytest
from parameterized import 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.g4dn.xlarge",
            "results": {"train_runtime": 650, "eval_accuracy": 0.6, "eval_loss": 0.9},
        },
        {
            "framework": "tensorflow",
            "script": "run_tf.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 600, "eval_accuracy": 0.3, "eval_loss": 0.9},
        },
    ]						)
class _lowerCamelCase    (					unittest.TestCase						):






   def 	_lowerCAmelCase   (		self       :		Optional[Any]   )			->						Union[str, 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=UpperCamelCase   ,     )
         assert hasattr(self   ,     """env"""   )






   def 	_lowerCAmelCase   (		self       :		Dict   ,     UpperCamelCase       :		List[str]=1   )			->						List[Any]:

         """simple docstring"""
         # 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=f"""{self.env.base_job_name}-single"""   ,     instance_count=UpperCamelCase   ,     instance_type=self.instance_type   ,     debugger_hook_config=UpperCamelCase   ,     hyperparameters={**self.env.hyperparameters, """model_name_or_path""": self.model_name_or_path}   ,     metric_definitions=self.env.metric_definitions   ,     py_version="""py36"""   ,     )






   def 	_lowerCAmelCase   (		self       :		Tuple   ,     UpperCamelCase       :		List[str]   )			->						str:

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






   def 	_lowerCAmelCase   (		self       :		Optional[int]   )			->						str:

         """simple docstring"""
         # create estimator
         lowerCAmelCase__      :			Union[str, Any]   =	self.create_estimator()

         # run training
         estimator.fit()

         # result dataframe
         lowerCAmelCase__      :			List[str]   =	TrainingJobAnalytics(estimator.latest_training_job.name   ).dataframe()

         # extract kpis
         lowerCAmelCase__      :			Union[str, Any]   =	list(result_metrics_df[result_metrics_df.metric_name == """eval_accuracy"""]["""value"""]   )
         lowerCAmelCase__      :			Optional[int]   =	list(result_metrics_df[result_metrics_df.metric_name == """eval_loss"""]["""value"""]   )
         # get train time from SageMaker job, this includes starting, preprocessing, stopping
         lowerCAmelCase__      :			Optional[int]   =	(
             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}   ,     UpperCamelCase   )











"""simple docstring"""


from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType


class _lowerCamelCase    (					a_						):
   _lowerCamelCase    :Dict     				=     ["image_processor", "tokenizer"]
   _lowerCamelCase    :Dict     				=     "BlipImageProcessor"
   _lowerCamelCase    :Any     				=     ("BertTokenizer", "BertTokenizerFast")






   def __init__(		self       :		Dict   ,     UpperCamelCase       :		Optional[int]   ,     UpperCamelCase       :		Union[str, Any]   )			->						str:

         """simple docstring"""
         lowerCAmelCase__      :			Optional[Any]   =	False
         super().__init__(UpperCamelCase   ,     UpperCamelCase   )
         lowerCAmelCase__      :			List[Any]   =	self.image_processor






   def __call__(		self       :		int   ,     UpperCamelCase       :		ImageInput = None   ,     UpperCamelCase       :		Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None   ,     UpperCamelCase       :		bool = True   ,     UpperCamelCase       :		Union[bool, str, PaddingStrategy] = False   ,     UpperCamelCase       :		Union[bool, str, TruncationStrategy] = None   ,     UpperCamelCase       :		Optional[int] = None   ,     UpperCamelCase       :		int = 0   ,     UpperCamelCase       :		Optional[int] = None   ,     UpperCamelCase       :		Optional[bool] = None   ,     UpperCamelCase       :		bool = False   ,     UpperCamelCase       :		bool = False   ,     UpperCamelCase       :		bool = False   ,     UpperCamelCase       :		bool = False   ,     UpperCamelCase       :		bool = False   ,     UpperCamelCase       :		bool = True   ,     UpperCamelCase       :		Optional[Union[str, TensorType]] = None   ,     **UpperCamelCase       :		Optional[int]   ,     )			->						BatchEncoding:

         """simple docstring"""
         if images is None and text is None:
               raise ValueError("""You have to specify either images or text."""   )

         # Get only text
         if images is None:
               lowerCAmelCase__      :			Any   =	self.tokenizer
               lowerCAmelCase__      :			Optional[int]   =	self.tokenizer(
                   text=UpperCamelCase   ,     add_special_tokens=UpperCamelCase   ,     padding=UpperCamelCase   ,     truncation=UpperCamelCase   ,     max_length=UpperCamelCase   ,     stride=UpperCamelCase   ,     pad_to_multiple_of=UpperCamelCase   ,     return_attention_mask=UpperCamelCase   ,     return_overflowing_tokens=UpperCamelCase   ,     return_special_tokens_mask=UpperCamelCase   ,     return_offsets_mapping=UpperCamelCase   ,     return_token_type_ids=UpperCamelCase   ,     return_length=UpperCamelCase   ,     verbose=UpperCamelCase   ,     return_tensors=UpperCamelCase   ,     **UpperCamelCase   ,     )
               return text_encoding

         # add pixel_values
         lowerCAmelCase__      :			Tuple   =	self.image_processor(UpperCamelCase   ,     return_tensors=UpperCamelCase   )

         if text is not None:
               lowerCAmelCase__      :			Optional[int]   =	self.tokenizer(
                   text=UpperCamelCase   ,     add_special_tokens=UpperCamelCase   ,     padding=UpperCamelCase   ,     truncation=UpperCamelCase   ,     max_length=UpperCamelCase   ,     stride=UpperCamelCase   ,     pad_to_multiple_of=UpperCamelCase   ,     return_attention_mask=UpperCamelCase   ,     return_overflowing_tokens=UpperCamelCase   ,     return_special_tokens_mask=UpperCamelCase   ,     return_offsets_mapping=UpperCamelCase   ,     return_token_type_ids=UpperCamelCase   ,     return_length=UpperCamelCase   ,     verbose=UpperCamelCase   ,     return_tensors=UpperCamelCase   ,     **UpperCamelCase   ,     )
         else:
               lowerCAmelCase__      :			Tuple   =	None

         if text_encoding is not None:
               encoding_image_processor.update(UpperCamelCase   )

         return encoding_image_processor






   def 	_lowerCAmelCase   (		self       :		int   ,     *UpperCamelCase       :		Union[str, Any]   ,     **UpperCamelCase       :		List[str]   )			->						Dict:

         """simple docstring"""
         return self.tokenizer.batch_decode(*UpperCamelCase   ,     **UpperCamelCase   )






   def 	_lowerCAmelCase   (		self       :		List[Any]   ,     *UpperCamelCase       :		Tuple   ,     **UpperCamelCase       :		List[str]   )			->						List[Any]:

         """simple docstring"""
         return self.tokenizer.decode(*UpperCamelCase   ,     **UpperCamelCase   )






   @property
   def 	_lowerCAmelCase   (		self       :		List[Any]   )			->						Tuple:

         """simple docstring"""
         lowerCAmelCase__      :			Optional[int]   =	self.tokenizer.model_input_names
         lowerCAmelCase__      :			Union[str, Any]   =	self.image_processor.model_input_names
         return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names   )   )





