





from typing import List, Union

import numpy as np

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, logging
from .base import PIPELINE_INIT_ARGS, ArgumentHandler, ChunkPipeline


_snake_case    :       Dict		       =    logging.get_logger(__name__)



class 							_UpperCAmelCase       (       UpperCAmelCase__       ):

  """simple docstring"""


  def 							lowercase						(							self					:						Optional[Any]   ,    lowerCAmelCase_					:						Optional[int]	)				->		Optional[Any]:






     if isinstance(__lowercase   ,    __lowercase	):
        __lowerCAmelCase				      =      [label.strip() for label in labels.split(','	) if label.strip()]
     return labels






  def __call__(							self					:						Optional[Any]   ,    lowerCAmelCase_					:						Optional[Any]   ,    lowerCAmelCase_					:						Any   ,    lowerCAmelCase_					:						str	)				->		str:






     if len(__lowercase	) == 0 or len(__lowercase	) == 0:
        raise ValueError('You must include at least one label and at least one sequence.'	)
     if hypothesis_template.format(labels[0]	) == hypothesis_template:
        raise ValueError(
            (
                'The provided hypothesis_template \"{}\" was not able to be formatted with the target labels. '
                'Make sure the passed template includes formatting syntax such as {{}} where the label should go.'
            ).format(__lowercase	)	)

     if isinstance(__lowercase   ,    __lowercase	):
        __lowerCAmelCase				      =      [sequences]

     __lowerCAmelCase				      =      []
     for sequence in sequences:
        sequence_pairs.extend([[sequence, hypothesis_template.format(__lowercase	)] for label in labels]	)

     return sequence_pairs, sequences





