


'''simple docstring'''





import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


__lowerCamelCase												=						get_tests_dir('''fixtures''')




class 				UpperCAmelCase			(			unittest.TestCase						):


 def        __SCREAMING_SNAKE_CASE					(      self							:	str ):
    # A mock response for an HTTP head request to emulate server down
    UpperCAmelCase__	:List[Any]      =      mock.Mock()
    UpperCAmelCase__	:Tuple      =      5_0_0
    UpperCAmelCase__	:Optional[int]      =      {}
    UpperCAmelCase__	:Any      =      HTTPError
    UpperCAmelCase__	:Tuple      =      {}

    # Download this model to make sure it's in the cache.
    UpperCAmelCase__	:Tuple      =      WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2''' )
    # Under the mock environment we get a 500 error when trying to reach the model.
    with mock.patch('''requests.Session.request'''			,       return_value=__lowerCamelCase ) as mock_head:
       UpperCAmelCase__	:str      =      WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2''' )
       # This check we did call the fake head request
       mock_head.assert_called()





 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    # This test is for deprecated behavior and can be removed in v5
    UpperCAmelCase__	:Union[str, Any]      =      WavaVecaFeatureExtractor.from_pretrained(
        '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json''' )




@is_staging_test
class 				UpperCAmelCase			(			unittest.TestCase						):


 @classmethod
 def        __SCREAMING_SNAKE_CASE					(      cls							:	List[str] ):
    UpperCAmelCase__	:Tuple      =      TOKEN
    HfFolder.save_token(__lowerCamelCase )


 @classmethod
 def        __SCREAMING_SNAKE_CASE					(      cls							:	str ):
    try:
       delete_repo(token=cls._token			,       repo_id='''test-feature-extractor''' )
    except HTTPError:
       pass

    try:
       delete_repo(token=cls._token			,       repo_id='''valid_org/test-feature-extractor-org''' )
    except HTTPError:
       pass

    try:
       delete_repo(token=cls._token			,       repo_id='''test-dynamic-feature-extractor''' )
    except HTTPError:
       pass


 def        __SCREAMING_SNAKE_CASE					(      self							:	Any ):
    UpperCAmelCase__	:Tuple      =      WavaVecaFeatureExtractor.from_pretrained(__lowerCamelCase )
    feature_extractor.push_to_hub('''test-feature-extractor'''			,       use_auth_token=self._token )

    UpperCAmelCase__	:str      =      WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor''' )
    for k, v in feature_extractor.__dict__.items():
       self.assertEqual(__lowerCamelCase			,       getattr(__lowerCamelCase			,       __lowerCamelCase ) )

    # Reset repo
    delete_repo(token=self._token			,       repo_id='''test-feature-extractor''' )

    # Push to hub via save_pretrained
    with tempfile.TemporaryDirectory() as tmp_dir:
       feature_extractor.save_pretrained(
           __lowerCamelCase			,       repo_id='''test-feature-extractor'''			,       push_to_hub=__lowerCamelCase			,       use_auth_token=self._token )

    UpperCAmelCase__	:Any      =      WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor''' )
    for k, v in feature_extractor.__dict__.items():
       self.assertEqual(__lowerCamelCase			,       getattr(__lowerCamelCase			,       __lowerCamelCase ) )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Any ):
    UpperCAmelCase__	:str      =      WavaVecaFeatureExtractor.from_pretrained(__lowerCamelCase )
    feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''			,       use_auth_token=self._token )

    UpperCAmelCase__	:str      =      WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor''' )
    for k, v in feature_extractor.__dict__.items():
       self.assertEqual(__lowerCamelCase			,       getattr(__lowerCamelCase			,       __lowerCamelCase ) )

    # Reset repo
    delete_repo(token=self._token			,       repo_id='''valid_org/test-feature-extractor''' )

    # Push to hub via save_pretrained
    with tempfile.TemporaryDirectory() as tmp_dir:
       feature_extractor.save_pretrained(
           __lowerCamelCase			,       repo_id='''valid_org/test-feature-extractor-org'''			,       push_to_hub=__lowerCamelCase			,       use_auth_token=self._token )

    UpperCAmelCase__	:Dict      =      WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor-org''' )
    for k, v in feature_extractor.__dict__.items():
       self.assertEqual(__lowerCamelCase			,       getattr(__lowerCamelCase			,       __lowerCamelCase ) )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Tuple ):
    CustomFeatureExtractor.register_for_auto_class()
    UpperCAmelCase__	:Dict      =      CustomFeatureExtractor.from_pretrained(__lowerCamelCase )

    feature_extractor.push_to_hub('''test-dynamic-feature-extractor'''			,       use_auth_token=self._token )

    # This has added the proper auto_map field to the config
    self.assertDictEqual(
        feature_extractor.auto_map			,       {'''AutoFeatureExtractor''': '''custom_feature_extraction.CustomFeatureExtractor'''}			,       )

    UpperCAmelCase__	:Tuple      =      AutoFeatureExtractor.from_pretrained(
        f'''{USER}/test-dynamic-feature-extractor'''			,       trust_remote_code=__lowerCamelCase )
    # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
    self.assertEqual(new_feature_extractor.__class__.__name__			,       '''CustomFeatureExtractor''' )







'''simple docstring'''





import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available




