


import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__lowercase    =				logging.get_logger(__name__)


__lowercase    =				{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

# See all BART models at https://huggingface.co/models?filter=bart
__lowercase    =				{
    '''vocab_file''': {
        '''facebook/bart-base''': '''https://huggingface.co/facebook/bart-base/resolve/main/vocab.json''',
        '''facebook/bart-large''': '''https://huggingface.co/facebook/bart-large/resolve/main/vocab.json''',
        '''facebook/bart-large-mnli''': '''https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json''',
        '''facebook/bart-large-cnn''': '''https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json''',
        '''facebook/bart-large-xsum''': '''https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json''',
        '''yjernite/bart_eli5''': '''https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''facebook/bart-base''': '''https://huggingface.co/facebook/bart-base/resolve/main/merges.txt''',
        '''facebook/bart-large''': '''https://huggingface.co/facebook/bart-large/resolve/main/merges.txt''',
        '''facebook/bart-large-mnli''': '''https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt''',
        '''facebook/bart-large-cnn''': '''https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt''',
        '''facebook/bart-large-xsum''': '''https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt''',
        '''yjernite/bart_eli5''': '''https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt''',
    },
}

__lowercase    =				{
    '''facebook/bart-base''': 1024,
    '''facebook/bart-large''': 1024,
    '''facebook/bart-large-mnli''': 1024,
    '''facebook/bart-large-cnn''': 1024,
    '''facebook/bart-large-xsum''': 1024,
    '''yjernite/bart_eli5''': 1024,
}


@lru_cache()
def 					lowercase      (    )->				List[Any]:
  '''simple docstring'''







  a						:		str											=      (
      list(range(ord("!"	)  ,       ord("~"	) + 1	)	) + list(range(ord("¡"	)  ,       ord("¬"	) + 1	)	) + list(range(ord("®"	)  ,       ord("ÿ"	) + 1	)	)
  )
  a						:		Optional[Any]											=      bs[:]
  a						:		int											=      0
  for b in range(2**8	):
    if b not in bs:
      bs.append(__lowerCAmelCase	)
      cs.append(2**8 + n	)
      n += 1
  a						:		Tuple											=      [chr(__lowerCAmelCase	) for n in cs]
  return dict(zip(__lowerCAmelCase  ,       __lowerCAmelCase	)	)


def 					lowercase      (    A_	)->				List[str]:
  '''simple docstring'''







  a						:		int											=      set()
  a						:		List[Any]											=      word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)	)
    a						:		List[str]											=      char
  return pairs







