





import warnings

from transformers import AutoTokenizer
from transformers.utils import is_torch_available
from transformers.utils.generic import ExplicitEnum

from ...processing_utils import ProcessorMixin


if is_torch_available():
     import torch



class 							_UpperCAmelCase       (       UpperCAmelCase__       ):

  """simple docstring"""


  a_              =					"""char"""
  a_              =					"""bpe"""
  a_              =					"""wp"""





_snake_case    :       List[Any]		       =    (DecodeType.CHARACTER, DecodeType.BPE, DecodeType.WORDPIECE)



class 							_UpperCAmelCase       (       UpperCAmelCase__       ):

  """simple docstring"""


  a_              =					["""image_processor""", """char_tokenizer"""]
  a_              =					"""ViTImageProcessor"""
  a_              =					"""MgpstrTokenizer"""






  def __init__(							self					:						Union[str, Any]   ,    lowerCAmelCase_					:						int=None   ,    lowerCAmelCase_					:						Dict=None   ,    **lowerCAmelCase_					:						Optional[Any]	)				->		List[Any]:
     __lowerCAmelCase				      =      None
     if "feature_extractor" in kwargs:
        warnings.warn(
            'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
            ' instead.'   ,    __lowercase   ,    )
        __lowerCAmelCase				      =      kwargs.pop('feature_extractor'	)

     __lowerCAmelCase				      =      image_processor if image_processor is not None else feature_extractor
     if image_processor is None:
        raise ValueError('You need to specify an `image_processor`.'	)
     if tokenizer is None:
        raise ValueError('You need to specify a `tokenizer`.'	)

     __lowerCAmelCase				      =      tokenizer
     __lowerCAmelCase				      =      AutoTokenizer.from_pretrained('gpt2'	)
     __lowerCAmelCase				      =      AutoTokenizer.from_pretrained('bert-base-uncased'	)

     super().__init__(__lowercase   ,    __lowercase	)






  def __call__(							self					:						List[str]   ,    lowerCAmelCase_					:						Dict=None   ,    lowerCAmelCase_					:						Dict=None   ,    lowerCAmelCase_					:						Optional[int]=None   ,    **lowerCAmelCase_					:						Optional[Any]	)				->		int:






     if images is None and text is None:
        raise ValueError('You need to specify either an `images` or `text` input to process.'	)

     if images is not None:
        __lowerCAmelCase				      =      self.image_processor(__lowercase   ,    return_tensors=__lowercase   ,    **__lowercase	)
     if text is not None:
        __lowerCAmelCase				      =      self.char_tokenizer(__lowercase   ,    return_tensors=__lowercase   ,    **__lowercase	)

     if text is None:
        return inputs
     elif images is None:
        return encodings
     else:
        __lowerCAmelCase				      =      encodings['input_ids']
        return inputs






  def 							lowercase						(							self					:						Union[str, Any]   ,    lowerCAmelCase_					:						Tuple	)				->		List[str]:
     __lowerCAmelCase      ,      __lowerCAmelCase      ,      __lowerCAmelCase				      =      sequences
     __lowerCAmelCase				      =      char_preds.size(0	)

     __lowerCAmelCase      ,      __lowerCAmelCase				      =      self._decode_helper(__lowercase   ,    'char'	)
     __lowerCAmelCase      ,      __lowerCAmelCase				      =      self._decode_helper(__lowercase   ,    'bpe'	)
     __lowerCAmelCase      ,      __lowerCAmelCase				      =      self._decode_helper(__lowercase   ,    'wp'	)

     __lowerCAmelCase				      =      []
     __lowerCAmelCase				      =      []
     for i in range(__lowercase	):
        __lowerCAmelCase				      =      [char_scores[i], bpe_scores[i], wp_scores[i]]
        __lowerCAmelCase				      =      [char_strs[i], bpe_strs[i], wp_strs[i]]
        __lowerCAmelCase				      =      scores.index(max(__lowercase	)	)
        final_strs.append(strs[max_score_index]	)
        final_scores.append(scores[max_score_index]	)

     __lowerCAmelCase				      =      {}
     __lowerCAmelCase				      =      final_strs
     __lowerCAmelCase				      =      final_scores
     __lowerCAmelCase				      =      char_strs
     __lowerCAmelCase				      =      bpe_strs
     __lowerCAmelCase				      =      wp_strs
     return out






  def 							lowercase						(							self					:						str   ,    lowerCAmelCase_					:						List[Any]   ,    lowerCAmelCase_					:						List[Any]	)				->		int:






     if format == DecodeType.CHARACTER:
        __lowerCAmelCase				      =      self.char_decode
        __lowerCAmelCase				      =      1
        __lowerCAmelCase				      =      '[s]'
     elif format == DecodeType.BPE:
        __lowerCAmelCase				      =      self.bpe_decode
        __lowerCAmelCase				      =      2
        __lowerCAmelCase				      =      '#'
     elif format == DecodeType.WORDPIECE:
        __lowerCAmelCase				      =      self.wp_decode
        __lowerCAmelCase				      =      1_0_2
        __lowerCAmelCase				      =      '[SEP]'
     else:
        raise ValueError(f"""Format {format} is not supported."""	)

     __lowerCAmelCase      ,      __lowerCAmelCase				      =      [], []
     __lowerCAmelCase				      =      pred_logits.size(0	)
     __lowerCAmelCase				      =      pred_logits.size(1	)
     __lowerCAmelCase      ,      __lowerCAmelCase				      =      pred_logits.topk(1   ,    dim=-1   ,    largest=__lowercase   ,    sorted=__lowercase	)
     __lowerCAmelCase				      =      preds_index.view(-1   ,    __lowercase	)[:, 1:]
     __lowerCAmelCase				      =      decoder(__lowercase	)
     __lowerCAmelCase      ,      __lowerCAmelCase				      =      torch.nn.functional.softmax(__lowercase   ,    dim=2	).max(dim=2	)
     __lowerCAmelCase				      =      preds_max_prob[:, 1:]

     for index in range(__lowercase	):
        __lowerCAmelCase				      =      preds_str[index].find(__lowercase	)
        __lowerCAmelCase				      =      preds_str[index][:pred_eos]
        __lowerCAmelCase				      =      preds_index[index].cpu().tolist()
        __lowerCAmelCase				      =      pred_index.index(__lowercase	) if eos_token in pred_index else -1
        __lowerCAmelCase				      =      preds_max_prob[index][: pred_eos_index + 1]
        __lowerCAmelCase				      =      pred_max_prob.cumprod(dim=0	)[-1] if pred_max_prob.nelement() != 0 else 0.0
        dec_strs.append(__lowercase	)
        conf_scores.append(__lowercase	)

     return dec_strs, conf_scores






  def 							lowercase						(							self					:						Optional[Any]   ,    lowerCAmelCase_					:						Optional[int]	)				->		Optional[int]:
     __lowerCAmelCase				      =      [seq.replace(' '   ,    ''	) for seq in self.char_tokenizer.batch_decode(__lowercase	)]
     return decode_strs






  def 							lowercase						(							self					:						Union[str, Any]   ,    lowerCAmelCase_					:						Any	)				->		List[str]:
     return self.bpe_tokenizer.batch_decode(__lowercase	)






  def 							lowercase						(							self					:						Any   ,    lowerCAmelCase_					:						Union[str, Any]	)				->		Dict:
     __lowerCAmelCase				      =      [seq.replace(' '   ,    ''	) for seq in self.wp_tokenizer.batch_decode(__lowercase	)]
     return decode_strs







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_model_parallelism.py''',
            '''model_name_or_path''': '''roberta-large''',
            '''instance_type''': '''ml.p3dn.24xlarge''',
            '''results''': {'''train_runtime''': 1600, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_glue.py''',
            '''model_name_or_path''': '''roberta-large''',
            '''instance_type''': '''ml.p3dn.24xlarge''',
            '''results''': {'''train_runtime''': 1600, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
    ]     )
