"""simple docstring"""





import re

from flax.core.frozen_dict import freeze
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.experimental import PartitionSpec as P


# Sentinels
lowerCAmelCase_     =	object()

# For specifying empty leaf dict `{}`
lowerCAmelCase_     =	object()



def      lowerCamelCase_(__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE  )->    str:
							_SCREAMING_SNAKE_CASE		:						List[str]			  =    tuple((re.compile(x + """$"""  ) for x in qs)  )
							for i in range(len(__SCREAMING_SNAKE_CASE  ) - len(__SCREAMING_SNAKE_CASE  ) + 1  ):
														_SCREAMING_SNAKE_CASE		:						List[Any]			  =    [x.match(__SCREAMING_SNAKE_CASE  ) for x, y in zip(__SCREAMING_SNAKE_CASE     ,	ks[i:]  )]
														if matches and all(__SCREAMING_SNAKE_CASE  ):
																					return True
							return False



def      lowerCamelCase_(__SCREAMING_SNAKE_CASE  )->    Dict:
							def replace(__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE  ):
														for rule, replacement in rules:
																					if _match(__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE  ):
																												return replacement
														return val

							return replace



def      lowerCamelCase_()->    Optional[int]:
							return [
							    # embeddings
							    (("transformer", "wpe", "embedding"), P("""mp"""     ,	__SCREAMING_SNAKE_CASE  )),
							    (("transformer", "wte", "embedding"), P("""mp"""     ,	__SCREAMING_SNAKE_CASE  )),
							    # atention
							    (("attention", "(q_proj|k_proj|v_proj)", "kernel"), P(__SCREAMING_SNAKE_CASE     ,	"""mp"""  )),
							    (("attention", "out_proj", "kernel"), P("""mp"""     ,	__SCREAMING_SNAKE_CASE  )),
							    (("attention", "out_proj", "bias"), None),
							    # mlp
							    (("mlp", "c_fc", "kernel"), P(__SCREAMING_SNAKE_CASE     ,	"""mp"""  )),
							    (("mlp", "c_fc", "bias"), P("""mp"""  )),
							    (("mlp", "c_proj", "kernel"), P("""mp"""     ,	__SCREAMING_SNAKE_CASE  )),
							    (("mlp", "c_proj", "bias"), None),
							    # layer norms
							    ((r"ln_\d+", "bias"), None),
							    ((r"\d+", r"ln_\d+", "scale"), None),
							    (("ln_f", "bias"), None),
							    (("ln_f", "scale"), None),
							]



def      lowerCamelCase_(__SCREAMING_SNAKE_CASE  )->    Optional[Any]:
							_SCREAMING_SNAKE_CASE		:						Tuple			  =    _get_partition_rules()
							_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    _replacement_rules(__SCREAMING_SNAKE_CASE  )
							_SCREAMING_SNAKE_CASE		:						Union[str, Any]			  =    {k: _unmatched for k in flatten_dict(__SCREAMING_SNAKE_CASE  )}
							_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    {k: replace(__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE  ) for k, v in initd.items()}
							assert _unmatched not in result.values(), "Incomplete partition spec."
							return freeze(unflatten_dict(__SCREAMING_SNAKE_CASE  )  )



"""simple docstring"""





import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
			import torch

			from transformers import AutoModelForCausalLM




@require_torch
class  _snake_case		(    unittest.TestCase ):







				"""simple docstring"""






				def 		_lowerCAmelCase		(      self							:							List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    -1

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Dict			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						Dict			  =    tokenizer.decode(greedy_ids[0])

											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Any			  =    TextStreamer(_A)
																		model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A      ,    streamer=_A)
											# The greedy text should be printed to stdout, except for the final "\n" in the streamer
											_SCREAMING_SNAKE_CASE		:						str			  =    cs.out[:-1]

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							Any):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Any			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    -1

											_SCREAMING_SNAKE_CASE		:						str			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    tokenizer.decode(greedy_ids[0])

											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    TextIteratorStreamer(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    {"""input_ids""": input_ids, """max_new_tokens""": 1_0, """do_sample""": False, """streamer""": streamer}
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    Thread(target=model.generate      ,    kwargs=_A)
											thread.start()
											_SCREAMING_SNAKE_CASE		:						Any			  =    """"""
											for new_text in streamer:
																		streamer_text += new_text

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						Dict			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Any			  =    -1

											_SCREAMING_SNAKE_CASE		:						Any			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A)
											_SCREAMING_SNAKE_CASE		:						str			  =    greedy_ids[:, input_ids.shape[1] :]
											_SCREAMING_SNAKE_CASE		:						Dict			  =    tokenizer.decode(new_greedy_ids[0])

											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Any			  =    TextStreamer(_A      ,    skip_prompt=_A)
																		model.generate(_A      ,    max_new_tokens=1_0      ,    do_sample=_A      ,    streamer=_A)
											# The greedy text should be printed to stdout, except for the final "\n" in the streamer
											_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    cs.out[:-1]

											self.assertEqual(_A      ,    _A)





				def 		_lowerCAmelCase		(      self							:							Dict):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						int			  =    AutoTokenizer.from_pretrained("""distilgpt2""")
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoModelForCausalLM.from_pretrained("""distilgpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						int			  =    -1

											_SCREAMING_SNAKE_CASE		:						List[str]			  =    torch.ones((1, 5)      ,    device=_A).long() * model.config.bos_token_id
											with CaptureStdout() as cs:
																		_SCREAMING_SNAKE_CASE		:						Optional[int]			  =    TextStreamer(_A      ,    skip_special_tokens=_A)
																		model.generate(_A      ,    max_new_tokens=1      ,    do_sample=_A      ,    streamer=_A)

											# The prompt contains a special token, so the streamer should not print it. As such, the output text, when
											# re-tokenized, must only contain one token
											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    cs.out[:-1]  # Remove the final "\n"
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    tokenizer(_A      ,    return_tensors="""pt""")
											self.assertEqual(streamer_text_tokenized.input_ids.shape      ,    (1, 1))





				def 		_lowerCAmelCase		(      self							:							str):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:						Optional[Any]			  =    AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2""")
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    AutoModelForCausalLM.from_pretrained("""hf-internal-testing/tiny-random-gpt2""").to(_A)
											_SCREAMING_SNAKE_CASE		:						Tuple			  =    -1

											_SCREAMING_SNAKE_CASE		:						Dict			  =    ids_tensor((1, 5)      ,    vocab_size=model.config.vocab_size).to(_A)
											_SCREAMING_SNAKE_CASE		:						int			  =    TextIteratorStreamer(_A      ,    timeout=0.001)
											_SCREAMING_SNAKE_CASE		:						List[Any]			  =    {"""input_ids""": input_ids, """max_new_tokens""": 1_0, """do_sample""": False, """streamer""": streamer}
											_SCREAMING_SNAKE_CASE		:						List[str]			  =    Thread(target=model.generate      ,    kwargs=_A)
											thread.start()

											# The streamer will timeout after 0.001 seconds, so an exception will be raised
											with self.assertRaises(_A):
																		_SCREAMING_SNAKE_CASE		:						str			  =    """"""
																		for new_text in streamer:
																									streamer_text += new_text


