






"""simple docstring"""



from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
						from ..models.auto import AutoTokenizer







class        lowercase__      :


			def      UpperCAmelCase__						(				self					:  Optional[int]       ,	snake_case__					:  Optional[Any] ):
										raise NotImplementedError()


			def      UpperCAmelCase__						(				self					:  Union[str, Any] ):
										raise NotImplementedError()







class        lowercase__      (							snake_case__       ):


			def __init__(				self					:  Optional[int]       ,	snake_case__					:  "AutoTokenizer"       ,	snake_case__					:  bool = False       ,	**snake_case__					:  int ):
										lowerCamelCase_	:							int									=tokenizer
										lowerCamelCase_	:							Optional[int]									=skip_prompt
										lowerCamelCase_	:							Optional[int]									=decode_kwargs

										# variables used in the streaming process
										lowerCamelCase_	:							Any									=[]
										lowerCamelCase_	:							int									=0
										lowerCamelCase_	:							Union[str, Any]									=True


			def      UpperCAmelCase__						(				self					:  Tuple       ,	snake_case__					:  str ):
										if len(value.shape ) > 1 and value.shape[0] > 1:
																	raise ValueError("TextStreamer only supports batch size 1" )
										elif len(value.shape ) > 1:
																	lowerCamelCase_	:							Tuple									=value[0]

										if self.skip_prompt and self.next_tokens_are_prompt:
																	lowerCamelCase_	:							List[str]									=False
																	return

										# Add the new token to the cache and decodes the entire thing.
										self.token_cache.extend(value.tolist() )
										lowerCamelCase_	:							List[str]									=self.tokenizer.decode(self.token_cache       ,	**self.decode_kwargs )

										# After the symbol for a new line, we flush the cache.
										if text.endswith("\n" ):
																	lowerCamelCase_	:							Any									=text[self.print_len :]
																	lowerCamelCase_	:							List[Any]									=[]
																	lowerCamelCase_	:							Tuple									=0
										# If the last token is a CJK character, we print the characters.
										elif len(snake_case__ ) > 0 and self._is_chinese_char(ord(text[-1] ) ):
																	lowerCamelCase_	:							List[Any]									=text[self.print_len :]
																	self.print_len += len(snake_case__ )
										# 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:
																	lowerCamelCase_	:							Optional[int]									=text[self.print_len : text.rfind(" " ) + 1]
																	self.print_len += len(snake_case__ )

										self.on_finalized_text(snake_case__ )


			def      UpperCAmelCase__						(				self					:  Any ):
										# Flush the cache, if it exists
										if len(self.token_cache ) > 0:
																	lowerCamelCase_	:							Dict									=self.tokenizer.decode(self.token_cache       ,	**self.decode_kwargs )
																	lowerCamelCase_	:							List[Any]									=text[self.print_len :]
																	lowerCamelCase_	:							int									=[]
																	lowerCamelCase_	:							List[str]									=0
										else:
																	lowerCamelCase_	:							List[Any]									=""

										lowerCamelCase_	:							str									=True
										self.on_finalized_text(snake_case__       ,	stream_end=snake_case__ )


			def      UpperCAmelCase__						(				self					:  int       ,	snake_case__					:  str       ,	snake_case__					:  bool = False ):
										print(snake_case__       ,	flush=snake_case__       ,	end="" if not stream_end else None )


			def      UpperCAmelCase__						(				self					:  Tuple       ,	snake_case__					:  int ):
										# This defines a "chinese character" as anything in the CJK Unicode block:
										#   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
										#
										# Note that the CJK Unicode block is NOT all Japanese and Korean characters,
										# despite its name. The modern Korean Hangul alphabet is a different block,
										# as is Japanese Hiragana and Katakana. Those alphabets are used to write
										# space-separated words, so they are not treated specially and handled
										# like the all of the other languages.
										if (
										    (cp >= 0x4_e00 and cp <= 0x9_fff)
										    or (cp >= 0x3_400 and cp <= 0x4_dbf)  #
										    or (cp >= 0x20_000 and cp <= 0x2a_6df)  #
										    or (cp >= 0x2a_700 and cp <= 0x2b_73f)  #
										    or (cp >= 0x2b_740 and cp <= 0x2b_81f)  #
										    or (cp >= 0x2b_820 and cp <= 0x2c_eaf)  #
										    or (cp >= 0xf_900 and cp <= 0xf_aff)
										    or (cp >= 0x2f_800 and cp <= 0x2f_a1f)  #
										):  #
																	return True

										return False







