



import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map

def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Union[dict, list, tuple, torch.Tensor]				)  ->     List[Tuple[int, ...]]:
   _A      =       []
   if isinstance(_snake_case	,   _snake_case				):
      for v in tree.values():
         shapes.extend(_fetch_dims(_snake_case				)				)
   elif isinstance(_snake_case	,   (list, tuple)				):
      for t in tree:
         shapes.extend(_fetch_dims(_snake_case				)				)
   elif isinstance(_snake_case	,   torch.Tensor				):
      shapes.append(tree.shape				)
   else:
      raise ValueError('''Not supported'''				)

   return shapes

@torch.jit.ignore
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:int	,   _snake_case	:Tuple[int, ...]				)  ->     Tuple[int, ...]:
   _A      =       []
   for d in reversed(_snake_case				):
      idx.append(flat_idx % d				)
      _A      =       flat_idx // d

   return tuple(reversed(_snake_case				)				)

@torch.jit.ignore
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Sequence[int]	,   _snake_case	:Sequence[int]	,   _snake_case	:Sequence[int]	,   _snake_case	:Optional[Sequence[bool]] = None	,   _snake_case	:Optional[Sequence[bool]] = None	,   )  ->     List[Tuple[slice, ...]]:

   # start_edges and end_edges both indicate whether, starting from any given
   # dimension, the start/end index is at the top/bottom edge of the
   # corresponding tensor, modeled as a tree
   def reduce_edge_list(_snake_case	:List[bool]				) -> None:
      _A      =       True
      for i in range(len(_snake_case				)				):
         _A      =       -1 * (i + 1)
         l[reversed_idx] &= tally
         _A      =       l[reversed_idx]

   if start_edges is None:
      _A      =       [s == 0 for s in start]
      reduce_edge_list(_snake_case				)
   if end_edges is None:
      _A      =       [e == (d - 1) for e, d in zip(_snake_case	,   _snake_case				)]
      reduce_edge_list(_snake_case				)

   # Base cases. Either start/end are empty and we're done, or the final,
   # one-dimensional tensor can be simply sliced
   if len(_snake_case				) == 0:
      return [()]
   elif len(_snake_case				) == 1:
      return [(slice(start[0]	,   end[0] + 1				),)]

   _A      =       []
   _A      =       []

   # Dimensions common to start and end can be selected directly
   for s, e in zip(_snake_case	,   _snake_case				):
      if s == e:
         path_list.append(slice(_snake_case	,   s + 1				)				)
      else:
         break

   _A      =       tuple(_snake_case				)
   _A      =       len(_snake_case				)

   # start == end, and we're done
   if divergence_idx == len(_snake_case				):
      return [path]

   def upper() -> Tuple[Tuple[slice, ...], ...]:
      assert start_edges is not None
      assert end_edges is not None

      _A      =       start[divergence_idx]
      return tuple(
          path + (slice(_snake_case	,   sdi + 1				),) + s
          for s in _get_minimal_slice_set(
              start[divergence_idx + 1 :]	,   [d - 1 for d in dims[divergence_idx + 1 :]]	,   dims[divergence_idx + 1 :]	,   start_edges=start_edges[divergence_idx + 1 :]	,   end_edges=[True for _ in end_edges[divergence_idx + 1 :]]	,   )				)

   def lower() -> Tuple[Tuple[slice, ...], ...]:
      assert start_edges is not None
      assert end_edges is not None

      _A      =       end[divergence_idx]
      return tuple(
          path + (slice(_snake_case	,   edi + 1				),) + s
          for s in _get_minimal_slice_set(
              [0 for _ in start[divergence_idx + 1 :]]	,   end[divergence_idx + 1 :]	,   dims[divergence_idx + 1 :]	,   start_edges=[True for _ in start_edges[divergence_idx + 1 :]]	,   end_edges=end_edges[divergence_idx + 1 :]	,   )				)

   # If both start and end are at the edges of the subtree rooted at
   # divergence_idx, we can just select the whole subtree at once
   if start_edges[divergence_idx] and end_edges[divergence_idx]:
      slices.append(path + (slice(start[divergence_idx]	,   end[divergence_idx] + 1				),)				)
   # If just start is at the edge, we can grab almost all of the subtree,
   # treating only the ragged bottom edge as an edge case
   elif start_edges[divergence_idx]:
      slices.append(path + (slice(start[divergence_idx]	,   end[divergence_idx]				),)				)
      slices.extend(lower()				)
   # Analogous to the previous case, but the top is ragged this time
   elif end_edges[divergence_idx]:
      slices.extend(upper()				)
      slices.append(path + (slice(start[divergence_idx] + 1	,   end[divergence_idx] + 1				),)				)
   # If both sides of the range are ragged, we need to handle both sides
   # separately. If there's contiguous meat in between them, we can index it
   # in one big chunk
   else:
      slices.extend(upper()				)
      _A      =       end[divergence_idx] - start[divergence_idx]
      if middle_ground > 1:
         slices.append(path + (slice(start[divergence_idx] + 1	,   end[divergence_idx]				),)				)
      slices.extend(lower()				)

   return slices

