






"""simple docstring"""




import enum
import os
from hashlib import shaaaa
from typing import Optional

from .. import config
from .logging import get_logger


UpperCamelCase							       =    get_logger(__name__)




class    lowercase_    (enum.Enum     ):
      A__	:		Optional[int]       			=				'''all_checks'''
      A__	:		List[Any]       			=				'''basic_checks'''
      A__	:		Union[str, Any]       			=				'''no_checks'''







class    lowercase_    (_UpperCAmelCase     ):
      pass







class    lowercase_    (_UpperCAmelCase     ):
      pass







class    lowercase_    (_UpperCAmelCase     ):
      pass







class    lowercase_    (_UpperCAmelCase     ):
      pass







def        lowerCAmelCase (      UpperCamelCase_:							Optional[dict]     , UpperCamelCase_:							dict     , UpperCamelCase_:							List[str]=None					)       ->							Any:







  '''simple docstring'''







  if expected_checksums is None:
    logger.info("Unable to verify checksums."					)
    return
  if len(set(UpperCamelCase_					) - set(UpperCamelCase_					)					) > 0:
    raise ExpectedMoreDownloadedFiles(str(set(UpperCamelCase_					) - set(UpperCamelCase_					)					)					)
  if len(set(UpperCamelCase_					) - set(UpperCamelCase_					)					) > 0:
    raise UnexpectedDownloadedFile(str(set(UpperCamelCase_					) - set(UpperCamelCase_					)					)					)
  _a							=	[url for url in expected_checksums if expected_checksums[url] != recorded_checksums[url]]
  _a							=	" for " + verification_name if verification_name is not None else ""
  if len(UpperCamelCase_					) > 0:
    raise NonMatchingChecksumError(
        f'''Checksums didn\'t match{for_verification_name}:\n'''
        f'''{bad_urls}\n'''
        "Set `verification_mode='no_checks'` to skip checksums verification and ignore this error"					)
  logger.info("All the checksums matched successfully" + for_verification_name					)




class    lowercase_    (_UpperCAmelCase     ):
      pass







class    lowercase_    (_UpperCAmelCase     ):
      pass







class    lowercase_    (_UpperCAmelCase     ):
      pass







class    lowercase_    (_UpperCAmelCase     ):
      pass







def        lowerCAmelCase (      UpperCamelCase_:							Optional[dict]     , UpperCamelCase_:							dict					)       ->							Optional[int]:







  '''simple docstring'''







  if expected_splits is None:
    logger.info("Unable to verify splits sizes."					)
    return
  if len(set(UpperCamelCase_					) - set(UpperCamelCase_					)					) > 0:
    raise ExpectedMoreSplits(str(set(UpperCamelCase_					) - set(UpperCamelCase_					)					)					)
  if len(set(UpperCamelCase_					) - set(UpperCamelCase_					)					) > 0:
    raise UnexpectedSplits(str(set(UpperCamelCase_					) - set(UpperCamelCase_					)					)					)
  _a							=	[
      {"expected": expected_splits[name], "recorded": recorded_splits[name]}
      for name in expected_splits
      if expected_splits[name].num_examples != recorded_splits[name].num_examples
  ]
  if len(UpperCamelCase_					) > 0:
    raise NonMatchingSplitsSizesError(str(UpperCamelCase_					)					)
  logger.info("All the splits matched successfully."					)






def        lowerCAmelCase (      UpperCamelCase_:							str     , UpperCamelCase_:							bool = True					)       ->							dict:







  '''simple docstring'''







  if record_checksum:
    _a							=	shaaaa()
    with open(UpperCamelCase_     , "rb"					) as f:
      for chunk in iter(lambda: f.read(1 << 20					)     , b""					):
        m.update(UpperCamelCase_					)
      _a							=	m.hexdigest()
  else:
    _a							=	None
  return {"num_bytes": os.path.getsize(UpperCamelCase_					), "checksum": checksum}






def        lowerCAmelCase (      UpperCamelCase_:							Union[str, Any]					)       ->							int:







  '''simple docstring'''







  if dataset_size and config.IN_MEMORY_MAX_SIZE:
    return dataset_size < config.IN_MEMORY_MAX_SIZE
  else:
    return False










"""simple docstring"""




import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import 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.g4dn.xlarge''',
            '''results''': {'''train_runtime''': 650, '''eval_accuracy''': 0.6, '''eval_loss''': 0.9},
        },
        {
            '''framework''': '''tensorflow''',
            '''script''': '''run_tf.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.g4dn.xlarge''',
            '''results''': {'''train_runtime''': 600, '''eval_accuracy''': 0.3, '''eval_loss''': 0.9},
        },
    ]     )
class    lowercase_    (unittest.TestCase     ):






      def 			lowerCamelCase__    (    self	)	->Union[str, 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 			lowerCamelCase__    (    self     ,     a_=1	)	->int:







        '''simple docstring'''


        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}-single'''     ,     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}     ,     metric_definitions=self.env.metric_definitions     ,     py_version="py36"     ,     )






      def 			lowerCamelCase__    (    self     ,     a_	)	->str:







        '''simple docstring'''


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






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        _a							=	self.create_estimator()

        # run training
        estimator.fit()

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

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


