



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": 16_00, "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": 16_00, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
    ]   )
class   __lowerCAmelCase  (		unittest.TestCase   ):
					"""simple docstring"""






					def      _a				(  self       :							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=_snake_case					,       )
								assert hasattr(self					,       'env'       )



					def      _a				(  self       :							Optional[int]					,       _snake_case       :							int       ):






								"""simple docstring"""


								A__	  =     {
								    'enabled': True,
								    'processes_per_host': 8,
								}
								A__	  =     {
								    'enabled': True,
								    'parameters': {
								        'microbatches': 4,
								        'placement_strategy': 'spread',
								        'pipeline': 'interleaved',
								        'optimize': 'speed',
								        'partitions': 4,
								        'ddp': True,
								    },
								}

								A__	  =     {'smdistributed': {'modelparallel': smp_options}, 'mpi': mpi_options}

								A__	  =     '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=_snake_case					,       instance_type=self.instance_type					,       debugger_hook_config=_snake_case					,       hyperparameters={
								        **self.env.hyperparameters,
								        'model_name_or_path': self.model_name_or_path,
								        'max_steps': 5_00,
								    }					,       metric_definitions=self.env.metric_definitions					,       distribution=_snake_case					,       py_version='py36'					,       )



					def      _a				(  self       :							List[str]					,       _snake_case       :							Optional[Any]       ):






								"""simple docstring"""


								TrainingJobAnalytics(_snake_case       ).export_csv(F'''{self.env.test_path}/{job_name}_metrics.csv'''       )



					@parameterized.expand([(1,)]       )
					def      _a				(  self       :							Dict					,       _snake_case       :							Optional[int]       ):






								"""simple docstring"""


								A__	  =     self.create_estimator(_snake_case       )

								# 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'					,       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}					,       _snake_case       )











"""simple docstring"""



from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    convert_to_rgb,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


a_					=     logging.get_logger(__name__)


if is_vision_available():
       import PIL






