




from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

lowerCamelCase							:						str					       =  logging.get_logger(__name__)







@add_end_docstrings(UpperCamelCase  )
class     A(		UpperCamelCase  ):



 '''simple docstring'''


 def __init__(							self   :    Optional[Any]		,				**A_   :    str    )  ->      Dict:

       """simple docstring"""






       super().__init__(**A_    )

       requires_backends(self		,				'vision'    )
       self.check_model_type(
           TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
           if self.framework == 'tf'
           else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING    )

 def __call__(							self   :    Optional[Any]		,				A_   :    Union[str, List[str], "Image", List["Image"]]		,				**A_   :    Union[str, Any]    )  ->      List[str]:

       """simple docstring"""






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

 def 							a__					(							self   :    Optional[int]		,				**A_   :    Tuple    )  ->      int:

       """simple docstring"""






       lowerCamelCase_							=       {}
       if "candidate_labels" in kwargs:
             lowerCamelCase_							=       kwargs['candidate_labels']
       if "hypothesis_template" in kwargs:
             lowerCamelCase_							=       kwargs['hypothesis_template']

       return preprocess_params, {}, {}

 def 							a__					(							self   :    Optional[Any]		,				A_   :    List[Any]		,				A_   :    str=None		,				A_   :    Tuple="This is a photo of {}."    )  ->      Tuple:

       """simple docstring"""






       lowerCamelCase_							=       load_image(A_    )
       lowerCamelCase_							=       self.image_processor(images=[image]		,				return_tensors=self.framework    )
       lowerCamelCase_							=       candidate_labels
       lowerCamelCase_							=       [hypothesis_template.format(A_    ) for x in candidate_labels]
       lowerCamelCase_							=       self.tokenizer(A_		,				return_tensors=self.framework		,				padding=A_    )
       lowerCamelCase_							=       [text_inputs]
       return inputs

 def 							a__					(							self   :    Any		,				A_   :    Optional[Any]    )  ->      List[str]:

       """simple docstring"""






       lowerCamelCase_							=       model_inputs.pop('candidate_labels'    )
       lowerCamelCase_							=       model_inputs.pop('text_inputs'    )
       if isinstance(text_inputs[0]		,				A_    ):
             lowerCamelCase_							=       text_inputs[0]
       else:
             # Batching case.
             lowerCamelCase_							=       text_inputs[0][0]

       lowerCamelCase_							=       self.model(**A_		,				**A_    )

       lowerCamelCase_							=       {
           'candidate_labels': candidate_labels,
           'logits': outputs.logits_per_image,
       }
       return model_outputs





 def 							a__					(							self   :    int		,				A_   :    Union[str, Any]    )  ->      str:

       """simple docstring"""






       lowerCamelCase_							=       model_outputs.pop('candidate_labels'    )
       lowerCamelCase_							=       model_outputs['logits'][0]
       if self.framework == "pt":
             lowerCamelCase_							=       logits.softmax(dim=-1    ).squeeze(-1    )
             lowerCamelCase_							=       probs.tolist()
             if not isinstance(A_		,				A_    ):
                   lowerCamelCase_							=       [scores]
       elif self.framework == "tf":
             lowerCamelCase_							=       stable_softmax(A_		,				axis=-1    )
             lowerCamelCase_							=       probs.numpy().tolist()
       else:
             raise ValueError(f"""Unsupported framework: {self.framework}"""    )

       lowerCamelCase_							=       [
           {'score': score, 'label': candidate_label}
           for score, candidate_label in sorted(zip(A_		,				A_    )		,				key=lambda A_    : -x[0]    )
       ]
       return result








'''simple docstring'''



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







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







						def        __lowercase     (	self :   Optional[int]					,     lowerCAmelCase__ :   List[Any]					):

													"""simple docstring"""
													# configuration for running training on smdistributed Model Parallel
													SCREAMING_SNAKE_CASE       :   Tuple         =      {
													    '''enabled''': True,
													    '''processes_per_host''': 8,
													}
													SCREAMING_SNAKE_CASE       :   Optional[int]         =      {
													    '''enabled''': True,
													    '''parameters''': {
													        '''microbatches''': 4,
													        '''placement_strategy''': '''spread''',
													        '''pipeline''': '''interleaved''',
													        '''optimize''': '''speed''',
													        '''partitions''': 4,
													        '''ddp''': True,
													    },
													}

													SCREAMING_SNAKE_CASE       :   str         =      {'''smdistributed''': {'''modelparallel''': smp_options}, '''mpi''': mpi_options}

													SCREAMING_SNAKE_CASE       :   Optional[Any]         =      '''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=lowerCAmelCase__					,     instance_type=self.instance_type					,     debugger_hook_config=lowerCAmelCase__					,     hyperparameters={
													        **self.env.hyperparameters,
													        '''model_name_or_path''': self.model_name_or_path,
													        '''max_steps''': 5_00,
													    }					,     metric_definitions=self.env.metric_definitions					,     distribution=lowerCAmelCase__					,     py_version='''py36'''					,     )







						def        __lowercase     (	self :   Dict					,     lowerCAmelCase__ :   int					):

													"""simple docstring"""
													TrainingJobAnalytics(lowerCAmelCase__					).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""					)







						@parameterized.expand([(1,)]					)
						def        __lowercase     (	self :   Dict					,     lowerCAmelCase__ :   List[Any]					):

													"""simple docstring"""
													# create estimator
													SCREAMING_SNAKE_CASE       :   Any         =      self.create_estimator(lowerCAmelCase__					)

													# run training
													estimator.fit()

													# result dataframe
													SCREAMING_SNAKE_CASE       :   Optional[Any]         =      TrainingJobAnalytics(estimator.latest_training_job.name					).dataframe()

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




