'''simple docstring'''

from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
  from ..models.auto import AutoTokenizer


class 					_lowercase						:
      '''simple docstring'''


      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_				)      ->      Any:



         '''simple docstring'''



         raise NotImplementedError()




      def 		lowerCAmelCase__       (					self				)      ->      Dict:



         '''simple docstring'''



         raise NotImplementedError()




class 					_lowercase						(   lowerCAmelCase ):
      '''simple docstring'''


      def __init__(					self       ,lowerCamelCase_       ,lowerCamelCase_ = False       ,**lowerCamelCase_				)      ->      List[Any]:



         '''simple docstring'''



         UpperCAmelCase__				:					Optional[int] 					=	tokenizer
         UpperCAmelCase__				:					Any 					=	skip_prompt
         UpperCAmelCase__				:					str 					=	decode_kwargs

         # variables used in the streaming process
         UpperCAmelCase__				:					Optional[int] 					=	[]
         UpperCAmelCase__				:					List[str] 					=	0
         UpperCAmelCase__				:					Dict 					=	True




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_				)      ->      Optional[Any]:



         '''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__				:					str 					=	value[0]

         if self.skip_prompt and self.next_tokens_are_prompt:
            UpperCAmelCase__				:					Optional[int] 					=	False
            return

         # Add the new token to the cache and decodes the entire thing.
         self.token_cache.extend(value.tolist()				)
         UpperCAmelCase__				:					Optional[Any] 					=	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__				:					Dict 					=	text[self.print_len :]
            UpperCAmelCase__				:					Tuple 					=	[]
            UpperCAmelCase__				:					Dict 					=	0
         # If the last token is a CJK character, we print the characters.
         elif len(lowerCamelCase_				) > 0 and self._is_chinese_char(ord(text[-1]				)				):
            UpperCAmelCase__				:					List[str] 					=	text[self.print_len :]
            self.print_len += len(lowerCamelCase_				)
         # 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__				:					Optional[Any] 					=	text[self.print_len : text.rfind(''' '''				) + 1]
            self.print_len += len(lowerCamelCase_				)

         self.on_finalized_text(lowerCamelCase_				)




      def 		lowerCAmelCase__       (					self				)      ->      List[str]:



         '''simple docstring'''



         if len(self.token_cache				) > 0:
            UpperCAmelCase__				:					Optional[Any] 					=	self.tokenizer.decode(self.token_cache       ,**self.decode_kwargs				)
            UpperCAmelCase__				:					Optional[Any] 					=	text[self.print_len :]
            UpperCAmelCase__				:					List[Any] 					=	[]
            UpperCAmelCase__				:					Dict 					=	0
         else:
            UpperCAmelCase__				:					Any 					=	''''''

         UpperCAmelCase__				:					Dict 					=	True
         self.on_finalized_text(lowerCamelCase_       ,stream_end=lowerCamelCase_				)




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_       ,lowerCamelCase_ = False				)      ->      Optional[Any]:



         '''simple docstring'''



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




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_				)      ->      List[Any]:



         '''simple docstring'''



         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						(   lowerCAmelCase ):
      '''simple docstring'''


      def __init__(					self       ,lowerCamelCase_       ,lowerCamelCase_ = False       ,lowerCamelCase_ = None       ,**lowerCamelCase_				)      ->      Tuple:



         '''simple docstring'''



         super().__init__(lowerCamelCase_       ,lowerCamelCase_       ,**lowerCamelCase_				)
         UpperCAmelCase__				:					int 					=	Queue()
         UpperCAmelCase__				:					Optional[Any] 					=	None
         UpperCAmelCase__				:					Optional[Any] 					=	timeout




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_       ,lowerCamelCase_ = False				)      ->      Dict:



         '''simple docstring'''



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




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



         '''simple docstring'''



         return self




      def 		lowerCAmelCase__       (					self				)      ->      str:



         '''simple docstring'''



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





'''simple docstring'''

import warnings


warnings.warn(
    'memory_utils has been reorganized to utils.memory. Import `find_executable_batchsize` from the main `__init__`: '
    '`from accelerate import find_executable_batch_size` to avoid this warning.',
    FutureWarning,
)




