
"""simple docstring"""






import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
				from transformers.trainer import DEFAULT_CALLBACKS

				from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel




class 			_SCREAMING_SNAKE_CASE					(			A__							):






			def __init__(					self       )			->     Tuple:
						lowerCAmelCase_    :Dict			     =  []






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     List[str]:
						self.events.append("""on_init_end"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Any:
						self.events.append("""on_train_begin"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Tuple:
						self.events.append("""on_train_end"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Any:
						self.events.append("""on_epoch_begin"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Optional[Any]:
						self.events.append("""on_epoch_end"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Optional[Any]:
						self.events.append("""on_step_begin"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Any:
						self.events.append("""on_step_end"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Any:
						self.events.append("""on_evaluate"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     int:
						self.events.append("""on_predict"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Union[str, Any]:
						self.events.append("""on_save"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Dict:
						self.events.append("""on_log"""       )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       ,   __A       ,   **__A       )			->     Optional[Any]:
						self.events.append("""on_prediction_step"""       )




@require_torch
class 			_SCREAMING_SNAKE_CASE					(			unittest.TestCase							):






			def  __lowerCAmelCase (					self       )			->     Tuple:
						lowerCAmelCase_    :Optional[int]			     =  tempfile.mkdtemp()






			def  __lowerCAmelCase (					self       )			->     Union[str, Any]:
						shutil.rmtree(self.output_dir       )






			def  __lowerCAmelCase (					self       ,   __A=0       ,   __A=0       ,   __A=64       ,   __A=64       ,   __A=None       ,   __A=False       ,   **__A       )			->     Optional[int]:
						# disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
						# its set to False since the tests later on depend on its value.
						lowerCAmelCase_    :Optional[Any]			     =  RegressionDataset(length=__A       )
						lowerCAmelCase_    :str			     =  RegressionDataset(length=__A       )
						lowerCAmelCase_    :Dict			     =  RegressionModelConfig(a=__A       ,   b=__A       )
						lowerCAmelCase_    :Dict			     =  RegressionPreTrainedModel(__A       )

						lowerCAmelCase_    :Any			     =  TrainingArguments(self.output_dir       ,   disable_tqdm=__A       ,   report_to=[]       ,   **__A       )
						return Trainer(
						    __A       ,   __A       ,   train_dataset=__A       ,   eval_dataset=__A       ,   callbacks=__A       ,   )






			def  __lowerCAmelCase (					self       ,   __A       ,   __A       )			->     List[Any]:
						self.assertEqual(len(__A       )       ,   len(__A       )       )

						# Order doesn't matter
						lowerCAmelCase_    :Optional[int]			     =  sorted(__A       ,   key=lambda __A       : cb.__name__ if isinstance(__A       ,   __A       ) else cb.__class__.__name__       )
						lowerCAmelCase_    :List[Any]			     =  sorted(__A       ,   key=lambda __A       : cb.__name__ if isinstance(__A       ,   __A       ) else cb.__class__.__name__       )

						for cba, cba in zip(__A       ,   __A       ):
									if isinstance(__A       ,   __A       ) and isinstance(__A       ,   __A       ):
												self.assertEqual(__A       ,   __A       )
									elif isinstance(__A       ,   __A       ) and not isinstance(__A       ,   __A       ):
												self.assertEqual(__A       ,   cba.__class__       )
									elif not isinstance(__A       ,   __A       ) and isinstance(__A       ,   __A       ):
												self.assertEqual(cba.__class__       ,   __A       )
									else:
												self.assertEqual(__A       ,   __A       )






			def  __lowerCAmelCase (					self       ,   __A       )			->     int:
						lowerCAmelCase_    :Optional[int]			     =  ["""on_init_end""", """on_train_begin"""]
						lowerCAmelCase_    :Any			     =  0
						lowerCAmelCase_    :Optional[Any]			     =  len(trainer.get_eval_dataloader()       )
						lowerCAmelCase_    :List[Any]			     =  ["""on_prediction_step"""] * len(trainer.get_eval_dataloader()       ) + ["""on_log""", """on_evaluate"""]
						for _ in range(trainer.state.num_train_epochs       ):
									expected_events.append("""on_epoch_begin"""       )
									for _ in range(__A       ):
												step += 1
												expected_events += ["on_step_begin", "on_step_end"]
												if step % trainer.args.logging_steps == 0:
															expected_events.append("""on_log"""       )
												if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
															expected_events += evaluation_events.copy()
												if step % trainer.args.save_steps == 0:
															expected_events.append("""on_save"""       )
									expected_events.append("""on_epoch_end"""       )
									if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
												expected_events += evaluation_events.copy()
						expected_events += ["on_log", "on_train_end"]
						return expected_events






			def  __lowerCAmelCase (					self       )			->     Tuple:
						lowerCAmelCase_    :Union[str, Any]			     =  self.get_trainer()
						lowerCAmelCase_    :Optional[int]			     =  DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )

						# Callbacks passed at init are added to the default callbacks
						lowerCAmelCase_    :Optional[Any]			     =  self.get_trainer(callbacks=[MyTestTrainerCallback]       )
						expected_callbacks.append(__A       )
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )

						# TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
						lowerCAmelCase_    :List[Any]			     =  self.get_trainer(disable_tqdm=__A       )
						lowerCAmelCase_    :Tuple			     =  DEFAULT_CALLBACKS.copy() + [PrinterCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )






			def  __lowerCAmelCase (					self       )			->     Tuple:
						lowerCAmelCase_    :Dict			     =  DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						lowerCAmelCase_    :Optional[int]			     =  self.get_trainer()

						# We can add, pop, or remove by class name
						trainer.remove_callback(__A       )
						expected_callbacks.remove(__A       )
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )

						lowerCAmelCase_    :int			     =  self.get_trainer()
						lowerCAmelCase_    :Tuple			     =  trainer.pop_callback(__A       )
						self.assertEqual(cb.__class__       ,   __A       )
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )

						trainer.add_callback(__A       )
						expected_callbacks.insert(0       ,   __A       )
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )

						# We can also add, pop, or remove by instance
						lowerCAmelCase_    :str			     =  self.get_trainer()
						lowerCAmelCase_    :Optional[int]			     =  trainer.callback_handler.callbacks[0]
						trainer.remove_callback(__A       )
						expected_callbacks.remove(__A       )
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )

						lowerCAmelCase_    :int			     =  self.get_trainer()
						lowerCAmelCase_    :Tuple			     =  trainer.callback_handler.callbacks[0]
						lowerCAmelCase_    :Dict			     =  trainer.pop_callback(__A       )
						self.assertEqual(__A       ,   __A       )
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )

						trainer.add_callback(__A       )
						expected_callbacks.insert(0       ,   __A       )
						self.check_callbacks_equality(trainer.callback_handler.callbacks       ,   __A       )






			def  __lowerCAmelCase (					self       )			->     Optional[Any]:
						import warnings

						# XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
						warnings.simplefilter(action="""ignore"""       ,   category=__A       )

						lowerCAmelCase_    :Tuple			     =  self.get_trainer(callbacks=[MyTestTrainerCallback]       )
						trainer.train()
						lowerCAmelCase_    :Tuple			     =  trainer.callback_handler.callbacks[-2].events
						self.assertEqual(__A       ,   self.get_expected_events(__A       )       )

						# Independent log/save/eval
						lowerCAmelCase_    :Union[str, Any]			     =  self.get_trainer(callbacks=[MyTestTrainerCallback]       ,   logging_steps=5       )
						trainer.train()
						lowerCAmelCase_    :Optional[int]			     =  trainer.callback_handler.callbacks[-2].events
						self.assertEqual(__A       ,   self.get_expected_events(__A       )       )

						lowerCAmelCase_    :Optional[Any]			     =  self.get_trainer(callbacks=[MyTestTrainerCallback]       ,   save_steps=5       )
						trainer.train()
						lowerCAmelCase_    :Dict			     =  trainer.callback_handler.callbacks[-2].events
						self.assertEqual(__A       ,   self.get_expected_events(__A       )       )

						lowerCAmelCase_    :Any			     =  self.get_trainer(callbacks=[MyTestTrainerCallback]       ,   eval_steps=5       ,   evaluation_strategy="""steps"""       )
						trainer.train()
						lowerCAmelCase_    :List[Any]			     =  trainer.callback_handler.callbacks[-2].events
						self.assertEqual(__A       ,   self.get_expected_events(__A       )       )

						lowerCAmelCase_    :Optional[Any]			     =  self.get_trainer(callbacks=[MyTestTrainerCallback]       ,   evaluation_strategy="""epoch"""       )
						trainer.train()
						lowerCAmelCase_    :Optional[Any]			     =  trainer.callback_handler.callbacks[-2].events
						self.assertEqual(__A       ,   self.get_expected_events(__A       )       )

						# A bit of everything
						lowerCAmelCase_    :Any			     =  self.get_trainer(
						    callbacks=[MyTestTrainerCallback]       ,   logging_steps=3       ,   save_steps=10       ,   eval_steps=5       ,   evaluation_strategy="""steps"""       ,   )
						trainer.train()
						lowerCAmelCase_    :Union[str, Any]			     =  trainer.callback_handler.callbacks[-2].events
						self.assertEqual(__A       ,   self.get_expected_events(__A       )       )

						# warning should be emitted for duplicated callbacks
						with patch("""transformers.trainer_callback.logger.warning"""       ) as warn_mock:
									lowerCAmelCase_    :Optional[int]			     =  self.get_trainer(
									    callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]       ,   )
									assert str(__A       ) in warn_mock.call_args[0][0]