class 						UpperCAmelCase  (							unittest.TestCase     ):







						'''simple docstring'''





						def 			snake_case__		(		self				:	int		):






							"""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				:	Any			,      __lowercase				:	Dict		):






							"""simple docstring"""






							snake_case_						       =	{
							    "enabled": True,
							    "processes_per_host": 8,
							}
							snake_case_						       =	{
							    "enabled": True,
							    "parameters": {
							        "microbatches": 4,
							        "placement_strategy": "spread",
							        "pipeline": "interleaved",
							        "optimize": "speed",
							        "partitions": 4,
							        "ddp": True,
							    },
							}

							snake_case_						       =	{"smdistributed": {"modelparallel": smp_options}, "mpi": mpi_options}

							snake_case_						       =	"trainer" if self.script == "run_glue.py" else "smtrainer"
							# 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}-{instance_count}-smp-{name_extension}"			,      instance_count=__lowercase			,      instance_type=self.instance_type			,      debugger_hook_config=__lowercase			,      hyperparameters={
							        **self.env.hyperparameters,
							        "model_name_or_path": self.model_name_or_path,
							        "max_steps": 5_00,
							    }			,      metric_definitions=self.env.metric_definitions			,      distribution=__lowercase			,      py_version="py36"			,      )






						def 			snake_case__		(		self				:	Tuple			,      __lowercase				:	Dict		):






							"""simple docstring"""






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






						@parameterized.expand([(1,)]		)
						def 			snake_case__		(		self				:	Any			,      __lowercase				:	int		):






							"""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		)
