






'''simple docstring'''

from typing import List, Optional, Union

import numpy as np
import PIL.Image

from ...image_processing_utils import BaseImageProcessor, BatchFeature
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    PILImageResampling,
    get_image_size,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


__a:			Optional[int]	       =       logging.get_logger(__name__)



class       UpperCAmelCase  (						__lowercase ):





       '''simple docstring'''



       SCREAMING_SNAKE_CASE 		=   ["pixel_values"]

       def __init__(						self     ,       __lowerCAmelCase = True     ,       __lowerCAmelCase = 32     ,       __lowerCAmelCase=PILImageResampling.BILINEAR     ,       __lowerCAmelCase = True     ,       **__lowerCAmelCase     ,       )  ->  Optional[Any]:
        lowercase__    :   List[str]						= do_resize
        lowercase__    :   int						= do_rescale
        lowercase__    :   List[Any]						= size_divisor
        lowercase__    :   List[Any]						= resample
        super().__init__(**snake_case_			)

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase     ,       __lowerCAmelCase     ,       __lowerCAmelCase     ,       __lowerCAmelCase = None     ,       **__lowerCAmelCase			)  ->  List[Any]:
        lowercase__    :   int						= get_image_size(snake_case_			)
        # Rounds the height and width down to the closest multiple of size_divisor
        lowercase__    :   List[str]						= height // size_divisor * size_divisor
        lowercase__    :   Optional[Any]						= width // size_divisor * size_divisor
        lowercase__    :   Dict						= resize(snake_case_     ,       (new_h, new_w)     ,       resample=snake_case_     ,       data_format=snake_case_     ,       **snake_case_			)
        return image

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase     ,       __lowerCAmelCase     ,       __lowerCAmelCase = None     ,       **__lowerCAmelCase			)  ->  int:
        return rescale(image=snake_case_     ,       scale=snake_case_     ,       data_format=snake_case_     ,       **snake_case_			)


       def _lowerCAmelCase(						self     ,       __lowerCAmelCase     ,       __lowerCAmelCase = None     ,       __lowerCAmelCase = None     ,       __lowerCAmelCase=None     ,       __lowerCAmelCase = None     ,       __lowerCAmelCase = None     ,       __lowerCAmelCase = ChannelDimension.FIRST     ,       **__lowerCAmelCase     ,       )  ->  Optional[int]:
        lowercase__    :   int						= do_resize if do_resize is not None else self.do_resize
        lowercase__    :   Dict						= do_rescale if do_rescale is not None else self.do_rescale
        lowercase__    :   List[Any]						= size_divisor if size_divisor is not None else self.size_divisor
        lowercase__    :   str						= resample if resample is not None else self.resample

        if do_resize and size_divisor is None:
         raise ValueError('''size_divisor is required for resizing'''			)

        lowercase__    :   Optional[Any]						= make_list_of_images(snake_case_			)

        if not valid_images(snake_case_			):
         raise ValueError('''Invalid image(s)'''			)

        # All transformations expect numpy arrays.
        lowercase__    :   Optional[Any]						= [to_numpy_array(snake_case_			) for img in images]

        if do_resize:
         lowercase__    :   Optional[Any]						= [self.resize(snake_case_     ,       size_divisor=snake_case_     ,       resample=snake_case_			) for image in images]

        if do_rescale:
         lowercase__    :   Any						= [self.rescale(snake_case_     ,       scale=1 / 255			) for image in images]

        lowercase__    :   Optional[Any]						= [to_channel_dimension_format(snake_case_     ,       snake_case_			) for image in images]

        lowercase__    :   int						= {'''pixel_values''': images}
        return BatchFeature(data=snake_case_     ,       tensor_type=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

__a:			Optional[Any]	       =       logging.get_logger(__name__)



@add_end_docstrings(a__ )
class       UpperCAmelCase  (						a__ ):





       '''simple docstring'''



       def __init__(						self     ,       **__lowerCAmelCase			)  ->  int:
        super().__init__(**__lowerCAmelCase			)

        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     ,       __lowerCAmelCase     ,       **__lowerCAmelCase			)  ->  List[str]:
        return super().__call__(__lowerCAmelCase     ,       **__lowerCAmelCase			)

       def _lowerCAmelCase(						self     ,       **__lowerCAmelCase			)  ->  Optional[Any]:
        lowercase__    :   str						= {}
        if "candidate_labels" in kwargs:
         lowercase__    :   str						= kwargs['''candidate_labels''']
        if "hypothesis_template" in kwargs:
         lowercase__    :   str						= kwargs['''hypothesis_template''']

        return preprocess_params, {}, {}

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

       def _lowerCAmelCase(						self     ,       __lowerCAmelCase			)  ->  Optional[Any]:
        lowercase__    :   Any						= model_inputs.pop('''candidate_labels'''			)
        lowercase__    :   int						= model_inputs.pop('''text_inputs'''			)
        if isinstance(text_inputs[0]     ,       __lowerCAmelCase			):
         lowercase__    :   Union[str, Any]						= text_inputs[0]
        else:
         # Batching case.
         lowercase__    :   Optional[Any]						= text_inputs[0][0]

        lowercase__    :   Any						= self.model(**__lowerCAmelCase     ,       **__lowerCAmelCase			)

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


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

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