"""simple docstring"""







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_   						=   logging.get_logger(__name__)
@add_end_docstrings(A_    )
class        __A       (	A_    ):







		'''simple docstring'''





		def __init__(	self						:  List[str]			,**_snake_case						:  Dict						)					->		List[Any]:






				"""simple docstring"""



				super().__init__(**_snake_case						)

				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[int]			,_snake_case						:  Union[str, List[str], "Image", List["Image"]]			,**_snake_case						:  int						)					->		Optional[Any]:






				"""simple docstring"""



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







		def 					UpperCAmelCase    (	self						:  Dict			,**_snake_case						:  Optional[int]						)					->		List[Any]:






				"""simple docstring"""



				lowercase__   :    List[str]				=			{}
				if "candidate_labels" in kwargs:
						lowercase__   :    Any				=			kwargs['''candidate_labels''']
				if "hypothesis_template" in kwargs:
						lowercase__   :    Optional[Any]				=			kwargs['''hypothesis_template''']

				return preprocess_params, {}, {}







		def 					UpperCAmelCase    (	self						:  Optional[Any]			,_snake_case						:  Optional[int]			,_snake_case						:  Dict=None			,_snake_case						:  Union[str, Any]="This is a photo of {}."						)					->		List[str]:






				"""simple docstring"""



				lowercase__   :    List[Any]				=			load_image(_snake_case						)
				lowercase__   :    int				=			self.image_processor(images=[image]			,return_tensors=self.framework						)
				lowercase__   :    str				=			candidate_labels
				lowercase__   :    Dict				=			[hypothesis_template.format(_snake_case						) for x in candidate_labels]
				lowercase__   :    Any				=			self.tokenizer(_snake_case			,return_tensors=self.framework			,padding=_snake_case						)
				lowercase__   :    Optional[int]				=			[text_inputs]
				return inputs







		def 					UpperCAmelCase    (	self						:  Optional[Any]			,_snake_case						:  Optional[int]						)					->		List[Any]:






				"""simple docstring"""



				lowercase__   :    Optional[int]				=			model_inputs.pop('''candidate_labels'''						)
				lowercase__   :    Union[str, Any]				=			model_inputs.pop('''text_inputs'''						)
				if isinstance(text_inputs[0]			,_snake_case						):
						lowercase__   :    List[str]				=			text_inputs[0]
				else:
						# Batching case.
						lowercase__   :    int				=			text_inputs[0][0]

				lowercase__   :    Tuple				=			self.model(**_snake_case			,**_snake_case						)

				lowercase__   :    Union[str, Any]				=			{
				    '''candidate_labels''': candidate_labels,
				    '''logits''': outputs.logits_per_image,
				}
				return model_outputs







		def 					UpperCAmelCase    (	self						:  Any			,_snake_case						:  Tuple						)					->		Any:






				"""simple docstring"""



				lowercase__   :    Dict				=			model_outputs.pop('''candidate_labels'''						)
				lowercase__   :    Optional[Any]				=			model_outputs['''logits'''][0]
				if self.framework == "pt":
						lowercase__   :    Optional[int]				=			logits.softmax(dim=-1						).squeeze(-1						)
						lowercase__   :    Tuple				=			probs.tolist()
						if not isinstance(_snake_case			,_snake_case						):
								lowercase__   :    Any				=			[scores]
				elif self.framework == "tf":
						lowercase__   :    List[str]				=			stable_softmax(_snake_case			,axis=-1						)
						lowercase__   :    Optional[Any]				=			probs.numpy().tolist()
				else:
						raise ValueError(f"""Unsupported framework: {self.framework}"""						)

				lowercase__   :    Union[str, Any]				=			[
				    {'''score''': score, '''label''': candidate_label}
				    for score, candidate_label in sorted(zip(_snake_case			,_snake_case						)			,key=lambda _snake_case						: -x[0]						)
				]
				return result
