



'''simple docstring'''





from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
 import tensorflow as tf


_lowerCAmelCase      						=	logging.get_logger(__name__)






@dataclass
class        lowerCAmelCase_(				SCREAMING_SNAKE_CASE_     ):
  '''simple docstring'''




  __lowercase							:      Any				=      [
      '''no_inference''',
      '''no_cuda''',
      '''no_tpu''',
      '''no_speed''',
      '''no_memory''',
      '''no_env_print''',
      '''no_multi_process''',
  ]




  def __init__(    self  ,**__UpperCAmelCase      )    ->		int:
      for deprecated_arg in self.deprecated_args:
          if deprecated_arg in kwargs:
              lowerCAmelCase__ :  str							      =					deprecated_arg[3:]
              lowerCAmelCase__ :  Union[str, Any]							      =					not kwargs.pop(__UpperCAmelCase      )
              logger.warning(
                  F"""{deprecated_arg} is depreciated. Please use --no-{positive_arg} or"""
                  F""" {positive_arg}={kwargs[positive_arg]}"""      )
      lowerCAmelCase__ :  Union[str, Any]							      =					kwargs.pop("""tpu_name"""  ,self.tpu_name      )
      lowerCAmelCase__ :  str							      =					kwargs.pop("""device_idx"""  ,self.device_idx      )
      lowerCAmelCase__ :  int							      =					kwargs.pop("""eager_mode"""  ,self.eager_mode      )
      lowerCAmelCase__ :  int							      =					kwargs.pop("""use_xla"""  ,self.use_xla      )
      super().__init__(**__UpperCAmelCase      )

  __lowercase							:      str				=      field(
      default=SCREAMING_SNAKE_CASE_						, metadata={'''help''': '''Name of TPU'''}						, )
  __lowercase							:      int				=      field(
      default=0						, metadata={'''help''': '''CPU / GPU device index. Defaults to 0.'''}						, )
  __lowercase							:      bool				=      field(default=SCREAMING_SNAKE_CASE_						, metadata={'''help''': '''Benchmark models in eager model.'''}     )
  __lowercase							:      bool				=      field(
      default=SCREAMING_SNAKE_CASE_						, metadata={
          '''help''': '''Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`.'''
      }						, )




  @cached_property
  def 							UpperCAmelCase_	(    self      )    ->		Tuple["tf.distribute.cluster_resolver.TPUClusterResolver"]:
      requires_backends(self  ,["""tf"""]      )
      lowerCAmelCase__ :  Any							      =					None
      if self.tpu:
          try:
              if self.tpu_name:
                  lowerCAmelCase__ :  List[Any]							      =					tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name      )
              else:
                  lowerCAmelCase__ :  Any							      =					tf.distribute.cluster_resolver.TPUClusterResolver()
          except ValueError:
              lowerCAmelCase__ :  Optional[Any]							      =					None
      return tpu




  @cached_property
  def 							UpperCAmelCase_	(    self      )    ->		Tuple["tf.distribute.Strategy", "tf.distribute.cluster_resolver.TPUClusterResolver"]:
      requires_backends(self  ,["""tf"""]      )
      if self.is_tpu:
          tf.config.experimental_connect_to_cluster(self._setup_tpu      )
          tf.tpu.experimental.initialize_tpu_system(self._setup_tpu      )

          lowerCAmelCase__ :  Any							      =					tf.distribute.TPUStrategy(self._setup_tpu      )
      else:
          # currently no multi gpu is allowed
          if self.is_gpu:
              # TODO: Currently only single GPU is supported
              tf.config.set_visible_devices(self.gpu_list[self.device_idx]  ,"""GPU"""      )
              lowerCAmelCase__ :  str							      =					tf.distribute.OneDeviceStrategy(device=F"""/gpu:{self.device_idx}"""      )
          else:
              tf.config.set_visible_devices([]  ,"""GPU"""      )  # disable GPU
              lowerCAmelCase__ :  str							      =					tf.distribute.OneDeviceStrategy(device=F"""/cpu:{self.device_idx}"""      )

      return strategy




  @property
  def 							UpperCAmelCase_	(    self      )    ->		bool:
      requires_backends(self  ,["""tf"""]      )
      return self._setup_tpu is not None




  @property
  def 							UpperCAmelCase_	(    self      )    ->		"tf.distribute.Strategy":
      requires_backends(self  ,["""tf"""]      )
      return self._setup_strategy




  @property
  def 							UpperCAmelCase_	(    self      )    ->		str:
      requires_backends(self  ,["""tf"""]      )
      return tf.config.list_physical_devices("""GPU"""      )




  @property
  def 							UpperCAmelCase_	(    self      )    ->		int:
      requires_backends(self  ,["""tf"""]      )
      if self.cuda:
          return len(self.gpu_list      )
      return 0






  @property
  def 							UpperCAmelCase_	(    self      )    ->		bool:
      return self.n_gpu > 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(SCREAMING_SNAKE_CASE_     )