class        lowercase__      (							snake_case__       ):


			def __init__(				self					:  List[str]       ,	snake_case__					:  "AutoTokenizer"       ,	snake_case__					:  bool = False       ,	snake_case__					:  Optional[float] = None       ,	**snake_case__					:  Dict ):
										super().__init__(snake_case__       ,	snake_case__       ,	**snake_case__ )
										lowerCamelCase_	:							Optional[Any]									=Queue()
										lowerCamelCase_	:							Union[str, Any]									=None
										lowerCamelCase_	:							List[Any]									=timeout


			def      UpperCAmelCase__						(				self					:  Any       ,	snake_case__					:  str       ,	snake_case__					:  bool = False ):
										self.text_queue.put(snake_case__       ,	timeout=self.timeout )
										if stream_end:
																	self.text_queue.put(self.stop_signal       ,	timeout=self.timeout )


			def __iter__(				self					:  Tuple ):
										return self


			def      UpperCAmelCase__						(				self					:  Union[str, Any] ):
										lowerCamelCase_	:							Optional[int]									=self.text_queue.get(timeout=self.timeout )
										if value == self.stop_signal:
																	raise StopIteration()
										else:
																	return value











"""simple docstring"""



import copy
import json
import os
import tempfile

from transformers import is_torch_available

from .test_configuration_utils import config_common_kwargs







