





import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


__UpperCAmelCase       :   Dict				       =							logging.get_logger(__name__)
__UpperCAmelCase       :   List[Any]				       =							{"vocab_file": "spiece.model"}

__UpperCAmelCase       :   int				       =							{
    "vocab_file": {
        "albert-base-v1": "https://huggingface.co/albert-base-v1/resolve/main/spiece.model",
        "albert-large-v1": "https://huggingface.co/albert-large-v1/resolve/main/spiece.model",
        "albert-xlarge-v1": "https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model",
        "albert-xxlarge-v1": "https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model",
        "albert-base-v2": "https://huggingface.co/albert-base-v2/resolve/main/spiece.model",
        "albert-large-v2": "https://huggingface.co/albert-large-v2/resolve/main/spiece.model",
        "albert-xlarge-v2": "https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model",
        "albert-xxlarge-v2": "https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model",
    }
}

__UpperCAmelCase       :   List[str]				       =							{
    "albert-base-v1": 512,
    "albert-large-v1": 512,
    "albert-xlarge-v1": 512,
    "albert-xxlarge-v1": 512,
    "albert-base-v2": 512,
    "albert-large-v2": 512,
    "albert-xlarge-v2": 512,
    "albert-xxlarge-v2": 512,
}

__UpperCAmelCase       :   Any				       =							"▁"







class     __snake_case       (							__lowerCamelCase			):





       '''simple docstring'''







       lowerCAmelCase__      		=     VOCAB_FILES_NAMES
       lowerCAmelCase__      		=     PRETRAINED_VOCAB_FILES_MAP
       lowerCAmelCase__      		=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES




       def __init__(  self :    Any							,				A :    Tuple							,				A :    Optional[int]=True							,				A :    List[str]=True							,				A :    Any=False							,				A :    str="[CLS]"							,				A :    Any="[SEP]"							,				A :    Tuple="<unk>"							,				A :    Any="[SEP]"							,				A :    Optional[int]="<pad>"							,				A :    Any="[CLS]"							,				A :    List[Any]="[MASK]"							,				A :    Optional[Dict[str, Any]] = None							,				**A :    Dict							,				):
         # Mask token behave like a normal word, i.e. include the space before it and
         # is included in the raw text, there should be a match in a non-normalized sentence.
         __snake_case:						str   =						(
             AddedToken(A							,				lstrip=A							,				rstrip=A							,				normalized=A		)
             if isinstance(A							,				A		)
             else mask_token
         )

         __snake_case:						Optional[int]   =						{} if sp_model_kwargs is None else sp_model_kwargs

         super().__init__(
             do_lower_case=A							,				remove_space=A							,				keep_accents=A							,				bos_token=A							,				eos_token=A							,				unk_token=A							,				sep_token=A							,				pad_token=A							,				cls_token=A							,				mask_token=A							,				sp_model_kwargs=self.sp_model_kwargs							,				**A							,				)

         __snake_case:						Tuple   =						do_lower_case
         __snake_case:						Union[str, Any]   =						remove_space
         __snake_case:						Union[str, Any]   =						keep_accents
         __snake_case:						Tuple   =						vocab_file

         __snake_case:						int   =						spm.SentencePieceProcessor(**self.sp_model_kwargs		)
         self.sp_model.Load(A		)




       @property
       def 							UpperCAmelCase__ (  self :    Dict		):
         return len(self.sp_model		)




       def 							UpperCAmelCase__ (  self :    List[Any]		):
         __snake_case:						Any   =						{self.convert_ids_to_tokens(A		): i for i in range(self.vocab_size		)}
         vocab.update(self.added_tokens_encoder		)
         return vocab




       def __getstate__(  self :    List[str]		):
         __snake_case:						Any   =						self.__dict__.copy()
         __snake_case:						List[Any]   =						None
         return state




       def __setstate__(  self :    str							,				A :    Dict		):
         __snake_case:						Dict   =						d

         # for backward compatibility
         if not hasattr(self							,				"""sp_model_kwargs"""		):
           __snake_case:						Optional[int]   =						{}

         __snake_case:						Optional[int]   =						spm.SentencePieceProcessor(**self.sp_model_kwargs		)
         self.sp_model.Load(self.vocab_file		)




       def 							UpperCAmelCase__ (  self :    List[Any]							,				A :    str		):
         if self.remove_space:
           __snake_case:						Optional[int]   =						""" """.join(inputs.strip().split()		)
         else:
           __snake_case:						str   =						inputs
         __snake_case:						Any   =						outputs.replace("""``"""							,				"""\""""		).replace("""''"""							,				"""\""""		)

         if not self.keep_accents:
           __snake_case:						Any   =						unicodedata.normalize("""NFKD"""							,				A		)
           __snake_case:						str   =						"""""".join([c for c in outputs if not unicodedata.combining(A		)]		)
         if self.do_lower_case:
           __snake_case:						int   =						outputs.lower()

         return outputs




       def 							UpperCAmelCase__ (  self :    Optional[int]							,				A :    str		):
         __snake_case:						Tuple   =						self.preprocess_text(A		)
         __snake_case:						Optional[Any]   =						self.sp_model.encode(A							,				out_type=A		)
         __snake_case:						int   =						[]
         for piece in pieces:
           if len(A		) > 1 and piece[-1] == str(""","""		) and piece[-2].isdigit():
             __snake_case:						Dict   =						self.sp_model.EncodeAsPieces(piece[:-1].replace(A							,				""""""		)		)
             if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
               if len(cur_pieces[0]		) == 1:
                 __snake_case:						Any   =						cur_pieces[1:]
               else:
                 __snake_case:						List[Any]   =						cur_pieces[0][1:]
             cur_pieces.append(piece[-1]		)
             new_pieces.extend(A		)
           else:
             new_pieces.append(A		)

         return new_pieces




       def 							UpperCAmelCase__ (  self :    List[str]							,				A :    Optional[int]		):
         return self.sp_model.PieceToId(A		)




       def 							UpperCAmelCase__ (  self :    int							,				A :    str		):
         return self.sp_model.IdToPiece(A		)




       def 							UpperCAmelCase__ (  self :    int							,				A :    int		):
         __snake_case:						str   =						[]
         __snake_case:						Any   =						""""""
         __snake_case:						Optional[int]   =						False
         for token in tokens:
           # make sure that special tokens are not decoded using sentencepiece model
           if token in self.all_special_tokens:
             if not prev_is_special:
               out_string += " "
             out_string += self.sp_model.decode(A		) + token
             __snake_case:						List[str]   =						True
             __snake_case:						int   =						[]
           else:
             current_sub_tokens.append(A		)
             __snake_case:						Dict   =						False
         out_string += self.sp_model.decode(A		)
         return out_string.strip()




       def 							UpperCAmelCase__ (  self :    str							,				A :    List[int]							,				A :    Optional[List[int]] = None		):
         __snake_case:						str   =						[self.sep_token_id]
         __snake_case:						Dict   =						[self.cls_token_id]
         if token_ids_a is None:
           return cls + token_ids_a + sep
         return cls + token_ids_a + sep + token_ids_a + sep




       def 							UpperCAmelCase__ (  self :    List[str]							,				A :    List[int]							,				A :    Optional[List[int]] = None							,				A :    bool = False		):

         if already_has_special_tokens:
           return super().get_special_tokens_mask(
               token_ids_a=A							,				token_ids_a=A							,				already_has_special_tokens=A		)

         if token_ids_a is not None:
           return [1] + ([0] * len(A		)) + [1] + ([0] * len(A		)) + [1]
         return [1] + ([0] * len(A		)) + [1]




       def 							UpperCAmelCase__ (  self :    Optional[Any]							,				A :    List[int]							,				A :    Optional[List[int]] = None		):
         __snake_case:						Union[str, Any]   =						[self.sep_token_id]
         __snake_case:						Any   =						[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		) * [0] + len(token_ids_a + sep		) * [1]






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

         if os.path.abspath(self.vocab_file		) != os.path.abspath(A		) and os.path.isfile(self.vocab_file		):
           copyfile(self.vocab_file							,				A		)
         elif not os.path.isfile(self.vocab_file		):
           with open(A							,				"""wb"""		) as fi:
             __snake_case:						Dict   =						self.sp_model.serialized_model_proto()
             fi.write(A		)

         return (out_vocab_file,)













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       :   Any				       =							logging.get_logger(__name__)


