






'''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							):







   def     SCREAMING_SNAKE_CASE__					(self)			->      Dict:
      """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     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      List[str]:
      """simple docstring"""






      __snake_case				:      str     							= F"""{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}"""
      # distributed data settings
      __snake_case				:      Dict     							= {'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=_lowercase					,   instance_count=_lowercase					,   instance_type=self.instance_type					,   debugger_hook_config=_lowercase					,   hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path}					,   metric_definitions=self.env.metric_definitions					,   distribution=_lowercase					,   py_version='py36'					,   )







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      Union[str, Any]:
      """simple docstring"""






      TrainingJobAnalytics(_lowercase).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv""")







   @parameterized.expand([(2,)])
   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      str:
      """simple docstring"""






      __snake_case				:      Optional[int]     							= self.create_estimator(_lowercase)

      # run training
      estimator.fit()

      # result dataframe
      __snake_case				:      Optional[Any]     							= TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()

      # extract kpis
      __snake_case				:      Tuple     							= list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value'])
      __snake_case				:      int     							= 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				:      int     							= (
          Session().describe_training_job(estimator.latest_training_job.name).get('TrainingTimeInSeconds'					,   9_9_9_9_9_9)
      )

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







'''simple docstring'''






def        _SCREAMING_SNAKE_CASE  (				A  :       float  ,		A  :       list[float]  )    -> float:







   """simple docstring"""
   if discount_rate < 0:
      raise ValueError('Discount rate cannot be negative'  )
   if not cash_flows:
      raise ValueError('Cash flows list cannot be empty'  )
   __snake_case				:      List[str]     							= sum(
       cash_flow / ((1 + discount_rate) ** i) for i, cash_flow in enumerate(A  )  )
   return round(A  ,		ndigits=2  )


if __name__ == "__main__":
      import doctest

      doctest.testmod()