





'''simple docstring'''




from __future__ import annotations

import time
from collections.abc import Sequence
from random import randint

from matplotlib import pyplot as plt
def 	UpperCAmelCase_							(					__lowercase							:      Sequence[float]			,     __lowercase							:      int			,     __lowercase							:      int       )		->       tuple[int | None, int | None, float]:


     '''simple docstring'''



     if not arr:
          return None, None, 0
     if low == high:
          return low, high, arr[low]

     _UpperCAmelCase     							=   (low + high) // 2
     _UpperCAmelCase							,  _UpperCAmelCase							,  _UpperCAmelCase     							=   max_subarray(__lowercase			,     __lowercase			,     __lowercase       )
     _UpperCAmelCase							,  _UpperCAmelCase							,  _UpperCAmelCase     							=   max_subarray(__lowercase			,     mid + 1			,     __lowercase       )
     _UpperCAmelCase							,  _UpperCAmelCase							,  _UpperCAmelCase     							=   max_cross_sum(__lowercase			,     __lowercase			,     __lowercase			,     __lowercase       )
     if left_sum >= right_sum and left_sum >= cross_sum:
          return left_low, left_high, left_sum
     elif right_sum >= left_sum and right_sum >= cross_sum:
          return right_low, right_high, right_sum
     return cross_left, cross_right, cross_sum
def 	UpperCAmelCase_							(					__lowercase							:      Sequence[float]			,     __lowercase							:      int			,     __lowercase							:      int			,     __lowercase							:      int       )		->       tuple[int, int, float]:


     '''simple docstring'''



     _UpperCAmelCase							,  _UpperCAmelCase     							=   float("-inf"       ), -1
     _UpperCAmelCase							,  _UpperCAmelCase     							=   float("-inf"       ), -1

     _UpperCAmelCase     							=   0
     for i in range(__lowercase			,     low - 1			,     -1       ):
          summ += arr[i]
          if summ > left_sum:
               _UpperCAmelCase     							=   summ
               _UpperCAmelCase     							=   i

     _UpperCAmelCase     							=   0
     for i in range(mid + 1			,     high + 1       ):
          summ += arr[i]
          if summ > right_sum:
               _UpperCAmelCase     							=   summ
               _UpperCAmelCase     							=   i

     return max_left, max_right, (left_sum + right_sum)
def 	UpperCAmelCase_							(					__lowercase							:      int       )		->       float:


     '''simple docstring'''



     _UpperCAmelCase     							=   [randint(1			,     __lowercase       ) for _ in range(__lowercase       )]
     _UpperCAmelCase     							=   time.time()
     max_subarray(__lowercase			,     0			,     input_size - 1       )
     _UpperCAmelCase     							=   time.time()
     return end - start
def 	UpperCAmelCase_							(					)		->       None:


     '''simple docstring'''



     _UpperCAmelCase     							=   [10, 100, 1000, 1_0000, 5_0000, 10_0000, 20_0000, 30_0000, 40_0000, 50_0000]
     _UpperCAmelCase     							=   [time_max_subarray(__lowercase       ) for input_size in input_sizes]
     print("No of Inputs\t\tTime Taken"       )
     for input_size, runtime in zip(__lowercase			,     __lowercase       ):
          print(__lowercase			,     "\t\t"			,     __lowercase       )
     plt.plot(__lowercase			,     __lowercase       )
     plt.xlabel("Number of Inputs"       )
     plt.ylabel("Time taken in seconds"       )
     plt.show()


if __name__ == "__main__":
  from doctest import testmod

  testmod()






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




	'''simple docstring'''




	def    _lowerCamelCase				(					self		:List[Any]						)		-> Any:
								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    _lowerCamelCase				(					self		:Any     ,   a		:Optional[Any]						)		-> Dict:
								__UpperCamelCase	:   str			   =					f'{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'
								# distributed data settings
								__UpperCamelCase	:   Optional[int]			   =					{"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    _lowerCamelCase				(					self		:Dict     ,   a		:Dict						)		-> Optional[int]:
								TrainingJobAnalytics(a						).export_csv(f'{self.env.test_path}/{job_name}_metrics.csv'						)







	@parameterized.expand([(2,)]						)
	def    _lowerCamelCase				(					self		:Dict     ,   a		:Tuple						)		-> List[Any]:
								# create estimator
								__UpperCamelCase	:   int			   =					self.create_estimator(a						)

								# run training
								estimator.fit()

								# result dataframe
								__UpperCamelCase	:   Optional[int]			   =					TrainingJobAnalytics(estimator.latest_training_job.name						).dataframe()

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