class 						__snake_case   (						SCREAMING_SNAKE_CASE__	):





 """simple docstring"""


 _lowerCamelCase								=	["""pixel_values"""]

 def __init__(       self   ,		__lowerCamelCase = True   ,		__lowerCamelCase = None   ,		__lowerCamelCase = PILImageResampling.BICUBIC   ,		__lowerCamelCase = True   ,		__lowerCamelCase = None   ,		__lowerCamelCase = True   ,		__lowerCamelCase = 1 / 255   ,		__lowerCamelCase = True   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = True   ,		**__lowerCamelCase   ,		):



     '''simple docstring'''




     super().__init__(**__lowerCamelCase    )
     __A       :		Optional[int]							=	size if size is not None else {'''shortest_edge''': 224}
     __A       :		Optional[int]							=	get_size_dict(__lowerCamelCase   ,		default_to_square=__lowerCamelCase    )
     __A       :		Dict							=	crop_size if crop_size is not None else {'''height''': 224, '''width''': 224}
     __A       :		str							=	get_size_dict(__lowerCamelCase   ,		default_to_square=__lowerCamelCase   ,		param_name='''crop_size'''    )

     __A       :		Dict							=	do_resize
     __A       :		Optional[Any]							=	size
     __A       :		Optional[Any]							=	resample
     __A       :		Dict							=	do_center_crop
     __A       :		Any							=	crop_size
     __A       :		Optional[Any]							=	do_rescale
     __A       :		List[Any]							=	rescale_factor
     __A       :		Union[str, Any]							=	do_normalize
     __A       :		Any							=	image_mean if image_mean is not None else OPENAI_CLIP_MEAN
     __A       :		Optional[int]							=	image_std if image_std is not None else OPENAI_CLIP_STD
     __A       :		Dict							=	do_convert_rgb

 def    UpperCamelCase__(       self   ,		__lowerCamelCase   ,		__lowerCamelCase   ,		__lowerCamelCase = PILImageResampling.BICUBIC   ,		__lowerCamelCase = None   ,		**__lowerCamelCase   ,		):



     '''simple docstring'''




     __A       :		Optional[Any]							=	get_size_dict(__lowerCamelCase   ,		default_to_square=__lowerCamelCase    )
     if "shortest_edge" not in size:
         raise ValueError(F"""The `size` parameter must contain the key `shortest_edge`. Got {size.keys()}"""    )
     __A       :		Optional[int]							=	get_resize_output_image_size(__lowerCamelCase   ,		size=size['''shortest_edge''']   ,		default_to_square=__lowerCamelCase    )
     return resize(__lowerCamelCase   ,		size=__lowerCamelCase   ,		resample=__lowerCamelCase   ,		data_format=__lowerCamelCase   ,		**__lowerCamelCase    )

 def    UpperCamelCase__(       self   ,		__lowerCamelCase   ,		__lowerCamelCase   ,		__lowerCamelCase = None   ,		**__lowerCamelCase   ,		):



     '''simple docstring'''




     __A       :		Optional[Any]							=	get_size_dict(__lowerCamelCase    )
     if "height" not in size or "width" not in size:
         raise ValueError(F"""The `size` parameter must contain the keys (height, width). Got {size.keys()}"""    )
     return center_crop(__lowerCamelCase   ,		size=(size['''height'''], size['''width'''])   ,		data_format=__lowerCamelCase   ,		**__lowerCamelCase    )

 def    UpperCamelCase__(       self   ,		__lowerCamelCase   ,		__lowerCamelCase   ,		__lowerCamelCase = None   ,		**__lowerCamelCase   ,		):



     '''simple docstring'''




     return rescale(__lowerCamelCase   ,		scale=__lowerCamelCase   ,		data_format=__lowerCamelCase   ,		**__lowerCamelCase    )

 def    UpperCamelCase__(       self   ,		__lowerCamelCase   ,		__lowerCamelCase   ,		__lowerCamelCase   ,		__lowerCamelCase = None   ,		**__lowerCamelCase   ,		):



     '''simple docstring'''




     return normalize(__lowerCamelCase   ,		mean=__lowerCamelCase   ,		std=__lowerCamelCase   ,		data_format=__lowerCamelCase   ,		**__lowerCamelCase    )







 def    UpperCamelCase__(       self   ,		__lowerCamelCase   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = None   ,		__lowerCamelCase = ChannelDimension.FIRST   ,		**__lowerCamelCase   ,		):



     '''simple docstring'''




     __A       :		Tuple							=	do_resize if do_resize is not None else self.do_resize
     __A       :		Any							=	size if size is not None else self.size
     __A       :		List[str]							=	get_size_dict(__lowerCamelCase   ,		param_name='''size'''   ,		default_to_square=__lowerCamelCase    )
     __A       :		List[Any]							=	resample if resample is not None else self.resample
     __A       :		Any							=	do_center_crop if do_center_crop is not None else self.do_center_crop
     __A       :		Dict							=	crop_size if crop_size is not None else self.crop_size
     __A       :		List[Any]							=	get_size_dict(__lowerCamelCase   ,		param_name='''crop_size'''   ,		default_to_square=__lowerCamelCase    )
     __A       :		List[str]							=	do_rescale if do_rescale is not None else self.do_rescale
     __A       :		List[Any]							=	rescale_factor if rescale_factor is not None else self.rescale_factor
     __A       :		Tuple							=	do_normalize if do_normalize is not None else self.do_normalize
     __A       :		str							=	image_mean if image_mean is not None else self.image_mean
     __A       :		int							=	image_std if image_std is not None else self.image_std
     __A       :		Union[str, Any]							=	do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

     __A       :		int							=	make_list_of_images(__lowerCamelCase    )

     if not valid_images(__lowerCamelCase    ):
         raise ValueError(
             '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
             '''torch.Tensor, tf.Tensor or jax.ndarray.'''    )

     if do_resize and size is None:
         raise ValueError('''Size must be specified if do_resize is True.'''    )

     if do_center_crop and crop_size is None:
         raise ValueError('''Crop size must be specified if do_center_crop is True.'''    )

     if do_rescale and rescale_factor is None:
         raise ValueError('''Rescale factor must be specified if do_rescale is True.'''    )

     if do_normalize and (image_mean is None or image_std is None):
         raise ValueError('''Image mean and std must be specified if do_normalize is True.'''    )

     # PIL RGBA images are converted to RGB
     if do_convert_rgb:
         __A       :		List[str]							=	[convert_to_rgb(__lowerCamelCase    ) for image in images]

     # All transformations expect numpy arrays.
     __A       :		Optional[int]							=	[to_numpy_array(__lowerCamelCase    ) for image in images]

     if do_resize:
         __A       :		List[str]							=	[self.resize(image=__lowerCamelCase   ,		size=__lowerCamelCase   ,		resample=__lowerCamelCase    ) for image in images]

     if do_center_crop:
         __A       :		Dict							=	[self.center_crop(image=__lowerCamelCase   ,		size=__lowerCamelCase    ) for image in images]

     if do_rescale:
         __A       :		str							=	[self.rescale(image=__lowerCamelCase   ,		scale=__lowerCamelCase    ) for image in images]

     if do_normalize:
         __A       :		Optional[Any]							=	[self.normalize(image=__lowerCamelCase   ,		mean=__lowerCamelCase   ,		std=__lowerCamelCase    ) for image in images]

     __A       :		str							=	[to_channel_dimension_format(__lowerCamelCase   ,		__lowerCamelCase    ) for image in images]

     __A       :		Union[str, Any]							=	{'''pixel_values''': images}
     return BatchFeature(data=__lowerCamelCase   ,		tensor_type=__lowerCamelCase    )




