'''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": 6_5_0, "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": 6_0_0, "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": 6_0_0, "eval_accuracy": 0.6, "eval_loss": 0.7},
        },
    ]			)
class    a__				(       unittest.TestCase			):






     def        SCREAMING_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=a  ,    )
         assert hasattr(self  ,    '''env'''   )






     def        SCREAMING_SNAKE_CASE__       (			self		:		Optional[int]  ,    a		:		Tuple   ):





         """simple docstring"""



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






     def        SCREAMING_SNAKE_CASE__       (			self		:		Dict  ,    a		:		Any   ):





         """simple docstring"""



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






     @parameterized.expand([(2,)]   )
     def        SCREAMING_SNAKE_CASE__       (			self		:		Union[str, Any]  ,    a		:		Tuple   ):





         """simple docstring"""



         __lowerCamelCase          =  self.create_estimator(a   )

         # run training
         estimator.fit()

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

         # extract kpis
         __lowerCamelCase          =  list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value''']   )
         __lowerCamelCase          =  list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value''']   )
         # get train time from SageMaker job, this includes starting, preprocessing, stopping
         __lowerCamelCase          =  (
             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}  ,    a   )

'''simple docstring'''




def __lowerCAmelCase			(							UpperCamelCase__    ,  UpperCamelCase__    ,  UpperCamelCase__    )     ->   float:
    if principal <= 0:
        raise Exception('''Principal borrowed must be > 0'''    )
    if rate_per_annum < 0:
        raise Exception('''Rate of interest must be >= 0'''    )
    if years_to_repay <= 0 or not isinstance(UpperCamelCase__    ,  UpperCamelCase__    ):
        raise Exception('''Years to repay must be an integer > 0'''    )

    # Yearly rate is divided by 12 to get monthly rate
    __lowerCamelCase          =  rate_per_annum / 12

    # Years to repay is multiplied by 12 to get number of payments as payment is monthly
    __lowerCamelCase          =  years_to_repay * 12

    return (
        principal
        * rate_per_month
        * (1 + rate_per_month) ** number_of_payments
        / ((1 + rate_per_month) ** number_of_payments - 1)
    )


if __name__ == "__main__":
  import doctest

  doctest.testmod()
