



import unittest

from transformers import (
    MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
    TextClassificationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_tf, require_torch, slow

from .test_pipelines_common import ANY


# These 2 model types require different inputs than those of the usual text models.
A__  :  Any								=					{'''LayoutLMv2Config''', '''LayoutLMv3Config'''}



@is_pipeline_test
class 				__snake_case						(     unittest.TestCase						):
							_a			      = MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
							_a			      = TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

							if model_mapping is not None:
														_a			      = {config: model for config, model in model_mapping.items() if config.__name__ not in _TO_SKIP}
							if tf_model_mapping is not None:
														_a			      = {
														    config: model for config, model in tf_model_mapping.items() if config.__name__ not in _TO_SKIP
														}




							@require_torch
							def    UpperCAmelCase__			(       self     :       List[Any]):
									lowerCAmelCase_			:     Dict									= pipeline(
									    task='''text-classification'''				,				model='''hf-internal-testing/tiny-random-distilbert'''				,				framework='''pt''')

									lowerCAmelCase_			:     Union[str, Any]									= text_classifier('''This is great !''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''LABEL_0''', '''score''': 0.504}])

									lowerCAmelCase_			:     Union[str, Any]									= text_classifier('''This is great !'''				,				top_k=2)
									self.assertEqual(
									    nested_simplify(A_)				,				[{'''label''': '''LABEL_0''', '''score''': 0.504}, {'''label''': '''LABEL_1''', '''score''': 0.496}])

									lowerCAmelCase_			:     int									= text_classifier(['''This is great !''', '''This is bad''']				,				top_k=2)
									self.assertEqual(
									    nested_simplify(A_)				,				[
									        [{'''label''': '''LABEL_0''', '''score''': 0.504}, {'''label''': '''LABEL_1''', '''score''': 0.496}],
									        [{'''label''': '''LABEL_0''', '''score''': 0.504}, {'''label''': '''LABEL_1''', '''score''': 0.496}],
									    ]				,				)

									lowerCAmelCase_			:     str									= text_classifier('''This is great !'''				,				top_k=1)
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''LABEL_0''', '''score''': 0.504}])

									# Legacy behavior
									lowerCAmelCase_			:     int									= text_classifier('''This is great !'''				,				return_all_scores=A_)
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''LABEL_0''', '''score''': 0.504}])

									lowerCAmelCase_			:     List[Any]									= text_classifier('''This is great !'''				,				return_all_scores=A_)
									self.assertEqual(
									    nested_simplify(A_)				,				[[{'''label''': '''LABEL_0''', '''score''': 0.504}, {'''label''': '''LABEL_1''', '''score''': 0.496}]])

									lowerCAmelCase_			:     Any									= text_classifier(['''This is great !''', '''Something else''']				,				return_all_scores=A_)
									self.assertEqual(
									    nested_simplify(A_)				,				[
									        [{'''label''': '''LABEL_0''', '''score''': 0.504}, {'''label''': '''LABEL_1''', '''score''': 0.496}],
									        [{'''label''': '''LABEL_0''', '''score''': 0.504}, {'''label''': '''LABEL_1''', '''score''': 0.496}],
									    ]				,				)

									lowerCAmelCase_			:     Optional[Any]									= text_classifier(['''This is great !''', '''Something else''']				,				return_all_scores=A_)
									self.assertEqual(
									    nested_simplify(A_)				,				[
									        {'''label''': '''LABEL_0''', '''score''': 0.504},
									        {'''label''': '''LABEL_0''', '''score''': 0.504},
									    ]				,				)




							@require_torch
							def    UpperCAmelCase__			(       self     :       Any):
									import torch

									lowerCAmelCase_			:     Optional[int]									= pipeline(
									    task='''text-classification'''				,				model='''hf-internal-testing/tiny-random-distilbert'''				,				framework='''pt'''				,				device=torch.device('''cpu''')				,				)

									lowerCAmelCase_			:     str									= text_classifier('''This is great !''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''LABEL_0''', '''score''': 0.504}])




							@require_tf
							def    UpperCAmelCase__			(       self     :       int):
									lowerCAmelCase_			:     int									= pipeline(
									    task='''text-classification'''				,				model='''hf-internal-testing/tiny-random-distilbert'''				,				framework='''tf''')

									lowerCAmelCase_			:     int									= text_classifier('''This is great !''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''LABEL_0''', '''score''': 0.504}])




							@slow
							@require_torch
							def    UpperCAmelCase__			(       self     :       Tuple):
									lowerCAmelCase_			:     Dict									= pipeline('''text-classification''')

									lowerCAmelCase_			:     Union[str, Any]									= text_classifier('''This is great !''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''POSITIVE''', '''score''': 1.0}])
									lowerCAmelCase_			:     Tuple									= text_classifier('''This is bad !''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''NEGATIVE''', '''score''': 1.0}])
									lowerCAmelCase_			:     Dict									= text_classifier('''Birds are a type of animal''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''POSITIVE''', '''score''': 0.988}])




							@slow
							@require_tf
							def    UpperCAmelCase__			(       self     :       List[str]):
									lowerCAmelCase_			:     Any									= pipeline('''text-classification'''				,				framework='''tf''')

									lowerCAmelCase_			:     Tuple									= text_classifier('''This is great !''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''POSITIVE''', '''score''': 1.0}])
									lowerCAmelCase_			:     str									= text_classifier('''This is bad !''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''NEGATIVE''', '''score''': 1.0}])
									lowerCAmelCase_			:     Any									= text_classifier('''Birds are a type of animal''')
									self.assertEqual(nested_simplify(A_)				,				[{'''label''': '''POSITIVE''', '''score''': 0.988}])




							def    UpperCAmelCase__			(       self     :       List[str]				,				A_     :       Union[str, Any]				,				A_     :       Any				,				A_     :       Tuple):
									lowerCAmelCase_			:     str									= TextClassificationPipeline(model=A_				,				tokenizer=A_)
									return text_classifier, ["HuggingFace is in", "This is another test"]







							def    UpperCAmelCase__			(       self     :       Any				,				A_     :       str				,				A_     :       Dict):
									lowerCAmelCase_			:     Optional[Any]									= text_classifier.model
									# Small inputs because BartTokenizer tiny has maximum position embeddings = 22
									lowerCAmelCase_			:     int									= '''HuggingFace is in'''
									lowerCAmelCase_			:     List[str]									= text_classifier(A_)

									self.assertEqual(nested_simplify(A_)				,				[{'''label''': ANY(A_), '''score''': ANY(A_)}])
									self.assertTrue(outputs[0]['''label'''] in model.config.idalabel.values())

									lowerCAmelCase_			:     Any									= ['''HuggingFace is in ''', '''Paris is in France''']
									lowerCAmelCase_			:     List[Any]									= text_classifier(A_)
									self.assertEqual(
									    nested_simplify(A_)				,				[{'''label''': ANY(A_), '''score''': ANY(A_)}, {'''label''': ANY(A_), '''score''': ANY(A_)}]				,				)
									self.assertTrue(outputs[0]['''label'''] in model.config.idalabel.values())
									self.assertTrue(outputs[1]['''label'''] in model.config.idalabel.values())

									# Forcing to get all results with `top_k=None`
									# This is NOT the legacy format
									lowerCAmelCase_			:     Dict									= text_classifier(A_				,				top_k=A_)
									lowerCAmelCase_			:     Any									= len(model.config.idalabel.values())
									self.assertEqual(
									    nested_simplify(A_)				,				[[{'''label''': ANY(A_), '''score''': ANY(A_)}] * N, [{'''label''': ANY(A_), '''score''': ANY(A_)}] * N]				,				)

									lowerCAmelCase_			:     List[Any]									= {'''text''': '''HuggingFace is in ''', '''text_pair''': '''Paris is in France'''}
									lowerCAmelCase_			:     Tuple									= text_classifier(A_)
									self.assertEqual(
									    nested_simplify(A_)				,				{'''label''': ANY(A_), '''score''': ANY(A_)}				,				)
									self.assertTrue(outputs['''label'''] in model.config.idalabel.values())

									# This might be used a text pair, but tokenizer + pipe interaction
									# makes it hard to understand that it's not using the pair properly
									# https://github.com/huggingface/transformers/issues/17305
									# We disabled this usage instead as it was outputting wrong outputs.
									lowerCAmelCase_			:     Tuple									= [['''HuggingFace is in ''', '''Paris is in France''']]
									with self.assertRaises(A_):
											text_classifier(A_)

									# This used to be valid for doing text pairs
									# We're keeping it working because of backward compatibility
									lowerCAmelCase_			:     Optional[Any]									= text_classifier([[['''HuggingFace is in ''', '''Paris is in France''']]])
									self.assertEqual(
									    nested_simplify(A_)				,				[{'''label''': ANY(A_), '''score''': ANY(A_)}]				,				)
									self.assertTrue(outputs[0]['''label'''] in model.config.idalabel.values())










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




							def    UpperCAmelCase__			(       self     :       int):
									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    UpperCAmelCase__			(       self     :       Union[str, Any]				,				A_     :       str=1):
									# 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}-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    UpperCAmelCase__			(       self     :       List[str]				,				A_     :       Optional[Any]):
									TrainingJobAnalytics(A_).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv""")







							def    UpperCAmelCase__			(       self     :       int):
									# create estimator
									lowerCAmelCase_			:     List[Any]									= self.create_estimator()

									# run training
									estimator.fit()

									# result dataframe
									lowerCAmelCase_			:     Union[str, Any]									= TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()

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





