

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


   def 						_lowerCamelCase       (			self  ):





      """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      ,    _UpperCamelCase  ):





      """simple docstring"""





      _lowercase      :				List[str]      					=       {
          "enabled": True,
          "processes_per_host": 8,
      }
      _lowercase      :				Tuple      					=       {
          "enabled": True,
          "parameters": {
              "microbatches": 4,
              "placement_strategy": "spread",
              "pipeline": "interleaved",
              "optimize": "speed",
              "partitions": 4,
              "ddp": True,
          },
      }

      _lowercase      :				Union[str, Any]      					=       {"smdistributed": {"modelparallel": smp_options}, "mpi": mpi_options}

      _lowercase      :				Optional[Any]      					=       "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=_UpperCamelCase      ,    instance_type=self.instance_type      ,    debugger_hook_config=_UpperCamelCase      ,    hyperparameters={
              **self.env.hyperparameters,
              "model_name_or_path": self.model_name_or_path,
              "max_steps": 500,
          }      ,    metric_definitions=self.env.metric_definitions      ,    distribution=_UpperCamelCase      ,    py_version="py36"      ,    )


   def 						_lowerCamelCase       (			self      ,    _UpperCamelCase  ):





      """simple docstring"""





      TrainingJobAnalytics(_UpperCamelCase  ).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''  )






   @parameterized.expand([(1,)]  )
   def 						_lowerCamelCase       (			self      ,    _UpperCamelCase  ):





      """simple docstring"""





      _lowercase      :				Union[str, Any]      					=       self.create_estimator(_UpperCamelCase  )

      # run training
      estimator.fit()

      # result dataframe
      _lowercase      :				str      					=       TrainingJobAnalytics(estimator.latest_training_job.name  ).dataframe()

      # extract kpis
      _lowercase      :				Optional[Any]      					=       list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]  )
      _lowercase      :				List[Any]      					=       list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]  )
      # get train time from SageMaker job, this includes starting, preprocessing, stopping
      _lowercase      :				Any      					=       (
          Session().describe_training_job(estimator.latest_training_job.name  ).get("TrainingTimeInSeconds"      ,    999999  )
      )

      # 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'''
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
from .tokenization_wavaveca import WavaVecaCTCTokenizer


class        a__  (							lowerCamelCase_  ):
   _SCREAMING_SNAKE_CASE   :					int     	=							'Wav2Vec2FeatureExtractor'
   _SCREAMING_SNAKE_CASE   :					List[str]     	=							'AutoTokenizer'


   def __init__(			self      ,    _UpperCamelCase      ,    _UpperCamelCase  ):





      """simple docstring"""





      super().__init__(_UpperCamelCase      ,    _UpperCamelCase  )
      _lowercase      :				List[Any]      					=       self.feature_extractor
      _lowercase      :				Optional[Any]      					=       False


   @classmethod
   def 						_lowerCamelCase       (			cls      ,    _UpperCamelCase      ,    **_UpperCamelCase  ):





      """simple docstring"""





      try:
         return super().from_pretrained(_UpperCamelCase      ,    **_UpperCamelCase  )
      except OSError:
         warnings.warn(
             f'''Loading a tokenizer inside {cls.__name__} from a config that does not'''
             " include a `tokenizer_class` attribute is deprecated and will be "
             "removed in v5. Please add `'tokenizer_class': 'Wav2Vec2CTCTokenizer'`"
             " attribute to either your `config.json` or `tokenizer_config.json` "
             "file to suppress this warning: "      ,    _UpperCamelCase      ,    )

         _lowercase      :				Dict      					=       WavaVecaFeatureExtractor.from_pretrained(_UpperCamelCase      ,    **_UpperCamelCase  )
         _lowercase      :				str      					=       WavaVecaCTCTokenizer.from_pretrained(_UpperCamelCase      ,    **_UpperCamelCase  )

         return cls(feature_extractor=_UpperCamelCase      ,    tokenizer=_UpperCamelCase  )


   def __call__(			self      ,    *_UpperCamelCase      ,    **_UpperCamelCase  ):





      """simple docstring"""





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

      if "raw_speech" in kwargs:
         warnings.warn("Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."  )
         _lowercase      :				int      					=       kwargs.pop("raw_speech"  )
      else:
         _lowercase      :				List[Any]      					=       kwargs.pop("audio"      ,    _UpperCamelCase  )
      _lowercase      :				List[Any]      					=       kwargs.pop("sampling_rate"      ,    _UpperCamelCase  )
      _lowercase      :				Union[str, Any]      					=       kwargs.pop("text"      ,    _UpperCamelCase  )
      if len(_UpperCamelCase  ) > 0:
         _lowercase      :				int      					=       args[0]
         _lowercase      :				Any      					=       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:
         _lowercase      :				Dict      					=       self.feature_extractor(_UpperCamelCase      ,    *_UpperCamelCase      ,    sampling_rate=_UpperCamelCase      ,    **_UpperCamelCase  )
      if text is not None:
         _lowercase      :				Union[str, Any]      					=       self.tokenizer(_UpperCamelCase      ,    **_UpperCamelCase  )

      if text is None:
         return inputs
      elif audio is None:
         return encodings
      else:
         _lowercase      :				int      					=       encodings["input_ids"]
         return inputs


   def 						_lowerCamelCase       (			self      ,    *_UpperCamelCase      ,    **_UpperCamelCase  ):





      """simple docstring"""





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

      _lowercase      :				List[Any]      					=       kwargs.pop("input_features"      ,    _UpperCamelCase  )
      _lowercase      :				Any      					=       kwargs.pop("labels"      ,    _UpperCamelCase  )
      if len(_UpperCamelCase  ) > 0:
         _lowercase      :				Any      					=       args[0]
         _lowercase      :				Any      					=       args[1:]

      if input_features is not None:
         _lowercase      :				Any      					=       self.feature_extractor.pad(_UpperCamelCase      ,    *_UpperCamelCase      ,    **_UpperCamelCase  )
      if labels is not None:
         _lowercase      :				int      					=       self.tokenizer.pad(_UpperCamelCase      ,    **_UpperCamelCase  )

      if labels is None:
         return input_features
      elif input_features is None:
         return labels
      else:
         _lowercase      :				Optional[Any]      					=       labels["input_ids"]
         return input_features


   def 						_lowerCamelCase       (			self      ,    *_UpperCamelCase      ,    **_UpperCamelCase  ):





      """simple docstring"""





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


   def 						_lowerCamelCase       (			self      ,    *_UpperCamelCase      ,    **_UpperCamelCase  ):





      """simple docstring"""





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






   @contextmanager
   def 						_lowerCamelCase       (			self  ):





      """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."  )
      _lowercase      :				Optional[Any]      					=       True
      _lowercase      :				Dict      					=       self.tokenizer
      yield
      _lowercase      :				List[str]      					=       self.feature_extractor
      _lowercase      :				List[str]      					=       False






