

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





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





  def 						A  (		self				:     Optional[Any]							, _a				:     Dict			)     ->   Any:





       '''simple docstring'''






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

       _SCREAMING_SNAKE_CASE							   ={'smdistributed': {'modelparallel': smp_options}, 'mpi': mpi_options}

       _SCREAMING_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=_a							, instance_type=self.instance_type							, debugger_hook_config=_a							, hyperparameters={
               **self.env.hyperparameters,
               'model_name_or_path': self.model_name_or_path,
               'max_steps': 500,
           }							, metric_definitions=self.env.metric_definitions							, distribution=_a							, py_version='py36'							, )





  def 						A  (		self				:     Dict							, _a				:     Dict			)     ->   Any:





       '''simple docstring'''






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





  @parameterized.expand([(1,)]			)
  def 						A  (		self				:     str							, _a				:     List[Any]			)     ->   Optional[int]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.create_estimator(_a			)

       # run training
       estimator.fit()

       # result dataframe
       _SCREAMING_SNAKE_CASE							   =TrainingJobAnalytics(estimator.latest_training_job.name			).dataframe()

       # extract kpis
       _SCREAMING_SNAKE_CASE							   =list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']			)
       _SCREAMING_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
       _SCREAMING_SNAKE_CASE							   =(
           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			)



'''simple docstring'''
import warnings
from contextlib import contextmanager

from ....processing_utils import ProcessorMixin







class 							A__   (	A__  ):
  A__					=    'MCTCTFeatureExtractor'
  A__					=    'AutoTokenizer'





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





       '''simple docstring'''






       super().__init__(_a							, _a			)
       _SCREAMING_SNAKE_CASE							   =self.feature_extractor
       _SCREAMING_SNAKE_CASE							   =False





  def __call__(		self				:     Dict							, *_a				:     str							, **_a				:     Dict			)     ->   Dict:





       '''simple docstring'''






       if self._in_target_context_manager:
            return self.current_processor(*_a							, **_a			)

       if "raw_speech" in kwargs:
            warnings.warn('Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'			)
            _SCREAMING_SNAKE_CASE							   =kwargs.pop('raw_speech'			)
       else:
            _SCREAMING_SNAKE_CASE							   =kwargs.pop('audio'							, _a			)
       _SCREAMING_SNAKE_CASE							   =kwargs.pop('sampling_rate'							, _a			)
       _SCREAMING_SNAKE_CASE							   =kwargs.pop('text'							, _a			)
       if len(_a			) > 0:
            _SCREAMING_SNAKE_CASE							   =args[0]
            _SCREAMING_SNAKE_CASE							   =args[1:]

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

       if audio is not None:
            _SCREAMING_SNAKE_CASE							   =self.feature_extractor(_a							, *_a							, sampling_rate=_a							, **_a			)
       if text is not None:
            _SCREAMING_SNAKE_CASE							   =self.tokenizer(_a							, **_a			)

       if text is None:
            return inputs
       elif audio is None:
            return encodings
       else:
            _SCREAMING_SNAKE_CASE							   =encodings['input_ids']
            return inputs





  def 						A  (		self				:     Any							, *_a				:     List[str]							, **_a				:     Optional[int]			)     ->   Optional[int]:





       '''simple docstring'''






       return self.tokenizer.batch_decode(*_a							, **_a			)





  def 						A  (		self				:     Dict							, *_a				:     Tuple							, **_a				:     Dict			)     ->   List[str]:





       '''simple docstring'''






       if self._in_target_context_manager:
            return self.current_processor.pad(*_a							, **_a			)

       _SCREAMING_SNAKE_CASE							   =kwargs.pop('input_features'							, _a			)
       _SCREAMING_SNAKE_CASE							   =kwargs.pop('labels'							, _a			)
       if len(_a			) > 0:
            _SCREAMING_SNAKE_CASE							   =args[0]
            _SCREAMING_SNAKE_CASE							   =args[1:]

       if input_features is not None:
            _SCREAMING_SNAKE_CASE							   =self.feature_extractor.pad(_a							, *_a							, **_a			)
       if labels is not None:
            _SCREAMING_SNAKE_CASE							   =self.tokenizer.pad(_a							, **_a			)

       if labels is None:
            return input_features
       elif input_features is None:
            return labels
       else:
            _SCREAMING_SNAKE_CASE							   =labels['input_ids']
            return input_features





  def 						A  (		self				:     Tuple							, *_a				:     Dict							, **_a				:     List[Any]			)     ->   Optional[int]:





       '''simple docstring'''






       return self.tokenizer.decode(*_a							, **_a			)





  @contextmanager
  def 						A  (		self				:     Optional[Any]			)     ->   Dict:





       '''simple docstring'''






       warnings.warn(
           '`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '
           'labels by using the argument `text` of the regular `__call__` method (either in the same call as '
           'your audio inputs, or in a separate call.'			)
       _SCREAMING_SNAKE_CASE							   =True
       _SCREAMING_SNAKE_CASE							   =self.tokenizer
       yield
       _SCREAMING_SNAKE_CASE							   =self.feature_extractor
       _SCREAMING_SNAKE_CASE							   =False
