




"""simple docstring"""




import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
			import jieba

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


a_	   =    logging.get_logger(__name__)

a_	   =    {'vocab_file': 'vocab.txt'}

a_	   =    {
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

a_	   =    {
    'openbmb/cpm-ant-10b': 1_0_2_4,
}







def 			__UpperCAmelCase			(      __UpperCamelCase     ):
			__lowercase       :   Any        =    collections.OrderedDict()
			with open(SCREAMING_SNAKE_CASE_     ,							'''r'''     ,							encoding='''utf-8'''     ) as reader:
						__lowercase       :   Dict        =    reader.readlines()
			for index, token in enumerate(SCREAMING_SNAKE_CASE_     ):
						__lowercase       :   str        =    token.rstrip('''\n'''     )
						__lowercase       :   Any        =    index
			return vocab

class    UpperCAmelCase_  (       snake_case						):
		def __init__(		self							,		UpperCamelCase_							,		UpperCamelCase_="<unk>"							,		UpperCamelCase_=2_00    )		->    Dict:
					__lowercase       :   List[str]        =    vocab
					__lowercase       :   Dict        =    unk_token
					__lowercase       :   Any        =    max_input_chars_per_word
		def      _lowerCamelCase				(		self							,		UpperCamelCase_    )		->    List[Any]:
					__lowercase       :   Dict        =    list(UpperCamelCase_    )
					if len(UpperCamelCase_    ) > self.max_input_chars_per_word:
								return [self.unk_token]

					__lowercase       :   Optional[int]        =    0
					__lowercase       :   List[str]        =    []
					while start < len(UpperCamelCase_    ):
								__lowercase       :   Dict        =    len(UpperCamelCase_    )
								__lowercase       :   Optional[Any]        =    None
								while start < end:
											__lowercase       :   List[str]        =    ''.join(chars[start:end]    )
											if substr in self.vocab:
														__lowercase       :   int        =    substr
														break
											end -= 1
								if cur_substr is None:
											sub_tokens.append(self.unk_token    )
											start += 1
								else:
											sub_tokens.append(UpperCamelCase_    )
											__lowercase       :   List[Any]        =    end

					return sub_tokens


class    UpperCAmelCase_  (       snake_case						):
		UpperCamelCase						 =VOCAB_FILES_NAMES
		UpperCamelCase						 =PRETRAINED_VOCAB_FILES_MAP
		UpperCamelCase						 =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		UpperCamelCase						 =["input_ids", "attention_mask"]
		UpperCamelCase						 =False
		def __init__(		self							,		UpperCamelCase_							,		UpperCamelCase_="<d>"							,		UpperCamelCase_="</d>"							,		UpperCamelCase_="<s>"							,		UpperCamelCase_="</s>"							,		UpperCamelCase_="<pad>"							,		UpperCamelCase_="<unk>"							,		UpperCamelCase_="</n>"							,		UpperCamelCase_="</_>"							,		UpperCamelCase_="left"							,		**UpperCamelCase_							,		)		->    Tuple:
					requires_backends(self							,		['''jieba''']    )
					super().__init__(
					    bod_token=UpperCamelCase_							,		eod_token=UpperCamelCase_							,		bos_token=UpperCamelCase_							,		eos_token=UpperCamelCase_							,		pad_token=UpperCamelCase_							,		unk_token=UpperCamelCase_							,		line_token=UpperCamelCase_							,		space_token=UpperCamelCase_							,		padding_side=UpperCamelCase_							,		**UpperCamelCase_							,		)
					__lowercase       :   Dict        =    bod_token
					__lowercase       :   List[Any]        =    eod_token
					__lowercase       :   Any        =    load_vocab(UpperCamelCase_    )
					__lowercase       :   Tuple        =    self.encoder[space_token]
					__lowercase       :   int        =    self.encoder[line_token]

					del self.encoder[space_token]
					del self.encoder[line_token]

					__lowercase       :   str        =    collections.OrderedDict(sorted(self.encoder.items()							,		key=lambda UpperCamelCase_    : x[1]    )    )
					__lowercase       :   Optional[Any]        =    {v: k for k, v in self.encoder.items()}

					__lowercase       :   int        =    WordpieceTokenizer(vocab=self.encoder							,		unk_token=self.unk_token    )
		@property
		def      _lowerCamelCase				(		self    )		->    Dict:
					return self.encoder[self.bod_token]
		@property
		def      _lowerCamelCase				(		self    )		->    Optional[int]:
					return self.encoder[self.eod_token]
		@property
		def      _lowerCamelCase				(		self    )		->    int:
					return self.encoder["\n"]
		@property
		def      _lowerCamelCase				(		self    )		->    int:
					return len(self.encoder    )
		def      _lowerCamelCase				(		self    )		->    Union[str, Any]:
					return dict(self.encoder							,		**self.added_tokens_encoder    )
		def      _lowerCamelCase				(		self							,		UpperCamelCase_    )		->    str:
					__lowercase       :   Optional[Any]        =    []
					for x in jieba.cut(UpperCamelCase_							,		cut_all=UpperCamelCase_    ):
								output_tokens.extend(self.wordpiece_tokenizer.tokenize(UpperCamelCase_    )    )
					return output_tokens
		def      _lowerCamelCase				(		self							,		UpperCamelCase_							,		**UpperCamelCase_    )		->    Optional[int]:
					__lowercase       :   Optional[int]        =    [i for i in token_ids if i >= 0]
					__lowercase       :   Optional[Any]        =    [
					    x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
					]
					return super()._decode(UpperCamelCase_							,		**UpperCamelCase_    )
		def      _lowerCamelCase				(		self							,		UpperCamelCase_    )		->    Dict:
					return token in self.encoder
		def      _lowerCamelCase				(		self							,		UpperCamelCase_    )		->    str:
					return "".join(UpperCamelCase_    )
		def      _lowerCamelCase				(		self							,		UpperCamelCase_    )		->    List[str]:
					return self.encoder.get(UpperCamelCase_							,		self.encoder.get(self.unk_token    )    )
		def      _lowerCamelCase				(		self							,		UpperCamelCase_    )		->    Optional[int]:
					return self.decoder.get(UpperCamelCase_							,		self.unk_token    )
		def      _lowerCamelCase				(		self							,		UpperCamelCase_							,		UpperCamelCase_ = None    )		->    Any:







					if os.path.isdir(UpperCamelCase_    ):
								__lowercase       :   List[str]        =    os.path.join(
								    UpperCamelCase_							,		(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']    )
					else:
								__lowercase       :   Tuple        =    (filename_prefix + '-' if filename_prefix else '') + save_directory
					__lowercase       :   List[str]        =    0
					if " " in self.encoder:
								__lowercase       :   Optional[Any]        =    self.encoder[' ']
								del self.encoder[" "]
					if "\n" in self.encoder:
								__lowercase       :   str        =    self.encoder['\n']
								del self.encoder["\n"]
					__lowercase       :   Any        =    collections.OrderedDict(sorted(self.encoder.items()							,		key=lambda UpperCamelCase_    : x[1]    )    )
					with open(UpperCamelCase_							,		'''w'''							,		encoding='''utf-8'''    ) as writer:
								for token, token_index in self.encoder.items():
											if index != token_index:
														logger.warning(
														    F"""Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."""
														    ''' Please check that the vocabulary is not corrupted!'''    )
														__lowercase       :   Tuple        =    token_index
											writer.write(token + '''\n'''    )
											index += 1
					return (vocab_file,)
		def      _lowerCamelCase				(		self							,		UpperCamelCase_							,		UpperCamelCase_ = None    )		->    List[str]:







					if token_ids_a is None:
								return [self.bos_token_id] + token_ids_a
					return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a
		def      _lowerCamelCase				(		self							,		UpperCamelCase_							,		UpperCamelCase_ = None							,		UpperCamelCase_ = False    )		->    Dict:







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

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





from typing import List, Union

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


if is_vision_available():
					from PIL import Image

					from ..image_utils import load_image

if is_tf_available():
					import tensorflow as tf

					from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
					from ..tf_utils import stable_softmax

if is_torch_available():
					from ..models.auto.modeling_auto import MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING

lowerCamelCase__          =		logging.get_logger(__name__)







@add_end_docstrings(__magic_name__	)
class      A__							(     __magic_name__	):






				def __init__(						self					:      int    ,     *a					:      Dict    ,     **a					:      Union[str, Any]   ):






							'''simple docstring'''







							super().__init__(*a    ,     **a   )
							requires_backends(self    ,     'vision'   )
							self.check_model_type(
							    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
							    if self.framework == 'tf'
							    else MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING   )






				def 							_lowerCamelCase		(						self					:      Dict    ,     a					:      List[str]=None   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Any         =		{}
							if top_k is not None:
										lowerCAmelCase__      :      Tuple         =		top_k
							return {}, {}, postprocess_params






				def __call__(						self					:      Any    ,     a					:      Union[str, List[str], "Image.Image", List["Image.Image"]]    ,     **a					:      List[Any]   ):






							'''simple docstring'''







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






				def 							_lowerCamelCase		(						self					:      Any    ,     a					:      Optional[Any]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      List[Any]         =		load_image(a   )
							lowerCAmelCase__      :      Optional[int]         =		self.image_processor(images=a    ,     return_tensors=self.framework   )
							return model_inputs






				def 							_lowerCamelCase		(						self					:      Optional[int]    ,     a					:      List[str]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Optional[Any]         =		self.model(**a   )
							return model_outputs






				def 							_lowerCamelCase		(						self					:      Optional[Any]    ,     a					:      List[Any]    ,     a					:      List[Any]=5   ):






							'''simple docstring'''







							if top_k > self.model.config.num_labels:
										lowerCAmelCase__      :      Optional[int]         =		self.model.config.num_labels

							if self.framework == "pt":
										lowerCAmelCase__      :      List[Any]         =		model_outputs.logits.softmax(-1   )[0]
										lowerCAmelCase__      ,				lowerCAmelCase__      :      Optional[int]         =		probs.topk(a   )
							elif self.framework == "tf":
										lowerCAmelCase__      :      Any         =		stable_softmax(model_outputs.logits    ,     axis=-1   )[0]
										lowerCAmelCase__      :      Any         =		tf.math.top_k(a    ,     k=a   )
										lowerCAmelCase__      ,				lowerCAmelCase__      :      int         =		topk.values.numpy(), topk.indices.numpy()
							else:
										raise ValueError(f'''Unsupported framework: {self.framework}'''   )

							lowerCAmelCase__      :      List[Any]         =		scores.tolist()
							lowerCAmelCase__      :      List[str]         =		ids.tolist()
							return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(a    ,     a   )]