class        lowerCAmelCase_(				SCREAMING_SNAKE_CASE_     ):
  '''simple docstring'''




  def __init__(    self  ,**__UpperCAmelCase      )    ->		Tuple:
      super().__init__(**__UpperCAmelCase      )

      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  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		str:
      return super().__call__(__UpperCAmelCase  ,**__UpperCAmelCase      )




  def 							UpperCAmelCase_	(    self  ,**__UpperCAmelCase      )    ->		str:
      lowerCAmelCase__ :  List[Any]							      =					{}
      if "candidate_labels" in kwargs:
          lowerCAmelCase__ :  int							      =					kwargs["""candidate_labels"""]
      if "hypothesis_template" in kwargs:
          lowerCAmelCase__ :  Optional[int]							      =					kwargs["""hypothesis_template"""]

      return preprocess_params, {}, {}




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase=None  ,__UpperCAmelCase="This is a photo of {}."      )    ->		int:
      lowerCAmelCase__ :  str							      =					load_image(__UpperCAmelCase      )
      lowerCAmelCase__ :  Dict							      =					self.image_processor(images=[image]  ,return_tensors=self.framework      )
      lowerCAmelCase__ :  List[Any]							      =					candidate_labels
      lowerCAmelCase__ :  List[str]							      =					[hypothesis_template.format(__UpperCAmelCase      ) for x in candidate_labels]
      lowerCAmelCase__ :  Optional[Any]							      =					self.tokenizer(__UpperCAmelCase  ,return_tensors=self.framework  ,padding=__UpperCAmelCase      )
      lowerCAmelCase__ :  Tuple							      =					[text_inputs]
      return inputs




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Union[str, Any]:
      lowerCAmelCase__ :  Tuple							      =					model_inputs.pop("""candidate_labels"""      )
      lowerCAmelCase__ :  Union[str, Any]							      =					model_inputs.pop("""text_inputs"""      )
      if isinstance(text_inputs[0]  ,__UpperCAmelCase      ):
          lowerCAmelCase__ :  int							      =					text_inputs[0]
      else:
          # Batching case.
          lowerCAmelCase__ :  Dict							      =					text_inputs[0][0]

      lowerCAmelCase__ :  Any							      =					self.model(**__UpperCAmelCase  ,**__UpperCAmelCase      )

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






  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Any:
      lowerCAmelCase__ :  Union[str, Any]							      =					model_outputs.pop("""candidate_labels"""      )
      lowerCAmelCase__ :  List[str]							      =					model_outputs["""logits"""][0]
      if self.framework == "pt":
          lowerCAmelCase__ :  List[str]							      =					logits.softmax(dim=-1      ).squeeze(-1      )
          lowerCAmelCase__ :  Optional[Any]							      =					probs.tolist()
          if not isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ):
              lowerCAmelCase__ :  Dict							      =					[scores]
      elif self.framework == "tf":
          lowerCAmelCase__ :  Any							      =					stable_softmax(__UpperCAmelCase  ,axis=-1      )
          lowerCAmelCase__ :  List[Any]							      =					probs.numpy().tolist()
      else:
          raise ValueError(F"""Unsupported framework: {self.framework}"""      )

      lowerCAmelCase__ :  Tuple							      =					[
          {"""score""": score, """label""": candidate_label}
          for score, candidate_label in sorted(zip(__UpperCAmelCase  ,__UpperCAmelCase      )  ,key=lambda __UpperCAmelCase      : -x[0]      )
      ]
      return result