@torch.jit.ignore
def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:torch.Tensor	,   _snake_case	:int	,   _snake_case	:int	,   _snake_case	:int				)  ->     torch.Tensor:
   _A      =       t.shape[:no_batch_dims]
   _A      =       list(_flat_idx_to_idx(_snake_case	,   _snake_case				)				)
   # _get_minimal_slice_set is inclusive
   _A      =       list(_flat_idx_to_idx(flat_end - 1	,   _snake_case				)				)

   # Get an ordered list of slices to perform
   _A      =       _get_minimal_slice_set(
       _snake_case	,   _snake_case	,   _snake_case	,   )

   _A      =       [t[s] for s in slices]

   return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]				) for s in sliced_tensors]				)

def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:Callable	,   _snake_case	:Dict[str, Any]	,   _snake_case	:int	,   _snake_case	:int	,   _snake_case	:bool = False	,   _snake_case	:Any = None	,   _snake_case	:bool = False	,   )  ->     Any:
   if not (len(_snake_case				) > 0):
      raise ValueError('''Must provide at least one input'''				)

   _A      =       [shape[:no_batch_dims] for shape in _fetch_dims(_snake_case				)]
   _A      =       tuple([max(_snake_case				) for s in zip(*_snake_case				)]				)

   def _prep_inputs(_snake_case	:torch.Tensor				) -> torch.Tensor:
      if not low_mem:
         if not sum(t.shape[:no_batch_dims]				) == no_batch_dims:
            _A      =       t.expand(orig_batch_dims + t.shape[no_batch_dims:]				)
         _A      =       t.reshape(-1	,   *t.shape[no_batch_dims:]				)
      else:
         _A      =       t.expand(orig_batch_dims + t.shape[no_batch_dims:]				)
      return t

   _A      =       tensor_tree_map(_prep_inputs	,   _snake_case				)
   _A      =       None
   if _out is not None:
      _A      =       tensor_tree_map(lambda _snake_case				: t.view([-1] + list(t.shape[no_batch_dims:]				)				)	,   _out				)

   _A      =       1
   for d in orig_batch_dims:
      flat_batch_dim *= d

   _A      =       flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

   def _select_chunk(_snake_case	:torch.Tensor				) -> torch.Tensor:
      return t[i : i + chunk_size] if t.shape[0] != 1 else t

   _A      =       0
   _A      =       prepped_outputs
   for _ in range(_snake_case				):
      # Chunk the input
      if not low_mem:
         _A      =       _select_chunk
      else:
         _A      =       partial(
             _chunk_slice	,   flat_start=_snake_case	,   flat_end=min(_snake_case	,   i + chunk_size				)	,   no_batch_dims=len(_snake_case				)	,   )

      _A      =       tensor_tree_map(_snake_case	,   _snake_case				)

      # Run the layer on the chunk
      _A      =       layer(**_snake_case				)

      # Allocate space for the output
      if out is None:
         _A      =       tensor_tree_map(lambda _snake_case				: t.new_zeros((flat_batch_dim,) + t.shape[1:]				)	,   _snake_case				)

      # Put the chunk in its pre-allocated space
      if isinstance(_snake_case	,   _snake_case				):

         def assign(_snake_case	:dict	,   _snake_case	:dict				) -> None:
            for k, v in da.items():
               if isinstance(_snake_case	,   _snake_case				):
                  assign(_snake_case	,   da[k]				)
               else:
                  if _add_into_out:
                     v[i : i + chunk_size] += da[k]
                  else:
                     _A      =       da[k]

         assign(_snake_case	,   _snake_case				)
      elif isinstance(_snake_case	,   _snake_case				):
         for xa, xa in zip(_snake_case	,   _snake_case				):
            if _add_into_out:
               xa[i : i + chunk_size] += xa
            else:
               _A      =       xa
      elif isinstance(_snake_case	,   torch.Tensor				):
         if _add_into_out:
            out[i : i + chunk_size] += output_chunk
         else:
            _A      =       output_chunk
      else:
         raise ValueError('''Not supported'''				)

      i += chunk_size

   _A      =       tensor_tree_map(lambda _snake_case				: t.view(orig_batch_dims + t.shape[1:]				)	,   _snake_case				)

   return out


