

import functools
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

import huggingface_hub.utils as hf_hub_utils
from tqdm import auto as tqdm_lib


lowerCAmelCase						     =  threading.Lock()
lowerCAmelCase						     =  None

lowerCAmelCase						     =  {
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
}

lowerCAmelCase						     =  logging.WARNING

lowerCAmelCase						     =  True

def    A_  (							):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     os.getenv('TRANSFORMERS_VERBOSITY'					,   a  )
      if env_level_str:
            if env_level_str in log_levels:
                  return log_levels[env_level_str]
            else:
                  logging.getLogger().warning(
                      f"Unknown option TRANSFORMERS_VERBOSITY={env_level_str}, "
                      f"has to be one of: { ', '.join(log_levels.keys()  ) }"  )
      return _default_log_level

def    A_  (							):

      """simple docstring"""



      return __name__.split('.'  )[0]

def    A_  (							):

      """simple docstring"""



      return logging.getLogger(_get_library_name()  )

def    A_  (							):

      """simple docstring"""



      global _default_handler

      with _lock:
            if _default_handler:
                  # This library has already configured the library root logger.
                  return
            SCREAMING_SNAKE_CASE_				:  int           =     logging.StreamHandler()  # Set sys.stderr as stream.
            SCREAMING_SNAKE_CASE_				:  int           =     sys.stderr.flush

            # Apply our default configuration to the library root logger.
            SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     _get_library_root_logger()
            library_root_logger.addHandler(_default_handler  )
            library_root_logger.setLevel(_get_default_logging_level()  )
            SCREAMING_SNAKE_CASE_				:  Tuple           =     False

def    A_  (							):

      """simple docstring"""



      global _default_handler

      with _lock:
            if not _default_handler:
                  return

            SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     _get_library_root_logger()
            library_root_logger.removeHandler(_default_handler  )
            library_root_logger.setLevel(logging.NOTSET  )
            SCREAMING_SNAKE_CASE_				:  Any           =     None

def    A_  (							):

      """simple docstring"""



      return log_levels

def    A_  (							a = None  ):

      """simple docstring"""



      if name is None:
            SCREAMING_SNAKE_CASE_				:  Any           =     _get_library_name()

      _configure_library_root_logger()
      return logging.getLogger(a  )

def    A_  (							):

      """simple docstring"""



      _configure_library_root_logger()
      return _get_library_root_logger().getEffectiveLevel()

def    A_  (							a  ):

      """simple docstring"""



      _configure_library_root_logger()
      _get_library_root_logger().setLevel(a  )

def    A_  (							):

      """simple docstring"""



      return set_verbosity(a  )

def    A_  (							):

      """simple docstring"""



      return set_verbosity(a  )

def    A_  (							):

      """simple docstring"""



      return set_verbosity(a  )

def    A_  (							):

      """simple docstring"""



      return set_verbosity(a  )

def    A_  (							):

      """simple docstring"""



      _configure_library_root_logger()

      assert _default_handler is not None
      _get_library_root_logger().removeHandler(_default_handler  )

def    A_  (							):

      """simple docstring"""



      _configure_library_root_logger()

      assert _default_handler is not None
      _get_library_root_logger().addHandler(_default_handler  )

def    A_  (							a  ):

      """simple docstring"""



      _configure_library_root_logger()

      assert handler is not None
      _get_library_root_logger().addHandler(a  )

def    A_  (							a  ):

      """simple docstring"""



      _configure_library_root_logger()

      assert handler is not None and handler not in _get_library_root_logger().handlers
      _get_library_root_logger().removeHandler(a  )

def    A_  (							):

      """simple docstring"""



      _configure_library_root_logger()
      SCREAMING_SNAKE_CASE_				:  Any           =     False

def    A_  (							):

      """simple docstring"""



      _configure_library_root_logger()
      SCREAMING_SNAKE_CASE_				:  int           =     True

def    A_  (							):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     _get_library_root_logger().handlers

      for handler in handlers:
            SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     logging.Formatter('[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s'  )
            handler.setFormatter(a  )

def    A_  (							):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     _get_library_root_logger().handlers

      for handler in handlers:
            handler.setFormatter(a  )

def    A_  (							self					,   *a					,   **a  ):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     os.getenv('TRANSFORMERS_NO_ADVISORY_WARNINGS'					,   a  )
      if no_advisory_warnings:
            return
      self.warning(*a					,   **a  )


lowerCAmelCase						     =  warning_advice

@functools.lru_cache(a  )
def    A_  (							self					,   *a					,   **a  ):

      """simple docstring"""



      self.warning(*a					,   **a  )