__UpperCAmelCase       :   str				       =							Dict[str, Any]
__UpperCAmelCase       :   int				       =							List[Prediction]







@add_end_docstrings(__lowerCamelCase			)
class     __snake_case       (							__lowerCamelCase			):





       '''simple docstring'''







       def __init__(  self :    int							,				*A :    Optional[int]							,				**A :    Optional[int]		):
         super().__init__(*A							,				**A		)

         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 :    List[str]							,				**A :    Tuple		):
         __snake_case:						List[str]   =						{}
         if "threshold" in kwargs:
           __snake_case:						Optional[Any]   =						kwargs["""threshold"""]
         return {}, {}, postprocess_kwargs




       def __call__(  self :    int							,				*A :    Optional[Any]							,				**A :    Tuple		):
         return super().__call__(*A							,				**A		)




       def 							UpperCAmelCase__ (  self :    Optional[int]							,				A :    str		):
         __snake_case:						Optional[Any]   =						load_image(A		)
         __snake_case:						Dict   =						torch.IntTensor([[image.height, image.width]]		)
         __snake_case:						str   =						self.image_processor(images=[image]							,				return_tensors="""pt"""		)
         if self.tokenizer is not None:
           __snake_case:						Optional[Any]   =						self.tokenizer(text=inputs["""words"""]							,				boxes=inputs["""boxes"""]							,				return_tensors="""pt"""		)
         __snake_case:						Any   =						target_size
         return inputs




       def 							UpperCAmelCase__ (  self :    Optional[int]							,				A :    Dict		):
         __snake_case:						int   =						model_inputs.pop("""target_size"""		)
         __snake_case:						int   =						self.model(**A		)
         __snake_case:						Any   =						outputs.__class__({"""target_size""": target_size, **outputs}		)
         if self.tokenizer is not None:
           __snake_case:						Optional[int]   =						model_inputs["""bbox"""]
         return model_outputs




       def 							UpperCAmelCase__ (  self :    List[Any]							,				A :    Optional[int]							,				A :    Union[str, Any]=0.9		):
         __snake_case:						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.
           __snake_case  ,   __snake_case:						Union[str, Any]   =						target_size[0].tolist()

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

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

           __snake_case:						Union[str, Any]   =						scores.tolist()
           __snake_case:						List[str]   =						[self.model.config.idalabel[label.item()] for label in labels]
           __snake_case:						List[str]   =						[self._get_bounding_box(A		) for box in boxes]

           # {"scores": [...], ...} --> [{"score":x, ...}, ...]
           __snake_case:						List[Any]   =						["""score""", """label""", """box"""]
           __snake_case:						Dict   =						[
               dict(zip(A							,				A		)		)
               for vals in zip(raw_annotation["""scores"""]							,				raw_annotation["""labels"""]							,				raw_annotation["""boxes"""]		)
           ]

         return annotation






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