class 				UpperCAmelCase			(			_snake_case						):
 UpperCAmelCase				=					["image_processor", "tokenizer"]
 UpperCAmelCase				=					"OwlViTImageProcessor"
 UpperCAmelCase				=					("CLIPTokenizer", "CLIPTokenizerFast")


 def __init__(      self							:	List[str]			,       __lowerCamelCase							:	Optional[int]=None			,       __lowerCamelCase							:	Optional[Any]=None			,       **__lowerCamelCase							:	str ):
    UpperCAmelCase__	:int      =      None
    if "feature_extractor" in kwargs:
       warnings.warn(
           '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
           ''' instead.'''			,       __lowerCamelCase			,       )
       UpperCAmelCase__	:Union[str, Any]      =      kwargs.pop('''feature_extractor''' )

    UpperCAmelCase__	:List[str]      =      image_processor if image_processor is not None else feature_extractor
    if image_processor is None:
       raise ValueError('''You need to specify an `image_processor`.''' )
    if tokenizer is None:
       raise ValueError('''You need to specify a `tokenizer`.''' )

    super().__init__(__lowerCamelCase			,       __lowerCamelCase )


 def __call__(      self							:	int			,       __lowerCamelCase							:	str=None			,       __lowerCamelCase							:	List[Any]=None			,       __lowerCamelCase							:	Dict=None			,       __lowerCamelCase							:	str="max_length"			,       __lowerCamelCase							:	Any="np"			,       **__lowerCamelCase							:	Tuple ):

    if text is None and query_images is None and images is None:
       raise ValueError(
           '''You have to specify at least one text or query image or image. All three cannot be none.''' )

    if text is not None:
       if isinstance(__lowerCamelCase			,       __lowerCamelCase ) or (isinstance(__lowerCamelCase			,       __lowerCamelCase ) and not isinstance(text[0]			,       __lowerCamelCase )):
          UpperCAmelCase__	:Any      =      [self.tokenizer(__lowerCamelCase			,       padding=__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )]

       elif isinstance(__lowerCamelCase			,       __lowerCamelCase ) and isinstance(text[0]			,       __lowerCamelCase ):
          UpperCAmelCase__	:Tuple      =      []

          # Maximum number of queries across batch
          UpperCAmelCase__	:List[str]      =      max([len(__lowerCamelCase ) for t in text] )

          # Pad all batch samples to max number of text queries
          for t in text:
             if len(__lowerCamelCase ) != max_num_queries:
                UpperCAmelCase__	:str      =      t + [''' '''] * (max_num_queries - len(__lowerCamelCase ))

             UpperCAmelCase__	:Tuple      =      self.tokenizer(__lowerCamelCase			,       padding=__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )
             encodings.append(__lowerCamelCase )
       else:
          raise TypeError('''Input text should be a string, a list of strings or a nested list of strings''' )

       if return_tensors == "np":
          UpperCAmelCase__	:List[Any]      =      np.concatenate([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:Any      =      np.concatenate([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       elif return_tensors == "jax" and is_flax_available():
          import jax.numpy as jnp

          UpperCAmelCase__	:List[Any]      =      jnp.concatenate([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:Union[str, Any]      =      jnp.concatenate([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       elif return_tensors == "pt" and is_torch_available():
          import torch

          UpperCAmelCase__	:Any      =      torch.cat([encoding['''input_ids'''] for encoding in encodings]			,       dim=0 )
          UpperCAmelCase__	:List[str]      =      torch.cat([encoding['''attention_mask'''] for encoding in encodings]			,       dim=0 )

       elif return_tensors == "tf" and is_tf_available():
          import tensorflow as tf

          UpperCAmelCase__	:Optional[Any]      =      tf.stack([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:int      =      tf.stack([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       else:
          raise ValueError('''Target return tensor type could not be returned''' )

       UpperCAmelCase__	:List[Any]      =      BatchEncoding()
       UpperCAmelCase__	:Union[str, Any]      =      input_ids
       UpperCAmelCase__	:Dict      =      attention_mask

    if query_images is not None:
       UpperCAmelCase__	:Tuple      =      BatchEncoding()
       UpperCAmelCase__	:int      =      self.image_processor(
           __lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase ).pixel_values
       UpperCAmelCase__	:Optional[int]      =      query_pixel_values

    if images is not None:
       UpperCAmelCase__	:str      =      self.image_processor(__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )

    if text is not None and images is not None:
       UpperCAmelCase__	:Dict      =      image_features.pixel_values
       return encoding
    elif query_images is not None and images is not None:
       UpperCAmelCase__	:Tuple      =      image_features.pixel_values
       return encoding
    elif text is not None or query_images is not None:
       return encoding
    else:
       return BatchEncoding(data=dict(**__lowerCamelCase )			,       tensor_type=__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       *__lowerCamelCase							:	Optional[int]			,       **__lowerCamelCase							:	int ):
    return self.image_processor.post_process(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict			,       *__lowerCamelCase							:	Any			,       **__lowerCamelCase							:	Tuple ):
    return self.image_processor.post_process_object_detection(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any]			,       *__lowerCamelCase							:	Tuple			,       **__lowerCamelCase							:	str ):
    return self.image_processor.post_process_image_guided_detection(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       *__lowerCamelCase							:	Optional[Any]			,       **__lowerCamelCase							:	Union[str, Any] ):
    return self.tokenizer.batch_decode(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any]			,       *__lowerCamelCase							:	Union[str, Any]			,       **__lowerCamelCase							:	Dict ):
    return self.tokenizer.decode(*__lowerCamelCase			,       **__lowerCamelCase )


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    warnings.warn(
        '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''			,       __lowerCamelCase			,       )
    return self.image_processor_class





 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    warnings.warn(
        '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''			,       __lowerCamelCase			,       )
    return self.image_processor