lowerCAmelCase						     =  warning_once
class  _A		:

  def __init__(				self ,					*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					):  # pylint: disable=unused-argument





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     args[0] if args else None

  def __iter__(				self					):





        """simple docstring"""







        return iter(self._iterator					)

  def __getattr__(				self ,					_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        def empty_fn(*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					):  # pylint: disable=unused-argument
              return

        return empty_fn

  def __enter__(				self					):





        """simple docstring"""







        return self



  def __exit__(				self ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        return

class  _A		:

  def __call__(				self ,					*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        if _tqdm_active:
              return tqdm_lib.tqdm(*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					)
        else:
              return EmptyTqdm(*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					)

  def  UpperCAmelCase    (				self ,					*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  Tuple           =     None
        if _tqdm_active:
              return tqdm_lib.tqdm.set_lock(*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					)



  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        if _tqdm_active:
              return tqdm_lib.tqdm.get_lock()

lowerCAmelCase						     =  _tqdm_cls()

def    A_  (							):

      """simple docstring"""



      global _tqdm_active
      return bool(_tqdm_active  )

def    A_  (							):

      """simple docstring"""



      global _tqdm_active
      SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     True
      hf_hub_utils.enable_progress_bars()



def    A_  (							):

      """simple docstring"""



      global _tqdm_active
      SCREAMING_SNAKE_CASE_				:  str           =     False
      hf_hub_utils.disable_progress_bars()









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

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


lowerCAmelCase     :   Optional[int]						     =  Dict[str, Any]
lowerCAmelCase     :   Any						     =  List[Prediction]
@add_end_docstrings(__magic_name__)
class  _A		(  __magic_name__):

  def __init__(				self ,					*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        super().__init__(*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_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  UpperCAmelCase    (				self ,					**_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  Any           =     {}
        if "threshold" in kwargs:
              SCREAMING_SNAKE_CASE_				:  List[str]           =     kwargs['threshold']
        return {}, {}, postprocess_kwargs

  def __call__(				self ,					*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        return super().__call__(*_SCREAMING_SNAKE_CASE ,					**_SCREAMING_SNAKE_CASE					)

  def  UpperCAmelCase    (				self ,					_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  Dict           =     load_image(_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     torch.IntTensor([[image.height, image.width]]					)
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     self.image_processor(images=[image] ,					return_tensors='pt'					)
        if self.tokenizer is not None:
              SCREAMING_SNAKE_CASE_				:  Tuple           =     self.tokenizer(text=inputs['words'] ,					boxes=inputs['boxes'] ,					return_tensors='pt'					)
        SCREAMING_SNAKE_CASE_				:  Optional[int]           =     target_size
        return inputs

  def  UpperCAmelCase    (				self ,					_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  Dict           =     model_inputs.pop('target_size'					)
        SCREAMING_SNAKE_CASE_				:  List[str]           =     self.model(**_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  Dict           =     outputs.__class__({'target_size': target_size, **outputs}					)
        if self.tokenizer is not None:
              SCREAMING_SNAKE_CASE_				:  str           =     model_inputs['bbox']
        return model_outputs

  def  UpperCAmelCase    (				self ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE=0.9					):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  List[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.
              SCREAMING_SNAKE_CASE_       ,						SCREAMING_SNAKE_CASE_				:  Any           =     target_size[0].tolist()

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

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

              SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     scores.tolist()
              SCREAMING_SNAKE_CASE_				:  Any           =     [self.model.config.idalabel[label.item()] for label in labels]
              SCREAMING_SNAKE_CASE_				:  Optional[int]           =     [self._get_bounding_box(_SCREAMING_SNAKE_CASE					) for box in boxes]

              # {"scores": [...], ...} --> [{"score":x, ...}, ...]
              SCREAMING_SNAKE_CASE_				:  str           =     ['score', 'label', 'box']
              SCREAMING_SNAKE_CASE_				:  Optional[int]           =     [
                  dict(zip(_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE					)					)
                  for vals in zip(raw_annotation['scores'] ,					raw_annotation['labels'] ,					raw_annotation['boxes']					)
              ]

        return annotation



  def  UpperCAmelCase    (				self ,					_SCREAMING_SNAKE_CASE					):





        """simple docstring"""







        if self.framework != "pt":
              raise ValueError('The ObjectDetectionPipeline is only available in PyTorch.'					)
        SCREAMING_SNAKE_CASE_       ,						SCREAMING_SNAKE_CASE_       ,						SCREAMING_SNAKE_CASE_       ,						SCREAMING_SNAKE_CASE_				:  List[Any]           =     box.int().tolist()
        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     {
            'xmin': xmin,
            'ymin': ymin,
            'xmax': xmax,
            'ymax': ymax,
        }
        return bbox






