


'''simple docstring'''



from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
	from ..models.auto import AutoTokenizer






class  lowerCamelCase_:




				def   a__(	self    :				List[Any]   ,					lowerCAmelCase    :				Optional[int]   )->  List[Any]:

										"""simple docstring"""




										raise NotImplementedError()





				def   a__(	self    :				int   )->  str:

										"""simple docstring"""




										raise NotImplementedError()







class  lowerCamelCase_(     __lowercase ):




				def __init__(	self    :				Union[str, Any]   ,					lowerCAmelCase    :				Dict   ,					lowerCAmelCase    :				Dict = False   ,					**lowerCAmelCase    :				Tuple   )->  Dict:

										"""simple docstring"""




										UpperCAmelCase         =			tokenizer
										UpperCAmelCase         =			skip_prompt
										UpperCAmelCase         =			decode_kwargs

										# variables used in the streaming process
										UpperCAmelCase         =			[]
										UpperCAmelCase         =			0
										UpperCAmelCase         =			True




				def   a__(	self    :				List[Any]   ,					lowerCAmelCase    :				int   )->  int:

										"""simple docstring"""




										if len(value.shape   ) > 1 and value.shape[0] > 1:
																raise ValueError('''TextStreamer only supports batch size 1'''   )
										elif len(value.shape   ) > 1:
																UpperCAmelCase         =			value[0]

										if self.skip_prompt and self.next_tokens_are_prompt:
																UpperCAmelCase         =			False
																return

										# Add the new token to the cache and decodes the entire thing.
										self.token_cache.extend(value.tolist()   )
										UpperCAmelCase         =			self.tokenizer.decode(self.token_cache   ,					**self.decode_kwargs   )

										# After the symbol for a new line, we flush the cache.
										if text.endswith('''\n'''   ):
																UpperCAmelCase         =			text[self.print_len :]
																UpperCAmelCase         =			[]
																UpperCAmelCase         =			0
										# If the last token is a CJK character, we print the characters.
										elif len(__A   ) > 0 and self._is_chinese_char(ord(text[-1]   )   ):
																UpperCAmelCase         =			text[self.print_len :]
																self.print_len += len(__A   )
										# Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
										# which may change with the subsequent token -- there are probably smarter ways to do this!)
										else:
																UpperCAmelCase         =			text[self.print_len : text.rfind(''' '''   ) + 1]
																self.print_len += len(__A   )

										self.on_finalized_text(__A   )




				def   a__(	self    :				Dict   )->  int:

										"""simple docstring"""




										if len(self.token_cache   ) > 0:
																UpperCAmelCase         =			self.tokenizer.decode(self.token_cache   ,					**self.decode_kwargs   )
																UpperCAmelCase         =			text[self.print_len :]
																UpperCAmelCase         =			[]
																UpperCAmelCase         =			0
										else:
																UpperCAmelCase         =			''''''

										UpperCAmelCase         =			True
										self.on_finalized_text(__A   ,					stream_end=__A   )




				def   a__(	self    :				Optional[Any]   ,					lowerCAmelCase    :				List[Any]   ,					lowerCAmelCase    :				Tuple = False   )->  Tuple:

										"""simple docstring"""




										print(__A   ,					flush=__A   ,					end='''''' if not stream_end else None   )





				def   a__(	self    :				int   ,					lowerCAmelCase    :				int   )->  str:

										"""simple docstring"""




										if (
										    (cp >= 0X4e_00 and cp <= 0X9f_ff)
										    or (cp >= 0X34_00 and cp <= 0X4d_bf)  #
										    or (cp >= 0X2_00_00 and cp <= 0X2_a6_df)  #
										    or (cp >= 0X2_a7_00 and cp <= 0X2_b7_3f)  #
										    or (cp >= 0X2_b7_40 and cp <= 0X2_b8_1f)  #
										    or (cp >= 0X2_b8_20 and cp <= 0X2_ce_af)  #
										    or (cp >= 0Xf9_00 and cp <= 0Xfa_ff)
										    or (cp >= 0X2_f8_00 and cp <= 0X2_fa_1f)  #
										):  #
																return True

										return False







class  lowerCamelCase_(     __lowercase ):




				def __init__(	self    :				Union[str, Any]   ,					lowerCAmelCase    :				Any   ,					lowerCAmelCase    :				int = False   ,					lowerCAmelCase    :				List[str] = None   ,					**lowerCAmelCase    :				str   )->  List[Any]:

										"""simple docstring"""




										super().__init__(__A   ,					__A   ,					**__A   )
										UpperCAmelCase         =			Queue()
										UpperCAmelCase         =			None
										UpperCAmelCase         =			timeout




				def   a__(	self    :				Any   ,					lowerCAmelCase    :				Dict   ,					lowerCAmelCase    :				int = False   )->  Any:

										"""simple docstring"""




										self.text_queue.put(__A   ,					timeout=self.timeout   )
										if stream_end:
																self.text_queue.put(self.stop_signal   ,					timeout=self.timeout   )




				def __iter__(	self    :				int   )->  Optional[int]:

										"""simple docstring"""




										return self





				def   a__(	self    :				Optional[Any]   )->  List[str]:

										"""simple docstring"""




										UpperCAmelCase         =			self.text_queue.get(timeout=self.timeout   )
										if value == self.stop_signal:
																raise StopIteration()
										else:
																return value









'''simple docstring'''



# Copyright 2022 The HuggingFace Team and The OpenBMB Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

# rely on isort to merge the imports
from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


_lowercase    :    Any			     =  {
    """configuration_cpmant""": ["""CPMANT_PRETRAINED_CONFIG_ARCHIVE_MAP""", """CpmAntConfig"""],
    """tokenization_cpmant""": ["""CpmAntTokenizer"""],
}

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	_lowercase    :    int			     =  [
	    """CPMANT_PRETRAINED_MODEL_ARCHIVE_LIST""",
	    """CpmAntForCausalLM""",
	    """CpmAntModel""",
	    """CpmAntPreTrainedModel""",
	]


if TYPE_CHECKING:
	from .configuration_cpmant import CPMANT_PRETRAINED_CONFIG_ARCHIVE_MAP, CpmAntConfig
	from .tokenization_cpmant import CpmAntTokenizer

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_cpmant import (
		    CPMANT_PRETRAINED_MODEL_ARCHIVE_LIST,
		    CpmAntForCausalLM,
		    CpmAntModel,
		    CpmAntPreTrainedModel,
		)


else:
	import sys

	_lowercase    :    Dict			     =  _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)