class 	lowerCamelCase__	:



 """simple docstring"""

 def __init__(    self					:					List[str]			,						__lowerCAmelCase					:					int = 5_12			,						)   ->	Optional[int]:
    _A      =       max_chunk_size
    _A      =       None
    _A      =       None

 def     snake_case_      (    self					:					int			,						__lowerCAmelCase					:					Callable			,						__lowerCAmelCase					:					tuple			,						__lowerCAmelCase					:					int      )   ->	int:
    logging.info('''Tuning chunk size...'''      )

    if min_chunk_size >= self.max_chunk_size:
       return min_chunk_size

    _A      =       [2**l for l in range(int(math.log(self.max_chunk_size			,						2      )      ) + 1      )]
    _A      =       [c for c in candidates if c > min_chunk_size]
    _A      =       [min_chunk_size] + candidates
    candidates[-1] += 4

    def test_chunk_size(__lowerCAmelCase					:					int      ) -> bool:
       try:
          with torch.no_grad():
             fn(*__lowerCAmelCase			,						chunk_size=__lowerCAmelCase      )
          return True
       except RuntimeError:
          return False

    _A      =       0
    _A      =       len(__lowerCAmelCase      ) - 1
    while i > min_viable_chunk_size_index:
       _A      =       test_chunk_size(candidates[i]      )
       if not viable:
          _A      =       (min_viable_chunk_size_index + i) // 2
       else:
          _A      =       i
          _A      =       (i + len(__lowerCAmelCase      ) - 1) // 2

    return candidates[min_viable_chunk_size_index]

 def     snake_case_      (    self					:					Union[str, Any]			,						__lowerCAmelCase					:					Iterable			,						__lowerCAmelCase					:					Iterable      )   ->	bool:
    _A      =       True
    for aa, aa in zip(__lowerCAmelCase			,						__lowerCAmelCase      ):
       assert type(__lowerCAmelCase      ) == type(__lowerCAmelCase      )
       if isinstance(__lowerCAmelCase			,						(list, tuple)      ):
          consistent &= self._compare_arg_caches(__lowerCAmelCase			,						__lowerCAmelCase      )
       elif isinstance(__lowerCAmelCase			,						__lowerCAmelCase      ):
          _A      =       [v for _, v in sorted(aa.items()			,						key=lambda __lowerCAmelCase      : x[0]      )]
          _A      =       [v for _, v in sorted(aa.items()			,						key=lambda __lowerCAmelCase      : x[0]      )]
          consistent &= self._compare_arg_caches(__lowerCAmelCase			,						__lowerCAmelCase      )
       else:
          consistent &= aa == aa

    return consistent

 def     snake_case_      (    self					:					Any			,						__lowerCAmelCase					:					Callable			,						__lowerCAmelCase					:					tuple			,						__lowerCAmelCase					:					int			,						)   ->	int:
    _A      =       True
    _A      =       tree_map(lambda __lowerCAmelCase      : a.shape if isinstance(__lowerCAmelCase			,						torch.Tensor      ) else a			,						__lowerCAmelCase			,						__lowerCAmelCase      )
    if self.cached_arg_data is not None:
       # If args have changed shape/value, we need to re-tune
       assert len(self.cached_arg_data      ) == len(__lowerCAmelCase      )
       _A      =       self._compare_arg_caches(self.cached_arg_data			,						__lowerCAmelCase      )
    else:
       # Otherwise, we can reuse the precomputed value
       _A      =       False

    if not consistent:
       _A      =       self._determine_favorable_chunk_size(
           __lowerCAmelCase			,						__lowerCAmelCase			,						__lowerCAmelCase			,						)
       _A      =       arg_data

    assert self.cached_chunk_size is not None

    return self.cached_chunk_size