class        lowercase__      (							snake_case__       ):


			def __init__(				self					:  Tuple       ,	snake_case__					:  Optional[int]       ,	snake_case__					:  int=None       ,	snake_case__					:  Union[str, Any]=True       ,	snake_case__					:  Optional[int]=None       ,	**snake_case__					:  Optional[int] ):
										lowerCamelCase_	:							Dict									=parent
										lowerCamelCase_	:							List[str]									=config_class
										lowerCamelCase_	:							Union[str, Any]									=has_text_modality
										lowerCamelCase_	:							Optional[int]									=kwargs
										lowerCamelCase_	:							List[str]									=common_properties


			def      UpperCAmelCase__						(				self					:  Optional[Any] ):
										lowerCamelCase_	:							List[str]									=self.config_class(**self.inputs_dict )
										lowerCamelCase_	:							Any									=(
										    ["hidden_size", "num_attention_heads", "num_hidden_layers"]
										    if self.common_properties is None
										    else self.common_properties
										)

										# Add common fields for text models
										if self.has_text_modality:
																	common_properties.extend(["vocab_size"] )

										# Test that config has the common properties as getters
										for prop in common_properties:
																	self.parent.assertTrue(hasattr(snake_case__       ,	snake_case__ )       ,	msg=F"""`{prop}` does not exist""" )

										# Test that config has the common properties as setter
										for idx, name in enumerate(snake_case__ ):
																	try:
																								setattr(snake_case__       ,	snake_case__       ,	snake_case__ )
																								self.parent.assertEqual(
																								    getattr(snake_case__       ,	snake_case__ )       ,	snake_case__       ,	msg=F"""`{name} value {idx} expected, but was {getattr(snake_case__       ,	snake_case__ )}""" )
																	except NotImplementedError:
																								# Some models might not be able to implement setters for common_properties
																								# In that case, a NotImplementedError is raised
																								pass

        # Test if config class can be called with Config(prop_name=..)
										for idx, name in enumerate(snake_case__ ):
																	try:
																								lowerCamelCase_	:							Dict									=self.config_class(**{name: idx} )
																								self.parent.assertEqual(
																								    getattr(snake_case__       ,	snake_case__ )       ,	snake_case__       ,	msg=F"""`{name} value {idx} expected, but was {getattr(snake_case__       ,	snake_case__ )}""" )
																	except NotImplementedError:
																								# Some models might not be able to implement setters for common_properties
																								# In that case, a NotImplementedError is raised
																								pass


			def      UpperCAmelCase__						(				self					:  Any ):
										lowerCamelCase_	:							Tuple									=self.config_class(**self.inputs_dict )
										lowerCamelCase_	:							Any									=json.loads(config.to_json_string() )
										for key, value in self.inputs_dict.items():
																	self.parent.assertEqual(obj[key]       ,	snake_case__ )


			def      UpperCAmelCase__						(				self					:  int ):
										lowerCamelCase_	:							Tuple									=self.config_class(**self.inputs_dict )

										with tempfile.TemporaryDirectory() as tmpdirname:
																	lowerCamelCase_	:							List[Any]									=os.path.join(snake_case__       ,	"config.json" )
																	config_first.to_json_file(snake_case__ )
																	lowerCamelCase_	:							Optional[int]									=self.config_class.from_json_file(snake_case__ )

										self.parent.assertEqual(config_second.to_dict()       ,	config_first.to_dict() )


			def      UpperCAmelCase__						(				self					:  Dict ):
										lowerCamelCase_	:							Dict									=self.config_class(**self.inputs_dict )

										with tempfile.TemporaryDirectory() as tmpdirname:
																	config_first.save_pretrained(snake_case__ )
																	lowerCamelCase_	:							Optional[int]									=self.config_class.from_pretrained(snake_case__ )

										self.parent.assertEqual(config_second.to_dict()       ,	config_first.to_dict() )


			def      UpperCAmelCase__						(				self					:  str ):
										lowerCamelCase_	:							Dict									=self.config_class(**self.inputs_dict )

										lowerCamelCase_	:							Dict									="test"
										with tempfile.TemporaryDirectory() as tmpdirname:
																	lowerCamelCase_	:							str									=os.path.join(snake_case__       ,	snake_case__ )
																	config_first.save_pretrained(snake_case__ )
																	lowerCamelCase_	:							Optional[Any]									=self.config_class.from_pretrained(snake_case__       ,	subfolder=snake_case__ )

										self.parent.assertEqual(config_second.to_dict()       ,	config_first.to_dict() )


			def      UpperCAmelCase__						(				self					:  Optional[Any] ):
										lowerCamelCase_	:							Optional[Any]									=self.config_class(**self.inputs_dict       ,	num_labels=5 )
										self.parent.assertEqual(len(config.idalabel )       ,	5 )
										self.parent.assertEqual(len(config.labelaid )       ,	5 )

										lowerCamelCase_	:							List[Any]									=3
										self.parent.assertEqual(len(config.idalabel )       ,	3 )
										self.parent.assertEqual(len(config.labelaid )       ,	3 )


			def      UpperCAmelCase__						(				self					:  List[Any] ):
										if self.config_class.is_composition:
																	return
										lowerCamelCase_	:							Tuple									=self.config_class()
										self.parent.assertIsNotNone(snake_case__ )


			def      UpperCAmelCase__						(				self					:  List[Any] ):
										lowerCamelCase_	:							List[str]									=copy.deepcopy(snake_case__ )
										lowerCamelCase_	:							Optional[int]									=self.config_class(**snake_case__ )
										lowerCamelCase_	:							Union[str, Any]									=[]
										for key, value in config_common_kwargs.items():
																	if key == "torch_dtype":
																								if not is_torch_available():
																															continue
																								else:
																															import torch

																															if config.torch_dtype != torch.floataa:
																																						wrong_values.append(("torch_dtype", config.torch_dtype, torch.floataa) )
																	elif getattr(snake_case__       ,	snake_case__ ) != value:
																								wrong_values.append((key, getattr(snake_case__       ,	snake_case__ ), value) )

										if len(snake_case__ ) > 0:
																	lowerCamelCase_	:							Any									="\n".join([F"""- {v[0]}: got {v[1]} instead of {v[2]}""" for v in wrong_values] )
																	raise ValueError(F"""The following keys were not properly set in the config:\n{errors}""" )


			def      UpperCAmelCase__						(				self					:  int ):
										self.create_and_test_config_common_properties()
										self.create_and_test_config_to_json_string()
										self.create_and_test_config_to_json_file()
										self.create_and_test_config_from_and_save_pretrained()
										self.create_and_test_config_from_and_save_pretrained_subfolder()
										self.create_and_test_config_with_num_labels()
										self.check_config_can_be_init_without_params()
										self.check_config_arguments_init()