class 						_A  (  lowerCamelCase_  ):







    """simple docstring"""




    UpperCAmelCase	:							Any     =	VOCAB_FILES_NAMES
    UpperCAmelCase	:							List[str]     =	PRETRAINED_VOCAB_FILES_MAP
    UpperCAmelCase	:							int     =	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    UpperCAmelCase	:							Union[str, Any]     =	['''input_ids''', '''attention_mask''']






    def __init__( self    :   int    ,			__UpperCAmelCase    :   int    ,			__UpperCAmelCase    :   int    ,			__UpperCAmelCase    :   Any="replace"    ,			__UpperCAmelCase    :   Optional[int]="<s>"    ,			__UpperCAmelCase    :   int="</s>"    ,			__UpperCAmelCase    :   Optional[Any]="</s>"    ,			__UpperCAmelCase    :   Union[str, Any]="<s>"    ,			__UpperCAmelCase    :   List[str]="<unk>"    ,			__UpperCAmelCase    :   Union[str, Any]="<pad>"    ,			__UpperCAmelCase    :   List[Any]="<mask>"    ,			__UpperCAmelCase    :   Optional[Any]=False    ,			**__UpperCAmelCase    :   str    ,			):
      a						:		Optional[int]											=      AddedToken(__snake_case    ,			lstrip=__snake_case    ,			rstrip=__snake_case) if isinstance(__snake_case    ,			__snake_case) else bos_token
      a						:		Dict											=      AddedToken(__snake_case    ,			lstrip=__snake_case    ,			rstrip=__snake_case) if isinstance(__snake_case    ,			__snake_case) else eos_token
      a						:		Union[str, Any]											=      AddedToken(__snake_case    ,			lstrip=__snake_case    ,			rstrip=__snake_case) if isinstance(__snake_case    ,			__snake_case) else sep_token
      a						:		Any											=      AddedToken(__snake_case    ,			lstrip=__snake_case    ,			rstrip=__snake_case) if isinstance(__snake_case    ,			__snake_case) else cls_token
      a						:		Optional[int]											=      AddedToken(__snake_case    ,			lstrip=__snake_case    ,			rstrip=__snake_case) if isinstance(__snake_case    ,			__snake_case) else unk_token
      a						:		Dict											=      AddedToken(__snake_case    ,			lstrip=__snake_case    ,			rstrip=__snake_case) if isinstance(__snake_case    ,			__snake_case) else pad_token

      # Mask token behave like a normal word, i.e. include the space before it
      a						:		int											=      AddedToken(__snake_case    ,			lstrip=__snake_case    ,			rstrip=__snake_case) if isinstance(__snake_case    ,			__snake_case) else mask_token

      super().__init__(
          errors=__snake_case    ,			bos_token=__snake_case    ,			eos_token=__snake_case    ,			unk_token=__snake_case    ,			sep_token=__snake_case    ,			cls_token=__snake_case    ,			pad_token=__snake_case    ,			mask_token=__snake_case    ,			add_prefix_space=__snake_case    ,			**__snake_case    ,			)

      with open(__snake_case    ,			encoding="utf-8") as vocab_handle:
        a						:		str											=      json.load(__snake_case)
      a						:		int											=      {v: k for k, v in self.encoder.items()}
      a						:		List[str]											=      errors  # how to handle errors in decoding
      a						:		str											=      bytes_to_unicode()
      a						:		Dict											=      {v: k for k, v in self.byte_encoder.items()}
      with open(__snake_case    ,			encoding="utf-8") as merges_handle:
        a						:		str											=      merges_handle.read().split("\n")[1:-1]
      a						:		Any											=      [tuple(merge.split()) for merge in bpe_merges]
      a						:		Any											=      dict(zip(__snake_case    ,			range(len(__snake_case))))
      a						:		Any											=      {}
      a						:		str											=      add_prefix_space

      # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
      a						:		str											=      re.compile(r"\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+")






    @property
    def      __snake_case							( self    :   int):
      return len(self.encoder)






    def      __snake_case							( self    :   Optional[int]):
      return dict(self.encoder    ,			**self.added_tokens_encoder)






    def      __snake_case							( self    :   Tuple    ,			__UpperCAmelCase    :   List[Any]):
      if token in self.cache:
        return self.cache[token]
      a						:		Any											=      tuple(__snake_case)
      a						:		Dict											=      get_pairs(__snake_case)

      if not pairs:
        return token

      while True:
        a						:		Optional[Any]											=      min(__snake_case    ,			key=lambda __UpperCAmelCase: self.bpe_ranks.get(__snake_case    ,			float("inf")))
        if bigram not in self.bpe_ranks:
          break
        a							,		a						:		Optional[Any]											=      bigram
        a						:		Union[str, Any]											=      []
        a						:		Dict											=      0
        while i < len(__snake_case):
          try:
            a						:		Union[str, Any]											=      word.index(__snake_case    ,			__snake_case)
          except ValueError:
            new_word.extend(word[i:])
            break
          else:
            new_word.extend(word[i:j])
            a						:		Tuple											=      j

          if word[i] == first and i < len(__snake_case) - 1 and word[i + 1] == second:
            new_word.append(first + second)
            i += 2
          else:
            new_word.append(word[i])
            i += 1
        a						:		Any											=      tuple(__snake_case)
        a						:		Union[str, Any]											=      new_word
        if len(__snake_case) == 1:
          break
        else:
          a						:		Optional[Any]											=      get_pairs(__snake_case)
      a						:		List[Any]											=      " ".join(__snake_case)
      a						:		Tuple											=      word
      return word






    def      __snake_case							( self    :   List[Any]    ,			__UpperCAmelCase    :   Optional[Any]):
      a						:		str											=      []
      for token in re.findall(self.pat    ,			__snake_case):
        a						:		int											=      "".join(
            self.byte_encoder[b] for b in token.encode("utf-8"))  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
        bpe_tokens.extend(bpe_token for bpe_token in self.bpe(__snake_case).split(" "))
      return bpe_tokens






    def      __snake_case							( self    :   List[str]    ,			__UpperCAmelCase    :   Dict):
      return self.encoder.get(__snake_case    ,			self.encoder.get(self.unk_token))






    def      __snake_case							( self    :   str    ,			__UpperCAmelCase    :   Dict):
      return self.decoder.get(__snake_case)






    def      __snake_case							( self    :   Any    ,			__UpperCAmelCase    :   List[str]):
      a						:		Dict											=      "".join(__snake_case)
      a						:		int											=      bytearray([self.byte_decoder[c] for c in text]).decode("utf-8"    ,			errors=self.errors)
      return text






    def      __snake_case							( self    :   str    ,			__UpperCAmelCase    :   str    ,			__UpperCAmelCase    :   Optional[str] = None):
      if not os.path.isdir(__snake_case):
        logger.error(f'''Vocabulary path ({save_directory}) should be a directory''')
        return
      a						:		Optional[Any]											=      os.path.join(
          __snake_case    ,			(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])
      a						:		Any											=      os.path.join(
          __snake_case    ,			(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"])

      with open(__snake_case    ,			"w"    ,			encoding="utf-8") as f:
        f.write(json.dumps(self.encoder    ,			indent=2    ,			sort_keys=__snake_case    ,			ensure_ascii=__snake_case) + "\n")

      a						:		Dict											=      0
      with open(__snake_case    ,			"w"    ,			encoding="utf-8") as writer:
        writer.write("#version: 0.2\n")
        for bpe_tokens, token_index in sorted(self.bpe_ranks.items()    ,			key=lambda __UpperCAmelCase: kv[1]):
          if index != token_index:
            logger.warning(
                f'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
                " Please check that the tokenizer is not corrupted!")
            a						:		str											=      token_index
          writer.write(" ".join(__snake_case) + "\n")
          index += 1

      return vocab_file, merge_file






    def      __snake_case							( self    :   List[Any]    ,			__UpperCAmelCase    :   List[int]    ,			__UpperCAmelCase    :   Optional[List[int]] = None):
      if token_ids_a is None:
        return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
      a						:		Optional[int]											=      [self.cls_token_id]
      a						:		int											=      [self.sep_token_id]
      return cls + token_ids_a + sep + sep + token_ids_a + sep






    def      __snake_case							( self    :   Union[str, Any]    ,			__UpperCAmelCase    :   List[int]    ,			__UpperCAmelCase    :   Optional[List[int]] = None    ,			__UpperCAmelCase    :   bool = False):
      if already_has_special_tokens:
        return super().get_special_tokens_mask(
            token_ids_a=__snake_case    ,			token_ids_a=__snake_case    ,			already_has_special_tokens=__snake_case)

      if token_ids_a is None:
        return [1] + ([0] * len(__snake_case)) + [1]
      return [1] + ([0] * len(__snake_case)) + [1, 1] + ([0] * len(__snake_case)) + [1]






    def      __snake_case							( self    :   List[Any]    ,			__UpperCAmelCase    :   List[int]    ,			__UpperCAmelCase    :   Optional[List[int]] = None):
      a						:		Dict											=      [self.sep_token_id]
      a						:		Dict											=      [self.cls_token_id]

      if token_ids_a is None:
        return len(cls + token_ids_a + sep) * [0]
      return len(cls + token_ids_a + sep + sep + token_ids_a + sep) * [0]






    def      __snake_case							( self    :   Optional[int]    ,			__UpperCAmelCase    :   int    ,			__UpperCAmelCase    :   Optional[Any]=False    ,			**__UpperCAmelCase    :   Tuple):
      a						:		List[str]											=      kwargs.pop("add_prefix_space"    ,			self.add_prefix_space)
      if (is_split_into_words or add_prefix_space) and (len(__snake_case) > 0 and not text[0].isspace()):
        a						:		List[str]											=      " " + text
      return (text, kwargs)










"""simple docstring"""







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, ChunkPipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_torch_available():
 import torch

 from transformers.modeling_outputs import BaseModelOutput

 from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

__lowercase    =				logging.get_logger(__name__)






@add_end_docstrings(_a  )
class 						_A  (  _a  ):







    """simple docstring"""




    def __init__( self    :   List[Any]    ,			**__UpperCAmelCase    :   List[Any]):
      super().__init__(**__UpperCAmelCase)

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

      requires_backends(self    ,			"vision")
      self.check_model_type(__UpperCAmelCase)






    def __call__( self    :   str    ,			__UpperCAmelCase    :   Union[str, "Image.Image", List[Dict[str, Any]]]    ,			__UpperCAmelCase    :   Union[str, List[str]] = None    ,			**__UpperCAmelCase    :   List[Any]    ,			):
      if "text_queries" in kwargs:
        a						:		List[Any]											=      kwargs.pop("text_queries")

      if isinstance(__UpperCAmelCase    ,			(str, Image.Image)):
        a						:		Any											=      {"image": image, "candidate_labels": candidate_labels}
      else:
        a						:		Optional[int]											=      image
      a						:		Optional[int]											=      super().__call__(__UpperCAmelCase    ,			**__UpperCAmelCase)
      return results






    def      __snake_case							( self    :   Optional[int]    ,			**__UpperCAmelCase    :   List[Any]):
      a						:		str											=      {}
      if "threshold" in kwargs:
        a						:		Dict											=      kwargs["threshold"]
      if "top_k" in kwargs:
        a						:		str											=      kwargs["top_k"]
      return {}, {}, postprocess_params






    def      __snake_case							( self    :   List[Any]    ,			__UpperCAmelCase    :   Optional[Any]):
      a						:		Union[str, Any]											=      load_image(inputs["image"])
      a						:		Any											=      inputs["candidate_labels"]
      if isinstance(__UpperCAmelCase    ,			__UpperCAmelCase):
        a						:		Optional[Any]											=      candidate_labels.split(",")

      a						:		Union[str, Any]											=      torch.tensor([[image.height, image.width]]    ,			dtype=torch.intaa)
      for i, candidate_label in enumerate(__UpperCAmelCase):
        a						:		int											=      self.tokenizer(__UpperCAmelCase    ,			return_tensors=self.framework)
        a						:		int											=      self.image_processor(__UpperCAmelCase    ,			return_tensors=self.framework)
        yield {
            "is_last": i == len(__UpperCAmelCase) - 1,
            "target_size": target_size,
            "candidate_label": candidate_label,
            **text_inputs,
            **image_features,
        }






    def      __snake_case							( self    :   Dict    ,			__UpperCAmelCase    :   Optional[int]):
      a						:		List[Any]											=      model_inputs.pop("target_size")
      a						:		Optional[int]											=      model_inputs.pop("candidate_label")
      a						:		List[Any]											=      model_inputs.pop("is_last")

      a						:		List[Any]											=      self.model(**__UpperCAmelCase)

      a						:		Union[str, Any]											=      {"target_size": target_size, "candidate_label": candidate_label, "is_last": is_last, **outputs}
      return model_outputs






    def      __snake_case							( self    :   str    ,			__UpperCAmelCase    :   Dict    ,			__UpperCAmelCase    :   Optional[int]=0.1    ,			__UpperCAmelCase    :   List[str]=None):
      a						:		Dict											=      []
      for model_output in model_outputs:
        a						:		int											=      model_output["candidate_label"]
        a						:		Any											=      BaseModelOutput(__UpperCAmelCase)
        a						:		Optional[Any]											=      self.image_processor.post_process_object_detection(
            outputs=__UpperCAmelCase    ,			threshold=__UpperCAmelCase    ,			target_sizes=model_output["target_size"])[0]

        for index in outputs["scores"].nonzero():
          a						:		Any											=      outputs["scores"][index].item()
          a						:		str											=      self._get_bounding_box(outputs["boxes"][index][0])

          a						:		Optional[Any]											=      {"score": score, "label": label, "box": box}
          results.append(__UpperCAmelCase)

      a						:		str											=      sorted(__UpperCAmelCase    ,			key=lambda __UpperCAmelCase: x["score"]    ,			reverse=__UpperCAmelCase)
      if top_k:
        a						:		Union[str, Any]											=      results[:top_k]

      return results






    def      __snake_case							( self    :   Union[str, Any]    ,			__UpperCAmelCase    :   "torch.Tensor"):
      if self.framework != "pt":
        raise ValueError("The ZeroShotObjectDetectionPipeline is only available in PyTorch.")
      a							,		a							,		a							,		a						:		List[Any]											=      box.int().tolist()
      a						:		str											=      {
          "xmin": xmin,
          "ymin": ymin,
          "xmax": xmax,
          "ymax": ymax,
      }
      return bbox