import unittest

from transformers import is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
   from PIL import Image
else:


   class 	lowerCamelCase__	:



    """simple docstring"""

    @staticmethod
    def     snake_case_      (    *__lowerCAmelCase					:					Optional[Any]			,						**__lowerCAmelCase					:					Any      )   ->	Any:
       pass


@is_pipeline_test
@require_vision
class 	lowerCamelCase__	(       unittest.TestCase):



 """simple docstring"""

 @require_torch
 def     snake_case_      (    self					:					Tuple      )   ->	Tuple:
    _A      =       pipeline(
        model='''hf-internal-testing/tiny-random-clip-zero-shot-image-classification'''			,						)
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''a''', '''b''', '''c''']      )

    # The floating scores are so close, we enter floating error approximation and the order is not guaranteed across
    # python and torch versions.
    self.assertIn(
        nested_simplify(__lowerCAmelCase      )			,						[
            [{'''score''': 0.333, '''label''': '''a'''}, {'''score''': 0.333, '''label''': '''b'''}, {'''score''': 0.333, '''label''': '''c'''}],
            [{'''score''': 0.333, '''label''': '''a'''}, {'''score''': 0.333, '''label''': '''c'''}, {'''score''': 0.333, '''label''': '''b'''}],
        ]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''A''', '''B''', '''C''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
        ]			,						)

 @require_tf
 def     snake_case_      (    self					:					int      )   ->	Optional[int]:
    _A      =       pipeline(
        model='''hf-internal-testing/tiny-random-clip-zero-shot-image-classification'''			,						framework='''tf'''      )
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''a''', '''b''', '''c''']      )

    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[{'''score''': 0.333, '''label''': '''a'''}, {'''score''': 0.333, '''label''': '''b'''}, {'''score''': 0.333, '''label''': '''c'''}]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''A''', '''B''', '''C''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
        ]			,						)

 @slow
 @require_torch
 def     snake_case_      (    self					:					Optional[int]      )   ->	int:
    _A      =       pipeline(
        task='''zero-shot-image-classification'''			,						model='''openai/clip-vit-base-patch32'''			,						)
    # This is an image of 2 cats with remotes and no planes
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''cat''', '''plane''', '''remote''']      )

    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            {'''score''': 0.511, '''label''': '''remote'''},
            {'''score''': 0.485, '''label''': '''cat'''},
            {'''score''': 0.004, '''label''': '''plane'''},
        ]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''cat''', '''plane''', '''remote''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.511, '''label''': '''remote'''},
                {'''score''': 0.485, '''label''': '''cat'''},
                {'''score''': 0.004, '''label''': '''plane'''},
            ],
        ]
        * 5			,						)

 @slow
 @require_tf
 def     snake_case_      (    self					:					Optional[int]      )   ->	Dict:
    _A      =       pipeline(
        task='''zero-shot-image-classification'''			,						model='''openai/clip-vit-base-patch32'''			,						framework='''tf'''      )
    # This is an image of 2 cats with remotes and no planes
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''cat''', '''plane''', '''remote''']      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            {'''score''': 0.511, '''label''': '''remote'''},
            {'''score''': 0.485, '''label''': '''cat'''},
            {'''score''': 0.004, '''label''': '''plane'''},
        ]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''cat''', '''plane''', '''remote''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.511, '''label''': '''remote'''},
                {'''score''': 0.485, '''label''': '''cat'''},
                {'''score''': 0.004, '''label''': '''plane'''},
            ],
        ]
        * 5			,						)



