


"""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							:    Optional[Any]					   =	get_tests_dir('''fixtures''')




class 							A_      (       unittest.TestCase ):


  def 							_lowercase					(			self:						int    ):







     '''simple docstring'''







     _lowerCamelCase   :				List[str]							=     mock.Mock()
     _lowerCamelCase   :				Union[str, Any]							=     500
     _lowerCamelCase   :				Optional[int]							=     {}
     _lowerCamelCase   :				str							=     HTTPError
     _lowerCamelCase   :				Dict							=     {}

     # Download this model to make sure it's in the cache.
     _lowerCamelCase   :				Optional[int]							=     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:
        _lowerCamelCase   :				Optional[int]							=     WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"    )
        # This check we did call the fake head request
        mock_head.assert_called()






  def 							_lowercase					(			self:						Tuple    ):







     '''simple docstring'''







     _lowerCamelCase   :				int							=     WavaVecaFeatureExtractor.from_pretrained(
         "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"    )




@is_staging_test
class 							A_      (       unittest.TestCase ):


  @classmethod
  def 							_lowercase					(			cls:						int    ):







     '''simple docstring'''







     _lowerCamelCase   :				Optional[Any]							=     TOKEN
     HfFolder.save_token(__lowerCAmelCase    )


  @classmethod
  def 							_lowercase					(			cls:						Optional[int]    ):







     '''simple docstring'''







     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 							_lowercase					(			self:						str    ):







     '''simple docstring'''







     _lowerCamelCase   :				List[Any]							=     WavaVecaFeatureExtractor.from_pretrained(__lowerCAmelCase    )
     feature_extractor.push_to_hub("test-feature-extractor"							,use_auth_token=self._token    )

     _lowerCamelCase   :				List[Any]							=     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    )

     _lowerCamelCase   :				Tuple							=     WavaVecaFeatureExtractor.from_pretrained(F"""{USER}/test-feature-extractor"""    )
     for k, v in feature_extractor.__dict__.items():
        self.assertEqual(__lowerCAmelCase							,getattr(__lowerCAmelCase							,__lowerCAmelCase    )    )


  def 							_lowercase					(			self:						Tuple    ):







     '''simple docstring'''







     _lowerCamelCase   :				Dict							=     WavaVecaFeatureExtractor.from_pretrained(__lowerCAmelCase    )
     feature_extractor.push_to_hub("valid_org/test-feature-extractor"							,use_auth_token=self._token    )

     _lowerCamelCase   :				Optional[int]							=     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    )

     _lowerCamelCase   :				str							=     WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"    )
     for k, v in feature_extractor.__dict__.items():
        self.assertEqual(__lowerCAmelCase							,getattr(__lowerCAmelCase							,__lowerCAmelCase    )    )






  def 							_lowercase					(			self:						List[str]    ):







     '''simple docstring'''







     CustomFeatureExtractor.register_for_auto_class()
     _lowerCamelCase   :				Optional[Any]							=     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"}							,)

     _lowerCamelCase   :				Optional[int]							=     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"    )

from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
							from ..image_utils import load_image


if is_torch_available():
							import torch

							from ..models.auto.modeling_auto import MODEL_FOR_OBJECT_DETECTION_MAPPING, MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING

UpperCamelCase_						:     int	       =							logging.get_logger(__name__)


UpperCamelCase_						:     List[Any]	       =							Dict[str, Any]
UpperCamelCase_						:     Optional[int]	       =							List[Prediction]




@add_end_docstrings(__snake_case       )
class   __lowercase						(						__snake_case       ):






	def __init__(self				: Dict      ,	*snake_case				: List[Any]      ,	**snake_case				: Tuple    ) ->			Any:
							super().__init__(*snake_case      ,	**snake_case    )

							if self.framework == "tf":
													raise ValueError(F"""The {self.__class__} is only available in PyTorch."""    )

							requires_backends(self      ,	"vision"    )
							self.check_model_type(
							    dict(MODEL_FOR_OBJECT_DETECTION_MAPPING.items() + MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING.items()    )    )






	def       _a(self				: Optional[int]      ,	**snake_case				: Union[str, Any]    ) ->			Any:
							_lowercase       :	List[Any]           =   {}
							if "threshold" in kwargs:
													_lowercase       :	Optional[Any]           =   kwargs["threshold"]
							return {}, {}, postprocess_kwargs






	def __call__(self				: str      ,	*snake_case				: Dict      ,	**snake_case				: Optional[int]    ) ->			Union[Predictions, List[Prediction]]:
							return super().__call__(*snake_case      ,	**snake_case    )






	def       _a(self				: Tuple      ,	snake_case				: str    ) ->			Dict:
							_lowercase       :	List[str]           =   load_image(snake_case    )
							_lowercase       :	str           =   torch.IntTensor([[image.height, image.width]]    )
							_lowercase       :	int           =   self.image_processor(images=[image]      ,	return_tensors="pt"    )
							if self.tokenizer is not None:
													_lowercase       :	List[Any]           =   self.tokenizer(text=inputs["words"]      ,	boxes=inputs["boxes"]      ,	return_tensors="pt"    )
							_lowercase       :	str           =   target_size
							return inputs






	def       _a(self				: List[Any]      ,	snake_case				: Any    ) ->			str:
							_lowercase       :	Dict           =   model_inputs.pop("target_size"    )
							_lowercase       :	str           =   self.model(**snake_case    )
							_lowercase       :	Union[str, Any]           =   outputs.__class__({"target_size": target_size, **outputs}    )
							if self.tokenizer is not None:
													_lowercase       :	str           =   model_inputs["bbox"]
							return model_outputs






	def       _a(self				: str      ,	snake_case				: Tuple      ,	snake_case				: Dict=0.9    ) ->			List[str]:
							_lowercase       :	Optional[Any]           =   model_outputs["target_size"]
							if self.tokenizer is not None:
													# This is a LayoutLMForTokenClassification variant.
													# The OCR got the boxes and the model classified the words.
													_lowercase      ,    _lowercase       :	str           =   target_size[0].tolist()

													def unnormalize(snake_case				: str    ):
																			return self._get_bounding_box(
																			    torch.Tensor(
																			        [
																			            (width * bbox[0] / 1000),
																			            (height * bbox[1] / 1000),
																			            (width * bbox[2] / 1000),
																			            (height * bbox[3] / 1000),
																			        ]    )    )

													_lowercase      ,    _lowercase       :	Any           =   model_outputs["logits"].squeeze(0    ).softmax(dim=-1    ).max(dim=-1    )
													_lowercase       :	Optional[Any]           =   [self.model.config.idalabel[prediction] for prediction in classes.tolist()]
													_lowercase       :	Dict           =   [unnormalize(snake_case    ) for bbox in model_outputs["bbox"].squeeze(0    )]
													_lowercase       :	Tuple           =   ["score", "label", "box"]
													_lowercase       :	int           =   [dict(zip(snake_case      ,	snake_case    )    ) for vals in zip(scores.tolist()      ,	snake_case      ,	snake_case    ) if vals[0] > threshold]
							else:
													# This is a regular ForObjectDetectionModel
													_lowercase       :	str           =   self.image_processor.post_process_object_detection(snake_case      ,	snake_case      ,	snake_case    )
													_lowercase       :	List[Any]           =   raw_annotations[0]
													_lowercase       :	str           =   raw_annotation["scores"]
													_lowercase       :	Optional[int]           =   raw_annotation["labels"]
													_lowercase       :	Dict           =   raw_annotation["boxes"]

													_lowercase       :	int           =   scores.tolist()
													_lowercase       :	List[str]           =   [self.model.config.idalabel[label.item()] for label in labels]
													_lowercase       :	List[Any]           =   [self._get_bounding_box(snake_case    ) for box in boxes]

													# {"scores": [...], ...} --> [{"score":x, ...}, ...]
													_lowercase       :	Union[str, Any]           =   ["score", "label", "box"]
													_lowercase       :	Tuple           =   [
													    dict(zip(snake_case      ,	snake_case    )    )
													    for vals in zip(raw_annotation["scores"]      ,	raw_annotation["labels"]      ,	raw_annotation["boxes"]    )
													]

							return annotation






	def       _a(self				: int      ,	snake_case				: "torch.Tensor"    ) ->			Dict[str, int]:
							if self.framework != "pt":
													raise ValueError("The ObjectDetectionPipeline is only available in PyTorch."    )
							_lowercase      ,    _lowercase      ,    _lowercase      ,    _lowercase       :	Optional[int]           =   box.int().tolist()
							_lowercase       :	Tuple           =   {
							    "xmin": xmin,
							    "ymin": ymin,
							    "xmax": xmax,
							    "ymax": ymax,
							}
							return bbox




