

import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
     import torch_xla.core.xla_model as xm
     import torch_xla.debug.metrics as met



class 						_A     (				__lowerCAmelCase				):





  '''simple docstring'''



  def __init__(  self   ,*SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_=None   ,SCREAMING_SNAKE_CASE_=None   ,**SCREAMING_SNAKE_CASE_    ):

      '''simple docstring'''
      super().__init__(*__SCREAMING_SNAKE_CASE   ,**__SCREAMING_SNAKE_CASE    )
      snake_case							:		List[Any]        =		eval_examples
      snake_case							:		Optional[int]        =		post_process_function





  def    snake_case_       (  self   ,SCREAMING_SNAKE_CASE_ = None   ,SCREAMING_SNAKE_CASE_=None   ,SCREAMING_SNAKE_CASE_ = None   ,SCREAMING_SNAKE_CASE_ = "eval"   ,**SCREAMING_SNAKE_CASE_   ,):

      '''simple docstring'''
      snake_case							:		Dict        =		gen_kwargs.copy()
      snake_case							:		Optional[Any]        =		(
          gen_kwargs["""max_length"""] if gen_kwargs.get("""max_length"""    ) is not None else self.args.generation_max_length
      )
      snake_case							:		Union[str, Any]        =		(
          gen_kwargs["""num_beams"""] if gen_kwargs.get("""num_beams"""    ) is not None else self.args.generation_num_beams
      )
      snake_case							:		Union[str, Any]        =		gen_kwargs

      snake_case							:		Optional[int]        =		self.eval_dataset if eval_dataset is None else eval_dataset
      snake_case							:		Any        =		self.get_eval_dataloader(__SCREAMING_SNAKE_CASE    )
      snake_case							:		Union[str, Any]        =		self.eval_examples if eval_examples is None else eval_examples

      # Temporarily disable metric computation, we will do it in the loop here.
      snake_case							:		str        =		self.compute_metrics
      snake_case							:		Tuple        =		None
      snake_case							:		Optional[Any]        =		time.time()
      snake_case							:		List[Any]        =		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
      try:
          snake_case							:		Tuple        =		eval_loop(
              __SCREAMING_SNAKE_CASE   ,description="""Evaluation"""   ,prediction_loss_only=True if compute_metrics is None else None   ,ignore_keys=__SCREAMING_SNAKE_CASE   ,metric_key_prefix=__SCREAMING_SNAKE_CASE   ,)
      finally:
          snake_case							:		List[str]        =		compute_metrics
      snake_case							:		Tuple        =		self.args.eval_batch_size * self.args.world_size
      if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
          start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
      output.metrics.update(
          speed_metrics(
              __SCREAMING_SNAKE_CASE   ,__SCREAMING_SNAKE_CASE   ,num_samples=output.num_samples   ,num_steps=math.ceil(output.num_samples / total_batch_size    )   ,)    )

      if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
          # Only the main node write the results by default
          snake_case							:		Optional[int]        =		self.post_process_function(__SCREAMING_SNAKE_CASE   ,__SCREAMING_SNAKE_CASE   ,__SCREAMING_SNAKE_CASE    )
          snake_case							:		Optional[Any]        =		self.compute_metrics(__SCREAMING_SNAKE_CASE    )

          # Prefix all keys with metric_key_prefix + '_'
          for key in list(metrics.keys()    ):
              if not key.startswith(F"""{metric_key_prefix}_"""    ):
                  snake_case							:		Union[str, Any]        =		metrics.pop(__SCREAMING_SNAKE_CASE    )

          metrics.update(output.metrics    )
      else:
          snake_case							:		Optional[Any]        =		output.metrics

      if self.args.should_log:
          # Only the main node log the results by default
          self.log(__SCREAMING_SNAKE_CASE    )

      if self.args.tpu_metrics_debug or self.args.debug:
          # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
          xm.master_print(met.metrics_report()    )

      snake_case							:		Optional[int]        =		self.callback_handler.on_evaluate(self.args   ,self.state   ,self.control   ,__SCREAMING_SNAKE_CASE    )
      return metrics





  def    snake_case_       (  self   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_=None   ,SCREAMING_SNAKE_CASE_ = "test"   ,**SCREAMING_SNAKE_CASE_    ):

      '''simple docstring'''
      snake_case							:		List[Any]        =		gen_kwargs.copy()

      snake_case							:		List[Any]        =		self.get_test_dataloader(__SCREAMING_SNAKE_CASE    )

      # Temporarily disable metric computation, we will do it in the loop here.
      snake_case							:		Optional[Any]        =		self.compute_metrics
      snake_case							:		Optional[Any]        =		None
      snake_case							:		str        =		time.time()
      snake_case							:		List[Any]        =		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
      try:
          snake_case							:		str        =		eval_loop(
              __SCREAMING_SNAKE_CASE   ,description="""Prediction"""   ,prediction_loss_only=True if compute_metrics is None else None   ,ignore_keys=__SCREAMING_SNAKE_CASE   ,metric_key_prefix=__SCREAMING_SNAKE_CASE   ,)
      finally:
          snake_case							:		Optional[int]        =		compute_metrics

      snake_case							:		Union[str, Any]        =		self.args.eval_batch_size * self.args.world_size
      if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
          start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
      output.metrics.update(
          speed_metrics(
              __SCREAMING_SNAKE_CASE   ,__SCREAMING_SNAKE_CASE   ,num_samples=output.num_samples   ,num_steps=math.ceil(output.num_samples / total_batch_size    )   ,)    )
      if self.post_process_function is None or self.compute_metrics is None:
          return output

      snake_case							:		Dict        =		self.post_process_function(__SCREAMING_SNAKE_CASE   ,__SCREAMING_SNAKE_CASE   ,__SCREAMING_SNAKE_CASE   ,"""predict"""    )
      snake_case							:		List[Any]        =		self.compute_metrics(__SCREAMING_SNAKE_CASE    )

      # Prefix all keys with metric_key_prefix + '_'
      for key in list(metrics.keys()    ):
          if not key.startswith(F"""{metric_key_prefix}_"""    ):
              snake_case							:		int        =		metrics.pop(__SCREAMING_SNAKE_CASE    )
      metrics.update(output.metrics    )
      return PredictionOutput(predictions=predictions.predictions   ,label_ids=predictions.label_ids   ,metrics=__SCREAMING_SNAKE_CASE    )









'''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

UpperCAmelCase_							:	str	     =		logging.get_logger(__name__)





@add_end_docstrings(__lowerCAmelCase)
class      lowerCAmelCase (			__lowerCAmelCase):


       def __init__( self     ,	**__SCREAMING_SNAKE_CASE	)   ->	Optional[int]:



            '''simple docstring'''




            super().__init__(**__SCREAMING_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     ,	__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE	)   ->	List[Any]:



            '''simple docstring'''




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


       def    lowerCAmelCase   ( self     ,	**__SCREAMING_SNAKE_CASE	)   ->	Tuple:



            '''simple docstring'''




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

            return preprocess_params, {}, {}


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None     ,	__SCREAMING_SNAKE_CASE="This is a photo of {}."	)   ->	Optional[Any]:



            '''simple docstring'''




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


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE	)   ->	Dict:



            '''simple docstring'''




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

            __snake_case            =	self.model(**__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE	)

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




       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE	)   ->	str:



            '''simple docstring'''




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

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