@add_end_docstrings(UpperCAmelCase__       )
class 							_UpperCAmelCase       (       UpperCAmelCase__       ):

  """simple docstring"""


  def __init__(							self					:						int   ,    lowerCAmelCase_					:						Any=ZeroShotClassificationArgumentHandler()   ,    *lowerCAmelCase_					:						Tuple   ,    **lowerCAmelCase_					:						Any	)				->		str:
     __lowerCAmelCase				      =      args_parser
     super().__init__(*__lowercase   ,    **__lowercase	)
     if self.entailment_id == -1:
        logger.warning(
            'Failed to determine \'entailment\' label id from the label2id mapping in the model config. Setting to '
            '-1. Define a descriptive label2id mapping in the model config to ensure correct outputs.'	)






  @property
  def 							lowercase						(							self					:						List[str]	)				->		List[str]:






     for label, ind in self.model.config.labelaid.items():
        if label.lower().startswith('entail'	):
           return ind
     return -1






  def 							lowercase						(							self					:						Optional[Any]   ,    lowerCAmelCase_					:						Optional[int]   ,    lowerCAmelCase_					:						Optional[Any]=True   ,    lowerCAmelCase_					:						Dict=True   ,    lowerCAmelCase_					:						Optional[int]=TruncationStrategy.ONLY_FIRST   ,    **lowerCAmelCase_					:						Dict	)				->		Union[str, Any]:
     __lowerCAmelCase				      =      self.framework
     if self.tokenizer.pad_token is None:
        # Override for tokenizers not supporting padding
        logger.error(
            'Tokenizer was not supporting padding necessary for zero-shot, attempting to use '
            ' `pad_token=eos_token`'	)
        __lowerCAmelCase				      =      self.tokenizer.eos_token
     try:
        __lowerCAmelCase				      =      self.tokenizer(
            __lowercase   ,    add_special_tokens=__lowercase   ,    return_tensors=__lowercase   ,    padding=__lowercase   ,    truncation=__lowercase   ,    )
     except Exception as e:
        if "too short" in str(__lowercase	):
           # tokenizers might yell that we want to truncate
           # to a value that is not even reached by the input.
           # In that case we don't want to truncate.
           # It seems there's not a really better way to catch that
           # exception.

           __lowerCAmelCase				      =      self.tokenizer(
               __lowercase   ,    add_special_tokens=__lowercase   ,    return_tensors=__lowercase   ,    padding=__lowercase   ,    truncation=TruncationStrategy.DO_NOT_TRUNCATE   ,    )
        else:
           raise e

     return inputs






  def 							lowercase						(							self					:						Union[str, Any]   ,    **lowerCAmelCase_					:						Tuple	)				->		List[str]:






     if kwargs.get('multi_class'   ,    __lowercase	) is not None:
        __lowerCAmelCase				      =      kwargs['multi_class']
        logger.warning(
            'The `multi_class` argument has been deprecated and renamed to `multi_label`. '
            '`multi_class` will be removed in a future version of Transformers.'	)
     __lowerCAmelCase				      =      {}
     if "candidate_labels" in kwargs:
        __lowerCAmelCase				      =      self._args_parser._parse_labels(kwargs['candidate_labels']	)
     if "hypothesis_template" in kwargs:
        __lowerCAmelCase				      =      kwargs['hypothesis_template']

     __lowerCAmelCase				      =      {}
     if "multi_label" in kwargs:
        __lowerCAmelCase				      =      kwargs['multi_label']
     return preprocess_params, {}, postprocess_params






  def __call__(							self					:						Tuple   ,    lowerCAmelCase_					:						Union[str, List[str]]   ,    *lowerCAmelCase_					:						int   ,    **lowerCAmelCase_					:						List[Any]   ,    )				->		Dict:






     if len(__lowercase	) == 0:
        pass
     elif len(__lowercase	) == 1 and "candidate_labels" not in kwargs:
        __lowerCAmelCase				      =      args[0]
     else:
        raise ValueError(f"""Unable to understand extra arguments {args}"""	)

     return super().__call__(__lowercase   ,    **__lowercase	)






  def 							lowercase						(							self					:						Any   ,    lowerCAmelCase_					:						Optional[Any]   ,    lowerCAmelCase_					:						Optional[Any]=None   ,    lowerCAmelCase_					:						Optional[Any]="This example is {}."	)				->		Any:
     __lowerCAmelCase      ,      __lowerCAmelCase				      =      self._args_parser(__lowercase   ,    __lowercase   ,    __lowercase	)

     for i, (candidate_label, sequence_pair) in enumerate(zip(__lowercase   ,    __lowercase	)	):
        __lowerCAmelCase				      =      self._parse_and_tokenize([sequence_pair]	)

        yield {
            "candidate_label": candidate_label,
            "sequence": sequences[0],
            "is_last": i == len(__lowercase	) - 1,
            **model_input,
        }






  def 							lowercase						(							self					:						Any   ,    lowerCAmelCase_					:						List[Any]	)				->		Union[str, Any]:
     __lowerCAmelCase				      =      inputs['candidate_label']
     __lowerCAmelCase				      =      inputs['sequence']
     __lowerCAmelCase				      =      {k: inputs[k] for k in self.tokenizer.model_input_names}
     __lowerCAmelCase				      =      self.model(**__lowercase	)

     __lowerCAmelCase				      =      {
         'candidate_label': candidate_label,
         'sequence': sequence,
         'is_last': inputs['is_last'],
         **outputs,
     }
     return model_outputs






  def 							lowercase						(							self					:						List[Any]   ,    lowerCAmelCase_					:						List[Any]   ,    lowerCAmelCase_					:						List[Any]=False	)				->		Optional[int]:
     __lowerCAmelCase				      =      [outputs['candidate_label'] for outputs in model_outputs]
     __lowerCAmelCase				      =      [outputs['sequence'] for outputs in model_outputs]
     __lowerCAmelCase				      =      np.concatenate([output['logits'].numpy() for output in model_outputs]	)
     __lowerCAmelCase				      =      logits.shape[0]
     __lowerCAmelCase				      =      len(__lowercase	)
     __lowerCAmelCase				      =      N // n
     __lowerCAmelCase				      =      logits.reshape((num_sequences, n, -1)	)

     if multi_label or len(__lowercase	) == 1:
        # softmax over the entailment vs. contradiction dim for each label independently
        __lowerCAmelCase				      =      self.entailment_id
        __lowerCAmelCase				      =      -1 if entailment_id == 0 else 0
        __lowerCAmelCase				      =      reshaped_outputs[..., [contradiction_id, entailment_id]]
        __lowerCAmelCase				      =      np.exp(__lowercase	) / np.exp(__lowercase	).sum(-1   ,    keepdims=__lowercase	)
        __lowerCAmelCase				      =      scores[..., 1]
     else:
        # softmax the "entailment" logits over all candidate labels
        __lowerCAmelCase				      =      reshaped_outputs[..., self.entailment_id]
        __lowerCAmelCase				      =      np.exp(__lowercase	) / np.exp(__lowercase	).sum(-1   ,    keepdims=__lowercase	)

     __lowerCAmelCase				      =      list(reversed(scores[0].argsort()	)	)
     return {
         "sequence": sequences[0],
         "labels": [candidate_labels[i] for i in top_inds],
         "scores": scores[0, top_inds].tolist(),
     }







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''': 1600, '''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''': 1600, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
    ]     )
class 						UpperCAmelCase  (							unittest.TestCase     ):







						'''simple docstring'''





						def 			snake_case__		(		self				:	int		):






							"""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=__lowercase			,      )
							assert hasattr(self			,      "env"		)






						def 			snake_case__		(		self				:	Any			,      __lowercase				:	Dict		):






							"""simple docstring"""






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

							snake_case_						       =	{"smdistributed": {"modelparallel": smp_options}, "mpi": mpi_options}

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






						def 			snake_case__		(		self				:	Tuple			,      __lowercase				:	Dict		):






							"""simple docstring"""






							TrainingJobAnalytics(__lowercase		).export_csv(f"{self.env.test_path}/{job_name}_metrics.csv"		)






						@parameterized.expand([(1,)]		)
						def 			snake_case__		(		self				:	Any			,      __lowercase				:	int		):






							"""simple docstring"""






							snake_case_						       =	self.create_estimator(__lowercase		)

							# run training
							estimator.fit()

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

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