
'''simple docstring'''







from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType






class  lowerCAmelCase__   (  lowerCamelCase_     ):
					lowerCAmelCase_        =     ['''image_processor''', '''tokenizer''']
					lowerCAmelCase_        =     '''BlipImageProcessor'''
					lowerCAmelCase_        =     ('''BertTokenizer''', '''BertTokenizerFast''')






					def __init__(   self				,							__SCREAMING_SNAKE_CASE				,							__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											lowercase_	:     Union[str, Any]													=       False
											super().__init__(__SCREAMING_SNAKE_CASE				,							__SCREAMING_SNAKE_CASE		)
											lowercase_	:     Tuple													=       self.image_processor






					def __call__(   self				,							__SCREAMING_SNAKE_CASE = None				,							__SCREAMING_SNAKE_CASE = None				,							__SCREAMING_SNAKE_CASE = True				,							__SCREAMING_SNAKE_CASE = False				,							__SCREAMING_SNAKE_CASE = None				,							__SCREAMING_SNAKE_CASE = None				,							__SCREAMING_SNAKE_CASE = 0				,							__SCREAMING_SNAKE_CASE = None				,							__SCREAMING_SNAKE_CASE = None				,							__SCREAMING_SNAKE_CASE = False				,							__SCREAMING_SNAKE_CASE = False				,							__SCREAMING_SNAKE_CASE = False				,							__SCREAMING_SNAKE_CASE = False				,							__SCREAMING_SNAKE_CASE = False				,							__SCREAMING_SNAKE_CASE = True				,							__SCREAMING_SNAKE_CASE = None				,							**__SCREAMING_SNAKE_CASE				,							):


											"""simple docstring"""





											if images is None and text is None:
																	raise ValueError('''You have to specify either images or text.'''		)

											# Get only text
											if images is None:
																	lowercase_	:     Dict													=       self.tokenizer
																	lowercase_	:     List[str]													=       self.tokenizer(
																	    text=__SCREAMING_SNAKE_CASE				,							add_special_tokens=__SCREAMING_SNAKE_CASE				,							padding=__SCREAMING_SNAKE_CASE				,							truncation=__SCREAMING_SNAKE_CASE				,							max_length=__SCREAMING_SNAKE_CASE				,							stride=__SCREAMING_SNAKE_CASE				,							pad_to_multiple_of=__SCREAMING_SNAKE_CASE				,							return_attention_mask=__SCREAMING_SNAKE_CASE				,							return_overflowing_tokens=__SCREAMING_SNAKE_CASE				,							return_special_tokens_mask=__SCREAMING_SNAKE_CASE				,							return_offsets_mapping=__SCREAMING_SNAKE_CASE				,							return_token_type_ids=__SCREAMING_SNAKE_CASE				,							return_length=__SCREAMING_SNAKE_CASE				,							verbose=__SCREAMING_SNAKE_CASE				,							return_tensors=__SCREAMING_SNAKE_CASE				,							**__SCREAMING_SNAKE_CASE				,							)
																	return text_encoding

											# add pixel_values
											lowercase_	:     Dict													=       self.image_processor(__SCREAMING_SNAKE_CASE				,							return_tensors=__SCREAMING_SNAKE_CASE		)

											if text is not None:
																	lowercase_	:     int													=       self.tokenizer(
																	    text=__SCREAMING_SNAKE_CASE				,							add_special_tokens=__SCREAMING_SNAKE_CASE				,							padding=__SCREAMING_SNAKE_CASE				,							truncation=__SCREAMING_SNAKE_CASE				,							max_length=__SCREAMING_SNAKE_CASE				,							stride=__SCREAMING_SNAKE_CASE				,							pad_to_multiple_of=__SCREAMING_SNAKE_CASE				,							return_attention_mask=__SCREAMING_SNAKE_CASE				,							return_overflowing_tokens=__SCREAMING_SNAKE_CASE				,							return_special_tokens_mask=__SCREAMING_SNAKE_CASE				,							return_offsets_mapping=__SCREAMING_SNAKE_CASE				,							return_token_type_ids=__SCREAMING_SNAKE_CASE				,							return_length=__SCREAMING_SNAKE_CASE				,							verbose=__SCREAMING_SNAKE_CASE				,							return_tensors=__SCREAMING_SNAKE_CASE				,							**__SCREAMING_SNAKE_CASE				,							)
											else:
																	lowercase_	:     Any													=       None

											if text_encoding is not None:
																	encoding_image_processor.update(__SCREAMING_SNAKE_CASE		)

											return encoding_image_processor






					def 	_snake_case    (   self				,							*__SCREAMING_SNAKE_CASE				,							**__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											return self.tokenizer.batch_decode(*__SCREAMING_SNAKE_CASE				,							**__SCREAMING_SNAKE_CASE		)






					def 	_snake_case    (   self				,							*__SCREAMING_SNAKE_CASE				,							**__SCREAMING_SNAKE_CASE		):


											"""simple docstring"""





											return self.tokenizer.decode(*__SCREAMING_SNAKE_CASE				,							**__SCREAMING_SNAKE_CASE		)






					@property
					def 	_snake_case    (   self		):


											"""simple docstring"""





											lowercase_	:     Optional[Any]													=       self.tokenizer.model_input_names
											lowercase_	:     str													=       self.image_processor.model_input_names
											return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names		)		)





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""": 1_600, """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""": 1_600, """eval_accuracy""": 0.3, """eval_loss""": 1.2},
        },
    ]			)
class    lowerCAmelCase__     (	unittest.TestCase			):






		def  __A       (		self		:							Optional[int]  )     ->							List[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=SCREAMING_SNAKE_CASE__							,			)
						assert hasattr(self							,			'''env'''  )






		def  __A       (		self		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							Optional[Any]:
						# configuration for running training on smdistributed Model Parallel
						__lowerCamelCase    					=	{
						    '''enabled''': True,
						    '''processes_per_host''': 8,
						}
						__lowerCamelCase    					=	{
						    '''enabled''': True,
						    '''parameters''': {
						        '''microbatches''': 4,
						        '''placement_strategy''': '''spread''',
						        '''pipeline''': '''interleaved''',
						        '''optimize''': '''speed''',
						        '''partitions''': 4,
						        '''ddp''': True,
						    },
						}

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

						__lowerCamelCase    					=	'''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=SCREAMING_SNAKE_CASE__							,			instance_type=self.instance_type							,			debugger_hook_config=SCREAMING_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=SCREAMING_SNAKE_CASE__							,			py_version='''py36'''							,			)






		def  __A       (		self		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							List[str]:
						TrainingJobAnalytics(SCREAMING_SNAKE_CASE__  ).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''  )







		@parameterized.expand([(1,)]  )
		def  __A       (		self		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Optional[int]:
						# create estimator
						__lowerCamelCase    					=	self.create_estimator(SCREAMING_SNAKE_CASE__  )

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



